import {
  getRedisUrl,
  initializeConfigFromDatabase,
  startConfigSubscription,
  useEnvironmentConfigFallback,
} from './config';
import dns from 'dns';
import Redis from 'ioredis';
import { randomUUID } from 'crypto';

// Set Google DNS for resolution as Tailscale might not resolve dev3.press.zone correctly
dns.setServers(['8.8.8.8', '4.4.4.4']);

/**
 * Background Job Worker
 *
 * Processes async translation jobs using Bull queue
 */

import { PrismaClient } from '@prisma/client';
import { logger } from './utils/logger';
import { geminiClient } from './services/geminiClient';
import { exceptionService } from './services/exceptionService';
import { verifyExceptions } from './services/exceptionVerifier';
import * as webhookService from './services/webhookService';

const deliverWebhook = webhookService.deliverWebhook;
import { mapStoredTranslation } from './services/translationService';
import {
  countSiteContentCharacters,
  siteContentCallbackMetadata,
  SiteContentJobSubmitRequest,
  translateSiteContentResource,
} from './services/siteContentTranslation';
import { countSourceCharacters, calculateCost, calculateCustomerCostByChars } from './utils/tokenCalculation';
import {
  canonicalizeStructuredFields,
  countStructuredCharacters,
  serializeStructuredFields,
  splitStructuredTranslation,
  StructuredFields,
  STRUCTURED_FIELD_KEY_PATTERN,
  MAX_STRUCTURED_FIELDS,
} from './utils/structuredFields';
import {
  GEMINI_CALL_CHAR_BUDGET,
  GEMINI_CALL_MAX_PAYLOAD_BYTES,
  GEMINI_CALL_MAX_STRINGS,
} from './config/batchLimits';
import { planBatches } from './utils/batchPlanner';
import { trackTranslationJob, trackTokensProcessed } from './utils/metrics';
import {
  TranslationStatus,
  BulkTranslationJobData,
  BulkContentTranslationJobData,
  BulkContentItemResult,
  Tone,
  WebhookPayload,
} from './types';
import { translationQueue } from './queue';
import { reconcilePermanentOrphanDisputes } from './workers/orphanDisputeReconciler';
import { WorkerHeartbeat, getWorkerHeartbeatKey } from './workerHeartbeat';
import { shutdownWorkerRuntime, startWorkerRuntime } from './workerLifecycle';
import { JobCancellationWonError, lockProcessingJob } from './workerLock';
import { configureBulkQueueDispatcher, startBulkQueueDispatcher, stopBulkQueueDispatcher } from './queue/bulkQueueDispatcher';

const prisma = new PrismaClient();
configureBulkQueueDispatcher(prisma.translationJob);
const workerId = (process.env.WORKER_ID || process.env.HOSTNAME)?.trim();
if (!workerId) {
  throw new Error('WORKER_ID or HOSTNAME is required for worker identity');
}
// Keep the Bull job concurrency separate from the provider-call ceiling. The latter is
// process-wide, so concurrent bulk-content jobs cannot multiply Gemini calls without bound.
const jobConcurrency = Math.max(1, Number(process.env.WORKER_JOB_CONCURRENCY || 8));

const DEFAULT_BULK_CONTENT_GEMINI_CONCURRENCY = 1;
const MAX_BULK_CONTENT_GEMINI_CONCURRENCY = 32;
const DEFAULT_BULK_CONTENT_GEMINI_MAX_RETRIES = 2;
const MAX_BULK_CONTENT_GEMINI_MAX_RETRIES = 3;
const DEFAULT_BULK_CONTENT_GEMINI_RETRY_BASE_DELAY_MS = 250;
const MAX_BULK_CONTENT_GEMINI_RETRY_BASE_DELAY_MS = 1_000;
const DEFAULT_BULK_CONTENT_GEMINI_RETRY_MAX_DELAY_MS = 2_000;
const MAX_BULK_CONTENT_GEMINI_RETRY_MAX_DELAY_MS = 5_000;

function parseBoundedPositiveInteger(value: string | undefined, fallback: number, maximum: number): number {
  if (!value?.trim()) {
    return fallback;
  }
  const parsed = Number(value);
  if (!Number.isSafeInteger(parsed) || parsed < 1) {
    return fallback;
  }
  return Math.min(parsed, maximum);
}

function parseBoundedNonNegativeInteger(value: string | undefined, fallback: number, maximum: number): number {
  if (!value?.trim()) {
    return fallback;
  }
  const parsed = Number(value);
  if (!Number.isSafeInteger(parsed) || parsed < 0) {
    return fallback;
  }
  return Math.min(parsed, maximum);
}

function configuredWorkerValue(...names: string[]): string | undefined {
  return names.map((name) => process.env[name]).find((value) => value !== undefined && value.trim() !== '');
}

// WORKER_GEMINI_CONCURRENCY is the canonical setting. The more specific aliases keep
// deployments that already name this limit after the bulk-content worker compatible.
const bulkContentGeminiConcurrency = parseBoundedPositiveInteger(
  configuredWorkerValue(
    'WORKER_GEMINI_CONCURRENCY',
    'WORKER_BULK_CONTENT_GEMINI_CONCURRENCY',
    'WORKER_BULK_CONTENT_CONCURRENCY'
  ),
  DEFAULT_BULK_CONTENT_GEMINI_CONCURRENCY,
  MAX_BULK_CONTENT_GEMINI_CONCURRENCY
);
const bulkContentGeminiMaxRetries = parseBoundedNonNegativeInteger(
  configuredWorkerValue('WORKER_GEMINI_MAX_RETRIES', 'WORKER_BULK_CONTENT_MAX_RETRIES'),
  DEFAULT_BULK_CONTENT_GEMINI_MAX_RETRIES,
  MAX_BULK_CONTENT_GEMINI_MAX_RETRIES
);
const bulkContentGeminiRetryBaseDelayMs = parseBoundedPositiveInteger(
  configuredWorkerValue('WORKER_GEMINI_RETRY_BASE_DELAY_MS', 'WORKER_BULK_CONTENT_RETRY_BASE_DELAY_MS'),
  DEFAULT_BULK_CONTENT_GEMINI_RETRY_BASE_DELAY_MS,
  MAX_BULK_CONTENT_GEMINI_RETRY_BASE_DELAY_MS
);
const bulkContentGeminiRetryMaxDelayMs = Math.max(
  bulkContentGeminiRetryBaseDelayMs,
  parseBoundedPositiveInteger(
    configuredWorkerValue('WORKER_GEMINI_RETRY_MAX_DELAY_MS', 'WORKER_BULK_CONTENT_RETRY_MAX_DELAY_MS'),
    DEFAULT_BULK_CONTENT_GEMINI_RETRY_MAX_DELAY_MS,
    MAX_BULK_CONTENT_GEMINI_RETRY_MAX_DELAY_MS
  )
);

class AsyncSemaphore {
  private inFlight = 0;
  private readonly waiters: Array<() => void> = [];

  constructor(private readonly limit: number) {}

  async acquire(): Promise<() => void> {
    if (this.inFlight >= this.limit || this.waiters.length > 0) {
      await new Promise<void>((resolve) => this.waiters.push(resolve));
    }
    this.inFlight += 1;
    let released = false;
    return () => {
      if (released) {
        return;
      }
      released = true;
      this.inFlight -= 1;
      this.waiters.shift()?.();
    };
  }

  async runExclusive<T>(operation: () => Promise<T>): Promise<T> {
    const release = await this.acquire();
    try {
      return await operation();
    } finally {
      release();
    }
  }
}

const bulkContentGeminiSemaphore = new AsyncSemaphore(bulkContentGeminiConcurrency);
let configSubscription: ReturnType<typeof startConfigSubscription> | undefined;
const workerHeartbeat = new WorkerHeartbeat({
  redis: new Redis(getRedisUrl()),
  checkQueueReady: () => translationQueue.isReady(),
  checkRuntimeReady: () => {
    if (!configSubscription) {
      return Promise.reject(new Error('Config subscription is not initialized'));
    }
    return configSubscription.checkReady();
  },
  checkDatabase: () => prisma.$queryRaw`SELECT 1`,
  key: getWorkerHeartbeatKey(workerId),
  ttlSeconds: Number(process.env.WORKER_HEARTBEAT_TTL_SECONDS || 15),
});

