/**
 * TASK-0124 — Database AI (generate database / extract data / generate formula).
 *
 * Endpoints (documented in `docs/api/ai.md`):
 *
 *   POST /workspaces/:workspaceId/ai/databases/generate     (SSE)
 *       body: { prompt, parentId? }. Frames: `tool_call` (the validated
 *       `createDatabase` call), `usage`, `error`, `done`. Pre-stream failures
 *       (permission 403, rate 429, budget 402) are a canonical JSON envelope;
 *       in-stream failures (invalid structure after the retry-once, provider
 *       outage) are in-band `error` + terminal `done {reason:"error"}`.
 *       `done` on success carries `{ databaseId, propertyIds, recordIds,
 *       tokensIn, tokensOut, assistantMessageId }`.
 *
 *   POST /workspaces/:workspaceId/ai/databases/extract      (SSE)
 *       body: { databaseId, text (≤50k) }. The model emits `insertRecords`
 *       (propertyId-keyed, ≤200). Unknown markdown columns → `ignoredColumns`
 *       (in the `done` frame).
 *
 *   POST /workspaces/:workspaceId/ai/databases/extract/preview  (JSON, no write)
 *       body: { databaseId, text } → the deterministic column→property mapping
 *       + row count (the UI's live preview; no LLM call, no budget).
 *
 *   POST /workspaces/:workspaceId/ai/databases/formulas     (JSON, non-streamed)
 *       body: { databaseId, description, name?, sampleRowIds? } → the model
 *       emits `generateFormula`; the expression is validated by the safe
 *       formula engine (parse + type-check + eval on ≤5 rows) BEFORE saving.
 *       Invalid after the retry-once → 422 `formula_invalid` (no Formula row).
 *
 * Security (spec §Security, AGENTS #7/18/19):
 *   - every write goes through the normal domain services (permissions,
 *     audit, realtime, search) — never raw SQL, never elevated permissions;
 *   - the tool-call output is UNTRUSTED: validated by zod before use;
 *   - formulas are only parsed/evaluated by the safe AST engine (zero eval,
 *     AGENTS #18);
 *   - hidden per-call conversations (`db:generate` / `db:extract` /
 *     `db:formula` titles, `metadata.feature` db.*) never list in the panel.
 */
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 { AuditService } from "../../common/audit.service.js";
import { ApiError, type ErrorCode, type ErrorDetail } 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 { AiRateLimitService } from "./ai-rate-limit.js";
import {
  type AiExtractInput,
  type AiExtractPreviewInput,
  type AiGenerateDatabaseInput,
  type AiGenerateFormulaInput,
} from "./database-ai.dto.js";
import {
  CREATE_DATABASE_TOOL,
  DB_EXTRACT_CONV_PREFIX,
  DB_FORMULA_CONV_PREFIX,
  DB_FORMULA_SYSTEM_PROMPT,
  DB_GEN_CONV_PREFIX,
  DB_GEN_SYSTEM_PROMPT,
  DB_EXTRACT_SYSTEM_PROMPT,
  GENERATE_FORMULA_TOOL,
  INSERT_RECORDS_TOOL,
  renderExtractPrompt,
  renderFormulaPrompt,
  renderGeneratePrompt,
  validateCreateDatabase,
  validateInsertRecords,
  previewExtractMapping,
  type DbPropMeta,
} from "./db-prompts.js";
import { lowerPropertyType, DB_FORMULA_MAX_SAMPLE_ROWS } from "@factory/ai";
import { DatabaseService } from "../databases/database.service.js";
import { DatabaseQueryService } from "../databases/query.service.js";
import { PageService } from "../pages/page.service.js";
import type { AiGateway } from "@factory/ai";

const TIMEOUT_MS = 60_000;

