import csv
import fcntl
import hashlib
import json
import os
from pathlib import Path
import secrets
import shutil
import signal
import subprocess
import sys
import tarfile
import time
from urllib.request import Request, build_opener, ProxyHandler
from urllib.error import HTTPError

import yaml

REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
HERE = Path(__file__).parent
ROOT = Path('/workspace/contextra-runtime-config')
ACTIVE = ROOT/'metadata-active/compose.yaml'
ENV = ROOT/'runtime.env'
TOKEN = ROOT/'secure-bridge/control-token'
TARGETS = ('web-cam08-test', 'web-cam01-test')
HEAD = 'b64446185d87d6ab21b45b669bd81c92613fbf22'
CANDIDATE = 'da0149bfe1d3f2bc81c6c3c28889f5c4b4f696c9'
sys.path.insert(0, str(REPO))
from tools.stream_benchmark import load_config
from tools.stream_benchmark_runtime import Runtime, file_hash


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


def read(name):
    return json.loads((HERE/name).read_text())


def cmd(args, timeout=25, stdin=None, cwd=None):
    p = subprocess.run(args, input=stdin, capture_output=True, text=True, timeout=timeout, cwd=cwd)
    if p.returncode:
        save('command-error-'+str(time.time_ns())+'.json', dict(command=args[:4], code=p.returncode,
             stdout=p.stdout[-3000:], stderr=p.stderr[-4000:]))
        raise RuntimeError('Local command failed; private evidence retained')
    return p.stdout


def runtime():
    cfg = load_config(ROOT/'test-run/event-artifact-next.yaml')
    cfg['limits']['container_ram_gib'] = 16
    return Runtime(cfg, HERE)


def compose():
    return ['docker', 'compose', '-p', 'contextra', '--project-directory', str(REPO),
            '--env-file', str(ENV), '-f', str(REPO/'docker-compose.yml'), '-f', str(ACTIVE)]


def ledger():
    query = """SELECT row_to_json(x) FROM (SELECT count(*) AS events,
      coalesce(sum(reservation),0) AS reserved_bytes,
      count(*) FILTER(WHERE reservation>0) AS reserved_jobs,
      count(*) FILTER(WHERE delivery_state IN ('PENDING','RETRY','CLAIMED','SENDING')) AS pending,
      md5(string_agg(md5(t::text),'' ORDER BY event_id)) AS fingerprint
      FROM contextra_event_artifacts t) x;"""
    return json.loads(cmd(['docker','exec','-i','contextra-postgresql-1','sh','-c',
        'exec psql -X -qAt -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB"'], stdin=query))


def other_containers():
    return cmd(['docker','ps','-a','--format','{{.ID}} {{.Names}}']).splitlines()


def artifact_off(rt):
    state = rt.api.call('GET', '/api/v1/metadata/status')['artifacts']
    assert state['enabled'] is False and state['http_enabled'] is True and state['workers'] == 0
    return state