interface TranslationJobData {
  jobId: string;
  /** Canonical submission identity; absent only for legacy queue payloads. */
  submissionId?: string;
  clientJobId?: string;
  userId: string;
  content?: string;
  fields?: StructuredFields;
  siteContent?: SiteContentJobSubmitRequest;
  sourceLang: string;
  targetLang: string;
  tone: string;
  exceptions?: unknown;
  callbackUrl?: string;
  callbackSecret?: string;
}

type LegacyWebhookOutboxEntry = {
  id: string;
  job_id: string;
  attempts: number;
  payload: unknown;
  claimed_at?: Date | null;
  next_attempt_at?: Date | null;
  job: { callback_url: string | null; callback_secret: string | null };
};

type WebhookOutboxDelegate = {
  findMany(_args: object): Promise<LegacyWebhookOutboxEntry[]>;
  findFirst?: (_args: object) => Promise<unknown>;
  updateMany(_args: object): Promise<{ count: number }>;
  create(_args: object): Promise<unknown>;
};

type PrismaWithWebhookOutbox = PrismaClient & { webhookOutbox: WebhookOutboxDelegate };

export const OUTBOX_MAX_ATTEMPTS = 5;
const OUTBOX_LEASE_MS = 60_000;
const OUTBOX_BASE_DELAY_MS = 5_000;
const OUTBOX_MAX_DELAY_MS = 60_000;
const OUTBOX_BATCH_SIZE = 100;

type OutboxDrainResult = {
  claimed: number;
  delivered: number;
  failed: number;
  deadLettered: number;
};

let outboxDrainPromise: Promise<OutboxDrainResult> | undefined;
let outboxStartPromise: Promise<void> | undefined;
let outboxTimer: ReturnType<typeof setTimeout> | undefined;
let outboxStarted = false;
let outboxStopRequested = false;
let outboxBackoffMs = OUTBOX_BASE_DELAY_MS;

const PROCESSING_LEASE_MS = 120_000;
const PROCESSING_LEASE_RENEWAL_MS = 30_000;

export type ProcessingLeaseClaim = {
  claimed: boolean;
  leaseId: string;
};

export async function claimTranslationJob(
  jobId: string,
  leaseId = randomUUID(),
  now = new Date()
): Promise<ProcessingLeaseClaim> {
  const expiresAt = new Date(now.getTime() + PROCESSING_LEASE_MS);
  const result = await prisma.translationJob.updateMany({
    where: {
      id: jobId,
      OR: [
        { status: 'pending' },
        { status: 'processing', processing_lease_expires_at: { lt: now } },
      ],
    },
    data: {
      status: 'processing' as TranslationStatus,
      processing_lease_id: leaseId,
      processing_lease_expires_at: expiresAt,
    },
  });
  return { claimed: result.count === 1, leaseId };
}

function startProcessingLeaseHeartbeat(jobId: string, leaseId: string): () => void {
  const timer = setInterval(() => {
    void prisma.translationJob.updateMany({
      where: { id: jobId, status: 'processing', processing_lease_id: leaseId },
      data: { processing_lease_expires_at: new Date(Date.now() + PROCESSING_LEASE_MS) },
    }).catch((error) => logger.error('Translation processing lease renewal failed', { jobId, error }));
  }, PROCESSING_LEASE_RENEWAL_MS);
  timer.unref?.();
  return () => clearInterval(timer);
}

async function releaseProcessingLease(jobId: string, leaseId: string): Promise<boolean> {
  const result = await prisma.translationJob.updateMany({
    where: { id: jobId, status: 'processing', processing_lease_id: leaseId },
    data: { status: 'pending' as TranslationStatus, processing_lease_id: null, processing_lease_expires_at: null },
  });
  return result.count === 1;
}

function isTerminalWorkerError(error: unknown): boolean {
  const message = error instanceof Error ? error.message : '';
  return /insufficient credits|validation|invalid/i.test(message);
}

type BulkContentItem = BulkContentTranslationJobData['items'][number];

type BulkContentItemOutcome = {
  result: BulkContentItemResult;
  charactersUsed: number;
};

type GeminiRetryClassifier = {
  isRetryableError?: (error: unknown) => boolean;
};

// Use the provider client's existing retry classification without duplicating its error rules.
// The method is intentionally private to GeminiClient; this runtime cast keeps the worker tied to
// the same classification while avoiding a second, inevitably divergent list of error strings.
function isRetryableGeminiError(error: unknown): boolean {
  const classifier = geminiClient as unknown as GeminiRetryClassifier;
  return typeof classifier.isRetryableError === 'function'
    ? classifier.isRetryableError.call(geminiClient, error)
    : false;
}

function getBulkContentRetryDelayMs(retryNumber: number): number {
  const exponentialDelay = Math.min(
    bulkContentGeminiRetryMaxDelayMs,
    bulkContentGeminiRetryBaseDelayMs * Math.pow(2, Math.max(0, retryNumber - 1))
  );
  // Full jitter prevents concurrently failing jobs from retrying in lockstep. The configured
  // retry count and delay caps bound the total added latency for an item.
  return Math.max(1, Math.floor(Math.random() * (exponentialDelay + 1)));
}

function waitForBulkContentRetry(delayMs: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, delayMs));
}

async function translateStructuredWithRateLimit(
  fields: StructuredFields,
  sourceLang: string,
  targetLang: string,
  tone: Tone,
  jobId: string,
  ref: string
) {
  let retryNumber = 0;

  while (true) {
    try {
      // Acquire for each provider attempt, not for the backoff sleep. This keeps the semaphore's
      // count equal to actual in-flight Gemini calls and lets other jobs use the released slot.
      return await bulkContentGeminiSemaphore.runExclusive(() =>
        geminiClient.translateStructured(fields, sourceLang, targetLang, tone)
      );
    } catch (error) {
      if (!isRetryableGeminiError(error) || retryNumber >= bulkContentGeminiMaxRetries) {
        throw error;
      }

      retryNumber += 1;
      const delayMs = getBulkContentRetryDelayMs(retryNumber);
      logger.warn('Retrying bulk-content Gemini item after a retryable error', {
        jobId,
        ref,
        retryNumber,
        delayMs,
        error,
      });
      await waitForBulkContentRetry(delayMs);
    }
  }
}

type NormalizedBulkContentItemFields = {
  fields: StructuredFields;
  /** True only for the post-migration queue shape carrying `fields`. */
  canonical: boolean;
};

/**
 * Validate the durable Bull payload at the worker boundary. New admissions carry one canonical
 * `fields` map and a submission identity; only pre-migration queue rows may use the three legacy
 * aliases. This is intentionally stricter than a type assertion because queue_payload is durable
 * input and can outlive the route validation that originally created it.
 */
export function normalizeBulkContentItemFields(item: BulkContentItem): NormalizedBulkContentItemFields {
  if (item.fields !== undefined) {
    if (item.title !== undefined || item.excerpt !== undefined || item.content !== undefined) {
      throw new Error('Canonical bulk-content items cannot mix fields with legacy aliases');
    }

    const rawFields: unknown = item.fields;
    if (!rawFields || typeof rawFields !== 'object' || Array.isArray(rawFields)) {
      throw new Error('Bulk-content fields must be an object');
    }
    const entries = Object.entries(rawFields);
    if (entries.length === 0 || entries.length > MAX_STRUCTURED_FIELDS) {
      throw new Error(`Bulk-content fields must contain between 1 and ${MAX_STRUCTURED_FIELDS} entries`);
    }
    for (const [key, value] of entries) {
      if (!STRUCTURED_FIELD_KEY_PATTERN.test(key)) {
        throw new Error(`Invalid structured field key: ${key}`);
      }
      if (typeof value !== 'string') {
        throw new Error(`Structured field value must be a string: ${key}`);
      }
    }

    const fields = canonicalizeStructuredFields(rawFields as StructuredFields);
    if (countStructuredCharacters(fields) === 0) {
      throw new Error('Bulk-content fields cannot be empty');
    }
    return { fields, canonical: true };
  }

  const fields: StructuredFields = {};
  if (item.title !== undefined) fields.title = item.title;
  if (item.excerpt !== undefined) fields.excerpt = item.excerpt;
  if (item.content !== undefined) fields.content = item.content;
  if (Object.keys(fields).length === 0 || countStructuredCharacters(fields) === 0) {
    throw new Error('Bulk-content item must contain at least one non-empty field');
  }
  return { fields, canonical: false };
}

