# Author: Sandeep Belgavi
"""Offline checkpoint example; no distributed exactly-once claim."""
import json
import sqlite3
import tempfile
from pathlib import Path


class Workflow:
    def __init__(self, path, version='v1'):
        self.version = version
        self.db = sqlite3.connect(path)
        self.db.executescript("""
        CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY, version TEXT, step INTEGER, evidence TEXT);
        CREATE TABLE IF NOT EXISTS receipts(id TEXT PRIMARY KEY, body TEXT);
        """)

    def start(self, run_id):
        if not isinstance(run_id, str) or not run_id:
            raise ValueError('run id')
        with self.db:
            self.db.execute('INSERT OR IGNORE INTO runs VALUES (?,?,0,NULL)', (run_id, self.version))
        return self.status(run_id)

    def status(self, run_id):
        row = self.db.execute('SELECT version,step,evidence FROM runs WHERE id=?', (run_id,)).fetchone()
        if row is None:
            raise KeyError(run_id)
        if row[0] != self.version:
            raise ValueError('workflow version mismatch')
        return {'step': row[1], 'evidence': json.loads(row[2]) if row[2] else None}

    def advance(self, run_id, adapter, crash=False):
        with self.db:
            self.db.execute('BEGIN IMMEDIATE')
            state = self.status(run_id)
            if state['step'] == 0:
                evidence = adapter.inspect(run_id)
                if not isinstance(evidence, dict) or not isinstance(evidence.get('summary'), str):
                    raise ValueError('invalid provider evidence')
                self.db.execute('UPDATE runs SET step=1,evidence=? WHERE id=?', (json.dumps(evidence, sort_keys=True), run_id))
            elif state['step'] == 1:
                body = json.dumps({'run': run_id, 'summary': state['evidence']['summary']}, sort_keys=True)
                self.db.execute('INSERT OR IGNORE INTO receipts VALUES (?,?)', (run_id, body))
                self.db.execute('UPDATE runs SET step=2 WHERE id=?', (run_id,))
            if crash:
                raise RuntimeError('injected before transaction commit')
        return self.status(run_id)

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


class FakeInspector:
    """Replace inspect with a read-only provider adapter; replay may call again."""
    def __init__(self):
        self.calls = 0
    def inspect(self, run_id):
        self.calls += 1
        return {'summary': 'Synthetic queue backlog', 'source': 'fixture-1'}


def demo():
    with tempfile.TemporaryDirectory() as folder:
        path = Path(folder) / 'workflow.db'
        provider = FakeInspector()
        flow = Workflow(path)
        flow.start('inc-42')
        flow.advance('inc-42', provider)
        flow.close()
        flow = Workflow(path)
        flow.advance('inc-42', provider)
        flow.advance('inc-42', provider)
        result = {'step': flow.status('inc-42')['step'], 'provider_calls': provider.calls, 'receipts': flow.db.execute('SELECT count(*) FROM receipts').fetchone()[0]}
        flow.close()
        return result


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