import fcntl
import hashlib
import json
from pathlib import Path
import re
import subprocess
import sys
import time

HERE = Path(__file__).parent
ROOT = Path('/data/contextra/artifacts')
PRIOR = Path('/tmp/codex/ltb1021/2026-09-17_contextra-event-ai01_r01')


def cmd(args, stdin=None):
    return subprocess.run(args, input=stdin, text=True, capture_output=True,
                          check=True, timeout=20).stdout


def pg(sql):
    return cmd(['docker', 'exec', '-i', 'contextra-postgresql-1', 'sh', '-c',
                'exec psql -X -qAt -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB"'], sql)


def save(name, value):
    with (HERE / name).open('x') as file:
        json.dump(value, file, indent=2)
    (HERE / name).chmod(0o600)


def inventory():
    container = json.loads(cmd(['docker', 'inspect', 'contextra-analytics-1']))[0]
    assert not container['State']['Running'] and container['State']['Pid'] == 0
    assert not ROOT.is_symlink()
    rows = json.loads(pg("""SELECT coalesce(json_agg(x),'[]') FROM (
        SELECT event_id,relative_path,payload->>'cameraId' AS camera,
        payload->>'streamId' AS stream,artifact_state,delivery_state,reservation,
        md5(t::text) AS fingerprint FROM contextra_event_artifacts t ORDER BY event_id) x;"""))
    historical = json.loads((PRIOR / 'plan.json').read_text())['old_events']
    historical += json.loads((PRIOR / 'summary.json').read_text())['events']
    ids = {row['event_id'] for row in historical}
    assert len(ids) == 174 and {row['event_id'] for row in rows} == ids
    allowed = []
    for row in rows:
        assert (row['stream'], row['camera']) in {
            ('web-cam08-test', '2'), ('web-cam08-test', '77'),
            ('web-cam01-test', '1'), ('web-cam01-test', '78')}
        assert re.fullmatch(r'[a-zA-Z0-9_-]+', row['event_id'])
        relative = Path(row['relative_path'])
        assert len(relative.parts) == 4 and relative.parts[:2] == ('cameras', row['camera'])
        assert relative.name.endswith('_' + row['event_id']) and '..' not in relative.parts
        allowed += [ROOT / relative, ROOT / '.staging' / row['event_id']]
    files, dirs = [], []
    for path in ROOT.rglob('*'):
        assert not path.is_symlink(), str(path)
        if path.is_dir():
            dirs.append(str(path.relative_to(ROOT)))
        elif path.is_file() and path.name not in ('.worker.lock', 'README.ko.md'):
            assert any(parent == path or parent in path.parents for parent in allowed), str(path)
            stat = path.stat()
            files.append(dict(path=str(path.relative_to(ROOT)), size=stat.st_size,
                              mtime_ns=stat.st_mtime_ns, inode=stat.st_ino))
        else:
            assert path in (ROOT / '.worker.lock', ROOT / 'README.ko.md'), str(path)
    return dict(container_id=container['Id'], rows=rows, files=sorted(files, key=lambda x:x['path']),
                dirs=sorted(dirs), bytes=sum(f['size'] for f in files),
                readme_sha256=hashlib.sha256((ROOT/'README.ko.md').read_bytes()).hexdigest())


with (ROOT / '.worker.lock').open('r+') as lock:
    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    current = inventory()
    if sys.argv[1] == 'inventory':
        save('inventory.json', current)
        print(json.dumps(dict(events=len(current['rows']), files=len(current['files']),
                              bytes=current['bytes'], held=sum(r['delivery_state']=='HELD' for r in current['rows']))))
    elif sys.argv[1] == 'apply':
        assert current == json.loads((HERE / 'inventory.json').read_text())
        assert not (HERE/'intent.json').exists()
        save('intent.json', dict(at=time.time(), action='Delete approved test Event rows and files; no restart'))
        values = ','.join("('%s','%s')" % (r['event_id'],r['fingerprint']) for r in current['rows'])
        pg("""BEGIN;
        SET LOCAL lock_timeout='2s'; SET LOCAL statement_timeout='10s';
        LOCK TABLE contextra_event_artifacts IN EXCLUSIVE MODE;
        CREATE TEMP TABLE cleanup_expected (event_id text PRIMARY KEY,fingerprint text) ON COMMIT DROP;
        INSERT INTO cleanup_expected VALUES """ + values + """;
        DO $$ BEGIN
          IF (SELECT count(*) FROM contextra_event_artifacts) <> 174 OR
             (SELECT count(*) FROM contextra_event_artifacts t JOIN cleanup_expected e
              ON t.event_id=e.event_id AND md5(t::text)=e.fingerprint) <> 174
          THEN RAISE EXCEPTION 'Event ledger drift; no deletion'; END IF;
        END $$;
        DELETE FROM contextra_event_artifacts t USING cleanup_expected e WHERE t.event_id=e.event_id;
        COMMIT;""")
        for item in current['files']:
            path = ROOT / item['path']
            stat = path.stat()
            assert not path.is_symlink() and (stat.st_size,stat.st_mtime_ns,stat.st_ino) == (item['size'],item['mtime_ns'],item['inode'])
            path.unlink()
        for relative in sorted(current['dirs'], key=lambda p:len(Path(p).parts), reverse=True):
            path = ROOT / relative
            if path not in (ROOT/'cameras', ROOT/'.staging'):
                path.rmdir()
        size = sum(p.stat().st_size for p in ROOT.rglob('*') if p.is_file())
        pg('UPDATE contextra_artifact_storage SET observed_bytes=%d,observed_at=%f WHERE id=1;' % (size,time.time()))
        assert pg('SELECT count(*) FROM contextra_event_artifacts;').strip() == '0'
        assert not list((ROOT/'cameras').iterdir()) and not list((ROOT/'.staging').iterdir())
        assert hashlib.sha256((ROOT/'README.ko.md').read_bytes()).hexdigest() == current['readme_sha256']
        state = json.loads(cmd(['docker','inspect','contextra-analytics-1']))[0]['State']
        assert not state['Running'] and state['Pid']==0
        result = dict(events_deleted=len(current['rows']),files_deleted=len(current['files']),
                      bytes_removed=current['bytes'],remaining_events=0,held=0,reserved_jobs=0,
                      analytics='stopped',readme_preserved=True,historical_evidence_preserved=True,
                      redis_untouched=True,web_records_untouched=True)
        save('result.json', result)
        print(json.dumps(result))
    else:
        raise ValueError('inventory or apply required')