/**
 * apps/worker — AI queue processor (embeddings + generation, stub-real).
 *
 * Jobs (queue `ai`):
 *   {type:"embed", workspaceId, texts: string[], chunkId?: string}
 *      → POST {AI_BASE_URL}/embeddings {model, input: texts}
 *      → vector MERGED onto the single content doc (chunkId IS the doc id —
 *        TASK-0096 removed the 5-index scheme) + an AiEmbedding row
 *        (workspace/chunkRef upsert). If chunkId is absent a stable auto id
 *        is used.
 *   {type:"generate", prompt: string, workspaceId?, conversationId?, userId?}
 *      → POST {AI_BASE_URL}/chat/completions {model, messages:[{user}]}
 *      → result returned; when conversationId+userId are present the
 *        assistant turn is persisted as an AiMessage row.
 *   {type:"embed:chunks", workspaceId, docType, refId, chunkIds?}
 *      → TASK-0125 real RAG embedding (ai-embedding.ts): re-derive the item's
 *        chunk set from the source, ONE AiGateway.embed call (AI_PROVIDER-
 *        honoring — mock works keyless), merge each vector onto its content
 *        doc (search.updateDoc partial merge), upsert the ai_embeddings
 *        registry row, prune stale rows. Dimension mismatch → BLOCK + the
 *        ai.reembed_required signal (alias rollover required).
 *   {type:"embed:reembed"}  (TASK-0125)
 *      → nightly full re-embed (registered as a repeatable in index.ts): walk
 *        every ai_embeddings row in batches of 100, re-embed + merge, prune
 *        orphans; count parity drift > 2% → ai.reembed_required.
 *   {type:"usage.rollup", workspaceId?}  (TASK-0121)
 *      → nightly (registered as a repeatable in index.ts) + manual: reconcile
 *        per-workspace month-to-date token totals (sums ai_messages of the
 *        workspace's own conversations since month start) and enqueue a
 *        budget.notify for any budget that was crossed but not yet notified
 *        (the 24 h slot is checked by the notify job — race-safe).
 *   {type:"budget.notify", workspaceId, used, limit}  (TASK-0121)
 *      → the budget-exceeded admin notification: IN_APP Notification rows for
 *        every OWNER/ADMIN + the notification.created stream event (email
 *        channel via the worker's email decision) + the ai.budget_exceeded
 *        audit row. Throttled to one per workspace per 24 h (the shared
 *        ai:budget:notified:<ws> Redis slot, SET NX).
 *
 * Dev-safe: without AI_API_KEY every job logs "skipping — no AI key" and
 * completes (Phase 12 wires the full provider gateway; the request shaping
 * here is already provider-agnostic and unit-tested with mocked fetch).
 */
import { randomUUID } from "node:crypto";
import { asJson } from "../lib/prisma.js";

import { Worker, type Job } from "bullmq";
import { withJobOtel } from "../lib/job-otel.js";
import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient } from "@prisma/client";
import { QUEUES, enqueue } from "../queues.js";
import { getLogger } from "../logger.js";
import type { WorkerEnv } from "../config.js";
import type { SearchClient } from "@factory/search";
import { embedChunksJobSchema, handleEmbedChunks, handleReembed, reembedJobSchema } from "./ai-embedding.js";

const log = getLogger("ai");

const embedJobSchema = z.object({
  type: z.literal("embed"),
  workspaceId: z.string().min(1),
  texts: z.array(z.string()).min(1).max(100),
  /** Content doc id the vector merges into, e.g. "page:abc" (auto if absent). */
  chunkId: z.string().min(1).optional(),
  /** DEPRECATED no-op (TASK-0096): the single content index has no per-type
   * override — accepted for backward compatibility with in-flight jobs. */
  index: z.string().optional()
});
const generateJobSchema = z.object({
  type: z.literal("generate"),
  prompt: z.string().min(1),
  workspaceId: z.string().optional(),
  conversationId: z.string().optional(),
  userId: z.string().optional(),
  system: z.string().optional()
});
/**
 * TASK-0121 — hourly AI usage rollup (worker-owned, registered as a
 * BullMQ repeatable in `apps/worker/src/index.ts`). Per workspace
 * (all workspaces when `workspaceId` is absent): sum `ai_messages`
 * `tokensIn + tokensOut` since month-start (the same range the API's
 * budget guard uses) and log the metric; when a workspace has crossed
 * `usageLimits.aiMonthlyTokens` and has NOT been notified this 24 h
 * (the shared `ai:budget:notified:<ws>` Redis slot), enqueue a
 * `budget.notify` job — so an admin is alerted within one rollup cycle
 * even if the workspace never makes another AI call after crossing.
 */
