jpmschweitzerandClaude Sonnet 5
8ec89336df
workspace D-70 ("`feedback` needs no address"): the failure channel must
not depend on the thing most likely to break. `scribe_messages.from_identity`
becomes nullable (migration 010); `POST /scribe/messages` accepts a missing
`X-Office-Identity` for exactly one shape — an office caller, `to` is
`["tower-control"]` alone, and `tags` contains `feedback` — stored with
`from_identity` null and `from_consumer` the caller's consumer. Every other
send without an identity is still 400 `identity_required`, enforced in
ScribeService.send_message rather than the caller dependency, since the
decision needs the request body. `MessageResponse.from_identity` is now
optional so mailbox/history/find/message_id answer such a message with a
null sender. Migration 010's downgrade refuses outright while any
sender-less row exists rather than inventing one.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
438 lines
18 KiB
Python
438 lines
18 KiB
Python
"""
|
|
Scribe Domain Service
|
|
|
|
Business logic for the Scribe — the household's message bus (D-66, D-68,
|
|
D-73). Every method starts by requiring a tenant on the caller and ends by
|
|
filtering its query on it: **an address or message in another tenant
|
|
answers exactly like one that does not exist** (dossier core-api T-5), so
|
|
no method here ever confirms an id across that boundary.
|
|
|
|
Methods take `session: AsyncSession` per call rather than storing it on
|
|
`self`, following src/domains/book/service.py.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from typing import List, Optional, Tuple
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.domains.auth.callers import Caller
|
|
from src.domains.scribe.models import (
|
|
RESERVED_ADDRESS_NAMES,
|
|
ScribeAddress,
|
|
ScribeMailboxEntry,
|
|
ScribeMessage,
|
|
)
|
|
from src.domains.scribe.schemas import MailboxPatchRequest, MessageSendRequest
|
|
from src.domains.tenants.service import TenantError, get_tenants_service
|
|
from src.shared.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class ScribeError(Exception):
|
|
"""Raised for a Scribe-domain failure; the controller maps it to a response."""
|
|
|
|
def __init__(self, code: str, status_code: int = 400, **extra):
|
|
self.code = code
|
|
self.status_code = status_code
|
|
self.extra = extra
|
|
super().__init__(code)
|
|
|
|
|
|
class ScribeService:
|
|
"""Service for Scribe operations."""
|
|
|
|
# =========================================================================
|
|
# Internal helpers
|
|
# =========================================================================
|
|
|
|
async def _require_tenant(self, session: AsyncSession, caller: Caller) -> str:
|
|
"""Every Scribe operation needs a tenant; a caller without one gets
|
|
a clear error rather than a query that silently matches nothing.
|
|
|
|
Delegates to the tenant registry's one enforcement point (core-api
|
|
T-8 / workspace D-73), which additionally refuses an unknown
|
|
tenant, a suspended one, or an Office consumer with no grant in it.
|
|
`TenantError` is translated to `ScribeError` so every call site and
|
|
the controller's single `except ScribeError` keep working
|
|
unchanged."""
|
|
try:
|
|
return await get_tenants_service().require_tenant(session, caller)
|
|
except TenantError as error:
|
|
raise ScribeError(error.code, error.status_code) from error
|
|
|
|
def _is_exempt_feedback(self, caller: Caller, recipients: List[str], tags: List[str]) -> bool:
|
|
"""core-api T-15 / workspace D-70 ("`feedback` needs no address"):
|
|
the one shape that may ship with no sender is an office caller's
|
|
`feedback` to `tower-control` alone — nothing broader. `recipients`
|
|
and `tags` are the de-duplicated lists `send_message` already
|
|
computed, so a second or different recipient, or a message missing
|
|
the `feedback` tag, never qualifies."""
|
|
return (
|
|
caller.kind == "office"
|
|
and recipients == ["tower-control"]
|
|
and "feedback" in tags
|
|
)
|
|
|
|
def _validate_address_name(self, name: str) -> None:
|
|
"""Beyond the schema's regex (shape only): not `group-*`, not reserved.
|
|
Mirrors src/domains/auth/callers.py's `_validate_identity` rule —
|
|
an address is named for an identity (D-66)."""
|
|
if name.startswith("group-") or name in RESERVED_ADDRESS_NAMES:
|
|
raise ScribeError("address_invalid", 400)
|
|
|
|
async def _get_address(self, session: AsyncSession, tenant: str, name: str) -> Optional[ScribeAddress]:
|
|
return await session.get(ScribeAddress, (tenant, name))
|
|
|
|
async def _require_owned_identity(self, session: AsyncSession, caller: Caller, tenant: str) -> None:
|
|
"""mailbox/find read the caller's own identity's mailbox: the identity
|
|
must be a registered address owned by the caller's consumer, else
|
|
403 identity_not_owned — the same refusal `send` uses for `from`."""
|
|
address = await self._get_address(session, tenant, caller.identity)
|
|
if address is None or address.owner_consumer != caller.consumer:
|
|
raise ScribeError("identity_not_owned", 403)
|
|
|
|
async def _messages_for_entries(
|
|
self, session: AsyncSession, entries: List[ScribeMailboxEntry]
|
|
) -> dict:
|
|
if not entries:
|
|
return {}
|
|
ids = [entry.message_id for entry in entries]
|
|
stmt = select(ScribeMessage).where(ScribeMessage.id.in_(ids))
|
|
messages = (await session.execute(stmt)).scalars().all()
|
|
return {message.id: message for message in messages}
|
|
|
|
# =========================================================================
|
|
# register / list addresses
|
|
# =========================================================================
|
|
|
|
async def register_address(
|
|
self, session: AsyncSession, name: str, caller: Caller
|
|
) -> Tuple[ScribeAddress, bool]:
|
|
"""`register` — idempotent for the same consumer; 409 address_taken
|
|
for another consumer in the tenant (D-66)."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
self._validate_address_name(name)
|
|
|
|
existing = await self._get_address(session, tenant, name)
|
|
if existing is not None:
|
|
if existing.owner_consumer == caller.consumer:
|
|
return existing, True
|
|
raise ScribeError("address_taken", 409)
|
|
|
|
address = ScribeAddress(
|
|
owner_tenant=tenant,
|
|
name=name,
|
|
owner_consumer=caller.consumer,
|
|
created_by=caller.identity,
|
|
)
|
|
session.add(address)
|
|
await session.commit()
|
|
await session.refresh(address)
|
|
|
|
logger.info(f"Scribe address registered: {tenant}/{name} by={caller.identity}")
|
|
return address, True
|
|
|
|
async def list_addresses(
|
|
self, session: AsyncSession, caller: Caller
|
|
) -> List[Tuple[ScribeAddress, bool]]:
|
|
"""`list` — the caller's tenant; mark which the caller's consumer owns."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
|
|
stmt = (
|
|
select(ScribeAddress)
|
|
.where(ScribeAddress.owner_tenant == tenant)
|
|
.order_by(ScribeAddress.name)
|
|
)
|
|
addresses = (await session.execute(stmt)).scalars().all()
|
|
return [(a, a.owner_consumer == caller.consumer) for a in addresses]
|
|
|
|
# =========================================================================
|
|
# send
|
|
# =========================================================================
|
|
|
|
async def send_message(
|
|
self, session: AsyncSession, data: MessageSendRequest, caller: Caller
|
|
) -> ScribeMessage:
|
|
"""`send` — `from` is the caller's own identity, never the body.
|
|
|
|
core-api T-15 / workspace D-70: `require_scribe_sender` (the
|
|
controller's dependency for this route) no longer enforces
|
|
`identity_required` itself, because that decision depends on the
|
|
body (`to`, `tags`), which a caller dependency cannot see. This
|
|
method is therefore the single place that enforces it: a missing
|
|
identity (`caller.identity == ""`, which only an office caller with
|
|
no `X-Office-Identity` header can produce) is allowed through
|
|
exactly for `_is_exempt_feedback`'s shape, and is still
|
|
`identity_required` for every other send — the tenant check stays
|
|
on this path too (no sender is not no tenant)."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
|
|
recipients = list(dict.fromkeys(data.to)) # de-dup, keep order
|
|
tags = list(dict.fromkeys(data.tags))
|
|
|
|
no_identity = caller.identity == ""
|
|
exempt = no_identity and self._is_exempt_feedback(caller, recipients, tags)
|
|
if no_identity and not exempt:
|
|
raise ScribeError("identity_required", 400)
|
|
|
|
if exempt:
|
|
from_identity = None
|
|
from_consumer = caller.consumer
|
|
else:
|
|
from_address = await self._get_address(session, tenant, caller.identity)
|
|
if from_address is None or from_address.owner_consumer != caller.consumer:
|
|
raise ScribeError("identity_not_owned", 403)
|
|
from_identity = caller.identity
|
|
from_consumer = caller.consumer
|
|
|
|
found_stmt = select(ScribeAddress.name).where(
|
|
ScribeAddress.owner_tenant == tenant, ScribeAddress.name.in_(recipients)
|
|
)
|
|
found = set((await session.execute(found_stmt)).scalars().all())
|
|
missing = [name for name in recipients if name not in found]
|
|
if missing:
|
|
raise ScribeError("unknown_address", 404, names=missing)
|
|
|
|
if data.reply_to is not None:
|
|
reply_target = await session.get(ScribeMessage, data.reply_to)
|
|
if reply_target is None or reply_target.owner_tenant != tenant:
|
|
raise ScribeError("not_found", 404)
|
|
|
|
message = ScribeMessage(
|
|
owner_tenant=tenant,
|
|
topics=recipients,
|
|
from_identity=from_identity,
|
|
from_consumer=from_consumer,
|
|
subject=data.subject,
|
|
body=data.body,
|
|
tags=tags,
|
|
reply_to=data.reply_to,
|
|
)
|
|
session.add(message)
|
|
await session.flush() # assign message.id
|
|
|
|
for recipient in recipients:
|
|
session.add(
|
|
ScribeMailboxEntry(address=recipient, owner_tenant=tenant, message_id=message.id)
|
|
)
|
|
|
|
await session.commit()
|
|
await session.refresh(message)
|
|
|
|
logger.info(
|
|
f"Scribe message sent: id={message.id} from={from_identity} "
|
|
f"from_consumer={from_consumer} to={recipients}"
|
|
)
|
|
return message
|
|
|
|
# =========================================================================
|
|
# mailbox / find
|
|
# =========================================================================
|
|
|
|
async def get_mailbox(
|
|
self,
|
|
session: AsyncSession,
|
|
caller: Caller,
|
|
view: str = "inbox",
|
|
unread: Optional[bool] = None,
|
|
tag: Optional[str] = None,
|
|
limit: int = 50,
|
|
before_id: Optional[int] = None,
|
|
message_id: Optional[int] = None,
|
|
) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]:
|
|
"""`mailbox` — the caller's identity's entries, unread first.
|
|
|
|
`message_id`, when given, is a direct lookup rather than a filtered
|
|
list: zero or one entry (the unique constraint on
|
|
(address, message_id) guarantees at most one), in any view — it
|
|
ignores `view`/`unread`/`tag`/`before_id` entirely, which is the
|
|
point (core-api T-9): a client that already knows the message
|
|
number never pages the unread-first/id-descending listing to find
|
|
it.
|
|
|
|
Otherwise `view` selects which of the independent `archived_at`/
|
|
`trashed_at` states to show: `inbox` (default) is neither,
|
|
`archive` is archived and not trashed, `trash` is trashed
|
|
(regardless of archived_at — trashing hides an entry from the
|
|
archive view too), `all` applies no filter on either column.
|
|
"""
|
|
tenant = await self._require_tenant(session, caller)
|
|
await self._require_owned_identity(session, caller, tenant)
|
|
|
|
if message_id is not None:
|
|
lookup_query = select(ScribeMailboxEntry).where(
|
|
ScribeMailboxEntry.owner_tenant == tenant,
|
|
ScribeMailboxEntry.address == caller.identity,
|
|
ScribeMailboxEntry.message_id == message_id,
|
|
)
|
|
entries = list((await session.execute(lookup_query)).scalars().all())
|
|
messages_by_id = await self._messages_for_entries(session, entries)
|
|
results = [(entry, messages_by_id[entry.message_id]) for entry in entries]
|
|
return results, len(results)
|
|
|
|
conditions = [
|
|
ScribeMailboxEntry.owner_tenant == tenant,
|
|
ScribeMailboxEntry.address == caller.identity,
|
|
]
|
|
if view == "archive":
|
|
conditions.append(ScribeMailboxEntry.archived_at.isnot(None))
|
|
conditions.append(ScribeMailboxEntry.trashed_at.is_(None))
|
|
elif view == "trash":
|
|
conditions.append(ScribeMailboxEntry.trashed_at.isnot(None))
|
|
elif view == "all":
|
|
pass
|
|
else: # "inbox" (default)
|
|
conditions.append(ScribeMailboxEntry.archived_at.is_(None))
|
|
conditions.append(ScribeMailboxEntry.trashed_at.is_(None))
|
|
if unread:
|
|
conditions.append(ScribeMailboxEntry.read_at.is_(None))
|
|
if tag is not None:
|
|
conditions.append(ScribeMailboxEntry.tags.contains([tag]))
|
|
if before_id is not None:
|
|
conditions.append(ScribeMailboxEntry.id < before_id)
|
|
|
|
base_query = select(ScribeMailboxEntry).where(*conditions)
|
|
count_query = select(func.count()).select_from(base_query.subquery())
|
|
total = (await session.execute(count_query)).scalar() or 0
|
|
|
|
query = (
|
|
base_query
|
|
.order_by(ScribeMailboxEntry.read_at.is_(None).desc(), ScribeMailboxEntry.id.desc())
|
|
.limit(limit)
|
|
)
|
|
entries = list((await session.execute(query)).scalars().all())
|
|
messages_by_id = await self._messages_for_entries(session, entries)
|
|
|
|
return [(entry, messages_by_id[entry.message_id]) for entry in entries], total
|
|
|
|
async def find_messages(
|
|
self, session: AsyncSession, caller: Caller, q: str, limit: int = 50
|
|
) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]:
|
|
"""`find` — full text over messages in the caller identity's
|
|
mailbox, every entry that is not trashed (core-api T-13: trash
|
|
takes an entry out of `find` the same way it takes it out of
|
|
`inbox` and `archive`); archived entries still match."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
await self._require_owned_identity(session, caller, tenant)
|
|
|
|
tsquery = func.websearch_to_tsquery("simple", q)
|
|
base_query = (
|
|
select(ScribeMailboxEntry)
|
|
.join(ScribeMessage, ScribeMailboxEntry.message_id == ScribeMessage.id)
|
|
.where(
|
|
ScribeMailboxEntry.owner_tenant == tenant,
|
|
ScribeMailboxEntry.address == caller.identity,
|
|
ScribeMailboxEntry.trashed_at.is_(None),
|
|
ScribeMessage.search.op("@@")(tsquery),
|
|
)
|
|
)
|
|
count_query = select(func.count()).select_from(base_query.subquery())
|
|
total = (await session.execute(count_query)).scalar() or 0
|
|
|
|
query = base_query.order_by(ScribeMailboxEntry.id.desc()).limit(limit)
|
|
entries = list((await session.execute(query)).scalars().all())
|
|
messages_by_id = await self._messages_for_entries(session, entries)
|
|
|
|
return [(entry, messages_by_id[entry.message_id]) for entry in entries], total
|
|
|
|
# =========================================================================
|
|
# history
|
|
# =========================================================================
|
|
|
|
async def get_history(
|
|
self,
|
|
session: AsyncSession,
|
|
name: str,
|
|
caller: Caller,
|
|
since_id: Optional[int] = None,
|
|
limit: int = 100,
|
|
) -> List[ScribeMessage]:
|
|
"""`history` — the retained messages sent to that address, oldest
|
|
first. Reading never consumes or marks: this reads scribe_messages
|
|
directly, never scribe_mailbox_entries."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
|
|
address = await self._get_address(session, tenant, name)
|
|
if address is None:
|
|
raise ScribeError("unknown_address", 404)
|
|
|
|
conditions = [
|
|
ScribeMessage.owner_tenant == tenant,
|
|
ScribeMessage.topics.contains([name]),
|
|
]
|
|
if since_id is not None:
|
|
conditions.append(ScribeMessage.id > since_id)
|
|
|
|
stmt = (
|
|
select(ScribeMessage)
|
|
.where(*conditions)
|
|
.order_by(ScribeMessage.id.asc())
|
|
.limit(limit)
|
|
)
|
|
return list((await session.execute(stmt)).scalars().all())
|
|
|
|
# =========================================================================
|
|
# mark / tag / untag
|
|
# =========================================================================
|
|
|
|
async def patch_mailbox_entry(
|
|
self, session: AsyncSession, entry_id: int, data: MailboxPatchRequest, caller: Caller
|
|
) -> Tuple[ScribeMailboxEntry, ScribeMessage]:
|
|
"""`mark` (`read`), `archive`/`unarchive`, `trash`/`restore` and
|
|
`tag`/`untag` (`add_tags`/`remove_tags`) share this route. An entry
|
|
outside the caller's tenant, or addressed to an identity the caller
|
|
is not reading as, answers 404 not_found — indistinguishable from a
|
|
nonexistent entry, so an id never confirms someone else's mail
|
|
exists. `archived`/`trashed` are independent timestamps (core-api
|
|
T-13): `true` sets the one not already set, `false` clears it —
|
|
trashing does not clear `archived_at`, so restoring a message that
|
|
was archived before it was trashed returns it to the archive view,
|
|
not the inbox."""
|
|
tenant = await self._require_tenant(session, caller)
|
|
|
|
entry = await session.get(ScribeMailboxEntry, entry_id)
|
|
if entry is None or entry.owner_tenant != tenant or entry.address != caller.identity:
|
|
raise ScribeError("not_found", 404)
|
|
|
|
if data.read is True and entry.read_at is None:
|
|
entry.read_at = datetime.now(timezone.utc)
|
|
elif data.read is False:
|
|
entry.read_at = None
|
|
|
|
if data.archived is True and entry.archived_at is None:
|
|
entry.archived_at = datetime.now(timezone.utc)
|
|
elif data.archived is False:
|
|
entry.archived_at = None
|
|
|
|
if data.trashed is True and entry.trashed_at is None:
|
|
entry.trashed_at = datetime.now(timezone.utc)
|
|
elif data.trashed is False:
|
|
entry.trashed_at = None
|
|
|
|
if data.add_tags:
|
|
current = set(entry.tags)
|
|
entry.tags = list(entry.tags) + [t for t in data.add_tags if t not in current]
|
|
if data.remove_tags:
|
|
remove = set(data.remove_tags)
|
|
entry.tags = [t for t in entry.tags if t not in remove]
|
|
|
|
await session.commit()
|
|
await session.refresh(entry)
|
|
|
|
message = await session.get(ScribeMessage, entry.message_id)
|
|
return entry, message
|
|
|
|
|
|
# Singleton instance
|
|
_scribe_service: Optional[ScribeService] = None
|
|
|
|
|
|
def get_scribe_service() -> ScribeService:
|
|
"""Get singleton Scribe service instance"""
|
|
global _scribe_service
|
|
if _scribe_service is None:
|
|
_scribe_service = ScribeService()
|
|
return _scribe_service
|