/**
 * TASK-0122 — Ask AI streaming service (SSE).
 *
 * `POST /workspaces/:workspaceId/ai/conversations/:conversationId/messages`
 *
 * Contract (documented in `docs/api/ai.md`):
 *
 *   - response: `text/event-stream` of JSON `data:` frames:
 *       `{type:"delta",  text}`                — assistant content increments
 *       `{type:"tool_call", id, name, arguments}`
 *       `{type:"usage",  tokensIn, tokensOut}`
 *       `{type:"error",  code, message, retryable?, retryAfterSeconds?}`
 *       `{type:"done",   assistantMessageId?, userMessageId?, reason?}`
 *     (plus `: keepalive` comment frames every 10 s).
 *   - context: the client sends RESOURCE IDS only (`{pageId?, blockId?,
 *     databaseId?}` — the conversation's stored context, optionally
 *     overridden per message). `prompts.ts` is the single choke point that
 *     re-checks permissions and assembles the system-prompt section (caps:
 *     40 blocks / 20 rows; unviewable resources are dropped, never leaked).
 *   - budget: checked BEFORE any SSE header is written, so an exhausted
 *     budget is the canonical JSON `402 AI_BUDGET_EXCEEDED` envelope (the
 *     client's error branch renders the banner from it); errors that arise
 *     MID-stream (provider outage / timeout) are SSE `error` deltas with the
 *     same `{code, message, retryable}` shape — raw provider text is never
 *     surfaced (gateway normalization + `mapAiError`).
 *   - rate limit: 30 messages / min per user (Redis sliding window) →
 *     `429 RATE_LIMITED` JSON (pre-stream) carrying `retryAfterSeconds`.
 *   - cancellation: the client's `AbortController` (fetch abort on panel
 *     close / the "Stop" button) propagates to `req.signal` → the provider
 *     stream is cancelled (TASK-0121 abort plumbing) and a PARTIAL assistant
 *     message is persisted with `metadata.truncated = true`.
 *   - retry: `body.retryOf = <userMessageId>` re-runs the turn WITHOUT
 *     duplicating the persisted user message (the UI's Retry button sends it
 *     after a mid-stream error).
 *   - auto-title: the first user message (first 60 chars) titles the
 *     conversation when it has no title yet.
 *   - persistence: user row (except on retry) + assistant row carrying
 *     `tokensIn/tokensOut/latencyMs/toolCalls` (the usage-ledger row).
 *
 * Pre-stream failures (ownership / budget / rate limit) throw `ApiError`
 * BEFORE a single byte is written → the global filter's JSON envelope.
 * Mid-stream failures are in-band `error` deltas.
 */
import { Inject, Injectable } from "@nestjs/common";
import type { Request, Response } from "express";

import type { PrismaClient } from "@prisma/client";
import { PrismaService } from "../../common/prisma.service.js";
import { ApiError } from "../../common/errors.js";
import { logger } from "../../common/logger.js";
import { assertCan } from "../../lib/authorize.js";
import {
  AI_GATEWAY,
  mapAiError,
  primeWorkspaceCache,
  type AiBudgetGuard,
} from "./gateway-provider.js";
import { AiBudgetGuardService } from "./budget-guard.service.js";
import { AI_CALLS_PER_USER_LIMIT, AiRateLimitService } from "./ai-rate-limit.js";
import { assembleContext, BASE_SYSTEM_PROMPT, type AiContextRef } from "./prompts.js";
import type { AiGateway, ChatChunk } from "@factory/ai";

/** 30 AI messages / min per user (spec §Security; 429 RATE_LIMITED).
 *  TASK-0123: the cap now lives in the shared `ai.calls` limiter (chat and
 *  writing count against the SAME window) — re-exported for the integration
 *  tests that pin the number. */
