import hashlib
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tarfile
import time
import yaml

REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
R=Path(__file__).parent
ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml')
ENV=Path('/workspace/contextra-runtime-config/runtime.env')
BASECFG=Path('/workspace/contextra-runtime-config/test-run/event-artifact-next.yaml')
CFG=Path('/workspace/contextra-runtime-config/test-run/artifact-parallel16.yaml')
sys.path.insert(0,str(REPO))
from tools.stream_benchmark import load_config,preflight,Benchmark
from tools.stream_benchmark_runtime import Runtime,file_hash

def save(name,value):
    p=R/name
    with p.open('x') as f:json.dump(value,f,indent=2)
    p.chmod(0o600)

def read(name):return json.loads((R/name).read_text())
def cmd(args,timeout=30,stdin=None):
    p=subprocess.run(args,input=stdin,capture_output=True,text=True,timeout=timeout)
    if p.returncode:
        raise RuntimeError(str(args[:3])+': '+p.stderr[-2000:])
    return p.stdout
def inspect():return json.loads(cmd(['docker','inspect','contextra-analytics-1']))[0]
def pg(sql):
    return 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=sql)
def compose():
    return ['docker','compose','-p','contextra','--env-file',str(ENV),'-f',str(REPO/'docker-compose.yml'),'-f',str(ACTIVE)]

def gate():
    c=inspect();assert not c['State']['Running'] and c['State']['Pid']==0
    assert cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip()=='aeae108acd1b439e96da5ec48a702682514e5adc'
    assert pg('SELECT count(*) FROM contextra_event_artifacts;').strip()=='0'
    assert shutil.disk_usage('/var/lib/docker').free>15*1024**3
    cfg=load_config(BASECFG)
    rows=json.loads(Path(cfg['identity_manifest']).read_text())
    save('gate.json',dict(head=cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip(),at=time.time(),
         old_image=c['Image'],compose_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),
         registry_hash=file_hash(cfg['registration_file']),manifest_hash=file_hash(cfg['identity_manifest']),
         mounts=[m for m in c['Mounts'] if m['Destination'].startswith('/opt/contextra/models/')],
         bounds=dict(build_seconds=900,free_disk_gib=15,observation_seconds=60,sample_seconds=2,
                     warmup_seconds=5,renderers=4,workers=8,expiry_seconds=120),
         artifacts=cfg['artifacts']))
    print('Gate: stopped, empty ledger, fixed model hashes, 15GiB floor')

def fixture():
    c=inspect();environment=dict(x.split('=',1) for x in c['Config']['Env'])
    name='contextra-artifact-fixture-'+str(os.getpid())
    args=['docker','run','--rm','--name',name,'--network','contextra_default',
          '-e','DATABASE_URL','-v',str(REPO)+':/fixture:ro','-v',str(R)+':/evidence',
          '--entrypoint','python3','-w','/fixture',c['Image'],'-m','unittest',
          *(sys.argv[2:] or ['tests.test_event_artifact','tests.test_event_artifact_ledger','tests.test_artifact_media_runtime']),'-v']
    env=dict(os.environ,DATABASE_URL=environment['DATABASE_URL'])
    save('fixture-intent-'+str(time.time_ns())+'.json',dict(name=name,at=time.time()))
    try:
        p=subprocess.run(args,env=env,capture_output=True,text=True,timeout=180)
        nameout='fixture-'+str(time.time_ns())+'.json'
        save(nameout,dict(code=p.returncode,stdout=p.stdout,stderr=p.stderr))
        print(p.stdout[-4000:]+p.stderr[-6000:])
        if p.returncode:raise RuntimeError('Fixture failed')
    finally:
        subprocess.run(['docker','rm','-f',name],capture_output=True,timeout=15)

def build():
    suffix='-final' if '--final' in sys.argv else ''
    assert not (R/('build'+suffix+'-intent.json')).exists()
    paths=cmd(['git','-C',str(REPO),'diff','--name-only']).splitlines()+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:artifact-parallel-aeae108-'+digest
    snapshot=R/('snapshot'+suffix);snapshot.mkdir()
    p=subprocess.Popen(['git','-C',str(REPO),'archive',read('gate.json')['head']],stdout=subprocess.PIPE)
    with tarfile.open(fileobj=p.stdout,mode='r|') as t:t.extractall(snapshot,filter='data')
    assert p.wait()==0
    for path in paths:
        (snapshot/path).parent.mkdir(parents=True,exist_ok=True);shutil.copy2(REPO/path,snapshot/path)
    args=['docker','build','--pull=false','--network=default','-f','docker/Dockerfile','--target','app','-t',tag,'.']
    save('build'+suffix+'-intent.json',dict(tag=tag,hashes=hashes,at=time.time()))
    with (R/('build'+suffix+'.log')).open('x') as log:
        p=subprocess.Popen(args,cwd=snapshot,stdout=log,stderr=log,start_new_session=True);start=time.monotonic()
        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 bound reached')
            time.sleep(2)
    assert p.returncode==0
    image=json.loads(cmd(['docker','image','inspect',tag]))[0]['Id']
    save('image'+suffix+'.json',dict(tag=tag,id=image,hashes=hashes,elapsed=time.monotonic()-start))
    print('Image built',tag,image)