function requireExactStructuredResult(expected: StructuredFields, candidate: unknown): StructuredFields {
  if (!candidate || typeof candidate !== 'object' || Array.isArray(candidate)) {
    throw new Error('Structured translation result must be an object');
  }
  const result = candidate as Record<string, unknown>;
  const expectedKeys = Object.keys(expected).sort();
  const resultKeys = Object.keys(result).sort();
  if (expectedKeys.length !== resultKeys.length || expectedKeys.some((key, index) => key !== resultKeys[index])) {
    throw new Error('Structured translation result keys do not match input fields');
  }
  for (const key of expectedKeys) {
    if (typeof result[key] !== 'string') {
      throw new Error(`Structured translation result value must be a string: ${key}`);
    }
  }
  return result as StructuredFields;
}

async function processBulkContentItem(
  data: BulkContentTranslationJobData,
  item: BulkContentItem,
  exceptionRules: ReturnType<typeof exceptionService.normalizeRules>
): Promise<BulkContentItemOutcome> {
  try {
    const normalized = normalizeBulkContentItemFields(item);
    const fields = normalized.fields;
    const processedFields: StructuredFields = {};
    const replacements = new Map<string, Map<string, string>>();
    for (const [key, value] of Object.entries(fields)) {
      const processed = exceptionService.replaceExceptions(exceptionRules, value);
      processedFields[key] = processed.processedText;
      if (processed.replacements.size > 0) {
        replacements.set(key, processed.replacements);
      }
    }

    const structuredResult = await translateStructuredWithRateLimit(
      processedFields,
      data.sourceLang,
      item.targetLang,
      (data.tone as Tone) || Tone.NEUTRAL,
      data.jobId,
      item.ref
    );
    const outputFields = requireExactStructuredResult(
      fields,
      structuredResult.translatedFields || structuredResult.fields
    );
    for (const [key, fieldReplacements] of replacements.entries()) {
      outputFields[key] = exceptionService.restoreExceptions(outputFields[key], fieldReplacements);
    }

    const split = splitStructuredTranslation(outputFields);
    return {
      result: normalized.canonical
        ? { ref: item.ref, status: 'completed', fields: { ...outputFields } }
        : {
          ref: item.ref,
          status: 'completed',
          translatedTitle: split.translatedTitle,
          translatedExcerpt: split.translatedExcerpt,
          translatedContent: split.translatedContent,
        },
      // This value is only reduced after every worker has settled, so concurrent items cannot
      // race on the customer billing accumulator.
      charactersUsed: countStructuredCharacters(fields),
    };
  } catch (itemError) {
    const itemErrorMessage = boundedWorkerErrorMessage(itemError, 'Translation failed');
    logger.error('Bulk-content item translation failed', {
      jobId: data.jobId,
      ref: item.ref,
      error: itemErrorMessage,
    });
    return {
      result: {
        ref: item.ref,
        status: 'failed',
        error: itemErrorMessage,
      },
      charactersUsed: 0,
    };
  }
}

function validateBulkContentJobData(data: BulkContentTranslationJobData): void {
  if (!Array.isArray(data.items) || data.items.length === 0) {
    throw new Error('Bulk-content job must contain items');
  }

  const hasSubmissionId = typeof data.submissionId === 'string' && data.submissionId.length > 0;
  const hasCanonicalItems = data.items.some((item) => item.fields !== undefined);
  const hasLegacyItems = data.items.some((item) => item.fields === undefined);
  if (hasCanonicalItems && hasLegacyItems) {
    throw new Error('Bulk-content jobs cannot mix canonical fields with legacy aliases');
  }
  if (hasSubmissionId !== hasCanonicalItems) {
    throw new Error('Canonical bulk-content jobs require submissionId and fields for every item');
  }
}

// Run a fixed number of item workers. Promise.all only joins this bounded worker set; it never
// creates one promise per item. The process-wide semaphore remains the final provider-call cap
// when multiple Bull jobs execute at once.
async function processBulkContentItems(
  data: BulkContentTranslationJobData,
  exceptionRules: ReturnType<typeof exceptionService.normalizeRules>
): Promise<BulkContentItemOutcome[]> {
  const outcomes = new Array<BulkContentItemOutcome>(data.items.length);
  let nextItemIndex = 0;

  const runWorker = async (): Promise<void> => {
    while (true) {
      const itemIndex = nextItemIndex;
      nextItemIndex += 1;
      if (itemIndex >= data.items.length) {
        return;
      }
      outcomes[itemIndex] = await processBulkContentItem(data, data.items[itemIndex], exceptionRules);
    }
  };

  const workerCount = Math.min(bulkContentGeminiConcurrency, data.items.length);
  await Promise.all(Array.from({ length: workerCount }, () => runWorker()));
  return outcomes;
}

type TerminalWebhookJobIdentity = {
  jobId: string;
  submissionId?: string;
  clientJobId?: string;
};

function boundedWorkerErrorMessage(error: unknown, fallback: string): string {
  const message = error instanceof Error ? error.message : fallback;
  return message.slice(0, 512) || fallback;
}

function terminalWebhookRefs(
  data: TerminalWebhookJobIdentity,
  refs: readonly string[] | undefined
): string[] {
  const candidateRefs = refs && refs.length > 0
    ? refs
    : [data.clientJobId || data.jobId];
  return [...new Set(candidateRefs.filter((ref): ref is string => typeof ref === 'string' && ref.length > 0))];
}

function bulkContentItemRefs(data: BulkContentTranslationJobData): string[] {
  const refs = Array.isArray(data.items)
    ? data.items
      .map((item) => item && typeof item.ref === 'string' ? item.ref : undefined)
      .filter((ref): ref is string => ref !== undefined)
    : undefined;
  return terminalWebhookRefs(data, refs);
}

/**
 * Persist one terminal callback in the same transaction as the winning job transition. The
 * service helper owns canonical validation, monotonic sequence allocation, and stable delivery
 * identity. The direct branch is only for old queue rows/tests whose payload predates those
 * required identities or whose legacy bulk-string result shape is intentionally different.
 */
async function createTerminalWebhookOutbox(
  tx: unknown,
  data: TerminalWebhookJobIdentity,
  event: string,
  payload: Record<string, unknown>,
  refs?: readonly string[]
): Promise<void> {
  const database = tx as PrismaWithWebhookOutbox;
  const outbox = database.webhookOutbox;
  const submissionId = typeof data.submissionId === 'string' && data.submissionId.length > 0
    ? data.submissionId
    : undefined;
  const clientJobId = typeof data.clientJobId === 'string' && data.clientJobId.length > 0
    ? data.clientJobId
    : undefined;
  const callbackPayload: Record<string, unknown> = {
    ...payload,
    jobId: data.jobId,
    job_id: data.jobId,
    ...(submissionId ? { submission_id: submissionId } : {}),
    ...(clientJobId ? { clientJobId } : {}),
  };
  const normalizedRefs = terminalWebhookRefs(data, refs);

  const helper = webhookService.createWebhookOutbox;
  const helperSupportsShape = typeof helper === 'function'
    && submissionId !== undefined
    && clientJobId !== undefined
    // Legacy bulk-string results use {id, translation, success}; the compatibility adapter
    // keeps their historical event shape while canonical bulk-content uses {ref, fields}.
    && !event.startsWith('bulk_translation.');
  if (helperSupportsShape) {
    await helper(database, {
      jobId: data.jobId,
      event,
      payload: callbackPayload,
      submissionId,
      clientJobId,
      refs: normalizedRefs,
      results: callbackPayload.results,
      failure: callbackPayload.failure,
    });
    return;
  }

  // Compatibility path for pre-manifest jobs and unit fixtures. It remains transactional because
  // callers invoke it before the surrounding Prisma transaction commits.
  let deliverySequence = 1;
  if (typeof outbox.findFirst === 'function') {
    const latest = await outbox.findFirst({
      where: { job_id: data.jobId },
      orderBy: { delivery_sequence: 'desc' },
      select: { delivery_sequence: true },
    });
    const latestSequence = (latest as { delivery_sequence?: unknown } | null)?.delivery_sequence;
    if (typeof latestSequence === 'number' && Number.isSafeInteger(latestSequence) && latestSequence >= 0) {
      deliverySequence = latestSequence + 1;
    }
  }
  const deliveryId = typeof callbackPayload.deliveryId === 'string'
    ? callbackPayload.deliveryId
    : randomUUID();
  callbackPayload.deliveryId = deliveryId;
  callbackPayload.event_sequence = deliverySequence;
  await outbox.create({
    data: {
      id: deliveryId,
      job_id: data.jobId,
      event,
      delivery_sequence: deliverySequence,
      is_final: true,
      payload: callbackPayload,
      status: 'pending',
      attempts: 0,
      next_attempt_at: new Date(),
    },
  });
}

