"""Private loopback-only observer service. Never publishes authentication or RPC errors."""
from pathlib import Path
import os,time,json,csv,io,signal,re,fcntl
from confirmation_observer import Observer,WINDOW
from confirmation_rpc import CoreRPC,ObservationError
from confirmation_archive import Archive,atomic,encode
BASE=Path(os.environ.get('LW_CONFIRMATION_DIR','/var/www/vhosts/litecoin.watch/.litecoin-private/core-observer'))
DATA=BASE/'data';OUTPUT=BASE/'public';OUTPUT.mkdir(mode=0o700,exist_ok=True)
lock=open(BASE/'collector.lock','a');fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
observer=Observer(BASE/'cohort.sqlite3');archive=Archive(OUTPUT/'archive',observer.db,int(time.time()))
rpc=CoreRPC(DATA);running=True;last_good={};last_problem=None
try:
    previous=json.loads((OUTPUT/'summary.json').read_text())['node']
    if previous.get('observed_at'):last_good=previous
except (OSError,KeyError,ValueError):pass

def stop(*args):
    global running;running=False
signal.signal(signal.SIGTERM,stop);signal.signal(signal.SIGINT,stop)
last_export=last_prune=last_archive=0;last_clock=None;archive_status={'state':'initializing'}
while running:
    started=time.monotonic();now=int(time.time());node=dict(last_good);node.update(state='unavailable',reason='initializing')
    if last_clock and abs((now-last_clock[0])-(started-last_clock[1]))>2:observer.unavailable(now,'clock_gap')
    last_clock=(now,started)
    try:
        # IBD can hold the chain lock for >4 seconds. Diagnostics may wait 20s;
        # eligible timing snapshots STILL have a strict 10s total budget below.
        chain=rpc('getblockchaininfo',timeout=20);net=rpc('getnetworkinfo')
        node={'state':'synchronizing','reason':'initial_block_download','observed_at':int(time.time()),'version':net['subversion'],'height':chain['blocks'],'headers':chain['headers'],'verification_progress':chain['verificationprogress'],'connections':net['connections'],'pruned':chain['pruned']}
        last_good=dict(node)
        if chain['chain']!='main':raise ObservationError('wrong_network')
        if chain['initialblockdownload']:
            observer.unavailable(int(time.time()),'synchronizing')
        elif net['connections']==0:
            node.update(state='disconnected',reason='no_peers');observer.unavailable(int(time.time()),'disconnected')
        elif chain['blocks']!=chain['headers']:
            node.update(state='catching_up',reason='headers_ahead');observer.unavailable(int(time.time()),'catching_up')
        else:
            tip=(chain['blocks'],chain['bestblockhash']);header=rpc('getblockheader',[tip[1]])
            if abs(int(time.time())-header['time'])>7200:raise ObservationError('stale_tip')
            if observer.db.execute('SELECT COUNT(*) FROM cohort').fetchone()[0]>=250000:raise ObservationError('cohort_capacity')
            pool=rpc('getmempoolinfo');node.update(mempool_count=pool['size'],max_mempool_bytes=pool['maxmempool'],mempool_min_fee_ltc_kvb=pool['mempoolminfee'])
            ids=rpc('getrawmempool')
            if not isinstance(ids,list) or len(ids)>100000:raise ObservationError('mempool_capacity')
            if not all(isinstance(x,str) and re.fullmatch('[0-9a-f]{64}',x) for x in ids):raise ObservationError('invalid_mempool')
            old=observer.last_tip();block=None;reorg=False
            if old and old!=tip:
                reorg=old[0]>tip[0] or rpc('getblockhash',[old[0]])!=old[1]
                if tip[0]==old[0]+1 and not reorg:block=rpc('getblock',[tip[1],1])
            if rpc('getbestblockhash')!=tip[1]:raise ObservationError('tip_changed')
            sampled_at=int(time.time());elapsed=time.monotonic()-started
            if elapsed>10:raise ObservationError('slow_snapshot')
            if abs((sampled_at-now)-elapsed)>2:raise ObservationError('clock_gap')
            observer.observe(sampled_at,tip,set(ids),block,reorg)
            canonical={height:rpc('getblockhash',[height]) if height<=tip[0] else None for height in observer.recheck_heights()}
            if rpc('getbestblockhash')!=tip[1]:raise ObservationError('tip_changed')
            observer.recheck(canonical);node.update(state='collecting',reason=None,observed_at=sampled_at)
            last_good=dict(node)
            if observer.previous and observer.db.execute('SELECT state FROM polls WHERE at=?',(sampled_at,)).fetchone()[0]=='collecting':archive.mark_ready(sampled_at)
    except Exception as error:
        code=error.code if isinstance(error,ObservationError) else 'internal_error'
        observer.unavailable(int(time.time()),code);node.update(state=code,reason=code)
        if code!=last_problem:print(json.dumps({'event':'observer_paused','reason':code}),flush=True)
        last_problem=code
    else:
        if last_problem:print(json.dumps({'event':'observer_response_restored','state':node['state']}),flush=True)
        last_problem=None
    now=int(time.time())
    if now-last_prune>=3600:observer.prune(now);last_prune=now
    if now-last_archive>=300 or (archive.meta('study_start') and archive_status.get('study_start') is None):
        try:
            index=archive.update(now);archive_status={k:index[k] for k in ['study_start','study_end','completed_calendar_days']};archive_status.update(state='ready',daily_files=len(index['daily']),updated_at=now,report_available=index['report'] is not None)
        except Exception:
            archive_status={'state':'export_error','updated_at':now};print('{"event":"archive_export_failed"}',flush=True)
        last_archive=now
    if now-last_export>=60:
        stream=io.StringIO(newline='');writer=csv.DictWriter(stream,fieldnames=['txid','first_seen','state','closed_at','confirmed_at','block_height','block_hash']);writer.writeheader();writer.writerows(observer.rows(now-WINDOW));atomic(OUTPUT/'cohort.csv',stream.getvalue().encode());last_export=now
    summary=observer.summary(now,node);summary['archive']=archive_status
    atomic(OUTPUT/'summary.json',encode(summary))
    while running and time.monotonic()-started<15:time.sleep(min(1,15-(time.monotonic()-started)))
observer.db.close()