const usageRollupJobSchema = z.object({
  type: z.literal("usage.rollup"),
  workspaceId: z.string().min(1).optional()
});
/**
 * TASK-0121 — budget-exceeded admin notification (worker-owned): create an
 * IN_APP Notification row for every workspace OWNER/ADMIN + publish the
 * `notification.created` domain event (the worker's email-decision consumer
 * sends the email channel per the user's prefs — the same channel every
 * notification type uses). The 24 h throttle is the shared Redis slot (the
 * API's guard hook and this job race on the same key — one notification per
 * workspace per day).
 */
const budgetNotifyJobSchema = z.object({
  type: z.literal("budget.notify"),
  workspaceId: z.string().min(1),
  used: z.number().int().nonnegative(),
  limit: z.number().int().positive()
});
export const aiJobSchema = z.discriminatedUnion("type", [
  embedJobSchema,
  generateJobSchema,
  embedChunksJobSchema,
  reembedJobSchema,
  usageRollupJobSchema,
  budgetNotifyJobSchema
]);
export type AiJobData = z.infer<typeof aiJobSchema>;

// ---------------------------------------------------------------------------
// Request shaping (exported for unit tests with mocked fetch)
// ---------------------------------------------------------------------------

export function embeddingsEndpoint(baseUrl: string): string {
  return baseUrl.replace(/\/+$/, "") + "/embeddings";
}
export function chatEndpoint(baseUrl: string): string {
  return baseUrl.replace(/\/+$/, "") + "/chat/completions";
}

/** Request body for POST /embeddings (shaped independently of the URL). */
export function buildEmbeddingsRequest(
  model: string,
  texts: string[]
): { model: string; input: string[] } {
  return { model, input: texts };
}

export function buildChatRequest(
  model: string,
  prompt: string,
  system?: string
): { model: string; messages: { role: string; content: string }[] } {
  const messages: { role: string; content: string }[] = [];
  if (system) messages.push({ role: "system", content: system });
  messages.push({ role: "user", content: prompt });
  return { model, messages };
}

async function aiHeaders(env: WorkerEnv): Promise<Record<string, string>> {
  return {
    "content-type": "application/json",
    authorization: `Bearer ${env.AI_API_KEY}`
  };
}

// ---------------------------------------------------------------------------
// Chunk → index mapping (vector updates land in the content index)
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// Chunk → index mapping (TASK-0096: single content index)
// ---------------------------------------------------------------------------

// The vector lands on the SAME doc the search indexer wrote: `chunkRef` IS the
// content doc id (`<type>:<refId>`, user: `user:<ws>:<uid>`). The 5-index
// scheme (factory_pages/…) was removed in TASK-0096 — there is one
// `factory-content` index now, so no per-type index map is needed.
const CONTENT_INDEX = "factory-content";

// ---------------------------------------------------------------------------
// Worker
// ---------------------------------------------------------------------------

