"""Append-only usage ledger with fixed-point quantities and immutable UTC period closes."""
from __future__ import annotations
import contextlib, datetime as dt, hashlib, json, re, sqlite3, time
from pathlib import Path

UNITS={'api_call':0,'compute_minute':3,'storage_gib_hour':3}
RULE_VERSION='usage-v1'
class Problem(ValueError):
    def __init__(self,message,status=400): super().__init__(message);self.status=status

def identifier(value):
    if not isinstance(value,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,95}',value):raise Problem('Invalid identifier')
    return value

def period(value):
    if not isinstance(value,str) or not re.fullmatch(r'\d{4}-(0[1-9]|1[0-2])',value) or not 2000<=int(value[:4])<=2100:raise Problem('Period must be YYYY-MM between 2000 and 2100')
    return value

def timestamp(value):
    if not isinstance(value,str):raise Problem('occurred_at must be an ISO timestamp with a timezone')
    try:
        parsed=dt.datetime.fromisoformat(value.replace('Z','+00:00'))
        if parsed.tzinfo is None:raise ValueError()
        parsed=parsed.astimezone(dt.timezone.utc)
        if parsed.microsecond:raise ValueError()
        normalized=parsed.isoformat(timespec='seconds').replace('+00:00','Z');period(normalized[:7]);return normalized
    except ValueError as e:raise Problem('Use a timestamp with whole seconds and an explicit timezone') from e

def quantity(value,unit):
    if unit not in UNITS:raise Problem('Unknown unit')
    precision=UNITS[unit]
    if not isinstance(value,str) or not re.fullmatch(r'\d+(?:\.\d+)?',value):raise Problem('quantity must be a nonnegative decimal string')
    parts=value.split('.')
    if len(parts)==2 and (len(parts[1])>precision or not precision):raise Problem(f'{unit} permits {precision} decimal places')
    amount=int(parts[0])*10**precision+(int(parts[1].ljust(precision,'0')) if len(parts)==2 else 0)
    if amount>10**15:raise Problem('Quantity exceeds per-event limit')
    return amount

def formatted(value,unit):
    precision=UNITS[unit];sign='-' if value<0 else '';value=abs(value)
    if not precision:return sign+str(value)
    return f'{sign}{value//10**precision}.{value%10**precision:0{precision}d}'

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

SCHEMA='''
CREATE TABLE IF NOT EXISTS events(
 seq INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT NOT NULL, source TEXT NOT NULL,event_id TEXT NOT NULL,
 kind TEXT NOT NULL CHECK(kind IN ('usage','correction')), unit TEXT NOT NULL, amount TEXT NOT NULL,
 occurred_at TEXT NOT NULL, origin_period TEXT NOT NULL, arrived_at TEXT NOT NULL,
 original_seq INTEGER REFERENCES events(seq), reason TEXT, fingerprint TEXT NOT NULL,
 UNIQUE(account,source,event_id));
CREATE UNIQUE INDEX IF NOT EXISTS one_correction_per_original ON events(original_seq) WHERE original_seq IS NOT NULL;
CREATE TABLE IF NOT EXISTS entries(
 id INTEGER PRIMARY KEY, event_seq INTEGER NOT NULL REFERENCES events(seq),
 kind TEXT NOT NULL CHECK(kind IN ('usage','reversal','replacement')), amount TEXT NOT NULL,
 UNIQUE(event_seq,kind));
CREATE TABLE IF NOT EXISTS periods(account TEXT NOT NULL,period TEXT NOT NULL,status TEXT NOT NULL CHECK(status IN ('open','closed')),PRIMARY KEY(account,period));
CREATE TABLE IF NOT EXISTS posting_batches(
 seq INTEGER PRIMARY KEY AUTOINCREMENT,account TEXT NOT NULL,period TEXT NOT NULL,event_seq INTEGER NOT NULL UNIQUE REFERENCES events(seq),
 actor TEXT NOT NULL,at TEXT NOT NULL,reason TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS postings(entry_id INTEGER PRIMARY KEY REFERENCES entries(id),batch_seq INTEGER NOT NULL REFERENCES posting_batches(seq));
CREATE TABLE IF NOT EXISTS totals(account TEXT NOT NULL,period TEXT NOT NULL,unit TEXT NOT NULL,amount TEXT NOT NULL,PRIMARY KEY(account,period,unit));
CREATE TABLE IF NOT EXISTS closes(
 id INTEGER PRIMARY KEY,account TEXT NOT NULL,period TEXT NOT NULL,version INTEGER NOT NULL,
 ingest_cutoff INTEGER NOT NULL,posting_cutoff INTEGER NOT NULL,rule_version TEXT NOT NULL,
 at TEXT NOT NULL,actor TEXT NOT NULL,totals_json TEXT NOT NULL,export_sha256 TEXT NOT NULL,
 UNIQUE(account,period,version));
CREATE TABLE IF NOT EXISTS audit(id INTEGER PRIMARY KEY,account TEXT NOT NULL,at TEXT NOT NULL,actor TEXT NOT NULL,action TEXT NOT NULL,subject TEXT NOT NULL,detail TEXT NOT NULL);
'''

