62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
"""Cursor reads for the durable project-scoped workbench event log."""
|
|
from __future__ import annotations
|
|
|
|
from psycopg import AsyncConnection, Connection
|
|
|
|
|
|
EVENT_ORDER_LOCK = "tjwater-workbench:event-order"
|
|
|
|
|
|
async def lock_event_transaction(connection: AsyncConnection) -> None:
|
|
"""Serialize v2 event-producing transactions before business locks/writes.
|
|
|
|
PostgreSQL sequence allocation is not commit order. All producers must use
|
|
this same transaction lock so a committed higher cursor cannot hide an
|
|
earlier, still-uncommitted event. Other BizDBs remain rollout opt-out.
|
|
"""
|
|
if getattr(getattr(connection, "info", None), "dbname", None) == "tjwater_v2":
|
|
await connection.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,)
|
|
)
|
|
|
|
|
|
def lock_event_transaction_sync(connection: Connection) -> None:
|
|
"""Synchronous counterpart using the identical PostgreSQL lock key."""
|
|
if getattr(getattr(connection, "info", None), "dbname", None) == "tjwater_v2":
|
|
connection.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,)
|
|
)
|
|
|
|
|
|
class WorkbenchEventRepository:
|
|
def __init__(self, connection: AsyncConnection, project_id):
|
|
self.connection = connection
|
|
self.project_id = str(project_id)
|
|
|
|
async def current_cursor(self) -> int:
|
|
cursor = await self.connection.execute(
|
|
"SELECT COALESCE(max(event_seq),0) AS cursor FROM workbench.domain_event_log WHERE project_id=%s",
|
|
(self.project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return int(row["cursor"])
|
|
|
|
async def oldest_cursor(self) -> int | None:
|
|
cursor = await self.connection.execute(
|
|
"SELECT min(event_seq) AS cursor FROM workbench.domain_event_log WHERE project_id=%s",
|
|
(self.project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return int(row["cursor"]) if row and row["cursor"] is not None else None
|
|
|
|
async def read_after(self, after: int, limit: int = 200) -> list[dict]:
|
|
cursor = await self.connection.execute(
|
|
"""SELECT event_seq,event_id,project_id,topic,aggregate_type,aggregate_id,
|
|
aggregate_revision,occurred_at,metadata
|
|
FROM workbench.domain_event_log
|
|
WHERE project_id=%s AND event_seq>%s
|
|
ORDER BY event_seq LIMIT %s""",
|
|
(self.project_id, after, limit),
|
|
)
|
|
return [dict(row) for row in await cursor.fetchall()]
|