export function createAiWorker(opts: {
  connection: Redis;
  prisma: PrismaClient;
  env: WorkerEnv;
  search: SearchClient;
  concurrency?: number;
  fetchImpl?: typeof fetch;
}): Worker {
  const { connection, prisma, env, search } = opts;
  const doFetch = opts.fetchImpl ?? fetch;
  const aiBase = env.AI_BASE_URL || "https://api.openai.com/v1";

  return new Worker(
    QUEUES.ai,
    withJobOtel(QUEUES.ai, async (job: Job) => {
      const started = Date.now();
      const data = aiJobSchema.parse(job.data);

      // RAG embedding jobs (TASK-0125) route BEFORE the AI_API_KEY gate:
      // they flow through the worker's own gateway (AI_PROVIDER-honoring —
      // `mock` works with no key; a named provider needs AI_API_KEY, which
      // the gateway resolves from the platform env via `env:AI_API_KEY`).
      if (data.type === "embed:chunks") {
        return handleEmbedChunks({ prisma, env, search, job: data, jobId: job.id });
      }
      if (data.type === "embed:reembed") {
        return handleReembed({ prisma, env, search, jobId: job.id });
      }

      // TASK-0121 usage/budget jobs are DB+Redis only (no provider call) —
      // they run regardless of AI_API_KEY. Logic lives in the exported
      // runUsageRollup / runBudgetNotify (unit-tested without BullMQ).
      if (data.type === "usage.rollup") {
        return runUsageRollup(
          {
            prisma,
            redis: connection,
            enqueueBudgetNotify: (d) => enqueue(connection, QUEUES.ai, d as Record<string, unknown>)
          },
          data,
          job.id ?? null
        );
      }
      if (data.type === "budget.notify") {
        return runBudgetNotify({ prisma, redis: connection }, data, job.id ?? null);
      }

      if (!env.AI_API_KEY) {
        log.info(
          { jobId: job.id, type: data.type },
          "skipping — no AI_API_KEY configured (Phase 12 wires the gateway)"
        );
        return { skipped: true, reason: "no ai api key" };
      }

      if (data.type === "embed") {
        return handleEmbed(job, data, started);
      }
      return handleGenerate(job, data, started);

      async function handleEmbed(
        job: Job,
        data: z.infer<typeof embedJobSchema>,
        started: number
      ): Promise<Record<string, unknown>> {
        const url = embeddingsEndpoint(aiBase);
        const body = buildEmbeddingsRequest(env.AI_MODEL, data.texts);
        const res = await doFetch(url, {
          method: "POST",
          headers: await aiHeaders(env),
          body: JSON.stringify(body)
        });
        if (!res.ok) {
          throw new Error(`embeddings HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`);
        }
        const json = (await res.json()) as {
          data: { index: number; embedding: number[] }[];
          usage?: { prompt_tokens: number; total_tokens: number };
        };
        const vectors = json.data.map((d) => d.embedding);
        if (vectors.length !== data.texts.length) {
          throw new Error(`embedding count mismatch: ${vectors.length} != ${data.texts.length}`);
        }

        const results: { chunkId: string; dimension: number; indexed: boolean }[] = [];
        for (let i = 0; i < data.texts.length; i++) {
          const chunkRef = data.chunkId ?? `auto:${randomUUID()}`;
          const docId = chunkRef; // chunkRef IS the content doc id
          const vector = vectors[i]!;
          let indexed = false;
          try {
            // Partial (merge) update of the `vector` field on the single
            // content doc — a full PUT here would clobber the doc's
            // title/body/grants (single-writer rule, TASK-0096).
            await search.updateDoc(docId, { vector });
            indexed = true;
          } catch (err) {
            log.warn(
              { err: (err as Error).message, docId },
              "vector merge into content doc failed (row still recorded)"
            );
          }
          await prisma.aiEmbedding.upsert({
            where: { workspaceId_chunkRef: { workspaceId: data.workspaceId, chunkRef } },
            create: {
              workspaceId: data.workspaceId,
              chunkRef,
              chunkText: data.texts[i]!,
              vectorId: docId,
              model: env.AI_MODEL,
              dimension: vector.length,
              metadata: asJson({ index: CONTENT_INDEX })
            },
            update: {
              chunkText: data.texts[i]!,
              model: env.AI_MODEL,
              dimension: vector.length,
              metadata: asJson({ index: CONTENT_INDEX })
            }
          });
          results.push({ chunkId: chunkRef, dimension: vector.length, indexed });
        }
        log.info(
          {
            jobId: job.id,
            texts: data.texts.length,
            usage: json.usage,
            ms: Date.now() - started
          },
          "embeddings stored"
        );
        return { embedded: results, usage: json.usage };
      }

      async function handleGenerate(
        job: Job,
        data: z.infer<typeof generateJobSchema>,
        started: number
      ): Promise<Record<string, unknown>> {
        const url = chatEndpoint(aiBase);
        const body = buildChatRequest(env.AI_MODEL, data.prompt, data.system);
        const res = await doFetch(url, {
          method: "POST",
          headers: await aiHeaders(env),
          body: JSON.stringify(body)
        });
        if (!res.ok) {
          throw new Error(`chat HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`);
        }
        const json = (await res.json()) as {
          choices?: { message?: { content?: string } }[];
          usage?: { prompt_tokens: number; completion_tokens: number; total_tokens: number };
        };
        const content = json.choices?.[0]?.message?.content ?? "";
        if (data.conversationId && data.userId) {
          await prisma.aiMessage.create({
            data: {
              conversationId: data.conversationId,
              role: "assistant",
              content,
              model: env.AI_MODEL,
              tokensOut: json.usage?.completion_tokens ?? null,
              latencyMs: Date.now() - started
            }
          });
        }
        log.info(
          { jobId: job.id, model: env.AI_MODEL, usage: json.usage, ms: Date.now() - started },
          "generation complete"
        );
        return { content, model: env.AI_MODEL, usage: json.usage };
      }
    }),
    {
      connection,
      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
    }
  );
}

