import importlib.util
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

os.umask(0o077)
R = Path(__file__).parent
s = importlib.util.spec_from_file_location('previous', '/tmp/codex/ltb1021/2026-09-15_contextra-redis-guide_r01/common.py')
m = importlib.util.module_from_spec(s)
s.loader.exec_module(m)
C, REPO, g = m.C, m.REPO, m.g
ADDED = ['getrange', 'strlen', 'mget', 'hlen', 'hexists', 'hmget', 'hkeys', 'hvals', 'hstrlen', 'pttl']
FILES = dict(acl=C/'users.acl', config=C/'redis.conf', compose=C/'compose.yaml', env=g.ENV, access=C/'access.json', firewall=C/'firewall.py')

def save(name, value):
    (R/name).write_text(json.dumps(value, ensure_ascii=False, indent=2))

def inspect(name):
    return json.loads(subprocess.check_output(['docker', 'inspect', name]))[0]

def streams():
    rows = m.api('/api/v1/streams')['streams']
    assert all(x['state']=='stopped' and x.get('desired_state', 'stopped')=='stopped' for x in rows), 'AI_NOT_STOPPED'
    return rows

def inventory():
    ids = subprocess.check_output(['docker','ps','-aq'], text=True).split()
    rows = json.loads(subprocess.check_output(['docker','inspect',*ids]))
    return {x['Name']:dict(id=x['Id'], started=x['State']['StartedAt'], running=x['State']['Running']) for x in rows if x['Name']!='/contextra-redis-1'}

def keys():
    info = m.query('admin','INFO','keyspace')
    assert all(line.startswith(('db0:', '#')) or not line for line in info.splitlines()), 'OTHER_DB_PRESENT'
    assert m.query('admin','DBSIZE') <= 100, 'KEY_CAP'
    cursor, found = '0', []
    for _ in range(20):
        cursor, batch = m.query('admin','SCAN',cursor,'COUNT','100')
        found.extend(batch)
        if str(cursor)=='0': break
    else: raise RuntimeError('SCAN_CAP')
    rows = []
    for key in sorted(set(found)):
        assert key.startswith('contextra:metadata:v1:'), 'OTHER_KEY_PRESENT'
        ttl = m.query('admin','PTTL',key)
        typ = m.query('admin','TYPE',key)
        assert ttl >= 0 or ttl == -2, 'NON_EXPIRING_KEY'
        assert not key.endswith(':frames'), 'UNEXPECTED_FRAMES'
        value = m.query('admin','GET',key) if typ=='string' else m.query('admin','HGETALL',key) if typ=='hash' else None
        rows.append(dict(key=key, type=typ, ttl_ms=ttl, sampled_at_ms=int(time.time()*1000), value=value))
    return rows

def compose_args():
    return ['docker','compose','-p','contextra','--env-file',str(g.ENV),'-f',str(REPO/'docker-compose.yml'),'-f',str(C/'compose.yaml')]

def checks():
    key='contextra:metadata:v1:channels'
    commands = [['GETRANGE',key,'0','127'],['STRLEN',key],['MGET',key],['HLEN',key],['HEXISTS',key,'demo'],['HMGET',key,'demo'],['HKEYS',key],['HVALS',key],['HSTRLEN',key,'demo'],['PTTL',key],['XREAD','COUNT','1','STREAMS','contextra:metadata:v1:camera:demo:frames','0-0'],['SCAN','0','MATCH','contextra:metadata:v1:*','COUNT','1'],['XINFO','STREAM',key]]
    denied = [['GET','outside:demo'],['MGET',key,'outside:demo'],['HGET','outside:demo','x'],['SET',key,'x'],['DEL',key],['XADD',key,'*','json','{}'],['XTRIM',key,'MAXLEN','1'],['EXPIRE',key,'1'],['CONFIG','GET','maxmemory'],['ACL','CAT'],['KEYS','*'],['MONITOR'],['EVAL','return 1','0'],['COMMAND','INFO','GET']]
    out=[]
    for allowed, cases in [(True,commands),(False,denied)]:
        for cmd in cases:
            try: result=m.query('admin','ACL','DRYRUN','contextra_reader',*cmd)
            except RuntimeError as e: result=str(e)
            ok=result=='OK'
            assert ok==allowed, (cmd[0],result)
            out.append(dict(command=cmd[0],allowed=allowed,pass_=True))
    return out

