/**
 * Cloudflare Queue consumer for `multideal-translation-jobs-preview`.
 *
 * Each message carries `{ jobId: string }`. The consumer:
 *   1. Checks the TRANSLATION_ENABLED feature flag.
 *   2. Calls verifyConfiguredModelExists to ensure the model is reachable.
 *   3. Atomically claims the job via a single UPDATE…RETURNING.
 *   4. Dispatches translateDealFields, writes DONE on success.
 *   5. Handles BudgetExhaustedError → PENDING_BUDGET.
 *   6. Handles TranslationProviderError with per-kind policy (retry / fail).
 *
 * DEVIATION NOTE — retry_max:
 *   The translation_jobs table has no retry_max column (schema confirmed).
 *   We use the module constant TRANSLATION_RETRY_MAX = 3. A future migration
 *   can add the column if per-job overrides are required.
 *
 * REQUEUE STRATEGY:
 *   max_retries = 0 in wrangler.toml — we manage retries ourselves via
 *   env.TRANSLATION_QUEUE.send({ jobId }, { delaySeconds }) so we get
 *   true exponential backoff (1m / 5m; 2 retries then terminal-fail) without
 *   relying on Cloudflare retry intervals. The job row's `attempt` column is the
 *   durable retry counter. Raising the retry budget to reach a 3rd (30m) backoff
 *   requires bumping TRANSLATION_RETRY_MAX — a deliberate product decision.
 */

import { z } from 'zod';
import { sql } from 'drizzle-orm';
import { getDb } from '../lib/db.js';
import type { DoDbClient } from '../lib/db.js';
import type { MultidealEnv } from '../lib/env.js';
import { TranslationEnabledSchema } from '../lib/env.js';
import { verifyConfiguredModelExists } from '@/server/translation/healthcheck.js';
import { translateDealFields } from '@/server/translation/fields/deal.js';
import { afterTranslate } from '@/server/translation/state.js';
import { failJob } from '@/server/translation/jobs/deadletter.js';
import { recordDeadLetter } from '@/server/monitor/dead-letter.js';
import {
  BudgetExhaustedError,
  TranslationProviderError,
} from '@/server/translation/provider/types.js';
import { TRANSLATION_RETRY_MAX } from '@/server/translation/jobs/constants.js';
import { scrubErrorForLog } from '@/server/observability/pii-scrub.js';

// ─── Constants ────────────────────────────────────────────────────────────────

/** Exponential backoff delays by post-claim attempt (reachable: 1 → 1m, 2 → 5m). */
const BACKOFF_SECONDS: Record<number, number> = {
  1: 60, // 1m  (after 1st failure)
  2: 300, // 5m  (after 2nd failure; attempt 3 hits TRANSLATION_RETRY_MAX → fail)
};

// ─── Zod validation ───────────────────────────────────────────────────────────

const TranslationJobMessageSchema = z.object({
  jobId: z.uuid(),
});

// ─── Consumer ─────────────────────────────────────────────────────────────────

export async function handleTranslationBatch(
  batch: MessageBatch<{ jobId: string }>,
  env: MultidealEnv,
  _ctx: ExecutionContext,
): Promise<void> {
  // ── Step 1: Feature flag check ──────────────────────────────────────────────
  const translationEnabled = TranslationEnabledSchema.parse(env.TRANSLATION_ENABLED);
  if (!translationEnabled) {
    console.warn(
      JSON.stringify({
        event: 'translation_consumer_noop',
        reason: 'flag_off',
        batchSize: batch.messages.length,
      }),
    );
    batch.ackAll();
    return;
  }

  const db: DoDbClient = getDb(env);

  // ── Step 2: Healthcheck — verify model is reachable ─────────────────────────
  const apiKey = env.GOOGLE_API_KEY ?? '';
  const healthcheck = await verifyConfiguredModelExists(db, apiKey);
  if (!healthcheck.ok) {
    // Ack all messages (they stay in DB as PENDING — cron will republish when healthy)
    // healthcheck.ts already writes the TRANSLATION_HEALTH_DEGRADED outbox alert.
    console.error(
      JSON.stringify({
        event: 'translation_consumer_healthcheck_failed',
        reason: healthcheck.reason,
        modelId: healthcheck.modelId,
        batchSize: batch.messages.length,
      }),
    );
    batch.ackAll();
    return;
  }

  // ── Per-message processing ──────────────────────────────────────────────────
  for (const message of batch.messages) {
    // Validate message body
    const parsed = TranslationJobMessageSchema.safeParse(message.body);
    if (!parsed.success) {
      console.error(
        JSON.stringify({
          event: 'translation_job_invalid_message',
          error: parsed.error.message,
          body: message.body,
        }),
      );
      message.ack();
      continue;
    }

    const { jobId } = parsed.data;
    // A single message's unexpected error (e.g. transient Neon error on the atomic
    // claim, under wrangler max_retries = 0) must not throw the whole batch into the
    // DLQ; the row stays PENDING/RUNNING and the */30 process-translation-jobs reaper
    // re-drives it (resets stale RUNNING, re-enqueues stuck PENDING).
    try {
      await processTranslationMessage(db, env, jobId, message);
    } catch (err) {
      console.error(
        JSON.stringify({
          event: 'translation_consumer_message_failed',
          jobId,
          messageId: message.id,
          error: scrubErrorForLog(err),
        }),
      );
      message.ack();
    }
  }
}

