"""Owner-reviewed identity changes in one tenant transaction.

Matching contact text only suggests a pair. Transfers use exact IDs, retain
the old rows and snapshots, and never reuse either customer's credentials.
"""
import json
import re
import unicodedata
from fastapi import HTTPException
from . import customer_review_models as m
from .store import one, many, now, uid, digest, audit

CONTACT = ('id','name','email','phone','address','version','created_at')
CONTACT_FIELDS = ('name','email','phone','address')


def fail(message, status=409):
    raise HTTPException(status, message)


def packed(value):
    return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',', ':'))


def exists(db):
    return bool(db.execute("SELECT 1 FROM sqlite_schema WHERE type='table' AND name='customer_records'").fetchone())


def register(db, cid):
    if not one(db, 'SELECT customer_id FROM customer_records WHERE customer_id=?', (cid,)):
        db.execute("INSERT INTO customer_records VALUES(?,'active',NULL,'',1)", (cid,))
        record(db, cid, 'initialisation', 'initialisation', 'Observed existing record; no earlier history invented.')


def setup(db):
    all_tables=set(m.TABLES);base=all_tables-{'customer_merge_corrections'}
    present={r[0] for r in db.execute("SELECT name FROM sqlite_schema WHERE type='table'")} & all_tables
    if present and present not in (base,all_tables):raise ValueError('Customer identity migration is incomplete; recover a verified copy.')
    def require(ok,message):
        if not ok:raise ValueError(message)
    if not present:
        db.executescript(m.SCHEMA)
        for row in many(db,'SELECT id FROM customers'):register(db,row['id'])
        return
    from .customer_moves import legacy,migrate
    if legacy(db):
        if present==all_tables:raise ValueError('Partial customer correction migration.')
        migrate(db)
    if present==base:
        from .customer_recovery_v13 import validate as previous
        db.execute('SAVEPOINT customer_separation_upgrade')
        try:
            previous(db,require)
            db.execute(m.CORRECTION_SCHEMA)
            from .customer_recovery import validate
            validate(db,require)
            db.execute('RELEASE customer_separation_upgrade')
        except BaseException:
            db.execute('ROLLBACK TO customer_separation_upgrade');db.execute('RELEASE customer_separation_upgrade');raise
    else:
        from .customer_recovery import validate
        validate(db,require)


def contact(db, cid):
    c = one(db, 'SELECT '+','.join(CONTACT)+',portal_version,CASE WHEN portal_hash IS NULL THEN 0 ELSE 1 END AS portal_enabled FROM customers WHERE id=?', (cid,))
    if not c: fail('Customer not found in this business.', 404)
    r = one(db, 'SELECT * FROM customer_records WHERE customer_id=?', (cid,))
    if not r: fail('Customer record history is missing.')
    return {**c, 'state':r['state'], 'canonical_id':r['canonical_id'], 'notes':r['notes'], 'record_version':r['version']}


def active(db, cid):
    c = contact(db, cid)
    if c['state'] != 'active': fail('Use an active exact customer record. A retired alias cannot be reused or invited.')
    return c


def active_access(db, cid):
    # Pinned older formats have no identity-review table.
    return not exists(db) or bool(db.execute("SELECT 1 FROM customer_records WHERE customer_id=? AND state='active'", (cid,)).fetchone())


def snapshot(db, cid):
    return {'record':one(db, 'SELECT * FROM customer_records WHERE customer_id=?', (cid,)),
            'contact':one(db, 'SELECT '+','.join(CONTACT)+' FROM customers WHERE id=?', (cid,))}


def record(db, cid, action, actor, reason='', *, advance=False):
    if advance: db.execute('UPDATE customer_records SET version=version+1 WHERE customer_id=?', (cid,))
    value = snapshot(db, cid)
    db.execute('INSERT INTO customer_record_history(customer_id,version,action,actor,snapshot,reason,created_at) VALUES(?,?,?,?,?,?,?)',
               (cid,value['record']['version'],action,actor,packed(value),reason,now()))


def historical_owner(db, jid, cid):
    j = db.execute('SELECT customer_id FROM jobs WHERE id=?',(jid,)).fetchone()
    if not j: return False
    if j[0] == cid: return True
    if not exists(db): return False
    from .customer_moves import chain
    return any(cid in (r['from_customer_id'],r['to_customer_id']) for r in chain(db,jid))


