/**
 * TASK-0125 — RAG workspace search (query path).
 *
 * `POST /api/v1/workspaces/:workspaceId/ai/search` (body `{ query, context? }`)
 * → SSE answer stream. The documented wire contract (docs/api/ai.md §9):
 *
 *   sources → delta* → usage → done
 *
 * The FIRST frame is `sources` (the server-side citations, built from the
 * retrieval results — deterministic, independent of the model). A real model
 * would stream inline `[n]` markers that the UI maps onto this frame; the
 * deterministic mock (`metadata.mode = "rag"`) instead streams a cited answer
 * whose `[n]` marker + source label are parsed from the SAME numbered context
 * it is given, so the integration/E2E path exercises the real
 * prompt → provider → server-citation contract.
 *
 * Retrieval (the permission-isolation proof): the viewer is resolved by the
 * search module's fail-closed resolver (non-member → 404; zero memberships →
 * empty) and the query runs the Phase-8 hybrid path (BM25 + kNN, RRF k=60,
 * both legs filtered by the requester's tenant + visibility snapshot) —
 * "import, don't duplicate" (the spec's Security requirement). A user can
 * therefore never retrieve a chunk of a page they cannot read; the kNN leg
 * carries the visibility filter in its own `filter` clause (queries.ts).
 *
 * Budget/cap: the query embed + the answer chat both count against the
 * workspace budget (the pre-stream `guard.check`) and the shared per-user
 * `ai.calls` limiter (30 chat+writing+rag/min). The answer is persisted in a
 * HIDDEN per-call conversation (`metadata.feature="rag"`, the citation list
 * in `toolResults`) for the usage ledger — it never appears in the panel.
 *
 * Top-12: the spec's "top-12 chunks". Chunk hits are de-duplicated to their
 * parent (a page over the 1600-char threshold can surface several chunks;
 * the UI cites the parent), so the de-duped set is capped at 12 — a page
 * with 3 retrieved chunks cites ONCE (its parent), freeing slots for other
 * docs (recorded decision: the spec counts chunks, the UI cites sources).
 */
import { Inject, Injectable } from "@nestjs/common";
import type { Request, Response } from "express";

import type { PrismaClient } from "@prisma/client";
import type { SearchResult, SearchType } from "@factory/search";
import type { AiGateway } from "@factory/ai";

import { PrismaService } from "../../common/prisma.service.js";
import { logger } from "../../common/logger.js";
import { ApiError, type ErrorCode, type ErrorDetail } from "../../common/errors.js";
import { assertCan } from "../../lib/authorize.js";
import { SearchService } from "../search/search.service.js";
import {
  AI_GATEWAY,
  mapAiError,
  primeWorkspaceCache,
  type AiBudgetGuard,
} from "./gateway-provider.js";
import { AiBudgetGuardService } from "./budget-guard.service.js";
import { AiRateLimitService } from "./ai-rate-limit.js";
import { RAG_CONV_PREFIX } from "./prompts.js";
import type { RagSearchInput } from "./rag.dto.js";

/** The spec's top-12 (de-duplicated to parent sources). */
const RAG_TOP_N = 12;
/** Hybrid candidates retrieved before de-dup + cap (the kNN leg's k). */
const RAG_CANDIDATES = 50;
/** The answer context's per-source text cap (keeps the prompt bounded). */
const RAG_CONTEXT_CHARS = 1200;
const RAG_TIMEOUT_MS = 60_000;
const KEEPALIVE_MS = 10_000;

/** One citation (the `sources` frame element + the persisted toolResults). */
export interface RagSource {
  /** 1-based citation number (the inline `[n]` marker). */
  n: number;
  /** The content doc id (`page:<id>`, `page:<id>#c<n>`, …). */
  docId: string;
  /** The parent resource id (page/record/… id, not the doc id). */
  id: string;
  /** The public type (chunk hits surface as their parent's kind). */
  type: SearchType;
  title: string;
  /** The deep link the citation chip navigates to. */
  url: string;
  /** A short snippet for the chip hover / sources list. */
  snippet: string;
  /** Which retrieval paths produced the hit: bm25 / vec / bm25+vec. */
  source: SearchResult["source"];
  /** Deep-link metadata (the chip navigates by the workspace slug + these). */
  pageId?: string;
  blockId?: string;
  databaseId?: string;
  recordId?: string;
}