def gate():
    os.chmod(HERE, 0o700)
    rt = runtime(); c = rt.inspect(); states = rt.states()
    assert cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip() == HEAD
    assert c['State']['Running'] and c['Image'] == 'sha256:5ca18768c2a1407baaa0eeebf6ccfa72650754c3da6968d829851d491efd4490'
    assert {sid for sid,q in states.items() if q.get('child_pid')} == set(TARGETS)
    registry = Path(rt.config['registration_file']); specs = yaml.safe_load(registry.read_text())
    for sid, camera in zip(TARGETS, (2,1)):
        s = next(s for s in specs['added'] if s['stream_id'] == sid)
        assert s['camera_id'] == camera and s['source_id'] == str(camera) and s['server_id'] == 'AI-01'
        assert states[sid]['state'] == 'running'
        v = rt.settings(sid)
        assert v['tracking']['enabled'] and v['intrusion']['enabled'] and not v['pose']['enabled'] and not v['fire']['enabled']
        save('before-settings-'+sid+'.json', v)
    save('before-registry.json', specs)
    model_files = [m['Source']+'/config.pbtxt' for m in c['Mounts'] if m['Destination'].startswith('/opt/contextra/models/triton/')]
    save('gate.json', dict(at=time.time(), head=HEAD, candidate=CANDIDATE, image=c['Image'],
        container=c['Id'], mounts=c['Mounts'], states=states, ledger=ledger(), artifacts=artifact_off(rt),
        compose_hash=file_hash(ACTIVE), env_hash=file_hash(ENV), registry_hash=file_hash(registry),
        registry_path=str(registry), model_configs={p:file_hash(p) for p in model_files},
        resources=rt.resources(), other_containers=other_containers(),
        limits=dict(build_seconds=900, disk_gib=15, start_seconds=90, stop_seconds=90,
                    ready_seconds=120, observation_seconds=20, host_available_gib=8,
                    analytics_ram_gib=16, gpu_mib=40000, helper_cpus=2, helper_ram_gib=2)))
    # Private settings snapshots contain existing deployment values, never copied into Git.
    for path, name in ((ACTIVE,'before-compose.yaml'),(ENV,'before-runtime.env')):
        shutil.copy2(path, HERE/name); (HERE/name).chmod(0o600)
    print('Gate PASS: 2 running; Artifact Off; preserved ledger; resource bounds fixed', flush=True)


def build():
    gate = read('gate.json')
    assert not (HERE/'build-intent.json').exists()
    paths = cmd(['git','-C',str(REPO),'diff','--name-only']).splitlines()
    paths += cmd(['git','-C',str(REPO),'ls-files','--others','--exclude-standard']).splitlines()
    hashes = {p:file_hash(REPO/p) for p in paths}
    digest = hashlib.sha256(json.dumps(hashes, sort_keys=True).encode()).hexdigest()[:8]
    tag = 'contextra-analytics:secure-rest-b644461-'+digest
    snapshot = HERE/'snapshot'; snapshot.mkdir()
    p = subprocess.Popen(['git','-C',str(REPO),'archive',HEAD], stdout=subprocess.PIPE)
    with tarfile.open(fileobj=p.stdout, mode='r|') as archive:
        archive.extractall(snapshot, filter='data')
    assert p.wait() == 0
    for path in paths:
        dest = snapshot/path; dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(REPO/path, dest)
    save('build-intent.json', dict(at=time.time(),tag=tag,hashes=hashes,head=HEAD,candidate=CANDIDATE))
    start = time.monotonic()
    with (HERE/'build.log').open('x') as log:
        p = subprocess.Popen(['docker','build','--pull=false','--network=default','-f','docker/Dockerfile',
                              '--target','app','-t',tag,'.'], cwd=snapshot, stdout=log, stderr=log, start_new_session=True)
        while p.poll() is None:
            if time.monotonic()-start > 900 or shutil.disk_usage('/var/lib/docker').free < 15*1024**3:
                os.killpg(p.pid,signal.SIGTERM); p.wait(20)
                raise RuntimeError('Build time/resource bound reached')
            time.sleep(2)
    assert p.returncode == 0
    image = json.loads(cmd(['docker','image','inspect',tag]))[0]['Id']
    save('image.json', dict(tag=tag,id=image,hashes=hashes,seconds=time.monotonic()-start))
    print('Built',tag,image,flush=True)


