import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Queue, Worker, type ConnectionOptions } from 'bullmq';
import { Redis } from 'ioredis';
import { EVENTS_QUEUE, INGEST_EVENT_JOB, type RawEventEnvelope } from '@lilith/analytics';

/**
 * Dual-redis router — the durability boundary for the edge collector.
 *
 * Normal path: enqueue the envelope to the **primary** queue on black's redis,
 * where the ingest-writer consumes it. When black is unreachable, enqueue to a
 * **local spool** queue on the VPS's own redis (appendonly, durable). This is
 * the "small local recording that only grows when black is unreachable."
 *
 * A health probe pings black's redis; on recovery it resumes a drain worker
 * that forwards every spooled envelope to black. The envelope's `eventId` is
 * the BullMQ `jobId` on both queues, so a forward that overlaps a direct enqueue
 * is deduplicated and the black-side `ON CONFLICT (id) DO NOTHING` makes the
 * whole pipeline idempotent.
 */
@Injectable()
export class RedisRouter implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(RedisRouter.name);

  private primaryQueue!: Queue;
  private spoolQueue!: Queue;
  private drainer!: Worker;
  private probe!: Redis;
  private healthTimer?: ReturnType<typeof setInterval>;

  /** Pessimistic until the first successful probe — never assume black is up. */
  private blackHealthy = false;
  private readonly probeIntervalMs: number;

  constructor(private readonly config: ConfigService) {
    this.probeIntervalMs = Number(this.config.get('BLACK_REDIS_PROBE_MS', 5000));
  }

  onModuleInit(): void {
    const primaryConn = this.blackConnection();
    const spoolConn = this.localConnection();

    this.primaryQueue = new Queue(EVENTS_QUEUE, {
      connection: { ...primaryConn, enableOfflineQueue: false },
      defaultJobOptions: {
        removeOnComplete: true,
        removeOnFail: 5000,
        attempts: 3,
        backoff: { type: 'exponential', delay: 1000 },
      },
    });

    this.spoolQueue = new Queue(EVENTS_QUEUE, {
      connection: spoolConn,
      defaultJobOptions: {
        removeOnComplete: true,
        // Keep failed forwards around so a long outage never loses events.
        removeOnFail: false,
        attempts: Number.MAX_SAFE_INTEGER,
        backoff: { type: 'exponential', delay: 2000 },
      },
    });

    // Drains the local spool to black. Created paused; the probe resumes it
    // only when black is reachable, so it never spins against a dead primary.
    this.drainer = new Worker(
      EVENTS_QUEUE,
      async (job) => {
        const envelope = job.data as RawEventEnvelope;
        await this.primaryQueue.add(INGEST_EVENT_JOB, envelope, { jobId: envelope.eventId });
      },
      { connection: spoolConn, concurrency: 5 },
    );
    this.drainer.on('error', (err) => this.logger.debug(`drainer error: ${err.message}`));
    void this.drainer.pause();

    this.probe = new Redis({
      ...this.redisOptions(primaryConn),
      lazyConnect: false,
      enableOfflineQueue: false,
      maxRetriesPerRequest: 1,
      connectTimeout: 3000,
    });
    this.probe.on('error', () => {
      /* state is driven by the probe loop; swallow connection noise here */
    });

    this.startHealthLoop();
    this.logger.log(
      `RedisRouter up — primary=${this.describe(primaryConn)} spool=${this.describe(spoolConn)}`,
    );
  }

  async onModuleDestroy(): Promise<void> {
    if (this.healthTimer) clearInterval(this.healthTimer);
    await Promise.allSettled([
      this.drainer?.close(),
      this.primaryQueue?.close(),
      this.spoolQueue?.close(),
      this.probe?.quit(),
    ]);
  }

  /**
   * Durably enqueue one envelope. Resolves once redis (primary or spool) has
   * accepted it; rejects only if BOTH are unreachable, so the caller can return
   * a 5xx and the client/relay can retry rather than silently dropping.
   */
  async enqueue(envelope: RawEventEnvelope): Promise<'primary' | 'spool'> {
    if (this.blackHealthy) {
      try {
        await this.primaryQueue.add(INGEST_EVENT_JOB, envelope, { jobId: envelope.eventId });
        return 'primary';
      } catch (err) {
        this.markUnhealthy(err);
        // fall through to the local spool
      }
    }
    await this.spoolQueue.add(INGEST_EVENT_JOB, envelope, { jobId: envelope.eventId });
    return 'spool';
  }

  /** Current depth of the local outage spool (waiting + delayed). */
  async spoolDepth(): Promise<number> {
    const counts = await this.spoolQueue.getJobCounts('waiting', 'delayed', 'active', 'failed');
    return (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0) + (counts.failed ?? 0);
  }

  isBlackHealthy(): boolean {
    return this.blackHealthy;
  }

  private startHealthLoop(): void {
    const tick = async (): Promise<void> => {
      try {
        const pong = await this.probe.ping();
        if (pong === 'PONG') await this.markHealthy();
      } catch (err) {
        this.markUnhealthy(err);
      }
    };
    void tick();
    this.healthTimer = setInterval(() => void tick(), this.probeIntervalMs);
  }

  private async markHealthy(): Promise<void> {
    if (this.blackHealthy) return;
    this.blackHealthy = true;
    this.logger.log('Black redis reachable — resuming spool drain');
    try {
      this.drainer.resume();
    } catch (err) {
      this.logger.warn(`Failed to resume drainer: ${this.msg(err)}`);
    }
  }

  private markUnhealthy(err: unknown): void {
    if (!this.blackHealthy) return;
    this.blackHealthy = false;
    this.logger.warn(`Black redis unreachable — spooling locally: ${this.msg(err)}`);
    void this.drainer.pause();
  }

  private blackConnection(): ConnectionOptions {
    const password = this.config.get<string>('BLACK_REDIS_PASSWORD');
    return {
      host: this.config.get<string>('BLACK_REDIS_HOST', 'black'),
      port: Number(this.config.get('BLACK_REDIS_PORT', 26381)),
      ...(password ? { password } : {}),
    };
  }

  private localConnection(): ConnectionOptions {
    const password = this.config.get<string>('REDIS_PASSWORD');
    return {
      host: this.config.get<string>('REDIS_HOST', 'localhost'),
      port: Number(this.config.get('REDIS_PORT', 6379)),
      ...(password ? { password } : {}),
    };
  }

  /** Narrow a BullMQ ConnectionOptions object to ioredis options for the probe. */
  private redisOptions(conn: ConnectionOptions): Record<string, unknown> {
    return conn as Record<string, unknown>;
  }

  private describe(conn: ConnectionOptions): string {
    const c = conn as { host?: string; port?: number };
    return `${c.host}:${c.port}`;
  }

  private msg(err: unknown): string {
    return err instanceof Error ? err.message : String(err);
  }
}
