"""Local sales records: sealed documents, one receipt ledger, explicit history.

No email, payment service, credit-note accounting or tax compliance is implied.
All mutations run inside the caller's idempotent BEGIN IMMEDIATE transaction.
"""
from datetime import date, datetime
from zoneinfo import ZoneInfo
import json
import re
from typing import Literal
from fastapi import HTTPException, Request
from pydantic import Field, StrictBool, field_validator
from .models import Strict, Quote, Version, Payment
from .store import one, many, uid, now, audit

INVOICE_SCHEMA = '''CREATE TABLE invoices_v5(id TEXT PRIMARY KEY, reference TEXT UNIQUE NOT NULL,
 quote_id TEXT REFERENCES quotes(id), job_id TEXT NOT NULL REFERENCES jobs(id),
 customer_id TEXT NOT NULL REFERENCES customers(id), snapshot TEXT NOT NULL,
 total_pence INTEGER NOT NULL CHECK(total_pence>0), created_at TEXT NOT NULL,
 status TEXT NOT NULL DEFAULT 'draft' CHECK(status IN ('draft','issued','void','cancelled')),
 version INTEGER NOT NULL DEFAULT 1, issued_at TEXT,
 replaces_id TEXT UNIQUE REFERENCES invoices(id))'''

STATEMENTS = [
 "CREATE UNIQUE INDEX IF NOT EXISTS invoice_active_quote ON invoices(quote_id) WHERE status IN ('draft','issued')",
 '''CREATE TABLE IF NOT EXISTS quote_links(quote_id TEXT PRIMARY KEY REFERENCES quotes(id),
 root_id TEXT NOT NULL REFERENCES quotes(id), previous_id TEXT UNIQUE REFERENCES quotes(id),
 revision INTEGER NOT NULL, current INTEGER NOT NULL CHECK(current IN (0,1)), shared_at TEXT, accepted_at TEXT)''',
 'CREATE UNIQUE INDEX IF NOT EXISTS quote_current ON quote_links(root_id) WHERE current=1',
 '''CREATE TABLE IF NOT EXISTS document_events(id INTEGER PRIMARY KEY AUTOINCREMENT,
 kind TEXT NOT NULL, document_id TEXT NOT NULL, action TEXT NOT NULL, before_value TEXT,
 after_value TEXT NOT NULL, reason TEXT NOT NULL, actor TEXT NOT NULL, created_at TEXT NOT NULL)''',
 'CREATE TABLE IF NOT EXISTS document_counters(kind TEXT PRIMARY KEY, next_value INTEGER NOT NULL)',
 '''CREATE TABLE IF NOT EXISTS payment_stages(id TEXT PRIMARY KEY, invoice_id TEXT NOT NULL REFERENCES invoices(id),
 position INTEGER NOT NULL, label TEXT NOT NULL, amount_pence INTEGER NOT NULL CHECK(amount_pence>0),
 due_date TEXT NOT NULL, requested_at TEXT, UNIQUE(invoice_id,position))''',
 '''CREATE TABLE IF NOT EXISTS payment_allocations(payment_id TEXT NOT NULL REFERENCES payments(id),
 stage_id TEXT NOT NULL REFERENCES payment_stages(id), amount_pence INTEGER NOT NULL CHECK(amount_pence>0),
 PRIMARY KEY(payment_id,stage_id))''',
 '''CREATE TABLE IF NOT EXISTS payment_reversals(payment_id TEXT PRIMARY KEY REFERENCES payments(id),
 reason TEXT NOT NULL, actor TEXT NOT NULL, created_at TEXT NOT NULL)''',
 '''CREATE TABLE IF NOT EXISTS payment_details(payment_id TEXT PRIMARY KEY REFERENCES payments(id),
 paid_on TEXT NOT NULL, method TEXT NOT NULL)''',
]