def transfer_for(db, jid):
    from .customer_moves import chain
    rows=chain(db,jid) if exists(db) else []
    return rows[-1] if rows else None


def directory(db, q='', state='active'):
    result=[];needle=unicodedata.normalize('NFC',q).strip().casefold()
    for row in many(db, 'SELECT id FROM customers ORDER BY name COLLATE NOCASE,id'):
        c=contact(db,row['id'])
        if state!='all' and c['state']!=state: continue
        if needle and needle not in unicodedata.normalize('NFC',' '.join(c[k] for k in ('id','name','email','phone','address','notes'))).casefold(): continue
        c['job_count']=db.execute('SELECT COUNT(*) FROM jobs WHERE customer_id=?',(c['id'],)).fetchone()[0]
        result.append(c)
    return result


def pair_context(db, left, right):
    ids=sorted([left,right])
    if ids[0]==ids[1]: fail('Select two different customer records.',422)
    rows=[active(db,cid) for cid in ids]
    context=[{k:r[k] for k in (*CONTACT,'state','record_version')} for r in rows]
    return digest(':'.join(ids))[:32],context,digest(packed(context))


def suggestions(db):
    customers=directory(db)
    if len(customers)>500: fail('Review duplicates in a dataset of at most 500 active local customers.',422)
    old={r['id']:r for r in many(db,'SELECT * FROM customer_pair_decisions')}
    result=[];matching=0
    def normal(value): return unicodedata.normalize('NFC',value).strip().casefold()
    def phone(value): return re.sub(r'[^0-9]','',value)
    for n,a in enumerate(customers):
        for b in customers[n+1:]:
            reasons=[]
            for field in ('name','email'):
                if normal(a[field]) and normal(a[field])==normal(b[field]): reasons.append('Same '+field+' text')
            pa,pb=phone(a['phone']),phone(b['phone'])
            if len(pa)>=7 and pa==pb: reasons.append('Same phone digits')
            ordered=sorted((a,b),key=lambda c:c['id']);key=digest(':'.join(c['id'] for c in ordered))[:32];saved=old.get(key)
            if not reasons and not saved: continue
            matching+=1
            if len(result)>=200:continue
            context=[{k:r[k] for k in (*CONTACT,'state','record_version')} for r in ordered];hashed=digest(packed(context))
            result.append({'id':key,'customers':context,'reasons':reasons or ['Previously reviewed exact pair'],
                           'conflicts':[k for k in CONTACT_FIELDS if a[k]!=b[k]],
                           'context_hash':hashed,'version':saved['version'] if saved else 0,
                           'state':saved['state'] if saved else 'review','reason':saved['reason'] if saved else '',
                           'details_changed':bool(saved and saved['context_hash']!=hashed)})
    return {'pairs':result,'scanned':len(customers),'matching_pairs':matching,'truncated':matching>len(result),'notice':'Similar text is a review suggestion only. It does not establish identity or grant access.'}


def decide_pair(db,payload,actor):
    key,context,hashed=pair_context(db,payload.left_id,payload.right_id)
    old=one(db,'SELECT * FROM customer_pair_decisions WHERE id=?',(key,))
    if payload.context_hash!=hashed or payload.version!=(old['version'] if old else 0): fail('The pair changed. Reload and review both exact records again.')
    row={'id':key,'left_id':context[0]['id'],'right_id':context[1]['id'],'state':payload.state,
         'version':payload.version+1,'reason':payload.reason,'context_hash':hashed,'context':packed(context),'updated_at':now()}
    db.execute('INSERT INTO customer_pair_decisions VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET state=excluded.state,version=excluded.version,reason=excluded.reason,context_hash=excluded.context_hash,context=excluded.context,updated_at=excluded.updated_at',tuple(row.values()))
    db.execute('INSERT INTO customer_pair_history(pair_id,version,actor,snapshot,created_at) VALUES(?,?,?,?,?)',(key,row['version'],actor,packed(row),now()))
    return {'id':key,'version':row['version'],'state':row['state']}