/** One SSE frame — the documented wire contract (docs/api/ai.md §9). */
export type RagSseEvent =
  | { type: "sources"; sources: RagSource[] }
  | { type: "delta"; text: string }
  | { type: "usage"; tokensIn: number; tokensOut: number }
  | { type: "error"; code: string; message: string; retryable?: boolean; retryAfterSeconds?: number }
  | { type: "done"; assistantMessageId?: string; userMessageId?: string; reason?: "ok" | "aborted" | "error" };

function sseFrame(event: RagSseEvent): string {
  return `data: ${JSON.stringify(event)}\n\n`;
}

/** A deduplicated source row (the de-duped, RRF-ordered retrieval result). */
interface RankedSource {
  hit: SearchResult;
  /** The full chunk text (from `_source.body`) for the prompt context. */
  body: string;
}

/**
 * De-duplicate the fused hybrid results to their PARENT source, preserving
 * RRF order, and cap at `n`. A chunk hit (`page:<id>#c<k>`) collapses onto
 * its parent (`page:<id>`); the FIRST (highest-ranked) hit per parent wins
 * its title/url/snippet, but the body kept is the FIRST chunk's (the text
 * that actually ranked — more specific than the whole doc for long pages).
 */
export function dedupeToParentSources(results: SearchResult[], n: number): SearchResult[] {
  const byParent = new Map<string, SearchResult>();
  for (const r of results) {
    const parent = r.docId.includes("#") ? r.docId.slice(0, r.docId.indexOf("#")) : r.docId;
    const existing = byParent.get(parent);
    if (!existing) byParent.set(parent, r);
    // keep the existing (higher-ranked) hit; nothing to merge (title/url are
    // identical for parent + chunks — they share the parent's facets)
  }
  return [...byParent.values()].slice(0, n);
}

/** The numbered context block the RAG prompt builder emits for one source. */
export function renderRagContext(n: number, label: string, body: string): string {
  const text = body.length > RAG_CONTEXT_CHARS ? `${body.slice(0, RAG_CONTEXT_CHARS)}…` : body;
  return `[${n}] Source: ${label}\n${text}`;
}

/** The RAG system prompt (a real model follows it; the mock ignores it). */
export const RAG_SYSTEM_PROMPT =
  "You are a search-augmented assistant. Answer the user's question using ONLY the numbered " +
  "workspace sources provided. Cite your sources inline with [n] markers that reference the " +
  "source number. If the sources do not contain the answer, say so. Never invent workspace " +
  "content.";

