import { Injectable, OnApplicationBootstrap, Optional } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
import { createHash } from 'crypto';
import { PluginLoaderService } from '../../core/plugins/plugin-loader.service';
import { IngressJobData } from '../queue/processors/ingress.processor';
import { IntegrationDeliveryFailure } from './entities/integration-delivery-failure.entity';
import { QUEUE_NAMES } from '../queue/queue-names';
import { createLogger } from '../../common/services/logger.service';
import { resolveNonNegativeIntEnv } from '../../config/configuration';

/**
 * Outcome of an enqueue attempt. 'queued' = handed to BullMQ; 'dispatched' = delivered inline; 'failed'
 * = inline dispatch threw and was swallowed (`error` carries the message). enqueue() never throws, so
 * callers use the outcome (not exceptions) to decide durability follow-up: a LIVE ingress delivery must
 * persist a dead-letter row on 'failed' (see buildIngressDeadLetterRow, wired at the IngressService
 * factory) so RedriveService can replay it; RedriveService itself calls enqueue() directly, because a
 * failed replay must keep its EXISTING dead-letter row redrivable rather than write a second one.
 */
export type EnqueueOutcome = { outcome: 'queued' | 'dispatched' | 'failed'; error?: string };

/**
 * Retry policy for inbound ingress jobs. Previously the enqueue passed only a jobId, so BullMQ ran a
 * SINGLE attempt — a transient plugin-sandbox 5xx went straight to the DLQ with no retry (asymmetric
 * with the webhook queue). A few exponential-backoff attempts absorb transient failures; the
 * final-attempt DLQ write in ingress.processor is gated on `job.opts.attempts`, so it still fires
 * exactly once, only after these are exhausted. Env-overridable (INGRESS_MAX_ATTEMPTS /
 * INGRESS_RETRY_DELAY_MS); an invalid value falls back to the default.
 */
export function resolveIngressJobOptions(): { attempts: number; backoff: { type: 'exponential'; delay: number } } {
  const attempts = Number(process.env.INGRESS_MAX_ATTEMPTS);
  return {
    attempts: Number.isInteger(attempts) && attempts >= 1 ? attempts : 3,
    backoff: { type: 'exponential', delay: resolveNonNegativeIntEnv(process.env.INGRESS_RETRY_DELAY_MS, 5000) },
  };
}

/**
 * BullMQ refuses several jobId shapes outright, at two validation sites: Job.validateOptions rejects
 * an exact integer string ("Custom Id cannot be integers") and a colon id that does not split into
 * exactly 3 parts ("Custom Id cannot contain :"); Queue.addJob additionally rejects '0' and any
 * '0:'-prefixed id ("JobId cannot be '0' or start with '0:'"). Providers send numeric dedup headers
 * (`svix-Id: 12345`), and the redrive path mints `redrive:<uuid>`, so these refusals happen in
 * practice, and because enqueue()'s catch-all treats ANY add() throw as "Redis unreachable", the job
 * silently degraded to inline dispatch with no retry, no backoff, and a blocked redrive loop. Map the
 * refused shapes to a deterministic sha256 prefix: BullMQ's exactly-once dedup only needs the id
 * STABLE per delivery, not recognizable, and the reconciler replays through this same function so
 * its dedup against an earlier enqueue is preserved.
 *
 * The hash input is namespaced with the plugin/instance pair. BullMQ dedups jobIds across the WHOLE
 * shared ingress queue, while the database dedup is (pluginId, instanceId, providerDeliveryId), and
 * numeric provider ids are exactly the short, per-account sequence style that two instances of one
 * provider can share. Without the namespace, the second instance's delivery would collide with the
 * first's job id and BullMQ would silently discard it (resolved as the existing job, no DLQ row);
 * with it, the queue-level dedup matches the database-level scope. A non-string id (a duplicated
 * header can surface as string[]) is coerced rather than trusted to reach BullMQ's own checks.
 */
export function sanitizeIngressJobId(jobId: string, namespace = ''): string {
  const raw = typeof jobId === 'string' ? jobId : String(jobId);
  const looksInteger = `${parseInt(raw, 10)}` === raw;
  const badColon = raw.includes(':') && raw.split(':').length !== 3;
  const zeroPrefixed = raw === '0' || raw.startsWith('0:');
  if (!looksInteger && !badColon && !zeroPrefixed) return raw;
  return `ing-${createHash('sha256').update(`${namespace}\u0000${raw}`).digest('hex').slice(0, 40)}`;
}

/**
 * Build the dead-letter row for an ingress delivery whose inline-dispatch fallback failed. The shape
 * mirrors the row IngressProcessor writes on a final-attempt failure (direction / pluginId / instanceId
 * / sessionId / deliveryId / attempts / lastError / payload / redriven), so RedriveService reads either
 * back identically. `attempts` is 1 — the inline path makes a single dispatch attempt (no BullMQ retries).
 */