def dependencies(db):
    # A conservative fence includes saved work in this business. Credentials are
    # never copied into a preview/history; their safe versions fence re-invites.
    tables=('jobs','quotes','quote_links','quote_responses','invoices','payments','payment_details','payment_reversals',
            'payment_stages','document_sources','field_logs','field_sessions','field_intervals','field_time_links',
            'job_materials','time_entries','staff_updates','calendar_bookings','job_photos','gallery_projects','recurring_contracts',
            'recurrence_occurrences','feedback_requests','feedback_consents','feedback_publications',
            'job_workflow','opportunity_reviews','opportunity_settings','customer_records','customer_transfer_jobs',
            'customer_pair_decisions','customer_completion_runs','customer_transfers','customer_links','job_workflow_history',
            'customer_record_history','customer_merge_corrections','document_events','photo_history','gallery_history','field_log_history','field_history',
            'source_history','opportunity_history','portal_events')
    values={table:many(db,'SELECT * FROM '+table+' ORDER BY rowid') for table in tables}
    values['customers']=many(db,'SELECT '+','.join(CONTACT)+',portal_version,CASE WHEN portal_hash IS NULL THEN 0 ELSE 1 END AS portal_enabled FROM customers ORDER BY id')
    values['portal_tokens']=many(db,'SELECT id,customer_id,kind,account_version,expires,used_at,created_at FROM portal_tokens ORDER BY id')
    values['portal_accounts']=many(db,'SELECT customer_id,state,version,updated_at FROM portal_accounts ORDER BY customer_id')
    return digest(packed(values))


def work_blockers(db,jid):
    from .workflow import today
    blockers=[]
    checks=[("SELECT 1 FROM jobs WHERE id=? AND status='in_progress'",(jid,),'Work currently in progress'),
            ("SELECT 1 FROM field_sessions WHERE job_id=? AND state NOT IN ('converted','discarded')",(jid,),'Unfinished field timer or review'),
            ("SELECT 1 FROM field_logs WHERE job_id=? AND reviewed=0 AND archived=0",(jid,),'Unreviewed field note'),
            ("SELECT 1 FROM recurring_contracts WHERE job_id=? AND state IN ('active','paused')",(jid,),'Active or paused repeat work'),
            ('SELECT 1 FROM calendar_bookings WHERE job_id=? AND archived=0 AND end_date>=?',(jid,today()),'Current or future booking')]
    for sql,args,message in checks:
        if db.execute(sql,args).fetchone(): blockers.append(message)
    return blockers


def work_inventory(db,source,target,jobs,kind='jobs'):
    errors=[]
    if len(jobs)>50: errors.append('Review at most 50 source jobs in this local stage.')
    if db.execute('SELECT COUNT(*) FROM customer_transfers').fetchone()[0]>=5000 or db.execute('SELECT COUNT(*) FROM customer_transfer_jobs').fetchone()[0]+len(jobs)>10000:errors.append('This local correction history has reached its supported bound.')
    affected_jobs=many(db,'SELECT id FROM jobs WHERE customer_id IN (?,?) ORDER BY id',(source['id'],target['id']))
    documents=[];associations={}
    tables=('job_materials','time_entries','staff_updates','calendar_bookings','job_photos','field_logs','field_sessions','document_sources','recurring_contracts','feedback_requests','opportunity_reviews')
    for j in jobs:
        if transfer_for(db,j['id']) and kind=='merge': errors.append(j['id']+': use selected-job correction for previously transferred work; further whole-customer merging is deferred.')
        errors.extend(j['id']+': '+why for why in work_blockers(db,j['id']))
        for table in ('quotes','invoices','job_photos','field_logs','field_sessions','document_sources'):
            if db.execute('SELECT 1 FROM '+table+' WHERE job_id=? AND customer_id<>?',(j['id'],source['id'])).fetchone():errors.append(j['id']+': '+table+' has an inconsistent customer association; recover or review the source evidence first.')
        associations[j['id']]={t:[r['id'] for r in many(db,'SELECT id FROM '+t+' WHERE job_id=? ORDER BY id',(j['id'],))] for t in tables}
        for kind,table in (('quote','quotes'),('invoice','invoices')):
            for d in many(db,'SELECT * FROM '+table+' WHERE job_id=? ORDER BY id',(j['id'],)):
                saved=json.loads(d['snapshot'])
                documents.append({'kind':kind,'id':d['id'],'job_id':j['id'],'customer_id':d['customer_id'],
                                  'reference':d['reference'],'status':d['status'],'version':d['version'],
                                  'total_pence':d['total_pence'],'snapshot_sha256':digest(d['snapshot']),
                                  'original_customer':saved['customer'],'created_at':d['created_at']})
    aid=[j['id'] for j in affected_jobs]
    paused={
        'portal_customer_ids':[source['id'],target['id']],
        'photo_ids':[p['id'] for p in many(db,'SELECT id,job_id FROM job_photos') if p['job_id'] in aid],
        'gallery_ids':[p['id'] for p in many(db,'SELECT id,source_job_id FROM gallery_projects WHERE published IS NOT NULL') if p['source_job_id'] in aid],
        'feedback_ids':[r['id'] for r in many(db,'SELECT id,customer_id FROM feedback_requests') if r['customer_id'] in (source['id'],target['id'])]}
    paused['photo_versions']={p['id']:p['version'] for p in many(db,'SELECT id,version FROM job_photos') if p['id'] in paused['photo_ids']}
    paused['gallery_versions']={p['id']:p['version'] for p in many(db,'SELECT id,version FROM gallery_projects') if p['id'] in paused['gallery_ids']}
    from .customer_moves import chain
    return {'jobs':jobs,'associations':associations,'documents':documents,'sharing_to_pause':paused,'errors':errors,
            'ownership_chains':{j['id']:chain(db,j['id']) for j in jobs}}


