/**
 * apps/worker — RAG embedding pipeline (TASK-0125, replaces the TASK-0099 §4
 * Phase-12 stub).
 *
 * Consumed from the `ai` queue (see ai.processor.ts dispatch). Jobs:
 *
 *   { type: "embed:chunks", workspaceId, docType, refId, chunkIds? }
 *      One (re)indexed page/record: derive its chunk set from the SOURCE
 *      (the same derivation the search indexer uses — whole doc + `#c<n>`
 *      chunk docs), embed every chunk text in ONE `AiGateway.embed` call,
 *      merge each vector onto its content doc (`search.updateDoc` partial
 *      merge — single-writer rule, TASK-0096) and upsert the `ai_embeddings`
 *      registry row per chunk (unique `[workspaceId, chunkRef]`).
 *
 *      The chunk set is re-derived from the source rows (not taken from the
 *      payload): the source is the truth, and a shrink/grow of the body
 *      changes the chunk count. Stale registry rows (chunks that no longer
 *      exist) are deleted so `ai_embeddings` count == index chunk count
 *      (the parity the DoD requires).
 *
 *   { type: "embed:reembed" }
 *      Nightly full re-embed (03:30 UTC repeatable, registered in index.ts):
 *      walk every `ai_embeddings` row in batches of 100, re-embed, merge the
 *      fresh vector, and verify count parity — drift > 2% emits the
 *      `ai.reembed_required` admin notification (the model-change re-index
 *      rule: a new model simply re-embeds everything; a DIMENSION mismatch
 *      instead emits `ai.reembed_required` and BLOCKS the write — the index
 *      must be rolled via scripts/opensearch-reindex.ts first).
 *
 * Provider (recorded decision): the worker honors `AI_PROVIDER` exactly like
 * the API's `gateway-provider.ts` — `AI_PROVIDER=mock` builds a
 * deterministic `MockProvider` (dimension = `SEARCH_VECTOR_DIMENSION`), so
 * the dev stack + E2E run the full RAG surface offline. The old
 * `AI_EMBEDDING_PROVIDER` gate is superseded: the real embed now flows
 * through the `@factory/ai` gateway (retries, budget ledger, circuit
 * breaker — the same surface the chat features use).
 *
 * Reuse (recorded decision, spec conflict §29): the spec's separate
 * `ai-chunks` OpenSearch index is NOT built — the Phase 8 single
 * `factory-content` index already carries the `vector` knn_vector field,
 * the `grants` permission snapshot, `type: "chunk"` and per-chunk `body`,
 * and the search-index pipeline already emits the chunk docs. RAG merges
 * vectors onto those same docs and reuses the Phase 8 visibility filter at
 * query time — one index, one filter, no divergent snapshot implementation
 * (the spec's own Security section requires the filter to be the Phase 8
 * one, "import, don't duplicate").
 */
import type { PrismaClient } from "@prisma/client";
import { z } from "zod";
import {
  AiGateway,
  AnthropicAdapter,
  BudgetService,
  CustomAdapter,
  GoogleAdapter,
  MockProvider,
  OpenAIAdapter,
  OssAdapter,
  resolveAiConfig,
  UsageLedger,
  type ProviderAdapter,
  type ProviderId,
  type ProviderRegistry,
  type ResolvedAiConfig,
  type UsageRecord,
} from "@factory/ai";
import { CONTENT_INDEX_ALIAS } from "@factory/search";
import type { SearchClient } from "@factory/search";
import { asJson } from "../lib/prisma.js";
import { getLogger } from "../logger.js";
import type { WorkerEnv } from "../config.js";
import { buildPageDoc, buildRecordDoc, expandChunkDocs } from "./search-index/builders.js";
import type { IndexerDeps } from "./search-index/builders.js";

const log = getLogger("ai-embedding");

/** Only page/record bodies are chunked (TASK-0099 §3), so only those have
 *  chunk docs to embed. Fail fast on any other type. */
