import { appendFile, open, readFile, rename, stat, writeFile } from 'node:fs/promises';
import { hostname } from 'node:os';
import { ANALYTICS_GROUP, ANALYTICS_STREAM, CallerAnalyticsStore, Prisma, PrismaClient, analyticsHash, parseAnalyticsEvent, validateAnalyticsEvent, type AnalyticsCall, type AnalyticsEvent } from '@lisglosips/database';
import { createRedisClient } from '@lisglosips/redis';
import { createLogger } from '@lisglosips/observability';
import { evaluateCallerAlerts, suspendCallerAlerts } from './analytics-alerts.js';

const logger = createLogger('caller-analytics'); const db = new PrismaClient();
const redis = createRedisClient(process.env.REDIS_URL ?? ''); const store = new CallerAnalyticsStore(db);
const spool = process.env.ANALYTICS_SPOOL ?? '/var/log/lisglosips/caller-analytics.log';
const checkpoint = process.env.ANALYTICS_CHECKPOINT ?? '/var/lib/lisglosips/caller-analytics-offset.json';
interface CollectorCursor { offset:number; ino:number; gap?:boolean; eventsSinceProducerStart?:number; lastProducerCount?:number }
let cursor:CollectorCursor={offset:0,ino:0,eventsSinceProducerStart:0}; let stopping=false; let lastAlerts=0; let lastTrim=0; let lastAlertSuspend=0;
let dataThrough:string|null=null; let ingestLagMs:number|null=null; let gap=false; let bootId='unknown';
const consumer=`${hostname()}-${process.pid}`;
process.on('SIGTERM',()=>{stopping=true;}); process.on('SIGINT',()=>{stopping=true;});