def transfer_plan(db,payload):
    source=active(db,payload.source_id);target=active(db,payload.target_id)
    if source['version']!=payload.source_version or target['version']!=payload.target_version: fail('Customer details changed. Reload both records before reviewing.')
    jobs=many(db,'SELECT * FROM jobs WHERE customer_id=? ORDER BY id',(source['id'],))
    if payload.kind=='jobs':
        by_id={j['id']:j for j in jobs}
        if any(jid not in by_id for jid in payload.job_ids): fail('Selected job not found on this exact source customer.',404)
        jobs=[by_id[jid] for jid in sorted(payload.job_ids)]
    inventory=work_inventory(db,source,target,jobs,payload.kind)
    errors=inventory['errors']
    if payload.kind=='merge' and db.execute('SELECT 1 FROM customer_records WHERE canonical_id=?',(source['id'],)).fetchone(): errors.append('This source retains merged aliases. Transitive customer merging needs a separate review design.')
    chosen={k:(source if choice=='source' else target)[k] for k,choice in payload.fields.model_dump().items()}
    def safe(c): return {k:c[k] for k in (*CONTACT,'state','notes','record_version','canonical_id')}
    plan={'selection':payload.model_dump(),'source':safe(source),'target':safe(target),'retained_fields':chosen,
          **inventory,
          'dependency_hash':dependencies(db),'errors':errors,'can_confirm':not errors,
          'effect':'Move only the listed current associations. Keep both records, job IDs, original documents, amounts and histories. Revoke both portal accounts, pause listed sharing, and exclude moved jobs from automatic follow-ups pending owner review. Retired repeat-work definitions remain historical and cannot restart under another customer. No one-click undo.'}
    from .customer_moves import chain
    plan['ownership_chains']={j['id']:chain(db,j['id']) for j in jobs}
    plan['effect']+=' Every correction appends a new move, including a return to an earlier active customer. Earlier access and permissions stay revoked.'
    plan['preview_hash']=digest(packed(plan))
    return plan


def pause_sharing(db,source,target,s):
    from .portal_access import revoke
    from .feedback import invalidate
    from .photos import photo_snapshot, project_snapshot
    for cid in dict.fromkeys((source,target)):
        revoke(db,cid,s['subject'])
        invalidate(db,customer_id=cid,actor=s['subject'],detail='Reviewed customer identity change; old permission remains with the original record.')
    for p in many(db,'SELECT p.* FROM job_photos p JOIN jobs j ON j.id=p.job_id WHERE j.customer_id IN (?,?)',(source,target)):
        if p['customer_share'] is not None or p['reviewed']:
            db.execute('UPDATE job_photos SET customer_share=NULL,reviewed=0,version=version+1,updated_at=? WHERE id=?',(now(),p['id']))
            photo_snapshot(db,p['id'],s,'photo.permission_suspended')
    for p in many(db,'SELECT g.* FROM gallery_projects g JOIN jobs j ON j.id=g.source_job_id WHERE j.customer_id IN (?,?) AND g.published IS NOT NULL',(source,target)):
        db.execute('UPDATE gallery_projects SET published=NULL,version=version+1,updated_at=? WHERE id=?',(now(),p['id']))
        project_snapshot(db,p['id'],s,'project.withdraw')


