import { NestFactory } from '@nestjs/core';
import IORedis from 'ioredis';
import { ProtocolLogWorkerModule } from './protocol-log-worker.module';
import { ProtocolLogsService, type ProtocolLogInput } from './protocol-logs/protocol-logs.service';

const STREAM = process.env.GATEWAY_PROTOCOL_LOG_STREAM ?? 'gateway.protocol.logs';
const GROUP = process.env.GATEWAY_PROTOCOL_LOG_GROUP ?? 'cmpp-protocol-log-writer';
const CONSUMER = process.env.GATEWAY_PROTOCOL_LOG_CONSUMER ?? `protocol-log-${process.pid}`;
const BATCH_SIZE = boundedEnv('PROTOCOL_LOG_STREAM_BATCH_SIZE', 250, 100, 500);

async function bootstrap() {
  process.env.CMPP_PROCESS_ROLE = 'protocol-log-worker';
  const app = await NestFactory.createApplicationContext(ProtocolLogWorkerModule, { logger: ['log', 'warn', 'error'] });
  const logs = app.get(ProtocolLogsService);
  const redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', { maxRetriesPerRequest: null });
  try { await redis.xgroup('CREATE', STREAM, GROUP, '0', 'MKSTREAM'); } catch (error) {
    if (!String(error).includes('BUSYGROUP')) throw error;
  }
  let stopping = false;
  const stop = () => { stopping = true; };
  process.on('SIGTERM', stop); process.on('SIGINT', stop);
  while (!stopping) {
    const claimed = await redis.xautoclaim(STREAM, GROUP, CONSUMER, 30_000, '0-0', 'COUNT', BATCH_SIZE) as unknown as [string, Array<[string, string[]]>];
    let messages = claimed[1] ?? [];
    if (!messages.length) {
      const reply = await redis.xreadgroup('GROUP', GROUP, CONSUMER, 'COUNT', BATCH_SIZE, 'BLOCK', 2000, 'STREAMS', STREAM, '>') as unknown as Array<[string, Array<[string, string[]]>]> | null;
      if (!reply) continue;
      messages = reply[0]?.[1] ?? [];
    }
    const accepted: string[] = [];
    const parsed: ProtocolLogInput[] = [];
    for (const [id, fields] of messages) {
      const dataIndex = fields.indexOf('data');
      try {
        if (dataIndex < 0) throw new Error('data field missing');
        parsed.push(JSON.parse(fields[dataIndex + 1]) as ProtocolLogInput);
        accepted.push(id);
      } catch (error) {
        console.error(`protocol log stream event ${id} is invalid`, error);
		await redis.xadd(`${STREAM}.dead`, '*', 'sourceId', id, 'error', String(error), 'data', dataIndex >= 0 ? fields[dataIndex + 1] : '');
		await redis.xack(STREAM, GROUP, id); await redis.xdel(STREAM, id);
      }
    }
    logs.recordMany(parsed);
    if (accepted.length && await logs.flushNow()) {
      const pipeline = redis.pipeline();
      for (const id of accepted) pipeline.xack(STREAM, GROUP, id).xdel(STREAM, id);
      await pipeline.exec();
    }
  }
  await logs.flushNow(); await redis.quit(); await app.close();
}

function boundedEnv(name: string, fallback: number, minimum: number, maximum: number) {
  const value = Number(process.env[name] ?? fallback);
  return Number.isInteger(value) && value >= minimum && value <= maximum ? value : fallback;
}

void bootstrap().catch((error) => { console.error(error); process.exitCode = 1; });