async function settleOutboxEntry(
  outboxPrisma: PrismaWithWebhookOutbox,
  entryId: string,
  claimedAt: Date,
  data: object
): Promise<boolean> {
  const result = await outboxPrisma.webhookOutbox.updateMany({
    where: { id: entryId, status: 'delivering', claimed_at: claimedAt },
    data,
  });
  return result.count === 1;
}

async function runCompletionOutboxDrain(jobId?: string): Promise<OutboxDrainResult> {
  const claimDue = webhookService.claimDueWebhookOutbox;
  const deliverPersisted = webhookService.deliverPersistedWebhook;
  const settlePersisted = webhookService.settleWebhookOutbox;
  if (typeof claimDue === 'function' && typeof deliverPersisted === 'function' && typeof settlePersisted === 'function') {
    const result: OutboxDrainResult = { claimed: 0, delivered: 0, failed: 0, deadLettered: 0 };
    // The documented helper claims due rows globally. A job-specific wakeup may therefore drain
    // a small neighboring batch too; durable due/lease predicates remain the authority and avoid
    // leaving a claimed row behind when the caller's job finishes.
    const entries = await claimDue(prisma, {
      leaseMs: OUTBOX_LEASE_MS,
      maxAttempts: OUTBOX_MAX_ATTEMPTS,
      batchSize: OUTBOX_BATCH_SIZE,
    });
    result.claimed = entries.length;

    for (const entry of entries) {
      const claimedAt = entry.claimedAt || entry.claimed_at || new Date();
      const callbackUrl = entry.job?.callback_url;
      const callbackSecret = entry.job?.callback_secret;
      if (!callbackUrl || !callbackSecret) {
        const settled = await settlePersisted(prisma, {
          entryId: entry.id,
          claimedAt,
          result: {
            success: false,
            status: 'dead',
            errorMessage: 'Webhook callback configuration is missing',
          },
        });
        if (settled) {
          result.failed += 1;
          result.deadLettered += 1;
        }
        continue;
      }

      try {
        const delivery = await deliverPersisted(entry, callbackUrl, callbackSecret, {
          maxAttempts: OUTBOX_MAX_ATTEMPTS,
        });
        const settled = await settlePersisted(prisma, {
          entryId: entry.id,
          claimedAt,
          result: {
            success: delivery.success,
            status: delivery.status,
            errorMessage: delivery.errorMessage,
            httpStatus: delivery.httpStatus,
            nextAttemptAt: delivery.nextAttemptAt,
          },
        });
        if (!settled) {
          continue;
        }
        if (delivery.success) {
          result.delivered += 1;
        } else {
          result.failed += 1;
          if (delivery.status === 'dead') {
            result.deadLettered += 1;
          }
        }
      } catch (error) {
        const attempts = entry.attemptNumber ?? Math.max(1, entry.attempts || 1);
        const dead = attempts >= OUTBOX_MAX_ATTEMPTS;
        const settled = await settlePersisted(prisma, {
          entryId: entry.id,
          claimedAt,
          result: {
            success: false,
            status: dead ? 'dead' : 'retry_wait',
            errorMessage: boundedWorkerErrorMessage(error, 'Webhook delivery failed'),
            nextAttemptAt: dead
              ? undefined
              : new Date(Date.now() + Math.min(OUTBOX_MAX_DELAY_MS, OUTBOX_BASE_DELAY_MS * Math.pow(2, attempts - 1))),
          },
        });
        if (settled) {
          result.failed += 1;
          if (dead) {
            result.deadLettered += 1;
          }
        }
      }
    }
    return result;
  }

  const result: OutboxDrainResult = { claimed: 0, delivered: 0, failed: 0, deadLettered: 0 };
  const outboxPrisma = prisma as PrismaWithWebhookOutbox;
  const leaseCutoff = new Date(Date.now() - OUTBOX_LEASE_MS);
  const jobFilter = jobId ? { job_id: jobId } : {};

  await outboxPrisma.webhookOutbox.updateMany({
    where: { ...jobFilter, status: 'delivering', claimed_at: { lt: leaseCutoff } },
    data: { status: 'pending', claimed_at: null },
  });
  const exhausted = await outboxPrisma.webhookOutbox.updateMany({
    where: { ...jobFilter, status: 'pending', attempts: { gte: OUTBOX_MAX_ATTEMPTS } },
    data: { status: 'dead', last_error: `Webhook delivery exhausted after ${OUTBOX_MAX_ATTEMPTS} attempts` },
  });
  result.deadLettered += exhausted.count;

  const outboxEntries = await outboxPrisma.webhookOutbox.findMany({
    where: {
      ...(jobId ? { job_id: jobId } : {}),
      event: {
        in: [
          'translation.completed',
          'translation.failed',
          'bulk_translation.completed',
          'bulk_translation.failed',
          'bulk_content_translation.completed',
          'bulk_content_translation.failed',
        ],
      },
      status: 'pending',
      attempts: { lt: OUTBOX_MAX_ATTEMPTS },
    },
    orderBy: { created_at: 'asc' },
    take: OUTBOX_BATCH_SIZE,
    include: { job: { select: { callback_url: true, callback_secret: true } } },
  });

  for (const entry of outboxEntries) {
    const claimedAt = new Date();
    const claimed = await outboxPrisma.webhookOutbox.updateMany({
      where: { id: entry.id, status: 'pending', attempts: { lt: OUTBOX_MAX_ATTEMPTS } },
      data: { status: 'delivering', attempts: { increment: 1 }, claimed_at: claimedAt },
    });
    if (claimed.count !== 1) {
      continue;
    }
    result.claimed += 1;
    const attemptNumber = (entry.attempts || 0) + 1;

    if (!entry.job.callback_url || !entry.job.callback_secret) {
      const deadLettered = await settleOutboxEntry(outboxPrisma, entry.id, claimedAt, {
        status: 'dead',
        last_error: 'Webhook callback configuration is missing',
      });
      if (deadLettered) {
        result.deadLettered += 1;
      }
      continue;
    }

    try {
      const delivery = await deliverWebhook(
        entry.job_id,
        entry.payload as unknown as WebhookPayload,
        entry.job.callback_url,
        entry.job.callback_secret,
        entry.id
      );
      if (delivery.success) {
        if (await settleOutboxEntry(outboxPrisma, entry.id, claimedAt, {
          status: 'delivered',
          delivered_at: new Date(),
          claimed_at: null,
          last_error: null,
        })) {
          result.delivered += 1;
        }
      } else {
        const status = attemptNumber >= OUTBOX_MAX_ATTEMPTS ? 'dead' : 'pending';
        if (await settleOutboxEntry(outboxPrisma, entry.id, claimedAt, {
          status,
          claimed_at: null,
          last_error: delivery.errorMessage || 'Webhook delivery failed',
        })) {
          result.failed += 1;
          if (status === 'dead') {
            result.deadLettered += 1;
          }
        }
      }
    } catch (error) {
      const status = attemptNumber >= OUTBOX_MAX_ATTEMPTS ? 'dead' : 'pending';
      if (await settleOutboxEntry(outboxPrisma, entry.id, claimedAt, {
        status,
        claimed_at: null,
        last_error: error instanceof Error ? error.message : 'Webhook delivery failed',
      })) {
        result.failed += 1;
        if (status === 'dead') {
          result.deadLettered += 1;
        }
      }
    }
  }

  return result;
}