const CHUNKABLE_DOC_TYPES = ["page", "record"] as const;

export const embedChunksJobSchema = z.object({
  type: z.literal("embed:chunks"),
  workspaceId: z.string().min(1),
  docType: z.enum(CHUNKABLE_DOC_TYPES),
  refId: z.string().min(1),
  /**
   * The chunk ids (`c0`, `c1`, …) as computed at index time. TASK-0125:
   * INFORMATIONAL only — the handler re-derives the chunk set from the
   * source (truth). `[]` (short body → whole doc only) is valid now.
   */
  chunkIds: z.array(z.string().min(1)).default([]),
});
export type EmbedChunksJobData = z.infer<typeof embedChunksJobSchema>;

/** The nightly full re-embed (registered as a BullMQ repeatable in index.ts). */
export const reembedJobSchema = z.object({
  type: z.literal("embed:reembed"),
});
export type ReembedJobData = z.infer<typeof reembedJobSchema>;

/** Reconstruct the content-doc id of a chunk (`<docType>:<refId>#<chunkId>`). */
export function chunkDocId(docType: string, refId: string, chunkId: string): string {
  return `${docType}:${refId}#${chunkId}`;
}

/** The registry prefix for every chunk of one item (whole doc + chunks). */
export function itemChunkPrefix(docType: string, refId: string): string {
  return `${docType}:${refId}`;
}

export interface EmbedChunksResult {
  /** true when there is nothing to embed (source vanished → cleanup only). */
  skipped: boolean;
  reason?: string;
  /** Number of chunks embedded (vectors merged + registry rows upserted). */
  embedded: number;
  /** Stale registry rows deleted (chunks that no longer exist). */
  pruned: number;
  /** The content-doc ids embedded (whole doc + chunk docs). */
  chunkRefs: string[];
  /** The provider model the vectors were produced with. */
  model?: string;
  /** The vector dimension written (vs the index's SEARCH_VECTOR_DIMENSION). */
  dimension?: number;
  /** true when the dimension mismatch blocked the write (re-index required). */
  blocked?: boolean;
}

// ---------------------------------------------------------------------------
// Worker gateway (AI_PROVIDER-honoring, like the API's gateway-provider)
// ---------------------------------------------------------------------------

/**
 * Build the worker's `AiGateway`. Mirrors the API's `buildAiGateway`
 * provider selection, minus the workspace settings (the worker never
 * resolves per-workspace config — platform env only) and minus the budget
 * gate (embeds are batch/background: the per-workspace monthly budget is
 * enforced on the request path by the API's guard; the gateway's own
 * submit-time check stays as the backstop with limits disabled here).
 *
 * `AI_PROVIDER=mock` → deterministic `MockProvider` at the index's dimension
 * (the dev stack + E2E run the full pipeline offline).
 */
