/**
 * Search service — TASK-0097 S3.
 *
 * The API's search module is now a thin, permission-scoped facade over the
 * single tested core, `@factory/search` (`SearchClient`): one index (alias
 * `factory-content`), one query shape, one permission filter. The Phase-0
 * hand-rolled 3-index client (`opensearch.ts`) and its Postgres-ILIKE
 * fallback are GONE (recorded decision, phase-08-search.md "Phase 8 Notes":
 * no Postgres-ILIKE fallback — OpenSearch down ⇒ 503 `SEARCH_UNAVAILABLE`).
 *
 * Security model (TASK-0097 §5 + Security): every query runs a FAIL-CLOSED
 * caller-scope resolver (userId → `ViewerContext { userId, workspaceIds,
 * teamIds }`) and the tenant clause is mandatory — the client never runs an
 * unfiltered query:
 *   - explicit `workspaceId` the caller is NOT a member of → 404 NOT_FOUND
 *     (no existence leak; consistent with the pages API),
 *   - "all workspaces" with zero memberships → empty result set (no query).
 *
 * Response shapes (TASK-0097 S3, recorded — the live web client pins these):
 *   - POST /search          → envelope { data, nextCursor, tookMs }
 *   - GET  /search/query    → BARE array of results (command palette client)
 *   - GET  /search/suggest  → BARE array of { id, type, title } (the old
 *     `{ data: string[] }` shape already mismatched the web typing; pinned
 *     to objects the suggest UI needs)
 *   - GET/PUT/DELETE /search/recent → { data: string[] } (Redis-backed, kept
 *     as-is: key `search:recent:{userId}`, cap 20, newest first)
 */
import { createHash } from "node:crypto";
import { Inject, Injectable } from "@nestjs/common";
import { z } from "zod";
// `SearchClient` is a VALUE here (the service constructs it lazily from
// env); `SearchType`/`ViewerContext` are types. Root import only — the API
// resolves under node10 (no subpaths), and the root re-exports the client.
import { SearchClient, type SearchSort, type SearchType, type ViewerContext } from "@factory/search";
import { m } from "@factory/observability";
import { loadEnv } from "../../config/env.js";
import { ActorContext } from "../../common/actor-context.js";
import { ApiError } from "../../common/errors.js";
import { logger } from "../../common/logger.js";
import { PrismaService } from "../../common/prisma.service.js";
import { RedisService } from "../../common/redis.service.js";
import { CacheService } from "../../common/cache.js";

const RECENT_KEY = (userId: string) => `search:recent:${userId}`;
const RECENT_MAX = 20;
/** TASK-0097 §7: structured log per request — never more than this of the raw query. */
const LOG_QUERY_PREFIX_CHARS = 64;
/** TASK-0098 §5: per-user rate limit on recent-search writes (30/min). */
const RECENT_WRITE_LIMIT = 30;
const RECENT_WRITE_WINDOW_MS = 60_000;
/** Suggest page size (the client defaults to 8; keep the server cap small). */
const SUGGEST_DEFAULT_SIZE = 8;
/** Suggest server cap (buildPrefixBody size guard). */
const SUGGEST_MAX_SIZE = 20;

export interface SearchInput {
  q: string;
  types?: SearchType[];
  workspaceId?: string;
  teamId?: string;
  createdBy?: string;
  updatedAfter?: string;
  updatedBefore?: string;
  /** T-155: result ordering (undefined/"relevance" = the default `_score` path). */
  sortBy?: SearchSort;
  cursor?: string;
  limit?: number;
}

export interface SearchEnvelope {
  data: import("@factory/search").SearchResult[];
  nextCursor: string | null;
  tookMs: number;
}

export interface Suggestion {
  id: string;
  type: string;
  title: string;
}

/** The fail-closed caller scope (TASK-0097 Security). */
interface CallerScope {
  /** Every non-deleted workspace the user is a member of. */
  workspaceIds: string[];
  /** Team ids the user belongs to, within those workspaces. */
  teamIds: string[];
}

type SearchResult = import("@factory/search").SearchResult;

