"""Adopt an existing Compose stack without moving its data or changing boot policy."""
import base64
import copy
import hashlib
import json
import os
from pathlib import Path
import shutil
import time

import field_profiles as fp
import server_context as server


def canonical(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()


def execute(args, timeout=30):
    from maintenance_process import run
    from deployment_kit.compose_snapshot import command_environment
    return run(args, check=True, capture_output=True, text=True, timeout=timeout,
               env=command_environment(), cwd=server.CURRENT['working_dir']).stdout


def compose_args(path=None):
    args = ['docker','compose','--project-directory',server.CURRENT['working_dir'],'-p',server.project()]
    if path is None:
        for p in server.CURRENT['env_files']:args += ['--env-file',p]
        for p in server.CURRENT['compose_files']:args += ['-f',p]
    else:args += ['-f',str(path)]
    return args


def render():
    return server.logical(json.loads(execute(compose_args()+['config','--format','json'])))


def state_path():
    return fp.RUNTIME/'state'


def active():
    return fp.read_json(state_path()/'active.json', 1024**2)


def current_path():
    p = fp.safe_path(active()['compose'])
    if not p.is_relative_to(state_path()/'transactions'):
        raise ValueError('Active Compose outside installation metadata')
    return p


def running():
    return execute(['docker','ps','--filter','label=com.docker.compose.project='+server.project(),'--format','{{.ID}}']).split()


def stopped():
    if running():raise ValueError('Target services must be stopped; use ct stop explicitly')


def containers():
    ids = execute(['docker','ps','-aq','--filter','label=com.docker.compose.project='+server.project()]).split()
    if not ids:return []
    return json.loads(execute(['docker','inspect',*ids]))


def preserve_resources(doc, rows):
    """Pin running images, actual anonymous volumes and project networks by identity."""
    result = copy.deepcopy(doc)
    by_name = {}
    for row in rows:
        labels = row['Config'].get('Labels') or {}
        role = next((r for r in server.ROLES if server.service(r) == labels.get('com.docker.compose.service')), None)
        if role is None or role in by_name:raise ValueError('Unexpected or replicated service during adoption')
        by_name[role] = row
    if set(by_name) != set(server.ROLES):raise ValueError('Five existing service containers required for adoption')
    for role, cfg in result['services'].items():
        row = by_name[role]
        cfg['image'] = row['Image'];cfg.pop('build',None);cfg.pop('pull_policy',None)
        mounts = {m['Destination']:m for m in row['Mounts']}
        for mount in cfg.get('volumes',[]):
            actual = mounts.get(mount['target'])
            if actual is None or mount['type'] != actual['Type']:
                raise ValueError('Configured/live storage mismatch: '+role)
            if mount['type'] == 'bind' and Path(mount['source']) != Path(actual['Source']):
                raise ValueError('Configured/live bind mismatch: '+role)
            if mount['type'] == 'volume':
                key = 'ct_'+role+'_'+hashlib.sha256(mount['target'].encode()).hexdigest()[:10]
                result.setdefault('volumes',{})[key] = {'external':True,'name':actual['Name']}
                mount['source'] = key
        configured_targets = {m['target'] for m in cfg.get('volumes',[])}
        for target, actual in mounts.items():
            if actual['Type'] == 'volume' and target not in configured_targets:
                key = 'ct_'+role+'_'+hashlib.sha256(target.encode()).hexdigest()[:10]
                result.setdefault('volumes',{})[key] = {'external':True,'name':actual['Name']}
                item={'type':'volume','source':key,'target':target}
                if not actual['RW']:item['read_only']=True
                cfg.setdefault('volumes',[]).append(item)
        attached = row['NetworkSettings']['Networks']
        for key in cfg.get('networks',{}):
            name = result.get('networks',{}).get(key,{}).get('name',server.project()+'_'+key)
            if name not in attached:raise ValueError('Configured/live network mismatch')
            result.setdefault('networks',{})[key] = {'external':True,'name':name,'ipam':{}}
    used = {m['source'] for cfg in result['services'].values() for m in cfg.get('volumes',[]) if m['type']=='volume'}
    result['volumes'] = {k:v for k,v in result.get('volumes',{}).items() if k in used}
    return result


def inspect():
    import field_runtime as runtime
    a = active(); cp = current_path()
    protected=fp.read_json(state_path()/'protected-bindings.json',1048576) if (state_path()/'protected-bindings.json').exists() else {}
    from deployment_kit.compose_snapshot import verify_bindings
    verify_bindings(protected)
    base = fp.read_json(cp, 8*1024**2)
    if canonical(base) != a['compose_semantic_sha256']:
        approved = fp.read_json(cp.with_name('approved.json'), 8*1024**2)
        if canonical(approved) != a['compose_semantic_sha256']:raise ValueError('Protected baseline damaged')
        runtime.check_drift(approved, base)
    else:approved = base
    old_input = fp.read_json(state_path()/'input-baseline.json',8*1024**2)
    source = render()
    runtime.check_drift(old_input, source)
    site = fp.read_json(state_path()/'site.json')
    return a,site,cp,approved,base,source


def verify_plan(plan):
    from field_runtime import sha
    inspect()
    if canonical(render()) != plan['baseline']['input_stamp'] or sha(current_path()) != plan['baseline']['compose_sha256']:
        raise ValueError('Compose changed since preparation plan')


def firewall_stamp():
    f=server.CURRENT['firewall']
    if not f:return {}
    from field_runtime import sha
    script=Path(f['script']);access=script.with_name('access.json')
    return {'script':sha(script),'access':canonical(fp.read_json(access))}


def verify_enrollment(record):
    inspect()
    if canonical(render())!=record['input_stamp'] or firewall_stamp()!=record['firewall_stamp']:
        raise ValueError('Local source/firewall changed since prepare; no remote request')


def prepare(p):
    import connection
    import field_network
    import network_prepare
    import field_runtime as runtime
    a,site,cp,base,live,source = inspect()
    stopped();server.guard_disks()
    if (fp.STATE/'pending.json').exists():raise ValueError('Pending stage preserved; status/resume/cancel first')
    # A different public edit must be imported/reviewed, never silently overwritten by a profile.
    accepted_input=fp.read_json(state_path()/'input-baseline.json',8*1024**2)
    source_values=runtime.public_values(source)
    if canonical(source)!=canonical(accepted_input) or live!=base:
        desired_values=dict(ai_ip=field_network.ai_ip(p),manager=fp.web(p)['manager'],web_host=fp.web(p)['host'],
            event_url=fp.web(p)['event_url'],control_port=p['control_port'],redis_port=p['redis_port'],
            artifact_lan_cidr=p['artifact_lan_cidr'],server_id=fp.web(p)['server_id'])
        changed=source_values if canonical(source)!=canonical(accepted_input) else runtime.public_values(live)
        if changed!=desired_values:raise ValueError('Public direct edits conflict with profile. Review ct emergency before prepare')
    ca=fp.ca_path(p);connection.verify_ca(ca)
    credentials=fp.read_credentials(fp.secret_path(p))
    wanted=connection.addresses(field_network.ai_ip(p),fp.web(p)['host'],fp.web(p)['manager'],p['control_port'],p['redis_port'])
    wanted.update(event_endpoint=fp.web(p)['event_url'],artifact_lan_cidr=p['artifact_lan_cidr'],redis_allow=p['redis_allow'])
    directory=fp.STATE/('field-'+time.strftime('%Y%m%dT%H%M%S')+'-'+str(time.time_ns()))
    fp.atomic(directory/'plan.json',dict(desired=wanted,baseline=dict(compose_sha256=runtime.sha(cp),input_stamp=canonical(source))))
    stage=network_prepare.prepare(directory/'plan.json',ca,runtime.sha(ca),credentials=credentials,
        server_id=fp.web(p)['server_id'],connection_version=fp.web(p).get('connection_version'))
    record=fp.read_json(stage/'record.json')
    record.update(profile=p['id'],profile_stamp=fp.stamp(p),profile_snapshot=p,redis_allow=p['redis_allow'],firewall_stamp=firewall_stamp(),
        profile_credentials_sha256=runtime.sha(fp.secret_path(p)),profile_ca_sha256=runtime.sha(ca),public_before=runtime.public_values(base))
    fp.atomic(stage/'record.json',record)
    record['prepared_files']={name:runtime.sha(stage/name) for name in ('compose.proposed.json','host-enrollment.json',
        'bridge/login-body.json','bridge/password','bridge/manager-ca.crt','bridge/control-token','bridge/relay-token','tls/control-tls.key')}
    fp.atomic(fp.STATE/'stages'/(stage.name+'.json'),record)
    fp.atomic(fp.STATE/'pending.json',{'stage':str(stage),'profile':p['id'],'phase':'PREPARED'})
    print('Prepared only. Next: ct config connect (explicit remote request)')
    return stage


def allowed_writes():
    paths={state_path()/x for x in ('active.json','site.json','input-baseline.json')}
    firewall=server.CURRENT['firewall']
    if firewall:
        paths.add(Path(firewall['script']).with_name('access.json'))
        paths.add(Path(firewall['script']))
    return paths


def replay(journal, reverse=False):
    writes=journal['writes']
    if len({x['path'] for x in writes})!=len(writes) or {Path(x['path']) for x in writes}!=allowed_writes():
        raise ValueError('Unexpected apply/restore targets')
    for item in writes:
        path=fp.safe_path(item['path']);current=path.read_bytes() if path.exists() else None
        if current not in (base64.b64decode(item['before']),base64.b64decode(item['after'])):
            raise ValueError('Concurrent local edit preserved: '+str(path))
    for item in writes:
        path=fp.safe_path(item['path']);current=path.read_bytes() if path.exists() else None
        if current not in (base64.b64decode(item['before']),base64.b64decode(item['after'])):
            raise ValueError('Concurrent local write preserved: '+str(path))
        fp.atomic(path,base64.b64decode(item['before' if reverse else 'after']))


def firewall_apply():
    f=server.CURRENT['firewall']
    if f:
        # Reconcile the owned chain even when restoring a legacy append-only launcher.
        import portable_firewall
        portable_firewall.apply()


def guard_images(doc):
    for cfg in doc['services'].values():
        ident=cfg['image']
        if not ident.startswith('sha256:'):raise ValueError('Pinned local image ID required')
        found=json.loads(execute(['docker','image','inspect',ident]))
        if len(found)!=1 or found[0]['Id']!=ident:raise ValueError('Pinned image missing')


def apply_stage(stage):
    import field_runtime as runtime
    import field_network
    record,p=runtime.verify_pending(stage)
    runtime.stage_ready(stage);stopped();server.guard_disks()
    if shutil.disk_usage('/').free<15*1024**3:raise ValueError('Root reserve 15GiB required')
    if record['desired']['ai_ip'] not in field_network.addresses(p['network']['interface']):
        raise ValueError('Assign the intended NIC address before app apply')
    pending=fp.STATE/'apply-pending.json'
    if pending.exists():
        journal=fp.read_json(pending,16*1024**2)
        if journal['stage']!=str(stage):raise ValueError('Different apply pending')
    else:
        inspect()
        if runtime.sha(state_path()/'active.json')!=record['base_active_sha256'] or canonical(render())!=record['input_stamp'] or firewall_stamp()!=record['firewall_stamp']:
            raise ValueError('Configuration changed since prepare; preserve this issued stage')
        doc=fp.read_json(stage/'compose.proposed.json',8*1024**2);guard_images(doc)
        tx=state_path()/'transactions'/stage.name
        tx.mkdir(mode=0o700,exist_ok=True)
        fp.atomic(tx/'compose.json',doc);fp.atomic(tx/'approved.json',doc)
        physical=server.physical(doc);fp.atomic(tx/'compose.physical.json',physical)
        reread=json.loads(execute(compose_args(tx/'compose.physical.json')+['config','--format','json']))
        from apply_source import normalize
        if normalize(server.logical(reread))!=normalize(doc):raise ValueError('Final Compose differs; no active change')
        a={'transaction':stage.name,'compose':str(tx/'compose.json'),'physical':str(tx/'compose.physical.json'),
           'compose_semantic_sha256':canonical(doc),'profile':p['id'],'enrolled':True}
        site={'ai_ip':record['desired']['ai_ip'],'env_files':server.CURRENT['env_files']}
        values={state_path()/'active.json':a,state_path()/'site.json':site,state_path()/'input-baseline.json':render()}
        f=server.CURRENT['firewall']
        if f:
            access=Path(f['script']).with_name('access.json')
            original=fp.read_json(access)
            if set(original)!={'host','port','readers'}:raise ValueError('Unsupported Redis firewall access contract')
            values[access]={'host':site['ai_ip'],'port':p['redis_port'],'readers':p['redis_allow']}
            loader=('import sys\nsys.path.insert(0, '+repr(str(Path(__file__).parent))+')\n'
                    'from portable_firewall import apply\napply()\n').encode()
            values[Path(f['script'])]=loader
        journal={'stage':str(stage),'phase':'PREPARED','writes':[]}
        for path,value in values.items():
            journal['writes'].append({'path':str(path),'before':base64.b64encode(path.read_bytes()).decode(),
                'after':base64.b64encode(value if isinstance(value,bytes) else (json.dumps(value,indent=2)+'\n').encode()).decode()})
        fp.atomic(pending,journal)
    replay(journal)
    firewall_apply()
    journal['phase']='APPLIED';fp.atomic(fp.STATE/'last-compose-apply.json',journal)
    fp.atomic(fp.STATE/'last-applied.json',{'stage':str(stage),'status':'APPLIED'})
    runtime.export_receipt(stage,'APPLIED')
    pending.unlink(missing_ok=True);(fp.STATE/'pending.json').unlink(missing_ok=True)
    print('Applied locally; boot policy unchanged; remote auth is not rollbackable. Next: ct start')


def restore_config():
    stopped();server.guard_disks()
    pending=fp.STATE/'apply-pending.json'
    journal=fp.read_json(pending if pending.exists() else fp.STATE/'last-compose-apply.json',16*1024**2)
    replay(journal,reverse=True);firewall_apply()
    journal['phase']='RESTORED';fp.atomic(fp.STATE/'last-compose-restore.json',journal)
    pending.unlink(missing_ok=True)
    print('Local configuration restored only. Web registration/tokens were NOT restored. No automatic Start.')


def runtime(action):
    from data_safety import lock
    with lock(fp.STATE/'data-maintenance.lock'):
        return _runtime(action)


def _runtime(action):
    from maintenance_process import run
    import field_runtime as shared
    if action=='stop':
        # Stopping must remain possible even when a config edit blocks Start.
        a=active()
    else:a,site,cp,base,live,source=inspect()
    if action=='start':
        server.guard_disks();guard_images(base)
        if not a.get('enrolled'):raise ValueError('Existing stack adopted but not enrolled by ct. Complete user-led config prepare/connect/apply first')
        if canonical(base)!=canonical(live):raise ValueError('Review direct changes through ct emergency before Start')
        source_base=fp.read_json(state_path()/'input-baseline.json',8*1024**2)
        if canonical(source)!=canonical(source_base):raise ValueError('Persistent source changed; review ct emergency before Start')
        if (fp.STATE/'apply-pending.json').exists():raise ValueError('Interrupted configuration apply: ct config resume')
        if (fp.STATE/'network-pending.json').exists():raise ValueError('Incomplete NIC operation; inspect ct net status/restore')
        if (fp.STATE/'pending.json').exists():
            stage=shared.pending_stage()
            if (stage/'enroll-intent.json').exists() or (stage/'enroll-result.json').exists():
                raise ValueError('Remote enrollment pending; reconcile ct config status/resume before Start')
        import field_network
        p=fp.load(a['profile'])
        if site['ai_ip'] not in field_network.addresses(p['network']['interface']):raise ValueError('Configured AI address is not assigned')
        shared.stage_ready(Path(fp.read_json(fp.STATE/'last-applied.json')['stage']))
        firewall_apply()
        execute(compose_args(a['physical'])+['up','-d','--pull','never','--no-build'],240)
    elif action=='stop':
        rows=containers();forced=[]
        for role in ('ai-gateway','analytics'):
            for row in rows:
                if not row['State']['Running'] or row['Config']['Labels'].get('com.docker.compose.service')!=server.service(role):continue
                if role=='ai-gateway':
                    execute(['docker','stop','--timeout','30',row['Id']],45)
                    state=json.loads(execute(['docker','inspect',row['Id']]))[0]['State']
                    if state.get('OOMKilled') or state.get('ExitCode')==137:forced.append(role)
                else:
                    helper=Path(__file__).parent/'stop_channels.py'
                    run(['docker','exec','-i',row['Id'],'python3','-'],input=helper.read_bytes(),check=True,timeout=240,capture_output=True)
        execute(compose_args(a['physical'])+['stop','--timeout','60'],330)
        stopped()
        if forced:raise RuntimeError('Stopped, but Gateway force termination observed; not graceful PASS')
    print('Services:',action,'No automatic camera Start.')


def status():
    import field_runtime as runtime
    a,site,cp,base,live,source=inspect()
    print('Managed Compose:',cp)
    print('Effective app values:',json.dumps(runtime.public_values(live),ensure_ascii=False))
    print('Public direct edit:',canonical(base)!=canonical(live),'Boot policy: unchanged')
    rows=containers()
    print('Services:',json.dumps({r['Config']['Labels']['com.docker.compose.service']:r['State']['Status'] for r in rows}))
    print('Login/certificate/WSS/Redis/display: not probed. Remote current version: unknown.')