def execute_transfer(db,payload,s,operation_id=None):
    if operation_id:
        if one(db,'SELECT id FROM customer_merge_corrections WHERE id=?',(operation_id,)):fail('This operation ID already belongs to a merge correction.')
        prior=one(db,'SELECT * FROM customer_transfers WHERE id=?',(operation_id,))
        if prior:
            saved=json.loads(prior['plan'])
            exact={**saved['selection'],'preview_hash':prior['preview_hash'],'reason':prior['reason'],'confirmed':True}
            if prior['actor']!=s['subject'] or exact!=payload.model_dump():fail('This operation ID already belongs to a different reviewed correction.')
            return json.loads(prior['result'])
    selection=m.TransferSelection.model_validate(payload.model_dump(include=set(m.TransferSelection.model_fields)))
    plan=transfer_plan(db,selection)
    if plan['preview_hash']!=payload.preview_hash: fail('Saved work or access changed since this preview. Keep your choices and review again.')
    if plan['errors']: fail('This transfer has unresolved blockers: '+ '; '.join(plan['errors']))
    tid=operation_id or uid();stamp=now();source=payload.source_id;target=payload.target_id
    result={'id':tid,'kind':payload.kind,'source_id':source,'target_id':target,'job_ids':[j['id'] for j in plan['jobs']],
            'source_version':payload.source_version+1,'target_version':payload.target_version+1,'portal_access':'Both revoked; fresh exact-customer invitation required.'}
    db.execute('INSERT INTO customer_transfers VALUES(?,?,?,?,?,?,?,?,?,?)',(tid,payload.kind,source,target,s['subject'],payload.reason,payload.preview_hash,packed(plan),packed(result),stamp))
    pause_sharing(db,source,target,s)
    from .workflow import metadata,record_job
    from .photos import photo_snapshot
    from .fieldwork import log_event,event as field_event
    from .billing_sources import log as source_event
    from .documents import event as document_event
    for j in plan['jobs']:
        jid=j['id'];prior=plan['ownership_chains'][jid]
        db.execute('INSERT INTO customer_transfer_jobs VALUES(?,?,?,?,?,?,?,?)',(jid,tid,source,target,j['version'],j['version']+1,len(prior)+1,prior[-1]['transfer_id'] if prior else None))
        db.execute('UPDATE jobs SET customer_id=?,version=version+1 WHERE id=?',(target,jid))
        meta=metadata(db,jid);meta['followup_excluded']=True
        db.execute('UPDATE job_workflow SET content=?,feedback_version=?,version=version+1 WHERE job_id=?',(json.dumps(meta),j['version']+1,jid));record_job(db,jid,s['subject'])
        for table in ('quotes','invoices'):
            for before in many(db,'SELECT * FROM '+table+' WHERE job_id=? ORDER BY id',(jid,)):
                db.execute('UPDATE '+table+' SET customer_id=?,version=version+1 WHERE id=?',(target,before['id']))
                document_event(db,'quote' if table=='quotes' else 'invoice',before['id'],'customer.transferred',before,one(db,'SELECT * FROM '+table+' WHERE id=?',(before['id'],)),payload.reason,s['subject'])
        for table in ('job_photos','field_logs','field_sessions'):
            for row in many(db,'SELECT id FROM '+table+' WHERE job_id=? ORDER BY id',(jid,)):
                db.execute('UPDATE '+table+' SET customer_id=?,version=version+1,updated_at=? WHERE id=?',(target,stamp,row['id']))
                if table=='job_photos': photo_snapshot(db,row['id'],s,'photo.customer_transferred')
                elif table=='field_logs': log_event(db,row['id'],'customer.transferred',s,payload.reason)
                else: field_event(db,row['id'],'customer.transferred',s,{'transfer_id':tid})
        for row in many(db,'SELECT * FROM document_sources WHERE job_id=? ORDER BY id',(jid,)):
            db.execute('UPDATE document_sources SET customer_id=?,updated_at=? WHERE id=?',(target,stamp,row['id']))
            source_event(db,one(db,'SELECT * FROM document_sources WHERE id=?',(row['id'],)),'customer.transferred',s['subject'])
        db.execute('INSERT INTO customer_links(job_id,from_customer_id,to_customer_id,actor,created_at) VALUES(?,?,?,?,?)',(jid,source,target,s['subject'],stamp))
    if payload.kind=='merge':
        db.execute("UPDATE customer_records SET state='merged',canonical_id=? WHERE customer_id=?",(target,source))
        values=plan['retained_fields'];db.execute('UPDATE customers SET name=?,email=?,phone=?,address=? WHERE id=?',(*(values[k] for k in CONTACT_FIELDS),target))
    for cid,role in ((source,'source'),(target,'target')):
        db.execute('UPDATE customers SET version=version+1 WHERE id=?',(cid,))
        record(db,cid,payload.kind+'.'+role,s['subject'],payload.reason,advance=True)
    audit(db,'customer.'+payload.kind+'.reviewed',tid)
    return result


