orchestrata_ultimate.postman.receipt_store

Reference › orchestrata_ultimate › postman › receipt_store

The durable store for delivery receipts, SPEC-002 story FR-007 (REQ-002-FR7-006 to -010) and story US-007 (REQ-002-US-007-019, -020). ReceiptStore implements IReceiptStore from interfaces and records the DeliveryReceipt values (from types) that prove a message reached, or failed to reach, a recipient. It follows the layering of message_store: SQLite primary, Redis cache and a graph audit queue. replay_engine takes it as a collaborator.

initialize creates the receipts table and, if missing, a messages table in a SQLite file (./data/receipts.db by default). store(receipt) makes sure a message row exists, inserts the receipt with its type and error message, appends it to a Redis list cached under receipts:<message id> for 30 minutes, and pushes a receipt_stored event onto the audit queue. get(message_id) returns all receipts for a message, and query filters by sender, recipient and time range with a default limit of 100. expire(older_than) deletes older receipts, clears the Redis receipt keys, records a receipts_expired event and returns the count.

At a glance

Class diagram

Dependency diagram

API

Classes

ReceiptStore

Bases: IReceiptStore · Decorators: traces(spec='002', story='FR-007', reqs=['REQ-002-FR7-006', 'REQ-002-FR7-007', 'REQ-002-FR7-008', 'REQ-002-FR7-010', 'R… · Line 62

Delivery receipt store.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/receipt_store.py

# mypy: ignore-errors
"""Receipt Store module for Orchestrata."""

from __future__ import annotations

import asyncio
import json
import pickle
import sqlite3
from collections.abc import Awaitable
from datetime import UTC, datetime
from typing import TYPE_CHECKING, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import DEFAULT_RECEIPT_QUERY_LIMIT
from orchestrata_ultimate.postman.interfaces import IReceiptStore

if TYPE_CHECKING:
    from orchestrata_ultimate.postman.types import DeliveryReceipt

# Spec: 002 - Postman Message Broker


"""
Receipt store using SQLite, Redis cache, graph audit.
Implements IReceiptStore interface.
"""


try:
    from redis.asyncio import Redis as AsyncRedis  # noqa: F401

    REDIS_AVAILABLE = True
except ImportError:
    REDIS_AVAILABLE = False


def _redis_result[T](value: Awaitable[T] | T) -> Awaitable[T]:
    """Narrow a redis-py command result back to its true async type.

    redis-py types command methods (e.g. ``lpush``) as ``Awaitable[T] | T``
    because the same command mixin backs both the sync and async clients
    (see ``redis.typing.ResponseT``). ``SQLiteReceiptStore`` only ever holds
    an async client (``create_async_redis``), which always returns an
    awaitable at runtime -- this documents that fact for pyright instead of
    suppressing the check.
    """
    return cast(Awaitable[T], value)


@traces(
    spec="002",
    story="FR-007",
    reqs=[
        "REQ-002-FR7-006",
        "REQ-002-FR7-007",
        "REQ-002-FR7-008",
        "REQ-002-FR7-010",
        "REQ-002-US-007-019",
    ],
)
class ReceiptStore(IReceiptStore):
    """Delivery receipt store."""

    def __init__(
        self,
        sqlite_path: str = "./data/receipts.db",
        redis_host: str | None = None,
        # No literal port default (RUL-003/RUL-011): "6379" would silently
        # aim the async client at nothing when REDIS_PORT isn't the target,
        # reading as an outage instead of the config gap it is. Resolved
        # lazily below via resolve_redis_port() so importing this module
        # never raises.
        redis_port: int | None = None,
        graph_uri: str | None = None,
        graph_user: str = "",
        graph_password: str | None = None,
    ) -> None:
        self.sqlite_path = sqlite_path
        self._init_sqlite()

        if REDIS_AVAILABLE:
            from orchestrata_ultimate.core.redis_factory import (
                create_async_redis,
                resolve_redis_host,
                resolve_redis_port,
            )

            self.redis_client = create_async_redis(
                host=redis_host if redis_host is not None else resolve_redis_host(),
                port=redis_port if redis_port is not None else resolve_redis_port(),
                decode_responses=False,
            )
        else:
            self.redis_client = None

    async def initialize(self) -> None:
        """Complete async initialisation.

        No-op: the SQLite schema is created synchronously in __init__.
        Kept for API compatibility with callers that await
        ``store.initialize()`` after construction, e.g.::

            store = ReceiptStore(...)
            await store.initialize()
        """
        return None

    @traces(
        spec="002",
        story="FR-007",
        reqs=["REQ-002-FR7-007", "REQ-002-FR7-008"],
    )
    def _init_sqlite(self) -> None:
        """Create receipts table and messages table if missing."""

        @traces(
            spec="002",
            story="FR-007",
            reqs=["REQ-002-FR7-007", "REQ-002-FR7-008"],
        )
        def sync_init() -> None:
            conn = sqlite3.connect(self.sqlite_path)
            cur = conn.cursor()
            # Ensure messages table exists (needed for joins)
            cur.execute("""
                CREATE TABLE IF NOT EXISTS messages (
                    message_id TEXT PRIMARY KEY,
                    sender TEXT,
                    recipient TEXT,
                    timestamp DATETIME
                )
            """)
            cur.execute("""
                CREATE TABLE IF NOT EXISTS receipts (
                    receipt_id TEXT PRIMARY KEY,
                    message_id TEXT,
                    recipient TEXT,
                    receipt_type TEXT,
                    timestamp DATETIME,
                    error_message TEXT,
                    FOREIGN KEY(message_id) REFERENCES messages(message_id)
                )
            """)
            cur.execute(
                "CREATE INDEX IF NOT EXISTS idx_receipt_message ON receipts(message_id)"
            )
            cur.execute(
                "CREATE INDEX IF NOT EXISTS idx_receipt_timestamp ON receipts(timestamp)"
            )
            conn.commit()
            conn.close()

        sync_init()

    async def _execute_sql(self, query: str, args: tuple = ()) -> list[tuple]:
        def sync_query() -> list[tuple]:
            conn = sqlite3.connect(self.sqlite_path)
            cur = conn.cursor()
            cur.execute(query, args)
            result = cur.fetchall()
            conn.commit()
            conn.close()
            return result

        return await asyncio.to_thread(sync_query)

    async def _execute_sql_write(self, query: str, args: tuple = ()) -> int:
        def sync_write() -> int:
            conn = sqlite3.connect(self.sqlite_path)
            cur = conn.cursor()
            cur.execute(query, args)
            rowcount = cur.rowcount
            conn.commit()
            conn.close()
            return rowcount

        return await asyncio.to_thread(sync_write)

    @traces(
        spec="002",
        story="FR-007",
        reqs=["REQ-002-FR7-006", "REQ-002-FR7-007", "REQ-002-US-007-019"],
    )
    async def store(self, receipt: DeliveryReceipt) -> None:
        """Store a delivery receipt."""
        # Ensure the referenced message exists in messages table (dummy)
        await self._execute_sql_write(
            "INSERT OR IGNORE INTO messages (message_id, sender, recipient, timestamp) VALUES (?, ?, ?, ?)",
            (receipt.message_id, "", "", datetime.now(UTC).isoformat()),
        )
        await self._execute_sql_write(
            """
            INSERT INTO receipts (receipt_id, message_id, recipient, receipt_type, timestamp, error_message)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                receipt.receipt_id,
                receipt.message_id,
                receipt.recipient,
                receipt.receipt_type.value if receipt.receipt_type else None,
                receipt.timestamp.isoformat(),
                receipt.error_message,
            ),
        )
        # Redis cache for receipts of a message
        if self.redis_client:
            key = f"receipts:{receipt.message_id}"
            cached = await self.redis_client.get(key)
            if cached:
                receipts_list = pickle.loads(cached)
            else:
                receipts_list = []
            receipts_list.append(receipt.to_dict())
            await self.redis_client.setex(key, 1800, pickle.dumps(receipts_list))

        # Graph audit via Redis queue
        if self.redis_client:
            audit_event = {
                "event_type": "receipt_stored",
                "receipt_id": receipt.receipt_id,
                "message_id": receipt.message_id,
                "recipient": receipt.recipient,
                "receipt_type": (
                    receipt.receipt_type.value if receipt.receipt_type else None
                ),
                "timestamp": receipt.timestamp.isoformat(),
                "error_message": receipt.error_message,
            }
            await _redis_result(
                self.redis_client.lpush("graph:audit:queue", json.dumps(audit_event))
            )

    @traces(spec="002", story="FR-007", reqs=["REQ-002-FR7-011"])
    async def get(self, message_id: str) -> list[DeliveryReceipt]:
        """Get all receipts for a message."""
        from orchestrata_ultimate.postman.types import DeliveryReceipt, ReceiptType

        if self.redis_client:
            cached = await self.redis_client.get(f"receipts:{message_id}")
            if cached:
                receipt_dicts = pickle.loads(cached)
                return [DeliveryReceipt.from_dict(d) for d in receipt_dicts]

        rows = await self._execute_sql(
            """
            SELECT receipt_id, message_id, recipient, receipt_type, timestamp, error_message
            FROM receipts WHERE message_id = ? ORDER BY timestamp
            """,
            (message_id,),
        )
        receipts = []
        for row in rows:
            receipt = DeliveryReceipt(
                receipt_id=row[0],
                message_id=row[1],
                recipient=row[2],
                receipt_type=ReceiptType(row[3]),
                timestamp=datetime.fromisoformat(row[4]),
                error_message=row[5],
            )
            receipts.append(receipt)

        # Cache for future
        if self.redis_client and receipts:
            receipt_dicts = [r.to_dict() for r in receipts]
            await self.redis_client.setex(
                f"receipts:{message_id}", 1800, pickle.dumps(receipt_dicts)
            )

        return receipts

    @traces(
        spec="002",
        story="FR-007",
        reqs=["REQ-002-FR7-012", "REQ-002-FR7-013", "REQ-002-FR7-014"],
    )
    async def query(
        self,
        sender: str | None = None,
        recipient: str | None = None,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        limit: int = DEFAULT_RECEIPT_QUERY_LIMIT,
    ) -> list[DeliveryReceipt]:
        """Query receipts, optionally joining messages for sender."""
        from orchestrata_ultimate.postman.types import DeliveryReceipt, ReceiptType

        query_parts = [
            "SELECT r.receipt_id, r.message_id, r.recipient, r.receipt_type, r.timestamp, r.error_message",
            "FROM receipts r",
        ]
        params = []
        if sender:
            query_parts.append("JOIN messages m ON r.message_id = m.message_id")
            query_parts.append("WHERE m.sender = ?")
            params.append(sender)
        else:
            query_parts.append("WHERE 1=1")
        if recipient:
            query_parts.append("AND r.recipient = ?")
            params.append(recipient)
        if start_time:
            query_parts.append("AND r.timestamp >= ?")
            params.append(start_time.isoformat())
        if end_time:
            query_parts.append("AND r.timestamp <= ?")
            params.append(end_time.isoformat())

        query_parts.append(f"ORDER BY r.timestamp DESC LIMIT {limit}")
        sql = " ".join(query_parts)
        rows = await self._execute_sql(sql, tuple(params))

        receipts = [
            DeliveryReceipt(
                receipt_id=row[0],
                message_id=row[1],
                recipient=row[2],
                receipt_type=ReceiptType(row[3]),
                timestamp=datetime.fromisoformat(row[4]),
                error_message=row[5],
            )
            for row in rows
        ]
        return receipts

    @traces(
        spec="002",
        story="FR-007",
        reqs=["REQ-002-FR7-010", "REQ-002-US-007-020"],
    )
    async def expire(self, older_than: datetime) -> int:
        """Expire receipts older than cutoff."""
        rows = await self._execute_sql(
            "SELECT COUNT(*) FROM receipts WHERE timestamp < ?",
            (older_than.isoformat(),),
        )
        count = rows[0][0] if rows else 0
        await self._execute_sql_write(
            "DELETE FROM receipts WHERE timestamp < ?",
            (older_than.isoformat(),),
        )
        # Invalidate Redis receipt cache keys
        if self.redis_client:
            async for key in self.redis_client.scan_iter(match="receipts:*"):
                await self.redis_client.delete(key)

        # Graph audit via Redis queue
        if self.redis_client:
            audit_event = {
                "event_type": "receipts_expired",
                "cutoff": older_than.isoformat(),
                "count": count,
            }
            await _redis_result(
                self.redis_client.lpush("graph:audit:queue", json.dumps(audit_event))
            )

        return count