export async function drainCompletionOutbox(jobId?: string): Promise<OutboxDrainResult> {
  if (!outboxDrainPromise) {
    outboxDrainPromise = runCompletionOutboxDrain(jobId).catch((error) => {
      logger.error('Completion webhook outbox drain failed', { error, jobId });
      return { claimed: 0, delivered: 0, failed: 1, deadLettered: 0 };
    });
  }
  try {
    return await outboxDrainPromise;
  } finally {
    outboxDrainPromise = undefined;
  }
}

function scheduleCompletionOutboxDrain(delayMs: number): void {
  if (!outboxStarted || outboxStopRequested) {
    return;
  }
  outboxTimer = setTimeout(() => {
    outboxTimer = undefined;
    void drainCompletionOutbox().then((result) => {
      outboxBackoffMs = result.failed > 0
        ? Math.min(outboxBackoffMs * 2, OUTBOX_MAX_DELAY_MS)
        : OUTBOX_BASE_DELAY_MS;
    }).catch((error) => {
      logger.error('Completion webhook outbox recurring drain failed', { error });
      outboxBackoffMs = Math.min(outboxBackoffMs * 2, OUTBOX_MAX_DELAY_MS);
    }).finally(() => scheduleCompletionOutboxDrain(outboxBackoffMs));
  }, delayMs);
  outboxTimer.unref?.();
}

export async function startCompletionOutboxDrainer(): Promise<void> {
  if (outboxStarted) {
    if (outboxStartPromise) {
      await outboxStartPromise;
    }
    return;
  }
  outboxStarted = true;
  outboxStopRequested = false;
  outboxBackoffMs = OUTBOX_BASE_DELAY_MS;
  outboxStartPromise = (async () => {
    const result = await drainCompletionOutbox();
    outboxBackoffMs = result.failed > 0
      ? Math.min(OUTBOX_BASE_DELAY_MS * 2, OUTBOX_MAX_DELAY_MS)
      : OUTBOX_BASE_DELAY_MS;
    scheduleCompletionOutboxDrain(outboxBackoffMs);
  })();
  try {
    await outboxStartPromise;
  } finally {
    outboxStartPromise = undefined;
  }
}

export async function stopCompletionOutboxDrainer(): Promise<void> {
  outboxStopRequested = true;
  outboxStarted = false;
  if (outboxTimer) {
    clearTimeout(outboxTimer);
    outboxTimer = undefined;
  }
  await outboxStartPromise;
  await outboxDrainPromise;
}