export type AiDatabaseSseEvent =
  | { 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; reason?: string; retryable?: boolean }
  | {
      type: "done";
      assistantMessageId?: string;
      reason?: "ok" | "aborted" | "error";
      databaseId?: string;
      propertyIds?: string[];
      recordIds?: string[];
      ignoredColumns?: string[];
      formulaId?: string;
      expression?: string;
      returnType?: string;
      explanation?: string;
      samples?: Array<{ recordId: string; value: unknown; error?: string }>;
      tokensIn?: number;
      tokensOut?: number;
    };

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

/** A record's stored property values → the formula engine's name→value map. */
function recordToFormulaProps(
  properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }>,
): Record<string, unknown> {
  const out: Record<string, unknown> = {};
  for (const rp of properties) {
    const { name, type, config } = rp.property;
    const v = rp.value;
    const cfg = (config ?? {}) as { options?: Array<{ id?: string; name?: string }> };
    const optName = (id: unknown): unknown => cfg.options?.find((o) => o.id === id)?.name ?? id;
    switch (type) {
      case "TEXT":
      case "RICH_TEXT":
        out[name] = typeof v === "object" && v !== null && "text" in v ? String((v as { text?: unknown }).text ?? "") : v == null ? null : String(v);
        break;
      case "NUMBER":
        out[name] = typeof v === "object" && v !== null && "value" in v ? (v as { value: unknown }).value ?? null : v;
        break;
      case "SELECT":
      case "STATUS":
        out[name] = typeof v === "object" && v !== null && "optionId" in v ? optName((v as { optionId: unknown }).optionId) : null;
        break;
      case "MULTI_SELECT":
        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { optionIds?: unknown[] }).optionIds) ? (v as { optionIds: unknown[] }).optionIds.map(optName) : [];
        break;
      case "CHECKBOX":
        out[name] = typeof v === "object" && v !== null && "value" in v ? Boolean((v as { value: unknown }).value) : Boolean(v);
        break;
      case "DATE":
      case "DATE_RANGE":
        out[name] = typeof v === "object" && v !== null && "start" in v ? (v as { start: unknown }).start ?? null : v;
        break;
      case "PERSON":
        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { userIds?: unknown[] }).userIds) ? (v as { userIds: unknown[] }).userIds : [];
        break;
      case "EMAIL":
      case "PHONE":
      case "URL":
        out[name] = typeof v === "object" && v !== null && "value" in v ? (v as { value: unknown }).value ?? null : v;
        break;
      case "FILES":
        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { fileIds?: unknown[] }).fileIds) ? (v as { fileIds: unknown[] }).fileIds : [];
        break;
      default:
        out[name] = null;
    }
  }
  return out;
}

/** A stable, short expression hash for audit (never the full expression). */
function hashExpression(expr: string): string {
  let h = 5381;
  for (let i = 0; i < expr.length; i++) h = ((h << 5) + h + expr.charCodeAt(i)) | 0;
  return `djb2:${(h >>> 0).toString(16)}`;
}

/**
 * The safe formula engine, loaded at RUNTIME (see `loadDatabaseEngine`).
 * Structural (not the package's types) so the engine's ESM source never
 * enters the API's commonjs/node10 type graph.
 */
interface ValidatedFormulaLike {
  ast: unknown;
  returnType: string;
  dependencies: string[];
  usesNow: boolean;
}
interface FormulaErrorLike {
  code: string;
  message: string;
}
interface DatabaseEngineModule {
  validateFormula(src: string, properties: Record<string, unknown>, propertyTypes?: Record<string, string>): ValidatedFormulaLike;
  evaluateFormula(ast: unknown, ctx: { now: number; properties: Record<string, unknown>; propertyTypes?: Record<string, string> }): { value: { v: unknown; t: string }; steps: number };
  FormulaError: new (code: string, message?: string) => FormulaErrorLike;
}

