"""Protected DB2DB shared-cell rendezvous. Operator auth is the existing SMESH key."""
from __future__ import annotations
import base64,hashlib,hmac,json,secrets,sqlite3,tempfile,time,zipfile
from pathlib import Path
from threading import RLock
_LOCK=RLock();_CELLS={};ROOT=Path(tempfile.gettempdir())/'signalmesh-db2db-cells';ROOT.mkdir(exist_ok=True)
def _sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest()
def create_cell(ttl=3600,band=None,site_fingerprint=None):
 sid=secrets.token_hex(4).upper();sid=f'{sid[:4]}-{sid[4:]}';key=secrets.token_urlsafe(32);now=int(time.time());band=(str(band).lower() if band else None)
 if band not in (None,'source','target'):raise ValueError('band must be source or target')
 with _LOCK:_CELLS[sid]={'id':sid,'session_key':key,'created_at':now,'expires_at':now+ttl,'requested_band':band,'requested_site_fingerprint':site_fingerprint,'bands':{'source':None,'target':None},'connections':{'source':None,'target':None},'handshake_nonces':set(),'map':None,'verification':None,'allow_apply':False}
 return {'coordinate':sid,'expires_at':now+ttl,'bands':['source','target'],'requested_band':band}
def assign_band(sid,band,site_fingerprint=None):
 band=str(band).lower()
 if band not in ('source','target'):raise ValueError('band must be source or target')
 with _LOCK:
  c=_CELLS.get(sid)
  if not c:raise KeyError('unknown migration coordinate')
  c['requested_band']=band;c['requested_site_fingerprint']=site_fingerprint;c['connections'][band]=None
 return status(sid)
def discover(site_fingerprint=None):
 now=time.time()
 with _LOCK:
  candidates=[c for c in _CELLS.values() if now<=c['expires_at'] and c.get('requested_band') and not c['bands'][c['requested_band']] and (not c.get('requested_site_fingerprint') or c.get('requested_site_fingerprint')==site_fingerprint)]
  if not candidates:return None
  c=max(candidates,key=lambda x:x['created_at']);band=c['requested_band']
  return {'pending':True,'coordinate':c['id'],'band':band,'session_key':c['session_key'],'relay_url':'/api/editor/api/migration/beacon','handshake_url':'/api/editor/api/migration/beacon/handshake','expires_at':c['expires_at'],'allow_apply':bool(c.get('allow_apply') and band=='target')}
def handshake(env):
 sid=str(env.get('coordinate',''));band=str(env.get('band','')).lower();fp=str(env.get('site_fingerprint',''));nonce=str(env.get('nonce',''));protocol=str(env.get('protocol','signalmesh.db2db.v2'))
 try:ts=int(env.get('timestamp',0))
 except Exception:raise PermissionError('invalid handshake timestamp')
 if protocol!='signalmesh.db2db.v2':raise PermissionError('unsupported DB2DB protocol')
 if band not in ('source','target') or not sid or not fp or not nonce:raise PermissionError('incomplete DB2DB handshake')
 now=int(time.time())
 if abs(now-ts)>120:raise TimeoutError('DB2DB handshake timestamp outside allowed window')
 with _LOCK:
  c=_CELLS.get(sid)
  if not c:raise KeyError('unknown migration coordinate')
  if now>c['expires_at']:raise TimeoutError('migration coordinate expired')
  if c.get('requested_band') and c['requested_band']!=band:raise PermissionError('wrong reality band for this session')
  if c.get('requested_site_fingerprint') and c['requested_site_fingerprint']!=fp:raise PermissionError('site fingerprint does not match listener')
  if nonce in c['handshake_nonces']:raise PermissionError('replayed DB2DB handshake')
  signed='|'.join([sid,band,fp,nonce,str(ts)]);want=hmac.new(c['session_key'].encode(),signed.encode(),hashlib.sha256).hexdigest()
  if not hmac.compare_digest(want,str(env.get('signature',''))):raise PermissionError('invalid handshake signature')
  c['handshake_nonces'].add(nonce);c['connections'][band]={'site_fingerprint':fp,'plugin_version':env.get('plugin_version'),'joomla_version':env.get('joomla_version'),'connected_at':now,'protocol':protocol}
 return {'connected':True,'coordinate':sid,'band':band,'site_fingerprint':fp,'connected_at':now,'protocol':protocol}
def authorize_apply(sid,allowed=True):
 with _LOCK:
  c=_CELLS.get(sid)
  if not c:raise KeyError('unknown migration coordinate')
  c['allow_apply']=bool(allowed)
 return status(sid)