function decodeLine(line:string):AnalyticsEvent|null {
  const marker=line.indexOf('CRA1|'); if(marker<0)return null; const parts=line.slice(marker).trim().split('|');
  if(parts.length!==20)throw new Error('Malformed producer record');
  const [uid,callId,customerId,customerGatewayId,vendorId,vendorGatewayId,caller,landingCaller,callee,city,carrier,attemptId]=parts.slice(8).map(s=>Buffer.from(s,'base64').toString('utf8'));
  const at=Number(parts[2])*1000+Math.floor(Number(parts[3])/1000); const startedAt=Number(parts[4])*1000+Math.floor(Number(parts[5])/1000); const receivedAt=Date.now();
  const fields={version:'2',eventId:analyticsHash(parts),callUid:analyticsHash([uid,customerId,hostname()]),callId,customerId,customerGatewayId,vendorId,vendorGatewayId,caller,landingCaller,callee,city,carrier,attemptId,
    kind:parts[1],at:String(at),startedAt:String(startedAt),code:parts[6],source:parts[7],talkMs:'-1',sequence:String(at*1000),nodeId:hostname(),bootId,receivedAt:String(receivedAt),method:'INVITE',evidenceVersion:'2'};
  return parseAnalyticsEvent(Object.entries(fields).flat());
}
async function recordGap(kind:string,detail:Record<string,unknown>){const id=analyticsHash([kind,detail,Date.now()]);await db.$executeRaw`INSERT INTO caller_analysis_gaps(id,node_id,kind,status,started_at,payload,created_at,updated_at) VALUES (${id},${hostname()},${kind},'OPEN',NOW(3),${JSON.stringify(detail)},NOW(3),NOW(3))`;}
async function persistCursor(){await writeFile(`${checkpoint}.tmp`,JSON.stringify({...cursor,gap}),{mode:0o600});await rename(`${checkpoint}.tmp`,checkpoint);}
async function collect(){
  const info=await stat(spool); if(cursor.ino&&(cursor.ino!==info.ino||info.size<cursor.offset)){gap=true;await recordGap('SPOOL_ROTATED_OR_TRUNCATED',{previousIno:cursor.ino,currentIno:info.ino,previousOffset:cursor.offset,currentSize:info.size});cursor.offset=0;} cursor.ino=info.ino;
  const file=await open(spool,'r'); try{const buffer=Buffer.alloc(Math.min(1024*1024,Math.max(0,info.size-cursor.offset)));const{bytesRead}=await file.read(buffer,0,buffer.length,cursor.offset);const last=buffer.subarray(0,bytesRead).lastIndexOf(10);if(last<0)return;
    for(const line of buffer.subarray(0,last+1).toString('utf8').split('\n').filter(Boolean)){try{const event=decodeLine(line);if(event){await redis.xadd(ANALYTICS_STREAM,'*','event',JSON.stringify(event));cursor.eventsSinceProducerStart=(cursor.eventsSinceProducerStart??0)+1;}}catch(error){const message=error instanceof Error?error.message:'decode error';await appendFile(`${checkpoint}.deadletter`,`${JSON.stringify({at:new Date(),line,error:message})}\n`,{mode:0o600});gap=true;await recordGap('MALFORMED_SPOOL_RECORD',{offset:cursor.offset,error:message});}}
    cursor.offset+=last+1;await persistCursor();}finally{await file.close();}
}
async function consumeEntries(items:Array<[string,string[]]>){for(const[id,fields]of items){try{const i=fields.findIndex(v=>v==='event');if(i<0||!fields[i+1])throw new Error('Missing event field');const event=validateAnalyticsEvent(JSON.parse(fields[i+1]));await store.consume(event);ingestLagMs=Math.max(0,(event.receivedAt??Date.now())-event.at);if(!dataThrough||event.at>Date.parse(dataThrough))dataThrough=new Date(event.at).toISOString();}catch(error){const message=error instanceof Error?error.message.slice(0,500):'Invalid queued event';const deadId=analyticsHash([ANALYTICS_STREAM,id]);await db.$executeRaw`INSERT IGNORE INTO caller_analysis_dead_letters(id,stream_id,payload,error,status,created_at,updated_at) VALUES (${deadId},${id},${JSON.stringify(fields)},${message},'OPEN',NOW(3),NOW(3))`;gap=true;}await redis.xack(ANALYTICS_STREAM,ANALYTICS_GROUP,id);}}
async function producerCount():Promise<number>{const response=await fetch(process.env.ANALYTICS_MI_URL??'http://127.0.0.1:8888/mi',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({jsonrpc:'2.0',id:1,method:'get_statistics',params:{statistics:['cra_event_total']}}),signal:AbortSignal.timeout(2000)});const body=await response.json()as{result?:Record<string,number>};const value=body.result?.['script:cra_event_total'];if(!response.ok||typeof value!=='number'||!Number.isSafeInteger(value))throw new Error('Producer counter unavailable');return value;}
async function observeActive(){const response=await fetch(process.env.ANALYTICS_MI_URL??'http://127.0.0.1:8888/mi',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({jsonrpc:'2.0',id:1,method:'dlg_list',params:[]}),signal:AbortSignal.timeout(2000)});const body=await response.json()as{result?:{Dialogs?:Array<{callid:string;state:number}>}};if(!response.ok||!body.result?.Dialogs)throw new Error('Dialog observation unavailable');const live=new Map(body.result.Dialogs.map(d=>[d.callid,Number(d.state)]));let lastId='';for(;;){const calls=await db.$queryRaw<Array<{id:string;payload:AnalyticsCall}>>`SELECT c.id,c.payload FROM caller_analysis_calls c WHERE c.id>${lastId} AND EXISTS (SELECT 1 FROM caller_analysis_states s WHERE s.call_key=c.id AND s.view='original' AND CAST(JSON_UNQUOTE(JSON_EXTRACT(s.counts,'$.activeCalls')) AS SIGNED)>0) ORDER BY c.id LIMIT 500`;if(!calls.length)break;const now=Date.now();for(const{payload}of calls)for(const leg of Object.values(payload.legs)){if(leg.endedAt!==null||leg.talkStartedAt==null||leg.talkMs>0)continue;if((live.get(leg.event.callId)??0)===4&&now>leg.talkStartedAt)await store.consume({...leg.event,kind:'DURATION',at:now,talkMs:now-leg.talkStartedAt,source:'dialog-acked-observation',eventId:analyticsHash([leg.event.callUid,leg.event.attemptId,'positive-duration'])});}lastId=calls[calls.length-1].id;}}
async function main(){try{bootId=(await readFile('/proc/sys/kernel/random/boot_id','utf8')).trim();}catch{bootId=`${hostname()}-unknown`;}try{const saved=JSON.parse(await readFile(checkpoint,'utf8'))as CollectorCursor;cursor={...cursor,...saved,eventsSinceProducerStart:saved.eventsSinceProducerStart??0};gap=saved.gap===true;}catch{/* first start */}await redis.connect();await db.$connect();try{await redis.xgroup('CREATE',ANALYTICS_STREAM,ANALYTICS_GROUP,'0','MKSTREAM');}catch(e){if(!(e instanceof Error)||!e.message.includes('BUSYGROUP'))throw e;}let pendingCursor='0-0';while(!stopping){let degraded=gap;let reason='';try{await collect();const produced=await producerCount();if(cursor.lastProducerCount!==undefined&&produced<cursor.lastProducerCount){gap=degraded=true;reason='PRODUCER_RESTART';await recordGap(reason,{previous:cursor.lastProducerCount,current:produced});cursor.eventsSinceProducerStart=0;}cursor.lastProducerCount=produced;await persistCursor();const claimed=await redis.xautoclaim(ANALYTICS_STREAM,ANALYTICS_GROUP,consumer,5000,pendingCursor,'COUNT',100)as[string,Array<[string,string[]]>];pendingCursor=claimed[0];await consumeEntries(claimed[1]);const fresh=await redis.xreadgroup('GROUP',ANALYTICS_GROUP,consumer,'COUNT',200,'BLOCK',100,'STREAMS',ANALYTICS_STREAM,'>')as Array<[string,Array<[string,string[]]>]>|null;if(fresh)for(const[,batch]of fresh)await consumeEntries(batch);await observeActive();const groups=await redis.xinfo('GROUPS',ANALYTICS_STREAM)as Array<Array<string|number>>;const group=groups.map(g=>Object.fromEntries(Array.from({length:g.length/2},(_,i)=>[g[i*2],g[i*2+1]]))).find(g=>g.name===ANALYTICS_GROUP);if(Number(group?.pending??0)||Number(group?.lag??0)){degraded=true;reason='QUEUE_BACKLOG';}if(produced!==(cursor.eventsSinceProducerStart??0)){degraded=true;reason='PRODUCER_COLLECTOR_COUNT_MISMATCH';}if(!Number(group?.pending??0)&&Date.now()-lastTrim>60000){await redis.xtrim(ANALYTICS_STREAM,'MAXLEN','~',Number(process.env.ANALYTICS_STREAM_MAXLEN??1000000));lastTrim=Date.now();}if(Date.now()-lastAlerts>15000&&!degraded){await evaluateCallerAlerts(db);lastAlerts=Date.now();}}catch(e){degraded=true;reason=e instanceof Error?e.message.slice(0,200):'Worker error';logger.error({reason},'Analytics cycle failed');}if(degraded&&Date.now()-lastAlertSuspend>15000){try{await suspendCallerAlerts(db,reason||'ANALYTICS_DEGRADED');lastAlertSuspend=Date.now();}catch{logger.error('Analytics alert suspension failed');}}try{const spoolSize=await stat(spool).then(s=>s.size).catch(()=>null);const coverageStatus=!degraded&&!gap&&cursor.lastProducerCount===cursor.eventsSinceProducerStart?'CONTINUOUS':'DEGRADED';const payload=JSON.stringify({degraded,reason,dataThrough,lagMs:ingestLagMs,coverageStatus,captureSince:process.env.ANALYTICS_CAPTURE_SINCE??null,spoolOffset:cursor.offset,spoolUnreadBytes:spoolSize===null?null:Math.max(0,spoolSize-cursor.offset),gap,producerEventCount:cursor.lastProducerCount??null,collectedEventCount:cursor.eventsSinceProducerStart??null,bootId,updatedAt:new Date().toISOString()});await db.$executeRaw(Prisma.sql`INSERT INTO caller_analysis_health(id,payload,updated_at) VALUES ('worker',${payload},NOW(3)) ON DUPLICATE KEY UPDATE payload=VALUES(payload),updated_at=VALUES(updated_at)`);}catch{logger.error('Analytics heartbeat write failed');}await new Promise(resolve=>setTimeout(resolve,700));}redis.disconnect();await db.$disconnect();}
void main().catch(e=>{logger.error({message:e instanceof Error?e.message:'Startup failed'},'Analytics failed');process.exitCode=1;redis.disconnect();void db.$disconnect();});