def setup(db):
    # Called immediately after the base DDL, before any tenant DML. SQLite's
    # documented create/copy/drop/rename migration preserves child FK names.
    if db.in_transaction:
        raise ValueError('Document migration must start outside a transaction.')
    db.execute('PRAGMA foreign_keys=OFF')
    try:
        db.execute('BEGIN IMMEDIATE')
        migrated = 'status' not in {r['name'] for r in db.execute('PRAGMA table_info(invoices)')}
        if migrated:
            db.execute(INVOICE_SCHEMA)
            db.execute("INSERT INTO invoices_v5(id,reference,quote_id,job_id,customer_id,snapshot,total_pence,created_at,status,issued_at) SELECT id,reference,quote_id,job_id,customer_id,snapshot,total_pence,created_at,'issued',created_at FROM invoices")
            db.execute('DROP TABLE invoices')
            db.execute('ALTER TABLE invoices_v5 RENAME TO invoices')
        for statement in STATEMENTS:
            db.execute(statement)
        for q in many(db, 'SELECT * FROM quotes WHERE id NOT IN (SELECT quote_id FROM quote_links)'):
            published = q['created_at'] if q['status'] in ('shared','accepted') else None
            db.execute('INSERT INTO quote_links VALUES(?,?,NULL,1,1,?,?)', (q['id'],q['id'],published,q['created_at'] if q['status']=='accepted' else None))
            event(db,'quote',q['id'],'legacy.imported',None,q,'Imported unchanged from the local v0.4 schema.','migration')
        if migrated:
            for invoice in many(db,'SELECT * FROM invoices'):
                event(db,'invoice',invoice['id'],'legacy.issued',None,invoice,'Existing local invoice retained as issued; its content and receipts are unchanged.','migration')
        for kind, table, prefix in (('quote','quotes','Q-'),('invoice','invoices','I-')):
            highest=max([int(r[0][len(prefix):]) for r in db.execute('SELECT reference FROM '+table) if re.fullmatch(prefix+r'\d+',r[0])]+[0])
            db.execute('INSERT OR IGNORE INTO document_counters VALUES(?,?)',(kind,highest+1))
            db.execute('UPDATE document_counters SET next_value=MAX(next_value,?) WHERE kind=?',(highest+1,kind))
        if db.execute('PRAGMA foreign_key_check').fetchone():
            raise ValueError('Document migration found broken record relationships.')
        db.commit()
    except BaseException:
        db.rollback()
        raise
    finally:
        db.execute('PRAGMA foreign_keys=ON')


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


def required(row):
    if not row: fail('Record not found in this business.',404)
    return row


def today():
    return datetime.now(ZoneInfo('Europe/London')).date().isoformat()


def event(db,kind,doc_id,action,before,after,reason,actor):
    db.execute('INSERT INTO document_events(kind,document_id,action,before_value,after_value,reason,actor,created_at) VALUES(?,?,?,?,?,?,?,?)',
               (kind,doc_id,action,json.dumps(before) if before is not None else None,json.dumps(after),reason,actor,now()))
    audit(db,kind+'.'+action,doc_id)


def number(db,kind):
    from .sales_outputs import allocate_reference
    return allocate_reference(db,kind)


def quote_row(db,qid):
    return required(one(db,'SELECT q.*,l.root_id,l.previous_id,l.revision,l.current,l.shared_at,l.accepted_at FROM quotes q JOIN quote_links l ON l.quote_id=q.id WHERE q.id=?',(qid,)))


def check_version(row,version):
    if row['version']!=version: fail('This record changed. Reload it and review before continuing.')


def snapshot(db,job_id,payload,original=None):
    job=required(one(db,'SELECT * FROM jobs WHERE id=?',(job_id,)))
    if original is None:
        customer=required(one(db,'SELECT id,name,email,phone,address FROM customers WHERE id=?',(job['customer_id'],)))
        from .websites import identity
        from .sales_outputs import apply_settings
        original={'business':apply_settings(db,identity(db)),'customer':customer,'service':job['service']}
    content=payload.model_dump(exclude={'version','reason'})
    subtotal=sum(item.total_pence() for item in payload.items)
    vat=(subtotal*payload.vat_percent+50)//100
    if not 0<subtotal+vat<=100_000_000: fail('Document total must be between £0.01 and £1,000,000.',422)
    return {**original,**content,'document_date':content.get('document_date') or today(),'line_totals_pence':[item.total_pence() for item in payload.items],'subtotal_pence':subtotal,'vat_pence':vat,'total_pence':subtotal+vat}


def net_paid(db,iid):
    return db.execute('SELECT COALESCE(SUM(amount_pence),0) FROM payments WHERE invoice_id=? AND id NOT IN (SELECT payment_id FROM payment_reversals)',(iid,)).fetchone()[0]