export function registerProcessors(): void {

/**
 * Process translation job
 */
translationQueue.process('translate', jobConcurrency, async (job) => {
  const startTime = Date.now();
  const data: TranslationJobData = job.data;
  const exceptionRules = exceptionService.normalizeRules(data.exceptions);

  logger.info('Processing translation job', { jobId: data.jobId, clientJobId: data.clientJobId });

  const claim = await claimTranslationJob(data.jobId);
  if (!claim.claimed) {
    const current = await prisma.translationJob.findUnique({
      where: { id: data.jobId },
      select: { status: true, translation: true },
    });
    if (current?.status === 'completed' || current?.status === 'failed') {
      await drainCompletionOutbox(data.jobId);
      if (current.status === 'completed') {
        return { translation: current.translation || undefined };
      }
      return { status: current.status };
    }
    return { status: current?.status || 'cancelled' };
  }
  const stopLeaseHeartbeat = startProcessingLeaseHeartbeat(data.jobId, claim.leaseId);

  try {

    let storedTranslation: string;
    let translatedFields: StructuredFields | undefined;
    let translatedOutputFields: StructuredFields | undefined;
    let tokensUsed: number;
    let inputTokens: number;
    let outputTokens: number;
    let modelUsed: string;

    if (data.siteContent) {
      const siteContentResult = await translateSiteContentResource(
        data.siteContent,
        async (fields) => {
          const processedFields: StructuredFields = {};
          const replacements = new Map<string, Map<string, string>>();
          for (const [key, value] of Object.entries(fields)) {
            const processed = exceptionService.replaceExceptions(exceptionRules, value);
            processedFields[key] = processed.processedText;
            if (processed.replacements.size > 0) {
              replacements.set(key, processed.replacements);
            }
          }

          const structuredResult = await geminiClient.translateStructured(
            processedFields,
            data.sourceLang,
            data.targetLang,
            data.tone as Tone
          );
          const outputFields = structuredResult.translatedFields || structuredResult.fields;
          for (const [key, fieldReplacements] of replacements.entries()) {
            const translated = outputFields[key];
            if (translated !== undefined) {
              outputFields[key] = exceptionService.restoreExceptions(translated, fieldReplacements);
            }
          }

          return {
            translatedFields: outputFields,
            tokens_used: structuredResult.tokens_used,
            input_tokens: structuredResult.input_tokens,
            output_tokens: structuredResult.output_tokens,
            processing_time_ms: structuredResult.processing_time_ms,
            model_used: structuredResult.model_used,
          };
        }
      );
      storedTranslation = siteContentResult.translation;
      tokensUsed = siteContentResult.tokens_used;
      inputTokens = siteContentResult.input_tokens;
      outputTokens = siteContentResult.output_tokens;
      modelUsed = siteContentResult.model_used;
    } else if (data.fields) {
      const processedFields: StructuredFields = {};
      const replacements = new Map<string, Map<string, string>>();

      for (const [key, value] of Object.entries(data.fields)) {
        const processed = exceptionService.replaceExceptions(exceptionRules, value);
        processedFields[key] = processed.processedText;
        if (processed.replacements.size > 0) {
          replacements.set(key, processed.replacements);
        }
      }

      const structuredResult = await geminiClient.translateStructured(
        processedFields,
        data.sourceLang,
        data.targetLang,
        data.tone as Tone
      );
      translatedOutputFields = structuredResult.translatedFields || structuredResult.fields;

      for (const [key, fieldReplacements] of replacements.entries()) {
        const translated = translatedOutputFields[key];
        if (translated !== undefined) {
          translatedOutputFields[key] = exceptionService.restoreExceptions(translated, fieldReplacements);
        }
      }

      translatedFields = splitStructuredTranslation(translatedOutputFields).translatedFields;
      storedTranslation = serializeStructuredFields(translatedOutputFields);
      tokensUsed = structuredResult.tokens_used;
      inputTokens = structuredResult.input_tokens;
      outputTokens = structuredResult.output_tokens;
      modelUsed = structuredResult.model_used;
    } else {
      const content = data.content || '';
      const { processedText, replacements, matchedExceptions } =
        exceptionService.replaceExceptions(exceptionRules, content);

      const result = await geminiClient.translate(
        processedText,
        data.sourceLang,
        data.targetLang,
        data.tone as Tone
      );

      storedTranslation = result.translation;
      if (replacements.size > 0) {
        storedTranslation = exceptionService.restoreExceptions(storedTranslation, replacements);

        if (matchedExceptions.length > 0) {
          const verification = verifyExceptions(storedTranslation, matchedExceptions);
          if (!verification.passed) {
            logger.warn('Worker: Exception verification warnings', {
              jobId: data.jobId,
              violations: verification.violations,
            });
          }
        }
      }

      tokensUsed = result.tokens_used;
      inputTokens = result.input_tokens;
      outputTokens = result.output_tokens;
      modelUsed = result.model_used;
    }

    const processingTime = Date.now() - startTime;
    const charactersUsed = data.siteContent
      ? countSiteContentCharacters(data.siteContent)
      : data.fields
        ? countStructuredCharacters(data.fields)
        : countSourceCharacters(data.content || '');
    const internalCost = calculateCost(inputTokens, outputTokens, modelUsed);

    // Look up user's subscription to get customer cost per character
    const subscription = await prisma.subscription.findFirst({
      where: { user_id: data.userId },
      select: { customer_cost_per_char: true },
    });
    const costPerChar = subscription?.customer_cost_per_char
      ? Number(subscription.customer_cost_per_char)
      : 0;
    const customerCost = calculateCustomerCostByChars(charactersUsed, costPerChar);

    const completionPayload = {
        event: 'translation.completed' as const,
        jobId: data.jobId,
        clientJobId: data.clientJobId,
        status: 'completed' as TranslationStatus,
        ...(data.siteContent ? siteContentCallbackMetadata(data.siteContent) : {}),
        ...mapStoredTranslation(storedTranslation),
        charactersUsed,
        cost: customerCost,
        processingTimeMs: processingTime,
        timestamp: new Date().toISOString(),
      };

    await prisma.$transaction(async (tx) => {
      await lockProcessingJob(tx, data.jobId, claim.leaseId);

      const user = await tx.user.findUnique({
        where: { id: data.userId },
        select: { id: true },
      });

      if (!user) {
        throw new Error('User not found');
      }

      const latestTransaction = await tx.creditTransaction.findFirst({
        where: { user_id: data.userId },
        orderBy: { created_at: 'desc' },
        select: { balance_after: true },
      });

      const currentBalance = latestTransaction?.balance_after ?? 0;
      const newBalance = currentBalance - charactersUsed;

      await tx.creditTransaction.create({
        data: {
          user_id: data.userId,
          type: 'deduction',
          amount: -charactersUsed,
          balance_after: newBalance,
          description: `Translation job ${data.jobId}: ${data.sourceLang} → ${data.targetLang}`,
          related_job_id: data.jobId,
        },
      });

      await tx.user.update({
        where: { id: data.userId },
        data: {
          updated_at: new Date(),
        },
      });

      await tx.translationJob.update({
        where: { id: data.jobId },
        data: {
          status: 'completed' as TranslationStatus,
          translation: storedTranslation,
          model: modelUsed,
          characters_used: charactersUsed,
          tokens_used: tokensUsed,
          input_tokens: inputTokens,
          output_tokens: outputTokens,
          cost: internalCost,
          customer_cost: customerCost,
          processing_time_ms: processingTime,
          completed_at: new Date(),
          processing_lease_id: null,
          processing_lease_expires_at: null,
        },
      });

      await createTerminalWebhookOutbox(
        tx,
        data,
        'translation.completed',
        completionPayload,
        data.clientJobId ? [data.clientJobId] : undefined
      );
    });

    // Track metrics
    trackTranslationJob('gemini', 'completed', 'async', processingTime);
    trackTokensProcessed('gemini', tokensUsed);

    await drainCompletionOutbox(data.jobId);

    logger.info('Translation job completed', {
      jobId: data.jobId,
      characters_used: charactersUsed,
      tokens_used: tokensUsed,
      cost: internalCost,
      processingTime,
    });

    return {
      translation: storedTranslation,
      translatedFields,
      tokensUsed,
      processingTimeMs: processingTime,
    };
  } catch (error: any) {
    const processingTime = Date.now() - startTime;

    if (error instanceof JobCancellationWonError) {
      return { status: 'cancelled' };
    }

    const publicErrorMessage = data.siteContent
      ? 'Site Content translation failed.'
      : error instanceof Error
        ? error.message
        : 'Translation failed';
    const outwardError = data.siteContent ? new Error(publicErrorMessage) : error;
    logger.error('Translation job failed', {
      jobId: data.jobId,
      error: data.siteContent ? publicErrorMessage : error,
    });

    const isFinalAttempt = (job.opts?.attempts ?? 1) <= job.attemptsMade + 1;
    if (!isFinalAttempt && !isTerminalWorkerError(error)) {
      await releaseProcessingLease(data.jobId, claim.leaseId);
      throw outwardError;
    }

    let failed = false;
    try {
      await prisma.$transaction(async (tx) => {
        await lockProcessingJob(tx, data.jobId, claim.leaseId);
        const transition = await tx.translationJob.updateMany({
          where: { id: data.jobId, status: 'processing', processing_lease_id: claim.leaseId },
          data: {
            status: 'failed' as TranslationStatus,
            processing_lease_id: null,
            processing_lease_expires_at: null,
            error_message: publicErrorMessage,
            processing_time_ms: processingTime,
            completed_at: new Date(),
          },
        });
        if (transition.count !== 1) {
          return;
        }
        failed = true;
        await createTerminalWebhookOutbox(
          tx,
          data,
          'translation.failed',
          {
            event: 'translation.failed',
            status: 'failed' as TranslationStatus,
            ...(data.siteContent ? siteContentCallbackMetadata(data.siteContent) : {}),
            errorMessage: publicErrorMessage,
            failure: { code: 'translation_failed', message: publicErrorMessage },
            processingTimeMs: processingTime,
            timestamp: new Date().toISOString(),
          },
          data.clientJobId ? [data.clientJobId] : undefined
        );
      });
    } catch (transitionError) {
      if (transitionError instanceof JobCancellationWonError) {
        const current = await prisma.translationJob.findUnique({
          where: { id: data.jobId },
          select: { status: true, translation: true },
        });
        if (current?.status === 'cancelled' || current?.status === 'completed') {
          return { status: current.status, translation: current.translation || undefined };
        }
      }
      throw transitionError;
    }
    if (!failed) {
      const current = await prisma.translationJob.findUnique({
        where: { id: data.jobId },
        select: { status: true, translation: true },
      });
      if (current?.status === 'cancelled' || current?.status === 'completed' || current?.status === 'failed') {
        if (current.status === 'completed' || current.status === 'failed') {
          await drainCompletionOutbox(data.jobId);
        }
        if (current.status === 'completed') {
          return { status: current.status, translation: current.translation || undefined };
        }
        return { status: current.status };
      }
      throw outwardError;
    }

    await drainCompletionOutbox(data.jobId);
    // Track metrics
    trackTranslationJob('gemini', 'failed', 'async', processingTime);

    throw outwardError;
  } finally {
    stopLeaseHeartbeat();
  }
});

/**
 * Process async bulk-strings translation job.
 * Translates all strings for each target language in budget-packed Gemini calls,
 * deducts credits once for the entire job, then delivers one webhook with full results.
 */
translationQueue.process('bulk-strings', jobConcurrency, async (job) => {
  const startTime = Date.now();
  const data: BulkTranslationJobData = job.data;
  const exceptionRules = exceptionService.normalizeRules(data.exceptions);

  logger.info('Processing bulk-strings job', {
    jobId: data.jobId,
    stringCount: data.strings.length,
    targetLangs: data.targetLangs,
  });

  const claim = await claimTranslationJob(data.jobId);
  if (!claim.claimed) {
    const current = await prisma.translationJob.findUnique({
      where: { id: data.jobId },
      select: { status: true, translation: true },
    });
    if (current?.status === 'completed' || current?.status === 'failed') {
      await drainCompletionOutbox(data.jobId);
      if (current.status === 'completed') {
        return { translation: current.translation || undefined };
      }
      return { status: current.status };
    }
    return { status: current?.status || 'cancelled' };
  }
  const stopLeaseHeartbeat = startProcessingLeaseHeartbeat(data.jobId, claim.leaseId);

  const resultsByLang: Record<string, Array<{ id: string; translation: string; success: boolean }>> = {};
  let totalCharactersUsed = 0;
  let totalFailedCount = 0;

  try {
  // Process each target language in series
  for (const targetLang of data.targetLangs) {
    const langResults: Array<{ id: string; translation: string; success: boolean }> = [];
    const langChars = data.strings.reduce((sum, s) => sum + countSourceCharacters(s.content), 0);
    totalCharactersUsed += langChars;

    // Pre-process: replace exceptions in each string for this language
    const perStringReplacements: Map<string, Map<string, string>> = new Map();
    const processedStrings = [];

    for (const s of data.strings) {
      const { processedText, replacements } =
        exceptionService.replaceExceptions(exceptionRules, s.content);
      processedStrings.push({ id: s.id, content: processedText });
      if (replacements.size > 0) {
        perStringReplacements.set(s.id, replacements);
      }
    }

    // Pack strings by the Gemini output-token budget, count cap, and payload cap.
    const batches = planBatches(processedStrings, {
      maxChars: GEMINI_CALL_CHAR_BUDGET,
      maxCount: GEMINI_CALL_MAX_STRINGS,
      maxPayloadBytes: GEMINI_CALL_MAX_PAYLOAD_BYTES,
    });
    for (const batch of batches) {
      const bulkResponse = await geminiClient.translateBulk(
        batch,
        data.sourceLang,
        targetLang,
        (data.tone as Tone) || Tone.NEUTRAL
      );

      // Post-process: restore exceptions in each result
      for (const result of bulkResponse.results) {
        const reps = perStringReplacements.get(result.id);
        if (reps && result.success && result.translation) {
          result.translation = exceptionService.restoreExceptions(result.translation, reps);
        }
      }

      langResults.push(...bulkResponse.results);
    }

    totalFailedCount += langResults.filter(r => !r.success).length;
    resultsByLang[targetLang] = langResults;
  }

  const processingTime = Date.now() - startTime;
  const storedTranslation = JSON.stringify(resultsByLang);
  const completionPayload = {
    event: 'bulk_translation.completed' as const,
    jobId: data.jobId,
    clientJobId: data.clientJobId,
    status: 'completed' as TranslationStatus,
    translation: storedTranslation,
    job_id: data.jobId,
    results_by_lang: resultsByLang,
    total_characters_used: totalCharactersUsed,
    failed_count: totalFailedCount,
    timestamp: new Date().toISOString(),
  };

  try {
    await prisma.$transaction(async (tx) => {
      await lockProcessingJob(tx, data.jobId, claim.leaseId);
      const latestTransaction = await tx.creditTransaction.findFirst({
        where: { user_id: data.userId },
        orderBy: { created_at: 'desc' },
        select: { balance_after: true },
      });
      const currentBalance = latestTransaction?.balance_after ?? 0;
      await tx.creditTransaction.create({
        data: {
          user_id: data.userId,
          type: 'deduction',
          amount: -totalCharactersUsed,
          balance_after: currentBalance - totalCharactersUsed,
          description: `Bulk-strings job ${data.jobId}: ${data.sourceLang} → ${data.targetLangs.join(', ')}`,
          related_job_id: data.jobId,
        },
      });
      await tx.user.update({ where: { id: data.userId }, data: { updated_at: new Date() } });
      await tx.translationJob.update({
        where: { id: data.jobId },
        data: {
          status: 'completed' as TranslationStatus,
          translation: storedTranslation,
          characters_used: totalCharactersUsed,
          processing_time_ms: processingTime,
          completed_at: new Date(),
          processing_lease_id: null,
          processing_lease_expires_at: null,
        },
      });
      await createTerminalWebhookOutbox(
        tx,
        data,
        'bulk_translation.completed',
        completionPayload,
        data.strings.map((stringItem) => stringItem.id)
      );
    });
  } catch (error) {
    if (error instanceof JobCancellationWonError) {
      return { status: 'cancelled' };
    }
    throw error;
  }

  await drainCompletionOutbox(data.jobId);

  logger.info('Bulk-strings job completed', {
    jobId: data.jobId,
    totalCharactersUsed,
    totalFailedCount,
    processingTimeMs: processingTime,
  });

  return {
    translation: storedTranslation,
    totalCharactersUsed,
    totalFailedCount,
    processingTimeMs: processingTime,
  };
  } catch (error) {
    if (error instanceof JobCancellationWonError) {
      return { status: 'cancelled' };
    }
    const isFinalAttempt = (job.opts?.attempts ?? 1) <= job.attemptsMade + 1;
    if (!isFinalAttempt && !isTerminalWorkerError(error)) {
      await releaseProcessingLease(data.jobId, claim.leaseId);
      throw error;
    }
    const failureMessage = boundedWorkerErrorMessage(error, 'Bulk translation failed');
    const failureIdentity = data as unknown as TerminalWebhookJobIdentity;
    let failed = false;
    try {
      await prisma.$transaction(async (tx) => {
        await lockProcessingJob(tx, data.jobId, claim.leaseId);
        const transition = await tx.translationJob.updateMany({
          where: { id: data.jobId, status: 'processing', processing_lease_id: claim.leaseId },
          data: {
            status: 'failed' as TranslationStatus,
            processing_lease_id: null,
            processing_lease_expires_at: null,
            error_message: failureMessage,
            processing_time_ms: Date.now() - startTime,
            completed_at: new Date(),
          },
        });
        if (transition.count !== 1) {
          return;
        }
        failed = true;
        await createTerminalWebhookOutbox(
          tx,
          failureIdentity,
          'bulk_translation.failed',
          {
            event: 'bulk_translation.failed',
            status: 'failed' as TranslationStatus,
            errorMessage: failureMessage,
            failure: { code: 'translation_failed', message: failureMessage },
            refs: data.strings.map((stringItem) => stringItem.id),
            timestamp: new Date().toISOString(),
          },
          data.strings.map((stringItem) => stringItem.id)
        );
      });
    } catch (transitionError) {
      if (transitionError instanceof JobCancellationWonError) {
        const current = await prisma.translationJob.findUnique({
          where: { id: data.jobId },
          select: { status: true, translation: true },
        });
        if (current?.status === 'cancelled' || current?.status === 'completed') {
          return { status: current.status, translation: current.translation || undefined };
        }
      }
      throw transitionError;
    }
    if (!failed) {
      const current = await prisma.translationJob.findUnique({
        where: { id: data.jobId },
        select: { status: true, translation: true },
      });
      if (current?.status === 'cancelled' || current?.status === 'completed' || current?.status === 'failed') {
        if (current.status === 'completed' || current.status === 'failed') {
          await drainCompletionOutbox(data.jobId);
        }
        return current.status === 'completed'
          ? { status: current.status, translation: current.translation || undefined }
          : { status: current.status };
      }
    }
    await drainCompletionOutbox(data.jobId);
    throw error;
  } finally {
    stopLeaseHeartbeat();
  }
});

/**
 * Process async bulk-content translation job.
 * Each item is a (post, language) pair; translated with one Gemini structured call per
 * item (preserving exact single-content translation behavior). A single item's failure
 * does not abort the batch — it is recorded as a failed result. Credits are deducted once
 * for the batch, billing only successfully translated items, then one webhook is delivered
 * with full per-item results.
 */
translationQueue.process('bulk-content', jobConcurrency, async (job) => {
  const startTime = Date.now();
  const data: BulkContentTranslationJobData = job.data;
  const exceptionRules = exceptionService.normalizeRules(undefined);

  logger.info('Processing bulk-content job', {
    jobId: data.jobId,
    itemCount: Array.isArray(data.items) ? data.items.length : 0,
  });

  const claim = await claimTranslationJob(data.jobId);
  if (!claim.claimed) {
    const current = await prisma.translationJob.findUnique({
      where: { id: data.jobId },
      select: { status: true, translation: true },
    });
    if (current?.status === 'completed' || current?.status === 'failed') {
      await drainCompletionOutbox(data.jobId);
      return current.status === 'completed'
        ? { translation: current.translation || undefined }
        : { status: 'failed' };
    }
    return { status: current?.status || 'cancelled' };
  }
  const stopLeaseHeartbeat = startProcessingLeaseHeartbeat(data.jobId, claim.leaseId);

  try {
    validateBulkContentJobData(data);
    const outcomes = await processBulkContentItems(data, exceptionRules);
    // Map by index after every worker settles to preserve the input order. Totals are reduced
    // here, in one synchronous pass, rather than mutated by concurrent item workers.
    const results = outcomes.map((outcome) => outcome.result);
    let totalCharactersUsed = 0;
    let totalFailedCount = 0;
    for (const outcome of outcomes) {
      if (outcome.result.status === 'completed') {
        totalCharactersUsed += outcome.charactersUsed;
      } else {
        totalFailedCount += 1;
      }
    }

    const processingTime = Date.now() - startTime;
    const storedTranslation = JSON.stringify(results);
    const completionPayload = {
      event: 'bulk_content_translation.completed' as const,
      jobId: data.jobId,
      clientJobId: data.clientJobId,
      status: 'completed' as TranslationStatus,
      translation: storedTranslation,
      job_id: data.jobId,
      results,
      total_characters_used: totalCharactersUsed,
      failed_count: totalFailedCount,
      timestamp: new Date().toISOString(),
    };

    try {
      await prisma.$transaction(async (tx) => {
        await lockProcessingJob(tx, data.jobId, claim.leaseId);
        const latestTransaction = await tx.creditTransaction.findFirst({
          where: { user_id: data.userId },
          orderBy: { created_at: 'desc' },
          select: { balance_after: true },
        });
        const currentBalance = latestTransaction?.balance_after ?? 0;
        await tx.creditTransaction.create({
          data: {
            user_id: data.userId,
            type: 'deduction',
            amount: -totalCharactersUsed,
            balance_after: currentBalance - totalCharactersUsed,
            description: `Bulk-content job ${data.jobId}: ${data.sourceLang} → ${data.items.length} item(s)`,
            related_job_id: data.jobId,
          },
        });
        await tx.user.update({ where: { id: data.userId }, data: { updated_at: new Date() } });
        await tx.translationJob.update({
          where: { id: data.jobId },
          data: {
            status: 'completed' as TranslationStatus,
            translation: storedTranslation,
            characters_used: totalCharactersUsed,
            processing_time_ms: processingTime,
            completed_at: new Date(),
            processing_lease_id: null,
            processing_lease_expires_at: null,
          },
        });
        await createTerminalWebhookOutbox(
          tx,
          data,
          'bulk_content_translation.completed',
          completionPayload,
          data.items.map((item) => item.ref)
        );
      });
    } catch (error) {
      if (error instanceof JobCancellationWonError) {
        return { status: 'cancelled' };
      }
      throw error;
    }

    await drainCompletionOutbox(data.jobId);

    logger.info('Bulk-content job completed', {
      jobId: data.jobId,
      totalCharactersUsed,
      totalFailedCount,
      processingTimeMs: processingTime,
    });

    return {
      translation: storedTranslation,
      totalCharactersUsed,
      totalFailedCount,
      processingTimeMs: processingTime,
    };
  } catch (error) {
    if (error instanceof JobCancellationWonError) {
      return { status: 'cancelled' };
    }
    const isFinalAttempt = (job.opts?.attempts ?? 1) <= job.attemptsMade + 1;
    if (!isFinalAttempt && !isTerminalWorkerError(error)) {
      await releaseProcessingLease(data.jobId, claim.leaseId);
      throw error;
    }
    const failureMessage = boundedWorkerErrorMessage(error, 'Bulk content translation failed');
    let failed = false;
    try {
      await prisma.$transaction(async (tx) => {
        await lockProcessingJob(tx, data.jobId, claim.leaseId);
        const transition = await tx.translationJob.updateMany({
          where: { id: data.jobId, status: 'processing', processing_lease_id: claim.leaseId },
          data: {
            status: 'failed' as TranslationStatus,
            processing_lease_id: null,
            processing_lease_expires_at: null,
            error_message: failureMessage,
            processing_time_ms: Date.now() - startTime,
            completed_at: new Date(),
          },
        });
        if (transition.count !== 1) {
          return;
        }
        failed = true;
        await createTerminalWebhookOutbox(
          tx,
          data,
          'bulk_content_translation.failed',
          {
            event: 'bulk_content_translation.failed',
            status: 'failed' as TranslationStatus,
            errorMessage: failureMessage,
            failure: { code: 'translation_failed', message: failureMessage },
            refs: bulkContentItemRefs(data),
            timestamp: new Date().toISOString(),
          },
          bulkContentItemRefs(data)
        );
      });
    } catch (transitionError) {
      if (transitionError instanceof JobCancellationWonError) {
        const current = await prisma.translationJob.findUnique({
          where: { id: data.jobId },
          select: { status: true, translation: true },
        });
        if (current?.status === 'cancelled' || current?.status === 'completed') {
          return { status: current.status, translation: current.translation || undefined };
        }
      }
      throw transitionError;
    }
    if (!failed) {
      const current = await prisma.translationJob.findUnique({
        where: { id: data.jobId },
        select: { status: true, translation: true },
      });
      if (current?.status === 'cancelled' || current?.status === 'completed' || current?.status === 'failed') {
        if (current.status === 'completed' || current.status === 'failed') {
          await drainCompletionOutbox(data.jobId);
        }
        return current.status === 'completed'
          ? { status: current.status, translation: current.translation || undefined }
          : { status: current.status };
      }
    }
    await drainCompletionOutbox(data.jobId);
    throw error;
  } finally {
    stopLeaseHeartbeat();
  }
});
}

