""" Federation Node — Envelope Store (Phase 2 Vault) ================================================ Contract: C-FED-NODE-001 v0.1.1 Persistent SQLite envelope storage backed by Cloudflare D1. """ from __future__ import annotations import uuid import json from datetime import datetime, timezone from typing import Optional from models import DeliveryStatus import d1_client class EnvelopeStore: def __init__(self, max_envelopes: int = 1000): self._max = max_envelopes async def get_count(self) -> int: rs = await d1_client.execute_sql("SELECT COUNT(*) as c FROM envelopes") return rs[0]["c"] if rs else 0 async def is_full(self) -> bool: return await self.get_count() >= self._max async def accept( self, *, sender_seal: str, sender_origin: str, message_class: str, payload: dict, delivery: dict, trust_tier: str, initial_status: DeliveryStatus = DeliveryStatus.QUEUED, ) -> dict: now = datetime.now(timezone.utc).isoformat() envelope_id = str(uuid.uuid4()) await d1_client.execute_sql( '''INSERT INTO envelopes (envelope_id, protocol_version, message_class, sender_seal, recipient_seal, payload_body, status, received_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', [ envelope_id, "1.0.0", message_class, sender_seal, "NODE", json.dumps(payload), initial_status.value, now ] ) history = [ {"status": DeliveryStatus.RECEIVED.value, "at": now}, {"status": DeliveryStatus.VALIDATED.value, "at": now}, ] await d1_client.execute_sql( "INSERT INTO status_history (envelope_id, status, transitioned_at) VALUES (?, ?, ?), (?, ?, ?)", [envelope_id, DeliveryStatus.RECEIVED.value, now, envelope_id, DeliveryStatus.VALIDATED.value, now] ) if initial_status == DeliveryStatus.QUARANTINED: history.append({"status": DeliveryStatus.QUARANTINED.value, "at": now}) await d1_client.execute_sql( "INSERT INTO status_history (envelope_id, status, transitioned_at) VALUES (?, ?, ?)", [envelope_id, DeliveryStatus.QUARANTINED.value, now] ) else: history.append({"status": DeliveryStatus.QUEUED.value, "at": now}) await d1_client.execute_sql( "INSERT INTO status_history (envelope_id, status, transitioned_at) VALUES (?, ?, ?)", [envelope_id, DeliveryStatus.QUEUED.value, now] ) return { "envelope_id": envelope_id, "status": initial_status, "sender_seal": sender_seal, "sender_origin": sender_origin, "message_class": message_class, "payload": payload, "delivery": delivery, "trust_tier": trust_tier, "received_at": now, "status_history": history, } async def get_envelope(self, envelope_id: str) -> Optional[dict]: rs = await d1_client.execute_sql("SELECT * FROM envelopes WHERE envelope_id = ?", [envelope_id]) if not rs: return None record = rs[0] try: payload = json.loads(record["payload_body"]) except: payload = {"body": record["payload_body"]} return { "envelope_id": record["envelope_id"], "status": record["status"], "sender_seal": record["sender_seal"], "message_class": record["message_class"], "received_at": record["received_at"], "payload": payload, } async def list_envelopes(self, limit: int = 20, status_filter: Optional[str] = None) -> dict: total_rs = await d1_client.execute_sql("SELECT COUNT(*) as c FROM envelopes") total = total_rs[0]["c"] if total_rs else 0 query = "SELECT * FROM envelopes" params = [] if status_filter: query += " WHERE status = ?" params.append(status_filter) query += " ORDER BY received_at DESC LIMIT ?" params.append(limit) rs = await d1_client.execute_sql(query, params) envelopes = [] for r in rs: try: payload = json.loads(r["payload_body"]) preview = str(payload.get("body", ""))[:120] except: preview = str(r["payload_body"])[:120] envelopes.append({ "envelope_id": r["envelope_id"], "status": r["status"], "sender_origin": "Federation", "sender_seal": r["sender_seal"], "message_class": r["message_class"], "received_at": r["received_at"], "payload_preview": preview, }) return { "total": total, "showing": len(envelopes), "envelopes": envelopes } async def get_status(self, envelope_id: str) -> Optional[dict]: rs = await d1_client.execute_sql("SELECT * FROM envelopes WHERE envelope_id = ?", [envelope_id]) if not rs: return None record = rs[0] hist_rs = await d1_client.execute_sql("SELECT status, transitioned_at as at FROM status_history WHERE envelope_id = ? ORDER BY id ASC", [envelope_id]) return { "envelope_id": record["envelope_id"], "status": record["status"], "sender_seal": record["sender_seal"], "message_class": record["message_class"], "trust_tier": "UNKNOWN", "received_at": record["received_at"], "status_history": hist_rs, }