def validate_records(db,legacy=False):
    """Recovery checks business relationships in addition to SQLite FKs."""
    checks=[
        "SELECT id FROM quotes WHERE typeof(total_pence)!='integer' OR total_pence<=0",
        "SELECT id FROM invoices WHERE typeof(total_pence)!='integer' OR total_pence<=0",
        "SELECT id FROM payments WHERE typeof(amount_pence)!='integer' OR amount_pence<=0",
        'SELECT q.id FROM quotes q JOIN jobs j ON j.id=q.job_id WHERE q.customer_id!=j.customer_id',
        'SELECT i.id FROM invoices i JOIN jobs j ON j.id=i.job_id WHERE i.customer_id!=j.customer_id',
        'SELECT i.id FROM invoices i JOIN quotes q ON q.id=i.quote_id WHERE i.customer_id!=q.customer_id OR i.job_id!=q.job_id',
    ]
    if not legacy:
        checks += [
            "SELECT id FROM quotes WHERE status NOT IN ('draft','shared','accepted','declined','superseded','cancelled')",
            "SELECT q.id FROM quotes q JOIN quote_links l ON l.quote_id=q.id WHERE (q.status='superseded' AND l.current!=0) OR (q.status!='superseded' AND l.current!=1)",
            "SELECT id FROM payment_stages WHERE typeof(amount_pence)!='integer' OR amount_pence<=0",
            "SELECT payment_id FROM payment_allocations WHERE typeof(amount_pence)!='integer' OR amount_pence<=0",
            'SELECT q.id FROM quotes q LEFT JOIN quote_links l ON l.quote_id=q.id WHERE l.quote_id IS NULL',
            'SELECT q.id FROM quote_links l JOIN quotes q ON q.id=l.quote_id JOIN quotes r ON r.id=l.root_id WHERE q.customer_id!=r.customer_id OR q.job_id!=r.job_id',
            'SELECT a.payment_id FROM payment_allocations a JOIN payments p ON p.id=a.payment_id JOIN payment_stages s ON s.id=a.stage_id WHERE p.invoice_id!=s.invoice_id',
            'SELECT i.id FROM invoices i WHERE EXISTS(SELECT 1 FROM payment_stages s WHERE s.invoice_id=i.id) AND (SELECT SUM(amount_pence) FROM payment_stages s WHERE s.invoice_id=i.id)!=i.total_pence',
            'SELECT p.id FROM payments p WHERE EXISTS(SELECT 1 FROM payment_stages s WHERE s.invoice_id=p.invoice_id) AND COALESCE((SELECT SUM(amount_pence) FROM payment_allocations a WHERE a.payment_id=p.id),0)!=p.amount_pence',
            'SELECT s.id FROM payment_stages s WHERE (SELECT COALESCE(SUM(a.amount_pence),0) FROM payment_allocations a WHERE a.stage_id=s.id AND a.payment_id NOT IN(SELECT payment_id FROM payment_reversals))>s.amount_pence',
            "SELECT i.id FROM invoices i WHERE (SELECT COALESCE(SUM(p.amount_pence),0) FROM payments p WHERE p.invoice_id=i.id AND p.id NOT IN(SELECT payment_id FROM payment_reversals))>CASE WHEN i.status='issued' THEN i.total_pence ELSE 0 END",
        ]
    for sql in checks:
        if db.execute(sql+' LIMIT 1').fetchone():
            raise ValueError('Document ownership, payment allocation or balance relationships are inconsistent.')
    for table in ('quotes','invoices'):
        for raw,total,cid,jid in db.execute('SELECT snapshot,total_pence,customer_id,job_id FROM '+table):
            content=json.loads(raw)
            payload=Quote.model_validate({key:content[key] for key in ('items','vat_percent','notes','document_date','valid_until') if key in content})
            subtotal=sum(line.total_pence() for line in payload.items)
            vat=(subtotal*payload.vat_percent+50)//100
            from .customer_review import historical_owner
            if not historical_owner(db,jid,content['customer']['id']) or not 0<total<=100_000_000 or total!=subtotal+vat or content['subtotal_pence']!=subtotal or content['vat_pence']!=vat or content['total_pence']!=total:
                raise ValueError('Document snapshot totals or customer identity are inconsistent.')
            if 'line_totals_pence' in content and content['line_totals_pence']!=[line.total_pence() for line in payload.items]:
                raise ValueError('Document line totals are inconsistent.')