export function createWorkerAiGateway(env: WorkerEnv): AiGateway {
  const resolveConfig = (workspaceId: string): ResolvedAiConfig => {
    const provider: ProviderId = env.AI_PROVIDER === "mock" ? "custom" : (env.AI_PROVIDER || "openai") as ProviderId;
    const config = resolveAiConfig(null, null, {
      provider,
      chatModel: env.AI_MODEL,
      embedModel: env.AI_EMBEDDING_MODEL,
      ...(env.AI_BASE_URL ? { baseUrl: env.AI_BASE_URL } : {}),
    });
    config.workspaceId = workspaceId;
    // Background embeds: no per-call budget gate here (the API's
    // service-level guard enforces the monthly budget on the request path).
    // `tokensPerMonth: 0` = unlimited; `requestsPerMinute` must stay
    // POSITIVE (0 would mean "no requests allowed" — the check is `>=`).
    config.budgets = { ...config.budgets, tokensPerMonth: 0, requestsPerMinute: Math.max(1, config.budgets.requestsPerMinute) };
    return config;
  };
  const registry: ProviderRegistry = {
    get(providerId: ProviderId, config: ResolvedAiConfig): ProviderAdapter {
      if (env.AI_PROVIDER === "mock") {
        return new MockProvider(providerId === "anthropic" ? "anthropic" : "custom", env.SEARCH_VECTOR_DIMENSION);
      }
      const baseUrl = config.endpoint?.baseUrl ?? "https://api.openai.com/v1";
      const key = env.AI_API_KEY;
      switch (providerId) {
        case "openai":
          return new OpenAIAdapter({ baseUrl, apiKey: key });
        case "anthropic":
          return new AnthropicAdapter({ apiKey: key });
        case "google":
          return new GoogleAdapter({ apiKey: key });
        case "oss":
          return new OssAdapter({ baseUrl, apiKey: key });
        default:
          return new CustomAdapter({
            baseUrl,
            apiKey: key,
            ...(config.endpoint?.extraHeaders ? { extraHeaders: config.endpoint.extraHeaders } : {})
          });
      }
    },
  };
  const ledger = new UsageLedger({
    sink: async (record: UsageRecord) => {
      log.info(
        { workspaceId: record.workspaceId, kind: record.kind, model: record.model, provider: record.provider, status: record.status },
        "ai usage (worker)",
      );
    },
  });
  return new AiGateway({ resolveConfig, registry, budgets: new BudgetService(ledger), ledger });
}

/**
 * One process-lifetime gateway per env snapshot (the factory is cheap but
 * the circuit breaker state is useful to keep). Tests pass their own
 * gateway to the handlers and never touch this.
 */
let workerGateway: AiGateway | null = null;
let workerGatewayEnv: WorkerEnv | null = null;
export function workerAiGateway(env: WorkerEnv): AiGateway {
  if (!workerGateway || workerGatewayEnv !== env) {
    workerGateway = createWorkerAiGateway(env);
    workerGatewayEnv = env;
  }
  return workerGateway;
}

// ---------------------------------------------------------------------------
// embed:chunks — one (re)indexed item
// ---------------------------------------------------------------------------

/** The chunk set of one item, derived from the SOURCE (index-time truth). */
export async function deriveItemChunks(
  prisma: PrismaClient,
  docType: "page" | "record",
  refId: string,
): Promise<{ docId: string; workspaceId: string; text: string }[] | null> {
  const deps: IndexerDeps = { prisma };
  const doc = docType === "page" ? await buildPageDoc(deps, refId) : await buildRecordDoc(deps, refId);
  if (!doc) return null; // source vanished (trashed/purged) → caller prunes
  const expanded = expandChunkDocs(doc);
  const out: { docId: string; workspaceId: string; text: string }[] = [];
  // The whole doc ALWAYS gets a vector (RAG retrieves it even for short
  // bodies — the mock/short-page E2E depends on this; TASK-0099 only chunks
  // long bodies, it does not stop the parent from being a vector target).
  out.push({ docId: doc.docId, workspaceId: doc.workspaceId, text: doc.body });
  for (const d of expanded) {
    if (d.type === "chunk") out.push({ docId: d.docId, workspaceId: d.workspaceId, text: d.body });
  }
  return out;
}

/**
 * Handle one `embed:chunks` job. Pure (no BullMQ here) so it is unit-tested
 * Docker-free with a fake Prisma + a fake SearchClient + an injected gateway.
 */