class Ledger:
    def __init__(self,path,clock=None,transaction_observer=None):
        self.path=str(path);Path(self.path).parent.mkdir(parents=True,exist_ok=True)
        self.clock=clock or (lambda:dt.datetime.now(dt.timezone.utc).isoformat(timespec='seconds').replace('+00:00','Z'))
        self.transaction_observer=transaction_observer
        with contextlib.closing(self.connect()) as db:
            db.executescript(SCHEMA)
            for table in ('events','entries','posting_batches','postings','closes','audit'):
                for action in ('UPDATE','DELETE'):
                    db.execute(f"CREATE TRIGGER IF NOT EXISTS immutable_{table}_{action.lower()} BEFORE {action} ON {table} BEGIN SELECT RAISE(ABORT,'append-only table'); END")
            db.execute("CREATE TRIGGER IF NOT EXISTS frozen_period BEFORE UPDATE ON periods WHEN OLD.status='closed' BEGIN SELECT RAISE(ABORT,'closed period is immutable'); END")
            db.execute("CREATE TRIGGER IF NOT EXISTS no_period_delete BEFORE DELETE ON periods BEGIN SELECT RAISE(ABORT,'period history is immutable'); END")
    def connect(self):
        db=sqlite3.connect(self.path,timeout=15,isolation_level=None);db.row_factory=sqlite3.Row
        db.execute('PRAGMA foreign_keys=ON');db.execute('PRAGMA journal_mode=WAL');db.execute('PRAGMA synchronous=FULL');return db
    @contextlib.contextmanager
    def transaction(self,operation):
        db=self.connect()
        try:
            db.execute('BEGIN IMMEDIATE')
            if self.transaction_observer:self.transaction_observer(operation)
            yield db;db.commit()
        except BaseException:db.rollback();raise
        finally:db.close()
    def audit(self,db,account,actor,action,subject,detail=''):
        db.execute('INSERT INTO audit(account,at,actor,action,subject,detail) VALUES(?,?,?,?,?,?)',(account,self.clock(),actor,action,str(subject),detail))
    def ensure_period(self,db,account,name):
        period(name);db.execute("INSERT OR IGNORE INTO periods VALUES(?,?,'open')",(account,name))
        return db.execute('SELECT status FROM periods WHERE account=? AND period=?',(account,name)).fetchone()[0]
    def _post(self,db,account,event_seq,target,actor,reason):
        if self.ensure_period(db,account,target)!='open':raise Problem('Target period is closed',409)
        event=db.execute('SELECT * FROM events WHERE account=? AND seq=?',(account,event_seq)).fetchone()
        if not event:raise Problem('Event not found',404)
        if db.execute('SELECT 1 FROM posting_batches WHERE event_seq=?',(event_seq,)).fetchone():raise Problem('Event already posted',409)
        batch=db.execute('INSERT INTO posting_batches(account,period,event_seq,actor,at,reason) VALUES(?,?,?,?,?,?)',(account,target,event_seq,actor,self.clock(),reason)).lastrowid
        entries=db.execute('SELECT id,amount FROM entries WHERE event_seq=?',(event_seq,)).fetchall()
        total=db.execute('SELECT amount FROM totals WHERE account=? AND period=? AND unit=?',(account,target,event['unit'])).fetchone()
        amount=int(total[0]) if total else 0
        for entry in entries:
            db.execute('INSERT INTO postings VALUES(?,?)',(entry['id'],batch));amount+=int(entry['amount'])
        db.execute('INSERT INTO totals VALUES(?,?,?,?) ON CONFLICT(account,period,unit) DO UPDATE SET amount=excluded.amount',(account,target,event['unit'],str(amount)))
        return batch
    def ingest(self,account,data,actor='ingest'):
        identifier(account)
        if not isinstance(data,dict):raise Problem('JSON object required')
        source=identifier(data.get('source'));event_id=identifier(data.get('event_id'));unit=data.get('unit');amount=quantity(data.get('quantity'),unit)
        occurred=timestamp(data.get('occurred_at'));origin=occurred[:7]
        fingerprint=hashlib.sha256(canonical({'source':source,'event_id':event_id,'unit':unit,'amount':str(amount),'occurred_at':occurred,'kind':'usage'}).encode()).hexdigest()
        with self.transaction('ingest') as db:
            old=db.execute('SELECT seq,fingerprint FROM events WHERE account=? AND source=? AND event_id=?',(account,source,event_id)).fetchone()
            if old:
                if old['fingerprint']!=fingerprint:raise Problem('Event identity conflicts with stored content',409)
                return {'seq':old['seq'],'duplicate':True}
            status=self.ensure_period(db,account,origin)
            seq=db.execute("INSERT INTO events(account,source,event_id,kind,unit,amount,occurred_at,origin_period,arrived_at,fingerprint) VALUES(?,?,?,'usage',?,?,?,?,?,?)",(account,source,event_id,unit,str(amount),occurred,origin,self.clock(),fingerprint)).lastrowid
            db.execute("INSERT INTO entries(event_seq,kind,amount) VALUES(?,'usage',?)",(seq,str(amount)))
            if status=='open':self._post(db,account,seq,origin,actor,'event_time_period')
            self.audit(db,account,actor,'ingest',seq,'pending late adjustment' if status=='closed' else origin)
            return {'seq':seq,'duplicate':False,'pending':status=='closed'}
    def correct(self,account,data,actor='operator'):
        source=identifier(data.get('source'));event_id=identifier(data.get('event_id'))
        original_source=identifier(data.get('original_source'));original_event=identifier(data.get('original_event_id'))
        reason=data.get('reason')
        if not isinstance(reason,str) or not 3<=len(reason.strip())<=500:raise Problem('Correction reason must be 3-500 characters')
        with self.transaction('correct') as db:
            original=db.execute('SELECT * FROM events WHERE account=? AND source=? AND event_id=?',(account,original_source,original_event)).fetchone()
            if not original:raise Problem('Original event not found in this account',404)
            amount=quantity(data.get('replacement_quantity'),original['unit'])
            fingerprint=hashlib.sha256(canonical({'kind':'correction','original_seq':original['seq'],'amount':str(amount),'reason':reason.strip()}).encode()).hexdigest()
            old=db.execute('SELECT seq,fingerprint FROM events WHERE account=? AND source=? AND event_id=?',(account,source,event_id)).fetchone()
            if old:
                if old['fingerprint']!=fingerprint:raise Problem('Correction identity conflicts with stored content',409)
                return {'seq':old['seq'],'duplicate':True}
            if db.execute('SELECT 1 FROM events WHERE original_seq=?',(original['seq'],)).fetchone():raise Problem('This event already has a correction; reference the latest correction to revise it again',409)
            seq=db.execute("INSERT INTO events(account,source,event_id,kind,unit,amount,occurred_at,origin_period,arrived_at,original_seq,reason,fingerprint) VALUES(?,?,?,'correction',?,?,?,?,?,?,?,?)",(account,source,event_id,original['unit'],str(amount),original['occurred_at'],original['origin_period'],self.clock(),original['seq'],reason.strip(),fingerprint)).lastrowid
            db.execute("INSERT INTO entries(event_seq,kind,amount) VALUES(?,'reversal',?)",(seq,str(-int(original['amount']))))
            db.execute("INSERT INTO entries(event_seq,kind,amount) VALUES(?,'replacement',?)",(seq,str(amount)))
            previous=db.execute('SELECT period FROM posting_batches WHERE event_seq=?',(original['seq'],)).fetchone()
            target=previous['period'] if previous else original['origin_period']
            is_open=previous is not None and self.ensure_period(db,account,target)=='open'
            if is_open:self._post(db,account,seq,target,actor,'correction_to_open_period')
            self.audit(db,account,actor,'correction',seq,reason.strip())
            return {'seq':seq,'duplicate':False,'pending':not is_open,'original_seq':original['seq']}
    def approve(self,account,event_seq,target,actor='operator'):
        period(target)
        with self.transaction('approve') as db:
            event=db.execute('SELECT * FROM events WHERE account=? AND seq=?',(account,event_seq)).fetchone()
            if not event:raise Problem('Event not found in this account',404)
            # A correction cannot be posted before the event it adjusts.
            parent=db.execute('SELECT period FROM posting_batches WHERE event_seq=?',(event['original_seq'],)).fetchone() if event['original_seq'] else None
            if event['original_seq'] and parent is None:raise Problem('Post the parent event before approving its correction',409)
            floor=max(event['origin_period'],parent['period'] if parent else event['origin_period'])
            if target<=floor:raise Problem('A pending adjustment must post to a later open period',409)
            batch=self._post(db,account,event_seq,target,actor,'approved_adjustment')
            self.audit(db,account,actor,'adjustment_approved',event_seq,target)
            return {'batch_seq':batch,'period':target}
    def _rows(self,db,account,name,cutoff=None):
        sql='''SELECT p.seq posting_seq,e.seq ingest_seq,e.source,e.event_id,e.kind event_kind,e.unit,e.occurred_at,e.origin_period,e.arrived_at,e.original_seq,e.reason,n.id entry_id,n.kind entry_kind,n.amount
            FROM posting_batches p JOIN postings k ON k.batch_seq=p.seq JOIN entries n ON n.id=k.entry_id JOIN events e ON e.seq=n.event_seq
            WHERE p.account=? AND p.period=?'''
        args=[account,name]
        if cutoff is not None:sql+=' AND p.seq<=?';args.append(cutoff)
        return [dict(x) for x in db.execute(sql+' ORDER BY p.seq,n.id',args)]
    @staticmethod
    def _totals(rows):
        result={}
        for row in rows:result[row['unit']]=result.get(row['unit'],0)+int(row['amount'])
        return {unit:{'quantity':formatted(amount,unit),'base_units':str(amount),'precision':UNITS[unit]} for unit,amount in sorted(result.items())}
    def reconcile(self,account,name):
        period(name)
        with contextlib.closing(self.connect()) as db:
            db.execute('BEGIN')
            rebuilt=self._totals(self._rows(db,account,name));cached={r['unit']:str(r['amount']) for r in db.execute('SELECT unit,amount FROM totals WHERE account=? AND period=?',(account,name))}
            expected={u:v['base_units'] for u,v in rebuilt.items()}
            return {'period':name,'matches':expected==cached,'rebuilt':rebuilt,'cached_base_units':cached}
    def close(self,account,name,actor='operator'):
        period(name)
        with self.transaction('close') as db:
            existing=db.execute('SELECT * FROM closes WHERE account=? AND period=? ORDER BY version DESC LIMIT 1',(account,name)).fetchone()
            if existing:return {'close_id':existing['id'],'version':existing['version'],'duplicate':True}
            self.ensure_period(db,account,name)
            ingest_cutoff=db.execute('SELECT COALESCE(MAX(seq),0) FROM events').fetchone()[0]
            posting_cutoff=db.execute('SELECT COALESCE(MAX(seq),0) FROM posting_batches').fetchone()[0]
            rows=self._rows(db,account,name,posting_cutoff);totals=self._totals(rows)
            cached={r['unit']:r['amount'] for r in db.execute('SELECT unit,amount FROM totals WHERE account=? AND period=?',(account,name))}
            if cached!={u:v['base_units'] for u,v in totals.items()}:raise Problem('Cached totals differ from the ledger; investigate before closing',409)
            export={'account':account,'period':name,'version':1,'timezone':'UTC','rule_version':RULE_VERSION,'ingest_cutoff':ingest_cutoff,'posting_cutoff':posting_cutoff,'totals':totals,'entries':rows}
            digest=hashlib.sha256(canonical(export).encode()).hexdigest()
            close_id=db.execute('INSERT INTO closes(account,period,version,ingest_cutoff,posting_cutoff,rule_version,at,actor,totals_json,export_sha256) VALUES(?,?,1,?,?,?,?,?,?,?)',(account,name,ingest_cutoff,posting_cutoff,RULE_VERSION,self.clock(),actor,canonical(totals),digest)).lastrowid
            db.execute("UPDATE periods SET status='closed' WHERE account=? AND period=?",(account,name));self.audit(db,account,actor,'period_closed',name,digest)
            return {'close_id':close_id,'version':1,'duplicate':False}
    def export(self,account,name,version=1):
        period(name)
        with contextlib.closing(self.connect()) as db:
            close=db.execute('SELECT * FROM closes WHERE account=? AND period=? AND version=?',(account,name,version)).fetchone()
            if not close:raise Problem('Closed version not found in this account',404)
            rows=self._rows(db,account,name,close['posting_cutoff'])
            result={'account':account,'period':name,'version':version,'timezone':'UTC','rule_version':close['rule_version'],'ingest_cutoff':close['ingest_cutoff'],'posting_cutoff':close['posting_cutoff'],'totals':self._totals(rows),'entries':rows}
            digest=hashlib.sha256(canonical(result).encode()).hexdigest()
            if digest!=close['export_sha256'] or result['totals']!=json.loads(close['totals_json']):raise Problem('Rebuilt export does not match its frozen close',409)
            return {**result,'sha256':digest,'verified':True}
    def state(self,account):
        identifier(account)
        with contextlib.closing(self.connect()) as db:
            db.execute('BEGIN')
            events=[dict(r) for r in db.execute('SELECT e.*,p.period posting_period,p.seq posting_seq FROM events e LEFT JOIN posting_batches p ON p.event_seq=e.seq WHERE e.account=? ORDER BY e.seq DESC LIMIT 500',(account,))]
            pending=[dict(r) for r in db.execute('SELECT e.*,NULL posting_period,NULL posting_seq FROM events e WHERE e.account=? AND NOT EXISTS(SELECT 1 FROM posting_batches p WHERE p.event_seq=e.seq) ORDER BY e.seq LIMIT 500',(account,))]
            pending_count=db.execute('SELECT COUNT(*) FROM events e WHERE e.account=? AND NOT EXISTS(SELECT 1 FROM posting_batches p WHERE p.event_seq=e.seq)',(account,)).fetchone()[0]
            for event in events+pending:
                event['quantity']=formatted(int(event['amount']),event['unit']);event.pop('fingerprint')
                entries=db.execute('SELECT amount FROM entries WHERE event_seq=?',(event['seq'],)).fetchall()
                event['net_quantity']=formatted(sum(int(r[0]) for r in entries),event['unit'])
            periods=[dict(r) for r in db.execute('SELECT * FROM periods WHERE account=? ORDER BY period DESC',(account,))]
            totals=[{**dict(r),'quantity':formatted(int(r['amount']),r['unit'])} for r in db.execute('SELECT * FROM totals WHERE account=? ORDER BY period DESC,unit',(account,))]
            closes=[dict(r) for r in db.execute('SELECT id,period,version,ingest_cutoff,posting_cutoff,at,actor,export_sha256 FROM closes WHERE account=? ORDER BY id DESC',(account,))]
            audit=[dict(r) for r in db.execute('SELECT * FROM audit WHERE account=? ORDER BY id DESC LIMIT 100',(account,))]
            return {'account':account,'units':UNITS,'timezone':'UTC','rule_version':RULE_VERSION,'events':events,'pending':pending,'pending_count':pending_count,'periods':periods,'totals':totals,'closes':closes,'audit':audit}