/**
 * TASK-0126 — AI page generation (prompt → structured tool calls executed
 * through the domain services with the user's permissions).
 *
 *   POST /workspaces/:workspaceId/ai/generate   (body `{ prompt, parentId? }`,
 *   header `Idempotency-Key?`)   → SSE progress stream.
 *
 * The wire contract (docs/api/ai.md §10):
 *
 *   tool* (per step: {type:"tool", name, status:"started"|"done", ref?})
 *     → done { pageId, created:{pages,blocks,databases,relations}, tokensIn, tokensOut }
 *   or an in-band error (a step failed / the plan was invalid / timeout /
 *   budget / rate-limit) + a terminal done { reason }.
 *
 * Security (spec §Security, AGENTS #7/#18/#19/#20/#27):
 *   - pre-stream: non-member → 404 (fail-closed, like the pages API + the
 *     RAG fix), no-EDIT → 403 FORBIDDEN, budget → 402 AI_BUDGET_EXCEEDED
 *     (the guard also fires the 24 h-throttled admin notify + audit),
 *     rate-limit → 429, validation → 400;
 *   - the LLM emits a PLAN (structured tool calls) — its UNTRUSTED output,
 *     validated by `parseGeneratePlan` (zod) before ANY write;
 *   - the executor runs the plan through the DOMAIN services with the
 *     REQUESTING user's context (the permission-context hook rejects an
 *     elevated context); a failing step rolls back the partial tree;
 *   - total capped at 120 s (injectable `timeoutMs`; the test overrides to
 *     2 s) → in-band `ai_timeout`; the client's SSE abort → the executor stops
 *     after the current step, rolls back, and emits `cancelled`;
 *   - cost: one hidden per-call conversation (`metadata.feature="generate"`)
 *     with full token counts; audit `ai.page_generated` / `ai.generation_failed`;
 *   - idempotency: the `Idempotency-Key` header memoizes the created page id
 *     (Redis, 24 h) so a retried cancelled generation does not double-create.
 */
import { Inject, Injectable } from "@nestjs/common";
import type { Request, Response } from "express";

import type { PrismaClient } from "@prisma/client";
import { parseGeneratePlan } from "@factory/ai";
import { PrismaService } from "../../../common/prisma.service.js";
import { AuditService } from "../../../common/audit.service.js";
import { ApiError } from "../../../common/errors.js";
import { logger } from "../../../common/logger.js";
import { actorMembership, assertCan } from "../../../lib/authorize.js";
import { RedisService } from "../../../common/redis.service.js";
import {
  AI_GATEWAY,
  mapAiError,
  primeWorkspaceCache,
} from "../gateway-provider.js";
import { GENERATE_PLAN_SYSTEM_PROMPT, GENERATE_PLAN_TOOL, normalizeGeneratePlanArgs } from "../prompts.js";
import { AiBudgetGuardService } from "../budget-guard.service.js";
import { AiRateLimitService } from "../ai-rate-limit.js";
import { PageService } from "../../pages/page.service.js";
import { BlockService } from "../../blocks/block.service.js";
import { DatabaseService } from "../../databases/database.service.js";
import { DatabaseQueryService } from "../../databases/query.service.js";
import type { AiGateway } from "@factory/ai";
import {
  GenerationAborted,
  GenerationExecutor,
  GenerationStepFailed,
  type StepName,
} from "./executor.js";
import type { AiGenerateInput } from "./generate.dto.js";

/** The hidden per-call conversation's title (never lists in the panel). */
export const GENERATE_CONV_PREFIX = "generate:";
/** The 24 h idempotency memo TTL (a cancelled run's key lives a day). */
const IDEM_TTL_SECONDS = 24 * 60 * 60;
/** The server-side total cap (the spec's 120 s). Read PER CALL (not at boot)
 *  so the integration/E2E timeout suite can override `AI_GENERATE_TIMEOUT_MS`
 *  to 2 s without a service-instance seam. */
const DEFAULT_TIMEOUT_MS = 120_000;
/** SSE comment-keepalive interval — well under any proxy/browser idle cap. */
const KEEPALIVE_MS = 15_000;

/** The SSE frame contract (mirrors the web client's `GenerateFrame` — the
 *  frontend treats exactly one of `done` / `cancelled` / `error` as the
 *  terminal frame per stream; `tool` frames precede it). */