def invoice_rows(db,where='',args=()):
    rows=many(db,'SELECT i.* FROM invoices i '+where+' ORDER BY i.created_at DESC,i.rowid DESC',args)
    for row in rows:
        row['paid_pence']=net_paid(db,row['id'])
        row['balance_pence']=row['total_pence']-row['paid_pence'] if row['status']=='issued' else 0
        due=json.loads(row['snapshot']).get('due_date','')
        row['payment_status']=row['status'] if row['status']!='issued' else ('paid' if row['balance_pence']==0 else 'overdue' if due and due<today() else 'part_paid' if row['paid_pence'] else 'unpaid')
    return rows


def stage_rows(db,iid):
    rows=many(db,'SELECT s.*,COALESCE((SELECT SUM(a.amount_pence) FROM payment_allocations a WHERE a.stage_id=s.id AND a.payment_id NOT IN (SELECT payment_id FROM payment_reversals)),0) AS paid_pence FROM payment_stages s WHERE invoice_id=? ORDER BY position',(iid,))
    invoice_state=db.execute('SELECT status FROM invoices WHERE id=?',(iid,)).fetchone()
    for row in rows:
        row['balance_pence']=row['amount_pence']-row['paid_pence']
        row['status']='paid' if row['balance_pence']==0 else 'part_paid' if row['paid_pence'] else 'requested' if row['requested_at'] else 'planned'
        if invoice_state and invoice_state[0] in ('void','cancelled'):
            row['status']=invoice_state[0]
            row['balance_pence']=0
    return rows


def get_document(db,kind,doc_id,customer_id=None):
    if kind not in ('quote','invoice'): fail('Document not found.',404)
    row=quote_row(db,doc_id) if kind=='quote' else required(next(iter(invoice_rows(db,'WHERE i.id=?',(doc_id,))),None))
    if customer_id and (row['customer_id']!=customer_id or not row['shared_at' if kind=='quote' else 'issued_at']):
        fail('Document not found.',404)
    row['snapshot']=json.loads(row['snapshot'])
    from .sales_outputs import snapshot_hash
    row['snapshot_hash']=snapshot_hash(row['snapshot'])
    if kind=='quote':
        response=one(db,'SELECT action,evidence,created_at FROM quote_responses WHERE quote_id=?',(doc_id,))
        if response:
            response['evidence']=json.loads(response['evidence'])
            if customer_id and response['evidence']['mode']=='owner-recorded':response['evidence']['actor']='business owner (manual record)'
        row['response']=response
    row['expired']=bool(kind=='quote' and row['snapshot'].get('valid_until') and row['snapshot']['valid_until']<today())
    if kind=='invoice':
        row['payments']=many(db,"SELECT p.*,d.paid_on,d.method,CASE WHEN r.payment_id IS NULL THEN 0 ELSE 1 END AS reversed FROM payments p LEFT JOIN payment_details d ON d.payment_id=p.id LEFT JOIN payment_reversals r ON r.payment_id=p.id WHERE p.invoice_id=? ORDER BY p.created_at,p.rowid",(doc_id,))
        row['schedule']=stage_rows(db,doc_id)
        row['replacement']=one(db,'SELECT id,reference,status,issued_at FROM invoices WHERE replaces_id=?',(doc_id,))
        if customer_id and row['replacement'] and not row['replacement']['issued_at']: row['replacement']=None
        for payment in row['payments']:
            payment['allocations']=many(db,'SELECT stage_id,amount_pence FROM payment_allocations WHERE payment_id=?',(payment['id'],))
            if customer_id: payment.pop('note',None)
    if not customer_id:
        from .billing_sources import document_projection
        row['source_review']=document_projection(db,kind,doc_id)
        row['history']=many(db,'SELECT * FROM document_events WHERE kind=? AND document_id=? ORDER BY id',(kind,doc_id))
        for entry in row['history']:
            entry['before_value']=json.loads(entry['before_value']) if entry['before_value'] else None
            entry['after_value']=json.loads(entry['after_value'])
    return row


class DocumentEdit(Quote):
    version: int = Field(strict=True,ge=1)