/**
 * TASK-0135: search results are cached for 5 s, keyed by
 * { workspaceId, query-hash, filters-hash, principalId } — permission
 * relevant, so the principal is part of the key (two viewers never share an
 * entry). The zod schema below matches the JSON shape of the LIVE envelope
 * `{ data: SearchResult[], nextCursor, tookMs }` so a corrupt entry is a
 * cache MISS, never a 500.
 */
const SEARCH_RESULT_CACHE = z.object({
  docId: z.string(),
  type: z.string(),
  id: z.string(),
  title: z.string(),
  snippet: z.string().optional(),
  url: z.string(),
  icon: z.string().optional(),
  createdByName: z.string().optional(),
  updatedAt: z.string(),
  pageId: z.string().optional(),
  blockId: z.string().optional(),
  databaseId: z.string().optional(),
  recordId: z.string().optional(),
  workspaceId: z.string(),
  score: z.number(),
  source: z.string()
});
const SEARCH_ENVELOPE_CACHE = z.object({
  data: z.array(SEARCH_RESULT_CACHE),
  nextCursor: z.string().nullable(),
  tookMs: z.number()
});

@Injectable()
export class SearchService {
  /**
   * The single `SearchClient` for the module (TASK-0097: built lazily so
   * unit tests can swap the instance via {@link setClientForTesting}).
   */
  private clientOverride: SearchClient | null = null;
  private lazyClient: SearchClient | null = null;

  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaService,
    @Inject(RedisService) private readonly redis: RedisService,
    @Inject(CacheService) private readonly cache: CacheService,
  ) {}

  /** Test seam: swap the client instance (a fake/fetch-backed `SearchClient`). */
  setClientForTesting(client: SearchClient | null): void {
    this.clientOverride = client;
    this.lazyClient = null;
  }

  private client(): SearchClient {
    if (this.clientOverride) return this.clientOverride;
    if (!this.lazyClient) {
      const env = loadEnv();
      this.lazyClient = new SearchClient({
        baseUrl: env.SEARCH_URL,
        ...(env.SEARCH_PASSWORD
          ? { auth: { username: env.SEARCH_USERNAME, password: env.SEARCH_PASSWORD } }
          : {}),
      });
    }
    return this.lazyClient;
  }

  /**
   * TASK-0125 — public access to the underlying client for the AI RAG query
   * path (it needs the raw client's `_mget` for full chunk context beyond the
   * 160-char snippet the search API surfaces). The client is the SAME
   * singleton the search API uses (config + test-override aware).
   */
  searchClient(): SearchClient {
    return this.client();
  }

  /**
   * Resolve the caller's scope (fail-closed, TASK-0097 Security):
   * workspaceIds = the user's non-deleted workspace memberships;
   * teamIds = `teamMember` rows joined through `team.workspaceId IN workspaceIds`.
   */
  async resolveScope(userId: string): Promise<CallerScope> {
    // TASK-0127: API keys are scoped to their OWN workspace (docs §7) —
    // never the creator's memberships. The key sees the whole workspace
    // (including team-scoped content — it acts as the workspace), so
    // teamIds = ALL teams of that workspace (a `teamId` filter then works).
    const actor = ActorContext.get();
    if (actor?.kind === "api" && actor.workspaceId) {
      const active = await this.filterActiveWorkspaces([actor.workspaceId]);
      let teamIds: string[] = [];
      if (active.length > 0) {
        const teamRows = await this.prisma.team.findMany({
          where: { workspaceId: actor.workspaceId },
          select: { id: true },
        });
        teamIds = teamRows.map((t) => t.id);
      }
      return { workspaceIds: active, teamIds };
    }
    const members = await this.prisma.workspaceMember.findMany({
      where: { userId },
      select: { workspaceId: true },
    });
    const workspaceIds = await this.filterActiveWorkspaces(members.map((m) => m.workspaceId));
    let teamIds: string[] = [];
    if (workspaceIds.length > 0) {
      const teamRows = await this.prisma.teamMember.findMany({
        where: { userId, team: { workspaceId: { in: workspaceIds } } },
        select: { teamId: true },
      });
      teamIds = [...new Set(teamRows.map((t) => t.teamId))];
    }
    return { workspaceIds, teamIds };
  }

  /**
   * TASK-0125 — public viewer resolution for the AI RAG query path (the RAG
   * service lives in the AI module; it must not reach into the search
   * service's private plumbing). Same fail-closed rules as the search API:
   * non-member workspace → 404; zero memberships → null (empty result, no
   * query sent).
   */
  async viewerForWorkspace(userId: string, workspaceId: string): Promise<ViewerContext | null> {
    const scope = await this.resolveScope(userId);
    return this.viewerFor(userId, scope, workspaceId);
  }

  /** Drop workspaces that no longer exist (deleted rows) — non-member ≠ member. */
  private async filterActiveWorkspaces(ids: string[]): Promise<string[]> {
    if (ids.length === 0) return [];
    const rows = await this.prisma.workspace.findMany({
      where: { id: { in: ids }, deletedAt: null },
      select: { id: true },
    });
    return rows.map((w) => w.id);
  }

  /**
   * The viewer context for a request, enforcing the fail-closed rules:
   *   - explicit `workspaceId` + not a member → 404 NOT_FOUND (no leak),
   *   - no `workspaceId` + zero memberships  → null (caller returns empty;
   *     the query is never sent — "nothing the caller can see").
   */
  private async viewerFor(
    userId: string,
    scope: CallerScope,
    workspaceId?: string,
  ): Promise<ViewerContext | null> {
    const actor = ActorContext.get();
    const isKey = actor?.kind === "api";
    let workspaceIds: string[];
    if (isKey) {
      // TASK-0127: a key can ONLY search its own workspace, explicitly.
      // No `workspaceId` (or a different one) → 404, never a cross-tenant
      // query over "all of the key's workspaces" (the key has exactly one).
      if (!workspaceId || workspaceId !== actor.workspaceId) {
        throw ApiError.notFound("Resource not found.");
      }
      workspaceIds = [workspaceId];
    } else if (workspaceId) {
      if (!scope.workspaceIds.includes(workspaceId)) {
        // 404 (NOT 403): consistent with the pages API; a workspace the
        // caller is not a member of simply does not exist for them.
        throw ApiError.notFound("Resource not found.");
      }
      workspaceIds = [workspaceId];
    } else {
      workspaceIds = scope.workspaceIds;
      if (workspaceIds.length === 0) return null;
    }
    // TASK-0127: for a key the viewer identity is SYNTHETIC — a real user
    // id (the creator's) in `grants.users` would surface the creator's
    // personal grants. The key acts as the workspace: the
    // `grants.workspace` + team clauses (all workspace teams) are its access.
    return {
      userId: isKey ? `apikey-${actor!.apiKeyId ?? "unknown"}` : userId,
      workspaceIds,
      teamIds: scope.teamIds,
    };
  }

  /**
   * Full-text search (POST /search). TASK-0135: the result envelope is
   * cached for 5 s, keyed by { workspaceId, query-hash, filters-hash,
   * principalId } — permission relevant, so the principal (userId) is part of
   * the key and two viewers never share an entry. On a cache hit the scope +
   * viewer + OpenSearch round-trip are skipped entirely.
   */
  async search(userId: string, input: SearchInput): Promise<SearchEnvelope> {
    const cacheKeyInput = {
      q: input.q.trim(),
      types: input.types,
      workspaceId: input.workspaceId,
      teamId: input.teamId,
      createdBy: input.createdBy,
      updatedAfter: input.updatedAfter,
      updatedBefore: input.updatedBefore,
      // T-155: a different ordering is a different result — part of the key so
      // a sortBy=createdAt query never returns a cached relevance envelope.
      sortBy: input.sortBy,
      cursor: input.cursor,
      limit: input.limit
    };
    const { data } = await this.cache.getSearch(
      input.workspaceId,
      userId,
      cacheKeyInput,
      () => this.searchLive(userId, input),
      SEARCH_ENVELOPE_CACHE
    );
    return data as SearchEnvelope;
  }

  /** The live (un-cached) search — the loader the 5 s cache calls. */
  private async searchLive(userId: string, input: SearchInput): Promise<SearchEnvelope> {
    const q = input.q.trim();
    const scope = await this.resolveScope(userId);
    const viewer = await this.viewerFor(userId, scope, input.workspaceId);
    if (!viewer) {
      // Fail-closed: zero visible workspaces → nothing can match. No query sent.
      this.logSearch(userId, { q, types: input.types, workspaceId: input.workspaceId }, 0, 0);
      return { data: [], nextCursor: null, tookMs: 0 };
    }

    // Clamp OUTSIDE tryClient: an out-of-range `limit` must surface as
    // 400 VALIDATION_ERROR, not 503 — tryClient's catch maps ANY non-
    // SEARCH_UNAVAILABLE throw to 503 SEARCH_UNAVAILABLE (mapSharedError),
    // which would swallow the 400 (recorded S3 contract, UI depends on it).
    const pageSize = clampLimit(input.limit);
    const started = Date.now();
    const res = await this.tryClient(() =>
      this.client().search(
        {
          q,
          ...(input.types && input.types.length > 0 ? { types: input.types } : {}),
          // T-155: additive date sort — undefined/"relevance" keeps the default.
          ...(input.sortBy && input.sortBy !== "relevance" ? { sortBy: input.sortBy } : {}),
          filters: {
            ...(input.teamId ? { teams: [input.teamId] } : {}),
            ...(input.createdBy ? { createdBy: input.createdBy } : {}),
            ...(input.updatedAfter ? { updatedAfter: input.updatedAfter } : {}),
            ...(input.updatedBefore ? { updatedBefore: input.updatedBefore } : {}),
          },
          page: input.cursor,
          pageSize,
        },
        viewer,
      ),
    );
    const tookMs = Date.now() - started;
    // TASK-0135: the spec's search_query_duration_seconds (type bm25 — the
    // live engine is BM25; kNN would be a second call site when/if hybrid
    // lands). The cache-HIT path (search()) never reaches here — cached
    // results are not "search queries" for the duration catalog.
    m.search.record({ type: "bm25" }, tookMs / 1000);
    this.logSearch(
      userId,
      { q, types: input.types, workspaceId: input.workspaceId },
      res.total,
      tookMs,
    );
    return { data: res.results, nextCursor: res.cursor ?? null, tookMs };
  }

  /**
   * Command-palette query (GET /search/query): same engine, BARE array
   * response (the web client types it `Promise<SearchResult[]>` and does
   * `searchApi.query(...).then(setResults)` — a record shape would break the
   * palette; envelope-vs-array is the recorded S3 decision).
   */
  async query(userId: string, input: SearchInput): Promise<SearchResult[]> {
    const q = input.q.trim();
    const scope = await this.resolveScope(userId);
    const viewer = await this.viewerFor(userId, scope, input.workspaceId);
    if (!viewer) return [];

    // Clamp outside tryClient (400 VALIDATION_ERROR, not 503 — see search()).
    const pageSize = clampLimit(input.limit);
    const res = await this.tryClient(() =>
      this.client().search(
        {
          q,
          ...(input.types && input.types.length > 0 ? { types: input.types } : {}),
          // T-155: additive date sort — undefined/"relevance" keeps the default.
          ...(input.sortBy && input.sortBy !== "relevance" ? { sortBy: input.sortBy } : {}),
          filters: {
            ...(input.teamId ? { teams: [input.teamId] } : {}),
            ...(input.createdBy ? { createdBy: input.createdBy } : {}),
            ...(input.updatedAfter ? { updatedAfter: input.updatedAfter } : {}),
            ...(input.updatedBefore ? { updatedBefore: input.updatedBefore } : {}),
          },
          page: input.cursor,
          pageSize,
        },
        viewer,
      ),
    );
    // TASK-0135: search_query_duration_seconds — the command-palette path
    // (GET /search/query) is the same BM25 engine.
    m.search.record({ type: "bm25" }, res.latencyMs / 1000);
    this.logSearch(
      userId,
      { q, types: input.types, workspaceId: input.workspaceId },
      res.total,
      res.latencyMs,
    );
    return res.results;
  }

  /**
   * Suggestions (GET /search/suggest): prefix match via the client's
   * `suggest()` (`buildPrefixBody` over the SAME permission filters). BARE
   * array of `{ id, type, title }` (recorded S3 decision: the old
   * `{ data: string[] }` mismatched the web typing `SearchResult[]`;
   * objects are what the suggest UI needs).
   */
  async suggest(userId: string, q: string, limit?: number): Promise<Suggestion[]> {
    const scope = await this.resolveScope(userId);
    const viewer = await this.viewerFor(userId, scope);
    if (!viewer) return [];
    // Clamp outside tryClient (400 VALIDATION_ERROR, not 503 — see search()).
    const size = clampSuggest(limit);
    const started = Date.now();
    const items = await this.tryClient(() => this.client().suggest(q, viewer, size));
    this.logSearch(userId, { q, suggest: true }, items.length, Date.now() - started);
    return items;
  }

  /** Run a client call, mapping the shared `ApiError` (see mapSharedError). */
  private async tryClient<T>(fn: () => Promise<T>): Promise<T> {
    try {
      return await fn();
    } catch (err) {
      this.mapSharedError(err);
    }
  }

  /**
   * Map the SHARED `@factory/shared/errors` ApiError thrown by `SearchClient`
   * onto the API-local `ApiError` the global exception filter understands.
   *
   * WHY: the filter's `instanceof ApiError` check is against the API-LOCAL
   * class (common/errors.ts); the shared class is a DIFFERENT class, so an
   * unmapped shared error would fall through every structural check to the
   * 500 INTERNAL_ERROR catch-all. The shared `SEARCH_UNAVAILABLE` (503)
   * therefore becomes the API-local `ApiError("SEARCH_UNAVAILABLE", 503, …)`
   * with a GENERIC message — backend details (host, path, upstream status)
   * are logged here and never returned (TASK-0097 §8: "do not leak
   * backend details").
   */
  private mapSharedError(err: unknown): never {
    const code: unknown = (err as { code?: unknown })?.code;
    const status: unknown = (err as { status?: unknown })?.status;
    const msg = err instanceof Error ? err.message : "search request failed";
    if (code === "SEARCH_UNAVAILABLE") {
      logger.warn("opensearch unavailable", { code, status, message: msg });
      throw new ApiError(
        "SEARCH_UNAVAILABLE",
        typeof status === "number" ? status : 503,
        "Search is temporarily unavailable.",
      );
    }
    // Unknown shared error codes (e.g. AI_ERROR from a mis-configured hybrid)
    // and non-ApiError throws (should not happen): 503 keeps the contract
    // "search down ⇒ 503 SEARCH_UNAVAILABLE" (TASK-0097 §8).
    logger.warn("search backend error", {
      code: code ?? null,
      status: status ?? null,
      message: msg,
    });
    throw new ApiError("SEARCH_UNAVAILABLE", 503, "Search is temporarily unavailable.");
  }

  /** TASK-0097 §7 structured log: types, tookMs, total + query hash. */
  private logSearch(
    userId: string,
    meta: { q: string; types?: SearchType[]; workspaceId?: string; suggest?: boolean },
    total: number,
    tookMs: number,
  ): void {
    // Never more than 64 raw chars in the log; the rest is a hash.
    const prefix = meta.q.slice(0, LOG_QUERY_PREFIX_CHARS);
    const hash = createHash("sha256").update(meta.q, "utf8").digest("hex").slice(0, 16);
    logger.info(meta.suggest ? "search:suggest" : "search:query", {
      userId,
      types: meta.types ?? null,
      workspaceId: meta.workspaceId ?? null,
      tookMs,
      total,
      queryPrefix: prefix,
      queryHash: hash,
    });
  }

  // ---------------------------------------------------------------------
  // Recent searches (Redis, kept as-is from the Phase-0 module: key
  // `search:recent:{userId}`, cap 20, newest first). The e2e spec pins the
  // `search-recent` / `search-recent-item` UI testids against these routes.
  // ---------------------------------------------------------------------

  /** Record a query (dedupe: lrem + lpush, cap 20, newest first). Best-effort. */
  async trackRecent(userId: string, q: string): Promise<void> {
    try {
      const key = RECENT_KEY(userId);
      const redis = this.redis.getRedis();
      await redis.lrem(key, 1, q);
      await redis.lpush(key, q);
      await redis.ltrim(key, 0, RECENT_MAX - 1);
    } catch (err) {
      logger.warn(`recent-search tracking failed: ${(err as Error).message}`);
    }
  }

  async recent(userId: string): Promise<{ data: string[] }> {
    try {
      const items = await this.redis.getRedis().lrange(RECENT_KEY(userId), 0, RECENT_MAX - 1);
      return { data: items };
    } catch {
      return { data: [] };
    }
  }

  /**
   * PUT /search/recent (TASK-0098 §5, recorded in docs/api/search.md):
   * upsert a query into the user's list (dedupe, cap 20, newest first) and
   * return the updated list. Reuses `trackRecent` (same Redis list); the
   * 30/min per-user write rate limit (TASK-0098 §5) is a small inline
   * sliding window on the SAME Redis — no new infra (recorded S3 decision:
   * the global RateLimitGuard is per-route-class, not per-action, and
   * building dedicated infra is out of scope for this slice).
   */
  async addRecent(userId: string, q: string): Promise<{ data: string[] }> {
    const raw = q.trim();
    if (raw.length === 0)
      throw ApiError.validation("q is required.", [{ field: "q", issue: "required" }]);
    await this.enforceRecentWriteLimit(userId);
    await this.trackRecent(userId, raw);
    return this.recent(userId);
  }

  private async enforceRecentWriteLimit(userId: string): Promise<void> {
    try {
      const key = `search:recent:rl:${userId}`;
      const now = Date.now();
      const pipe = this.redis.getRedis().pipeline();
      pipe.zremrangebyscore(key, 0, now - RECENT_WRITE_WINDOW_MS);
      pipe.zadd(key, now, `${now}:${Math.random().toString(36).slice(2)}`);
      pipe.zcard(key);
      pipe.pexpire(key, RECENT_WRITE_WINDOW_MS);
      const results = await pipe.exec();
      const card = results?.[2]?.[1];
      const count = typeof card === "number" ? card : 1;
      if (count > RECENT_WRITE_LIMIT) {
        throw new ApiError("RATE_LIMITED", 429, "Too many recent-search writes.", [
          { retryAfterSeconds: Math.ceil(RECENT_WRITE_WINDOW_MS / 1000) },
        ]);
      }
    } catch (err) {
      // Degrade OPEN like the global guard (availability over strictness);
      // a 429 thrown above is re-thrown as-is.
      if (err instanceof ApiError) throw err;
      logger.warn(
        `recent-write rate limit degraded (redis unavailable): ${(err as Error).message}`,
      );
    }
  }

  async clearRecent(userId: string): Promise<{ cleared: boolean }> {
    try {
      await this.redis.getRedis().del(RECENT_KEY(userId));
      return { cleared: true };
    } catch {
      return { cleared: false };
    }
  }
}

/** limit: 1..50 (TASK-0097 §2; LIMITS.MAX_SEARCH_LIMIT = 50), default 20. */
function clampLimit(limit: number | undefined): number {
  if (limit === undefined) return 20;
  if (!Number.isInteger(limit) || limit < 1 || limit > 50) {
    throw ApiError.validation("limit must be an integer between 1 and 50.", [
      { field: "limit", issue: "1..50", received: String(limit) },
    ]);
  }
  return limit;
}

function clampSuggest(limit: number | undefined): number {
  if (limit === undefined) return SUGGEST_DEFAULT_SIZE;
  if (!Number.isInteger(limit) || limit < 1) {
    throw ApiError.validation("limit must be an integer between 1 and 20.", [
      { field: "limit", issue: "1..20", received: String(limit) },
    ]);
  }
  return Math.min(limit, SUGGEST_MAX_SIZE);
}