feat(reporting-storage): add PostgreSQL stores for reporting subscriptions

packages/adcp-reporting-storage/pyproject.toml

@@ -84,6 +84,7 @@ adcp_reporting_storage = [
     "projection/*.json",
     "production/*.json",
     "production/*.sql",
+    "subscriptions/*.sql",
 ]
 
 # Inside this repository the dependency is the checkout, not PyPI. uv ignores

packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py

@@ -205,6 +205,12 @@
     from adcp_reporting_storage.submissions.pg import (
         PgReportingSubmissionIntentStore as PgReportingSubmissionIntentStore,
     )
+    from adcp_reporting_storage.subscriptions.pg import (
+        PgPrincipalRecordStore as PgPrincipalRecordStore,
+    )
+    from adcp_reporting_storage.subscriptions.pg import (
+        PgReportingSubscriptionStateStore as PgReportingSubscriptionStateStore,
+    )
 
 #: Public name -> (module under this package, attribute). Resolved on first use.
 _LAZY_EXPORTS: dict[str, tuple[str, str]] = {
@@ -214,6 +220,11 @@
     "PgReportingMaterializerStore": ("materializer.pg", "PgReportingMaterializerStore"),
     "PgReportingReceiptStore": ("receipts.pg", "PgReportingReceiptStore"),
     "PgReportingFeedStore": ("feed.pg", "PgReportingFeedStore"),
+    "PgPrincipalRecordStore": ("subscriptions.pg", "PgPrincipalRecordStore"),
+    "PgReportingSubscriptionStateStore": (
+        "subscriptions.pg",
+        "PgReportingSubscriptionStateStore",
+    ),
     "PgReportingProjectionStore": ("projection.pg", "PgReportingProjectionStore"),
     "PgReportingStatusProjection": ("projection.pg", "PgReportingStatusProjection"),
     "PgReportingProjectionOutbox": ("projection.notifications", "PgReportingProjectionOutbox"),
@@ -320,6 +331,7 @@
         "receipts",
         "service",
         "submissions",
+        "subscriptions",
     }
 )
 

packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py

@@ -0,0 +1,15 @@
+"""PostgreSQL stores for AdCP 3.3 reporting subscriptions."""
+
+from __future__ import annotations
+
+from adcp_reporting_storage.subscriptions.pg import (
+    PgPrincipalRecordStore,
+    PgReportingSubscriptionStateStore,
+    reporting_subscriptions_schema_sql,
+)
+
+__all__ = [
+    "PgPrincipalRecordStore",
+    "PgReportingSubscriptionStateStore",
+    "reporting_subscriptions_schema_sql",
+]

packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py

