#!/usr/bin/env python3
"""Read-only LND -> Optaimaze outcome collector. Credentials remain local.
Docs: https://lightning.engineering/api-docs/api/lnd/lightning/list-payments/
"""
import argparse, datetime, fcntl, hashlib, hmac, json, os, secrets, ssl, sys, time
import urllib.error, urllib.parse, urllib.request
from pathlib import Path

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None

def outcome(payment, salt):
    status=payment.get('status')
    if status not in ('SUCCEEDED','FAILED'):
        return None
    created=int(payment.get('creation_time_ns') or int(payment['creation_date'])*1_000_000_000)
    resolved=max([int(h.get('resolve_time_ns') or 0) for h in payment.get('htlcs',[])]+[created])
    ended=max(created,resolved)
    event={'event_id':'lnd:'+hmac.new(bytes.fromhex(salt),payment['payment_hash'].encode(),hashlib.sha256).hexdigest(),
           'outcome':'succeeded' if status=='SUCCEEDED' else 'failed',
           'amount_sats':int(payment['value_sat']),'fee_msat':int(payment.get('fee_msat') or 0),
           'latency_ms':min(4294967295,max(0,(ended-created)//1_000_000)),
           # HTLC shards do not identify logical retries. Do not count shards as retries.
           'attempts':1,
           'occurred_at':datetime.datetime.fromtimestamp(ended//1_000_000_000,datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}
    if status=='FAILED':
        reason=payment.get('failure_reason') or 'UNKNOWN'
        event['failure_code']=reason if isinstance(reason,str) and reason.replace('_','').isalnum() and reason.isupper() and len(reason)<=60 else 'UNKNOWN'
    return event

def collect_once(fetch, submit, state, now=None, max_pages=10, on_progress=None):
    now=int(time.time() if now is None else now)
    checkpoint=int(state.get('checkpoint',0));offset=checkpoint;pending=None;high=checkpoint;sent=state.setdefault('delivered',{});submitted=0
    for _ in range(max_pages):
        page=fetch(offset);payments=sorted(page.get('payments',[]),key=lambda p:int(p['payment_index']))
        if not payments:
            break
        for payment in payments:
            index=int(payment['payment_index'])
            if index<=offset:
                continue
            high=max(high,index)
            event=outcome(payment,state['salt'])
            if event is None:
                pending=index if pending is None else min(pending,index)
                state['checkpoint']=min(high,pending-1)
                if on_progress:on_progress(state)
                continue
            # The service retains only the most recent 90 days.
            timestamp=int(datetime.datetime.strptime(event['occurred_at'],'%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=datetime.timezone.utc).timestamp())
            if timestamp<now-90*86400:
                state['checkpoint']=min(high,pending-1) if pending is not None else high
                if on_progress:on_progress(state)
                continue
            if str(index) not in sent:
                submit(event);sent[str(index)]=True;submitted+=1
            state['checkpoint']=min(high,pending-1) if pending is not None else high
            if on_progress:on_progress(state)
        next_offset=int(page.get('last_index_offset') or high)
        if next_offset<=offset:
            raise RuntimeError('LND pagination did not advance.')
        offset=next_offset
        if len(payments)<100:
            break
    state['checkpoint']=min(high,pending-1) if pending is not None else high
    state['delivered']={key:value for key,value in sent.items() if int(key)>state['checkpoint']}
    if len(state['delivered'])>10000:
        # Removing this local optimization may cause harmless server-deduplicated resubmission.
        keep=sorted(state['delivered'],key=int)[-10000:]
        state['delivered']={key:True for key in keep}
    return submitted

def save_state(path, state):
    path.parent.mkdir(parents=True,exist_ok=True)
    tmp=path.with_name(path.name+'.tmp');fd=os.open(tmp,os.O_CREAT|os.O_TRUNC|os.O_WRONLY,0o600)
    with os.fdopen(fd,'w') as out:
        json.dump(state,out);out.flush();os.fsync(out.fileno())
    os.replace(tmp,path)

def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--state',type=Path,default=Path.home()/'.local/state/optaimaze/lnd.json')
    parser.add_argument('--max-pages',type=int,default=10)
    args=parser.parse_args()
    if not 1<=args.max_pages<=100:
        parser.error('--max-pages must be between 1 and 100')
    node=os.environ.get('LND_REST_URL','https://localhost:8080').rstrip('/')
    api=os.environ.get('OPTAIMAZE_URL','https://optaimaze.com').rstrip('/')
    for url in (node,api):
        parsed=urllib.parse.urlparse(url)
        if parsed.scheme!='https' or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
            raise RuntimeError('Use HTTPS origins without credentials or query parameters.')
    macaroon=Path(os.environ['LND_READONLY_MACAROON']).read_bytes().hex()
    context=ssl.create_default_context(cafile=os.environ['LND_TLS_CERT'])
    node_client=urllib.request.build_opener(urllib.request.ProxyHandler({}),urllib.request.HTTPSHandler(context=context),NoRedirect())
    api_client=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect())
    key=os.environ['OPTAIMAZE_TELEMETRY_KEY']
    args.state.parent.mkdir(parents=True,exist_ok=True)
    lock_fd=os.open(args.state.with_name(args.state.name+'.lock'),os.O_CREAT|os.O_RDWR,0o600)
    with os.fdopen(lock_fd,'w') as lock:
        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
        state=json.loads(args.state.read_text()) if args.state.exists() else {'checkpoint':0,'salt':secrets.token_hex(32),'delivered':{}}
        save_state(args.state,state) # Persist stable hashing key before the first submission.
        def fetch(offset):
            query=urllib.parse.urlencode({'include_incomplete':'true','index_offset':offset,'max_payments':100,'reversed':'false','creation_date_start':int(time.time())-90*86400,'omit_hops':'true'})
            request=urllib.request.Request(node+'/v1/payments?'+query,headers={'Grpc-Metadata-macaroon':macaroon})
            with node_client.open(request,timeout=20) as response:
                return json.load(response)
        def submit(event):
            request=urllib.request.Request(api+'/api/telemetry',data=json.dumps(event).encode(),headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'})
            with api_client.open(request,timeout=20) as response:
                result=json.load(response)
                if not (result.get('accepted') or result.get('duplicate')):
                    raise RuntimeError('Outcome was not acknowledged.')
            time.sleep(1.1) # Respect the pilot's per-account API rate limit.
        try:
            count=collect_once(fetch,submit,state,max_pages=args.max_pages,on_progress=lambda current:save_state(args.state,current))
            print('Collector complete. New outcomes acknowledged:',count)
        finally:
            save_state(args.state,state)

if __name__=='__main__':
    try:
        main()
    except urllib.error.HTTPError as error:
        print('Collector request failed with HTTP',error.code,'; checkpoint retained. Retry later.',file=sys.stderr);sys.exit(1)
    except Exception as error:
        # Never log raw node responses, URLs, headers, macaroons, or event identifiers.
        print('Collector could not finish ('+type(error).__name__+'). Check configuration; checkpoint retained.',file=sys.stderr);sys.exit(1)