"""Strict format-11 customer identity, transfer and completion provenance.""" import json import re import sqlite3 from datetime import datetime from .customer_review import CONTACT,CONTACT_FIELDS,packed,historical_owner from .customer_review_models import TABLES,TransferSelection,CompletionRow from .customers import Contact from .store import digest def validate(db,require): previous=db.row_factory;db.row_factory=sqlite3.Row try: rows=lambda table:[dict(r) for r in db.execute('SELECT * FROM '+table+' ORDER BY rowid')] customers={c['id']:c for c in rows('customers')};jobs={j['id']:j for j in rows('jobs')} records={r['customer_id']:r for r in rows('customer_records')} transfers={t['id']:t for t in rows('customer_transfers')};moves={r['job_id']:r for r in rows('customer_transfer_jobs')} hid=lambda v:isinstance(v,str) and re.fullmatch(r'[a-f0-9]{32}',v) is not None hashid=lambda v:isinstance(v,str) and re.fullmatch(r'[a-f0-9]{64}',v) is not None integer=lambda v:type(v) is int and v>=1 def stamp(v):require(isinstance(v,str) and datetime.fromisoformat(v).tzinfo is not None,'Invalid customer history date.') def reason(v):require(isinstance(v,str) and 5<=len(v.strip())<=1000,'Missing customer review reason.') require(set(records)==set(customers) and all(hid(cid) for cid in customers),'Customer identity baseline is missing or invalid.') require(len(transfers)<=5000 and len(moves)<=10000,'Customer transfer history exceeds supported bounds.') histories={cid:[] for cid in customers} for h in rows('customer_record_history'): require(h['customer_id'] in customers,'Customer history has no exact record.');stamp(h['created_at']);reason(h['reason']) value=json.loads(h['snapshot']) require(set(value)=={'record','contact'} and set(value['record'])==set(records[h['customer_id']]) and set(value['contact'])==set(CONTACT),'Unexpected customer history fields.') r,c=value['record'],value['contact'];cid=h['customer_id'] require(r['customer_id']==c['id']==cid and r['version']==h['version'] and integer(c['version']),'Customer history identity/version mismatch.') require(r['state'] in ('active','archived','merged') and isinstance(r['notes'],str) and len(r['notes'])<=6000,'Invalid customer record state/notes.') require((r['canonical_id'] is not None)==(r['state']=='merged') and (r['canonical_id'] is None or r['canonical_id'] in customers and r['canonical_id']!=cid),'Invalid customer alias target.') Contact.model_validate({k:c[k] for k in CONTACT_FIELDS});stamp(c['created_at']) histories[cid].append((h,value)) actions={'initialisation','notes','contact','archive','restore','merge.source','merge.target','jobs.source','jobs.target'} for cid,r in records.items(): entries=histories[cid] require(integer(r['version']) and [h['version'] for h,v in entries]==list(range(1,r['version']+1)),'Customer record history is incomplete.') require(entries[-1][1]=={'record':r,'contact':{k:customers[cid][k] for k in CONTACT}},'Customer details differ from their retained history.') first=entries[0];require(first[0]['action']==first[0]['actor']=='initialisation' and first[1]['record']=={'customer_id':cid,'state':'active','canonical_id':None,'notes':'','version':1},'Invalid observed customer baseline.') for n,(h,v) in enumerate(entries): require(h['action'] in actions,'Unknown customer history action.') if not n:continue old=entries[n-1][1];a=h['action'];rc=v['record'];contact=v['contact'] require(a!='initialisation' and contact['created_at']==old['contact']['created_at'],'Customer identity/creation history was replaced.') require(contact['version']==old['contact']['version']+(a!='notes'),'Customer contact versions are not contiguous.') if a=='notes':require(contact==old['contact'] and all(rc[k]==old['record'][k] for k in ('state','canonical_id')),'Private notes changed identity or state.') else:require(rc['notes']==old['record']['notes'],'Private notes changed outside a notes action.') if a in ('contact','notes'):require(rc['state']==old['record']['state']=='active' and rc['canonical_id'] is None,'Inactive customer was edited.') if a in ('archive','restore'): require(rc['state']==('archived' if a=='archive' else 'active') and old['record']['state']==('active' if a=='archive' else 'archived'),'Invalid archive/restore transition.') require(all(contact[k]==old['contact'][k] for k in CONTACT_FIELDS),'Archive/restore changed contact details.') if a.endswith(('.source','.target')): kind,side=a.split('.') related=[t for t in transfers.values() if t['kind']==kind and t[side+'_id']==cid and json.loads(t['plan'])[side]['version']==old['contact']['version']] require(len(related)==1 and related[0]['actor']==h['actor'] and related[0]['reason']==h['reason'],'Customer state has no unique reviewed transfer.') t=related[0];plan=json.loads(t['plan']) require(old['record']['state']=='active' and rc['state']==('merged' if a=='merge.source' else 'active') and rc['canonical_id']==(t['target_id'] if a=='merge.source' else None),'Transfer alias/state transition is inconsistent.') selected=plan['retained_fields'] if a=='merge.target' else old['contact'] require(all(contact[k]==selected[k] for k in CONTACT_FIELDS),'Transfer retained different contact fields.') if r['state']=='merged': require(records[r['canonical_id']]['state']!='merged','Merged aliases cannot chain or cycle.') require(not any(j['customer_id']==cid for j in jobs.values()),'A retired alias still owns current work.') require(customers[cid]['portal_hash'] is None,'A retired alias has active customer access.') a=db.execute('SELECT state,password FROM portal_accounts WHERE customer_id=?',(cid,)).fetchone() require(a is None or (a['state']=='revoked' and not a['password']),'A retired alias can still authenticate.') plan_keys={'selection','source','target','retained_fields','jobs','associations','documents','sharing_to_pause','dependency_hash','errors','can_confirm','effect','preview_hash'} safe_keys=set(CONTACT)|{'state','notes','record_version','canonical_id'} assoc_tables=('job_materials','time_entries','staff_updates','calendar_bookings','job_photos','field_logs','field_sessions','document_sources','recurring_contracts','feedback_requests','opportunity_reviews') for tid,t in transfers.items(): require(hid(tid) and t['kind'] in ('merge','jobs') and t['source_id'] in customers and t['target_id'] in customers and t['source_id']!=t['target_id'],'Invalid exact transfer identity.') stamp(t['created_at']);reason(t['reason']);plan=json.loads(t['plan']);result=json.loads(t['result']) require(set(plan)==plan_keys and plan['preview_hash']==t['preview_hash']==digest(packed({k:v for k,v in plan.items() if k!='preview_hash'})),'Transfer preview hash/shape is inconsistent.') selection=TransferSelection.model_validate(plan['selection']);require(selection.model_dump()==plan['selection'],'Noncanonical transfer selection.') require(selection.source_id==t['source_id'] and selection.target_id==t['target_id'] and selection.kind==t['kind'] and plan['can_confirm'] is True and plan['errors']==[] and hashid(plan['dependency_hash']),'Transfer was not confirmable.') for side in ('source','target'): saved=plan[side];require(set(saved)==safe_keys and saved['id']==t[side+'_id'] and saved['version']==getattr(selection,side+'_version') and saved['state']=='active' and saved['canonical_id'] is None,'Transfer source/target is inconsistent.') require(any(v['contact']=={k:saved[k] for k in CONTACT} and v['record']=={'customer_id':saved['id'],'state':'active','canonical_id':None,'notes':saved['notes'],'version':saved['record_version']} for _,v in histories[saved['id']]),'Transfer source/target has no retained version.') require(plan['retained_fields']=={k:plan[side][k] for k,side in selection.fields.model_dump().items()},'Transfer contact choices changed.') selected=plan['jobs'];ids=[j['id'] for j in selected] require(len(ids)==len(set(ids))<=50 and set(plan['associations'])==set(ids),'Transfer job inventory is inconsistent.') if t['kind']=='jobs':require(ids==sorted(selection.job_ids),'Transfer moved unselected jobs.') expected={'id':tid,'kind':t['kind'],'source_id':t['source_id'],'target_id':t['target_id'],'job_ids':ids,'source_version':selection.source_version+1,'target_version':selection.target_version+1,'portal_access':'Both revoked; fresh exact-customer invitation required.'} require(result==expected,'Transfer result differs from its reviewed selection.') require({jid for jid,move in moves.items() if move['transfer_id']==tid}==set(ids),'Transfer job ledger is incomplete.') for before in selected: jid=before['id'];require(jid in jobs and set(before)==set(jobs[jid]) and before['customer_id']==t['source_id'],'Transfer has an invalid original job.') require(moves[jid]=={'job_id':jid,'transfer_id':tid,'from_customer_id':t['source_id'],'to_customer_id':t['target_id'],'before_version':before['version'],'after_version':before['version']+1} and jobs[jid]['customer_id']==t['target_id'] and jobs[jid]['version']>=before['version']+1,'Current job ownership lacks its exact reviewed transfer.') from .workflow import JOB_FIELDS h=db.execute('SELECT actor,snapshot FROM job_workflow_history WHERE job_id=? AND job_version=?',(jid,before['version']+1)).fetchall() require(len(h)==1 and h[0]['actor']==t['actor'],'Transfer has no unique workflow event.') v=json.loads(h[0]['snapshot']);require(v['job']=={**{k:before[k] for k in JOB_FIELDS},'customer_id':t['target_id'],'version':before['version']+1} and v['feedback_version']==before['version']+1 and v['metadata']['followup_excluded'] is True,'Transfer altered work details or failed to exclude automatic follow-ups.') assoc=plan['associations'][jid];require(set(assoc)==set(assoc_tables),'Unsupported transfer associations.') for table,names in assoc.items(): require(isinstance(names,list) and len(names)==len(set(names)),'Duplicate transfer association.') for rid in names: row=db.execute('SELECT * FROM '+table+' WHERE id=? AND job_id=?',(rid,jid)).fetchone();require(row is not None,'A transferred association was lost or moved to another job.') for r in db.execute('SELECT state,customer_id FROM recurring_contracts WHERE job_id=?',(jid,)): require(r['state']=='retired' and r['customer_id']==t['source_id'],'Transferred repeat work is not a retained retired definition.') for d in plan['documents']: require(d['kind'] in ('quote','invoice') and d['job_id'] in ids,'Transfer document belongs to another job.') table='quotes' if d['kind']=='quote' else 'invoices';current=db.execute('SELECT * FROM '+table+' WHERE id=?',(d['id'],)).fetchone() require(current is not None and current['job_id']==d['job_id'] and current['customer_id']==t['target_id'] and current['version']>=d['version']+1 and d['customer_id']==t['source_id'],'Transferred document ownership/version differs.') require(current['reference']==d['reference'] and current['created_at']==d['created_at'] and json.loads(current['snapshot'])['customer']==d['original_customer'],'Original document identity/reference changed during transfer.') if d['status']!='draft':require(digest(current['snapshot'])==d['snapshot_sha256'] and current['total_pence']==d['total_pence'],'A sealed transferred document changed its saved contents/value.') events=db.execute("SELECT before_value,after_value,actor FROM document_events WHERE kind=? AND document_id=? AND action='customer.transferred'",(d['kind'],d['id'])).fetchall() require(len(events)==1 and events[0]['actor']==t['actor'],'Transfer document has no unique audited change.') old,new=(json.loads(events[0][k]) for k in ('before_value','after_value')) require(new=={**old,'customer_id':t['target_id'],'version':old['version']+1} and old['customer_id']==t['source_id'] and digest(old['snapshot'])==d['snapshot_sha256'],'Transfer rewrote document evidence.') require(set(plan['sharing_to_pause'])=={'portal_customer_ids','photo_ids','gallery_ids','feedback_ids','photo_versions','gallery_versions'} and plan['sharing_to_pause']['portal_customer_ids']==[t['source_id'],t['target_id']],'Transfer access scope is inconsistent.') ended=next(h['created_at'] for h,v in histories[t['source_id']] if h['action']==t['kind']+'.source' and v['contact']['version']==selection.source_version+1) in_window=lambda row:t['created_at']<=row['created_at']<=ended and row.get('actor',row.get('actor_id'))==t['actor'] for cid in (t['source_id'],t['target_id']): require(any(in_window(h) and h['customer_id']==cid and h['action']=='revoked' for h in rows('portal_events')),'Transfer did not retain both portal revocations.') for pid in plan['sharing_to_pause']['photo_ids']: history=[h for h in rows('photo_history') if h['photo_id']==pid] before=[json.loads(h['snapshot']) for h in history if json.loads(h['snapshot'])['version']==plan['sharing_to_pause']['photo_versions'][pid]] require(len(before)==1,'Transferred photo has no original reviewed version.') if before[-1]['reviewed'] or before[-1]['customer_share'] is not None: require(any(in_window(h) and h['action']=='photo.permission_suspended' and json.loads(h['snapshot'])['version']==before[0]['version']+1 and json.loads(h['snapshot'])['customer_share'] is None and not json.loads(h['snapshot'])['reviewed'] for h in history),'Transfer has no recorded suspension of photo permission.') for gid in plan['sharing_to_pause']['gallery_ids']: require(any(in_window(h) and h['project_id']==gid and h['action']=='project.withdraw' and h['version']==plan['sharing_to_pause']['gallery_versions'][gid]+1 and json.loads(h['snapshot'])['published'] is None for h in rows('gallery_history')),'Transfer has no recorded gallery withdrawal.') for jid,assoc in plan['associations'].items(): for table,key,kind,nested,action in [('photo_history','photo_id','job_photos',None,'photo.customer_transferred'),('field_log_history','log_id','field_logs',None,'customer.transferred'),('field_history','session_id','field_sessions','session','customer.transferred'),('source_history','claim_id','document_sources',None,'customer.transferred')]: for rid in assoc[kind]: hs=[h for h in rows(table) if h[key]==rid];changes=[] for prior_event,current in zip(hs,hs[1:]): old,new=json.loads(prior_event['snapshot']),json.loads(current['snapshot']) if nested:old,new=old[nested],new[nested] if old['customer_id']==new['customer_id']:continue changes.append(current) require(current['action']==action and in_window(current) and old['customer_id']==t['source_id'] and new['customer_id']==t['target_id'],'Associated history changed customer outside its reviewed transfer.') mutable={'customer_id','updated_at'}|({'version'} if kind!='document_sources' else set()) require(all(old[k]==new[k] for k in old if k not in mutable),'Associated transfer rewrote its original evidence.') if kind!='document_sources':require(new['version']==old['version']+1,'Associated transfer skipped a version.') require(len(changes)==1,'Associated record lacks its original customer and exact transfer event.') require(all(m['transfer_id'] in transfers for m in moves.values()),'Transfer ledger has an orphan operation.') # Every current operational binding is exact. Older history can retain # the former customer only through the immutable transfer ledger. for table in ('quotes','invoices','job_photos','field_logs','field_sessions','document_sources'): for row in rows(table):require(row['job_id'] in jobs and row['customer_id']==jobs[row['job_id']]['customer_id'],'Unrecorded current ownership change.') for jid in jobs: hs=[json.loads(h['snapshot']) for h in rows('job_workflow_history') if h['job_id']==jid] for old,new in zip(hs,hs[1:]): if old['job']['customer_id']==new['job']['customer_id']:continue a,b=old['job']['customer_id'],new['job']['customer_id'] require(db.execute('SELECT 1 FROM customer_links WHERE job_id=? AND from_customer_id=? AND to_customer_id=?',(jid,a,b)).fetchone() is not None,'Job ownership changed without a retained link review.') if jid in moves:require(moves[jid]['after_version']==new['job']['version'] or new['job']['version']