state-hub/api/services/message_receipts.py

"""Per-recipient broadcast receipts and standing notices (STATE-WP-0093).

Broadcast visibility is per reader: a ``message_receipts`` row per
(message, agent) records delivery, read and acknowledgement. Direct messages
never get receipts and keep the global ``AgentMessage.read_at``.

- ``news`` broadcasts stay in a reader's unread inbox until that reader has
  any receipt (delivery at orientation is enough).
- ``standing`` broadcasts stay until the reader acknowledges, the notice
  expires, or it is superseded (which archives it).
"""
from __future__ import annotations

import uuid
from collections.abc import Iterable, Sequence
from datetime import datetime, timedelta, timezone

from sqlalchemy import and_, exists, func, not_, or_, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from api.models.agent_message import BROADCAST, AgentMessage, MessageReceipt
from api.models.managed_repo import ManagedRepo
from api.schemas.agent_message import NoticeStatus, StandingNoticeDigest

NEWS_DEFAULT_TTL = timedelta(days=30)


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def is_broadcast(message: AgentMessage) -> bool:
    return message.to_agent == BROADCAST


def not_expired(now: datetime):
    return or_(AgentMessage.expires_at.is_(None), AgentMessage.expires_at > now)


def broadcast_unread_clause(reader_values: Sequence[str], now: datetime):
    """SQL clause: broadcast is still unread for a reader (any of its slugs)."""
    has_any_receipt = exists().where(
        and_(
            MessageReceipt.message_id == AgentMessage.id,
            MessageReceipt.agent.in_(reader_values),
        )
    )
    has_ack = exists().where(
        and_(
            MessageReceipt.message_id == AgentMessage.id,
            MessageReceipt.agent.in_(reader_values),
            MessageReceipt.acknowledged_at.is_not(None),
        )
    )
    return and_(
        AgentMessage.to_agent == BROADCAST,
        not_expired(now),
        or_(
            and_(AgentMessage.kind == "standing", not_(has_ack)),
            and_(AgentMessage.kind != "standing", not_(has_any_receipt)),
        ),
    )


async def upsert_receipt(
    session: AsyncSession,
    message_ids: Iterable[uuid.UUID],
    agent: str,
    *,
    delivered: bool = False,
    read: bool = False,
    acknowledged: bool = False,
    at: datetime | None = None,
) -> None:
    """Idempotently stamp receipt fields; the first timestamp wins."""
    ids = list(message_ids)
    if not ids or not agent or agent == BROADCAST:
        return
    at = at or utcnow()
    rows = [
        {
            "message_id": mid,
            "agent": agent,
            "delivered_at": at if delivered else None,
            "read_at": at if (read or acknowledged) else None,
            "acknowledged_at": at if acknowledged else None,
        }
        for mid in ids
    ]
    stmt = pg_insert(MessageReceipt).values(rows)
    table = MessageReceipt.__table__
    stmt = stmt.on_conflict_do_update(
        index_elements=[table.c.message_id, table.c.agent],
        set_={
            col: func.coalesce(table.c[col], stmt.excluded[col])
            for col in ("delivered_at", "read_at", "acknowledged_at")
        },
    )
    await session.execute(stmt)


async def receipts_for(
    session: AsyncSession,
    message_ids: Iterable[uuid.UUID],
    reader_values: Sequence[str],
) -> dict[uuid.UUID, dict[str, datetime | None]]:
    """Merged receipt state per message across a reader's historical slugs."""
    ids = list(message_ids)
    if not ids:
        return {}
    result = await session.execute(
        select(MessageReceipt).where(
            MessageReceipt.message_id.in_(ids),
            MessageReceipt.agent.in_(reader_values),
        )
    )
    merged: dict[uuid.UUID, dict[str, datetime | None]] = {}
    for receipt in result.scalars().all():
        state = merged.setdefault(
            receipt.message_id,
            {"delivered_at": None, "read_at": None, "acknowledged_at": None},
        )
        for field in state:
            value = getattr(receipt, field)
            if value is not None and (state[field] is None or value < state[field]):
                state[field] = value
    return merged


async def _active_repo_slugs(session: AsyncSession) -> list[str]:
    result = await session.execute(
        select(ManagedRepo.slug).where(ManagedRepo.status == "active").order_by(ManagedRepo.slug)
    )
    return list(result.scalars().all())


async def live_standing_notices(
    session: AsyncSession, now: datetime | None = None
) -> list[AgentMessage]:
    now = now or utcnow()
    result = await session.execute(
        select(AgentMessage)
        .where(
            AgentMessage.to_agent == BROADCAST,
            AgentMessage.kind == "standing",
            AgentMessage.archived_at.is_(None),
            not_expired(now),
        )
        .order_by(AgentMessage.created_at.desc())
    )
    return list(result.scalars().all())


async def notice_statuses(session: AsyncSession) -> list[NoticeStatus]:
    notices = await live_standing_notices(session)
    if not notices:
        return []
    repos = await _active_repo_slugs(session)
    result = await session.execute(
        select(MessageReceipt).where(
            MessageReceipt.message_id.in_([n.id for n in notices])
        )
    )
    by_notice: dict[uuid.UUID, list[MessageReceipt]] = {}
    for receipt in result.scalars().all():
        by_notice.setdefault(receipt.message_id, []).append(receipt)
    statuses: list[NoticeStatus] = []
    for notice in notices:
        receipts = by_notice.get(notice.id, [])
        acked = sorted({r.agent for r in receipts if r.acknowledged_at is not None})
        seen = {r.agent for r in receipts}
        delivered_only = sorted(seen - set(acked))
        unreached = [slug for slug in repos if slug not in seen]
        statuses.append(
            NoticeStatus(
                id=notice.id,
                from_agent=notice.from_agent,
                subject=notice.subject,
                body=notice.body,
                created_at=notice.created_at,
                expires_at=notice.expires_at,
                supersedes_id=notice.supersedes_id,
                acknowledged=acked,
                delivered_only=delivered_only,
                unreached=unreached,
                acked_count=len(acked),
                delivered_only_count=len(delivered_only),
                unreached_count=len(unreached),
            )
        )
    return statuses


async def standing_notice_digests(session: AsyncSession) -> list[StandingNoticeDigest]:
    statuses = await notice_statuses(session)
    if not statuses:
        return []
    repos = set(await _active_repo_slugs(session))
    digests = []
    for status in statuses:
        acked_repos = len(repos.intersection(status.acknowledged))
        digests.append(
            StandingNoticeDigest(
                id=status.id,
                subject=status.subject,
                expires_at=status.expires_at,
                acked=acked_repos,
                pending=len(repos) - acked_repos,
            )
        )
    return digests