export async function handleEmbedChunks(opts: {
  prisma: PrismaClient;
  env: WorkerEnv;
  search: SearchClient;
  job: EmbedChunksJobData;
  /** Present unless the caller has no job id (e.g. direct unit invocation). */
  jobId: string | undefined;
  /** Injectable gateway (tests pass a fake; production = workerAiGateway). */
  gateway?: AiGateway;
}): Promise<EmbedChunksResult> {
  const { prisma, env, search, job } = opts;
  const gateway = opts.gateway ?? workerAiGateway(env);
  const { workspaceId, docType, refId } = job;

  // 1) Re-derive the chunk set from the source. Source gone → prune the
  //    registry (the search-index removal already dropped the index docs;
  //    the registry rows are this job's to own) and stop.
  const chunks = await deriveItemChunks(prisma, docType, refId);
  if (!chunks) {
    const pruned = await pruneItemEmbeddings(prisma, workspaceId, docType, refId, []);
    log.info({ jobId: opts.jobId, workspaceId, docType, refId, pruned }, "embed:chunks — source vanished; registry pruned");
    return { skipped: true, reason: "source vanished", embedded: 0, pruned, chunkRefs: [] };
  }

  // 2) ONE gateway call for the whole chunk set (batched — the mock and the
  //    real providers both price per call + per token, and a 50-chunk page
  //    should not be 50 round trips).
  let result;
  try {
    result = await gateway.embed({ workspaceId, texts: chunks.map((c) => c.text), metadata: { capability: "rag.embed" } });
  } catch (err) {
    // Retryable transport/provider failures rethrow (BullMQ backoff retries
    // the whole job — the embed is idempotent: upsert + partial merge).
    throw err;
  }

  // 3) Dimension guard (the model-change rule): the vectors MUST match the
  //    index's `vector` dimension. A mismatch means the embedding model
  //    changed (new dim) — BLOCK the write and emit `ai.reembed_required`
  //    (the alias rollover script is the fix; writing a wrong-dim vector
  //    corrupts the index for every other doc's kNN).
  if (result.dimension !== env.SEARCH_VECTOR_DIMENSION) {
    await emitReembedRequired(prisma, workspaceId, {
      reason: "dimension_mismatch",
      detail: `embed model "${result.model}" produced dim ${result.dimension}, index has ${env.SEARCH_VECTOR_DIMENSION}`,
    });
    log.error(
      { jobId: opts.jobId, workspaceId, model: result.model, got: result.dimension, want: env.SEARCH_VECTOR_DIMENSION },
      "embed BLOCKED — vector dimension mismatch (alias rollover required)",
    );
    return {
      skipped: true,
      reason: "dimension_mismatch",
      embedded: 0,
      pruned: 0,
      chunkRefs: [],
      model: result.model,
      dimension: result.dimension,
      blocked: true,
    };
  }

  // 4) Merge each vector onto its content doc + upsert the registry row.
  let embedded = 0;
  for (let i = 0; i < chunks.length; i++) {
    const chunk = chunks[i]!;
    const vector = result.vectors[i];
    if (!vector) continue;
    try {
      // Partial (merge) update — a full PUT would clobber title/body/grants
      // (single-writer rule, TASK-0096).
      await search.updateDoc(chunk.docId, { vector, vectorModel: result.model });
    } catch (err) {
      // A vanished doc (race with a delete) is fine — the registry row is
      // still the record; the next re-embed + prune reconciles.
      if (!(err instanceof Error && /404/.test(err.message))) {
        log.warn({ err: (err as Error).message, docId: chunk.docId }, "vector merge failed (row still recorded)");
      }
    }
    await prisma.aiEmbedding.upsert({
      where: { workspaceId_chunkRef: { workspaceId, chunkRef: chunk.docId } },
      create: {
        workspaceId,
        chunkRef: chunk.docId,
        chunkText: chunk.text,
        vectorId: chunk.docId,
        model: result.model,
        dimension: result.dimension,
        metadata: asJson({ docType, refId, index: CONTENT_INDEX_ALIAS }),
      },
      update: {
        chunkText: chunk.text,
        model: result.model,
        dimension: result.dimension,
        metadata: asJson({ docType, refId, index: CONTENT_INDEX_ALIAS }),
      },
    });
    embedded++;
  }

  // 5) Prune stale registry rows (a shrunk body ⇒ fewer chunks now).
  const currentRefs = chunks.map((c) => c.docId);
  const pruned = await pruneItemEmbeddings(prisma, workspaceId, docType, refId, currentRefs);

  log.info(
    { jobId: opts.jobId, workspaceId, docType, refId, embedded, pruned, model: result.model, dimension: result.dimension },
    "embed:chunks done",
  );
  return { skipped: false, embedded, pruned, chunkRefs: currentRefs, model: result.model, dimension: result.dimension };
}