// ─── Per-message handler ──────────────────────────────────────────────────────

async function processTranslationMessage(
  db: DoDbClient,
  env: MultidealEnv,
  jobId: string,
  message: Message<{ jobId: string }>,
): Promise<void> {
  // ── Step 3: Atomic claim ──────────────────────────────────────────────────
  // Single UPDATE…RETURNING — never read-then-write (concurrency guard).
  const claimResult = await db.execute(sql`
    UPDATE translation_jobs
       SET status     = 'RUNNING',
           attempt    = attempt + 1,
           started_at = now()
     WHERE id = ${jobId}
       AND status IN ('PENDING', 'PENDING_BUDGET')
       AND (scheduled_at IS NULL OR scheduled_at <= now())
    RETURNING
      id,
      deal_id,
      target_locale,
      attempt,
      force_fresh
  `);

  // Type-narrow the Neon HTTP result rows
  type ClaimRow = {
    id: string;
    deal_id: string;
    target_locale: string;
    attempt: number;
    force_fresh: boolean;
  };

  const rows = claimResult.rows as ClaimRow[];

  if (rows.length === 0) {
    // Already claimed by another consumer or in a terminal state — ack silently.
    message.ack();
    return;
  }

  // Non-null assertion safe: rows.length > 0 is guaranteed by the guard above.
  const claimed = rows[0]!;

  // ── Step 4: Translate ──────────────────────────────────────────────────────
  try {
    const result = await translateDealFields(db, {
      dealId: claimed.deal_id,
      targetLocale: claimed.target_locale,
      forceFresh: claimed.force_fresh,
    });

    // Mark DONE with cost/token accounting
    await db.execute(sql`
      UPDATE translation_jobs
         SET status       = 'DONE',
             cost_usd     = ${result.costUsd ?? null},
             input_tokens = ${result.inputTokens ?? null},
             output_tokens = ${result.outputTokens ?? null},
             finished_at  = now()
       WHERE id = ${jobId}
    `);

    // Recompute deal.translation_status
    await afterTranslate(db, claimed.deal_id);

    console.warn(
      JSON.stringify({
        event: 'translation_job_done',
        jobId,
        dealId: claimed.deal_id,
        targetLocale: claimed.target_locale,
        attempt: claimed.attempt,
        costUsd: result.costUsd,
      }),
    );

    message.ack();
  } catch (err) {
    await handleTranslationError(db, env, jobId, claimed, err, message);
  }
}

// ─── Error handling ───────────────────────────────────────────────────────────