def fixtures():
    image = read('image.json')
    name = 'contextra-secure-fixture-'+str(os.getpid())
    args = ['docker','run','--rm','--name',name,'--network','none','--cpus','2','--memory','2g',
            '--pids-limit','192','--cap-drop','ALL','--security-opt','no-new-privileges',
            '-e','PYTHONPATH=/opt/contextra:/fixture','-e','PYTHONDONTWRITEBYTECODE=1',
            '-v',str(HERE/'snapshot/tests')+':/fixture/tests:ro',
            '-v',str(HERE/'snapshot/control_sidecar')+':/fixture/control_sidecar:ro',
            '--entrypoint','python3','-w','/opt/contextra',image['id'],'-m','pytest','-q','-p','no:cacheprovider',
            '/fixture/tests/test_secure_bridge.py','/fixture/tests/test_channel_process.py','-k',
            'control_auth or known_error or late_save or pending_command or mapping_preserves or replayed_anchor or stop_cleanup or control_rejects or internal_route or command_owner or artifact_http_only or artifact_attempt_owner or artifact_blocked or unauthorized_relay or relay_returns or relay_transport or secure_factory or first_roi or new_camera or settings_cas_noop_and_save_failure_never_start']
    save('fixture-intent.json', dict(at=time.time(), name=name, image=image['id']))
    try:
        p = subprocess.run(args, capture_output=True, text=True, timeout=150)
        save('fixture-result.json',dict(code=p.returncode,stdout=p.stdout,stderr=p.stderr))
        print(p.stdout[-6000:]+p.stderr[-2000:],flush=True)
        assert p.returncode == 0
    finally:
        subprocess.run(['docker','rm','-f',name],capture_output=True,timeout=15)


def atomic(path, content, expected):
    assert file_hash(path) == expected
    temp = path.with_name(path.name+'.secure-rest-pending')
    with temp.open('x') as handle:
        handle.write(content)
    temp.chmod(0o600)
    assert file_hash(path) == expected
    os.replace(temp,path)


def stop_old():
    rt = runtime(); gate = read('gate.json')
    assert rt.inspect()['Id'] == gate['container']
    save('stop-old-intent.json',dict(at=time.time(),runs={s:rt.states()[s]['run_id'] for s in TARGETS}))
    out = {}
    for sid in TARGETS:
        out[sid] = rt.stop(sid)
    save('stop-old-result.json', out)
    assert not any(q.get('child_pid') for q in rt.states().values())
    cmd(['docker','stop','--time','85',rt.container],100)
    assert not rt.inspect()['State']['Running']
    print('Old two children reaped; Analytics stopped',flush=True)


