"""Immutable daily exports and a prospective 30-calendar-day report, standard library only.
Daily filenames refer to poll coverage; cohort states are AS OF the recorded export time.
No historic states are reconstructed. Never rewrite a published daily export or report.
"""
import csv,io,json,hashlib,os,zipfile,datetime,math
from pathlib import Path
from confirmation_observer import RETENTION
DAY=86400
COLUMNS=['txid','first_seen','state','closed_at','confirmed_at','block_height','block_hash']
POLL_COLUMNS=['at','state','covered_seconds','baseline_excluded']
PROTOCOL='lw-confirmations-2026-09-27-v2'

def csv_bytes(rows,columns):
    s=io.StringIO(newline='');w=csv.DictWriter(s,fieldnames=columns);w.writeheader();w.writerows(rows)
    return s.getvalue().encode()

def atomic(path,data):
    path=Path(path);temp=path.with_suffix(path.suffix+'.tmp');temp.write_bytes(data);temp.chmod(0o600);os.replace(temp,path)

def encode(value):return (json.dumps(value,sort_keys=True,separators=(',',':'))+'\n').encode()

def day_name(at):return datetime.datetime.fromtimestamp(at,datetime.timezone.utc).strftime('%Y-%m-%d')

def digest(data):return hashlib.sha256(data).hexdigest()

