225 lines
8.8 KiB
Python
225 lines
8.8 KiB
Python
"""Tenant-scoped append-only message ledger and exact quote resolver."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from contextlib import closing
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Iterable
|
|
|
|
from .tenant_context import (
|
|
CanonicalMessage,
|
|
ConflictError,
|
|
ScopeError,
|
|
ValidationError,
|
|
isoformat,
|
|
normalize_message_id,
|
|
parse_time,
|
|
utc_now,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LedgerRecord:
|
|
tenant_id: str
|
|
cell_id: str
|
|
channel: str
|
|
channel_account_id: str
|
|
peer_id: str
|
|
direction: str
|
|
message_id: str
|
|
event_id: str
|
|
body: str
|
|
occurred_at: str
|
|
expires_at: str
|
|
|
|
|
|
class MessageLedger:
|
|
"""Exact-ID ledger.
|
|
|
|
The resolver never uses time windows, prefixes, similarity, or nearest-message
|
|
fallbacks. Ambiguity and absence fail closed.
|
|
"""
|
|
|
|
def __init__(self, database: str | Path, *, ttl_days: int = 7) -> None:
|
|
if not 1 <= ttl_days <= 365:
|
|
raise ValidationError("ttl_days must be between 1 and 365")
|
|
self.database = Path(database)
|
|
self.ttl_days = ttl_days
|
|
self.database.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
try:
|
|
os.chmod(self.database, 0o600)
|
|
except OSError:
|
|
pass
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
db = sqlite3.connect(str(self.database), timeout=30)
|
|
db.row_factory = sqlite3.Row
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("PRAGMA synchronous=FULL")
|
|
db.execute("PRAGMA busy_timeout=30000")
|
|
return db
|
|
|
|
def _initialize(self) -> None:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
db.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS message_ledger (
|
|
tenant_id TEXT NOT NULL,
|
|
cell_id TEXT NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
channel_account_id TEXT NOT NULL,
|
|
peer_id TEXT NOT NULL,
|
|
direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
|
|
message_id TEXT NOT NULL,
|
|
event_id TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
occurred_at TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
PRIMARY KEY(
|
|
tenant_id, cell_id, channel, channel_account_id,
|
|
peer_id, direction, message_id
|
|
),
|
|
UNIQUE(tenant_id, cell_id, event_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
|
|
tenant_id, cell_id, channel, channel_account_id,
|
|
peer_id, message_id, expires_at
|
|
);
|
|
"""
|
|
)
|
|
|
|
def append(self, message: CanonicalMessage) -> LedgerRecord:
|
|
"""Append one exact message through the runtime write path.
|
|
|
|
This is intentionally the owner of validation, idempotency, and the
|
|
immediate SQLite transaction. Runtime callers must not depend on the
|
|
legacy ``remember`` compatibility name.
|
|
"""
|
|
message_id = normalize_message_id(message.message_id)
|
|
now = utc_now()
|
|
expires = isoformat(now + timedelta(days=self.ttl_days))
|
|
key = (
|
|
message.tenant_id,
|
|
message.cell_id,
|
|
message.channel,
|
|
message.channel_account_id,
|
|
message.peer_id,
|
|
message.direction,
|
|
message_id,
|
|
)
|
|
storage_error: ConflictError | None = None
|
|
try:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
existing = db.execute(
|
|
"SELECT event_id, body, occurred_at, expires_at FROM message_ledger "
|
|
"WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
|
|
"AND peer_id=? AND direction=? AND message_id=?",
|
|
key,
|
|
).fetchone()
|
|
if existing:
|
|
if existing["body"] != message.body:
|
|
db.rollback()
|
|
raise ConflictError("message ID collision with different content")
|
|
db.commit()
|
|
return LedgerRecord(
|
|
*key, existing["event_id"], existing["body"],
|
|
existing["occurred_at"], existing["expires_at"],
|
|
)
|
|
db.execute(
|
|
"INSERT INTO message_ledger VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(*key, message.event_id, message.body, message.occurred_at, expires, isoformat(now)),
|
|
)
|
|
db.commit()
|
|
except sqlite3.Error:
|
|
storage_error = ConflictError("ledger storage unavailable")
|
|
if storage_error is not None:
|
|
# Raise outside the sqlite exception handler so callers cannot inspect
|
|
# a driver exception through __cause__ or __context__.
|
|
raise storage_error
|
|
return LedgerRecord(*key, message.event_id, message.body, message.occurred_at, expires)
|
|
|
|
def remember(self, message: CanonicalMessage) -> LedgerRecord:
|
|
"""Backward-compatible API wrapper for callers outside the runtime path."""
|
|
return self.append(message)
|
|
|
|
def matches(self, message: CanonicalMessage) -> bool:
|
|
"""Whether this exact message is already durably represented."""
|
|
message_id = normalize_message_id(message.message_id)
|
|
with closing(self._connect()) as connection, connection as db:
|
|
row = db.execute(
|
|
"SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
|
|
"cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
|
|
"AND direction=? AND message_id=?",
|
|
(
|
|
message.tenant_id, message.cell_id, message.channel,
|
|
message.channel_account_id, message.peer_id, message.direction,
|
|
message_id,
|
|
),
|
|
).fetchone()
|
|
return bool(row and row["event_id"] == message.event_id and row["body"] == message.body)
|
|
|
|
def remove(self, record: LedgerRecord) -> None:
|
|
"""Compensate a newly appended record if a later ingress step fails."""
|
|
with closing(self._connect()) as connection, connection as db:
|
|
db.execute(
|
|
"DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
|
|
"channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
|
|
"AND message_id=? AND event_id=?",
|
|
(
|
|
record.tenant_id, record.cell_id, record.channel,
|
|
record.channel_account_id, record.peer_id, record.direction,
|
|
record.message_id, record.event_id,
|
|
),
|
|
)
|
|
|
|
def resolve_quote(self, message: CanonicalMessage) -> str:
|
|
if message.quoted_body is not None:
|
|
return message.quoted_body
|
|
if message.reply_to_message_id is None:
|
|
raise ScopeError("quoted_text_unavailable")
|
|
reply_id = normalize_message_id(
|
|
message.reply_to_message_id,
|
|
field_name="reply_to_message_id",
|
|
)
|
|
with closing(self._connect()) as connection, connection as db:
|
|
rows = db.execute(
|
|
"SELECT direction, body, event_id FROM message_ledger "
|
|
"WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
|
|
"AND peer_id=? AND message_id=? AND expires_at>?",
|
|
(
|
|
message.tenant_id,
|
|
message.cell_id,
|
|
message.channel,
|
|
message.channel_account_id,
|
|
message.peer_id,
|
|
reply_id,
|
|
isoformat(),
|
|
),
|
|
).fetchall()
|
|
if len(rows) == 1:
|
|
return str(rows[0]["body"])
|
|
if len(rows) > 1:
|
|
raise ConflictError("ambiguous reply ID across message directions")
|
|
raise ScopeError("quoted_text_unavailable")
|
|
|
|
def purge_expired(self) -> int:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
cursor = db.execute(
|
|
"DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
|
|
)
|
|
return cursor.rowcount
|
|
|
|
def integrity_check(self) -> str:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
return str(db.execute("PRAGMA integrity_check").fetchone()[0])
|
|
|
|
def count(self) -> int:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
|