def write_yaml(path,value,expected):
    assert file_hash(path)==expected
    temporary=path.with_suffix(path.suffix+'.artifact-pending')
    with temporary.open('x') as f:yaml.safe_dump(value,f,sort_keys=False)
    temporary.chmod(0o600)
    assert file_hash(path)==expected
    os.replace(temporary,path)

def reflect():
    gate=read('gate.json');image=read('image.json');assert not inspect()['State']['Running']
    assert file_hash(ACTIVE)==gate['compose_hash'] and file_hash(ENV)==gate['env_hash']
    for item in gate['artifacts']:assert file_hash(item['path'])==item['sha256']
    save('reflect-intent.json',dict(image=image['id'],old_image=gate['old_image'],at=time.time()))
    value=yaml.safe_load(ACTIVE.read_text());service=value['services']['analytics']
    service['image']=image['id']
    service['environment'].update(CONTEXTRA_EVENT_ARTIFACT_ENABLED='1',CONTEXTRA_ARTIFACT_HTTP_ENABLED='1',
                                  CONTEXTRA_ARTIFACT_DELIVERY_CAMERAS='1,2')
    write_yaml(ACTIVE,value,gate['compose_hash'])
    lines=ENV.read_text().splitlines();assert sum(x.startswith('CONTEXTRA_IMAGE=') for x in lines)==1
    lines=[('CONTEXTRA_IMAGE='+image['tag']) if x.startswith('CONTEXTRA_IMAGE=') else x for x in lines]
    temporary=ENV.with_suffix('.artifact-pending')
    with temporary.open('x') as f:f.write('\n'.join(lines)+'\n')
    temporary.chmod(0o600);assert file_hash(ENV)==gate['env_hash'];os.replace(temporary,ENV)
    cmd(compose()+['up','-d','--no-deps','--no-build','--pull','never','analytics'],180)
    cfg=load_config(BASECFG);cfg['expected_image']=image['id'];cfg['stages']=[dict(channels=16,seconds=60)]
    cfg['sample_seconds']=2;cfg['limits']['container_ram_gib']=24
    with CFG.open('x') as f:yaml.safe_dump(cfg,f,sort_keys=False)
    CFG.chmod(0o600)
    rt=Runtime(cfg,R);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')
    state=inspect();assert state['Image']==image['id']
    assert all(not s.get('child_pid') and s['state'] in ('stopped','error') for s in rt.states().values())
    hashes={p:h for p,h in image['hashes'].items() if p.startswith('app/')}
    code="import hashlib,json,sys;from pathlib import Path;import app.event_artifact_service as m; paths=json.loads(sys.argv[1]);print(json.dumps(dict(module=m.__file__,hashes={p:hashlib.sha256(Path(p).read_bytes()).hexdigest() for p in paths})))"
    reflected=json.loads(cmd(['docker','exec','-w','/opt/contextra',rt.container,'python3','-c',code,json.dumps(list(hashes))]))
    assert reflected['module']=='/opt/contextra/app/event_artifact_service.py' and reflected['hashes']==hashes
    plan=preflight(rt)
    save('runtime-plan.json',plan)
    save('reflected.json',dict(image=state['Image'],source=reflected,compose_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),
                               artifact_status=rt.api.call('GET','/api/v1/metadata/status')['artifacts']))
    print('Reflection/preflight passed: 16 stopped identities, model unchanged, image-owned source')

def ledger():
    return json.loads(pg("SELECT coalesce(json_agg(x),'[]') FROM (SELECT * FROM contextra_event_artifacts ORDER BY created_at,event_id) x;"))

def test():
    rt=Runtime(load_config(CFG),R);plan=read('runtime-plan.json')
    save('test-intent.json',dict(at=time.time(),stages=[16],seconds=60))
    code=Benchmark(rt).run(plan)
    save('events-after-test.json',ledger())
    print('Benchmark exit',code,flush=True)
    if code:
        cmd(['docker','stop','--time','85',rt.container],100)
        save('failure-stopped.json',dict(state=inspect()['State'],at=time.time()))
        raise RuntimeError('Runtime failed; stopped without rollback')

