/**
 * T-156 — org-admin search reindex trigger.
 *
 * The reindex CORE is the worker's (search-index processor, `reindex` job —
 * the same one the 15-minute `search-reconcile` sweep and the
 * `workspace-resync` path enqueue). This service only (1) gates on
 * `assertOrgAdmin`, (2) AUDITS the trigger (a sensitive, expensive op), and
 * (3) enqueues the SAME job with `orgId` so the worker limits the scan to
 * THIS org's workspaces (one org's admin can never reindex — or rewrite the
 * grant snapshots of — another org's data).
 */
import { Inject, Injectable } from "@nestjs/common";
import type { Request } from "express";
import { randomUUID } from "node:crypto";
import { PrismaService } from "../../common/prisma.service.js";
import { RedisService } from "../../common/redis.service.js";
import { AuditService } from "../../common/audit.service.js";
import { jobTraceMetadata } from "../../common/trace-metadata.js";
import { ApiError } from "../../common/errors.js";
import { requireUser } from "../../common/types.js";
import { SearchService } from "../search/search.service.js";
import { assertOrgAdmin } from "./guards.js";

@Injectable()
export class OrgSearchService {
  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaService,
    @Inject(RedisService) private readonly redis: RedisService,
    @Inject(AuditService) private readonly audit: AuditService,
    @Inject(SearchService) private readonly search: SearchService
  ) {}

  /**
   * `POST /orgs/:orgId/admin/search/reindex` — enqueue a full (org-scoped)
   * content-index rebuild. Optional `workspaceId` narrows it to one workspace
   * (the console's per-workspace "reindex" action). 202 + the job id; the
   * rebuild is async (thousands of docs) and idempotent (source→index).
   */
  async enqueueReindex(
    orgId: string,
    req: Request,
    workspaceId?: string
  ): Promise<{ jobId: string; status: "processing"; workspaceId: string | null }> {
    const userId = requireUser(req);
    await assertOrgAdmin(this.prisma, orgId, userId);
    // A workspace-scoped run must be a workspace of THIS org.
    if (workspaceId) {
      const ws = await this.prisma.workspace.findUnique({
        where: { id: workspaceId },
        select: { organizationId: true }
      });
      if (!ws || ws.organizationId !== orgId) {
        // No-leak: do not reveal that the id exists in another org.
        throw ApiError.notFound("Workspace not found in this organization.");
      }
    }
    const jobId = randomUUID();
    await this.audit.audit(req, req.auth, {
      action: "search.reindex_requested",
      resourceType: "organization",
      resourceId: orgId,
      metadata: { orgId, workspaceId: workspaceId ?? null, jobId }
    });
    const { Queue } = await import("bullmq");
    const q = new Queue("search-index", {
      connection: this.redis.getRedis().options as never
    });
    try {
      await q.add(
        "search-index",
        { type: "reindex", ...(workspaceId ? { workspaceId } : {}), orgId },
        { jobId, ...jobTraceMetadata() }
      );
    } finally {
      await q.close();
    }
    return { jobId, status: "processing", workspaceId: workspaceId ?? null };
  }

  /**
   * T-177 — `GET /orgs/:orgId/admin/search/health` — the SEARCH HEALTH tab
   * (doc count, last reconcile, lag, index health).
   *
   *   - `sourceCount` — the `searchDocument` mirror rows for the org's
   *     workspaces (the SAME rows the worker's reconcile sweep treats as the
   *     source of truth — so `sourceCount` vs `docCount` is an apples-to-
   *     apples drift figure, not a raw-table count the index would never
   *     match);
   *   - `docCount`    — whole docs in OpenSearch (`type != "chunk"`, tenant-
   *     scoped to the org's workspaces); a search failure → `healthy:false`
   *     with the upstream error, NEVER a guessed 0;
   *   - `lagMs`       — now − the newest `searchDocument.indexedAt` (how
   *     stale the freshest indexed doc is; null when the org has no mirror
   *     rows yet);
   *   - `lastReconcileAt` — the Redis stamp the worker writes when a
   *     reconcile sweep completes (`search:reconcile:last`, T-177 worker
   *     change); null until the first sweep after the stamp was introduced.
   */
  async health(orgId: string, req: Request) {
    const userId = requireUser(req);
    await assertOrgAdmin(this.prisma, orgId, userId);
    const workspaces = await this.prisma.workspace.findMany({
      where: { organizationId: orgId },
      select: { id: true }
    });
    const workspaceIds = workspaces.map((w) => w.id);

    // Source of truth (Postgres) — always available.
    const [sourceCount, newest] = await Promise.all([
      this.prisma.searchDocument.count({ where: { workspaceId: { in: workspaceIds } } }),
      this.prisma.searchDocument.findFirst({
        where: { workspaceId: { in: workspaceIds } },
        // NULLs sort LAST in DESC (Postgres) → the first row is the newest
        // non-null indexedAt; null-safe below (no `not: null` filter — the
        // column is nullable and the filter type rejects `null`).
        orderBy: { indexedAt: "desc" },
        select: { indexedAt: true }
      })
    ]);
    const lastIndexedAt = newest?.indexedAt ?? null;
    const lagMs = lastIndexedAt ? Date.now() - lastIndexedAt.getTime() : null;

    const lastReconcileRaw = await this.redis
      .getRedis()
      .get("search:reconcile:last")
      .catch(() => null);

    // Index (OpenSearch) — may be down; that IS the health signal.
    let docCount: number | null = null;
    let healthy = true;
    let error: string | null = null;
    try {
      docCount = await this.search.searchClient().countDocs(workspaceIds.length ? workspaceIds : undefined);
    } catch (e) {
      healthy = false;
      // `ApiError.code` is the stable signal (SEARCH_UNAVAILABLE for any
      // OpenSearch failure); fall back to the error name for anything else.
      error =
        (e as { code?: string })?.code ??
        (e instanceof Error ? e.name : "SEARCH_UNAVAILABLE");
    }

    return {
      healthy,
      error,
      workspaces: workspaceIds.length,
      sourceCount,
      docCount,
      drift: docCount !== null ? docCount - sourceCount : null,
      lagMs,
      lastIndexedAt: lastIndexedAt ? lastIndexedAt.toISOString() : null,
      lastReconcileAt: lastReconcileRaw ?? null
    };
  }
}