export type AiGenerateSseEvent =
  | { type: "tool"; name: StepName; status: "started" | "done"; ref?: string }
  | {
      type: "done";
      pageId: string;
      created: { pages: string[]; blocks: number; databases: string[]; relations: number };
      tokensIn?: number;
      tokensOut?: number;
    }
  | { type: "cancelled" }
  | { type: "error"; code: string; message: string; reason?: string; retryable?: boolean };

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

/** True when the error is the gateway's caller-abort (a mid-stream abort
 *  maps to a shared `AI_ERROR` with `details.reason === "ai_aborted"`). Used
 *  to distinguish a TIMEOUT / client-cancel that hit the plan fetch (before
 *  the executor created anything) from a genuine provider failure. */
function isAiAborted(err: unknown): boolean {
  const d = (err as { details?: unknown } | null)?.details;
  if (d && typeof d === "object") {
    if (Array.isArray(d)) return d.some((x) => (x as { reason?: unknown })?.reason === "ai_aborted");
    return (d as { reason?: unknown }).reason === "ai_aborted";
  }
  return false;
}

@Injectable()
export class AiGenerateService {
  /** The total generation cap (per call; the timeout suite overrides the env). */
  get timeoutMs(): number {
    const envMs = Number(process.env.AI_GENERATE_TIMEOUT_MS);
    return Number.isFinite(envMs) && envMs > 0 ? envMs : DEFAULT_TIMEOUT_MS;
  }

  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(RedisService) private readonly redis: RedisService,
    @Inject(PageService) private readonly pages: PageService,
    @Inject(BlockService) private readonly blocks: BlockService,
    @Inject(DatabaseService) private readonly databases: DatabaseService,
    @Inject(DatabaseQueryService) private readonly records: DatabaseQueryService,
    @Inject(AuditService) private readonly audit: AuditService,
  ) {}

  /**
   * The generate endpoint. Pre-stream failures throw (→ JSON envelope); the
   * SSE stream carries the progress `tool` frames + the terminal `done` /
   * in-band `error`.
   */
  async generate(
    workspaceId: string,
    userId: string,
    input: AiGenerateInput,
    idempotencyKey: string | undefined,
    req: Request,
    res: Response,
  ): Promise<void> {
    // ---- pre-stream gates (before any SSE byte) ----
    // Non-member → 404 (fail-closed: hide the workspace's existence, like the
    // pages API + the RAG gate-order fix). A member without EDIT → 403.
    // TASK-0127: actor-aware — an API key of the workspace is a member-
    // equivalent (its scope, not the creator's membership, decides).
    const membership = await actorMembership(this.prisma, userId, workspaceId, "EDIT");
    if (!membership) throw ApiError.notFound("Workspace not found.");
    await assertCan(this.prisma, { workspaceId, userId }, "EDIT");

    await this.rateLimitCheck(userId);

    // Budget pre-check (before ANY model call) + the 24 h admin notify/audit.
    await primeWorkspaceCache(this.prisma, workspaceId);
    try {
      await this.guardService.guard.check(workspaceId, undefined, req);
    } catch (err) {
      throw mapAiError(err);
    }

    // Idempotency: a memoized (cancelled/failed-then-retried) key replays the
    // created page without re-running the executor (no double-create).
    if (idempotencyKey) {
      const memo = await this.readIdem(workspaceId, userId, idempotencyKey);
      if (memo) {
        this.sseHeaders(res);
        const finish = (e: AiGenerateSseEvent): void => {
          if (!res.writableEnded) {
            res.write(sseFrame(e));
            res.end();
          }
        };
        finish({
          type: "done",
          pageId: memo.pageId,
          created: memo.created,
          tokensIn: memo.tokensIn,
          tokensOut: memo.tokensOut,
        });
        return;
      }
    }

    // ---- SSE ----
    this.sseHeaders(res);
    const ac = new AbortController();
    let clientClosed = false;
    // The total-timeout watchdog fired (distinct from a client cancel): the
    // gateway surfaces a mid-stream abort as a shared `AI_ERROR`/`ai_aborted`
    // (NOT via `signal.reason` — that's getter-only in strict mode), so the
    // service classifies timeouts by THIS flag, not the error type.
    let timedOut = false;
    const onClose = (): void => {
      if (res.writableEnded) return;
      clientClosed = true;
      ac.abort(); // the executor stops after the current step + rolls back
    };
    res.on("close", onClose);
    // The total cap (the spec's 120 s; the test overrides `this.timeoutMs`).
    const timer = setTimeout(() => {
      timedOut = true;
      ac.abort();
    }, this.timeoutMs);
    // SSE keepalive: the provider's thinking phase can emit ZERO bytes for
    // minutes (a successful run has been observed silent for ~150 s), and a
    // silent stream looks dead to the client (the dialog sat on "Drafting the
    // page…" with no way to distinguish "working" from "stuck"). A comment
    // frame every 15 s is invisible to the client's SSE parser (it only
    // reads `data:` lines) but resets every proxy/browser idle timer and the
    // client's own idle watchdog can now detect a GENUINELY dead stream.
    const keepAlive = setInterval(() => {
      if (!clientClosed && !res.writableEnded) res.write(": keepalive\n\n");
    }, KEEPALIVE_MS);
    const cleanup = (): void => {
      clearTimeout(timer);
      clearInterval(keepAlive);
      res.off("close", onClose);
    };
    const finish = (event: AiGenerateSseEvent): void => {
      cleanup();
      if (!res.writableEnded) {
        res.write(sseFrame(event));
        res.end();
      }
    };

    // The hidden per-call conversation (cost ledger; never lists in the panel).
    const conversation = await this.prisma.aiConversation.create({
      data: {
        workspaceId,
        userId,
        title: GENERATE_CONV_PREFIX,
        context: { feature: "generate", parentId: input.parentId ?? null } as object,
      },
    });
    await this.prisma.aiMessage.create({
      data: { conversationId: conversation.id, role: "user", content: `generate: "${input.prompt.slice(0, 200)}"` },
    });

    let usage = { in: 0, out: 0 };
    // The resolved chat model, captured ONCE now — before the (minutes-long)
    // model call. Recorded decision (2025, root cause of an API crash): the
    // earlier code re-resolved via `this.gateway.resolve(workspaceId)` in the
    // post-stream persist calls, which re-reads the 30 s TTL workspace-row
    // cache; a model call longer than 30 s expired it, `resolve` threw 404
    // ("Workspace not found"), and inside the catch block that throw escaped
    // into the exception filter AFTER the SSE headers were already sent →
    // uncaught `ERR_HTTP_HEADERS_SENT` → the API process died. The model name
    // is a static property of the resolved config for the lifetime of this
    // call, so capturing it early is exact.
    const chatModel = this.gateway.resolve(workspaceId).chatModel;
    try {
      // 1) The model emits the PLAN (one tool call; the mock is deterministic).
      const plan = await this.chatPlan(workspaceId, userId, input.prompt, ac.signal);
      usage = plan.usage;

      // 2) Validate the UNTRUSTED plan (zod) before ANY write.
      const parsed = parseGeneratePlan(plan.args);
      if (!parsed.ok) {
        // Observability (AGENTS #26): the plan is untrusted model output —
        // log WHY it failed so a model-shape regression is diagnosable.
        logger.warn(
          `ai generate: invalid plan from model ${chatModel}: ` +
            parsed.errors.issues.map((i) => `${i.path.join(".") || "(root)"} ${i.message}`).slice(0, 8).join("; "),
        );
        await this.persistAssistant(conversation.id, chatModel, usage, {
          feature: "generate",
          error: "ai_invalid_plan",
        });
        // Terminal in-band error (no resource was created — nothing to roll
        // back; the plan never ran).
        finish({
          type: "error",
          code: "UNPROCESSABLE_ENTITY",
          message: "The AI returned an invalid page plan.",
          reason: "ai_invalid_plan",
        });
        return;
      }

      // 3) Execute the plan through the domain services (the user's context).
      const executor = new GenerationExecutor(
        { pages: this.pages, blocks: this.blocks, databases: this.databases, records: this.records, prisma: this.prisma },
        {
          workspaceId,
          userId,
          expectedUserId: userId,
          req,
          signal: ac.signal,
          onStep: (name, status, ref) => {
            if (!clientClosed && !res.writableEnded) res.write(sseFrame({ type: "tool", name, status, ...(ref ? { ref } : {}) }));
          },
        },
      );
      const result = await executor.execute(parsed.plan, input.parentId);

      // 4) Success: audit + memo the idempotency key + persist usage.
      await this.audit.audit(req, req?.auth, {
        action: "ai.page_generated",
        resourceType: "page",
        resourceId: result.pageId,
        workspaceId,
        metadata: { steps: parsed.plan.tools.length, created: result.created },
      });
      await this.persistAssistant(conversation.id, chatModel, usage, {
        feature: "generate",
        pageId: result.pageId,
        created: result.created,
      });
      await this.writeIdem(workspaceId, userId, idempotencyKey, {
        pageId: result.pageId,
        created: result.created,
        tokensIn: usage.in,
        tokensOut: usage.out,
      });
      finish({
        type: "done",
        pageId: result.pageId,
        created: result.created,
        tokensIn: usage.in,
        tokensOut: usage.out,
      });
    } catch (err) {
      // ---- terminal classification (exactly ONE frame per path) ----
      // Total-timeout watchdog fired (or the gateway's abort surfaced before
      // the executor ran, while the client is still here): the partial tree
      // is rolled back (by the executor when it was running, by the abort
      // otherwise) and the run did NOT complete — a retryable in-band error.
      if (timedOut || (isAiAborted(err) && !clientClosed)) {
        await this.audit.audit(req, req?.auth, {
          action: "ai.generation_failed",
          resourceType: "page",
          resourceId: conversation.id,
          workspaceId,
          metadata: { failedStep: "timeout", reason: "ai_timeout" },
        });
        if (!clientClosed) {
          finish({
            type: "error",
            code: "SERVICE_UNAVAILABLE",
            message: "Generation took too long and was cancelled (partial work rolled back).",
            reason: "ai_timeout",
            retryable: true,
          });
        } else {
          cleanup();
        }
        return;
      }
      if (err instanceof GenerationAborted) {
        // The client went away (or cancelled): if the socket is still open
        // emit the terminal `cancelled`; otherwise just release resources.
        if (!clientClosed) finish({ type: "cancelled" });
        else cleanup();
        return;
      }
      if (err instanceof GenerationStepFailed) {
        // The executor already rolled back the partial tree.
        // Observability (AGENTS #26): the step error (usually the model's
        // non-conforming arg shape) is the thing to diagnose next.
        const stepMsg = (err.error as Error | null)?.message ?? String(err.error);
        logger.warn(`ai generate: step ${err.failedStep} failed: ${stepMsg.slice(0, 500)}`);
        await this.audit.audit(req, req?.auth, {
          action: "ai.generation_failed",
          resourceType: "page",
          resourceId: conversation.id,
          workspaceId,
          // Truncated: the audit row is also user-visible; keep it short.
          metadata: { failedStep: err.failedStep, reason: stepMsg.slice(0, 300) },
        });
        await this.persistAssistant(conversation.id, chatModel, usage, {
          feature: "generate",
          error: "generation_failed",
          failedStep: err.failedStep,
        });
        if (!clientClosed) {
          finish({
            type: "error",
            code: "UNPROCESSABLE_ENTITY",
            message: `Page generation failed at step ${err.failedStep} (partial work rolled back).`,
            reason: "generation_failed",
            retryable: true,
          });
        } else {
          cleanup();
        }
        return;
      }
      if (err instanceof ApiError && err.code === "RATE_LIMITED") {
        // Provider 429 (the gateway's normalizeProviderError). Pre-fix this
        // fell through to the generic `INTERNAL` frame below, which the
        // client rendered as an opaque failure with no retry guidance; now the
        // dialog shows the rate-limit banner + a Retry affordance.
        logger.warn(`ai generate: provider rate-limited: ${(err as Error).message}`);
        if (!clientClosed) {
          finish({
            type: "error",
            code: "RATE_LIMITED",
            message: err.message || "AI provider rate limit reached. Try again shortly.",
            reason: "rate_limited",
            retryable: true,
          });
        } else {
          cleanup();
        }
        return;
      }
      // A gateway/network error that slipped through (defensive).
      logger.warn(`ai generate failed: ${(err as Error).message}`);
      if (!clientClosed) {
        finish({ type: "error", code: "INTERNAL", message: "Page generation failed.", reason: "ai_error", retryable: true });
      } else {
        cleanup();
      }
    }
  }

  // ------------------------------------------------------------- helpers ---

  private async rateLimitCheck(userId: string): Promise<void> {
    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 },
      ]);
    }
  }

  private sseHeaders(res: Response): void {
    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 LLM's plan (one tool call; the mock is the deterministic plan).
   *
   * The `generatePlan` tool is FORCED (`toolChoice`) — recorded decision
   * (2025, root cause of the "Generation took too long" failure): with only a
   * system-prompt instruction, the OSS reasoning model streams minutes of
   * prose and never emits the plan, so the 120 s watchdog always fired. The
   * adapter sends `tool_choice: {type:"function", function:{name}}` for this.
   * The model's args are UNTRUSTED and get a small, total normalization pass
   * (`normalizeGeneratePlanArgs`) before the strict zod validation in the
   * caller (repairs the observed tools-as-string + `$page`-placeholder
   * shapes; see prompts.ts).
   */
  private async chatPlan(
    workspaceId: string,
    userId: string,
    prompt: string,
    signal: AbortSignal,
  ): Promise<{ args: unknown; usage: { in: number; out: number } }> {
    const messages: Array<{ role: "system" | "user"; content: string }> = [
      { role: "system", content: GENERATE_PLAN_SYSTEM_PROMPT },
      { role: "user", content: prompt },
    ];
    let args: unknown;
    let usage = { in: 0, out: 0 };
    for await (const chunk of this.gateway.chat({
      workspaceId,
      userId,
      capability: "chat",
      messages,
      tools: [GENERATE_PLAN_TOOL],
      toolChoice: { name: "generatePlan" },
      metadata: { mode: "generate" },
      signal,
    })) {
      if (chunk.toolCallDelta) args = chunk.toolCallDelta.arguments;
      if (chunk.done && chunk.usage) usage = { in: chunk.usage.in, out: chunk.usage.out };
    }
    if (args === undefined) {
      throw ApiError.unprocessable("The model returned no plan.", [{ reason: "ai_invalid_plan", issue: "no tool call" }]);
    }
    return { args: normalizeGeneratePlanArgs(args), usage };
  }

  private async persistAssistant(
    conversationId: string,
    model: string,
    usage: { in: number; out: number },
    metadata: Record<string, unknown>,
  ): Promise<string | undefined> {
    try {
      const row = await this.prisma.aiMessage.create({
        data: {
          conversationId,
          role: "assistant",
          content: metadata.error ? `error: ${String(metadata.error)}` : "ok",
          model,
          ...(usage.in > 0 ? { tokensIn: usage.in } : {}),
          ...(usage.out > 0 ? { tokensOut: usage.out } : {}),
          metadata: metadata as object,
        },
      });
      return row.id;
    } catch (err) {
      logger.error(`ai generate assistant persist failed: ${(err as Error).message}`);
      return undefined;
    }
  }

  private idemKey(workspaceId: string, userId: string, key: string): string {
    return `ai:generate:idem:${workspaceId}:${userId}:${key}`;
  }

  private async readIdem(
    workspaceId: string,
    userId: string,
    key: string,
  ): Promise<{ pageId: string; created: { pages: string[]; blocks: number; databases: string[]; relations: number }; tokensIn: number; tokensOut: number } | null> {
    try {
      const raw = await this.redis.getRedis().get(this.idemKey(workspaceId, userId, key));
      if (!raw) return null;
      return JSON.parse(raw);
    } catch {
      return null;
    }
  }

  private async writeIdem(
    workspaceId: string,
    userId: string,
    key: string | undefined,
    value: { pageId: string; created: { pages: string[]; blocks: number; databases: string[]; relations: number }; tokensIn: number; tokensOut: number },
  ): Promise<void> {
    if (!key) return;
    try {
      await this.redis.getRedis().set(this.idemKey(workspaceId, userId, key), JSON.stringify(value), "EX", IDEM_TTL_SECONDS);
    } catch (err) {
      logger.warn(`ai generate idem store failed: ${(err as Error).message}`);
    }
  }
}