"""Origin-declared observation scope and bounded known-cohort reporting.

The HTTPS registry is an origin statement, not evidence of complete traffic
capture. Reduction consumes independently authenticated intakes and trusted
expectations; it does not authenticate signatures or prove a reported action.
"""
from dataclasses import dataclass
import os
from pathlib import Path
import sqlite3
import stat
import time
from collections import defaultdict
from contextlib import closing

from odexa_ref import crypto, policy
from . import free_contracts as f, storage_contracts as sc, storage_sessions as ss
from .authority_transport import utc_now

VERSION = f.VERSION
PROFILE = 'observation_coverage_v1'
PATH = '/odexa-observation.json'
KINDS = {'gateway','agent','collector','unobserved'}
STATES = {'instrumented','partial','unknown','unobserved'}
EVENT_PROFILES = {'core',ss.PROFILE}
SUBJECTS = {'delivery','action','storage_start','storage_checkpoint','storage_cessation','report'}
need, keys = f.need, f.keys


def when(value): return f.timestamp(value,'time')
def raw(value): return crypto.json_bytes(value)
def sha(value): return crypto.digest(value)


def validate_registry(value, policy_bytes):
    """Closed declaration, exact policy binding; not an operational appointment."""
    crypto.strict_json(raw(value),131072)
    need(isinstance(policy_bytes,bytes),'policy','exact bytes required')
    pol = policy.validate_policy(crypto.strict_json(policy_bytes))
    keys(value,'protocol_version profile registry_id origin revision policy_id policy_revision policy_digest issued_at expires_at boundaries'.split(),'registry')
    need(value['protocol_version']==VERSION and value['profile']==PROFILE,'profile','explicit coverage profile required')
    need(value['origin']==pol['origin'] and value['registry_id']==pol['origin']+PATH,'origin','same-origin registry required')
    need(value['policy_id']==pol['policy_id'] and value['policy_revision']==pol['revision'] and value['policy_digest']==sha(policy_bytes),
         'policy','exact origin policy binding required')
    f.integer(value['revision'],'revision',1)
    issued, expires = when(value['issued_at']), when(value['expires_at'])
    need(when(pol['issued_at']) <= issued < expires <= when(pol['expires_at']),'interval','registry outside policy interval')
    f.array(value['boundaries'],'boundaries',0,128); seen=set(); resource_ids={r['id'] for r in pol['resources']}
    for b in value['boundaries']:
        keys(b,'id kind observer_id binding resource_ids actions purposes starts_at ends_at state event_profiles'.split(),'boundary')
        f.reference(b['id'],'boundary.id',value['origin']); need(b['id'] not in seen,'id','duplicate boundary'); seen.add(b['id'])
        need(b['kind'] in KINDS and b['state'] in STATES,'kind','unknown boundary/state')
        policy.enum_array(b['resource_ids'],resource_ids,'resource_ids')
        policy.enum_array(b['actions'],policy.ACTIONS,'actions'); policy.enum_array(b['purposes'],policy.PURPOSES,'purposes')
        policy.enum_array(b['event_profiles'],EVENT_PROFILES,'event_profiles',0)
        need(issued <= when(b['starts_at']) < when(b['ends_at']) <= expires,'interval','boundary outside registry interval')
        if b['kind']=='unobserved':
            need(b['state'] in {'unobserved','unknown'} and b['observer_id'] is None and b['binding'] is None
                 and b['event_profiles']==[],'unobserved','unobserved boundary cannot appoint an observer')
        else:
            need(b['state']!='unobserved' and b['event_profiles'],'state','observer profile required')
            f.reference(b['observer_id'],'observer_id'); f.validate_binding(b['binding'])
            need(b['binding']['origin']==value['origin'],'binding','wrong observation origin')
            if b['kind']=='collector': need(b['observer_id']==b['binding']['service_id'],'collector','collector service identity required')
            if b['kind']=='gateway': need(b['event_profiles']==['core'],'profile','gateway uses core transport observations')
    return value


@dataclass(frozen=True)
class RegistrySnapshot:
    registry_bytes: bytes
    policy_bytes: bytes
    observation: dict