export function buildIngressDeadLetterRow(data: IngressJobData, error?: string): Partial<IntegrationDeliveryFailure> {
  return {
    direction: 'inbound',
    pluginId: data.pluginId,
    instanceId: data.instanceId,
    sessionId: data.sessionId ?? null,
    deliveryId: data.deliveryId,
    attempts: 1,
    lastError: error ?? 'inline ingress dispatch failed',
    payload: {
      route: data.route,
      method: data.method,
      providerConversationId: data.providerConversationId,
      ingress: data.payload,
    },
    redriven: false,
  };
}

/**
 * Shared queue-or-inline enqueue for inbound ingress jobs. Extracted out of IngressService's DI
 * factory (integration.module.ts) so RedriveService can reuse the exact same behavior when replaying
 * DLQ rows: same queue.add args, same inline dispatch-after-persist fallback, same error swallow.
 * The ingress queue stays @Optional so a queue-off boot (QUEUE_ENABLED unset/false) needs no
 * QueueModule — and no Redis — at all; a missing injection then means inline dispatch, mirroring
 * WebhookService's direct fallback. Under QUEUE_ENABLED=true, IntegrationModule imports QueueModule
 * so the queue provider MUST resolve; onApplicationBootstrap below fails the boot if it did not, so
 * a broken queue wiring crashes at startup instead of silently running inline forever (the queued
 * dispatch contract — fast-ack, retries, ordered processing — would otherwise be dead with no signal).
 */
@Injectable()
export class IngressEnqueueService implements OnApplicationBootstrap {
  private readonly logger = createLogger('IngressEnqueueService');

  constructor(
    private readonly loader: PluginLoaderService,
    private readonly config: ConfigService,
    @Optional() @InjectQueue(QUEUE_NAMES.INGRESS) private readonly ingressQueue?: Queue<IngressJobData>,
  ) {}

  onApplicationBootstrap(): void {
    // Reads the same module-eval signal as the conditional QueueModule imports (integration.module.ts /
    // webhook.module.ts), not the runtime config, so the check guards exactly the wiring that should
    // have happened at import time.
    if (process.env.QUEUE_ENABLED === 'true' && !this.ingressQueue) {
      throw new Error(
        `QUEUE_ENABLED=true but the '${QUEUE_NAMES.INGRESS}' BullMQ queue did not resolve — ` +
          'IntegrationModule must import QueueModule (see the QUEUE_ENABLED conditional in integration.module.ts). ' +
          'Refusing to boot: ingress deliveries would silently dispatch inline, defeating the queued-dispatch contract.',
      );
    }
  }

  async enqueue(data: IngressJobData, jobId: string): Promise<EnqueueOutcome> {
    const queueEnabled = this.config.get<boolean>('queue.enabled', false);
    const useQueue = queueEnabled && !!this.ingressQueue;

    if (useQueue && this.ingressQueue) {
      try {
        // jobId = deliveryId gives BullMQ exactly-once enqueue semantics; the retry policy adds bounded
        // exponential-backoff attempts so a transient failure retries before landing in the DLQ. The id
        // is sanitized because BullMQ refuses several id shapes at add() (see sanitizeIngressJobId),
        // which would otherwise read as a Redis failure here and fall through to inline dispatch.
        await this.ingressQueue.add('ingress', data, {
          jobId: sanitizeIngressJobId(jobId, `${data.pluginId}\u0000${data.instanceId}`),
          ...resolveIngressJobOptions(),
        });
        return { outcome: 'queued' };
      } catch (err) {
        // Redis unreachable (enableOfflineQueue:false makes add() reject) — fall through to inline
        // dispatch. Without this, the already-persisted event would be lost forever: the throw would
        // 500 the ingress request, the provider retries, dedup returns "duplicate", and no job was
        // ever enqueued (no DLQ row either). Mirrors WebhookService's queue-add fallback.
        this.logger.error(
          'Ingress queue add failed; dispatching inline',
          err instanceof Error ? err.message : String(err),
          {
            pluginId: data.pluginId,
            instanceId: data.instanceId,
            route: data.route,
            deliveryId: data.deliveryId,
            action: 'ingress_queue_add_failed',
          },
        );
      }
    }
    // Queue disabled OR queue.add() failed: dispatch inline AFTER the ingress_events row was persisted
    // (persist-before-dispatch still holds), mirroring the webhook direct-delivery fallback.
    try {
      await this.loader.dispatchWebhookForInstance(data);
      return { outcome: 'dispatched' };
    } catch (err) {
      // A duplicate delivery already 200s before this point, so a failure here is a real dispatch error.
      // Log and swallow so the provider still gets its 202 (at-least-once, like the webhook fallback).
      // enqueue() intentionally does NOT write a dead-letter row here — it is shared with RedriveService
      // (a failed replay must not spawn a second DLQ row). The 'failed' outcome + error is returned so the
      // caller decides durability: the live-ingress wiring persists a DLQ row (buildIngressDeadLetterRow).
      const error = err instanceof Error ? err.message : String(err);
      this.logger.error('Inline ingress dispatch failed', error, {
        pluginId: data.pluginId,
        instanceId: data.instanceId,
        route: data.route,
        deliveryId: data.deliveryId,
        action: 'ingress_inline_dispatch_failed',
      });
      return { outcome: 'failed', error };
    }
  }
}