# Author: Sandeep Belgavi
"""Offline event collaboration: transactional inbox, incident state and outbox."""
import json
import sqlite3
import tempfile
from pathlib import Path


class EventHub:
    def __init__(self, path):
        self.db = sqlite3.connect(path)
        self.db.executescript("""
        CREATE TABLE IF NOT EXISTS inbox(id TEXT PRIMARY KEY, body TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS incidents(id TEXT PRIMARY KEY, severity INTEGER);
        CREATE TABLE IF NOT EXISTS outbox(id TEXT PRIMARY KEY, body TEXT NOT NULL, sent INTEGER DEFAULT 0);
        """)

    def accept(self, event, fail=False):
        required = {'id', 'incident', 'severity', 'version'}
        if set(event) != required or event['version'] != 1:
            raise ValueError('schema')
        if not all(isinstance(event[k], str) and event[k] for k in ('id', 'incident')):
            raise ValueError('identifier')
        if type(event['severity']) is not int or not 1 <= event['severity'] <= 5:
            raise ValueError('severity')
        body = json.dumps(event, sort_keys=True)
        with self.db:
            self.db.execute('BEGIN IMMEDIATE')
            prior = self.db.execute('SELECT body FROM inbox WHERE id=?', (event['id'],)).fetchone()
            if prior:
                if prior[0] != body:
                    raise ValueError('event id reused with different content')
                return 'duplicate'
            self.db.execute('INSERT INTO inbox VALUES (?,?)', (event['id'], body))
            self.db.execute('INSERT INTO incidents VALUES (?,?) ON CONFLICT(id) DO UPDATE SET severity=max(severity,excluded.severity)', (event['incident'], event['severity']))
            output = json.dumps({'incident': event['incident'], 'route': 'investigate'}, sort_keys=True)
            self.db.execute('INSERT INTO outbox(id,body) VALUES (?,?)', (event['id'], output))
            if fail:
                raise RuntimeError('injected before commit')
        return 'accepted'

    def drain(self, adapter, fail_after_send=False):
        # Adapter contract: publish(key, payload) must tolerate repeated keys.
        for key, body in self.db.execute('SELECT id,body FROM outbox WHERE sent=0 ORDER BY id').fetchall():
            adapter.publish(key, json.loads(body))
            if fail_after_send:
                raise RuntimeError('injected after delivery')
            with self.db:
                self.db.execute('UPDATE outbox SET sent=1 WHERE id=?', (key,))

    def close(self):
        self.db.close()


class FakeWorker:
    """Deterministic in-memory provider seam, not an LLM or durable broker."""
    def __init__(self):
        self.receipts = {}

    def publish(self, key, payload):
        if key in self.receipts and self.receipts[key] != payload:
            raise ValueError('delivery conflict')
        self.receipts[key] = payload


def demo():
    with tempfile.TemporaryDirectory() as folder:
        hub = EventHub(Path(folder) / 'events.db')
        event = {'id': 'evt-1', 'incident': 'inc-42', 'severity': 4, 'version': 1}
        first, second = hub.accept(event), hub.accept(event)
        worker = FakeWorker()
        hub.drain(worker)
        result = {'first': first, 'repeat': second, 'deliveries': len(worker.receipts), 'severity': hub.db.execute('SELECT severity FROM incidents').fetchone()[0]}
        hub.close()
        return result


if __name__ == '__main__':
    print(json.dumps(demo(), sort_keys=True))