def finish():
    rt=Runtime(load_config(CFG),R);result=read('result.json')
    assert result['runtime_error'] is None
    save('finish-intent.json',dict(at=time.time(),artifact_off=True,http_remains=True))
    cleanup={}
    for row in read('runtime-plan.json')['rows']:
        if row['original']:cleanup[row['stream_id']]=rt.stop(row['stream_id'])
    save('original-stop.json',cleanup)
    deadline=time.monotonic()+135
    while time.monotonic()<deadline:
        rows=ledger()
        active=[e for e in rows if e['reservation']>0 or e['delivery_state'] in ('PENDING','RETRY','CLAIMED','SENDING')]
        if not active:break
        time.sleep(2)
    save('drained-events.json',ledger())
    active=[e for e in ledger() if e['reservation']>0 or e['delivery_state'] in ('PENDING','RETRY','CLAIMED','SENDING')]
    if active:raise RuntimeError('Bounded drain incomplete; no restart')
    gets=[]
    for camera in (1,2):
        matches=[e for e in ledger() if e['payload']['cameraId']==camera and e['artifact_state']=='READY' and e['delivery_state']=='SENT']
        if not matches:continue
        item=matches[-1];body=item['body']['events'][0]
        import urllib.request
        for key in ('clipUrl','thumbnailUrl'):
            if key not in body:continue
            with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(body[key],timeout=5) as response:
                count=0
                while chunk:=response.read(65536):
                    count+=len(chunk)
                    if count>16*1024**2:raise RuntimeError('GET bound')
                gets.append(dict(camera_id=camera,event_id=item['event_id'],media=key,status=response.status,bytes=count))
    save('media-get.json',gets)
    cmd(['docker','stop','--time','85',rt.container],100)
    value=yaml.safe_load(ACTIVE.read_text());before=file_hash(ACTIVE)
    value['services']['analytics']['environment']['CONTEXTRA_EVENT_ARTIFACT_ENABLED']='0'
    write_yaml(ACTIVE,value,before)
    save('off-reflect-intent.json',dict(compose_hash=file_hash(ACTIVE),at=time.time()))
    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('Off readiness timeout')
    assert rt.api.call('GET','/api/v1/metadata/status')['artifacts']['enabled'] is False
    states={}
    for row in read('runtime-plan.json')['rows']:
        if row['original']:states[row['stream_id']]=rt.start(row['stream_id'])
    status=rt.api.call('GET','/api/v1/metadata/status')['artifacts']
    assert not status['enabled'] and status['workers']==0
    save('final-state.json',dict(states=rt.states(),artifacts=status,image=inspect()['Image'],compose_hash=file_hash(ACTIVE),
                               env_hash=file_hash(ENV),events=len(ledger()),at=time.time()))
    print('Finished: originals running with Artifact Off/HTTP only, extra14 stopped')

def stage_final():
    assert not inspect()['State']['Running'] and inspect()['State']['Pid']==0
    previous=read('reflected.json');image=read('image-final.json')
    assert file_hash(ACTIVE)==previous['compose_hash'] and file_hash(ENV)==previous['env_hash']
    save('final-image-intent.json',dict(image=image['id'],at=time.time(),start=False,reason='Expose unchanged 80GiB warning threshold'))
    value=yaml.safe_load(ACTIVE.read_text());value['services']['analytics']['image']=image['id']
    write_yaml(ACTIVE,value,previous['compose_hash'])
    lines=ENV.read_text().splitlines();lines=[('CONTEXTRA_IMAGE='+image['tag']) if x.startswith('CONTEXTRA_IMAGE=') else x for x in lines]
    temp=ENV.with_suffix('.artifact-pending')
    with temp.open('x') as f:f.write('\n'.join(lines)+'\n')
    temp.chmod(0o600);assert file_hash(ENV)==previous['env_hash'];os.replace(temp,ENV)
    cfg=yaml.safe_load(CFG.read_text());cfg['expected_image']=image['id'];write_yaml(CFG,cfg,file_hash(CFG))
    cmd(compose()+['up','--no-start','--no-deps','--no-build','--pull','never','analytics'],180)
    c=inspect();assert c['Image']==image['id'] and not c['State']['Running'] and c['State']['Pid']==0
    hashes={p:h for p,h in image['hashes'].items() if p.startswith('app/')}
    code="import hashlib,json,sys;from pathlib import Path;import app.event_artifact_service 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])})))"
    actual=json.loads(cmd(['docker','run','--rm','--network','none','--entrypoint','python3','-w','/opt/contextra',image['id'],'-c',code,json.dumps(list(hashes))]))
    assert actual['hashes']==hashes and actual['module']=='/opt/contextra/app/event_artifact_service.py'
    save('final-state.json',dict(image=image['id'],tag=image['tag'],runtime_test_image=read('image.json')['id'],
        state=c['State']['Status'],pid=c['State']['Pid'],source=actual,compose_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),
        artifact_enabled_next_start=True,http_enabled_next_start=True,restarted_channels=False,
        next_start_note='Existing pending Events retain deadlines; expired jobs are cleaned without POST'))
    print('Final warning-only image staged, container not started, no channel restarted')

if __name__=='__main__':globals()[sys.argv[1]]()