diff --git a/README.md b/README.md index df1e8abdc..1176949c1 100644 --- a/README.md +++ b/README.md @@ -373,7 +373,8 @@ outcome = await client.accounts.ensure(key, billing="operator") # AccountReady - **Storage.** One object; process-local storage is refused in production. Apply schema changes with `python -m adcp migrate`. - **Grants.** The SDK is the only source of `authorized_at`. End a relationship with `wind_down_grant()` instead of revoking. - **Explain.** Seller `reporting.explain_coverage(...)` and buyer `client.reporting.explain(...)` return the same `CoverageTrace`; `python -m adcp reporting diff` compares them. -- **Coming in later betas:** buyer events and the test kit. +- **Events.** `client.reporting.poll()` / `events()` derive events from saved state with deterministic ids; `handle_webhook()` treats a verified webhook as a doorbell. +- **Coming in later betas:** the test kit. Guides: [seller](docs/reporting-subscriptions-seller.md) · [buyer](docs/reporting-subscriptions-buyer.md) · [coverage](docs/reporting-subscriptions-coverage.md) · [accounts](docs/accounts-ensure-and-change.md) · [migrating from account-level configs](docs/reporting-subscriptions-migration.md) · [testing](docs/reporting-subscriptions-testing.md) · [errors](docs/reporting-subscriptions-errors.md) diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 886e1a260..588ae8543 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/docs/reporting-subscriptions-buyer.md @@ -93,7 +93,41 @@ traces = await client.reporting.explain(account_id="acc_a", subscription_id="ana ## Events -`client.reporting.poll()`, `events()` and `handle_webhook()` arrive in a later beta. Until then, pass each `principal.changed` payload to `tracker.handle_notification(...)`. +```python +# A long-running worker: each batch is acked once the loop body ran for all of it. +async for event in client.reporting.events(poll_every=timedelta(minutes=5)): + match event.type: + case "revision.available": await ingest(event.reporting_revision_id) + case "period.late": await alert_seller(event) + case "account.unauthorized": await pause_dashboards(event.account_id) + +# Serverless: one batch per invocation. +batch = await client.reporting.poll() +await handle(batch.events) +await batch.ack() + +# Webhook endpoint: a doorbell, never the event content. +outcome = await receiver.receive(...) # verifies the signature +await client.reporting.handle_webhook(VerifiedWebhook.from_outcome(outcome)) +``` + +- **Derived, not pushed.** Each poll runs `reconcile` and reads the subscription readback and coverage, then compares the result with the snapshot saved at the last `ack`. A crash before `ack` redelivers the same events with the same ids, so deliver at least once and deduplicate on `event.id`. +- **Deterministic ids.** `event.id` is `sha256(agent, type, natural key, version)`. +- **Durable state.** Pass `ReportingClient(client, event_store=...)` for a store that survives restarts. It implements `load(agent_id)` and a compare-and-set `save(...)`. The default `InMemoryReportingEventStore` is process-local. Two workers that ack the same snapshot get `ReportingEventConflictError` on the second ack. +- **Cold start.** `poll(baseline=True)` on a consumer with no snapshot records the present without emitting it. +- **Webhooks.** `handle_webhook` takes a `VerifiedWebhook` (or a verified, non-rejected `WebhookOutcome`) and refuses anything else. With an `events()` loop running it wakes the loop; otherwise it polls and returns the batch for you to handle and `ack`. +- **Partial reads.** An account that could not be read keeps its previous state, so it never produces a spurious "closed" or "gone" event. + +| Event | When | Key fields | +|---|---|---| +| `revision.available` | a revision appears for an owed period | `reporting_revision_id`, `reporting_obligation_id`, `period`, `finality` | +| `revision.adjusted` | a new revision replaces one you saw | `superseded_reporting_revision_id` | +| `period.late`, `period.missing_obligation`, `period.withheld` | a period reaches that status | the period line | +| `coverage.exception_opened`, `coverage.exception_closed` | an `excluded[]` entry opens or closes | `reason`, `from`, `until`, `media_buy_id` | +| `account.covered` | an account enters a subscription | `subscription_id`, `account_id` | +| `account.unauthorized`, `account.regranted` | authority is lost, or returns | `unauthorized_at`, `regranted_at`, `chain` | +| `subscription.state_changed` | `active`, `inactive` or `action_required` | `state`, `previous_state`, `pending_action` | +| `timezone.cutover` | a `timezone_cutovers[]` entry appears | `effective_at`, `timezone`, `previous_timezone` | ## Receipts for billing feeds diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 0dd9751db..7a405a0a8 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -81,6 +81,14 @@ decode_principal_state, encode_principal_state, ) +from adcp.reporting.subscriptions.events import ( + EventBatch, + InMemoryReportingEventStore, + ReportingEvent, + ReportingEventConflictError, + ReportingEventStore, + VerifiedWebhook, +) from adcp.reporting.subscriptions.explain import ( EXPLAIN_TEMPLATES, TRACE_VERSION, @@ -140,6 +148,12 @@ from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant __all__ = [ + "EventBatch", + "InMemoryReportingEventStore", + "ReportingEvent", + "ReportingEventConflictError", + "ReportingEventStore", + "VerifiedWebhook", "BuyerReportingState", "CoverageExplainer", "PeriodLine", diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py new file mode 100644 index 000000000..0421fa79b --- /dev/null +++ b/src/adcp/reporting/subscriptions/events.py @@ -0,0 +1,644 @@ +"""Buyer events for reporting subscriptions: ``poll``, ``events``, ``handle_webhook`` (AdCP 3.3). + +Events are derived from durable state, never taken from a webhook body: + +1. :meth:`ReportingClient.poll` runs ``reconcile`` and reads the subscription + readback and each account's ``subscription_coverage[]``. +2. It compares that snapshot with the one this consumer last acknowledged, + and emits one :class:`ReportingEvent` per difference. +3. ``await batch.ack()`` saves the new snapshot. Until then nothing moves: + a crash before the ack replays the same events, with the same ids, on the + next poll. Delivery is at least once; deduplicate on ``event.id``. + +Event ids are ``sha256(agent, type, natural key, version)``, so a replay +after a crash, or the same change seen by two workers, produces the same id. + +A seller's ``principal.changed`` webhook is a doorbell: pass the verified +delivery to :meth:`ReportingClient.handle_webhook`, which wakes a running +:meth:`ReportingClient.events` loop, or polls inline when none runs. + +Event types (every event has ``id``, ``type``, ``agent_id``, ``occurred_at``, +``observed_at`` and a ``data`` mapping): + +* ``revision.available`` and ``revision.adjusted`` (a new revision replaced + one this consumer saw); +* ``period.late``, ``period.missing_obligation``, ``period.withheld``; +* ``coverage.exception_opened`` and ``coverage.exception_closed``, with + ``reason``, ``from`` and ``until``; +* ``account.covered``, ``account.unauthorized``, ``account.regranted``; +* ``subscription.state_changed`` (``active``, ``inactive``, + ``action_required``); +* ``timezone.cutover``. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +from collections.abc import AsyncIterator, Mapping +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Protocol + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.subscriptions._clock import iso_z, utc_instant +from adcp.reporting.subscriptions.reconcile import ( + DEFAULT_GRACE, + BuyerReportingState, + ReconcileReport, + _reconcile, +) + +#: Version of the saved snapshot document; bump on a breaking change. +SNAPSHOT_VERSION = 1 + +EVENT_TYPES = frozenset( + { + "revision.available", + "revision.adjusted", + "period.late", + "period.missing_obligation", + "period.withheld", + "coverage.exception_opened", + "coverage.exception_closed", + "account.covered", + "account.unauthorized", + "account.regranted", + "subscription.state_changed", + "timezone.cutover", + } +) + + +@dataclass(frozen=True) +class ReportingEvent: + """One change this consumer has not acknowledged yet.""" + + id: str + type: str + agent_id: str + occurred_at: str + observed_at: str + data: Mapping[str, Any] = field(default_factory=dict) + + def __getattr__(self, name: str) -> Any: + # ``event.account_id``, ``event.reporting_revision_id`` and friends. + data = object.__getattribute__(self, "data") + if name in data: + return data[name] + raise AttributeError(name) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "type": self.type, + "agent_id": self.agent_id, + "occurred_at": self.occurred_at, + "observed_at": self.observed_at, + "data": dict(self.data), + } + + +class ReportingEventStore(Protocol): + """Where a consumer's acknowledged snapshot lives, per agent. + + ``save`` must refuse (raise) when the stored revision is no longer + ``expected_revision``, so two workers never both acknowledge. + """ + + async def load(self, agent_id: str) -> tuple[Mapping[str, Any], int] | None: ... + + async def save( + self, agent_id: str, snapshot: Mapping[str, Any], *, expected_revision: int | None + ) -> int: ... + + +class ReportingEventConflictError(RuntimeError): + """Another worker acknowledged first; poll again.""" + + code = "REPORTING_EVENTS_CONFLICT" + recovery = "transient" + + +class InMemoryReportingEventStore: + """Process-local snapshots: development and tests only.""" + + def __init__(self) -> None: + self._items: dict[str, tuple[dict[str, Any], int]] = {} + + async def load(self, agent_id: str) -> tuple[Mapping[str, Any], int] | None: + item = self._items.get(agent_id) + return None if item is None else (json.loads(json.dumps(item[0])), item[1]) + + async def save( + self, agent_id: str, snapshot: Mapping[str, Any], *, expected_revision: int | None + ) -> int: + current = self._items.get(agent_id) + if (current[1] if current else None) != expected_revision: + raise ReportingEventConflictError( + "Another worker acknowledged these events first. Poll again." + ) + revision = (current[1] if current else 0) + 1 + self._items[agent_id] = (json.loads(json.dumps(snapshot)), revision) + return revision + + +@dataclass +class EventBatch: + """One poll: the events since the last acknowledgement, and its ``ack``.""" + + events: list[ReportingEvent] + report: ReconcileReport + _store: ReportingEventStore + _agent_id: str + _snapshot: Mapping[str, Any] + _revision: int | None + acked: bool = False + + async def ack(self) -> None: + """Save this poll's snapshot, so its events are not delivered again.""" + if self.acked: + return + await self._store.save(self._agent_id, self._snapshot, expected_revision=self._revision) + self.acked = True + + +class VerifiedWebhook: + """A webhook delivery whose signature the SDK verified. + + Build it with :meth:`from_outcome` from a :class:`~adcp.webhook_receiver.WebhookReceiver` + outcome; :meth:`ReportingClient.handle_webhook` refuses anything else at + runtime, so an unverified body never reaches the poller. + """ + + __slots__ = ("payload", "sender_identity") + + def __init__(self, payload: Any, sender_identity: str, *, _token: object = None) -> None: + if _token is not _VERIFIED: + raise TypeError("use VerifiedWebhook.from_outcome(outcome)") + self.payload = payload + self.sender_identity = sender_identity + + @classmethod + def from_outcome(cls, outcome: Any) -> VerifiedWebhook: + """From a ``WebhookReceiver.receive`` outcome that verified and was not rejected.""" + if getattr(outcome, "rejected", True) or not getattr(outcome, "sender_identity", None): + raise ValueError("the webhook was not verified; it cannot wake the poller") + return cls(outcome.payload, outcome.sender_identity, _token=_VERIFIED) + + +_VERIFIED = object() + + +# -- snapshots ------------------------------------------------------------------------- + + +def _key(*parts: Any) -> str: + return json.dumps([str(part) if part is not None else None for part in parts]) + + +def _snapshot(report: ReconcileReport, context: Any) -> dict[str, Any]: + """The state events are derived from, as a JSON document.""" + periods: dict[str, dict[str, Any]] = {} + for line in report.periods: + if line.status == "not_owed": + continue + periods[ + _key( + line.account_id, + line.subscription_id, + line.subscription_version, + line.offering_id, + line.period["start"], + line.period["end"], + ) + ] = { + "status": line.status, + "reporting_revision_id": line.reporting_revision_id, + "reporting_obligation_id": line.reporting_obligation_id, + "finality": line.finality, + "expected_at": line.expected_at, + "period": dict(line.period), + "account_id": line.account_id, + "subscription_id": line.subscription_id, + "subscription_version": line.subscription_version, + "offering_id": line.offering_id, + "media_buy_ids": list(line.media_buy_ids), + } + subscriptions: dict[str, dict[str, Any]] = {} + unauthorized: dict[str, str] = {} + for state in context.subscriptions: + value = state.model_dump(mode="json", exclude_none=True) + issues = value.get("issues") or [] + subscriptions[str(state.subscription_id)] = { + "state": value.get("state"), + "subscription_version": value.get("subscription_version"), + "pending_action": issues[0] if issues else None, + } + for entry in value.get("accounts_unauthorized") or []: + unauthorized[entry["account_id"]] = entry["unauthorized_at"] + covered: list[str] = sorted( + {_key(line.subscription_id, line.account_id) for line in report.periods} + ) + exceptions: dict[str, dict[str, Any]] = {} + cutovers: dict[str, dict[str, Any]] = {} + authorizations: dict[str, dict[str, Any]] = {} + for account_id in report.accounts: + record = context.state.accounts.get(account_id) + if record is None: + continue + if record.listed: + authorizations[account_id] = { + "authorized_at": None + if record.authorized_at is None + else iso_z(record.authorized_at), + "regranted_at": None if record.regranted_at is None else iso_z(record.regranted_at), + } + for subscription_id, entry in record.coverage.items(): + covered.append(_key(subscription_id, account_id)) + for item in entry.get("excluded") or (): + exceptions[ + _key( + subscription_id, + account_id, + item.get("reason"), + item.get("media_buy_id"), + item.get("from"), + ",".join(item.get("package_ids") or ()), + ) + ] = { + "subscription_id": subscription_id, + "account_id": account_id, + "reason": item.get("reason"), + "media_buy_id": item.get("media_buy_id"), + "from": item.get("from"), + "until": item.get("until"), + "package_ids": list(item.get("package_ids") or ()), + } + for item in entry.get("timezone_cutovers") or (): + cutovers[ + _key( + subscription_id, + account_id, + item.get("offering_id"), + item.get("effective_at"), + ) + ] = { + "subscription_id": subscription_id, + "account_id": account_id, + "offering_id": item.get("offering_id"), + "effective_at": item.get("effective_at"), + "timezone": item.get("timezone"), + "previous_timezone": item.get("previous_timezone"), + } + return { + "snapshot_version": SNAPSHOT_VERSION, + "periods": periods, + "subscriptions": subscriptions, + "covered": sorted(set(covered)), + "unauthorized": unauthorized, + "authorizations": authorizations, + "exceptions": exceptions, + "cutovers": cutovers, + } + + +def _event_id(agent_id: str, type_: str, natural_key: str, version: Any) -> str: + digest = hashlib.sha256( + canonical_json_utf8_v1([agent_id, type_, natural_key, version]) + ).hexdigest() + return f"rse_{digest[:40]}" + + +def derive_events( + previous: Mapping[str, Any] | None, + current: Mapping[str, Any], + *, + agent_id: str, + observed_at: datetime, + grace: timedelta = DEFAULT_GRACE, +) -> list[ReportingEvent]: + """The events between two snapshots, in a deterministic order. + + With no ``previous`` snapshot every present fact is new: each delivered + revision, open exception, covered account and so on is emitted once. + """ + before: Mapping[str, Any] = previous or {} + observed = iso_z(observed_at) + events: list[ReportingEvent] = [] + + def emit( + type_: str, natural_key: str, version: Any, occurred: str | None, data: Mapping[str, Any] + ) -> None: + events.append( + ReportingEvent( + id=_event_id(agent_id, type_, natural_key, version), + type=type_, + agent_id=agent_id, + occurred_at=occurred or observed, + observed_at=observed, + data=dict(data), + ) + ) + + old_subscriptions = before.get("subscriptions") or {} + for subscription_id, value in sorted((current.get("subscriptions") or {}).items()): + old = old_subscriptions.get(subscription_id) + if old is None or old.get("state") != value.get("state"): + emit( + "subscription.state_changed", + _key(subscription_id), + [value.get("subscription_version"), value.get("state")], + None, + { + "subscription_id": subscription_id, + "state": value.get("state"), + "previous_state": None if old is None else old.get("state"), + "subscription_version": value.get("subscription_version"), + "pending_action": value.get("pending_action"), + }, + ) + + old_covered = set(before.get("covered") or ()) + # Each time an account (re)joins a subscription is a new occurrence, so a + # rejoin is never deduplicated against the first join. + generations = dict(before.get("covered_generations") or {}) + for item in current.get("covered") or (): + if item not in old_covered: + generations[item] = int(generations.get(item, 0)) + 1 + subscription_id, account_id = json.loads(item) + emit( + "account.covered", + item, + generations[item], + None, + {"subscription_id": subscription_id, "account_id": account_id}, + ) + if isinstance(current, dict): + current["covered_generations"] = generations + + old_unauthorized = before.get("unauthorized") or {} + new_unauthorized = current.get("unauthorized") or {} + for account_id, lost in sorted(new_unauthorized.items()): + if old_unauthorized.get(account_id) != lost: + emit( + "account.unauthorized", + _key(account_id), + lost, + lost, + {"account_id": account_id, "unauthorized_at": lost}, + ) + old_authorizations = before.get("authorizations") or {} + for account_id, lost in sorted(old_unauthorized.items()): + if account_id in new_unauthorized: + continue + granted = (current.get("authorizations") or {}).get(account_id) + if granted is None: + continue + regranted = granted.get("regranted_at") or granted.get("authorized_at") + emit( + "account.regranted", + _key(account_id), + [lost, regranted], + regranted, + { + "account_id": account_id, + "unauthorized_at": lost, + "regranted_at": regranted, + "authorized_at": granted.get("authorized_at"), + "chain": "continued" + if granted.get("authorized_at") + == (old_authorizations.get(account_id) or {}).get("authorized_at") + else "new", + }, + ) + + old_exceptions = before.get("exceptions") or {} + new_exceptions = current.get("exceptions") or {} + for key, value in sorted(new_exceptions.items()): + old = old_exceptions.get(key) + if value.get("until") is None and (old is None or old.get("until") is not None): + emit("coverage.exception_opened", key, value.get("from"), value.get("from"), value) + if value.get("until") is not None and (old is not None and old.get("until") is None): + emit("coverage.exception_closed", key, value.get("until"), value.get("until"), value) + for key, old in sorted(old_exceptions.items()): + if key not in new_exceptions and old.get("until") is None: + emit("coverage.exception_closed", key, "gone", None, {**old, "until": None}) + + old_cutovers = before.get("cutovers") or {} + for key, value in sorted((current.get("cutovers") or {}).items()): + if key not in old_cutovers: + emit("timezone.cutover", key, 1, value.get("effective_at"), value) + + old_periods = before.get("periods") or {} + for key, value in sorted((current.get("periods") or {}).items()): + old = old_periods.get(key) or {} + line = {k: value[k] for k in value if k not in {"status"}} + revision = value.get("reporting_revision_id") + old_revision = old.get("reporting_revision_id") + if revision is not None and revision != old_revision: + type_ = "revision.adjusted" if old_revision is not None else "revision.available" + emit( + type_, + key, + revision, + None, + {**line, "superseded_reporting_revision_id": old_revision}, + ) + status = value.get("status") + if status != old.get("status") and status in {"late", "missing_obligation", "withheld"}: + occurred = None + if status == "late" and value.get("expected_at"): + occurred = iso_z(_parse(value["expected_at"]) + grace) + elif status == "missing_obligation": + occurred = iso_z(_parse(value["period"]["end"]) + grace) + emit(f"period.{status}", key, status, occurred, line) + return events + + +def _parse(value: str) -> datetime: + return utc_instant(datetime.fromisoformat(value.replace("Z", "+00:00"))) + + +# -- the consumer ---------------------------------------------------------------------- + + +class ReportingEvents: + """``poll``, ``events`` and ``handle_webhook`` for one client (see the module doc).""" + + def __init__( + self, + client: Any, + state: BuyerReportingState, + *, + store: ReportingEventStore | None = None, + agent_id: str | None = None, + ) -> None: + self._client = client + self._state = state + self.store: ReportingEventStore = store or InMemoryReportingEventStore() + self._agent_id = agent_id + self._doorbell: asyncio.Event | None = None + self._listening = 0 + + @property + def agent_id(self) -> str: + if self._agent_id is not None: + return self._agent_id + config = getattr(self._client, "agent_config", None) + identifier = getattr(config, "id", None) or getattr(config, "agent_uri", None) + return str(identifier or "default") + + async def poll( + self, + *, + grace: timedelta = DEFAULT_GRACE, + baseline: bool = False, + now: datetime | None = None, + ) -> EventBatch: + """One batch of events since the last acknowledgement; call ``batch.ack()`` after handling. + + ``baseline=True`` on a consumer with no snapshot yet records the + current state without emitting it. + """ + instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) + report, context = await _reconcile( + self._client, + state=self._state, + since=None, + accounts="covered", + subscription_id=None, + grace=grace, + include_not_owed=False, + now=instant, + ) + current = _snapshot(report, context) + loaded = await self.store.load(self.agent_id) + previous, revision = (None, None) if loaded is None else loaded + if previous is not None and previous.get("snapshot_version") != SNAPSHOT_VERSION: + previous = None + if not report.definitive and previous is not None: + # Never treat an unread account as a change: keep the previous view of + # what this run could not read. + current = _merge_partial(previous, current, report) + if loaded is None and baseline: + events: list[ReportingEvent] = [] + else: + events = derive_events( + previous, current, agent_id=self.agent_id, observed_at=instant, grace=grace + ) + return EventBatch(events, report, self.store, self.agent_id, current, revision) + + async def events( + self, + *, + poll_every: timedelta = timedelta(minutes=5), + stop: asyncio.Event | None = None, + grace: timedelta = DEFAULT_GRACE, + ) -> AsyncIterator[ReportingEvent]: + """Every event, forever (or until ``stop`` is set). + + Each batch is acknowledged once the loop body has run for all of its + events, so an exception in the body redelivers the batch. + """ + if self._doorbell is None: + self._doorbell = asyncio.Event() + self._listening += 1 + try: + while stop is None or not stop.is_set(): + # Consume the ring before polling: one that lands during the + # poll, the loop body or the ack survives and polls again at once. + self._doorbell.clear() + batch = await self.poll(grace=grace) + for event in batch.events: + yield event + try: + await batch.ack() + except ReportingEventConflictError: + # Another worker acknowledged first: poll again at once. + # Events it already acknowledged are not redelivered. + continue + waiters = [asyncio.ensure_future(self._doorbell.wait())] + if stop is not None: + waiters.append(asyncio.ensure_future(stop.wait())) + done, pending = await asyncio.wait( + waiters, + timeout=poll_every.total_seconds(), + return_when=asyncio.FIRST_COMPLETED, + ) + for item in pending: + item.cancel() + finally: + self._listening -= 1 + + async def handle_webhook(self, verified: Any) -> EventBatch | None: + """A verified seller webhook: wake the ``events()`` loop, or poll inline. + + ``verified`` is a :class:`VerifiedWebhook` or a + :class:`~adcp.webhook_receiver.WebhookOutcome` that verified and was not + rejected; anything else is refused. The body is never trusted as event + content: it only says "something changed". With a running ``events()`` + loop this returns ``None``; otherwise it returns the polled batch, for + the caller to handle and ``ack``. + """ + if not isinstance(verified, VerifiedWebhook): + from adcp.webhook_receiver import WebhookOutcome + + if not isinstance(verified, WebhookOutcome): + raise TypeError( + "handle_webhook takes a verified delivery: VerifiedWebhook.from_outcome(" + "await receiver.receive(...)). Verify the signature first." + ) + verified = VerifiedWebhook.from_outcome(verified) + if self._listening and self._doorbell is not None: + self._doorbell.set() + return None + return await self.poll() + + +def _merge_partial( + previous: Mapping[str, Any], current: dict[str, Any], report: ReconcileReport +) -> dict[str, Any]: + unread = {issue.account_id for issue in report.issues if issue.account_id} + everything = not report.accounts or any(issue.account_id is None for issue in report.issues) + merged = dict(current) + for section in ("periods", "exceptions", "cutovers"): + values = dict(current.get(section) or {}) + for key, value in (previous.get(section) or {}).items(): + if everything or value.get("account_id") in unread: + values.setdefault(key, value) + merged[section] = values + if everything: + # A run that could not read the principal or the roster keeps the + # previous subscriptions and losses, so nothing reads as changed. + merged["subscriptions"] = { + **dict(previous.get("subscriptions") or {}), + **dict(current.get("subscriptions") or {}), + } + merged["unauthorized"] = { + **dict(previous.get("unauthorized") or {}), + **dict(current.get("unauthorized") or {}), + } + merged["covered"] = sorted( + set(current.get("covered") or ()) | set(previous.get("covered") or ()) + ) + merged["authorizations"] = { + **dict(previous.get("authorizations") or {}), + **dict(current.get("authorizations") or {}), + } + return merged + + +__all__ = [ + "EVENT_TYPES", + "SNAPSHOT_VERSION", + "EventBatch", + "InMemoryReportingEventStore", + "ReportingEvent", + "ReportingEventConflictError", + "ReportingEventStore", + "ReportingEvents", + "VerifiedWebhook", + "derive_events", +] diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 76714f24c..616f58c07 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -35,6 +35,14 @@ from adcp.types import GetPrincipalRequest, SyncPrincipalRequest if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from adcp.reporting.subscriptions.events import ( + EventBatch, + ReportingEvent, + ReportingEvents, + ReportingEventStore, + ) from adcp.reporting.subscriptions.explain import CoverageTrace from adcp.reporting.subscriptions.reconcile import ReconcileReport @@ -267,6 +275,7 @@ def __init__( *, max_attempts: int = _DEFAULT_ATTEMPTS, idempotency_key: Callable[[], str] | None = None, + event_store: ReportingEventStore | None = None, ) -> None: if type(max_attempts) is not int or not 1 <= max_attempts <= 10: raise ValueError("max_attempts must be an integer in 1..10") @@ -276,6 +285,8 @@ def __init__( from adcp.reporting.subscriptions.reconcile import BuyerReportingState self._state = BuyerReportingState() + self._event_store = event_store + self._events: ReportingEvents | None = None # -- public API --------------------------------------------------------------- @@ -473,6 +484,54 @@ async def explain( now=now, ) + def _event_source(self) -> ReportingEvents: + if self._events is None: + from adcp.reporting.subscriptions.events import ReportingEvents + + self._events = ReportingEvents(self._client, self._state, store=self._event_store) + return self._events + + async def poll( + self, + *, + grace: timedelta = DEFAULT_GRACE, + baseline: bool = False, + now: datetime | None = None, + ) -> EventBatch: + """One batch of reporting events since the last ``ack``; serverless-safe. + + ``batch = await client.reporting.poll(); await handle(batch.events); + await batch.ack()``. Events are derived by comparing this run's + ``reconcile`` and coverage with the snapshot saved at the last ack, so + a crash before ``ack`` redelivers the same events with the same ids. + Pass ``event_store=`` to :class:`ReportingClient` for a durable store. + """ + return await self._event_source().poll(grace=grace, baseline=baseline, now=now) + + def events( + self, + *, + poll_every: timedelta = timedelta(minutes=5), + stop: Any = None, + grace: timedelta = DEFAULT_GRACE, + ) -> AsyncIterator[ReportingEvent]: + """``async for event in client.reporting.events(): ...``, acked per batch. + + Polls every ``poll_every``, or at once when :meth:`handle_webhook` + rings. ``stop`` is an ``asyncio.Event`` that ends the loop. + """ + return self._event_source().events(poll_every=poll_every, stop=stop, grace=grace) + + async def handle_webhook(self, verified: Any) -> EventBatch | None: + """Treat a verified seller webhook as a doorbell. + + Pass ``VerifiedWebhook.from_outcome(await receiver.receive(...))``. + With a running :meth:`events` loop it wakes the loop and returns + ``None``; otherwise it polls now and returns the batch, which the + caller handles and acks. The body is never trusted as event content. + """ + return await self._event_source().handle_webhook(verified) + # -- planning ------------------------------------------------------------------- async def _plan_subscribe( diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index 1e3585261..8c7d2b82e 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -754,6 +754,30 @@ async def reconcile_subscriptions( now: datetime | None = None, ) -> ReconcileReport: """What every covered account is owed, and its status. See :meth:`ReportingClient.reconcile`.""" + report, _context = await _reconcile( + client, + state=state, + since=since, + accounts=accounts, + subscription_id=subscription_id, + grace=grace, + include_not_owed=include_not_owed, + now=now, + ) + return report + + +async def _reconcile( + client: Any, + *, + state: BuyerReportingState | None, + since: Any, + accounts: Sequence[str] | Literal["covered"], + subscription_id: str | None, + grace: timedelta, + include_not_owed: bool, + now: datetime | None, +) -> tuple[ReconcileReport, _Context]: instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) if not isinstance(grace, timedelta) or grace < timedelta(0): raise ValueError("grace must be a non-negative timedelta") @@ -835,7 +859,7 @@ async def reconcile_subscriptions( counts[line.status] += 1 if line.status != "not_owed": counts["owed"] += 1 - return ReconcileReport( + report = ReconcileReport( as_of=iso_z(instant), definitive=complete and not context.issues, summary=ReconcileSummary(**counts), @@ -843,6 +867,7 @@ async def reconcile_subscriptions( issues=tuple(context.issues), accounts=tuple(record.account_id for record in records), ) + return report, context def _not_owed_line(trace: CoverageTrace) -> PeriodLine: diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py new file mode 100644 index 000000000..e6d117977 --- /dev/null +++ b/tests/test_reporting_subscription_events.py @@ -0,0 +1,239 @@ +"""Buyer reporting events: ``poll``, ``events`` and ``handle_webhook`` (AdCP 3.3). + +Events come from comparing durable snapshots of what the buyer reads, so a +replay before ``ack`` repeats the same ids, and a webhook is only a doorbell. +""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta + +import pytest + +from adcp.reporting.subscriptions import ( + ReportingEventConflictError, + VerifiedWebhook, +) +from adcp.reporting.subscriptions.facade import ReportingClient +from adcp.webhook_receiver import WebhookOutcome +from tests.reporting_inprocess import InProcessBuyer +from tests.test_reporting_subscription_managed import ACCOUNT, START, _ctx, _World + +pytestmark = pytest.mark.asyncio + + +def _types(batch) -> list[str]: + return sorted(event.type for event in batch.events) + + +async def test_the_first_poll_reports_the_present_and_ack_moves_on() -> None: + world = _World() + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + batch = await reporting.poll(now=world.clock()) + assert _types(batch) == [ + "account.covered", + "revision.available", + "revision.available", + "subscription.state_changed", + ] + revision = next(e for e in batch.events if e.type == "revision.available") + assert revision.account_id == ACCOUNT and revision.reporting_revision_id + assert revision.id.startswith("rse_") and revision.agent_id + # Not acked: the same events, with the same ids, come again. + again = await reporting.poll(now=world.clock()) + assert [e.id for e in again.events] == [e.id for e in batch.events] + await again.ack() + after = await reporting.poll(now=world.clock()) + assert after.events == [] + # The stale batch can no longer be acknowledged. + with pytest.raises(ReportingEventConflictError): + await batch.ack() + + +async def test_missing_obligations_then_revisions_are_events() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + world.clock.set(START + timedelta(minutes=30)) + await world.service.run_worker() + first = await reporting.poll(baseline=True, now=world.clock()) + assert first.events == [] and first.report.definitive, first.report.issues + await first.ack() + late = await reporting.poll(now=START + timedelta(hours=3, minutes=30)) + assert _types(late) == ["period.missing_obligation"] + assert late.events[0].occurred_at == "2026-11-01T03:00:00Z" + await late.ack() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + delivered = await reporting.poll(now=world.clock()) + assert _types(delivered) == ["revision.available", "revision.available"] + + +async def test_losing_and_regaining_authority_are_events() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + world.clock.set(START + timedelta(minutes=30)) + await (await reporting.poll(baseline=True, now=world.clock())).ack() + account = world.seller.accounts.pop(ACCOUNT) + await world.service.touch(account_id=ACCOUNT) + await world.service.drain_subscription_triggers() + lost = await reporting.poll(now=world.clock()) + assert _types(lost) == ["account.unauthorized"] + await lost.ack() + world.clock.set(START + timedelta(hours=1)) + world.seller.accounts[ACCOUNT] = account + await world.service.touch(account_id=ACCOUNT) + await world.service.drain_subscription_triggers() + back = await reporting.poll(now=world.clock()) + assert "account.regranted" in _types(back) + + +async def test_handle_webhook_needs_a_verified_delivery_and_polls_inline() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + with pytest.raises(TypeError, match="verified"): + await reporting.handle_webhook({"notification_type": "principal.changed"}) + with pytest.raises(ValueError, match="not verified"): + await reporting.handle_webhook(WebhookOutcome(rejected=True)) + with pytest.raises(TypeError): + VerifiedWebhook({}, "seller") + outcome = WebhookOutcome(sender_identity="https://seller.example|kid", payload=None) + batch = await reporting.handle_webhook(outcome) + assert batch is not None and "subscription.state_changed" in _types(batch) + + +async def test_the_events_loop_acks_each_batch_and_wakes_on_the_doorbell() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + stop = asyncio.Event() + seen: list[str] = [] + + async def consume() -> None: + async for event in reporting.events(poll_every=timedelta(hours=1), stop=stop): + seen.append(event.type) + + task = asyncio.create_task(consume()) + for _ in range(200): + if seen: + break + await asyncio.sleep(0.05) + assert "subscription.state_changed" in seen + # A verified webhook wakes the running loop instead of polling inline. + ringing = await reporting.handle_webhook( + VerifiedWebhook.from_outcome(WebhookOutcome(sender_identity="seller", payload=None)) + ) + assert ringing is None + stop.set() + await asyncio.wait_for(task, timeout=30) + # The loop acknowledged what it delivered: only clock-driven period events remain. + remaining = await reporting.poll() + assert "subscription.state_changed" not in _types(remaining) + + +async def test_a_failed_read_never_fabricates_events() -> None: + world = _World() + await world.subscribe() + client = InProcessBuyer(world.platform, _ctx) + reporting = ReportingClient(client) + world.clock.set(START + timedelta(minutes=30)) + await world.service.run_worker() + account = world.seller.accounts.pop(ACCOUNT) + await world.service.touch(account_id=ACCOUNT) + await world.service.drain_subscription_triggers() + await (await reporting.poll(now=world.clock())).ack() + for task in ("get_principal", "list_accounts"): + client.fail.add(task) + batch = await reporting.poll(now=world.clock()) + assert batch.events == [], (task, [e.type for e in batch.events]) + await batch.ack() + client.fail.discard(task) + world.seller.accounts[ACCOUNT] = account + + +async def test_a_ring_during_a_cycle_is_not_swallowed() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + source = reporting._event_source() + polls = 0 + original = source.poll + + async def counting(**kwargs): + nonlocal polls + polls += 1 + batch = await original(**kwargs) + if polls == 1: + # A webhook rings while the first cycle is still running. + source._doorbell.set() + return batch + + source.poll = counting + stop = asyncio.Event() + + async def consume() -> None: + async for _event in reporting.events(poll_every=timedelta(hours=1), stop=stop): + pass + + task = asyncio.create_task(consume()) + for _ in range(200): + if polls >= 2: + break + await asyncio.sleep(0.05) + stop.set() + await asyncio.wait_for(task, timeout=30) + assert polls >= 2 + + +async def test_the_events_loop_survives_an_acknowledgement_conflict() -> None: + world = _World() + await world.subscribe() + reporting = ReportingClient(InProcessBuyer(world.platform, _ctx)) + other = ReportingClient(InProcessBuyer(world.platform, _ctx), event_store=None) + source = reporting._event_source() + other._event_source().store = source.store # two workers, one store + stop = asyncio.Event() + seen: list[str] = [] + raced = False + + async def consume() -> None: + nonlocal raced + async for event in reporting.events(poll_every=timedelta(hours=1), stop=stop): + seen.append(event.type) + if not raced: + raced = True + # The other worker acknowledges the same snapshot first. + await (await other.poll()).ack() + + task = asyncio.create_task(consume()) + for _ in range(200): + if raced and task.done() is False and len(seen) >= 1: + break + await asyncio.sleep(0.05) + await asyncio.sleep(0.5) + assert not task.done(), task.exception() if task.done() else None + stop.set() + await asyncio.wait_for(task, timeout=30) + + +def test_a_rejoin_is_a_new_account_covered_event() -> None: + from datetime import datetime, timezone + + from adcp.reporting.subscriptions.events import derive_events + + now = datetime(2026, 11, 1, tzinfo=timezone.utc) + key = '["analytics", "acc_a"]' + joined = {"covered": [key]} + first = derive_events({}, joined, agent_id="a", observed_at=now) + left = {"covered": [], "covered_generations": joined["covered_generations"]} + assert derive_events(joined, left, agent_id="a", observed_at=now) == [] + rejoined = {"covered": [key]} + again = derive_events(left, rejoined, agent_id="a", observed_at=now) + assert [e.type for e in first] == [e.type for e in again] == ["account.covered"] + assert first[0].id != again[0].id