import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import time
import urllib.request
import urllib.error
import yaml

HERE=Path(__file__).parent
REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
ROOT=Path('/workspace/contextra-runtime-config')
sys.path.insert(0,str(REPO))
from tools.stream_benchmark import load_config
from tools.stream_benchmark_runtime import Runtime
os.umask(0o077)
cfg=load_config(ROOT/'test-run/event-artifact-next.yaml')
rt=Runtime(cfg,HERE)
TARGETS=('web-cam08-test','web-cam01-test')
def save(name,data):
    with (HERE/name).open('x') as f:json.dump(data,f,indent=2,ensure_ascii=False)
def cmd(args,timeout=120,input=None):
    p=subprocess.run(args,input=input,text=True,capture_output=True,timeout=timeout)
    if p.returncode:
        save('deploy-error-'+str(time.time_ns())+'.json',dict(code=p.returncode,stdout=p.stdout[-2000:],stderr=p.stderr[-3000:]))
        raise RuntimeError('DEPLOY_COMMAND_FAILED')
    return p.stdout
def compose():
    return ['docker','compose','-p','contextra','--project-directory',str(REPO),
        '--env-file',str(ROOT/'runtime.env'),'--env-file',str(ROOT/'secure-bridge.env'),
        '-f',str(REPO/'docker-compose.yml'),'-f',str(ROOT/'metadata-active/compose.yaml'),
        '-f',str(REPO/'deploy/secure-bridge.compose.yml')]
def ledger():
    sql="SELECT json_build_object('count',count(*),'fingerprint',md5(string_agg(md5(t::text),'' ORDER BY event_id))) FROM contextra_event_artifacts t;"
    return json.loads(cmd(['docker','exec','-i','contextra-postgresql-1','sh','-c',
        'exec psql -X -qAt -U "$POSTGRES_USER" -d "$POSTGRES_DB"'],input=sql))
def secure(action,body):
    headers={'Authorization':'Bearer '+(ROOT/'secure-bridge/control-token').read_text().strip(),
             'Content-Type':'application/json'}
    req=urllib.request.Request('http://127.0.0.1:35000/internal/secure-control/'+action,
                               data=json.dumps(body).encode(),headers=headers,method='POST')
    with urllib.request.urlopen(req,timeout=5) as r:return r.status,json.load(r)

if sys.argv[1]=='reflect':
    assert json.loads((HERE/'enroll-after-manager-fix-result.json').read_text())['code']==0
    assert json.loads((HERE/'release-fixture.json').read_text())['code']==0
    assert not (HERE/'reflect-intent.json').exists()
    status=rt.api.call('GET','/api/v1/metadata/status')['artifacts']
    assert not status['enabled'] and status['workers']==0
    states=rt.states()
    assert {s for s,v in states.items() if v.get('child_pid')}==set(TARGETS)
    registry=yaml.safe_load(Path(cfg['registration_file']).read_text())
    save('deploy-before.json',dict(ledger=ledger(),settings={s:rt.settings(s) for s in TARGETS},
        registry=registry,states=states,container=rt.inspect()['Id']))
    save('reflect-intent.json',dict(targets=TARGETS,at=time.time()))
    for sid in TARGETS:
        print('STOP',sid,flush=True)
        result=rt.stop(sid)
        save('deployment-stop-'+sid+'.json',result)
        assert not rt.states()[sid].get('child_pid')
    assert not any(q.get('child_pid') for q in rt.states().values())
    cmd(['docker','stop','--time','85',rt.container],100)
    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('RUNTIME_READY_TIMEOUT')
    assert rt.inspect()['Image']==json.loads((HERE/'analytics-image.json').read_text())['id']
    assert not any(q.get('child_pid') for q in rt.states().values())
    assert ledger()==json.loads((HERE/'deploy-before.json').read_text())['ledger']
    cmd(compose()+['up','-d','--no-deps','--no-build','--pull','never','ai-control-sidecar','ai-gateway'],120)
    save('reflect-complete.json',dict(image=rt.inspect()['Image'],ledger=ledger()))
    print('CANDIDATES_DEPLOYED_CHANNELS_STOPPED',flush=True)
elif sys.argv[1]=='start':
    before=json.loads((HERE/'deploy-before.json').read_text())
    for sid in TARGETS:
        spec=next(s for s in before['registry']['added'] if s['stream_id']==sid)
        intrusion=spec['intrusion']
        body=dict(cameraId=spec['camera_id'],rtspUrl=spec['uri'],
            startWallTimeMs=spec['frame_time_anchor']['startWallTimeMs'],rtpClockRate=90000,
            scenarios=[dict(scenarioCode=c,enabled=c=='INTRUSION' and intrusion['enabled'],
                confidence=intrusion['confidence'] if c=='INTRUSION' else None,
                rois=[dict(label=intrusion['roi_name'],points=intrusion['roi'])] if c=='INTRUSION' else [])
                for c in ('FIRE','FALL','INTRUSION','LOITERING','FIGHT','CROWD')])
        save('start-'+sid+'-intent.json',dict(at=time.time(),prior_run=rt.states()[sid]['run_id']))
        code,response=secure('start',body)
        save('start-'+sid+'-response.json',dict(code=code,response=response))
        assert code==202 and response['result']=='COMMAND_ACCEPTED'
        deadline=time.monotonic()+90
        while time.monotonic()<deadline:
            state=rt.states()[sid]
            if state['state']=='error':raise RuntimeError('CHANNEL_START_FAILED')
            if state['state']=='running' and state.get('metadata_path',{}).get('processed',0)>0:
                save('start-'+sid+'-first-frame.json',state)
                break
            time.sleep(.5)
        else:raise RuntimeError('FIRST_FRAME_TIMEOUT')
        assert state['inference_profile']['active_models']=={'det':'RFDETRMedium'}
        after=rt.settings(sid)
        assert all(v==after.get(k) for k,v in before['settings'][sid].items() if k!='revision')
        print('FIRST_FRAME',sid,'PID',state['child_pid'],flush=True)
    assert ledger()==before['ledger']
    save('start-complete.json',dict(ledger=ledger(),artifacts=rt.api.call('GET','/api/v1/metadata/status')['artifacts']))