if (process.env.NODE_ENV !== 'test') {
  registerProcessors();
}

// ---------------------------------------------------------------------------
// Orphan Dispute Reconciler — runs every hour (at :17 to avoid stampede
// with other hourly crons). The named jobId makes Bull dedupe this repeatable
// across worker restarts: re-adding the same cron+jobId is idempotent.
// ---------------------------------------------------------------------------
translationQueue.process('reconcile-orphan-disputes', async () => {
  const summary = await reconcilePermanentOrphanDisputes();
  logger.info('Orphan dispute reconciliation pass complete', summary);
  return summary;
});

translationQueue
  .add(
    'reconcile-orphan-disputes',
    {},
    {
      repeat: { cron: '17 * * * *' },
      jobId: 'reconcile-orphan-disputes-cron',
    }
  )
  .catch((err) => logger.error('Failed to schedule orphan dispute reconciler', { error: err }));

// Queue event handlers
translationQueue.on('completed', (job) => {
  logger.info('Job completed', { jobId: job.id });
});

translationQueue.on('failed', (job, error) => {
  logger.error('Job failed', { jobId: job?.id, error: error.message });
});

translationQueue.on('error', (error) => {
  logger.error('Queue error', { error: error.message });
});

// Graceful shutdown
let shutdownStarted = false;