@@ -0,0 +1,172 @@
+"""PostgreSQL stores for AdCP 3.3 reporting subscriptions (experimental).
+
+:class:`PgReportingSubscriptionStateStore` implements
+``adcp.reporting.subscriptions.ReportingSubscriptionStateStore`` and
+:class:`PgPrincipalRecordStore` implements
+``adcp.server.PrincipalRecordStore``. Together they make a seller's
+``sync_principal`` subscriptions and the engine's grants, exceptions and
+version snapshots survive restarts and agree across workers::
+
+    from psycopg_pool import AsyncConnectionPool
+    from adcp.reporting.subscriptions import ReportingSubscriptionEngine
+    from adcp.server import PrincipalService
+    from adcp_reporting_storage import (
+        PgPrincipalRecordStore,
+        PgReportingSubscriptionStateStore,
+    )
+
+    async with AsyncConnectionPool(dsn, min_size=2, max_size=10) as pool:
+        records = PgPrincipalRecordStore(pool=pool)
+        state = PgReportingSubscriptionStateStore(pool=pool)
+        await state.create_schema()  # creates both tables
+        principals = PrincipalService(store=records, emit_change=...)
+        engine = ReportingSubscriptionEngine(
+            principals=principals, host=..., offerings=..., store=state
+        )
+
+Each store keeps one JSONB document per subject and replaces it with a
+single conditional statement on its revision, so the compare-and-swap is
+atomic across every worker sharing the database. The pool is borrowed: the
+stores never open, own or close it. Requires ``adcp`` 10.0.0b3 or later.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from importlib.resources import files
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from psycopg_pool import AsyncConnectionPool
+
+    from adcp.reporting.subscriptions import StoredReportingSubscriptionState
+    from adcp.server import PrincipalRecord
+
+__all__ = [
+    "PgPrincipalRecordStore",
+    "PgReportingSubscriptionStateStore",
+    "reporting_subscriptions_schema_sql",
+]
+
+_REQUIRES = "adcp_reporting_storage.subscriptions requires adcp 10.0.0b3 or later"
+
+
+def reporting_subscriptions_schema_sql() -> str:
+    """The idempotent DDL for both reporting-subscription tables."""
+    return (
+        files("adcp_reporting_storage.subscriptions")
+        .joinpath("reporting_subscriptions.sql")
+        .read_text()
+    )
+
+
+def _document(value: Any) -> dict[str, Any]:
+    # psycopg decodes JSONB to Python by default; a custom loader may not.
+    decoded = json.loads(value) if isinstance(value, (str, bytes, bytearray)) else value
+    if not isinstance(decoded, dict):
+        raise ValueError("stored reporting-subscription document is not a JSON object")
+    return decoded
+
+
+class _PgDocumentStore:
+    def __init__(self, *, pool: AsyncConnectionPool) -> None:
+        self._pool = pool
+
+    async def create_schema(self) -> None:
+        """Create both reporting-subscription tables if they do not exist."""
+        async with self._pool.connection() as connection:
+            await connection.execute(reporting_subscriptions_schema_sql())
+
+
+class PgReportingSubscriptionStateStore(_PgDocumentStore):
+    """Engine state for reporting subscriptions, one CAS document per principal."""
+
+    async def load(self, subject: str) -> StoredReportingSubscriptionState | None:
+        try:
+            from adcp.reporting.subscriptions import StoredReportingSubscriptionState
+        except ImportError as error:  # pragma: no cover - older adcp
+            raise ImportError(_REQUIRES) from error
+        async with self._pool.connection() as connection:
+            cursor = await connection.execute(
+                "SELECT document, revision FROM adcp_reporting_subscription_state"
+                " WHERE subject = %s",
+                (subject,),
+            )
+            row = await cursor.fetchone()
+        if row is None:
+            return None
+        return StoredReportingSubscriptionState(_document(row[0]), int(row[1]))
+
+    async def compare_and_swap(
+        self,
+        subject: str,
+        expected_revision: int | None,
+        document: Mapping[str, Any],
+    ) -> bool:
+        payload = json.dumps(document, separators=(",", ":"), sort_keys=True)
+        async with self._pool.connection() as connection:
+            if expected_revision is None:
+                cursor = await connection.execute(
+                    "INSERT INTO adcp_reporting_subscription_state (subject, revision, document)"
+                    " VALUES (%s, 1, %s::jsonb) ON CONFLICT (subject) DO NOTHING",
+                    (subject, payload),
+                )
+            else:
+                cursor = await connection.execute(
+                    "UPDATE adcp_reporting_subscription_state"
+                    " SET revision = revision + 1, document = %s::jsonb, updated_at = now()"
+                    " WHERE subject = %s AND revision = %s",
+                    (payload, subject, expected_revision),
+                )
+            return bool(cursor.rowcount == 1)
+
+
+class PgPrincipalRecordStore(_PgDocumentStore):
+    """``sync_principal`` records, including reporting subscriptions and their versions."""
+
+    async def get(self, subject: str) -> PrincipalRecord | None:
+        try:
+            from adcp.server import principal_record_from_document
+        except ImportError as error:  # pragma: no cover - older adcp
+            raise ImportError(_REQUIRES) from error
+        async with self._pool.connection() as connection:
+            cursor = await connection.execute(
+                "SELECT document, store_revision FROM adcp_principal_records WHERE subject = %s",
+                (subject,),
+            )
+            row = await cursor.fetchone()
+        if row is None:
+            return None
+        return principal_record_from_document(_document(row[0]), store_revision=int(row[1]))
+
+    async def compare_and_swap(
+        self,
+        subject: str,
+        expected_store_revision: int | None,
+        record: PrincipalRecord,
+    ) -> bool:
+        try:
+            from adcp.server import principal_record_to_document
+        except ImportError as error:  # pragma: no cover - older adcp
+            raise ImportError(_REQUIRES) from error
+        if record.subject != subject:
+            raise ValueError("a principal record is stored under its own subject")
+        payload = json.dumps(
+            principal_record_to_document(record), separators=(",", ":"), sort_keys=True
+        )
+        async with self._pool.connection() as connection:
+            if expected_store_revision is None:
+                cursor = await connection.execute(
+                    "INSERT INTO adcp_principal_records (subject, store_revision, document)"
+                    " VALUES (%s, %s, %s::jsonb) ON CONFLICT (subject) DO NOTHING",
+                    (subject, record.store_revision, payload),
+                )
+            else:
+                cursor = await connection.execute(
+                    "UPDATE adcp_principal_records"
+                    " SET store_revision = %s, document = %s::jsonb, updated_at = now()"
+                    " WHERE subject = %s AND store_revision = %s",
+                    (record.store_revision, payload, subject, expected_store_revision),
+                )
+            return bool(cursor.rowcount == 1)

packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql

@@ -0,0 +1,18 @@
+-- AdCP 3.3 reporting subscriptions: durable principal records and engine state.
+-- Each row is one JSON document per subject, replaced by compare-and-swap on
+-- its revision. Rows are never pruned here: the documents keep every
+-- subscription version for as long as the obligations that cite it.
+
+CREATE TABLE IF NOT EXISTS adcp_principal_records (
+    subject TEXT PRIMARY KEY,
+    store_revision BIGINT NOT NULL CHECK (store_revision >= 1),
+    document JSONB NOT NULL CHECK (jsonb_typeof(document) = 'object'),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS adcp_reporting_subscription_state (
+    subject TEXT PRIMARY KEY,
+    revision BIGINT NOT NULL CHECK (revision >= 1),
+    document JSONB NOT NULL CHECK (jsonb_typeof(document) = 'object'),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);

tests/fixtures/public_api_snapshot.json

@@ -8941,6 +8941,7 @@
   },
   "adcp_reporting_storage": {
     "resolves": {
+      "PgPrincipalRecordStore": "adcp_reporting_storage.subscriptions.pg",
       "PgReportingActivityUnionStore": "adcp_reporting_storage.outbox.status_activity_pg",
       "PgReportingFeedStore": "adcp_reporting_storage.feed.pg",
       "PgReportingLedgerStore": "adcp_reporting_storage.ledger.pg",
@@ -8957,6 +8958,7 @@
       "PgReportingStatusOutbox": "adcp_reporting_storage.outbox.status_pg",
       "PgReportingStatusProjection": "adcp_reporting_storage.projection.pg",
       "PgReportingSubmissionIntentStore": "adcp_reporting_storage.submissions.pg",
+      "PgReportingSubscriptionStateStore": "adcp_reporting_storage.subscriptions.pg",
       "PgStatusNotificationStore": "adcp_reporting_storage.outbox.status_pg",
       "ReportingChangeCursorExpiredError": "adcp_reporting_storage.ledger.change_feed",
       "ReportingChangeKindV1": "union[value:'adjustment',value:'obligation',value:'retirement',value:'revision']",