def validate_snapshot(snapshot):
    need(isinstance(snapshot,RegistrySnapshot),'snapshot','trusted registry snapshot required')
    need(isinstance(snapshot.registry_bytes,bytes) and isinstance(snapshot.policy_bytes,bytes),'snapshot','exact bytes required')
    r = validate_registry(crypto.strict_json(snapshot.registry_bytes),snapshot.policy_bytes)
    o = snapshot.observation
    keys(o,'origin registry_url policy_url registry_digest policy_digest checked_at tls_verified redirected revalidated'.split(),'observation')
    need(o['origin']==r['origin'] and o['registry_url']==r['registry_id'] and o['policy_url']==r['origin']+'/odexa.json',
         'observation','wrong publication origin/URL')
    need(o['registry_digest']==sha(snapshot.registry_bytes) and o['policy_digest']==sha(snapshot.policy_bytes),
         'observation','exact publication bytes mismatch')
    need(o['tls_verified'] is True and o['redirected'] is False and o['revalidated'] is True,'observation','verified fresh HTTPS acquisition required')
    need(when(r['issued_at']) <= when(o['checked_at']) < when(r['expires_at']),'observation','publication not valid when observed')
    return r


class RegistryHistory:
    """Private monotonic publication memory, not a live permission cache."""
    def __init__(self,path):
        self.path=Path(path).absolute(); self.path.parent.mkdir(mode=0o700,parents=True,exist_ok=True)
        info=self.path.parent.lstat()
        need(stat.S_ISDIR(info.st_mode) and info.st_uid==os.getuid() and not info.st_mode & 0o077,'history','private owned directory required')
        fd=os.open(self.path,os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600)
        try:
            info=os.fstat(fd)
            need(stat.S_ISREG(info.st_mode) and info.st_uid==os.getuid() and not info.st_mode & 0o077 and info.st_nlink==1,
                 'history','private owned file required')
        finally: os.close(fd)
        with closing(sqlite3.connect(self.path)) as db:
            tables={row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
            need(tables <= {'registries','registry_reads'},'history','separate registry history required')
            db.execute('CREATE TABLE IF NOT EXISTS registries(origin TEXT,revision INTEGER,policy_id TEXT,policy_revision INTEGER,registry BLOB,policy BLOB,observation BLOB,PRIMARY KEY(origin,revision))')
            db.execute('CREATE TABLE IF NOT EXISTS registry_reads(sequence INTEGER PRIMARY KEY,origin TEXT,revision INTEGER,checked_at TEXT,recorded_at TEXT)')
            for table in ('registries','registry_reads'):
                for action in ('UPDATE','DELETE'):
                    db.execute(f"CREATE TRIGGER IF NOT EXISTS {table}_no_{action} BEFORE {action} ON {table} BEGIN SELECT RAISE(ABORT,'immutable registry'); END")
            db.commit()

    def observe(self,snapshot,*,clock=utc_now):
        # Copy before locking: no caller mutation can change the stored testimony.
        snap=RegistrySnapshot(snapshot.registry_bytes,snapshot.policy_bytes,crypto.strict_json(raw(snapshot.observation)))
        r=validate_snapshot(snap)
        db=sqlite3.connect(self.path,timeout=10,isolation_level=None)
        try:
            db.execute('PRAGMA synchronous=FULL'); db.execute('PRAGMA recursive_triggers=ON'); db.execute('BEGIN IMMEDIATE')
            instant=when(clock())
            need(0 <= (instant-when(snap.observation['checked_at'])).total_seconds() <= 5
                 and instant < when(r['expires_at']),'freshness','registry observation expired before commit')
            previous=db.execute('SELECT revision,policy_id,policy_revision,registry,policy,observation FROM registries WHERE origin=? ORDER BY revision DESC LIMIT 1',(r['origin'],)).fetchone()
            last=db.execute('SELECT checked_at,recorded_at FROM registry_reads WHERE origin=? ORDER BY sequence DESC LIMIT 1',(r['origin'],)).fetchone()
            if last: need(when(snap.observation['checked_at'])>=when(last[0]) and instant>=when(last[1]),'history','observation clock rollback')
            if previous:
                need(r['revision']>=previous[0] and r['policy_id']==previous[1] and r['policy_revision']>=previous[2],
                     'history','registry/policy rollback or changed lineage')
                need(snap.observation['checked_at']>=crypto.strict_json(previous[5])['checked_at'],'history','observation time rollback')
                if r['policy_revision']==previous[2]: need(snap.policy_bytes==previous[4],'history','same policy revision changed bytes')
                if r['revision']==previous[0]: need(snap.registry_bytes==previous[3] and snap.policy_bytes==previous[4],
                     'history','same registry revision changed bytes')
            db.execute('INSERT OR IGNORE INTO registries VALUES(?,?,?,?,?,?,?)',(r['origin'],r['revision'],r['policy_id'],r['policy_revision'],snap.registry_bytes,snap.policy_bytes,raw(snap.observation)))
            db.execute('INSERT INTO registry_reads(origin,revision,checked_at,recorded_at) VALUES(?,?,?,?)',
                       (r['origin'],r['revision'],snap.observation['checked_at'],instant.strftime('%Y-%m-%dT%H:%M:%SZ')))
            db.commit()
        except BaseException: db.rollback(); raise
        finally: db.close()
        return snap

    def get(self,origin,revision):
        policy.origin(origin); f.integer(revision,'revision',1)
        with closing(sqlite3.connect(self.path)) as db:
            row=db.execute('SELECT registry,policy,observation FROM registries WHERE origin=? AND revision=?',(origin,revision)).fetchone()
        need(row is not None,'history','unknown registry')
        return RegistrySnapshot(bytes(row[0]),bytes(row[1]),crypto.strict_json(bytes(row[2])))


def fetch_registry(transport,origin,*,history,clock=utc_now):
    """Credential-free same-origin HTTPS. Does not require a service/provider."""
    policy.origin(origin)
    checked_at=clock(); started=time.monotonic()
    pol=transport.request(origin,'/odexa.json',max_bytes=131072).body
    need(0 <= time.monotonic()-started <= 5,'freshness','registry acquisition exceeded five seconds')
    body=transport.request(origin,PATH,max_bytes=131072).body
    need(0 <= time.monotonic()-started <= 5,'freshness','registry acquisition exceeded five seconds')
    snap=RegistrySnapshot(body,pol,dict(origin=origin,registry_url=origin+PATH,policy_url=origin+'/odexa.json',
        registry_digest=sha(body),policy_digest=sha(pol),checked_at=checked_at,tls_verified=True,redirected=False,revalidated=True))
    return history.observe(snap,clock=clock)


def _resources(pol,url):
    parsed=policy.parse_url(url)
    if parsed['origin']!=pol['origin']: return set()
    return {r['id'] for r in pol['resources'] if any(
        (s['query']=='any' or not parsed['has_query']) and
        (parsed['path']==s['path'] if s['type']=='exact' else parsed['path'].startswith(s['path']))
        for s in r['selectors'])}


def _subject(value):
    keys(value,'kind id reporter_id ingress_id checkpoint_index payload_digest'.split(),'subject')
    need(value['kind'] in SUBJECTS,'subject','unknown expected observation')
    f.identifier(value['id'],'subject.id'); f.reference(value['reporter_id'],'subject.reporter_id')
    if value['kind']=='delivery': f.reference(value['ingress_id'],'ingress_id')
    else: need(value['ingress_id'] is None,'ingress_id','only delivery has ingress')
    if value['kind']=='storage_checkpoint': f.integer(value['checkpoint_index'],'checkpoint_index',1)
    else: need(value['checkpoint_index'] is None,'checkpoint_index','only checkpoint has index')
    if value['kind']=='report': f.hash_value(value['payload_digest'],'payload_digest')
    else: need(value['payload_digest'] is None,'payload_digest','only exact report receipt has digest')
    return tuple(value[k] for k in ('kind','id','reporter_id','ingress_id','checkpoint_index','payload_digest'))


def validate_expectation(value,origin):
    keys(value,'id boundary_id resource_url action purpose agreement_id expected_at due_at subject event_profile basis'.split(),'expectation')
    f.identifier(value['id'],'id'); f.reference(value['boundary_id'],'boundary_id',origin)
    f.reference(value['resource_url'],'resource_url',origin,query=True)
    need(value['action'] in policy.ACTIONS and value['purpose'] in policy.PURPOSES,'scope','unknown tuple')
    need(when(value['expected_at']) <= when(value['due_at']),'deadline','deadline before expectation')
    need(value['event_profile'] in EVENT_PROFILES,'profile','explicit expected event profile required')
    need(value['basis'] in {'instrumented_ingress','gateway_admission','accepted_duty','report_outbox'},'basis','known expectation source required')
    if value['agreement_id'] is not None: f.identifier(value['agreement_id'],'agreement_id')
    else: need(value['basis'] not in {'gateway_admission','accepted_duty'},'agreement_id','accepted agreement context required')
    subject=_subject(value['subject']); kind=subject[0]
    need((kind=='delivery' and value['basis'] in {'instrumented_ingress','gateway_admission'}) or
         (kind in {'action','storage_start','storage_checkpoint','storage_cessation'} and value['basis']=='accepted_duty') or
         (kind=='report' and value['basis']=='report_outbox'),'basis','expectation basis does not establish this cohort')
    need((kind.startswith('storage_') and value['event_profile']==ss.PROFILE) or not kind.startswith('storage_'),
         'profile','storage expectation requires selected session profile')
    return value


def _event_subject(event,boundary):
    kind=boundary['kind']; reporter=event['reporter_id']
    if kind=='collector': return ('report',event['event_id'],reporter,None,None,sha(raw(event)))
    if kind=='gateway':
        h=event['http']
        if event['source']!='origin_observed' or h is None or h['hop_role']!='end_client' or h['boundary_id']!=boundary['id']: return None
        return ('delivery',h['delivery_id'],reporter,h['ingress_id'],None,None)
    if kind=='agent' and event['source']=='client_reported' and event['event_type']=='use.reported':
        op=event['operation']
        if op.get('kind')=='storage_session':
            trigger=op['storage']['trigger']
            return ('storage_'+trigger,op['id'],reporter,None,op['storage']['checkpoint_index'] if trigger=='checkpoint' else None,None)
        if op['state'] in {'completed','failed'}: return ('action',op['id'],reporter,None,None,None)
    return None


def _ratio(n,d): return dict(numerator=n,denominator=d,percent=round(100*n/d,6) if d else None)


def reduce_coverage(snapshot,*,query,verified_intakes,expectations,as_of):
    """One URL/action/purpose/time cohort; all intakes must be authenticated first.

    Expectations come from an independent trusted ingress/admission/duty/outbox
    journal, never guessed from received reports. No global traffic denominator.
    Historical registry snapshots remain origin declarations at observation time.
    """
    registry=validate_snapshot(snapshot); pol=policy.validate_policy(crypto.strict_json(snapshot.policy_bytes))
    keys(query,'url action purpose starts_at ends_at'.split(),'query')
    f.reference(query['url'],'query.url',registry['origin'],query=True)
    need(query['action'] in policy.ACTIONS and query['purpose'] in policy.PURPOSES,'query','unknown tuple')
    start,end,instant=when(query['starts_at']),when(query['ends_at']),when(as_of)
    need(start < end <= instant,'query','closed historical query interval required')
    f.array(verified_intakes,'verified_intakes',0,4096); f.array(expectations,'expectations',0,4096)
    selected=_resources(pol,query['url'])
    bounds={b['id']:b for b in registry['boundaries'] if selected and selected<=set(b['resource_ids'])
            and query['action'] in b['actions'] and query['purpose'] in b['purposes']}
    result=dict(profile=PROFILE,registry_digest=sha(snapshot.registry_bytes),registry_revision=registry['revision'],
        registry_observed_at=snapshot.observation['checked_at'],registry_assurance='origin_declared_at_observation',
        query=dict(query),as_of=as_of,scope_segments=[],boundaries=[],unknown_expectations=[],
        invalid_records=0,unverified_records=0,future_records=0,exact_retries=0,event_conflicts=0,
        expectation_retries=0,expectation_conflicts=0,unmatched_records=0,
        denominator_scope='known_expected_boundary_observations',global_traffic_denominator=None,
        global_capture_percent=None,downstream_use_verified=False,resource_permission_granted=False,
        boundary_counts_are_not_additive=True)
    points=sorted({start,end}|{max(start,min(end,when(b[t]))) for b in bounds.values() for t in ('starts_at','ends_at')})
    for a,z in zip(points,points[1:]):
        covering=[b for b in bounds.values() if when(b['starts_at'])<=a and z<=when(b['ends_at'])]
        states={b['state'] for b in covering}
        state='unknown' if not covering or 'unknown' in states else 'partial' if 'partial' in states or len(states)>1 else next(iter(states))
        result['scope_segments'].append(dict(starts_at=a.strftime('%Y-%m-%dT%H:%M:%SZ'),ends_at=z.strftime('%Y-%m-%dT%H:%M:%SZ'),
            declared_state=state,boundary_ids=sorted(b['id'] for b in covering),overlap=len(covering)>1))
    groups=defaultdict(list)
    for intake in verified_intakes:
        try:
            payload=crypto.unb64u(intake['payload_b64url']); event=crypto.strict_json(payload)
            profile=event.get('event_profile','core')
            need(profile in EVENT_PROFILES,'profile','unselected event profile')
            (sc.validate_intake if profile==ss.PROFILE else f.validate_intake)(intake)
            if intake['assurance']=='unverified': result['unverified_records']+=1; continue
            received,occurred=when(intake['received_at']),when(event['occurred_at'])
            need(occurred<=received,'time','receipt before claimed observation')
            if received>instant or occurred>instant: result['future_records']+=1; continue
            need(event['origin']==registry['origin'],'origin','wrong origin')
            groups[(event['reporter_id'],event['event_id'])].append((intake,event,payload,profile))
        except (ValueError,KeyError,TypeError): result['invalid_records']+=1
    records=[]
    for key,group in sorted(groups.items()):
        if len({row[2] for row in group})>1: result['event_conflicts']+=1; continue
        result['exact_retries']+=len(group)-1
        records.append(min(group,key=lambda r:(r[0]['received_at'],r[0]['record_id'])))
    by_id=defaultdict(list)
    def in_cohort(e):
        return e['resource_url']==query['url'] and e['action']==query['action'] and e['purpose']==query['purpose'] and start<=when(e['expected_at'])<end
    for e in expectations:
        validate_expectation(e,registry['origin']); by_id[e['id']].append(e)
    candidates=[]; conflicted=set()
    for group in by_id.values():
        shapes={raw(e) for e in group}
        if len(shapes)>1:
            for e in group:
                if in_cohort(e): conflicted.add((e['boundary_id'],_subject(e['subject'])))
            candidates.extend({raw(e):e for e in group}.values())
        else:
            result['expectation_retries']+=len(group)-1; candidates.append(group[0])
    expected=defaultdict(list)
    for e in candidates:
        if in_cohort(e):
            expected[(e['boundary_id'],_subject(e['subject']))].append(e)
    for key,group in expected.items():
        if len({raw({k:v for k,v in e.items() if k!='id'}) for e in group})>1: conflicted.add(key)
        else: result['expectation_retries']+=len(group)-1
    # Conflicting expectations retain a visible denominator entry even if the
    # conflicting identity variants were discarded before cohort selection.
    for key in conflicted: expected.setdefault(key,[])
    result['expectation_conflicts']=len(conflicted)
    used=set()
    for bid,b in sorted(bounds.items()):
        row=dict(boundary_id=bid,kind=b['kind'],declared_state=b['state'],observer_id=b['observer_id'],
            expected=0,timely=0,late=0,missing=0,pending=0,conflicted=0,unresolved_admissions=0,
            observed=0,profile_mismatches=0,unexpected=0,outcomes={},expectations=[])
        targets={s:entries for (boundary,s),entries in expected.items() if boundary==bid}
        evidence=defaultdict(list)
        for intake,event,payload,profile in records:
            event_time=when(intake['received_at'] if b['kind']=='collector' else event['occurred_at'])
            if b['kind']=='unobserved' or not when(b['starts_at'])<=event_time<when(b['ends_at']): continue
            actor=intake['collector_id'] if b['kind']=='collector' else event['reporter_id']
            if actor!=b['observer_id'] or any(intake[k]!=b['binding'][k] for k in f.BINDING): continue
            if event['resource_url']!=query['url'] or query['action'] not in event['actions'] or query['purpose'] not in event['purposes']: continue
            if event['policy_id']!=registry['policy_id'] or event['policy_revision']!=registry['policy_revision']: continue
            if not set(event['actions'])<=set(b['actions']) or not set(event['purposes'])<=set(b['purposes']): continue
            subject=_event_subject(event,b)
            if subject is None: continue
            if b['kind']=='collector': subject=(*subject[:-1],sha(payload))  # Original bytes, never reserialized digest.
            if not start<=event_time<end and subject not in targets: continue
            if profile not in b['event_profiles']: row['profile_mismatches']+=1; continue
            semantic=raw({k:v for k,v in event.items() if k!='event_id'}) if b['kind']!='collector' else payload
            evidence[subject].append((intake,event,semantic,event_time))
            if start<=event_time<end: row['observed']+=1
            used.add((event['reporter_id'],event['event_id']))
        for subject,entries in sorted(targets.items(),key=lambda v:raw(list(v[0]))):
            row['expected']+=1
            chosen=entries[0] if entries else None
            matches=evidence.get(subject,[])
            shape_count=len({x[2] for x in matches})
            if (bid,subject) in conflicted or shape_count>1: status='conflicted'
            elif chosen is None or b['kind']=='unobserved' or not when(b['starts_at'])<=when(chosen['expected_at'])<when(b['ends_at']): status='unscoped'
            elif chosen['event_profile'] not in b['event_profiles']: status='profile_mismatch'
            elif ((b['kind']=='collector' and subject[0]!='report') or
                  (b['kind']=='gateway' and subject[0]!='delivery') or
                  (b['kind']=='agent' and subject[0] in {'delivery','report'}) or
                  (b['kind']!='collector' and subject[2]!=b['observer_id'])): status='binding_mismatch'
            elif matches and not any(x[1].get('event_profile','core')==chosen['event_profile'] for x in matches): status='profile_mismatch'
            elif matches and not any(x[1]['agreement_id']==chosen['agreement_id'] for x in matches): status='binding_mismatch'
            else:
                eligible=[x for x in matches if x[3]>=when(chosen['expected_at']) and
                          x[1]['agreement_id']==chosen['agreement_id'] and x[1].get('event_profile','core')==chosen['event_profile']]
                if eligible:
                    first=min(eligible,key=lambda x:x[0]['received_at'])
                    status='timely' if when(first[0]['received_at'])<=when(chosen['due_at']) else 'late'
                    event=first[1]
                    gateway_outcome=('gateway_'+event['http']['kind']+'_write_completed' if event['event_type']=='delivery.completed'
                        else 'gateway_write_failed' if event['event_type']=='delivery.failed' else 'gateway_response_only') if b['kind']=='gateway' else None
                    outcome='collector_received' if b['kind']=='collector' else gateway_outcome if b['kind']=='gateway' else (
                        'client_storage_'+event['operation']['storage']['trigger'] if event['operation'].get('kind')=='storage_session'
                        else 'client_action_'+event['operation']['state'])
                    row['outcomes'][outcome]=row['outcomes'].get(outcome,0)+1
                else: status='missing' if instant>when(chosen['due_at']) else 'pending'
            if status in row: row[status]+=1
            else: row.setdefault(status,0); row[status]+=1
            if chosen and chosen['basis']=='gateway_admission' and status not in {'timely','late'}: row['unresolved_admissions']+=1
            row['expectations'].append(dict(subject=dict(zip(('kind','id','reporter_id','ingress_id','checkpoint_index','payload_digest'),subject)),status=status))
        row['unexpected']=len(set(evidence)-set(targets))
        row['timely_ratio']=_ratio(row['timely'],row['expected'])
        row['received_ratio']=_ratio(row['timely']+row['late'],row['expected'])
        result['boundaries'].append(row)
    for (bid,subject),entries in sorted(expected.items(),key=lambda v:raw([v[0][0],list(v[0][1])])):
        if bid not in bounds: result['unknown_expectations'].append(dict(boundary_id=bid,subject=list(subject),status='unknown_boundary_or_scope'))
    result['unmatched_records']=len(records)-len(used)
    return result