class InvoiceInput(Quote):
    due_date: str = ''

    @field_validator('due_date')
    @classmethod
    def valid_date(cls,v):
        return Quote.valid_date(v)


class InvoiceEdit(InvoiceInput):
    version: int = Field(strict=True,ge=1)


class Action(Version):
    reason: str = Field(default='',max_length=1000)


class Stage(Strict):
    label: str = Field(min_length=1,max_length=100)
    amount_pence: int = Field(strict=True,gt=0,le=100_000_000)
    due_date: str = ''

    @field_validator('due_date')
    @classmethod
    def valid_date(cls,v): return Quote.valid_date(v)


class Schedule(Version):
    stages: list[Stage] = Field(min_length=1,max_length=12)


class Correction(Action):
    recording_error: StrictBool
    replacement: Payment | None = None


def reason_required(payload):
    if len(payload.reason)<5: fail('Add a clear reason (at least five characters).',422)


def record_payment(db,iid,payload,actor):
    invoice=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)))
    if invoice['status']!='issued': fail('Only an issued invoice can receive a payment record.')
    if payload.paid_on and payload.paid_on>today(): fail('A received payment cannot be dated after today in Europe/London.',422)
    paid=net_paid(db,iid)
    if payload.amount_pence>invoice['total_pence']-paid: fail('This exceeds the current balance. Reload the invoice.')
    stages=stage_rows(db,iid)
    if payload.stage_id:
        stages=[required(next((s for s in stages if s['id']==payload.stage_id),None))]
        if payload.amount_pence>stages[0]['balance_pence']: fail('This exceeds the selected stage balance.')
    pid=uid()
    db.execute('INSERT INTO payments VALUES(?,?,?,?,?)',(pid,iid,payload.amount_pence,payload.note,now()))
    db.execute('INSERT INTO payment_details VALUES(?,?,?)',(pid,payload.paid_on or today(),payload.method))
    left=payload.amount_pence
    for stage in stages:
        allocated=min(left,stage['balance_pence'])
        if allocated: db.execute('INSERT INTO payment_allocations VALUES(?,?,?)',(pid,stage['id'],allocated))
        left-=allocated
    if stages and left: fail('The schedule cannot allocate this receipt. Reload and review.')
    db.execute('UPDATE invoices SET version=version+1 WHERE id=?',(iid,))
    event(db,'invoice',iid,'payment.recorded',None,{'payment_id':pid,**payload.model_dump()},'',actor)
    return {'id':pid,'paid_pence':paid+payload.amount_pence}


def accept_quote(db,qid,version,actor,customer_id=None,reason='',response=None):
    q=quote_row(db,qid)
    if customer_id and (q['customer_id']!=customer_id or not q['shared_at']): fail('Document not found.',404)
    check_version(q,version)
    if not q['current'] or q['status'] not in (('shared',) if customer_id else ('draft','shared')): fail('Only the current available quote can be accepted.')
    if q['status']=='draft':
        from .billing_sources import unchanged
        unchanged(db,'quote',qid)
    content=json.loads(q['snapshot'])
    if content.get('valid_until') and content['valid_until']<today(): fail('This quote has expired. Ask for a new revision.')
    from .sales_outputs import record_response
    record_response(db,q,'accepted',actor,response)
    db.execute("UPDATE quotes SET status='accepted',version=version+1 WHERE id=?",(qid,))
    db.execute('UPDATE quote_links SET shared_at=COALESCE(shared_at,?),accepted_at=? WHERE quote_id=?',(now(),now(),qid))
    event(db,'quote',qid,'accepted',q,quote_row(db,qid),'Customer portal acceptance.' if customer_id else ('Customer acceptance recorded manually in the local demo. '+reason).strip(),actor)
    return {'ok':True}