/**
 * Load the safe formula engine (`packages/database-engine`) at runtime.
 *
 * A static import is impossible here: the engine is a source-first ESM
 * package (`"type":"module"`, `main` → `./src/*.ts`) whose source does not
 * compile under the API's commonjs/node10 tsconfig — a static import would
 * drag the whole engine into the API's type graph. The runtime file-URL
 * import skips compile-time resolution entirely (same approach as the
 * legacy `loadFormulaEngine` in `database.service.ts`). AGENTS #18: the
 * engine is a safe AST parser/evaluator — zero `eval`/`new Function`.
 *
 * Returns `undefined` when the engine source is not resolvable (e.g. a
 * plain-`node` production runtime without a built engine) — the formula
 * endpoint then answers 503 `formula_engine_unavailable` (recorded decision).
 */
async function loadDatabaseEngine(): Promise<DatabaseEngineModule | undefined> {
  const { existsSync } = await import("node:fs");
  const { join, resolve } = await import("node:path");
  const { pathToFileURL } = await import("node:url");
  const rel = "packages/database-engine/src/formula/index.ts";
  const candidates: string[] = [join(process.cwd(), rel)];
  // The importing file's directory (apps/api/{src,dist}/modules/ai is 5 levels
  // under the repo root). Guarded: `__dirname` is CJS-only (undefined in ESM).
  if (typeof __dirname !== "undefined") {
    candidates.push(resolve(__dirname, "..", "..", "..", "..", "..", rel));
  }
  // Walk up from the process cwd — robust under vitest (cwd = a tests/ subdir),
  // a monorepo tool, or any process launched from a subdirectory.
  let dir = process.cwd();
  for (let i = 0; i < 8; i++) {
    candidates.push(join(dir, rel));
    const parent = resolve(dir, "..");
    if (parent === dir) break;
    dir = parent;
  }
  for (const candidate of new Set(candidates)) {
    if (existsSync(candidate)) {
      try {
        const mod = (await import(/* webpackIgnore: true */ pathToFileURL(candidate).href)) as DatabaseEngineModule;
        if (typeof mod.validateFormula === "function" && typeof mod.evaluateFormula === "function") return mod;
      } catch {
        // broken source — keep trying
      }
    }
  }
  return undefined;
}

