"""CTCLR-04: isolated maintenance worker, no Runtime startup or external Event calls."""
from contextlib import ExitStack, contextmanager
from datetime import datetime, timezone, timedelta
import json
import os
from pathlib import Path
import re
import shutil
import sqlite3
import sys
import time
import traceback

from data_safety import atomic, child, digest, lock, remove_training, safe, tree_size

TERMINAL = {'SENT', 'FAILED', 'EXPIRED', 'SKIPPED', 'BLOCKED_TIME'}
LIMIT = 10000


def day(stamp):
    return datetime.fromtimestamp(float(stamp), timezone(timedelta(hours=9))).date().isoformat()


def eligible_event(row):
    return (not row['reservation'] and row['claim_token'] is None
            and row['artifact_state'] in ('READY', 'FAILED', 'EXPIRED')
            and row['delivery_state'] in TERMINAL
            and row['media_retention_state'] in ('PRESENT', 'DELETED', 'DELETING'))


def stable_event(row):
    return digest({k:v for k,v in row.items() if k not in
        ('media_retention_state','media_retention_at','media_retention_error','stored_bytes')})


def stable_training(row):
    return digest({k:v for k,v in row.items() if k != 'state'})


def selected(row, filters):
    return ((not filters.get('camera') or str(row.get('camera')) == filters['camera'])
            and (not filters.get('before') or row.get('date', '') < filters['before'])
            and (not filters.get('ids') or row['id'] in filters['ids']))