def mount_documents(app,store,session,once):
    from . import billing_sources as sources
    def mutate(request,payload,fn):
        s=session(request,csrf=True)
        with store.tenant(s['business_id']) as db:
            return once(db,request,payload.model_dump(),lambda:fn(db,s['subject']))

    @app.post('/api/jobs/{job_id}/quotes')
    def create_quote(job_id:str,payload:Quote,request:Request):
        def perform(db,actor):
            job=required(one(db,'SELECT * FROM jobs WHERE id=?',(job_id,)))
            content=snapshot(db,job_id,payload);qid=uid();reference=number(db,'quote')
            db.execute('INSERT INTO quotes(id,reference,job_id,customer_id,status,snapshot,total_pence,created_at) VALUES(?,?,?,?,?,?,?,?)',(qid,reference,job_id,job['customer_id'],'draft',json.dumps(content),content['total_pence'],now()))
            db.execute('INSERT INTO quote_links VALUES(?,?,NULL,1,1,NULL,NULL)',(qid,qid))
            event(db,'quote',qid,'created',None,quote_row(db,qid),'',actor)
            return {'id':qid,'reference':reference}
        return mutate(request,payload,perform)

    @app.put('/api/quotes/{qid}')
    def edit_quote(qid:str,payload:DocumentEdit,request:Request):
        def perform(db,actor):
            q=quote_row(db,qid);check_version(q,payload.version)
            if q['status']!='draft' or not q['current']: fail('Shared or accepted content is sealed. Create a new revision.')
            sources.guard_edit(db,'quote',qid)
            content=snapshot(db,q['job_id'],payload,json.loads(q['snapshot']))
            db.execute('UPDATE quotes SET snapshot=?,total_pence=?,version=version+1 WHERE id=?',(json.dumps(content),content['total_pence'],qid))
            event(db,'quote',qid,'draft.edited',q,quote_row(db,qid),'',actor)
            return {'ok':True}
        return mutate(request,payload,perform)

    @app.post('/api/quotes/{qid}/{action}')
    def quote_action(qid:str,action:str,payload:Action,request:Request):
        def perform(db,actor):
            q=quote_row(db,qid)
            if action=='accept': return accept_quote(db,qid,payload.version,actor,reason=payload.reason)
            if action=='invoice':
                existing=one(db,"SELECT id,reference FROM invoices WHERE quote_id=? AND status IN ('draft','issued')",(qid,))
                if existing:return existing
            check_version(q,payload.version)
            if not q['current']: fail('This quote has been superseded.')
            if action=='share':
                if q['status']!='draft': fail('Only a draft can be shared.')
                sources.unchanged(db,'quote',qid)
                if (json.loads(q['snapshot']).get('valid_until') or '9999')<today(): fail('Set a current validity date before sharing.')
                db.execute("UPDATE quotes SET status='shared',version=version+1 WHERE id=?",(qid,))
                db.execute('UPDATE quote_links SET shared_at=? WHERE quote_id=?',(now(),qid))
            elif action=='invoice':
                if q['status']!='accepted': fail('Record acceptance before drafting an invoice.')
                return new_invoice(db,q['job_id'],q['customer_id'],{**json.loads(q['snapshot']),'document_date':today(),'due_date':''},actor,quote_id=qid)
            elif action in ('revise','cancel'):
                reason_required(payload)
                if q['status'] not in ('draft','shared','accepted','declined'): fail('This quote is already closed.')
                if one(db,"SELECT i.id FROM invoices i JOIN quote_links l ON l.quote_id=i.quote_id WHERE l.root_id=? AND i.status IN ('draft','issued')",(q['root_id'],)): fail('Resolve the linked invoice before changing the quote family.')
                if action=='revise':
                    if q['status']=='draft': fail('Edit this draft before sharing it.')
                    db.execute("UPDATE quotes SET status='superseded',version=version+1 WHERE id=?",(qid,))
                    db.execute('UPDATE quote_links SET current=0 WHERE quote_id=?',(qid,))
                    new_id=uid();ref=number(db,'quote')
                    db.execute('INSERT INTO quotes(id,reference,job_id,customer_id,status,snapshot,total_pence,created_at) VALUES(?,?,?,?,?,?,?,?)',(new_id,ref,q['job_id'],q['customer_id'],'draft',q['snapshot'],q['total_pence'],now()))
                    db.execute('INSERT INTO quote_links VALUES(?,?,?,?,1,NULL,NULL)',(new_id,q['root_id'],qid,q['revision']+1))
                    event(db,'quote',qid,'superseded',q,quote_row(db,qid),payload.reason,actor)
                    event(db,'quote',new_id,'revision.created',None,quote_row(db,new_id),payload.reason,actor)
                    sources.transfer(db,'quote',qid,'quote',new_id,actor)
                    return {'id':new_id,'reference':ref}
                db.execute("UPDATE quotes SET status='cancelled',version=version+1 WHERE id=?",(qid,))
                sources.release(db,'quote',qid,actor)
            else: fail('Action not found.',404)
            event(db,'quote',qid,action,q,quote_row(db,qid),payload.reason,actor)
            return {'ok':True}
        return mutate(request,payload,perform)

    @app.post('/api/jobs/{job_id}/invoices')
    def create_invoice(job_id:str,payload:InvoiceInput,request:Request):
        def perform(db,actor):
            job=required(one(db,'SELECT * FROM jobs WHERE id=?',(job_id,)))
            return new_invoice(db,job_id,job['customer_id'],snapshot(db,job_id,payload),actor)
        return mutate(request,payload,perform)

    @app.put('/api/invoices/{iid}')
    def edit_invoice(iid:str,payload:InvoiceEdit,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            if inv['status']!='draft': fail('Issued content is sealed. Use the explicit correction workflow.')
            sources.guard_edit(db,'invoice',iid)
            content=snapshot(db,inv['job_id'],payload,json.loads(inv['snapshot']))
            stages=stage_rows(db,iid)
            if stages and sum(s['amount_pence'] for s in stages)!=content['total_pence']: fail('Clear or replace the draft schedule before changing its total.')
            db.execute('UPDATE invoices SET snapshot=?,total_pence=?,version=version+1 WHERE id=?',(json.dumps(content),content['total_pence'],iid))
            event(db,'invoice',iid,'draft.edited',inv,one(db,'SELECT * FROM invoices WHERE id=?',(iid,)),'',actor)
            return {'ok':True}
        return mutate(request,payload,perform)

    @app.post('/api/invoices/{iid}/schedule')
    def save_schedule(iid:str,payload:Schedule,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            if inv['status'] not in ('draft','issued'): fail('This invoice is closed.')
            previous=stage_rows(db,iid)
            if any(s['requested_at'] for s in previous) or one(db,'SELECT id FROM payments WHERE invoice_id=?',(iid,)): fail('The schedule is locked after a request or receipt. Existing history cannot be replaced.')
            if sum(s.amount_pence for s in payload.stages)!=inv['total_pence']: fail('Stage amounts must add up exactly to the invoice total.',422)
            db.execute('DELETE FROM payment_stages WHERE invoice_id=?',(iid,))
            for position,stage in enumerate(payload.stages,1):
                db.execute('INSERT INTO payment_stages VALUES(?,?,?,?,?,?,NULL)',(uid(),iid,position,stage.label,stage.amount_pence,stage.due_date))
            db.execute('UPDATE invoices SET version=version+1 WHERE id=?',(iid,))
            event(db,'invoice',iid,'schedule.saved',previous,stage_rows(db,iid),'',actor)
            return {'ok':True}
        return mutate(request,payload,perform)

    @app.post('/api/invoices/{iid}/schedule/{stage_id}/request')
    def request_stage(iid:str,stage_id:str,payload:Version,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            stage=required(next((s for s in stage_rows(db,iid) if s['id']==stage_id),None))
            if inv['status']!='issued' or stage['balance_pence']==0: fail('Only an unpaid stage on an issued invoice can be requested.')
            db.execute('UPDATE payment_stages SET requested_at=COALESCE(requested_at,?) WHERE id=?',(now(),stage_id))
            db.execute('UPDATE invoices SET version=version+1 WHERE id=?',(iid,))
            from .sales_outputs import freeze_request
            freeze_request(db,iid,stage_id,legacy_review=bool(stage['requested_at']))
            event(db,'invoice',iid,'stage.requested',stage,{'stage_id':stage_id,'remaining_pence':stage['balance_pence']},'Shared locally; no email or payment taken.',actor)
            return {'ok':True,'document_url':'/documents/invoice/'+iid+'?stage='+stage_id}
        return mutate(request,payload,perform)

    @app.post('/api/invoices/{iid}/schedule/clear')
    def clear_schedule(iid:str,payload:Version,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            previous=stage_rows(db,iid)
            if inv['status'] not in ('draft','issued') or any(s['requested_at'] for s in previous) or one(db,'SELECT id FROM payments WHERE invoice_id=?',(iid,)):
                fail('The schedule is locked after a request or receipt, or this invoice is closed.')
            db.execute('DELETE FROM payment_stages WHERE invoice_id=?',(iid,))
            db.execute('UPDATE invoices SET version=version+1 WHERE id=?',(iid,))
            event(db,'invoice',iid,'schedule.cleared',previous,[],'',actor)
            return {'ok':True}
        return mutate(request,payload,perform)

    @app.post('/api/invoices/{iid}/payments')
    def payment(iid:str,payload:Payment,request:Request):
        return mutate(request,payload,lambda db,actor:record_payment(db,iid,payload,actor))

    @app.post('/api/invoices/{iid}/payments/{pid}/correct')
    def correct_payment(iid:str,pid:str,payload:Correction,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            payment=required(one(db,'SELECT * FROM payments WHERE id=? AND invoice_id=?',(pid,iid)))
            reason_required(payload)
            if not payload.recording_error: fail('This corrects a mistaken record only; it does not process a refund.',422)
            if inv['status']!='issued' or one(db,'SELECT payment_id FROM payment_reversals WHERE payment_id=?',(pid,)): fail('This payment is already reversed or the invoice is closed.')
            db.execute('INSERT INTO payment_reversals VALUES(?,?,?,?)',(pid,payload.reason,actor,now()))
            db.execute('UPDATE invoices SET version=version+1 WHERE id=?',(iid,))
            replacement=record_payment(db,iid,payload.replacement,actor) if payload.replacement else None
            event(db,'invoice',iid,'payment.corrected',payment,{'reversed_payment_id':pid,'replacement':replacement},payload.reason,actor)
            return {'ok':True,'replacement':replacement,'paid_pence':net_paid(db,iid)}
        return mutate(request,payload,perform)

    @app.post('/api/invoices/{iid}/{action}')
    def invoice_action(iid:str,action:str,payload:Action,request:Request):
        def perform(db,actor):
            inv=required(one(db,'SELECT * FROM invoices WHERE id=?',(iid,)));check_version(inv,payload.version)
            if action=='issue':
                if inv['status']!='draft': fail('Only a draft invoice can be issued.')
                sources.issued(db,iid,actor)
                db.execute("UPDATE invoices SET status='issued',issued_at=?,version=version+1 WHERE id=?",(now(),iid))
            elif action in ('cancel','void','correct'):
                reason_required(payload)
                expected='draft' if action=='cancel' else 'issued'
                if inv['status']!=expected: fail('This action does not apply to the current invoice state.')
                if net_paid(db,iid): fail('Money is recorded against this invoice. Refund/credit settlement needs manual review; do not erase real receipts.')
                db.execute('UPDATE invoices SET status=?,version=version+1 WHERE id=?',('cancelled' if action=='cancel' else 'void',iid))
                if action=='correct':
                    result=new_invoice(db,inv['job_id'],inv['customer_id'],json.loads(inv['snapshot']),actor,inv['quote_id'],iid)
                    event(db,'invoice',iid,'void.for_correction',inv,{**one(db,'SELECT * FROM invoices WHERE id=?',(iid,)),'replacement_id':result['id']},payload.reason,actor)
                    return result
                sources.release(db,'invoice',iid,actor)
            else: fail('Action not found.',404)
            event(db,'invoice',iid,action,inv,one(db,'SELECT * FROM invoices WHERE id=?',(iid,)),payload.reason,actor)
            return {'ok':True}
        return mutate(request,payload,perform)


def new_invoice(db,job_id,customer_id,content,actor,quote_id=None,replaces_id=None):
    iid=uid();reference=number(db,'invoice')
    db.execute('INSERT INTO invoices(id,reference,quote_id,job_id,customer_id,snapshot,total_pence,created_at,replaces_id) VALUES(?,?,?,?,?,?,?,?,?)',
               (iid,reference,quote_id,job_id,customer_id,json.dumps(content),content['total_pence'],now(),replaces_id))
    event(db,'invoice',iid,'draft.created',None,one(db,'SELECT * FROM invoices WHERE id=?',(iid,)),'Replacement draft.' if replaces_id else '',actor)
    from . import billing_sources as sources
    if replaces_id:sources.transfer(db,'invoice',replaces_id,'invoice',iid,actor)
    elif quote_id:sources.transfer(db,'quote',quote_id,'invoice',iid,actor,check=True)
    return {'id':iid,'reference':reference}