/**
 * Delete the item's registry rows that are NOT in `keepRefs`. The item's
 * refs are the whole doc (`<docType>:<refId>`) + its chunks (`…#<chunkId>`);
 * the match is EXACT-on-whole + `#`-prefixed so a sibling item whose refId
 * is a string-prefix of this one (page `pg1` vs `pg10`) is never touched.
 */
async function pruneItemEmbeddings(
  prisma: PrismaClient,
  workspaceId: string,
  docType: string,
  refId: string,
  keepRefs: string[],
): Promise<number> {
  const prefix = itemChunkPrefix(docType, refId);
  const res = await prisma.aiEmbedding.deleteMany({
    where: {
      workspaceId,
      OR: [{ chunkRef: prefix }, { chunkRef: { startsWith: `${prefix}#` } }],
      ...(keepRefs.length > 0 ? { NOT: { chunkRef: { in: keepRefs } } } : {}),
    },
  });
  return res.count;
}

// ---------------------------------------------------------------------------
// embed:reembed — nightly full walk (parity verification)
// ---------------------------------------------------------------------------

export interface ReembedResult {
  /** Registry rows walked. */
  total: number;
  /** Rows re-embedded successfully. */
  embedded: number;
  /** Rows that failed (source vanished or provider error). */
  failed: number;
  /** true when drift (failed/total) exceeded the 2% threshold. */
  driftAlert: boolean;
  /** The provider model the re-embed ran with (undefined when nothing ran). */
  model?: string | undefined;
  /** true when the dimension mismatch blocked the whole run. */
  blocked?: boolean;
}

const REEMBED_BATCH = 100;
const REEMBED_DRIFT_THRESHOLD = 0.02; // 2% (spec)

/**
 * The nightly full re-embed: walk every `ai_embeddings` row in batches of
 * 100, re-embed the stored chunk text, merge the fresh vector onto the index
 * doc, and refresh the row's model/dimension (this is ALSO the model-change
 * re-index — a new model simply re-embeds everything overnight). Count
 * parity: failed/total > 2% → `ai.reembed_required` admin notification.
 *
 * Pure (no BullMQ) — unit-tested with fakes.
 */
