"""Receptor persistente de ejemplo. Python 3, sin dependencias ni red.

Precondición: mensajes autenticados por un mecanismo externo no implementado.
Solo proyecta estados completos en SQLite; no llama a Moodle. Los dos cambios
locales comparten transacción, algo que no se extiende a sistemas remotos.
Ejecutar: python3 laboratorio-eventos.py
"""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import hashlib
import json
import sqlite3
import tempfile


def initialize(path):
    with sqlite3.connect(path) as db:
        db.executescript('''
            CREATE TABLE inbox (source TEXT, event_id TEXT, digest TEXT, decision TEXT,
                                PRIMARY KEY(source, event_id));
            CREATE TABLE projection (relation TEXT PRIMARY KEY, revision INTEGER, state TEXT);
            CREATE TABLE incidents (event_id TEXT, reason TEXT);
        ''')


def process(path, event, fail_before_commit=False):
    if event['source'] != 'ACADEMIC_DEMO' or event['kind'] != 'complete_state':
        raise ValueError('El ejemplo solo admite la fuente acordada y estados completos')
    if event['state'] not in ['active', 'suspended'] or event['revision'] < 1:
        raise ValueError('Estado o revisión inválidos')
    digest = hashlib.sha256(json.dumps(event, sort_keys=True).encode()).hexdigest()
    db = sqlite3.connect(path, timeout=10)
    try:
        db.execute('BEGIN IMMEDIATE')
        old = db.execute('SELECT digest FROM inbox WHERE source=? AND event_id=?',
                         (event['source'], event['event_id'])).fetchone()
        if old:
            decision = 'duplicate' if old[0] == digest else 'quarantine-id-collision'
            if old[0] != digest:
                db.execute('INSERT INTO incidents VALUES (?, ?)', (event['event_id'], decision))
            db.commit()
            return decision
        current = db.execute('SELECT revision, state FROM projection WHERE relation=?',
                             (event['relation'],)).fetchone()
        if current and event['revision'] < current[0]:
            decision = 'stale'
        elif current and event['revision'] == current[0]:
            decision = 'redundant' if current[1] == event['state'] else 'quarantine-revision-conflict'
            if decision.startswith('quarantine'):
                db.execute('INSERT INTO incidents VALUES (?, ?)', (event['event_id'], decision))
        else:
            decision = 'applied'
            db.execute('INSERT INTO projection VALUES (?, ?, ?) ON CONFLICT(relation) DO UPDATE '
                       'SET revision=excluded.revision, state=excluded.state',
                       (event['relation'], event['revision'], event['state']))
        db.execute('INSERT INTO inbox VALUES (?, ?, ?, ?)',
                   (event['source'], event['event_id'], digest, decision))
        if fail_before_commit:
            raise RuntimeError('Interrupción simulada antes de confirmar la transacción local')
        db.commit()
        return decision
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()


def message(eid, revision, state):
    return {'source': 'ACADEMIC_DEMO', 'event_id': eid, 'kind': 'complete_state',
            'relation': 'ACADEMIC_DEMO:ED_DEMO:PERSON_DEMO:ENROL_DEMO',
            'revision': revision, 'state': state}


def state(path):
    with sqlite3.connect(path) as db:
        row = db.execute('SELECT revision, state FROM projection').fetchone()
        return {'revision': row[0], 'state': row[1]} if row else None


def main():
    with tempfile.TemporaryDirectory(prefix='inbox-educativo-') as folder:
        path = Path(folder) / 'order.sqlite'
        initialize(path)
        events = [message('E2', 2, 'suspended'), message('E1', 1, 'active'),
                  message('E2', 2, 'suspended'), message('E2', 2, 'active'),
                  message('E_CONFLICT', 2, 'active')]
        decisions = [process(path, e) for e in events]
        assert decisions == ['applied', 'stale', 'duplicate', 'quarantine-id-collision',
                             'quarantine-revision-conflict']
        assert state(path) == {'revision': 2, 'state': 'suspended'}
        ordered = {'decisions': decisions, 'final_state': state(path)}
        path2 = Path(folder) / 'failure.sqlite'
        initialize(path2)
        try:
            process(path2, message('F1', 1, 'active'), fail_before_commit=True)
        except RuntimeError:
            pass
        assert state(path2) is None
        after_failure = state(path2)
        retry = process(path2, message('F1', 1, 'active'))
        assert retry == 'applied'
        failure = {'state_after_rollback': after_failure, 'retry': retry, 'final_state': state(path2)}
        path3 = Path(folder) / 'concurrent.sqlite'
        initialize(path3)
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(pool.map(lambda _: process(path3, message('C1', 1, 'active')), range(2)))
        assert sorted(results) == ['applied', 'duplicate']
        with sqlite3.connect(path3) as db:
            count = db.execute('SELECT count(*) FROM inbox').fetchone()[0]
        assert count == 1
        concurrent = {'worker_decisions': sorted(results), 'inbox_rows': count, 'final_state': state(path3)}
        print(json.dumps({'scope': 'Laboratorio SQLite con mensajes ya autenticados; sin red, Moodle ni cola distribuida.',
                          'ordering_and_duplicates': ordered, 'rollback_and_retry': failure,
                          'two_workers_same_event': concurrent}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()