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

REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
HERE=Path(__file__).parent
ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml')
ENV=Path('/workspace/contextra-runtime-config/runtime.env')
CONFIG=Path('/workspace/contextra-runtime-config/test-run/artifact-parallel16.yaml')
TARGETS=('web-cam08-test','web-cam01-test')
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 f:json.dump(value,f,indent=2)
    (HERE/name).chmod(0o600)

def command(args,timeout=20,stdin=None):
    p=subprocess.run(args,input=stdin,text=True,capture_output=True,timeout=timeout)
    if p.returncode:raise RuntimeError('Command failed: '+p.stderr[-1500:])
    return p.stdout

def pg(query):
    return command(['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 ledger():
    return json.loads(pg("""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;"""))

rt=Runtime(load_config(CONFIG),HERE)
before=rt.inspect()
assert not before['State']['Running'] and before['State']['Pid']==0
assert before['Image']==rt.config['expected_image']
assert shutil.disk_usage('/var/lib/docker').free>15*1024**3
specs=yaml.safe_load(Path(rt.config['registration_file']).read_text())['added']
assert all(not s.get('autostart') for s in specs)
for sid,camera in zip(TARGETS,('2','1')):
    spec=next(s for s in specs if s['stream_id']==sid)
    assert spec['source_id']==camera and spec['server_id']=='AI-01'
    assert spec['intrusion']['enabled']
before_ledger=ledger();compose_hash=file_hash(ACTIVE)
save('plan.json',dict(at=time.time(),image=before['Image'],compose_hash=compose_hash,env_hash=file_hash(ENV),
    registration_hash=file_hash(rt.config['registration_file']),ledger=before_ledger,
    targets=TARGETS,old_artifact_enabled='1',new_artifact_enabled='0',http_enabled='1'))
value=yaml.safe_load(ACTIVE.read_text())
assert value['services']['analytics']['environment']['CONTEXTRA_EVENT_ARTIFACT_ENABLED']=='1'
value['services']['analytics']['environment']['CONTEXTRA_EVENT_ARTIFACT_ENABLED']='0'
assert value['services']['analytics']['environment']['CONTEXTRA_ARTIFACT_HTTP_ENABLED']=='1'
temporary=ACTIVE.with_suffix('.artifact-off-pending')
with temporary.open('x') as f:yaml.safe_dump(value,f,sort_keys=False)
temporary.chmod(0o600);assert file_hash(ACTIVE)==compose_hash;os.replace(temporary,ACTIVE)
args=['docker','compose','-p','contextra','--env-file',str(ENV),'-f',str(REPO/'docker-compose.yml'),
      '-f',str(ACTIVE),'up','-d','--no-deps','--no-build','--pull','never','analytics']
save('reflect-intent.json',dict(at=time.time(),compose_hash=file_hash(ACTIVE)))
try:
    command(args,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')
    status=rt.api.call('GET','/api/v1/metadata/status')['artifacts']
    assert not status['enabled'] and status['workers']==0 and status['http_enabled']
    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['version_policy']['specific']['versions']==['4'] or model['version_policy']['specific']['versions']==[4]
    for sid in TARGETS:
        settings=rt.settings(sid)
        assert settings['tracking']['enabled'] and settings['intrusion']['enabled']
        assert not settings['fire']['enabled'] and not settings['pose']['enabled']
        save('settings-'+sid+'.json',settings)
        state=rt.start(sid)
        assert state['tracking']['enabled'] if 'tracking' in state else True
        save('started-'+sid+'.json',state)
        print('Running',sid,'PID',state['child_pid'],flush=True)
    states=rt.states()
    assert {s for s,q in states.items() if q['state']=='running'}==set(TARGETS)
    assert len({states[s]['child_pid'] for s in TARGETS})==2
    for sid in TARGETS:
        assert states[sid]['metadata_path']['processed']>0
        assert states[sid]['metadata_path'].get('artifacts') is None
    after_ledger=ledger();assert after_ledger==before_ledger
    sample=json.loads(pg("SELECT row_to_json(x) FROM (SELECT event_id,relative_path FROM contextra_event_artifacts WHERE artifact_state='READY' ORDER BY created_at DESC LIMIT 1) x;"))
    env=dict(x.split('=',1) for x in rt.inspect()['Config']['Env'])
    url=env['CONTEXTRA_ARTIFACT_PUBLIC_BASE'].rstrip('/')+'/'+sample['relative_path']+'/clip.mp4'
    with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(urllib.request.Request(url,method='HEAD'),timeout=5) as response:
        media=dict(event_id=sample['event_id'],method='HEAD',status=response.status,size=response.headers.get('Content-Length'))
    assert file_hash(rt.config['registration_file'])==json.loads((HERE/'plan.json').read_text())['registration_hash']
    save('summary.json',dict(at=time.time(),image=rt.inspect()['Image'],states=states,
        artifacts=rt.api.call('GET','/api/v1/metadata/status')['artifacts'],ledger=after_ledger,
        ledger_unchanged=True,media=media,model=model,resources=rt.resources(),compose_hash=file_hash(ACTIVE)))
    print('Artifact Off, HTTP On, ledger unchanged, originals running; no performance window',flush=True)
except Exception as exc:
    save('failure.json',dict(type=type(exc).__name__,message=str(exc)[:300]))
    for sid in reversed(TARGETS):
        try:save('cleanup-'+sid+'.json',rt.stop(sid))
        except Exception as error:save('cleanup-error-'+sid+'.json',dict(type=type(error).__name__))
    command(['docker','stop','--time','85',rt.container],100)
    raise