export const AI_CHAT_PER_USER_LIMIT = AI_CALLS_PER_USER_LIMIT;
/** Client-side watchdog: "took too long" if no frame arrives within 30 s. */
const AI_CHAT_TIMEOUT_MS = 30_000;
const KEEPALIVE_MS = 10_000;

export interface AiChatStreamInput {
  content: string;
  context?: AiContextRef;
  /** Re-send this (already-persisted) user message without duplicating it. */
  retryOf?: string;
}

/** One SSE frame (the documented wire contract). */
export type AiSseEvent =
  | { type: "delta"; text: string }
  | { type: "tool_call"; id: string; name: string; arguments: Record<string, unknown> }
  | { 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: AiSseEvent): string {
  return `data: ${JSON.stringify(event)}\n\n`;
}

@Injectable()
export class AiChatService {
  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,
  ) {}

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

  /**
   * Stream one Ask AI turn. Pre-stream errors throw (JSON envelope);
   * mid-stream errors are `error` deltas followed by a terminal `done`.
   */
  async sendMessage(
    workspaceId: string,
    userId: string,
    conversationId: string,
    input: AiChatStreamInput,
    req: Request,
    res: Response,
  ): Promise<void> {
    // ---- pre-stream gates (canonical JSON envelope on failure) ----
    await assertCan(this.prisma, { workspaceId, userId }, "VIEW");
    const conversation = await this.prisma.aiConversation.findUnique({ where: { id: conversationId } });
    // Own-conversation scoping server-side: another member's conversation is
    // a 404 (existence hidden across the boundary — error-model convention).
    if (!conversation || conversation.userId !== userId || conversation.workspaceId !== workspaceId || conversation.deletedAt) {
      throw ApiError.notFound("Conversation not found.");
    }
    await primeWorkspaceCache(this.prisma, workspaceId);
    try {
      await this.guard.check(workspaceId, undefined, req);
    } catch (err) {
      throw mapAiError(err);
    }

    // per-user message rate limit (sliding window; degrades OPEN when Redis
    // is down — availability over strictness, same policy as the form limiter)
    const rl = await this.checkRateLimit(userId);
    if (!rl.allowed) {
      throw new ApiError("RATE_LIMITED", 429, "AI message limit reached.", [
        { reason: "ai_message_rate_limited", retryAfterSeconds: rl.retryAfterSeconds },
      ]);
    }

    // ---- context assembly (permission re-check + caps; never leaks) ----
    const ref: AiContextRef = {
      ...(input.context ?? (conversation.context as AiContextRef | null) ?? {}),
    };
    let ctx;
    try {
      ctx = await assembleContext(this.prisma, workspaceId, userId, ref);
    } catch (err) {
      logger.warn(`ai context assembly failed for ${userId}: ${(err as Error).message}`);
      ctx = { systemPrompt: undefined, note: "Context is unavailable for this message.", dropped: ["context"] };
    }

    // ---- persist the user message (or resolve the retry) ----
    let userRow;
    if (input.retryOf) {
      const found = await this.prisma.aiMessage.findFirst({
        where: { conversationId: conversation.id, id: input.retryOf, role: "user" },
      });
      if (!found) throw ApiError.notFound("Message not found.");
      userRow = found;
    } else {
      userRow = await this.prisma.aiMessage.create({
        data: { conversationId: conversation.id, role: "user", content: input.content },
      });
      const needsTitle = !conversation.title;
      const updates: { updatedAt: Date; title?: string } = { updatedAt: new Date() };
      if (needsTitle) updates.title = input.content.slice(0, 60);
      await this.prisma.aiConversation.update({ where: { id: conversation.id }, data: updates });
    }

    // ---- 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);

    // abort plumbing: client fetch abort → cancel the provider stream.
    // `res`'s "close" fires on BOTH normal completion and client departure —
    // the `writableEnded` check distinguishes them (on a clean end,
    // "finish" already flipped it; on a hang-up it is still false). This is
    // the reliable signal: `req`'s "close" can fire late/never for a
    // mid-body fetch abort.
    const ac = new AbortController();
    let clientClosed = false;
    let reason = "ok";
    const onResClose = (): void => {
      if (res.writableEnded) return; // normal completion
      clientClosed = true;
      ac.abort();
    };
    res.on("close", onResClose);
    const keepalive = setInterval(() => {
      if (!res.writableEnded) res.write(": keepalive\n\n");
    }, KEEPALIVE_MS);
    // client-side watchdog: no provider frame for 30 s → "took too long".
    let timer: ReturnType<typeof setTimeout> | undefined;
    const armTimeout = (): void => {
      if (timer) clearTimeout(timer);
      timer = setTimeout(() => {
        ac.abort();
        reason = "timeout";
      }, AI_CHAT_TIMEOUT_MS);
    };
    armTimeout();

    let content = "";
    const toolCalls: Array<{ id: string; name: string; arguments: Record<string, unknown> }> = [];
    let usage = { in: 0, out: 0 };
    let finished = false;
    const chatModel = this.gateway.resolve(workspaceId).chatModel;
    const startedAt = Date.now();

    const messages: Array<{ role: "system" | "user" | "assistant"; content: string }> = [
      { role: "system", content: BASE_SYSTEM_PROMPT },
    ];
    if (ctx.systemPrompt) messages.push({ role: "system", content: ctx.systemPrompt });
    const history = await this.prisma.aiMessage.findMany({
      where: { conversationId: conversation.id, id: { not: userRow.id } },
      orderBy: { createdAt: "desc" },
      take: 20,
      select: { role: true, content: true },
    });
    for (const m of [...history].reverse()) {
      if (m.role === "user" || m.role === "assistant") messages.push({ role: m.role, content: m.content });
    }
    messages.push({ role: "user", content: input.content });

    // context-unavailable note (permission drop) — the FIRST delta, so the
    // panel can render it even when the provider call fails before deltas.
    if (ctx.note) res.write(sseFrame({ type: "delta", text: `> ${ctx.note}\n\n` }));

    const persistAssistant = async (opts: { content: string; truncated: boolean; error?: string }) => {
      try {
        // Only persist when there is visible content — an empty assistant
        // row (abort before the first delta / provider error pre-output)
        // would pollute the history; the user row + the error banner cover
        // the UI state in those cases.
        if (!opts.content) return undefined;
        return await this.prisma.aiMessage.create({
          data: {
            conversationId: conversation.id,
            role: "assistant",
            content: opts.content,
            model: chatModel,
            ...(usage.in > 0 ? { tokensIn: usage.in } : {}),
            ...(usage.out > 0 ? { tokensOut: usage.out } : {}),
            latencyMs: Date.now() - startedAt,
            ...(toolCalls.length > 0 ? { toolCalls: toolCalls as unknown as object } : {}),
            metadata: {
              feature: "chat",
              ...(opts.truncated ? { truncated: true } : {}),
              ...(opts.error ? { error: opts.error } : {}),
            } as object,
          },
        });
      } catch (err) {
        // persistence must never crash the stream teardown
        logger.error(`ai chat assistant persist failed: ${(err as Error).message}`);
        return undefined;
      }
    };
    const finish = (event: AiSseEvent): 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>;
      // `retryable` rides on the mapped details (the gateway's normalization
      // sets it for 429/5xx/timeout; provider 4xx and budget 402 → false).
      // A rate limit is inherently transient → retryable (the UI shows a
      // countdown from `retryAfterSeconds`).
      const retryable = detail.retryable === true || detail.retryAfterSeconds != null || mapped.code === "RATE_LIMITED";
      const event: AiSseEvent = {
        type: "error",
        code: mapped.code,
        // raw provider text is never surfaced — mapped.message is the
        // normalized (no-secret) message from the gateway error map.
        message: mapped.message,
        ...(retryable ? { retryable: true } : {}),
        ...(typeof detail.retryAfterSeconds === "number" ? { retryAfterSeconds: detail.retryAfterSeconds } : {}),
      };
      res.write(sseFrame(event));
      logger.info(
        `ai chat stream error workspace=${workspaceId} conversation=${conversation.id} code=${mapped.code} retryable=${retryable}`,
        { event: "ai.chat.error", workspaceId, conversationId, code: mapped.code, retryable },
      );
    };

    try {
      for await (const chunk of this.gateway.chat({
        workspaceId,
        userId,
        capability: "chat",
        messages,
        metadata: { conversationId: conversation.id },
        signal: ac.signal,
      })) {
        if (timer) armTimeout();
        if (chunk.delta) {
          content += chunk.delta;
          res.write(sseFrame({ type: "delta", text: chunk.delta }));
        }
        if (chunk.toolCallDelta) {
          toolCalls.push({
            id: chunk.toolCallDelta.id,
            name: chunk.toolCallDelta.name,
            arguments: chunk.toolCallDelta.arguments,
          });
          res.write(sseFrame({ type: "tool_call", ...chunk.toolCallDelta }));
        }
        if (chunk.done) {
          if (chunk.usage) {
            usage = { in: chunk.usage.in, out: chunk.usage.out };
            res.write(sseFrame({ type: "usage", tokensIn: usage.in, tokensOut: usage.out }));
          }
          const row = await persistAssistant({ content, truncated: false });
          finish({ type: "done", assistantMessageId: row?.id, userMessageId: userRow.id, reason: "ok" });
        }
      }
      // provider stream ended without a done chunk (defensive) — persist
      // once (the done-chunk branch above already finishes when it fires).
      if (!finished) {
        const row = await persistAssistant({ content, truncated: false });
        finish({ type: "done", assistantMessageId: row?.id, userMessageId: userRow.id, reason: "ok" });
      }
    } catch (err) {
      if (timer) clearTimeout(timer);
      clearInterval(keepalive);
      res.off("close", onResClose);

      if (reason === "timeout") {
        // idle watchdog: the provider stopped sending frames (30 s) — the
        // client's fetch is still connected; tell it in-band and persist the
        // partial (metadata.truncated).
        const row = await persistAssistant({ content, truncated: content.length > 0, error: "timeout" });
        if (!clientClosed) {
          res.write(sseFrame({ type: "error", code: "TIMEOUT", message: "The AI request took too long.", retryable: true }));
        }
        finish({ type: "done", assistantMessageId: row?.id, userMessageId: userRow.id, reason: "error" });
        return;
      }
      if (clientClosed || ac.signal.aborted) {
        // client hang-up (panel closed / tab switched) or the "Stop" button:
        // the provider stream was cancelled; persist the partial with
        // metadata.truncated (the "Stop" case still gets the terminal done).
        const row = await persistAssistant({ content, truncated: content.length > 0, error: "aborted" });
        if (!clientClosed && !res.writableEnded) {
          res.write(sseFrame({ type: "done", assistantMessageId: row?.id, userMessageId: userRow.id, reason: "aborted" }));
          res.end();
        }
        return;
      }
      // provider failure (normalized — raw provider text never surfaces):
      // error delta + terminal done; persist any partial content.
      sendErrorDelta(err);
      const row = await persistAssistant({ content, truncated: content.length > 0, error: "provider_error" });
      finish({ type: "done", assistantMessageId: row?.id, userMessageId: userRow.id, reason: "error" });
    }
  }

  /**
   * 30 AI messages / min per user — the SHARED `ai.calls` window
   * (`ai:calls:rl:<userId>`, TASK-0123 `AiRateLimitService`): chat and the
   * writing tools count against the SAME limiter. Degrades OPEN when Redis is
   * unavailable (same policy as the form limiter).
   */
  private checkRateLimit(userId: string): Promise<{ allowed: boolean; remaining: number; retryAfterSeconds: number }> {
    return this.rateLimit.check(userId);
  }
}