A SQLite lab for approvals, history, conflicts and retries

"""Independent synthetic SQLite lab; no HTTP, real accounts, messages or ERP data.

Actor is a trusted test context, NOT authentication. Never trust client-supplied roles.
Results, document change, approval, event and outbox row commit in one local transaction.
SQLite's writer lock serializes this example; no distributed-system guarantee is implied.
"""
from dataclasses import dataclass
from pathlib import Path
import sqlite3,json,hashlib,re
from contextlib import closing

class ClosingConnection(sqlite3.Connection):
    def __exit__(self,*args):
        try:return super().__exit__(*args)
        finally:self.close()

def packed(value):return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=False,allow_nan=False)
def identifier(value):return isinstance(value,str) and re.fullmatch(r'[A-Za-z0-9_-]{1,40}',value) is not None

@dataclass(frozen=True)
class Actor:
    company_id:str
    user_id:str
    role:str
    active:bool=True

class WorkflowLab:
    def __init__(self,path):
        self.path=str(Path(path))
        if Path(path).exists() and Path(path).stat().st_size:
            with closing(sqlite3.connect(path)) as db:
                marker=db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='synthetic_lab_marker'").fetchone()
                if not marker or db.execute('SELECT version FROM synthetic_lab_marker').fetchall()!=[('synthetic-atomic-v1',)]:raise ValueError('Not a recognized synthetic lab')
        else:
            with self.connect() as db:
                db.executescript('''
                CREATE TABLE synthetic_lab_marker(version TEXT NOT NULL);
                INSERT INTO synthetic_lab_marker VALUES('synthetic-atomic-v1');
                CREATE TABLE documents(company_id TEXT,id TEXT,assigned_to TEXT,version INTEGER,content_revision INTEGER,quantity INTEGER,state TEXT,PRIMARY KEY(company_id,id));
                CREATE TABLE approvals(company_id TEXT,document_id TEXT,round INTEGER,target_revision INTEGER,requester TEXT,status TEXT,reviewer TEXT,reason TEXT,PRIMARY KEY(company_id,document_id,round));
                CREATE TABLE events(sequence INTEGER PRIMARY KEY,company_id TEXT,document_id TEXT,actor TEXT,action TEXT,before_json TEXT,after_json TEXT,reason TEXT);
                CREATE TABLE requests(company_id TEXT,user_id TEXT,request_key TEXT,fingerprint TEXT,result_json TEXT,PRIMARY KEY(company_id,user_id,request_key));
                CREATE TABLE outbox(id INTEGER PRIMARY KEY,company_id TEXT,event_sequence INTEGER,target TEXT,state TEXT,attempts INTEGER DEFAULT 0,UNIQUE(company_id,event_sequence,target));
                ''')
                db.execute("INSERT INTO documents VALUES('A','S01','USER_A',0,0,6,'draft')")
    def connect(self):
        db=sqlite3.connect(self.path,timeout=10,isolation_level=None,factory=ClosingConnection);db.row_factory=sqlite3.Row
        return db
    def _context(self,actor):
        return isinstance(actor,Actor) and actor.active is True and identifier(actor.company_id) and identifier(actor.user_id) and actor.role in ['staff','coordinator','owner']
    def read(self,actor):
        if not self._context(actor):return []
        with self.connect() as db:
            rows=db.execute('SELECT * FROM documents WHERE company_id=? ORDER BY id',[actor.company_id]).fetchall()
            return [dict(r) for r in rows if actor.role!='staff' or r['assigned_to']==actor.user_id]
    def history(self,actor):
        ids={r['id'] for r in self.read(actor)}
        with self.connect() as db:
            return [dict(r) for r in db.execute('SELECT * FROM events WHERE company_id=? ORDER BY sequence',[actor.company_id]) if r['document_id'] in ids]
    def inbox(self,actor):
        if not self._context(actor) or actor.role!='coordinator':return []
        with self.connect() as db:
            return [dict(r) for r in db.execute('SELECT * FROM approvals WHERE company_id=? AND status=? AND requester<>? ORDER BY document_id,round',[actor.company_id,'pending',actor.user_id])]
    def execute(self,actor,request_key,command,*,fail_before_commit=False):
        if not self._context(actor) or not identifier(request_key) or not isinstance(command,dict):return {'ok':False,'reason':'denied'}
        action=command.get('action');keys={'action','document_id','expected_version','reason'}|({'quantity'} if action=='edit' else set())
        if action not in ['edit','submit','approve','reject','execute'] or set(command)!=keys or not identifier(command.get('document_id')) or type(command.get('expected_version')) is not int or command['expected_version']<0 or not isinstance(command.get('reason'),str) or not 1<=len(command['reason'])<=160:return {'ok':False,'reason':'invalid_command'}
        if action=='edit' and (type(command['quantity']) is not int or not 1<=command['quantity']<=99):return {'ok':False,'reason':'invalid_quantity'}
        if actor.role=='owner' or (action in ['approve','reject','execute'] and actor.role!='coordinator'):return {'ok':False,'reason':'denied'}
        fingerprint=hashlib.sha256(packed(command).encode('utf-8')).hexdigest()
        db=self.connect()
        try:
            db.execute('BEGIN IMMEDIATE')
            def stop(reason):db.rollback();return {'ok':False,'reason':reason}
            row=db.execute('SELECT * FROM documents WHERE company_id=? AND id=?',[actor.company_id,command['document_id']]).fetchone()
            if not row or (actor.role=='staff' and row['assigned_to']!=actor.user_id):return stop('denied')
            previous=db.execute('SELECT * FROM requests WHERE company_id=? AND user_id=? AND request_key=?',[actor.company_id,actor.user_id,request_key]).fetchone()
            if previous:
                if previous['fingerprint']!=fingerprint:return stop('request_key_reused')
                result=json.loads(previous['result_json']);db.commit();return {**result,'replayed':True}
            before=dict(row)
            if before['version']!=command['expected_version']:return stop('version_conflict')
            after={**before,'version':before['version']+1}
            if after['version']>9007199254740991:return stop('version_exhausted')
            latest=db.execute('SELECT * FROM approvals WHERE company_id=? AND document_id=? ORDER BY round DESC LIMIT 1',[actor.company_id,row['id']]).fetchone()
            if action=='edit':
                if row['state']=='executed':return stop('state_conflict')
                after.update(quantity=command['quantity'],content_revision=row['content_revision']+1,state='draft')
                db.execute("UPDATE approvals SET status='superseded' WHERE company_id=? AND document_id=? AND status='pending'",[actor.company_id,row['id']])
            elif action=='submit':
                if row['state'] not in ['draft','rejected']:return stop('state_conflict')
                round_number=(latest['round'] if latest else 0)+1
                db.execute('INSERT INTO approvals VALUES(?,?,?,?,?,?,?,?)',[actor.company_id,row['id'],round_number,row['content_revision'],actor.user_id,'pending',None,command['reason']]);after['state']='pending'
            elif action in ['approve','reject']:
                if row['state']!='pending' or not latest or latest['status']!='pending' or latest['target_revision']!=row['content_revision']:return stop('approval_stale')
                if latest['requester']==actor.user_id:return stop('self_approval')
                state='approved' if action=='approve' else 'rejected'
                db.execute('UPDATE approvals SET status=?,reviewer=?,reason=? WHERE company_id=? AND document_id=? AND round=?',[state,actor.user_id,command['reason'],actor.company_id,row['id'],latest['round']]);after['state']=state
            elif action=='execute':
                if row['state']!='approved' or not latest or latest['status']!='approved' or latest['target_revision']!=row['content_revision']:return stop('not_approved')
                after['state']='executed' # Simulation marker only. No network operation.
            db.execute('UPDATE documents SET version=?,content_revision=?,quantity=?,state=? WHERE company_id=? AND id=? AND version=?',[after['version'],after['content_revision'],after['quantity'],after['state'],actor.company_id,row['id'],before['version']])
            event=db.execute('INSERT INTO events(company_id,document_id,actor,action,before_json,after_json,reason) VALUES(?,?,?,?,?,?,?)',[actor.company_id,row['id'],actor.user_id,action,packed(before),packed(after),command['reason']]).lastrowid
            db.execute("INSERT INTO outbox(company_id,event_sequence,target,state) VALUES(?,?,?,'queued')",[actor.company_id,event,'synthetic-reviewer'])
            result={'ok':True,'document':after,'event_sequence':event,'replayed':False}
            db.execute('INSERT INTO requests VALUES(?,?,?,?,?)',[actor.company_id,actor.user_id,request_key,fingerprint,packed(result)])
            if fail_before_commit:raise RuntimeError('Synthetic injected transaction failure')
            db.commit();return result
        except BaseException:
            db.rollback();raise
        finally:db.close()
    def delivery_result(self,actor,outbox_id,outcome):
        if not self._context(actor) or actor.role!='coordinator' or outcome not in ['delivered','unknown','failed']:return False
        with self.connect() as db:
            db.execute('BEGIN IMMEDIATE')
            row=db.execute('SELECT * FROM outbox WHERE company_id=? AND id=?',[actor.company_id,outbox_id]).fetchone()
            if not row or row['state'] in ['delivered','unknown']:db.rollback();return False
            db.execute('UPDATE outbox SET state=?,attempts=attempts+1 WHERE id=?',[outcome,outbox_id]);db.commit();return True
    def outbox(self,actor):
        if not self._context(actor) or actor.role!='coordinator':return []
        with self.connect() as db:return [dict(r) for r in db.execute('SELECT * FROM outbox WHERE company_id=? ORDER BY id',[actor.company_id])]

def command(action,version,**values):
    return {'action':action,'document_id':'S01','expected_version':version,'reason':'Synthetic review',**values}