diff --git a/packages/adcp-reporting-storage/pyproject.toml b/packages/adcp-reporting-storage/pyproject.toml index 2fc517009..b1fac8367 100644 --- a/packages/adcp-reporting-storage/pyproject.toml +++ b/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 diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py index 8979c0afc..e8b9f72bf 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py +++ b/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", } ) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py new file mode 100644 index 000000000..20ed35401 --- /dev/null +++ b/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", +] diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py new file mode 100644 index 000000000..efcf23f49 --- /dev/null +++ b/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) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql new file mode 100644 index 000000000..074f5dd02 --- /dev/null +++ b/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() +); diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 1dabac09d..ca91211e1 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/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']",