#!/usr/bin/env python3 """Private reference receiver. No listener, credentials or notification sending.""" import argparse from datetime import datetime, timezone import hashlib import json from pathlib import Path import sqlite3 import sys import uuid STATES = {'healthy', 'stale', 'missing', 'unavailable', 'failed'} MAX_BYTES = 32768 def instant(value): parsed = datetime.fromisoformat(value.replace('Z', '+00:00')) if parsed.tzinfo is None: raise ValueError('timezone required') return parsed.astimezone(timezone.utc) def strict_json(raw): def pairs(items): result = {} for key, value in items: if key in result: raise ValueError('duplicate field') result[key] = value return result if len(raw) > MAX_BYTES: raise ValueError('payload too large') return json.loads(raw, object_pairs_hook=pairs) def read_json(path): with path.open('rb') as stream: return strict_json(stream.read(MAX_BYTES + 1)) def contract_check(c): if set(c) != {'schema', 'stream', 'producer', 'recipient', 'signals', 'max_event_age_seconds', 'heartbeat_seconds', 'retention_days'}: raise ValueError('contract fields') if c['schema'] != 'railiance-telemetry.stream.v1': raise ValueError('contract version') for key in ('stream', 'producer', 'recipient'): if not isinstance(c[key], str) or not c[key] or len(c[key]) > 100: raise ValueError('identity missing') if (not isinstance(c['signals'], list) or not c['signals'] or not all(isinstance(s, str) and 0 < len(s) <= 100 for s in c['signals']) or len(c['signals']) != len(set(c['signals']))): raise ValueError('signal inventory') for key in ('max_event_age_seconds', 'heartbeat_seconds', 'retention_days'): if type(c[key]) is not int or not 0 < c[key] <= 31536000: raise ValueError('invalid budget') class Receiver: def __init__(self, database, contract): contract_check(contract) self.contract = contract self.db = sqlite3.connect(database, timeout=10) self.db.executescript(''' CREATE TABLE IF NOT EXISTS binding (id INTEGER PRIMARY KEY, digest TEXT NOT NULL); CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, digest TEXT NOT NULL, emitted REAL NOT NULL, received REAL NOT NULL, states TEXT NOT NULL); CREATE TABLE IF NOT EXISTS notices ( id INTEGER PRIMARY KEY, kind TEXT NOT NULL, created REAL NOT NULL, event_id TEXT, acknowledged REAL); ''') digest = hashlib.sha256(json.dumps(contract, sort_keys=True).encode()).hexdigest() with self.db: self.db.execute('INSERT OR IGNORE INTO binding VALUES (1, ?)', (digest,)) if self.db.execute('SELECT digest FROM binding WHERE id=1').fetchone()[0] != digest: self.db.close() raise ValueError('contract drift: explicit migration required') def close(self): self.db.close() def ingest(self, event, now): c = self.contract if set(event) != {'schema', 'id', 'stream', 'producer', 'observed_at', 'states'}: raise ValueError('event fields') if (event['schema'] != 'railiance-telemetry.signal.v1' or event['stream'] != c['stream'] or event['producer'] != c['producer'] or str(uuid.UUID(event['id'])) != event['id']): raise ValueError('event identity') states = event['states'] if (not isinstance(states, dict) or set(states) != set(c['signals']) or not all(isinstance(s, str) and s in STATES for s in states.values())): raise ValueError('signal scope') emitted = instant(event['observed_at']).timestamp() age = now.timestamp() - emitted if not 0 <= age <= c['max_event_age_seconds']: raise ValueError('stale or future event') digest = hashlib.sha256(json.dumps(event, sort_keys=True).encode()).hexdigest() with self.db: self.db.execute('BEGIN IMMEDIATE') existing = self.db.execute('SELECT digest FROM events WHERE id=?', (event['id'],)).fetchone() if existing: if existing[0] != digest: raise ValueError('event id collision') return {'status': 'duplicate', 'id': event['id']} latest = self.db.execute('SELECT MAX(emitted) FROM events').fetchone()[0] if latest is not None and emitted <= latest: raise ValueError('out of order event') if self.db.execute('SELECT COUNT(*) FROM events').fetchone()[0] >= 10000: raise ValueError('capacity reached; explicit retention required') self.db.execute('INSERT INTO events VALUES (?, ?, ?, ?, ?)', (event['id'], digest, emitted, now.timestamp(), json.dumps(states, sort_keys=True))) self.db.execute('INSERT INTO notices(kind, created, event_id) VALUES (?, ?, ?)', ('healthy' if all(s == 'healthy' for s in states.values()) else 'unhealthy', now.timestamp(), event['id'])) return {'status': 'accepted', 'id': event['id'], 'delivery': 'local-inbox-only'} def check(self, now): """Must be invoked by an independent scheduler; absence is not self-executing.""" with self.db: self.db.execute('BEGIN IMMEDIATE') last = self.db.execute('SELECT MAX(emitted) FROM events').fetchone()[0] if last is not None and now.timestamp() < last: raise ValueError('clock moved backwards') absent = last is None or now.timestamp() - last > self.contract['heartbeat_seconds'] previous = self.db.execute('SELECT kind FROM notices ORDER BY id DESC LIMIT 1').fetchone() if absent and (previous is None or previous[0] != 'missing-emission'): self.db.execute('INSERT INTO notices(kind, created) VALUES (?, ?)', ('missing-emission', now.timestamp())) return {'state': 'missing-emission' if absent else 'receiving', 'recipient': self.contract['recipient'], 'delivery': 'local-inbox-only'} def inbox(self): rows = self.db.execute('''SELECT n.id, n.kind, n.created, n.event_id, e.states FROM notices n LEFT JOIN events e ON e.id=n.event_id WHERE n.acknowledged IS NULL ORDER BY n.id''').fetchall() return {'recipient': self.contract['recipient'], 'delivery': 'local-inbox-only', 'notices': [dict(id=r[0], kind=r[1], created_at=datetime.fromtimestamp(r[2], timezone.utc).isoformat(), event_id=r[3], states=json.loads(r[4]) if r[4] else None) for r in rows]} def ack(self, notice_id, now): with self.db: updated = self.db.execute('UPDATE notices SET acknowledged=? WHERE id=? AND acknowledged IS NULL', (now.timestamp(), notice_id)).rowcount return {'acknowledged': bool(updated), 'id': notice_id} def prune(self, now): cutoff = now.timestamp() - self.contract['retention_days'] * 86400 with self.db: self.db.execute('BEGIN IMMEDIATE') # Keep the latest observation and notice as replay/absence anchors. self.db.execute('''DELETE FROM notices WHERE created < ? AND acknowledged IS NOT NULL AND id != (SELECT MAX(id) FROM notices)''', (cutoff,)) deleted = self.db.execute('''DELETE FROM events WHERE received < ? AND emitted != (SELECT MAX(emitted) FROM events) AND id NOT IN (SELECT event_id FROM notices WHERE event_id IS NOT NULL)''', (cutoff,)).rowcount return {'events_pruned': deleted} def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--contract', required=True, type=Path) p.add_argument('--database', required=True, type=Path) sub = p.add_subparsers(dest='command', required=True) sub.add_parser('ingest').add_argument('event', type=Path) for name in ('check', 'inbox', 'prune'): sub.add_parser(name) sub.add_parser('ack').add_argument('notice_id', type=int) a = p.parse_args() receiver = None try: receiver = Receiver(a.database, read_json(a.contract)) now = datetime.now(timezone.utc) if a.command == 'ingest': result = receiver.ingest(read_json(a.event), now) elif a.command == 'inbox': result = receiver.inbox() elif a.command == 'ack': result = receiver.ack(a.notice_id, now) else: result = getattr(receiver, a.command)(now) print(json.dumps(result)) return 0 except (ValueError, TypeError, KeyError, AttributeError, OSError, sqlite3.Error): print(json.dumps({'status': 'rejected', 'error': 'invalid-input-or-unavailable'})) return 2 finally: if receiver is not None: receiver.close() if __name__ == '__main__': sys.exit(main())