async function handleTranslationError(
  db: DoDbClient,
  env: MultidealEnv,
  jobId: string,
  claimed: { deal_id: string; target_locale: string; attempt: number },
  err: unknown,
  message: Message<{ jobId: string }>,
): Promise<void> {
  // ── Step 5: BudgetExhaustedError ──────────────────────────────────────────
  if (err instanceof BudgetExhaustedError) {
    await db.execute(sql`
      UPDATE translation_jobs
         SET status    = 'PENDING_BUDGET',
             last_error = 'Budget exhausted',
             finished_at = now()
       WHERE id = ${jobId}
    `);
    console.warn(
      JSON.stringify({
        event: 'translation_job_budget_exhausted',
        jobId,
        dealId: claimed.deal_id,
        targetLocale: claimed.target_locale,
        attempt: claimed.attempt,
      }),
    );
    // Budget reset cron will re-enqueue PENDING_BUDGET jobs — do not requeue here.
    message.ack();
    return;
  }

  // ── Step 6: TranslationProviderError ─────────────────────────────────────
  if (err instanceof TranslationProviderError) {
    const errMsg = err.message;

    if (err.kind === 'rate_limit' || err.kind === 'transport') {
      // Exponential backoff retry — up to TRANSLATION_RETRY_MAX attempts
      if (claimed.attempt < TRANSLATION_RETRY_MAX) {
        const delaySeconds = BACKOFF_SECONDS[claimed.attempt] ?? 300;

        await db.execute(sql`
          UPDATE translation_jobs
             SET status       = 'PENDING',
                 last_error   = ${errMsg},
                 started_at   = NULL,
                 scheduled_at = now() + (${delaySeconds} || ' seconds')::interval
           WHERE id = ${jobId}
        `);

        // Re-enqueue with delay (self-requeue pattern)
        await requeueWithDelay(env, jobId, delaySeconds);

        console.warn(
          JSON.stringify({
            event: 'translation_job_retry',
            jobId,
            dealId: claimed.deal_id,
            kind: err.kind,
            attempt: claimed.attempt,
            nextDelaySeconds: delaySeconds,
          }),
        );
      } else {
        // Exhausted retry_max — permanently fail
        const reason = `${err.kind}: ${errMsg} (max retries exhausted)`;
        await failJob(db, jobId, reason);
        await recordDeadLetter(db, 'translation-jobs-dlq', reason);
        console.error(
          JSON.stringify({
            event: 'translation_job_max_retries',
            jobId,
            dealId: claimed.deal_id,
            targetLocale: claimed.target_locale,
            attempt: claimed.attempt,
            kind: err.kind,
          }),
        );
        await afterTranslate(db, claimed.deal_id);
      }

      message.ack();
      return;
    }

    // ── Step 7: auth | content_filter | invalid_response → immediate FAILED ──
    const reason = `${err.kind}: ${errMsg}`;
    await failJob(db, jobId, reason);
    await recordDeadLetter(db, 'translation-jobs-dlq', reason);
    console.error(
      JSON.stringify({
        event: 'translation_job_permanent_failure',
        jobId,
        dealId: claimed.deal_id,
        targetLocale: claimed.target_locale,
        kind: err.kind,
        error: errMsg,
      }),
    );
    await afterTranslate(db, claimed.deal_id);
    message.ack();
    return;
  }

  // ── Unexpected error — fail permanently ──────────────────────────────────
  const errMsg = err instanceof Error ? err.message : String(err);
  const reason = `unexpected: ${errMsg}`;
  await failJob(db, jobId, reason);
  await recordDeadLetter(db, 'translation-jobs-dlq', reason);
  console.error(
    JSON.stringify({
      event: 'translation_job_unexpected_error',
      jobId,
      dealId: claimed.deal_id,
      targetLocale: claimed.target_locale,
      error: errMsg,
    }),
  );
  await afterTranslate(db, claimed.deal_id);
  message.ack();
}

// ─── Requeue helper ───────────────────────────────────────────────────────────

async function requeueWithDelay(
  env: MultidealEnv,
  jobId: string,
  delaySeconds: number,
): Promise<void> {
  if (!env.TRANSLATION_QUEUE) {
    console.error(
      JSON.stringify({
        event: 'translation_requeue_failed',
        reason: 'TRANSLATION_QUEUE binding not set on multideal-preview',
        jobId,
      }),
    );
    return;
  }

  try {
    await env.TRANSLATION_QUEUE.send({ jobId }, { delaySeconds });
  } catch (sendErr) {
    // Non-fatal — the budget-reset cron will republish PENDING jobs as a backstop.
    console.error(
      JSON.stringify({
        event: 'translation_requeue_send_failed',
        jobId,
        delaySeconds,
        error: sendErr instanceof Error ? sendErr.message : String(sendErr),
      }),
    );
  }
}