#!/usr/bin/env python3
import fcntl,json,sqlite3,subprocess,sys,time
from datetime import datetime,timezone
from pathlib import Path
DB=Path('/opt/struktur/youtube-research/knowledge.db')
WRITER='/opt/struktur/obsidian-graphiti-import/aa043/aa043_single_writer.py'
LOCK=Path('/run/graphiti-control/aa049-queue-worker.lock')
LOG=Path('/opt/struktur/reports/aa049-worker-events.jsonl')
def now(): return datetime.now(timezone.utc)
def due(v):
    if not v: return True
    try:
        x=datetime.fromisoformat(v.replace('Z','+00:00'))
        if x.tzinfo is None: x=x.replace(tzinfo=timezone.utc)
        return x<=now()
    except Exception: return False
def pick():
    c=sqlite3.connect(f'file:{DB}?mode=ro',uri=True); c.row_factory=sqlite3.Row
    rows=c.execute("select id,video_id,graphiti_status,next_attempt_at,graphiti_release from graphiti_import_queue where source_type='youtube' and video_id is not null and graphiti_status in ('queued','retry_wait') order by id").fetchall(); c.close()
    for r in rows:
        if (r['graphiti_release'] is None or int(r['graphiti_release'])==0) and due(r['next_attempt_at']): return dict(r)
    return None
def log(x):
    x={'ts_utc':now().isoformat(),**x}; LOG.parent.mkdir(parents=True,exist_ok=True); LOG.open('a').write(json.dumps(x,ensure_ascii=False)+'\n'); print(json.dumps(x,ensure_ascii=False),flush=True)
LOCK.parent.mkdir(parents=True,exist_ok=True)
with LOCK.open('w') as f:
    try: fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB)
    except BlockingIOError: log({'event':'SKIP_LOCK_HELD'}); sys.exit(0)
    row=pick()
    if not row: log({'event':'NO_DUE_YOUTUBE'}); sys.exit(0)
    log({'event':'START','queue_id':row['id'],'video_id':row['video_id'],'status':row['graphiti_status'],'next_attempt_at':row['next_attempt_at']})
    p=subprocess.run(['/opt/struktur/youtube-research/venv/bin/python',WRITER,'--queue-id',str(row['id'])],capture_output=True,text=True,timeout=1100)
    log({'event':'END','queue_id':row['id'],'returncode':p.returncode,'stdout':p.stdout[-2000:],'stderr':p.stderr[-2000:]})
    # A failure is source-local; the next timer invocation must remain available.
    sys.exit(0)