@Injectable()
export class DatabaseAiService {
  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(DatabaseService) private readonly databases: DatabaseService,
    @Inject(DatabaseQueryService) private readonly records: DatabaseQueryService,
    @Inject(PageService) private readonly pages: PageService,
    @Inject(AuditService) private readonly audit: AuditService,
  ) {}

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

  /** Shared pre-stream gates: rate cap + budget (canonical JSON on failure). */
  private async preStream(workspaceId: string, userId: string, req: Request): 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 },
      ]);
    }
    await primeWorkspaceCache(this.prisma, workspaceId);
    try {
      await this.guard.check(workspaceId, undefined, req);
    } catch (err) {
      throw mapAiError(err);
    }
  }

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

  /**
   * Run a tool-calling chat, collecting the first `toolCallDelta` + the final
   * usage. Throws `no_tool_call` (422) when the model returns none. The
   * `feedback` parameter is the retry-once seam (the previous error is fed
   * back into the prompt).
   */
  private async chatToolCall(opts: {
    workspaceId: string;
    userId: string;
    system: string;
    user: string;
    tool: typeof CREATE_DATABASE_TOOL | typeof INSERT_RECORDS_TOOL | typeof GENERATE_FORMULA_TOOL;
    action: "generate" | "extract" | "formula";
    feedback?: string;
    signal?: AbortSignal;
  }): Promise<{ args: unknown; toolId: string; usage: { in: number; out: number } }> {
    const messages: Array<{ role: "system" | "user"; content: string }> = [
      { role: "system", content: opts.system },
      { role: "user", content: opts.feedback ? `${opts.user}\n\nPrevious attempt was INVALID: ${opts.feedback}\nCorrect it and call the tool again.` : opts.user },
    ];
    let args: unknown;
    let toolId = "call";
    let usage = { in: 0, out: 0 };
    for await (const chunk of this.gateway.chat({
      workspaceId: opts.workspaceId,
      userId: opts.userId,
      capability: "chat",
      messages,
      tools: [opts.tool],
      toolChoice: { name: opts.tool.name },
      metadata: { mode: "database", action: opts.action },
      signal: opts.signal,
    })) {
      if (chunk.toolCallDelta) {
        args = chunk.toolCallDelta.arguments;
        toolId = chunk.toolCallDelta.id;
      }
      if (chunk.done && chunk.usage) usage = { in: chunk.usage.in, out: chunk.usage.out };
    }
    if (args === undefined) {
      throw ApiError.unprocessable("The model returned no tool call.", [{ reason: "ai_invalid_structure", issue: "no tool call" }]);
    }
    return { args, toolId, usage };
  }

  // ---------------------------------------------------------------- generate ---

  async generate(workspaceId: string, userId: string, input: AiGenerateDatabaseInput, req: Request, res: Response): Promise<void> {
    // EDIT gate: creation goes through PageService.create (workspace EDIT).
    await assertCan(this.prisma, { workspaceId, userId }, "EDIT");
    await this.preStream(workspaceId, userId, req);

    const conversation = await this.prisma.aiConversation.create({
      data: {
        workspaceId,
        userId,
        title: DB_GEN_CONV_PREFIX,
        context: { feature: "db.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)}"` },
    });

    this.sseHeaders(res);
    const ac = new AbortController();
    let clientClosed = false;
    const onClose = (): void => {
      if (res.writableEnded) return;
      clientClosed = true;
      ac.abort();
    };
    res.on("close", onClose);
    const timer = setTimeout(() => ac.abort(), TIMEOUT_MS);
    const cleanup = (): void => {
      clearTimeout(timer);
      res.off("close", onClose);
    };
    const finish = (event: AiDatabaseSseEvent): void => {
      cleanup();
      if (!res.writableEnded) {
        res.write(sseFrame(event));
        res.end();
      }
    };

    try {
      const userPrompt = renderGeneratePrompt(input.prompt);
      let attempt = await this.chatToolCall({
        workspaceId,
        userId,
        system: DB_GEN_SYSTEM_PROMPT,
        user: userPrompt,
        tool: CREATE_DATABASE_TOOL,
        action: "generate",
        signal: ac.signal,
      });
      let parsed = validateCreateDatabase(attempt.args);
      if (!parsed.ok) {
        attempt = await this.chatToolCall({
          workspaceId,
          userId,
          system: DB_GEN_SYSTEM_PROMPT,
          user: userPrompt,
          tool: CREATE_DATABASE_TOOL,
          action: "generate",
          feedback: parsed.feedback,
          signal: ac.signal,
        });
        parsed = validateCreateDatabase(attempt.args);
      }

      if (!parsed.ok) {
        // No partial write — nothing has been created yet.
        const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
          feature: "db.generate",
          error: "ai_invalid_structure",
          issue: parsed.feedback,
        });
        if (!clientClosed && !res.writableEnded) res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "createDatabase", arguments: attempt.args as Record<string, unknown> }));
        if (!clientClosed && !res.writableEnded) {
          res.write(sseFrame({
            type: "error",
            code: "UNPROCESSABLE_ENTITY",
            message: "The AI returned an invalid database structure.",
            reason: "ai_invalid_structure",
          }));
        }
        finish({ type: "done", assistantMessageId: assistantId, reason: "error" });
        return;
      }

      const normalized = parsed.value;

      // ---- write through the normal services (permission/audit/realtime) ----
      const page = await this.pages.create(workspaceId, userId, { title: normalized.title, parentId: input.parentId, type: "database" }, req);
      // PageService.create returns a union (the template path is a different
      // shape); we always pass type:"database" and never a templateId, so the
      // `{ ..., databaseId }` member applies.
      const databaseId = (page as { databaseId?: string }).databaseId!;

      const propertyIds: string[] = [];
      for (const p of normalized.properties) {
        const prop = await this.databases.createProperty(
          databaseId,
          userId,
          {
            name: p.name,
            type: p.type,
            ...(p.options ? { config: { options: p.options } } : { config: {} }),
            ...(p.required !== undefined ? { required: p.required } : {}),
          },
          req,
        );
        propertyIds.push(prop.id);
      }

      // property NAME → id (the seeded Title/Status + the new ones).
      const allProps = await this.prisma.databaseProperty.findMany({
        where: { databaseId, deletedAt: null },
        select: { id: true, name: true },
      });
      const idByName = new Map(allProps.map((ap) => [ap.name.toLowerCase(), ap.id]));

      const recordIds: string[] = [];
      for (const rec of normalized.sampleRecords) {
        const title = typeof rec.Title === "string" ? rec.Title : undefined;
        const properties: Record<string, unknown> = {};
        for (const [name, value] of Object.entries(rec)) {
          if (name === "Title") continue;
          const pid = idByName.get(name.toLowerCase());
          if (pid) properties[pid] = value;
        }
        const recDto = await this.records.createRecord(databaseId, userId, { title, properties, links: {} }, req);
        recordIds.push(String((recDto as { id: string }).id));
      }

      const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
        feature: "db.generate",
        databaseId,
        propertyIds,
        recordIds,
      });

      await this.audit.audit(req, req?.auth, {
        action: "ai.database_generated",
        resourceType: "database",
        resourceId: databaseId,
        workspaceId,
        metadata: { count: recordIds.length, propertyCount: propertyIds.length },
      });

      if (!clientClosed && !res.writableEnded) {
        res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "createDatabase", arguments: attempt.args as Record<string, unknown> }));
        res.write(sseFrame({ type: "usage", tokensIn: attempt.usage.in, tokensOut: attempt.usage.out }));
      }
      finish({
        type: "done",
        assistantMessageId: assistantId,
        reason: "ok",
        databaseId,
        propertyIds,
        recordIds,
        tokensIn: attempt.usage.in,
        tokensOut: attempt.usage.out,
      });
    } catch (err) {
      cleanup();
      if (clientClosed || ac.signal.aborted) {
        if (!clientClosed && !res.writableEnded) {
          res.write(sseFrame({ type: "done", reason: "aborted" }));
          res.end();
        }
        return;
      }
      const mapped = mapAiError(err);
      if (!clientClosed && !res.writableEnded) {
        res.write(sseFrame({ type: "error", code: mapped.code, message: mapped.message }));
        res.write(sseFrame({ type: "done", reason: "error" }));
        res.end();
      }
      logger.info(`ai database generate stream error workspace=${workspaceId} code=${mapped.code}`);
    }
  }

  // ----------------------------------------------------------------- extract ---

  async extract(workspaceId: string, userId: string, input: AiExtractInput, req: Request, res: Response): Promise<void> {
    const db = await this.prisma.database.findFirst({
      where: { id: input.databaseId, workspaceId, isArchived: false },
      select: { id: true, title: true },
    });
    if (!db) throw ApiError.notFound("Database not found.");
    // EDIT on the database (record creation is an EDIT action).
    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
    await this.preStream(workspaceId, userId, req);

    const properties = await this.loadProps(db.id);

    const conversation = await this.prisma.aiConversation.create({
      data: {
        workspaceId,
        userId,
        title: DB_EXTRACT_CONV_PREFIX,
        context: { feature: "db.extract", databaseId: db.id } as object,
      },
    });
    await this.prisma.aiMessage.create({
      data: { conversationId: conversation.id, role: "user", content: `extract: ${input.text.length} chars → ${db.title}` },
    });

    this.sseHeaders(res);
    const ac = new AbortController();
    let clientClosed = false;
    const onClose = (): void => {
      if (res.writableEnded) return;
      clientClosed = true;
      ac.abort();
    };
    res.on("close", onClose);
    const timer = setTimeout(() => ac.abort(), TIMEOUT_MS);
    const cleanup = (): void => {
      clearTimeout(timer);
      res.off("close", onClose);
    };
    const finish = (event: AiDatabaseSseEvent): void => {
      cleanup();
      if (!res.writableEnded) {
        res.write(sseFrame(event));
        res.end();
      }
    };

    try {
      const prompt = renderExtractPrompt({ databaseTitle: db.title, properties, text: input.text });
      let attempt = await this.chatToolCall({ workspaceId, userId, system: DB_EXTRACT_SYSTEM_PROMPT, user: prompt, tool: INSERT_RECORDS_TOOL, action: "extract", signal: ac.signal });
      let parsed = validateInsertRecords(attempt.args, properties);
      if (!parsed.ok) {
        attempt = await this.chatToolCall({ workspaceId, userId, system: DB_EXTRACT_SYSTEM_PROMPT, user: prompt, tool: INSERT_RECORDS_TOOL, action: "extract", feedback: parsed.feedback, signal: ac.signal });
        parsed = validateInsertRecords(attempt.args, properties);
      }

      if (!parsed.ok) {
        const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
          feature: "db.extract",
          databaseId: db.id,
          error: "ai_invalid_structure",
          issue: parsed.feedback,
        });
        if (!clientClosed && !res.writableEnded) res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "insertRecords", arguments: attempt.args as Record<string, unknown> }));
        if (!clientClosed && !res.writableEnded) {
          res.write(sseFrame({
            type: "error",
            code: "UNPROCESSABLE_ENTITY",
            message: "The AI returned an invalid record structure.",
            reason: "ai_invalid_structure",
          }));
        }
        finish({ type: "done", assistantMessageId: assistantId, reason: "error" });
        return;
      }

      const normalized = parsed.value;
      const recordIds: string[] = [];
      for (const rec of normalized.records) {
        const title = typeof rec.properties.__title__ === "string" ? (rec.properties.__title__ as string) : undefined;
        const { __title__, ...properties } = rec.properties as Record<string, unknown> & { __title__?: string };
        const recDto = await this.records.createRecord(db.id, userId, { title, properties, links: {} }, req);
        recordIds.push(String((recDto as { id: string }).id));
      }

      const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
        feature: "db.extract",
        databaseId: db.id,
        recordIds,
        ignoredColumns: normalized.ignoredColumns,
      });

      await this.audit.audit(req, req?.auth, {
        action: "ai.records_extracted",
        resourceType: "database",
        resourceId: db.id,
        workspaceId,
        metadata: { count: recordIds.length, ignoredColumns: normalized.ignoredColumns },
      });

      if (!clientClosed && !res.writableEnded) {
        res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "insertRecords", arguments: attempt.args as Record<string, unknown> }));
        res.write(sseFrame({ type: "usage", tokensIn: attempt.usage.in, tokensOut: attempt.usage.out }));
      }
      finish({
        type: "done",
        assistantMessageId: assistantId,
        reason: "ok",
        databaseId: db.id,
        recordIds,
        ignoredColumns: normalized.ignoredColumns,
        tokensIn: attempt.usage.in,
        tokensOut: attempt.usage.out,
      });
    } catch (err) {
      cleanup();
      if (clientClosed || ac.signal.aborted) {
        if (!clientClosed && !res.writableEnded) {
          res.write(sseFrame({ type: "done", reason: "aborted" }));
          res.end();
        }
        return;
      }
      const mapped = mapAiError(err);
      if (!clientClosed && !res.writableEnded) {
        res.write(sseFrame({ type: "error", code: mapped.code, message: mapped.message }));
        res.write(sseFrame({ type: "done", reason: "error" }));
        res.end();
      }
      logger.info(`ai database extract stream error workspace=${workspaceId} code=${mapped.code}`);
    }
  }

  // ----------------------------------------------------------------- preview ---

  /** Deterministic column→property mapping (no LLM, no write, no budget). */
  async preview(workspaceId: string, userId: string, input: AiExtractPreviewInput) {
    const db = await this.prisma.database.findFirst({
      where: { id: input.databaseId, workspaceId, isArchived: false },
      select: { id: true },
    });
    if (!db) throw ApiError.notFound("Database not found.");
    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
    const properties = await this.loadProps(db.id);
    const mapping = previewExtractMapping(input.text, properties);
    return { databaseId: db.id, ...mapping };
  }

  // ----------------------------------------------------------------- formula ---

  async formula(
    workspaceId: string,
    userId: string,
    input: AiGenerateFormulaInput,
    req: Request,
  ): Promise<{
    formulaId: string;
    propertyId: string;
    expression: string;
    returnType: string;
    explanation?: string;
    samples: Array<{ recordId: string; value: unknown; error?: string }>;
    tokensIn: number;
    tokensOut: number;
  }> {
    const db = await this.prisma.database.findFirst({
      where: { id: input.databaseId, workspaceId, isArchived: false },
      select: { id: true },
    });
    if (!db) throw ApiError.notFound("Database not found.");
    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
    await this.preStream(workspaceId, userId, req);

    const engine = await loadDatabaseEngine();
    if (!engine) {
      throw new ApiError("SERVICE_UNAVAILABLE", 503, "The formula engine is unavailable.", [
        { reason: "formula_engine_unavailable" },
      ]);
    }

    const properties = await this.loadProps(db.id);
    const propNames = properties.map((p) => ({ name: p.name, type: p.type }));
    const propertyTypes = new Map(properties.map((p) => [p.name, lowerPropertyType(p.type)]));
    const sampleRows = await this.loadSampleRows(db.id, input.sampleRowIds);

    const conversation = await this.prisma.aiConversation.create({
      data: {
        workspaceId,
        userId,
        title: DB_FORMULA_CONV_PREFIX,
        context: { feature: "db.formula", databaseId: db.id } as object,
      },
    });
    await this.prisma.aiMessage.create({
      data: { conversationId: conversation.id, role: "user", content: `formula: "${input.description.slice(0, 200)}"` },
    });

    const userPrompt = renderFormulaPrompt({ properties: propNames, description: input.description });
    let attempt = await this.chatToolCall({ workspaceId, userId, system: DB_FORMULA_SYSTEM_PROMPT, user: userPrompt, tool: GENERATE_FORMULA_TOOL, action: "formula" });
    let expression = String((attempt.args as { expression?: unknown })?.expression ?? "");
    let explanation = typeof (attempt.args as { explanation?: unknown })?.explanation === "string" ? ((attempt.args as { explanation?: string }).explanation as string) : undefined;

    // Validate with the safe engine BEFORE saving (parse + type-check).
    // A record's formula values = its stored properties + its TITLE (the
    // title lives on the page, not a RecordProperty — but prop("Title") must
    // resolve). Missing properties are null (the engine normalizes null).
    const formulaPropsFor = (row: { title: string | null; properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }> }): Record<string, unknown> => {
      const out = recordToFormulaProps(row.properties);
      out.Title = row.title ?? "";
      return out;
    };
    const sampleValues: Record<string, unknown> = {};
    for (const p of properties) sampleValues[p.name] = null;
    if (sampleRows.length > 0) Object.assign(sampleValues, formulaPropsFor(sampleRows[0]!));
    const typeMap = Object.fromEntries(propertyTypes);

    let validation: ValidatedFormulaLike | null = null;
    let parseError: string | undefined;
    const validate = (expr: string): boolean => {
      try {
        validation = engine.validateFormula(expr, sampleValues, typeMap);
        return true;
      } catch (err) {
        parseError = err instanceof engine.FormulaError ? `${err.code}: ${err.message}` : (err as Error).message;
        return false;
      }
    };

    if (!validate(expression)) {
      // Retry once with the parse error as feedback.
      attempt = await this.chatToolCall({ workspaceId, userId, system: DB_FORMULA_SYSTEM_PROMPT, user: userPrompt, tool: GENERATE_FORMULA_TOOL, action: "formula", feedback: parseError ?? "the expression is invalid" });
      expression = String((attempt.args as { expression?: unknown })?.expression ?? "");
      explanation = typeof (attempt.args as { explanation?: unknown })?.explanation === "string" ? ((attempt.args as { explanation?: string }).explanation as string) : undefined;
      if (!validate(expression)) {
        // No Formula row written.
        await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
          feature: "db.formula",
          databaseId: db.id,
          error: "formula_invalid",
          parseError,
        });
        throw ApiError.unprocessable("The AI generated a formula that failed validation.", [
          { reason: "formula_invalid", parseError: parseError ?? null },
        ]);
      }
    }

    // Evaluate on the sample rows (read-only; the engine's guards apply).
    // The evaluator returns a typed EValue `{v, t}` — `v` is the raw value.
    const samples = sampleRows.map((row) => {
      try {
        const { value } = engine.evaluateFormula(validation!.ast, {
          now: Date.now(),
          properties: formulaPropsFor(row),
          propertyTypes: typeMap,
        });
        return { recordId: row.id, value: value.v };
      } catch (err) {
        return { recordId: row.id, value: null, error: err instanceof engine.FormulaError ? err.message : (err as Error).message };
      }
    });

    // ---- save: formula property + Formula row through the normal path
    // (createFormula re-checks EDIT, audits `database.formula_created`, and
    // publishes `database.updated` realtime). The engine-validated AST +
    // returnType are stored verbatim (spec: the expression is engine-validated
    // BEFORE saving).
    const name = input.name ?? "Formula";
    const formula = await this.databases.createFormula(
      db.id,
      userId,
      { name, expression, ast: validation!.ast as object, returnType: validation!.returnType },
      req,
    );
    const propertyId = formula.propertyId;

    await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
      feature: "db.formula",
      databaseId: db.id,
      formulaId: formula.id,
      propertyId,
      expression,
      samples,
    });

    await this.audit.audit(req, req?.auth, {
      action: "ai.formula_generated",
      resourceType: "formula",
      resourceId: formula.id,
      workspaceId,
      metadata: { databaseId: db.id, expressionHash: hashExpression(expression) },
    });

    return {
      formulaId: formula.id,
      propertyId: propertyId!,
      expression,
      returnType: validation!.returnType,
      ...(explanation ? { explanation } : {}),
      samples,
      tokensIn: attempt.usage.in,
      tokensOut: attempt.usage.out,
    };
  }

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

  private async loadProps(databaseId: string): Promise<DbPropMeta[]> {
    const rows = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      orderBy: { position: "asc" },
      select: { id: true, name: true, type: true, isTitle: true, config: true },
    });
    return rows.map((r) => ({ id: r.id, name: r.name, type: r.type, isTitle: r.isTitle, config: r.config }));
  }

  private async loadSampleRows(
    databaseId: string,
    sampleRowIds?: string[],
  ): Promise<Array<{ id: string; title: string | null; properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }> }>> {
    const where =
      sampleRowIds && sampleRowIds.length > 0
        ? { id: { in: sampleRowIds.slice(0, DB_FORMULA_MAX_SAMPLE_ROWS) }, databaseId, archivedAt: null }
        : { databaseId, archivedAt: null };
    const rows = await this.prisma.databaseRecord.findMany({
      where,
      include: {
        properties: { select: { property: { select: { id: true, name: true, type: true, config: true } }, value: true } },
        // The record IS a page — the TITLE property lives on the page, and a
        // formula may reference it (prop("Title")). Carried separately because
        // it is not a RecordProperty row.
        page: { select: { title: true } },
      },
      orderBy: { position: "asc" },
      take: sampleRowIds ? undefined : DB_FORMULA_MAX_SAMPLE_ROWS,
    });
    // The record's title lives on its page (the record IS a page) — the
    // Prisma row carries it at `page.title`, but the formula values need a
    // flat `title` (a formula may reference prop("Title")).
    return rows.map((r) => ({
      id: r.id,
      title: r.page?.title ?? null,
      properties: r.properties,
    }));
  }

  private async persistAssistant(
    conversationId: string,
    model: string,
    usage: { in: number; out: number },
    _now: 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 database assistant persist failed: ${(err as Error).message}`);
      return undefined;
    }
  }
}