class Worker:
    def __init__(self, request):
        self.request = request
        self.action = request['kind']
        if self.action not in ('data', 'artifact', 'training', 'db', 'camera', 'redis'):
            raise ValueError('UNKNOWN_KIND')
        self.job = safe('/ct-job')
        self.artifact = Path(os.environ.get('CONTEXTRA_ARTIFACT_ROOT', '/data/contextra/artifacts'))
        self.training = Path(os.environ['CONTEXTRA_TRAINING_ROOT'])
        self.camera_path = safe(os.environ['STREAMS_RUNTIME_PATH'])
        self.filters = request.get('filters', {})
        self.store = None
        self.stack = ExitStack()

    def __enter__(self):
        try:
            if self.action in ('data', 'artifact', 'db'):
                self.stack.enter_context(lock(self.artifact / '.worker.lock'))
                self.stack.enter_context(lock(self.artifact / '.retention.lock'))
            if self.action in ('data', 'training', 'db'):
                self.stack.enter_context(lock(self.training / '.worker-lock'))
                self.stack.enter_context(lock(self.training / '.lock'))
            if self.action in ('camera', 'redis'):
                self.stack.enter_context(lock(self.camera_path.parent / 'channel-parent.lock'))
            return self
        except BaseException:
            self.stack.close()
            raise

    def __exit__(self, *args):
        if self.store: self.store.close()
        self.stack.close()

    def events(self):
        if self.store is None:
            from app.event_artifact_contract import ArtifactConfig
            from app.event_artifact_store import ArtifactStore
            cfg = ArtifactConfig.from_environment()
            if cfg is None: raise ValueError('ARTIFACT_CONFIGURATION_REQUIRED')
            self.store = ArtifactStore(cfg)
        return self.store

    @contextmanager
    def train_db(self, writable=False):
        path = safe(self.training / 'index.sqlite')
        if not path.is_file(): raise ValueError('TRAINING_INDEX_MISSING')
        db = sqlite3.connect(path.as_uri() + ('?mode=rw' if writable else '?mode=ro'), uri=True, timeout=3)
        db.row_factory = sqlite3.Row
        try:
            with db:yield db
        finally:db.close()

    def cameras(self):
        import yaml
        doc = yaml.safe_load(self.camera_path.read_text())
        if not isinstance(doc, dict) or set(doc) != {'added', 'removed'}:
            raise ValueError('CAMERA_OVERLAY_SCHEMA')
        return doc

    def event_rows(self):
        from app.event_artifact_contract import relative_directory
        with self.events().transaction() as cur:
            clauses=[];args=[]
            if self.filters.get('camera'):
                clauses.append("payload->>'cameraId'=%s");args.append(self.filters['camera'])
            if self.filters.get('before'):
                clauses.append('COALESCE(confirmed_at,created_at)<EXTRACT(EPOCH FROM %s::timestamptz)')
                args.append(self.filters['before']+'T00:00:00+09:00')
            where=' WHERE '+' AND '.join(clauses) if clauses else ''
            cur.execute('SELECT * FROM contextra_event_artifacts'+where+' ORDER BY confirmed_at,event_id LIMIT %s', (*args,LIMIT+1))
            rows = cur.fetchall()
        if len(rows) > LIMIT: raise ValueError('BATCH_LIMIT_FILTER_REQUIRED')
        for r in rows:
            item = dict(kind='artifact', id=r['event_id'], camera=str(r['payload']['cameraId']),
                        date=day(r['confirmed_at'] or r['created_at']), path=r['relative_path'],
                        bytes=r['stored_bytes'], stamp=digest(r), stable=stable_event(r), identity=digest(r['payload']), blocked=None)
            if not selected(item, self.filters): continue
            if not eligible_event(r): item['blocked'] = 'INFLIGHT_OR_PROTECTED'
            elif r['relative_path'] != relative_directory(r['payload']): item['blocked'] = 'OWNERSHIP'
            else:
                p = child(self.artifact, r['relative_path'])
                try:
                    if p.exists(): item['bytes'] = tree_size(self.artifact, p)
                    elif r['media_retention_state'] != 'DELETED': item['blocked'] = 'MISSING_MEDIA'
                except (OSError, ValueError): item['blocked'] = 'UNSAFE_BUNDLE'
            if self.action != 'db' and r['media_retention_state'] == 'DELETED': continue
            yield item

    def training_rows(self):
        from app.training_store import MARKER
        with self.train_db() as db:
            clauses=[];args=[]
            if self.filters.get('camera'):
                clauses.append('camera=?');args.append(self.filters['camera'])
            if self.filters.get('before'):
                clauses.append('created<?');args.append(datetime.fromisoformat(self.filters['before']+'T00:00:00+09:00').timestamp())
            where=' WHERE '+' AND '.join(clauses) if clauses else ''
            rows = db.execute('SELECT * FROM jobs'+where+' ORDER BY created,id LIMIT ?', (*args,LIMIT+1)).fetchall()
        if len(rows) > LIMIT: raise ValueError('BATCH_LIMIT_FILTER_REQUIRED')
        for raw in rows:
            r = dict(raw)
            item = dict(kind='training', id=r['id'], camera=str(r['camera']), date=day(r['created']),
                        path=r['path'], bytes=0, stamp=digest(r), stable=stable_training(r), blocked=None)
            if not selected(item, self.filters): continue
            if r['state'] not in ('READY', 'DELETED'): item['blocked'] = 'INFLIGHT_OR_UNOWNED'
            elif r['state'] == 'READY':
                try:
                    p = child(self.training, r['path'])
                    meta = json.loads(safe(p/'metadata.json').read_text())
                    if meta.get('owner') != MARKER or meta.get('jobId') != r['id']: raise ValueError()
                    item['bytes'] = tree_size(self.training, p)
                except (OSError, ValueError): item['blocked'] = 'UNSAFE_OR_MISSING_BUNDLE'
            if self.action != 'db' and r['state'] == 'DELETED': continue
            yield item

    def plan(self):
        rows = []
        if self.action in ('data', 'artifact', 'db'): rows.extend(self.event_rows())
        if self.action in ('data', 'training', 'db'): rows.extend(self.training_rows())
        if self.action == 'camera':
            for r in self.cameras()['added']:
                item = dict(kind='camera', id=r['stream_id'], camera=str(r.get('camera_id', r.get('source_id', ''))),
                            date='', path=str(self.camera_path), bytes=0, stamp=digest(r), blocked=None)
                if not r.get('camera_id') or r.get('source_type') != 'rtsp': item['blocked'] = 'NOT_WEB_CAMERA'
                if selected(item, self.filters): rows.append(item)
        if self.action == 'redis':
            from data_worker_redis import plan
            rows.extend(plan(self.cameras()['added'], self.filters))
        if self.action == 'db':
            from data_worker_sql import legacy_plan
            rows.extend(legacy_plan(self.events(), self.filters))
        result = dict(version=1, kind=self.action, filters=self.filters, rows=rows)
        result['hash'] = digest(result)
        return result

    def event_apply(self, item, resetting, started):
        store = self.events()
        with store.transaction() as cur:
            cur.execute('SELECT * FROM contextra_event_artifacts WHERE event_id=%s FOR UPDATE', (item['id'],))
            row = cur.fetchone()
            if row is None:
                if resetting and started: return
                raise ValueError('EVENT_MISSING')
            if not eligible_event(row): raise ValueError('EVENT_PROTECTED')
            if row['relative_path'] != item['path']: raise ValueError('EVENT_PATH_CHANGED')
            if digest(row['payload']) != item['identity']: raise ValueError('EVENT_IDENTITY_CHANGED')
            if stable_event(row) != item['stable']: raise ValueError('EVENT_CHANGED')
            if not started and digest(row) != item['stamp']: raise ValueError('EVENT_CHANGED')
            if row['media_retention_state'] != 'DELETED':
                cur.execute('SELECT observed_at FROM contextra_artifact_storage WHERE id=1 FOR UPDATE')
                row = dict(row, observed_at=cur.fetchone()['observed_at'])
                cur.execute("UPDATE contextra_event_artifacts SET media_retention_state='DELETING' WHERE event_id=%s", (item['id'],))
                cur.execute('UPDATE contextra_artifact_storage SET accounting_revision=accounting_revision+1 WHERE id=1')
        if row['media_retention_state'] != 'DELETED':
            from app.artifact_retention import _remove
            _remove(store, row)
        if resetting:
            with store.transaction() as cur:
                cur.execute("DELETE FROM contextra_event_artifacts WHERE event_id=%s AND media_retention_state='DELETED' AND reservation=0 AND claim_token IS NULL", (item['id'],))
                cur.execute('''DELETE FROM contextra_artifact_channels c WHERE c.channel_key=%s
                    AND c.reserved_jobs=0 AND c.render_token IS NULL AND c.send_token IS NULL
                    AND NOT EXISTS (SELECT 1 FROM contextra_event_artifacts e WHERE e.channel_key=c.channel_key)''', (row['channel_key'],))

    def training_apply(self, item, resetting, started):
        with self.train_db(True) as db:
            row = db.execute('SELECT * FROM jobs WHERE id=?', (item['id'],)).fetchone()
            if row is None:
                if resetting and started: return
                raise ValueError('TRAINING_MISSING')
            r = dict(row)
            if r['state'] not in ('READY', 'DELETED') or r['path'] != item['path']:
                raise ValueError('TRAINING_PROTECTED')
            if not started and digest(r) != item['stamp']: raise ValueError('TRAINING_CHANGED')
            if stable_training(r) != item['stable']: raise ValueError('TRAINING_CHANGED')
            if r['state'] != 'DELETED': remove_training(self.training, r['path'], r['id'])
            if resetting:
                db.execute('DELETE FROM jobs WHERE id=?', (r['id'],))
                db.execute('DELETE FROM cameras WHERE camera=? AND NOT EXISTS (SELECT 1 FROM jobs WHERE camera=?)', (r['camera'], r['camera']))
            else: db.execute("UPDATE jobs SET state='DELETED' WHERE id=?", (r['id'],))

    def camera_apply(self, item, started):
        import yaml
        doc = self.cameras()
        matches = [r for r in doc['added'] if r['stream_id'] == item['id']]
        if not matches and started: return
        if len(matches) != 1 or digest(matches[0]) != item['stamp']: raise ValueError('CAMERA_CHANGED')
        doc['added'] = [r for r in doc['added'] if r['stream_id'] != item['id']]
        atomic(self.camera_path, yaml.safe_dump(doc, allow_unicode=True).encode())

    def backup(self, plan):
        # Backups never go to the public outbox. Payloads can contain input credentials.
        if (self.job/'backup-complete.json').exists():
            receipt=json.loads((self.job/'backup-complete.json').read_text())
            raw=(self.job/'records.backup.json').read_bytes()
            if receipt['sha256'] != __import__('hashlib').sha256(raw).hexdigest():raise ValueError('BACKUP_CHANGED')
            return
        output = {}
        for item in plan['rows']:
            if item['blocked']: continue
            if item['kind'] == 'artifact':
                with self.events().transaction() as cur:
                    cur.execute('SELECT * FROM contextra_event_artifacts WHERE event_id=%s', (item['id'],))
                    row = cur.fetchone()
                if row is None or digest(row) != item['stamp']: raise ValueError('BACKUP_EVENT_CHANGED')
                output[item['kind']+':'+item['id']] = row
            elif item['kind'] == 'training':
                with self.train_db() as db:
                    row = db.execute('SELECT * FROM jobs WHERE id=?', (item['id'],)).fetchone()
                    timing = db.execute('SELECT * FROM cameras WHERE camera=?', (item['camera'],)).fetchone()
                if row is None or digest(dict(row)) != item['stamp']: raise ValueError('BACKUP_TRAINING_CHANGED')
                output['training:'+item['id']] = dict(job=dict(row), timing=dict(timing) if timing else None)
            elif item['kind'] == 'legacy':
                from data_worker_sql import snapshot
                data = snapshot(self.events(), self.filters)
                if digest(data) != item['stamp']: raise ValueError('BACKUP_LEGACY_CHANGED')
                output['legacy'] = data
            elif item['kind'] == 'camera': output['camera_overlay'] = self.cameras()
        raw = json.dumps(output, default=str).encode()
        if len(raw) > 128*1024**2: raise ValueError('BACKUP_SIZE_LIMIT')
        if shutil.disk_usage(self.job).free < len(raw)*2 + 15*1024**3: raise ValueError('BACKUP_SPACE_LOW')
        atomic(self.job/'records.backup.json', raw)
        atomic(self.job/'backup-complete.json', {'sha256': __import__('hashlib').sha256(raw).hexdigest(), 'bytes':len(raw)})

    def apply(self):
        plan = json.loads((self.job/'plan.json').read_text())
        expected = plan.pop('hash')
        if expected != digest(plan) or expected != self.request['plan_hash']: raise ValueError('PLAN_CHANGED')
        plan['hash'] = expected
        if plan['kind'] != self.action or plan['filters'] != self.filters: raise ValueError('PLAN_SCOPE')
        state_path = self.job/'progress.json'
        state = json.loads(state_path.read_text()) if state_path.exists() else {'started':[], 'done':[], 'backup':not self.request.get('no_backup', False)}
        if state['backup'] != (not self.request.get('no_backup',False)):raise ValueError('BACKUP_MODE_CHANGED')
        resetting = self.action in ('db', 'camera')
        if resetting and state['backup']: self.backup(plan)
        deadline = time.monotonic() + 180
        for item in plan['rows']:
            key = item['kind']+':'+item['id']
            if item['blocked'] or key in state['done']: continue
            if time.monotonic() > deadline: raise TimeoutError('RESUME_SAME_JOB')
            started = key in state['started']
            if not started: state['started'].append(key); atomic(state_path, state)
            if item['kind'] == 'artifact': self.event_apply(item, resetting, started)
            elif item['kind'] == 'training': self.training_apply(item, resetting, started)
            elif item['kind'] == 'camera': self.camera_apply(item, started)
            elif item['kind'] == 'legacy':
                from data_worker_sql import apply
                apply(self.events(), item, self.filters, started)
            elif item['kind'] == 'redis':
                from data_worker_redis import apply
                apply(item, started)
            state['done'].append(key); atomic(state_path, state)
        if self.store: self.store.scan_storage()
        state['complete'] = True; atomic(state_path, state)
        return dict(complete=True, deleted=len(state['done']), protected=sum(bool(x['blocked']) for x in plan['rows']))


def safe_error(error):
    result = {'error': type(error).__name__,
              'code': str(error) if re.fullmatch('[A-Z_]+', str(error)) else 'MAINTENANCE_FAILED'}
    if isinstance(error, OSError) and isinstance(error.errno, int):
        result['errno'] = error.errno
    pgcode = getattr(error, 'pgcode', None)
    if isinstance(pgcode, str) and re.fullmatch('[A-Z0-9]{5}', pgcode):
        result['sqlstate'] = pgcode
    frames = traceback.extract_tb(error.__traceback__)
    if frames:
        frame = frames[-1]
        result['location'] = Path(frame.filename).name + ':' + str(frame.lineno)
    return result


def main():
    os.umask(0o077)
    request = json.load(sys.stdin)
    if request.get('operation') not in ('plan', 'apply'): raise ValueError('OPERATION')
    with Worker(request) as worker:
        result = worker.plan() if request['operation'] == 'plan' else worker.apply()
    print(json.dumps(result, ensure_ascii=False, default=str))


if __name__ == '__main__':
    try: main()
    except Exception as e:
        # No SQL/DSN/credential-bearing path from exception text is printed.
        print(json.dumps(safe_error(e)))
        sys.exit(2)