def status(sid):
 with _LOCK:c=_CELLS.get(sid)
 if not c:return None
 return {'coordinate':sid,'expires_at':c['expires_at'],'requested_band':c.get('requested_band'),'connections':{k:(None if not v else dict(v)) for k,v in c.get('connections',{}).items()},'bands':{k:(None if not v else {x:v[x] for x in ('site_fingerprint','joomla_version','artifact_sha256','received_at')}) for k,v in c['bands'].items()},'ready':all(c['bands'].values()),'map':c.get('map'),'verification':c.get('verification'),'allow_apply':bool(c.get('allow_apply'))}
def receive(env,replace=False):
 sid=str(env.get('session_id',''));band=str(env.get('band','')).lower()
 if band not in ('source','target'):raise ValueError('band must be source or target')
 with _LOCK:
  c=_CELLS.get(sid)
  if not c:raise KeyError('unknown migration coordinate')
  if time.time()>c['expires_at']:raise TimeoutError('migration coordinate expired')
  if c.get('requested_band') and c['requested_band']!=band:raise PermissionError('wrong reality band for this session')
  connection=c.get('connections',{}).get(band)
  if not connection:raise PermissionError('DB2DB handshake required before artifact transmission')
  if connection.get('site_fingerprint')!=str(env.get('site_fingerprint','')):raise PermissionError('artifact site does not match connected beacon')
  if c['bands'][band] and not replace:raise FileExistsError(f'{band} band is locked')
  secret=c['session_key']
 signed='|'.join([sid,band,str(env.get('artifact_sha256','')),str(env.get('manifest_sha256',''))]);want=hmac.new(secret.encode(),signed.encode(),hashlib.sha256).hexdigest()
 if not hmac.compare_digest(want,str(env.get('signature',''))):raise PermissionError('invalid envelope signature')
 raw=base64.b64decode(env['artifact']);nonce=base64.b64decode(env['nonce']);key=hashlib.sha256(secret.encode()).digest();cipher=env.get('cipher')
 if cipher=='aes-256-gcm':
  from cryptography.hazmat.primitives.ciphers.aead import AESGCM;raw=AESGCM(key).decrypt(nonce,raw+base64.b64decode(env['tag']),None)
 elif cipher=='xsalsa20-poly1305':
  from nacl.secret import SecretBox;raw=SecretBox(key).decrypt(raw,nonce)
 else:raise ValueError('unsupported cipher')
 got=hashlib.sha256(raw).hexdigest()
 if not hmac.compare_digest(got,str(env.get('artifact_sha256',''))):raise ValueError('artifact hash mismatch')
 d=ROOT/sid;d.mkdir(exist_ok=True);p=d/f'{band}.db';p.write_bytes(raw);rec={'site_fingerprint':env.get('site_fingerprint'),'joomla_version':env.get('joomla_version'),'artifact_sha256':got,'manifest':env.get('manifest',{}),'received_at':int(time.time()),'path':str(p)}
 with _LOCK:_CELLS[sid]['bands'][band]=rec;_CELLS[sid]['requested_band']=('target' if band=='source' and not _CELLS[sid]['bands']['target'] else None);_CELLS[sid]['requested_site_fingerprint']=None;_CELLS[sid]['map']=None;_CELLS[sid]['verification']=None
 return status(sid)
def _read_portable(path):
 db=sqlite3.connect(path);db.row_factory=sqlite3.Row;tables={r['name']:{'columns':json.loads(r['columns_json'] or '{}'),'rows':[]} for r in db.execute('select name,columns_json from migration_tables')}
 for r in db.execute('select table_name,row_no,row_json from migration_rows order by table_name,row_no'):
  if r['table_name'] in tables:tables[r['table_name']]['rows'].append(json.loads(r['row_json']))
 meta={r['k']:r['v'] for r in db.execute('select k,v from migration_meta')};db.close();return meta,tables
def _identity(row,columns):
 for k in ('id','extension_id','asset_id','user_id','catid','menuid','field_id','group_id'):
  if k in row and row[k] not in (None,''):return(k,str(row[k]))
 for k in ('alias','element','name','title','username','email'):
  if k in row and row[k] not in (None,''):return(k,str(row[k]))
 return('hash',hashlib.sha256(json.dumps(row,sort_keys=True,default=str).encode()).hexdigest())
