import json,os,pathlib,subprocess,tempfile,time,urllib.request
from build import g,R,HEAD
os.umask(0o077)
plan=json.loads((R/'plan.json').read_text());built=json.loads((R/'build.json').read_text())
C=pathlib.Path(plan['controls']);env=g.load_env_file(str(g.ENV));overlay=pathlib.Path(env['CONTEXTRA_RUNTIME_STATE_ROOT'])/'streams_runtime.yaml'
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
def api(path):
 with opener.open(base+path,timeout=3) as f:return json.load(f)
def cmd(args,timeout=30):return subprocess.run(args,capture_output=True,text=True,check=True,timeout=timeout)
def atomic(path,data,expected):
 assert g.sha(path)==expected
 fd,name=tempfile.mkstemp(dir=path.parent)
 with os.fdopen(fd,'wb') as f:os.fchmod(f.fileno(),0o600);f.write(data);f.flush();os.fsync(f.fileno())
 assert g.sha(path)==expected;os.replace(name,path);return g.sha(path)
def restart_arg(policy):
 return policy['Name']+(':'+str(policy['MaximumRetryCount']) if policy.get('MaximumRetryCount') else '')
def rollback(state):
 if state.get('analytics_reflected'):
  cmd(['docker','stop','--timeout','20',g.NAME],40)
 if state.get('env_after'):
  atomic(g.ENV,(R/'env.before').read_bytes(),state['env_after'])
 for name,key in [('redis','redis_reflected'),('analytics','analytics_reflected')]:
  if not state.get(key):continue
  old=plan[name];files=plan['previous_compose_files'][old['Id']]
  assert all(g.sha(p)==h for p,h in files.items())
  args=['docker','compose','--env-file',str(g.ENV)]
  for p in files:args+=['-f',p]
  cmd(args+['up','-d','--no-deps','--force-recreate','--pull','never',name],120)
 if state.get('firewall_applied'):
  redis=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0]
  assert not redis['HostConfig']['PortBindings']
  cmd(['systemctl','disable','--now','contextra-metadata-firewall.service'])
  cmd(['python3',str(C/'firewall.py'),'remove'])
  pathlib.Path(plan['unit']).unlink();cmd(['systemctl','daemon-reload'])
 state['rollback']='applied'
def main():
 assert not (R/'apply.json').exists()
 state=dict(stage='starting',runtime_windows=0)
 def save():(R/'apply.json').write_text(json.dumps(state,indent=2))
 save()
 try:
  assert g.sh(['git','-C',str(g.REPO),'rev-parse','HEAD']).strip()==HEAD
  assert all(g.sha(g.REPO/p)==h for p,h in json.loads((R/'source.json').read_text())['hashes'].items())
  assert g.sha(g.ENV)==plan['env_hash'] and g.sha(overlay)==plan['overlay_hash']
  assert g.inspect()['Id']==plan['analytics']['Id']
  r=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0];assert r['Id']==plan['redis']['Id']
  assert g.sh(['docker','exec',r['Id'],'redis-cli','DBSIZE']).strip()=='0'
  assert all(g.sha(p)==h for p,h in plan['control_hashes'].items())
  assert api('/api/v1/streams')['streams'][0]['state']=='stopped'
  assert not pathlib.Path(plan['unit']).exists()
  cmd(['install','-m','644',str(R/'contextra-metadata-firewall.service'),plan['unit']])
  cmd(['systemd-analyze','verify',plan['unit']])
  cmd(['systemctl','daemon-reload']);cmd(['systemctl','enable','--now','contextra-metadata-firewall.service'])
  state['firewall_applied']=True;save()
  cmd(['python3',str(C/'firewall.py'),'check'])
  lines=g.ENV.read_text().splitlines(keepends=True)
  indices=[i for i,x in enumerate(lines) if x.split('=',1)[0]=='CONTEXTRA_IMAGE'];assert len(indices)==1
  lines[indices[0]]='CONTEXTRA_IMAGE='+built['tag']+'\n'
  state['env_after']=atomic(g.ENV,''.join(lines).encode(),plan['env_hash']);save()
  state['redis_reflected']=True;save()
  cmd(plan['compose']+['up','-d','--no-deps','--force-recreate','--pull','never','redis'],120)
  end=time.monotonic()+60
  while time.monotonic()<end:
   r=json.loads(g.sh(['docker','inspect','contextra-redis-1']))[0]
   assert not r['RestartCount'] and r['State']['Status']=='running'
   if r['State'].get('Health',{}).get('Status')=='healthy':break
   time.sleep(1)
  assert r['State']['Health']['Status']=='healthy'
  assert g.sh(['docker','exec',r['Id'],'redis-cli','--raw','PING']).startswith('NOAUTH')
  state['redis_id']=r['Id'];state['redis_auth']='PASS';save()
  state['analytics_reflected']=True;save()
  cmd(plan['compose']+['up','-d','--no-deps','--force-recreate','--pull','never','analytics'],120)
  end=time.monotonic()+300
  while time.monotonic()<end:
   c=g.inspect();assert c['Image']==built['image_id'] and c['State']['Status']=='running' and not c['RestartCount']
   assert not c['State']['OOMKilled'] and g.gpu()-plan['gpu_mib']<8192
   try:
    for p in ['/live','/ready','/health']:api(p)
    if c['State'].get('Health',{}).get('Status')=='healthy':break
   except Exception:pass
   time.sleep(2)
  assert c['State']['Health']['Status']=='healthy' and api('/ready')['ready']
  assert api('/api/v1/streams')['streams'][0]['state']=='stopped'
  assert api('/api/v1/metadata/status')['enabled'] is True
  state.update(stage='ready',analytics_id=c['Id'],metadata=api('/api/v1/metadata/status'))
  save();print(json.dumps({'stage':'ready','redis_auth':'PASS','analytics':'healthy','stream':'stopped'}))
 except Exception as error:
  state.update(stage='failed',error=type(error).__name__);save()
  # 복원으로 container가 교체되기 전에 제한된 실패 근거를 보존한다.
  try:
   failed=g.inspect()
   state['failed_state']={k:failed['State'].get(k) for k in ['Status','ExitCode','OOMKilled']}
   state['failed_image']=failed['Image']
   logs=subprocess.run(['docker','logs','--tail','150',g.NAME],capture_output=True,timeout=10)
   (R/'analytics-failure.private.log').write_bytes((logs.stdout+logs.stderr)[-1024*1024:])
  except Exception as observation:state['observation_error']=type(observation).__name__
  save()
  try:rollback(state)
  except Exception as restore:state['rollback_error']=type(restore).__name__
  save();print(json.dumps({k:state.get(k) for k in ['stage','error','rollback','rollback_error']}));raise
if __name__=='__main__':main()