class Archive:
    def __init__(self,root,db,now):
        self.root=Path(root);self.root.mkdir(exist_ok=True);self.db=db
        self.db.execute('CREATE TABLE IF NOT EXISTS study_metadata(key TEXT PRIMARY KEY,value INTEGER NOT NULL)')
        self.db.execute("INSERT OR IGNORE INTO study_metadata VALUES('archive_started',?)",(now,));self.db.commit()
    def meta(self,key):
        row=self.db.execute('SELECT value FROM study_metadata WHERE key=?',(key,)).fetchone();return row[0] if row else None
    def mark_ready(self,now):
        # The first full UTC day after the first valid collecting interval starts the cohort.
        self.db.execute("INSERT OR IGNORE INTO study_metadata VALUES('study_start',?)",((now//DAY+1)*DAY,));self.db.commit()
    def freeze(self,day,now):
        name=day_name(day)+'.zip';dest=self.root/name
        if dest.exists():return
        end=day+DAY
        polls=[dict(r) for r in self.db.execute('SELECT * FROM polls WHERE at>=? AND at<? ORDER BY at',(day,end))]
        cohort=[dict(r) for r in self.db.execute('SELECT * FROM cohort WHERE first_seen>=? AND first_seen<? ORDER BY first_seen,txid',(max(end-RETENTION,now-RETENTION),end))]
        states={}
        for poll in polls: states[poll['state']]=states.get(poll['state'],0)+1
        files={'cohort.csv':csv_bytes(cohort,COLUMNS),'polls.csv':csv_bytes(polls,POLL_COLUMNS)}
        manifest={'schema':1,'protocol':PROTOCOL,'coverage_day':day_name(day),'coverage_start':day,'coverage_end':end,'snapshot_at':now,'cohort_since':max(end-RETENTION,now-RETENTION),'cohort_before':end,'archive_started':self.meta('archive_started'),'study_start':self.meta('study_start'),'covered_seconds':sum(p['covered_seconds'] for p in polls),'poll_states':states,'baseline_exclusions':sum(p['baseline_excluded'] for p in polls),'cohort_rows':len(cohort),'files':{k:digest(v) for k,v in files.items()},'interpretation':'Polls cover the named UTC day, including zero coverage. Cohort states reflect snapshot_at, not a reconstructed midnight state. Retained cohorts can recur in successive snapshots; never concatenate cohort files as separate transactions.'}
        output=io.BytesIO()
        with zipfile.ZipFile(output,'w',zipfile.ZIP_DEFLATED,compresslevel=6) as z:
            for key,value in files.items():z.writestr(key,value)
            z.writestr('manifest.json',encode(manifest))
        atomic(dest,output.getvalue())
    def update(self,now):
        # Check already published filenames BEFORE generating any missing days.
        # A lost immutable file must be restored, never silently reconstructed.
        prior={}
        if (self.root/'index.json').exists():
            prior={e['file']:e for e in json.loads((self.root/'index.json').read_text())['daily']}
        if any(not (self.root/name).is_file() for name in prior):raise ValueError('Immutable archive missing')
        start=max(self.meta('archive_started')//DAY*DAY,(now-RETENTION)//DAY*DAY)
        for day in range(start,now//DAY*DAY,DAY):self.freeze(day,now)
        # Reuse hashes of unchanged immutable files rather than rereading a growing
        # archive every five minutes on the same CPU quota as the collector.
        paths=sorted(self.root.glob('????-??-??.zip'))
        entries=[]
        for path in paths:
            stat=path.stat();old=prior.get(path.name)
            if old and old.get('mtime_ns')==stat.st_mtime_ns and old['bytes']==stat.st_size:
                entries.append(old);continue
            checksum=digest(path.read_bytes())
            if old and old['sha256']!=checksum:raise ValueError('Immutable archive changed')
            with zipfile.ZipFile(path) as z:manifest=json.loads(z.read('manifest.json'))
            entries.append({'day':manifest['coverage_day'],'snapshot_at':manifest['snapshot_at'],'covered_seconds':manifest['covered_seconds'],'poll_states':manifest['poll_states'],'bytes':stat.st_size,'mtime_ns':stat.st_mtime_ns,'sha256':checksum,'file':path.name})
        start=self.meta('study_start');end=start+30*DAY if start else None
        status={'schema':1,'protocol':PROTOCOL,'generated_at':now,'archive_started':self.meta('archive_started'),'retention_days':90,'daily':entries,'study_start':start,'study_end':end,'completed_calendar_days':max(0,min(30,(now-start)//DAY)) if start else 0,'report':None}
        report_path=self.root/'report-30d.json'
        if start and now>=end and not report_path.exists():
            days=[day_name(d) for d in range(start,end,DAY)]
            if all((self.root/(d+'.zip')).exists() for d in days):
                report=build_report(self.root,start)
                atomic(report_path,encode(report))
        if report_path.exists():status['report']={'file':report_path.name,'sha256':digest(report_path.read_bytes())}
        atomic(self.root/'index.json',encode(status))
        return status

def build_report(root,start):
    root=Path(root);end=start+30*DAY;daily=[];inputs=[];last=None
    for day in range(start,end,DAY):
        path=root/(day_name(day)+'.zip');raw=path.read_bytes()
        with zipfile.ZipFile(io.BytesIO(raw)) as z:
            m=json.loads(z.read('manifest.json'))
            if m['protocol']!=PROTOCOL or m['coverage_start']!=day:raise ValueError('Archive protocol/day mismatch')
            for name,sha in m['files'].items():
                if digest(z.read(name))!=sha:raise ValueError('Archive checksum mismatch')
            rows=list(csv.DictReader(io.StringIO(z.read('polls.csv').decode())))
            coverage=sum(int(p['covered_seconds']) for p in rows)
            if coverage!=m['covered_seconds']:raise ValueError('Coverage mismatch')
            daily.append({'day':m['coverage_day'],'covered_seconds':coverage,'recorded_polls':len(rows),'baseline_exclusions':m['baseline_exclusions'],'poll_states':m['poll_states']})
            if day==end-DAY:
                last=list(csv.DictReader(io.StringIO(z.read('cohort.csv').decode())));snapshot_at=m['snapshot_at']
        inputs.append({'file':path.name,'sha256':digest(raw)})
    cohort=[r for r in last if start<=int(r['first_seen'])<end];counts={};durations=[]
    for r in cohort:
        counts[r['state']]=counts.get(r['state'],0)+1
        if r['state']=='confirmed':durations.append(int(r['confirmed_at'])-int(r['first_seen']))
    durations.sort()
    def quantile(q):
        if len(durations)<20:return None
        i=(len(durations)-1)*q;lo=math.floor(i);hi=math.ceil(i);return round(durations[lo]+(durations[hi]-durations[lo])*(i-lo),1)
    return {'schema':1,'protocol':PROTOCOL,'cohort_start':start,'cohort_end':end,'outcomes_as_of':snapshot_at,'daily':daily,'covered_seconds':sum(d['covered_seconds'] for d in daily),'scheduled_seconds':30*DAY,'sample_denominator':16,'counts':counts,'confirmed_median_seconds':quantile(.5),'confirmed_p90_seconds':quantile(.9),'inputs':inputs,'limits':'Single-node descriptive sample. Outcomes frozen at the last daily snapshot; open or censored waits remain in counts. Confirmed-only quantiles are not completion probabilities. Gaps never extend or restart this fixed calendar period. A later reorg can invalidate an old confirmation; an immutable report is not a claim of permanent finality.'}

if __name__=='__main__':
    import argparse
    p=argparse.ArgumentParser();p.add_argument('archive_directory');p.add_argument('start',help='First UTC cohort date YYYY-MM-DD');p.add_argument('output');args=p.parse_args()
    start=int(datetime.datetime.fromisoformat(args.start).replace(tzinfo=datetime.timezone.utc).timestamp())
    Path(args.output).write_bytes(encode(build_report(args.archive_directory,start)))