def completion_plan(db,payload):
    from .workflow import metadata,today
    rows=[];seen=set()
    for n,item in enumerate(payload.rows,1):
        errors=[];c=contact(db,item.customer_id)
        j=one(db,'SELECT * FROM jobs WHERE id=? AND customer_id=?',(item.job_id,item.customer_id))
        inv=one(db,'SELECT id,reference,job_id,customer_id,status FROM invoices WHERE id=? AND customer_id=?',(item.invoice_id,item.customer_id))
        if not j or not inv or inv['job_id']!=item.job_id: fail('An exact job/invoice link was not found in this business.',404)
        if c['state']!='active': errors.append('Restore an archived customer before reviewing its work.')
        if j['version']!=item.job_version or c['version']!=item.customer_version: errors.append('Customer or job version changed.')
        meta=metadata(db,j['id'])
        if j['status']!='completed' or meta['completed_date'] or meta['archived']: errors.append('Only an unarchived, already-completed job with an unknown date is eligible.')
        if inv['status']!='issued': errors.append('Select a currently issued invoice for this exact job.')
        if item.completed_date>today(): errors.append('A recorded completion date cannot be in the future.')
        if item.job_id in seen: errors.append('Review each job once, even when it has multiple invoices.')
        seen.add(item.job_id);errors.extend(work_blockers(db,item.job_id))
        rows.append({'row':n,'input':item.model_dump(),'customer_name':c['name'],'service':j['service'],'invoice_reference':inv['reference'],'errors':errors})
    plan={'rows':rows,'dependency_hash':dependencies(db),'can_confirm':all(not r['errors'] for r in rows),
          'effect':'Record the explicitly evidenced completion dates on existing exact jobs. No job or receipt is invented; payment dates are not completion evidence. Material-date changes suspend old feedback permission.'}
    plan['preview_hash']=digest(packed(plan));return plan


def complete_dates(db,payload,s):
    from .workflow import job,metadata,edit_job
    from .workflow_models import WorkflowEdit
    plan=completion_plan(db,payload)
    if plan['preview_hash']!=payload.preview_hash or not plan['can_confirm']: fail('The completion review changed or contains row errors. Review again; no rows were saved.')
    results=[]
    for row in payload.rows:
        j=job(db,row.job_id);meta=metadata(db,row.job_id);meta['completed_date']=row.completed_date
        results.append(edit_job(db,j['id'],WorkflowEdit(version=j['version'],status=j['status'],notes=j['notes'],scheduled_date=j['scheduled_date'],metadata=meta),s['subject']))
    rid=uid();result={'id':rid,'jobs':results}
    db.execute('INSERT INTO customer_completion_runs VALUES(?,?,?,?,?,?,?)',(rid,s['subject'],payload.reason,payload.preview_hash,packed(plan),packed(result),now()))
    return result