def build_map(sid):
 with _LOCK:c=_CELLS.get(sid)
 if not c or not all(c['bands'].values()):raise ValueError('source and target are required')
 d=ROOT/sid;sm,source=_read_portable(d/'source.db');tm,target=_read_portable(d/'target.db');plan=[];blockers=[];stats={'preserve_target':0,'insert_source':0,'merge_source':0,'source_only_tables':0,'target_only_tables':0}
 for name in sorted(set(source)|set(target)):
  if name not in source:stats['target_only_tables']+=1;plan.append({'table':name,'action':'preserve-target-table'});continue
  if name not in target:stats['source_only_tables']+=1;blockers.append({'table':name,'reason':'table absent from target; extension/schema must exist before migration'});continue
  sc=set(source[name]['columns']);tc=set(target[name]['columns'])
  if not sc&tc:blockers.append({'table':name,'reason':'no compatible columns'});continue
  target_index={_identity(r,tc):r for r in target[name]['rows']}
  for row in source[name]['rows']:
   ident=_identity(row,sc);clean={k:v for k,v in row.items() if k in tc}
   if ident in target_index:
    patch={k:v for k,v in clean.items() if target_index[ident].get(k) in (None,'') and v not in (None,'')}
    if patch:plan.append({'table':name,'action':'hydrate','identity':list(ident),'values':patch});stats['merge_source']+=1
    else:stats['preserve_target']+=1
   else:plan.append({'table':name,'action':'insert','identity':list(ident),'values':clean});stats['insert_source']+=1
 result={'schema':'signalmesh.db2db.map.v1','source':sm,'target':tm,'stats':stats,'blockers':blockers,'operations':plan,'ready':not blockers,'created_at':int(time.time())};(d/'migration-map.json').write_text(json.dumps(result,indent=2,default=str))
 with _LOCK:_CELLS[sid]['map']={'ready':result['ready'],'stats':stats,'blockers':blockers,'operation_count':len(plan)}
 return result
def construct_merged(sid):
 plan=build_map(sid);d=ROOT/sid
 if plan['blockers']:raise ValueError('migration map has blockers; install/map required target schemas first')
 merged=d/'merged.db';merged.write_bytes((d/'target.db').read_bytes());db=sqlite3.connect(merged)
 for op in plan['operations']:
  if op['action'] not in ('insert','hydrate'):continue
  table=op['table'];ident_key,ident_val=op['identity'];values=op['values']
  if op['action']=='insert':row_no=db.execute('select coalesce(max(row_no),-1)+1 from migration_rows where table_name=?',(table,)).fetchone()[0];db.execute('insert into migration_rows(table_name,row_no,row_json) values(?,?,?)',(table,row_no,json.dumps(values,default=str)))
  else:
   for rn,rj in db.execute('select row_no,row_json from migration_rows where table_name=?',(table,)).fetchall():
    row=json.loads(rj)
    if _identity(row,set(row))==(ident_key,ident_val):row.update(values);db.execute('update migration_rows set row_json=? where table_name=? and row_no=?',(json.dumps(row,default=str),table,rn));break
 db.commit();db.close();return str(merged),plan
def verify(sid,actual_path=None):
 d=ROOT/sid;expected=d/'merged.db';actual=Path(actual_path) if actual_path else d/'target.db'
 if not expected.exists():raise ValueError('merged future has not been constructed')
 _,e=_read_portable(expected);_,a=_read_portable(actual);diffs=[]
 for name in sorted(set(e)|set(a)):
  er=e.get(name,{}).get('rows',[]);ar=a.get(name,{}).get('rows',[])
  if len(er)!=len(ar):diffs.append({'table':name,'expected_rows':len(er),'actual_rows':len(ar)})
 result={'schema':'signalmesh.db2db.verify.v1','verified':not diffs,'differences':diffs,'expected_sha256':_sha(expected),'actual_sha256':_sha(actual),'checked_at':int(time.time())};(d/'verification.json').write_text(json.dumps(result,indent=2))
 with _LOCK:_CELLS[sid]['verification']=result
 return result
def build_bundle(sid):
 d=ROOT/sid;merged,plan=construct_merged(sid);source=d/'source.db';target=d/'target.db';manifest={'schema':'signalmesh.db2db.bundle.v2','coordinate':sid,'source_sha256':_sha(source),'target_sha256':_sha(target),'merged_sha256':_sha(merged),'map_ready':plan['ready'],'operations':len(plan['operations']),'created_at':int(time.time())};(d/'manifest.json').write_text(json.dumps(manifest,indent=2));out=d/'migration-bundle.zip'
 with zipfile.ZipFile(out,'w',zipfile.ZIP_DEFLATED) as z:z.write(source,'01-source.db');z.write(target,'02-target.db');z.write(merged,'03-merged.db');z.write(d/'migration-map.json','migration-map.json');z.write(d/'manifest.json','manifest.json')
 return str(out),manifest