def reflect():
    gate = read('gate.json'); image=read('image.json'); rt=runtime()
    assert not rt.inspect()['State']['Running']
    assert read('fixture-result.json')['code'] == 0
    assert file_hash(ACTIVE) == gate['compose_hash'] and file_hash(ENV) == gate['env_hash']
    assert file_hash(Path(gate['registry_path'])) == gate['registry_hash']
    for path, h in gate['model_configs'].items(): assert file_hash(path) == h
    save('reflect-intent.json',dict(at=time.time(),image=image['id'],external_auth='UNCONFIGURED'))
    TOKEN.parent.mkdir(mode=0o700,exist_ok=True)
    assert not TOKEN.exists()
    with TOKEN.open('x') as handle: handle.write(secrets.token_urlsafe(32)+'\n')
    TOKEN.chmod(0o600)
    value = yaml.safe_load(ACTIVE.read_text()); service=value['services']['analytics']
    service['image'] = image['id']
    service['environment'].update(CONTEXTRA_SECURE_CONTROL_ENABLED='1',AI_CONTROL_SERVER_ID='AI-01',
        CONTEXTRA_CONTROL_TOKEN_FILE='/run/bridge/control-token',
        CONTEXTRA_EVENT_ARTIFACT_ENABLED='0',CONTEXTRA_ARTIFACT_HTTP_ENABLED='1')
    service['volumes'].append(dict(type='bind',source=str(TOKEN),target='/run/bridge/control-token',
                                  read_only=True,bind=dict(create_host_path=False)))
    atomic(ACTIVE,yaml.safe_dump(value,sort_keys=False),gate['compose_hash'])
    lines=ENV.read_text().splitlines()
    assert sum(s.startswith('CONTEXTRA_IMAGE=') for s in lines)==1
    atomic(ENV,'\n'.join('CONTEXTRA_IMAGE='+image['tag'] if s.startswith('CONTEXTRA_IMAGE=') else s for s in lines)+'\n',gate['env_hash'])
    config=json.loads(cmd(compose()+['config','--format','json']))
    a=config['services']['analytics']; assert a['image']==image['id']
    assert a['environment']['CONTEXTRA_EVENT_ARTIFACT_ENABLED']=='0'
    assert any(int(p['target'])==5000 and p['host_ip']=='127.0.0.1' for p in a['ports'])
    assert not set(config['services']) & {'ai-control-sidecar','ai-gateway'}
    save('compose-check.json',dict(services=list(config['services']),ports=a['ports'],
                                 image=a['image'],external_services='not deployed'))
    cmd(compose()+['up','-d','--no-deps','--no-build','--pull','never','analytics'],180)
    deadline=time.monotonic()+120
    while time.monotonic()<deadline:
        try:
            if rt.api.call('GET','/ready'): break
        except Exception: pass
        time.sleep(2)
    else: raise RuntimeError('Analytics readiness timeout')
    c=rt.inspect(); assert c['Image']==image['id']
    assert not any(q.get('child_pid') for q in rt.states().values())
    relevant={p:h for p,h in image['hashes'].items() if p.startswith('app/')}
    code="import hashlib,json,sys;from pathlib import Path;import app.secure_control as m;print(json.dumps(dict(module=m.__file__,hashes={p:hashlib.sha256(Path(p).read_bytes()).hexdigest() for p in json.loads(sys.argv[1])})))"
    source=json.loads(cmd(['docker','exec','-w','/opt/contextra',rt.container,'python3','-c',code,json.dumps(list(relevant))]))
    assert source['module']=='/opt/contextra/app/secure_control.py' and source['hashes']==relevant
    model=rt.triton('v2/models/RFDETRMedium/config')
    assert model['max_batch_size']==4 and sum(g['count'] for g in model['instance_group'])==8
    assert model['input'][0]['data_type']=='TYPE_FP16'
    assert str(model['version_policy']['specific']['versions'][0])=='4'
    assert {m['Destination']:m['Source'] for m in c['Mounts'] if '/models/' in m['Destination']}=={
        m['Destination']:m['Source'] for m in gate['mounts'] if '/models/' in m['Destination']}
    assert ledger()==gate['ledger']
    save('reflected.json',dict(at=time.time(),image=c['Image'],container=c['Id'],source=source,
        model=model,artifacts=artifact_off(rt),compose_hash=file_hash(ACTIVE),env_hash=file_hash(ENV)))
    print('Reflected: image-owned secure Runtime; authentication enabled; Artifact Off; external services not started',flush=True)


def secure(method,path,body=None,authorized=True):
    headers={'Content-Type':'application/json'}
    if authorized: headers['Authorization']='Bearer '+TOKEN.read_text().strip()
    request=Request('http://127.0.0.1:35000'+path, data=None if body is None else json.dumps(body).encode(), headers=headers,method=method)
    try:
        with build_opener(ProxyHandler({})).open(request,timeout=5) as response:
            return response.status,json.load(response)
    except HTTPError as error:
        return error.code,json.load(error)