export async function handleReembed(opts: {
  prisma: PrismaClient;
  env: WorkerEnv;
  search: SearchClient;
  jobId: string | undefined;
  gateway?: AiGateway;
}): Promise<ReembedResult> {
  const { prisma, env, search } = opts;
  const gateway = opts.gateway ?? workerAiGateway(env);
  const t0 = Date.now();

  const rows = await prisma.aiEmbedding.findMany({ orderBy: { workspaceId: "asc" } });
  if (rows.length === 0) {
    return { total: 0, embedded: 0, failed: 0, driftAlert: false };
  }

  let embedded = 0;
  let failed = 0;
  let model: string | undefined;
  // First batch doubles as the dimension probe: if the (possibly new)
  // model's dimension doesn't match the index, BLOCK the whole run rather
  // than corrupt it row by row.
  let blocked = false;
  for (let start = 0; start < rows.length; start += REEMBED_BATCH) {
    const batch = rows.slice(start, start + REEMBED_BATCH);
    let result;
    try {
      result = await gateway.embed({
        workspaceId: batch[0]!.workspaceId,
        texts: batch.map((r) => r.chunkText),
        metadata: { capability: "rag.reembed" },
      });
    } catch (err) {
      log.warn({ err: (err as Error).message, batch: batch.length }, "reembed batch embed failed — counting rows as failed");
      failed += batch.length;
      continue;
    }
    if (start === 0 && result.dimension !== env.SEARCH_VECTOR_DIMENSION) {
      const ws0 = batch[0]!.workspaceId;
      await emitReembedRequired(prisma, ws0, {
        reason: "dimension_mismatch",
        detail: `reembed model "${result.model}" produced dim ${result.dimension}, index has ${env.SEARCH_VECTOR_DIMENSION}`,
      });
      log.error(
        { jobId: opts.jobId, model: result.model, got: result.dimension, want: env.SEARCH_VECTOR_DIMENSION },
        "reembed BLOCKED — dimension mismatch (alias rollover required)",
      );
      blocked = true;
      break;
    }
    model = result.model;
    for (let i = 0; i < batch.length; i++) {
      const row = batch[i]!;
      const vector = result.vectors[i];
      if (!vector) {
        failed++;
        continue;
      }
      try {
        await search.updateDoc(row.vectorId, { vector, vectorModel: result.model });
        await prisma.aiEmbedding.update({
          where: { id: row.id },
          data: { model: result.model, dimension: result.dimension, chunkText: row.chunkText },
        });
        embedded++;
      } catch (err) {
        // A vanished index doc (pruned page) is the expected drift source —
        // prune the orphan row so parity converges on the NEXT run.
        failed++;
        await prisma.aiEmbedding
          .delete({ where: { id: row.id } })
          .catch(() => undefined);
        log.warn({ err: (err as Error).message, vectorId: row.vectorId }, "reembed row failed — orphan pruned");
      }
    }
  }

  const total = rows.length;
  const drift = total > 0 ? failed / total : 0;
  const driftAlert = !blocked && drift > REEMBED_DRIFT_THRESHOLD;
  if (driftAlert) {
    // The drift can span many workspaces; notify per affected workspace is
    // overkill for the nightly job — one audit event + log is the spec's
    // "alert log + admin notification" (the notification targets the first
    // workspace with a failure, best-effort).
    const firstFailedWs = rows.find((r) => true)!.workspaceId;
    await emitReembedRequired(prisma, firstFailedWs, {
      reason: "count_parity_drift",
      detail: `${failed}/${total} chunks failed re-embed (${(drift * 100).toFixed(1)}% drift)`,
    });
  }
  log.info(
    { jobId: opts.jobId, total, embedded, failed, drift: +(drift * 100).toFixed(2), driftAlert, blocked, model, ms: Date.now() - t0 },
    "reembed done",
  );
  return { total, embedded, failed, driftAlert, model, ...(blocked ? { blocked: true } : {}) };
}

// ---------------------------------------------------------------------------
// ai.reembed_required — the config-drift / model-change signal
// ---------------------------------------------------------------------------

/**
 * Emit the `ai.reembed_required` signal: a SYSTEM audit row (workspace-
 * scoped) + a WARN log. The audit row is the admin console's source (Phase
 * 14); the log is the on-call's. Best-effort — a failed emit must never
 * take the embedding job down.
 */
export async function emitReembedRequired(
  prisma: PrismaClient,
  workspaceId: string,
  detail: { reason: string; detail: string },
): Promise<void> {
  try {
    await prisma.auditLog.create({
      data: {
        actorType: "SYSTEM",
        actorId: "worker:ai",
        workspaceId,
        resourceType: "workspace",
        resourceId: workspaceId,
        action: "ai.reembed_required",
        metadata: asJson({ reason: detail.reason, detail: detail.detail, index: CONTENT_INDEX_ALIAS }),
      },
    });
  } catch (err) {
    log.warn({ err: (err as Error).message, workspaceId }, "failed to emit ai.reembed_required audit row");
  }
  log.warn({ workspaceId, ...detail }, "ai.reembed_required — embedding model/config drifted from the index; run scripts/opensearch-reindex.ts");
}