// ---------------------------------------------------------------------------
// TASK-0121 — usage.rollup / budget.notify (exported, unit-testable without
// BullMQ: the deps are structural, so tests pass plain mocks).
// ---------------------------------------------------------------------------

/** The 24 h budget-notification slot key (shared with the API's guard). */
export function budgetNotifySlot(workspaceId: string): string {
  return `ai:budget:notified:${workspaceId}`;
}

/**
 * The monthly AI token budget for a workspace (mirrors the API's
 * `tokensPerMonthFor`): `max(usageLimits.aiMonthlyTokens,
 * settings.ai.budgets.tokensPerMonth)`; 0 = unlimited.
 */
export function aiMonthlyBudget(ws: {
  usageLimits: unknown;
  settings: unknown;
}): number {
  const usageLimits = (ws.usageLimits ?? {}) as { aiMonthlyTokens?: number };
  const settings = (ws.settings ?? {}) as { ai?: { budgets?: { tokensPerMonth?: number } } };
  const fromLimits =
    typeof usageLimits.aiMonthlyTokens === "number" && Number.isFinite(usageLimits.aiMonthlyTokens)
      ? Math.max(0, usageLimits.aiMonthlyTokens)
      : 0;
  const fromSettings =
    typeof settings.ai?.budgets?.tokensPerMonth === "number"
      ? Math.max(0, settings.ai.budgets.tokensPerMonth)
      : 0;
  return Math.max(fromLimits, fromSettings);
}

export interface RunUsageRollupDeps {
  /** Structural deps — the worker passes its PrismaClient; unit tests pass a
   * targeted fake (`as unknown as PrismaClient`, the ai-embedding convention). */
  prisma: PrismaClient;
  redis: Redis;
  /** Enqueue a `budget.notify` job (the worker passes the BullMQ helper;
   * tests pass a spy). */
  enqueueBudgetNotify: (data: {
    type: "budget.notify";
    workspaceId: string;
    used: number;
    limit: number;
  }) => Promise<unknown>;
}

/**
 * TASK-0121 — `usage.rollup`: reconcile the per-workspace month-to-date
 * token totals (the AI ledger is `ai_messages`, same range the API budget
 * guard uses) and sweep for workspaces that CROSSED their budget but have
 * gone quiet — the API's submit-time guard only notifies on the next call.
 * A workspace that crossed AND has no 24 h notification slot set gets a
 * `budget.notify` job (the throttle is checked THERE, race-safe).
 */
export async function runUsageRollup(
  deps: RunUsageRollupDeps,
  job: z.infer<typeof usageRollupJobSchema>,
  jobId: string | number | null
): Promise<Record<string, unknown>> {
  const t0 = Date.now();
  const monthStart = new Date();
  monthStart.setUTCDate(1);
  monthStart.setUTCHours(0, 0, 0, 0);

  // Workspaces to reconcile: the single requested one, or — in the
  // scheduled run — every workspace with at least one AI conversation
  // (a workspace with none has used 0 and cannot have crossed).
  const scope = job.workspaceId
    ? { where: { id: job.workspaceId } }
    : { where: { aiConversation: { some: {} } } };
  const workspaces = (await deps.prisma.workspace.findMany({
    where: scope.where,
    select: { id: true, settings: true, usageLimits: true }
  })) as { id: string; settings: unknown; usageLimits: unknown }[];

  let reconciled = 0;
  let budgetsExceeded = 0;
  let enqueued = 0;
  for (const ws of workspaces) {
    // Month-to-date tokens for THIS workspace: its conversations' messages,
    // createdAt >= month start. The aggregate is scoped to the workspace's
    // own conversation ids — never another workspace's rows.
    const conversations = await deps.prisma.aiConversation.findMany({
      where: { workspaceId: ws.id },
      select: { id: true }
    });
    let used = 0;
    if (conversations.length > 0) {
      const sums = await deps.prisma.aiMessage.aggregate({
        where: {
          conversationId: { in: conversations.map((c) => c.id) },
          createdAt: { gte: monthStart }
        },
        _sum: { tokensIn: true, tokensOut: true }
      });
      used = (sums._sum.tokensIn ?? 0) + (sums._sum.tokensOut ?? 0);
    }

    const limit = aiMonthlyBudget(ws);
    reconciled++;

    if (limit > 0 && used >= limit) {
      budgetsExceeded++;
      const notified = await deps.redis.get(budgetNotifySlot(ws.id)).catch(() => null);
      if (!notified) {
        await deps.enqueueBudgetNotify({ type: "budget.notify", workspaceId: ws.id, used, limit });
        enqueued++;
      }
    }

    log.info(
      { jobId, workspaceId: ws.id, used, limit, ms: Date.now() - t0 },
      "ai usage rollup (month to date)"
    );
  }

  return { reconciled, budgetsExceeded, enqueued, ms: Date.now() - t0 };
}