def secure_command(sid,action,phase):
    rt=runtime(); original=next(s for s in read('before-registry.json')['added'] if s['stream_id']==sid)
    camera=original['camera_id']; body={'cameraId':camera}
    if action=='start':
        roi=original['intrusion']
        body.update(rtspUrl=original['uri'],startWallTimeMs=original['frame_time_anchor']['startWallTimeMs'],rtpClockRate=90000,
                    scenarios=[dict(scenarioCode=c,enabled=c=='INTRUSION' and roi['enabled'],
                        confidence=roi['confidence'] if c=='INTRUSION' else None,
                        rois=[dict(label=roi['roi_name'],points=roi['roi'])] if c=='INTRUSION' else [])
                        for c in ['FIRE','FALL','INTRUSION','LOITERING','FIGHT','CROWD']])
        for k,v in [('recServerId','rec_server_id'),('recCameraId','rec_camera_id')]:
            if v in original: body[k]=original[v]
    before=rt.states()[sid]
    save(phase+'-intent.json',dict(at=time.time(),sid=sid,action=action,prior_run=before['run_id'],
                               generation=before['run_generation'],request_sha256=hashlib.sha256(json.dumps(body,sort_keys=True).encode()).hexdigest()))
    code,response=secure('POST','/internal/secure-control/'+action,body)
    save(phase+'-response.json',dict(code=code,response=response))
    assert code==202 and response['result']=='COMMAND_ACCEPTED'
    op=response['operationId']; deadline=time.monotonic()+90
    while time.monotonic()<deadline:
        _,operation=secure('GET','/internal/secure-control/operations/'+op)
        if operation['status']=='FAILED': raise RuntimeError('Secure command failed; inspect operation')
        q=rt.states()[sid]
        if q['state']=='error': save(phase+'-error.json',q); raise RuntimeError('Channel error')
        if action=='start' and q['state']=='running' and q.get('metadata_path',{}).get('processed',0)>0:
            assert q['inference_profile']['active_models']=={'det':'RFDETRMedium'}
            assert q['run_id']!=before['run_id']
            save(phase+'-complete.json',dict(state=q,operation=operation)); print('RUNNING',sid,'PID',q['child_pid'],flush=True)
            return q
        if action=='stop' and q.get('child_pid') is None:
            clean=q.get('last_cleanup') or {}
            if clean.get('generation')==before['run_generation'] and clean.get('child_reaped'):
                assert clean.get('completed') and not clean.get('forced') and clean.get('writer_released')
                save(phase+'-complete.json',dict(state=q,operation=operation));print('REAPED',sid,flush=True)
                return q
        time.sleep(.5)
    save(phase+'-timeout.json',dict(state=q,operation=operation))
    raise RuntimeError('Command execution timeout; request not repeated')


def settings_unchanged(rt, sid):
    before=read('before-settings-'+sid+'.json'); after=rt.settings(sid)
    differences=[k for k in before.keys() | after.keys() if before.get(k)!=after.get(k)]
    assert set(differences)<= {'revision'}, differences
    return dict(changed_fields=differences,before_revision=before['revision'],after_revision=after['revision'])


def start_two(prefix=''):
    rt=runtime(); assert artifact_off(rt)
    code,_=secure('POST','/internal/secure-control/start',{'cameraId':2},authorized=False)
    assert code==401 and not any(q.get('child_pid') for q in rt.states().values())
    save(prefix+'unauthorized.json',dict(status=code,children=0))
    for sid in TARGETS: secure_command(sid,'start',prefix+'start-'+sid)
    save(prefix+'settings-parity.json',{sid:settings_unchanged(rt,sid) for sid in TARGETS})
    print('Two first Frames and effective settings preserved',flush=True)


def confirm_helper_fix():
    rt=runtime(); assert not rt.inspect()['State']['Running']
    assert rt.inspect()['Image']==read('image.json')['id']
    assert ledger()==read('gate.json')['ledger']
    save('helper-correction-confirm-intent.json',dict(at=time.time(),reason='revision is spec hash, not effective setting',
                                                   image_unchanged=True,additional_execution=1))
    cmd(['docker','start',rt.container],120)
    deadline=time.monotonic()+120
    while time.monotonic()<deadline:
        try:
            if rt.api.call('GET','/ready'):break
        except Exception:pass
        time.sleep(2)
    else:raise RuntimeError('Readiness timeout')
    save('helper-correction-before-start.json',{sid:settings_unchanged(rt,sid) for sid in TARGETS})
    start_two('confirm-')


def cycle_stop():
    secure_command(TARGETS[0],'stop','cycle-stop')


def cycle_start():
    secure_command(TARGETS[0],'start','cycle-start')