async function gracefulShutdown(signal: string) {
  if (shutdownStarted) {
    return;
  }
  shutdownStarted = true;
  logger.info(`Received ${signal}, shutting down worker...`);

  try {
    await shutdownWorkerRuntime({
      closeQueue: () => translationQueue.close(),
      stopHeartbeat: () => workerHeartbeat.stop(),
      closeConfigSubscription: () => configSubscription?.close() ?? Promise.resolve(),
      disconnectDatabase: () => prisma.$disconnect(),
      stopOutboxDrainer: async () => {
        await stopBulkQueueDispatcher();
        await stopCompletionOutboxDrainer();
      },
    });
    logger.info('Worker shut down successfully');
    process.exit(0);
  } catch (error) {
    logger.error('Worker shutdown completed with errors', { error });
    process.exit(1);
  }
}

if (process.env.NODE_ENV !== 'test') {
  process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
  process.on('SIGINT', () => gracefulShutdown('SIGINT'));

  startWorkerRuntime({
    initializeConfig: initializeConfigFromDatabase,
    waitForQueue: () => translationQueue.isReady(),
    startOutboxDrainer: async () => {
      await startCompletionOutboxDrainer();
      await startBulkQueueDispatcher();
    },
    startHeartbeat: () => {
      configSubscription = startConfigSubscription({
        workerId,
        categories: ['gemini', 'pricing', 'credits', 'rateLimits'],
      });
      workerHeartbeat.start(5000, (error) => {
        logger.error('Worker heartbeat failed', { error });
      });
    },
    onConfigFallback: (error) => {
      useEnvironmentConfigFallback();
      logger.error('Worker config init from database failed, using explicit .env fallback', { error });
    },
  })
    .then(() => {
      logger.info('Translation worker started', {
        redis: getRedisUrl().replace(/:[^:]*@/, ':***@'),
        workerId,
      });
    })
    .catch((error) => {
      logger.error('Translation worker startup failed', { error });
      process.exit(1);
    });
}

export { translationQueue };