def gate():
    assert not (R/'plan.json').exists(), 'ALREADY_PLANNED'
    assert subprocess.check_output(['git','rev-parse','HEAD'],cwd=REPO,text=True).strip()=='7fa1a496d47b3773c6793e55b6d99680d3896f6e'
    assert not subprocess.check_output(['git','status','--porcelain'],cwd=REPO,text=True).strip()
    before=inspect('contextra-redis-1')
    cfg=json.loads(subprocess.check_output(compose_args()+['config','--format','json'],cwd=REPO))
    service=cfg['services']['redis']
    assert service['command'][service['command'].index('--maxmemory')+1]=='8gb'
    assert service['image']==before['Config']['Image'], 'IMAGE_CHANGE'
    volumes=[x for x in before['Mounts'] if x['Destination']=='/data']
    assert len(volumes)==1 and volumes[0]['Type']=='volume'
    data=Path(volumes[0]['Source'])
    assert shutil.disk_usage(data).free>=15*1024**3
    assert sum(p.stat().st_size for p in data.rglob('*') if p.is_file())<16*1024**2, 'DATA_FILE_CAP'
    for name,path in FILES.items(): shutil.copy2(path,R/(name+'.before'))
    if (data/'dump.rdb').exists(): shutil.copy2(data/'dump.rdb',R/'dump.before.rdb')
    plan=dict(started=time.time(),budget_seconds=1200, extension_max_seconds=2100, limits=dict(helper_seconds=120,key_count=100,data_bytes=16*1024**2,gpu=0,ai_start=0,redis_recreate=1,extra_restart=0), redis=before, others=inventory(), streams=streams(), keys=keys(), reader=m.query('admin','ACL','GETUSER','contextra_reader'), config=m.query('admin','CONFIG','GET','maxmemory','maxmemory-policy','save','appendonly','dir','dbfilename'), hashes={k:g.sha(v) for k,v in FILES.items()}, compose_redis=service, rollback='Restore exact ACL/config backups on CAS; same image/data volume; restore prior restart policy; no FLUSH. Shutdown SAVE failure leaves old service running. Recreate failure: inspect once, restart old if present or recreate from fixed active config using retained RDB; do not replay mutation blindly.')
    assert plan['config']['dir']=='/data' and plan['config']['dbfilename']=='dump.rdb'
    save('plan.json',plan)
    save('ledger.json',dict(gate=1,acl=0,shutdown_save=0,recreate=0,ai_start=0))
    print(json.dumps(dict(gate='PASS',keys=len(plan['keys']),streams_stopped=len(plan['streams']),prior_maxmemory='2gb',desired='8gb',data_volume_retained=True)))

def acl():
    plan=json.loads((R/'plan.json').read_text());assert not (R/'acl-started.json').exists()
    assert g.sha(FILES['acl'])==plan['hashes']['acl']
    assert m.query('admin','ACL','GETUSER','contextra_reader')==plan['reader']
    save('acl-started.json',dict(at=time.time()))
    original=FILES['acl'].read_text();lines=original.splitlines(keepends=True)
    indexes=[i for i,l in enumerate(lines) if l.startswith('user contextra_reader ')]
    assert len(indexes)==1
    i=indexes[0];lines[i]=lines[i].rstrip()+' '+ ' '.join('+'+c for c in ADDED)+'\n'
    fd,tmp=tempfile.mkstemp(dir=C,prefix='.reader-acl-')
    try:
        os.fchmod(fd,0o600)
        with os.fdopen(fd,'w') as f:f.write(''.join(lines));f.flush();os.fsync(f.fileno())
        assert m.query('admin','ACL','SETUSER','contextra_reader',*['+'+c for c in ADDED])=='OK'
        os.replace(tmp,FILES['acl'])
        result=checks()
        now=m.query('admin','ACL','GETUSER','contextra_reader')
        assert set(now['commands'].split())-set(plan['reader']['commands'].split())=={'+'+c for c in ADDED}
        assert all(now[k]==plan['reader'][k] for k in ['passwords','keys','channels','flags','selectors'])
    except Exception:
        m.query('admin','ACL','SETUSER','contextra_reader',*['-'+c for c in ADDED])
        FILES['acl'].write_text(original)
        raise
    finally:
        if os.path.exists(tmp):os.unlink(tmp)
    save('acl-result.json',dict(checks=result,hash=g.sha(FILES['acl']),actual_read=dict(hlen=m.query('reader','HLEN','contextra:metadata:v1:channels'),pttl=m.query('reader','PTTL','contextra:metadata:v1:channels'))))
    print('PASS: 10 ACL additions, 27 dryrun cases, bounded reader queries')

