"""Reset only known analysis rows, in dependency order; preserve schema and settings."""
from data_safety import digest

TABLES = {
    'active_learning_samples': {'id','event_id','stream_id','created_at'},
    'detections': {'id','frame_id','private_key','created_at'},
    'tracks': {'id','stream_id','timestamp_utc_ms'},
    'events': {'id','stream_id','timestamp_utc_ms'},
    'frames': {'id','stream_id','timestamp_utc_ms'},
    'system_metrics': {'id','timestamp'},
}


def snapshot_cursor(cur, filters):
    from psycopg2 import sql
    cur.execute("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'")
    available = {}
    for r in cur.fetchall(): available.setdefault(r['table_name'],set()).add(r['column_name'])
    result = {}
    for table, required in TABLES.items():
        if table not in available: continue
        if not required <= available[table]: raise ValueError('UNKNOWN_ANALYSIS_SCHEMA')
        where, args = [], []
        if filters.get('camera'):
            if table == 'system_metrics': continue
            # Web identity is stored as a generated stream ID; resolve only from local overlay.
            streams = filters.get('stream_ids', [])
            if not streams: continue
            if table == 'detections': where.append('t.frame_id IN (SELECT id FROM frames WHERE stream_id=ANY(%s))')
            else: where.append('t.stream_id=ANY(%s)')
            args.append(streams)
        if filters.get('before'):
            if 'timestamp_utc_ms' in available[table]: where.append('t.timestamp_utc_ms < EXTRACT(EPOCH FROM %s::timestamptz)*1000')
            else: where.append('t.'+('timestamp' if table=='system_metrics' else 'created_at')+' < %s::timestamptz')
            args.append(filters['before']+'T00:00:00+09:00')
        query = sql.SQL('SELECT row_to_json(t) AS row FROM {} t {} ORDER BY id LIMIT 10001').format(
            sql.Identifier(table),sql.SQL('WHERE '+' AND '.join(where) if where else ''))
        cur.execute(query,args)
        rows = [r['row'] for r in cur.fetchall()]
        if len(rows)>10000: raise ValueError('ANALYSIS_BATCH_LIMIT_FILTER_REQUIRED')
        result[table] = rows
    return result


def snapshot(store, filters):
    with store.transaction() as cur: return snapshot_cursor(cur, filters)


def legacy_plan(store, filters):
    data = snapshot(store, filters)
    n = sum(len(rows) for rows in data.values())
    return [] if not n else [dict(kind='legacy',id='analysis-records',camera=filters.get('camera','*'),date='',
        path='PostgreSQL: managed analysis records',bytes=0,records=n,stamp=digest(data),blocked=None)]


def apply(store, item, filters, started):
    from psycopg2 import sql
    with store.transaction() as cur:
        cur.execute("SELECT to_regclass('public.' || x) AS name FROM unnest(%s::text[]) x", (list(TABLES),))
        names = [r['name'] for r in cur.fetchall() if r['name']]
        for table in names:
            if table not in TABLES: raise ValueError('UNKNOWN_ANALYSIS_TABLE')
            cur.execute(sql.SQL('LOCK TABLE {} IN EXCLUSIVE MODE').format(sql.Identifier(table)))
        # Unknown inbound FKs could cascade outside the allowed records.
        cur.execute("""SELECT c.conrelid::regclass::text AS child FROM pg_constraint c
                       WHERE c.contype='f' AND c.confrelid=ANY(%s::regclass[])""", (names,))
        if any(r['child'] not in TABLES for r in cur.fetchall()): raise ValueError('EXTERNAL_DATABASE_DEPENDENCY')
        data = snapshot_cursor(cur, filters)
        if started and not any(data.values()): return
        if digest(data) != item['stamp']: raise ValueError('ANALYSIS_CHANGED')
        # Avoid implicit ON DELETE CASCADE beyond selected detection rows.
        frames = [r['id'] for r in data.get('frames',[])]
        detections = {r['id'] for r in data.get('detections',[])}
        if frames and 'detections' in data:
            cur.execute('SELECT id FROM detections WHERE frame_id=ANY(%s)',(frames,))
            if any(r['id'] not in detections for r in cur.fetchall()): raise ValueError('DEPENDENT_RECORD_NOT_SELECTED')
        for table in TABLES:
            ids = [r['id'] for r in data.get(table,[])]
            if not ids: continue
            # JSON roundtrips UUIDs as strings; individual parameter inference handles UUID/BIGINT.
            cur.executemany(sql.SQL('DELETE FROM {} WHERE id=%s').format(sql.Identifier(table)), [(i,) for i in ids])