export interface RunBudgetNotifyDeps {
  prisma: PrismaClient;
  redis: Redis;
}

/**
 * TASK-0121 — `budget.notify`: the budget-exceeded admin notification,
 * worker-owned so a workspace that crossed its budget mid-month and never
 * calls the API again is still alerted (the API's guard hook only fires on
 * the NEXT call). Creates an IN_APP Notification for every OWNER/ADMIN +
 * publishes `notification.created` (the worker's email-decision consumer
 * owns the email channel — the same path every notification uses).
 * Throttle: the shared `ai:budget:notified:<ws>` slot (SET NX 24 h) — the
 * API hook and this job race on it; the loser skips (one per workspace per
 * 24 h).
 */
export async function runBudgetNotify(
  deps: RunBudgetNotifyDeps,
  job: { type: "budget.notify"; workspaceId: string; used: number; limit: number },
  jobId: string | number | null
): Promise<Record<string, unknown>> {
  const { workspaceId } = job;
  const claimed = await deps.redis
    .set(budgetNotifySlot(workspaceId), "1", "EX", 24 * 3600, "NX")
    .catch(() => "OK" as const);
  if (claimed !== "OK") {
    log.info({ jobId, workspaceId }, "budget notification already sent this 24 h — skipped");
    return { skipped: true, reason: "already notified" };
  }

  const admins = await deps.prisma.workspaceMember.findMany({
    where: { workspaceId, role: { in: ["OWNER", "ADMIN"] } },
    select: { userId: true }
  });
  if (admins.length === 0) {
    log.warn({ jobId, workspaceId }, "budget exceeded but no OWNER/ADMIN member — no notification");
    return { notified: 0 };
  }

  const body =
    `The workspace AI token budget was reached (${job.used} used of ${job.limit} monthly). ` +
    `New AI calls are blocked until the limit is raised (workspace settings → usageLimits.aiMonthlyTokens).`;
  const title = "AI budget exceeded";
  let notified = 0;
  for (const admin of admins) {
    const row = await deps.prisma.notification.create({
      data: {
        workspaceId,
        userId: admin.userId,
        type: "AUTOMATION",
        channel: "IN_APP",
        title,
        body,
        link: asJson({ type: "workspace", id: workspaceId })
      },
      select: { id: true }
    });
    // The worker owns the email channel: publish the same
    // `notification.created` event the API publishes (the email-decision
    // consumer applies per-user prefs; no actorId → SYSTEM actor).
    await deps.redis
      .xadd(
        "factory:events",
        "*",
        "event",
        "notification.created",
        "payload",
        JSON.stringify({
          workspaceId,
          notificationId: row.id,
          userId: admin.userId,
          type: "AUTOMATION",
          title,
          deepLink: { type: "workspace", id: workspaceId }
        }),
        "ts",
        new Date().toISOString()
      )
      .catch(() => undefined);
    notified++;
  }

  await deps.prisma.auditLog.create({
    data: {
      actorType: "SYSTEM",
      actorId: "worker:ai",
      workspaceId,
      resourceType: "workspace",
      resourceId: workspaceId,
      action: "ai.budget_exceeded",
      metadata: asJson({ used: job.used, limit: job.limit, adminsNotified: notified, source: "usage.rollup" })
    }
  });

  log.info({ jobId, workspaceId, notified }, "ai budget notification sent to admins");
  return { notified };
}