@Injectable()
export class RagService {
  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaClient,
    @Inject(AI_GATEWAY) private readonly gateway: AiGateway,
    @Inject(AiBudgetGuardService) private readonly guardService: AiBudgetGuardService,
    @Inject(AiRateLimitService) private readonly rateLimit: AiRateLimitService,
    @Inject(SearchService) private readonly search: SearchService,
  ) {}

  private get guard(): AiBudgetGuard {
    return this.guardService.guard;
  }

  async ask(workspaceId: string, userId: string, input: RagSearchInput, req: Request, res: Response): Promise<void> {
    const q = input.query.trim();

    // ---- fail-closed workspace gate (the search viewer's documented rule) ----
    // The viewer is the search module's fail-closed resolver: an explicit
    // `workspaceId` the caller is NOT a member of simply does not exist for
    // them → 404 NOT_FOUND (existence hidden, NOT 403). This MUST run before
    // `assertCan` below — assertCan 403s on a NONE level and would otherwise
    // mask the 404 boundary. Zero memberships + no workspaceId → null → empty
    // (the query is never sent).
    const viewer = await this.search.viewerForWorkspace(userId, workspaceId);

    // ---- pre-stream gates (canonical JSON envelope on failure) ----
    // Members always have ≥ VIEW (GUEST→VIEW, MEMBER→EDIT), so this passes
    // for everyone the viewer resolution just admitted; it is the belt-
    // and-braces level gate + the budget gate's precondition.
    await assertCan(this.prisma, { workspaceId, userId }, "VIEW");
    const rl = await this.rateLimit.check(userId);
    if (!rl.allowed) {
      throw new ApiError("RATE_LIMITED", 429, "AI call limit reached.", [
        { reason: "ai_call_rate_limited", retryAfterSeconds: rl.retryAfterSeconds },
      ]);
    }
    await primeWorkspaceCache(this.prisma, workspaceId);
    try {
      await this.guard.check(workspaceId, undefined, req);
    } catch (err) {
      throw mapAiError(err);
    }

    // ---- retrieval (the permission-isolation proof) ----
    const client = this.search.searchClient();
    let ranked: RankedSource[] = [];
    const wantedTypes = input.context?.types;
    if (viewer) {
      const queryVector = (await this.gateway.embed({ workspaceId, texts: [q], metadata: { capability: "rag.query" } })).vectors[0];
      const response = await client.search({ q, mode: "hybrid", pageSize: RAG_CANDIDATES }, viewer, queryVector);
      // The search page's type chips → a metadata filter. The kNN leg cannot
      // filter chunk docs by their PARENT's public type (a chunk doc carries
      // `type:"chunk"`, recorded conflict, task file), so the filter is
      // applied to the FUSED hits by their resolved display type — a clean,
      // non-leaky guarantee that only the requested types are cited.
      const hits =
        wantedTypes && wantedTypes.length > 0
          ? response.results.filter((r) => wantedTypes.includes(r.type))
          : response.results;
      const top = dedupeToParentSources(hits, RAG_TOP_N);
      // Full chunk text for the prompt context (the 160-char snippet is too
      // thin); a doc deleted between retrieval + mget drops out of context
      // (its source row stays — the citation is still valid).
      const bodies = await client.getDocs(top.map((r) => r.docId));
      const bodyById = new Map(bodies.map((b) => [b.docId, String(b.source.body ?? "")]));
      ranked = top.map((hit) => ({ hit, body: bodyById.get(hit.docId) ?? hit.snippet ?? "" }));
    }

    const sources: RagSource[] = ranked.map((r, i) => ({
      n: i + 1,
      docId: r.hit.docId,
      id: r.hit.id,
      type: r.hit.type,
      title: r.hit.title,
      url: r.hit.url,
      snippet: r.hit.snippet ?? "",
      source: r.hit.source,
      // Deep-link metadata (chunk docs inherit the parent's pageId /
      // databaseId + recordId — the chip navigates by the slug + these).
      ...(r.hit.pageId ? { pageId: r.hit.pageId } : {}),
      ...(r.hit.blockId ? { blockId: r.hit.blockId } : {}),
      ...(r.hit.databaseId ? { databaseId: r.hit.databaseId } : {}),
      ...(r.hit.recordId ? { recordId: r.hit.recordId } : {}),
    }));

    // ---- hidden per-call conversation (never listed in the panel) ----
    const conversation = await this.prisma.aiConversation.create({
      data: {
        workspaceId,
        userId,
        title: `${RAG_CONV_PREFIX}${q.slice(0, 60)}`,
        context: { query: q, context: input.context ?? null } as object,
      },
    });
    const userRow = await this.prisma.aiMessage.create({
      data: { conversationId: conversation.id, role: "user", content: q },
    });

    // ---- SSE header (the first byte — after this, errors are in-band) ----
    res.status(200);
    res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
    res.setHeader("Cache-Control", "no-cache, no-transform");
    res.setHeader("Connection", "keep-alive");
    res.setHeader("X-Accel-Buffering", "no");
    res.flushHeaders?.();
    if (typeof res.socket?.setNoDelay === "function") res.socket.setNoDelay(true);

    // The FIRST frame: the server-side citations (deterministic — the UI's
    // chips + sources list render from it regardless of the model's output).
    res.write(sseFrame({ type: "sources", sources }));

    // abort plumbing (same as Ask AI / writing: res close + writableEnded)
    const ac = new AbortController();
    let clientClosed = false;
    const onResClose = (): void => {
      if (res.writableEnded) return;
      clientClosed = true;
      ac.abort();
    };
    res.on("close", onResClose);
    const keepalive = setInterval(() => {
      if (!res.writableEnded) res.write(": keepalive\n\n");
    }, KEEPALIVE_MS);
    let timer: ReturnType<typeof setTimeout> | undefined;
    const armTimeout = (): void => {
      if (timer) clearTimeout(timer);
      timer = setTimeout(() => ac.abort(), RAG_TIMEOUT_MS);
    };
    armTimeout();

    let finished = false;
    const finish = (event: RagSseEvent): void => {
      if (finished) return;
      finished = true;
      if (timer) clearTimeout(timer);
      clearInterval(keepalive);
      res.off("close", onResClose);
      if (!res.writableEnded) {
        res.write(sseFrame(event));
        res.end();
      }
    };
    const sendErrorDelta = (err: unknown): void => {
      const mapped = mapAiError(err);
      const detail = (mapped.details?.[0] ?? {}) as Record<string, unknown>;
      const retryable = detail.retryable === true || detail.retryAfterSeconds != null || mapped.code === "RATE_LIMITED";
      res.write(
        sseFrame({
          type: "error",
          code: mapped.code,
          message: mapped.message,
          ...(retryable ? { retryable: true } : {}),
          ...(typeof detail.retryAfterSeconds === "number" ? { retryAfterSeconds: detail.retryAfterSeconds } : {}),
        }),
      );
      logger.warn(`rag stream error workspace=${workspaceId} code=${mapped.code}`, {
        event: "ai.rag.error",
        workspaceId,
        code: mapped.code,
      });
    };

    const persistAssistant = async (content: string, truncated: boolean, error?: string): Promise<string | undefined> => {
      try {
        const created = await this.prisma.aiMessage.create({
          data: {
            conversationId: conversation.id,
            role: "assistant",
            content: content || "(empty)",
            model: this.gateway.resolve(workspaceId).chatModel,
            ...(usage.in > 0 ? { tokensIn: usage.in } : {}),
            ...(usage.out > 0 ? { tokensOut: usage.out } : {}),
            latencyMs: Date.now() - startedAt,
            toolResults: { query: q, sources } as unknown as object,
            metadata: {
              feature: "rag",
              ...(truncated ? { truncated: true } : {}),
              ...(error ? { error } : {}),
            } as object,
          },
        });
        return created.id;
      } catch (err) {
        logger.error(`rag assistant persist failed: ${(err as Error).message}`);
        return undefined;
      }
    };

    const chatModel = this.gateway.resolve(workspaceId).chatModel;
    const startedAt = Date.now();
    const usage = { in: 0, out: 0 };

    // ---- no results: a deterministic empty-state answer (no chat call) ----
    if (ranked.length === 0) {
      const text = "I could not find any workspace content that answers this question. Try different wording.";
      for (const word of text.split(" ")) res.write(sseFrame({ type: "delta", text: word + " " }));
      const assistantId = await persistAssistant(text, false);
      finish({ type: "done", assistantMessageId: assistantId, userMessageId: userRow.id, reason: "ok" });
      return;
    }

    // ---- the answer (numbered context + question → the model) ----
    const contextBlocks = ranked.map((r, i) => renderRagContext(i + 1, r.hit.title || "(untitled)", r.body)).join("\n\n");
    const messages: Array<{ role: "system" | "user"; content: string }> = [
      { role: "system", content: RAG_SYSTEM_PROMPT },
      { role: "user", content: `${contextBlocks}\n\nQuestion: ${q}` },
    ];

    let content = "";
    let doneSeen = false;
    try {
      for await (const chunk of this.gateway.chat({
        workspaceId,
        userId,
        model: chatModel,
        messages,
        stream: true,
        capability: "rag",
        metadata: { mode: "rag" },
        signal: ac.signal,
      })) {
        if (timer) armTimeout();
        if (chunk.delta) {
          content += chunk.delta;
          res.write(sseFrame({ type: "delta", text: chunk.delta }));
        }
        if (chunk.done) {
          doneSeen = true;
          if (chunk.usage) {
            usage.in = chunk.usage.in;
            usage.out = chunk.usage.out;
            res.write(sseFrame({ type: "usage", tokensIn: chunk.usage.in, tokensOut: chunk.usage.out }));
          }
          const assistantId = await persistAssistant(content, false);
          finish({ type: "done", assistantMessageId: assistantId, userMessageId: userRow.id, reason: "ok" });
          return;
        }
      }
      // provider stream ended without a done chunk (defensive) — persist.
      if (!doneSeen) {
        const assistantId = await persistAssistant(content, false);
        finish({ type: "done", assistantMessageId: assistantId, userMessageId: userRow.id, reason: "ok" });
      }
    } catch (err) {
      if (clientClosed || ac.signal.aborted) {
        // the client went away (or the watchdog fired) — persist the partial.
        const assistantId = await persistAssistant(content, content.length > 0, "aborted");
        if (!clientClosed) {
          res.write(sseFrame({ type: "error", code: "TIMEOUT", message: "The AI request took too long.", retryable: true }));
        }
        finish({ type: "done", assistantMessageId: assistantId, userMessageId: userRow.id, reason: clientClosed ? "aborted" : "error" });
        return;
      }
      sendErrorDelta(err);
      const assistantId = await persistAssistant(content, content.length > 0, "error");
      finish({ type: "done", assistantMessageId: assistantId, userMessageId: userRow.id, reason: "error" });
    }
  }
}