def observe():
    rt=runtime(); rows=[dict(stream_id=sid,source_id=str(c)) for sid,c in zip(TARGETS,(2,1))]
    samples=[]; save('observation-intent.json',dict(at=time.time(),seconds=20,warmup_seconds=3))
    time.sleep(3)
    for i in range(3):
        if i: time.sleep(10)
        sample=rt.sample(rows); samples.append(sample)
        assert artifact_off(rt)
        for q in sample['channels'].values():
            assert q['state']=='running' and not q['error']
        save('sample-'+str(i)+'.json',sample)
    summary={}
    for sid in TARGETS:
        a,b=samples[0]['channels'][sid],samples[-1]['channels'][sid]
        elapsed=samples[-1]['at']-samples[0]['at']
        summary[sid]=dict(result_fps=(b['processed']-a['processed'])/elapsed,
            redis_fps=(b['redis']['entries']-a['redis']['entries'])/(b['redis']['at']-a['redis']['at']),
            queue_max=max(s['channels'][sid]['queue'] for s in samples),
            stale=b['stale']-a['stale'],drop=b['drop']-a['drop'],
            publisher_errors=b['publisher'].get('send_error',0)-a['publisher'].get('send_error',0),
            redis_length=b['redis']['length'],redis_pttl=b['redis']['pttl'],pid=b['pid'])
        assert summary[sid]['stale']==0 and summary[sid]['drop']==0 and summary[sid]['queue_max']<8
        assert summary[sid]['result_fps']>8 and summary[sid]['redis_fps']>8
    states=rt.states()
    assert {sid for sid,q in states.items() if q.get('child_pid')}==set(TARGETS)
    assert len({states[s]['child_pid'] for s in TARGETS})==2
    after=ledger(); assert after==read('gate.json')['ledger']
    original=read('before-registry.json'); now=yaml.safe_load(Path(read('gate.json')['registry_path']).read_text())
    for old in original['added']:
        new=next(s for s in now['added'] if s['stream_id']==old['stream_id'])
        if old['stream_id'] not in TARGETS: assert old==new
        else:
            settings_unchanged(rt,old['stream_id'])
            assert {k:v for k,v in old.items() if k not in ['tracking','secure_control_projection']}=={
                   k:v for k,v in new.items() if k not in ['tracking','secure_control_projection']}
    gate=read('gate.json')
    others=lambda x:sorted(s for s in x if not s.endswith(' contextra-analytics-1'))
    assert others(other_containers())==others(gate['other_containers'])
    save('summary.json',dict(at=time.time(),result='PASS',head=HEAD,candidate=CANDIDATE,image=read('image.json'),
        channels=summary,states=states,artifacts=artifact_off(rt),ledger_unchanged=after==gate['ledger'],
        resources=[s['resource'] for s in samples],seconds=elapsed,other_containers_unchanged=True,
        external_auth='NOT_CONFIGURED',external_services='NOT_DEPLOYED',visual_feedback='USER_PENDING',
        compose_hash=file_hash(ACTIVE),env_hash=file_hash(ENV)))
    with (HERE/'summary.csv').open('x') as handle:
        writer=csv.DictWriter(handle,fieldnames=['stream_id']+list(next(iter(summary.values()))))
        writer.writeheader();writer.writerows(dict(stream_id=k,**v) for k,v in summary.items())
    print(json.dumps(summary,indent=2),flush=True)


def safe_stop():
    rt=runtime(); outcomes={}
    if rt.inspect()['State']['Running']:
        try:
            for sid in TARGETS:
                q=rt.states()[sid]
                if q.get('child_pid') and q['state']!='stopping':
                    outcomes[sid]=rt.stop(sid)
        finally:
            cmd(['docker','stop','--time','85',rt.container],100)
    save('failure-cleanup-'+str(time.time_ns())+'.json',dict(at=time.time(),state=rt.inspect()['State'],outcomes=outcomes))


if __name__=='__main__':
    with (ROOT/'test-run/benchmark.lock').open('a') as lock:
        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
        action=sys.argv[1]
        try:
            globals()[action]()
        except Exception as error:
            save('failure-'+action+'-'+str(time.time_ns())+'.json',dict(at=time.time(),type=type(error).__name__,message=str(error)))
            if action in {'stop_old','reflect','start_two','confirm_helper_fix','cycle_stop','cycle_start','observe'}:
                safe_stop()
            raise