def recreate():
    plan=json.loads((R/'plan.json').read_text());assert not (R/'recreate-started.json').exists()
    assert time.time()-plan['started']<plan['budget_seconds']
    streams(); before=inspect('contextra-redis-1');assert before['Id']==plan['redis']['Id']
    assert inventory()==plan['others']
    for k,p in FILES.items():assert g.sha(p)==(json.loads((R/'acl-result.json').read_text())['hash'] if k=='acl' else plan['hashes'][k]), k
    save('pre-shutdown-keys.json',keys())
    save('recreate-started.json',dict(at=time.time(),old_id=before['Id']))
    subprocess.run(['docker','update','--restart=no',before['Id']],check=True,capture_output=True,timeout=10)
    # SAVE snapshots the final accepted writes and preserves absolute expiry times.
    shutdown=m.query('admin','SHUTDOWN','SAVE')
    save('shutdown-result.json',dict(response=shutdown,at=time.time()))
    for _ in range(30):
        old=inspect(before['Id'])
        if not old['State']['Running']:break
        time.sleep(.1)
    else:raise RuntimeError('SHUTDOWN_NOT_COMPLETED')
    data=Path(next(x['Source'] for x in before['Mounts'] if x['Destination']=='/data'))
    assert (data/'dump.rdb').is_file()
    shutil.copy2(data/'dump.rdb',R/'shutdown.rdb')
    save('snapshot.json',dict(hash=g.sha(R/'shutdown.rdb'),bytes=(R/'shutdown.rdb').stat().st_size))
    p=subprocess.run(compose_args()+['up','-d','--no-deps','--no-build','--pull','never','--force-recreate','redis'],cwd=REPO,capture_output=True,text=True,timeout=90)
    save('recreate-command.json',dict(returncode=p.returncode,stdout=p.stdout,stderr=p.stderr))
    assert p.returncode==0,'RECREATE_FAILED'
    for _ in range(30):
        now=inspect('contextra-redis-1')
        if now['State'].get('Health',{}).get('Status')=='healthy':break
        time.sleep(1)
    else:raise RuntimeError('REDIS_NOT_HEALTHY')
    assert now['Id']!=before['Id'] and now['Image']==before['Image']
    assert next(x['Name'] for x in now['Mounts'] if x['Destination']=='/data')==next(x['Name'] for x in before['Mounts'] if x['Destination']=='/data')
    assert now['Config']['Cmd'][now['Config']['Cmd'].index('--maxmemory')+1]=='8gb'
    assert now['HostConfig']['RestartPolicy']==before['HostConfig']['RestartPolicy']
    assert now['HostConfig']['PortBindings']==before['HostConfig']['PortBindings']
    assert m.query('admin','CONFIG','GET','maxmemory')['maxmemory']=='8589934592'
    assert inventory()==plan['others']
    result=checks()
    save('recreate-result.json',dict(new_id=now['Id'],new_started=now['State']['StartedAt'],cmd=now['Config']['Cmd'],config=m.query('admin','CONFIG','GET','maxmemory','maxmemory-policy','save','appendonly'),checks=result,keys=keys(),publisher=m.api('/api/v1/metadata/status'),streams=streams(),others_preserved=True,completed_at=time.time()))
    save('ledger.json',dict(gate=1,acl=1,shutdown_save=1,recreate=1,ai_start=0,extra_restart=0))
    print('PASS: same image/volume, new container8gb, ACL/readiness, other containers preserved')

if __name__=='__main__':
    {'gate':gate,'acl':acl,'recreate':recreate}[sys.argv[1]]()