/**
 * AI Software Factory — E2E index-lag helpers (TASK-0100 req 1 + Backend
 * Changes + Security).
 *
 * These are API / worker-surface helpers (the lag SLAs are measured without a
 * browser; only `measureCreateLagViaSearchUi` drives the real UI, and
 * `runSearchIndexBurst` publishes directly to the worker's Redis input stream)
 * so the lag SLA can be measured deterministically against the REAL pipeline:
 *
 *   page write → API publishes `factory:events` → worker stream consumer →
 *   OpenSearch upsert → query-time permission filter → POST /api/v1/search.
 *
 *   - `waitUntilSearchable`   create-side lag (t0 → first hit, 2 s polls, < 30 s)
 *   - `waitUntilUnsearchable` delete-side  lag (t0 → last hit,  2 s polls, < 30 s)
 *   - `assertNeverSearchable` permission negative (never visible in a window)
 *   - `getSearchIndexLag`     the worker's `search-index:lag` metric (worker
 *                             health server `/health/search-index:lag`)
 *   - `runSearchIndexBurst`   1k-event burst → worker drains the backlog
 *   - `measureCreateLagViaSearchUi` UI-side create lag (page via UI → found in
 *                             the search dialog → exact-page URL)
 *   - `setPagePrivateDirectDb`the one field no public API sets (Page.isPrivate)
 *
 * All lag helpers resolve the measured `lagMs` and the number of 2 s poll
 * rounds, so a caller can log the curve / attach it to the completion note.
 */
import { PrismaClient } from "@prisma/client";
import { expect, type APIRequestContext, type Locator, type Page } from "@playwright/test";

/** Index-lag SLA (spec req 1): create + delete must be < 30 s. */
export const SEARCH_LAG_TIMEOUT_MS = 30_000;
/** Spec-mandated poll cadence for POST /api/v1/search. */
export const SEARCH_POLL_INTERVAL_MS = 2_000;
/**
 * TASK-0100 S5: the search dialog DEBOUNCES its input (SEARCH_DEBOUNCE_MS =
 * 225 ms) and then runs the query + suggest fetches, so a result row renders
 * ≈ 400 ms AFTER `input.fill(...)`. A one-shot `row.isVisible()` immediately
 * after the fill therefore sees the pre-debounce state (no rows yet) even
 * when the page IS already searchable (the API finds it in < 2 s) — the
 * helpers must bound-wait for the row within each poll round.
 */
export const SEARCH_ROW_RENDER_WINDOW_MS = 2_500;

/**
 * Bounded wait for a search-result row to render (dialog debounce + fetch +
 * paint). Returns true when the row becomes visible within the window.
 * Shared by the create-lag UI poll and the click-through re-open so both
 * honour the dialog's debounce instead of racing it.
 */
export async function waitForSearchRow(row: Locator, windowMs: number = SEARCH_ROW_RENDER_WINDOW_MS): Promise<boolean> {
  try {
    await expect(row.first()).toBeVisible({ timeout: windowMs });
    return true;
  } catch {
    return false;
  }
}

/**
 * Plain sleep. `APIRequestContext` has NO `waitForTimeout` (that is a Page /
 * Locator method) — the polling helpers are browser-free on purpose (they run
 * against the API/worker surface), so they sleep on `setTimeout`.
 */
const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/** Worker health server port (docs/development/development-guide.md). */
export const WORKER_HEALTH_URL = process.env.WORKER_HEALTH_URL ?? "http://localhost:4003";

export interface LagResult {
  /** Milliseconds from the caller's t0 to the terminal observation. */
  lagMs: number;
  /** How many 2 s poll rounds ran before the terminal observation. */
  polls: number;
  /** Server-side `tookMs` of the terminal query (cross-check, optional). */
  tookMs?: number;
}

interface SearchEnvelope {
  data?: { id?: string; docId?: string; url?: string; title?: string; snippet?: string }[];
  nextCursor?: string | null;
  tookMs?: number;
}

export interface SearchResponse {
  status: number;
  rows: SearchEnvelope["data"];
  tookMs?: number;
  /** True when the API answered `SEARCH_UNAVAILABLE` (fail-closed). */
  unavailable?: boolean;
}

/**
 * One POST /api/v1/search (the live TASK-0097 contract: `{ q }` body,
 * envelope `{ data, nextCursor, tookMs }`).
 *
 * "Unavailable" is detected by the `SEARCH_UNAVAILABLE` ERROR CODE, not the
 * HTTP status: the API's `mapSharedError` propagates the upstream OpenSearch
 * status (503 when OS is down, but e.g. 400 while the index alias still has
 * no backing index, 429 on throttling — all fail-closed `SEARCH_UNAVAILABLE`).
 * Unavailable responses return empty rows + `unavailable: true` — the
 * polling helpers treat them as "not yet" and keep going to the SLA deadline
 * (a persistent unavailability then surfaces as the SLA timeout).
 */
export async function search(ctx: APIRequestContext, q: string, limit = 10): Promise<SearchResponse> {
  const res = await ctx.post("/api/v1/search", { data: { q, limit } });
  if (res.status() !== 200) {
    const text = await res.text();
    let code: unknown;
    try {
      code = (JSON.parse(text) as { error?: { code?: unknown } }).error?.code;
    } catch {
      code = undefined; // non-JSON error body — treat as a hard failure below
    }
    if (code === "SEARCH_UNAVAILABLE") return { status: res.status(), rows: [], tookMs: 0, unavailable: true };
    throw new Error(`search: POST /api/v1/search "${q}" → ${res.status}: ${text.slice(0, 300)}`);
  }
  const body = (await res.json()) as SearchEnvelope;
  return { status: 200, rows: body.data ?? [], tookMs: body.tookMs };
}

/**
 * Does the result set contain the doc? The search `q` is a globally unique
 * token (a uuid), so a substring match across the serialized rows cannot
 * false-positive on unrelated docs.
 */
function rowsContain(rows: SearchResponse["rows"], q: string): boolean {
  return (rows ?? []).some((r) => JSON.stringify(r).includes(q));
}

/**
 * Create-side lag (spec req 1 a–d): poll POST /api/v1/search every 2 s from
 * `t0`, PASS when `q` appears, FAIL after the SLA. Returns the measured lag.
 */
export async function waitUntilSearchable(
  ctx: APIRequestContext,
  q: string,
  t0: number,
  opts: { timeoutMs?: number; intervalMs?: number } = {},
): Promise<LagResult> {
  const timeoutMs = opts.timeoutMs ?? SEARCH_LAG_TIMEOUT_MS;
  const intervalMs = opts.intervalMs ?? SEARCH_POLL_INTERVAL_MS;
  const deadline = Date.now() + timeoutMs;
  let polls = 0;
  let lastStatus = 0;
  // eslint-disable-next-line no-constant-condition
  while (true) {
    polls += 1;
    const { status, rows, tookMs } = await search(ctx, q).catch(() => ({
      status: 0,
      rows: [] as SearchResponse["rows"],
      tookMs: 0,
      unavailable: true,
    }));
    lastStatus = status;
    if (rowsContain(rows, q)) {
      return { lagMs: Date.now() - t0, polls, tookMs };
    }
    if (Date.now() >= deadline) {
      throw new Error(
        `index lag (create): "${q}" not searchable within ${timeoutMs} ms after ${polls} polls ` +
          `(last search status ${lastStatus})`,
      );
    }
    await sleep(intervalMs);
  }
}

/**
 * Delete-side lag (spec req 1 / Tests): the doc was visible; poll until it is
 * gone (the worker removes the doc on `page.deleted`). Returns the measured
 * lag from the deletion t0.
 */
export async function waitUntilUnsearchable(
  ctx: APIRequestContext,
  q: string,
  t0: number,
  opts: { timeoutMs?: number; intervalMs?: number } = {},
): Promise<LagResult> {
  const timeoutMs = opts.timeoutMs ?? SEARCH_LAG_TIMEOUT_MS;
  const intervalMs = opts.intervalMs ?? SEARCH_POLL_INTERVAL_MS;
  const deadline = Date.now() + timeoutMs;
  let polls = 0;
  // eslint-disable-next-line no-constant-condition
  while (true) {
    polls += 1;
    const { rows } = await search(ctx, q).catch(() => ({ rows: [] as SearchResponse["rows"] }));
    if (!rowsContain(rows, q)) {
      return { lagMs: Date.now() - t0, polls };
    }
    if (Date.now() >= deadline) {
      throw new Error(
        `index lag (delete): "${q}" still searchable ${timeoutMs} ms after deletion ` +
          `(the worker must remove the doc on page.deleted)`,
      );
    }
    await sleep(intervalMs);
  }
}

/**
 * Permission negative (spec Security): a doc that must NEVER be visible to
 * this viewer — e.g. another member's PRIVATE page — is polled for the whole
 * window and must return ZERO rows the entire time. Fails the moment the doc
 * appears.
 */
export async function assertNeverSearchable(
  ctx: APIRequestContext,
  q: string,
  opts: { durationMs?: number; intervalMs?: number; description?: string } = {},
): Promise<{ polls: number }> {
  const durationMs = opts.durationMs ?? SEARCH_LAG_TIMEOUT_MS;
  const intervalMs = opts.intervalMs ?? SEARCH_POLL_INTERVAL_MS;
  const deadline = Date.now() + durationMs;
  let polls = 0;
  // eslint-disable-next-line no-constant-condition
  while (true) {
    polls += 1;
    const { rows } = await search(ctx, q).catch(() => ({ rows: [] as SearchResponse["rows"] }));
    if (rowsContain(rows, q)) {
      throw new Error(
        `${opts.description ?? "permission negative"}: "${q}" WAS visible after ${polls} poll(s) — ` +
          `it must never be found`,
      );
    }
    if (Date.now() >= deadline) break;
    await sleep(intervalMs);
  }
  return { polls };
}

// ---------------------------------------------------------------------------
// UI-side create lag (spec req 1, test 1: "create page (UI) → searchable < 30 s")
// ---------------------------------------------------------------------------

export interface CreateLagViaUiOptions {
  /** Fired once the new page's id is known (read from the post-create URL). */
  onCreated?: (pageId: string) => void;
  /** SLA deadline (default 30 s). */
  timeoutMs?: number;
}

/**
 * Create-side index lag measured through the REAL search UI (spec req 1 /
 * Tests test 1). Drives the actual browser surface so the event path is the
 * genuine one (page.created + the page.updated title write → `factory:events`
 * → worker → OpenSearch):
 *
 *   1. navigate to the workspace (`/w/{id}` — the API resolves by id; the
 *      search dialog scopes by the resolved workspace object, not the param)
 *   2. click the sidebar "New page" button (aria-label="New page")
 *   3. type the globally-unique `token` into the title input (aria-label="Page
 *      title") and blur it so the optimistic title mutation flushes
 *   4. open the ⌘K dialog, type `token`, and poll (2 s) until a `search-result`
 *      row containing `token` is visible
 *
 * Returns the measured lag (t0 → first visible result row) and fires
 * `onCreated(pageId)` once the sidebar's `router.push(/page/{id})` has made the
 * id readable from the URL. The search dialog is CLOSED again before returning
 * so a caller can re-open it deterministically (the button toggles).
 */
export async function measureCreateLagViaSearchUi(
  page: Page,
  workspaceId: string,
  token: string,
  opts: CreateLagViaUiOptions = {},
): Promise<LagResult> {
  const timeoutMs = opts.timeoutMs ?? SEARCH_LAG_TIMEOUT_MS;
  const intervalMs = SEARCH_POLL_INTERVAL_MS;

  await page.goto(`/w/${workspaceId}`);
  const newPageBtn = page.getByRole("button", { name: "New page" });
  await expect(newPageBtn, 'the sidebar "New page" button').toBeVisible({ timeout: 15_000 });

  const t0 = Date.now();
  await newPageBtn.click();

  // The sidebar creates the page (title "") and router.push's to /page/{id}.
  const title = page.getByRole("textbox", { name: "Page title" });
  await expect(title, 'the "Page title" input').toBeVisible({ timeout: 15_000 });
  await title.click();
  await title.fill(token);
  await title.press("Tab"); // blur → flush the optimistic title mutation

  // The sidebar pushed /w/{slugOrId}/page/{id} — the id is the last segment.
  const pageId = new URL(page.url()).pathname.split("/").filter(Boolean).pop() ?? "";
  if (pageId) opts.onCreated?.(pageId);

  const deadline = t0 + timeoutMs;
  let polls = 0;
  const dialog = page.getByTestId("search-dialog");
  const input = page.getByTestId("search-input");
  const row = page.getByTestId("search-result").filter({ hasText: token });
  // eslint-disable-next-line no-constant-condition
  for (;;) {
    polls += 1;
    await page.getByTestId("search-button").click();
    await expect(dialog).toBeVisible();
    await input.fill(token);
    // Bounded wait: the dialog debounces the fill (225 ms) + fetches, so a
    // one-shot isVisible() always saw the pre-debounce state.
    const visible = await waitForSearchRow(row);
    if (visible) {
      await page.keyboard.press("Escape"); // leave the dialog closed for the caller
      return { lagMs: Date.now() - t0, polls };
    }
    await page.keyboard.press("Escape");
    if (Date.now() >= deadline) {
      throw new Error(
        `measureCreateLagViaSearchUi: "${token}" not found in the search UI within ` +
          `${timeoutMs} ms after ${polls} polls`,
      );
    }
    await sleep(intervalMs);
  }
}

// ---------------------------------------------------------------------------
// Worker consumer-lag metric (search-index:lag) — the burst assertion
// ---------------------------------------------------------------------------

export interface SearchIndexLag {
  metric: "search-index:lag";
  /** False when the endpoint is missing / the worker predates TASK-0100. */
  available: boolean;
  /** Delivered to the group but not ACKed (in-flight / redelivery). */
  pending: number;
  /** Stream entries not yet delivered to the group (tail − last-delivered). */
  lagEntries: number;
  /** pending + lagEntries — the backlog the burst assertion drains. */
  backlog: number;
}

/**
 * Read `GET /health/search-index:lag` from the worker health server.
 * Resolves `{ available: false, ... }` (never throws) when the worker does not
 * expose the metric — callers skip the assertion instead of failing the run.
 */
export async function getSearchIndexLag(): Promise<SearchIndexLag> {
  const fallback: SearchIndexLag = {
    metric: "search-index:lag",
    available: false,
    pending: 0,
    lagEntries: 0,
    backlog: 0
  };
  try {
    const res = await fetch(`${WORKER_HEALTH_URL}/health/search-index:lag`, {
      signal: AbortSignal.timeout(5_000)
    });
    if (res.status !== 200) return fallback;
    const body = (await res.json()) as Partial<SearchIndexLag>;
    if (typeof body.available !== "boolean") return fallback;
    return {
      metric: "search-index:lag",
      available: body.available,
      pending: body.pending ?? 0,
      lagEntries: body.lagEntries ?? 0,
      backlog: body.backlog ?? 0
    };
  } catch {
    return fallback;
  }
}

export interface BurstResult {
  /** Number of events published to the stream. */
  published: number;
  /** Wall-clock ms to XADD the whole burst. */
  burstMs: number;
  /** Wall-clock ms from burst start to backlog === 0 (the drain). */
  drainMs: number;
  /** Per-interval backlog samples (tMs from burst start; -1 = unavailable). */
  samples: { tMs: number; backlog: number }[];
}

/**
 * Burst assertion (spec Backend Changes: "verify the worker consumes a 1k-event
 * burst without backlog growth > 30 s — assert in the lag suite via the
 * consumer-lag metric `search-index:lag`").
 *
 * The events are published DIRECTLY to the worker's input stream
 * (`factory:events`) as real `page.updated` envelopes referencing `pageIds` —
 * each one makes the worker run a genuine re-index (DB read + OpenSearch
 * upsert), so this is real indexing load, not a no-op. Publishing to the
 * stream (rather than through the public API) is deliberate: the API's
 * per-user rate limit (100/min) makes a true 1k-event burst infeasible via
 * HTTP, while the stream is the worker's actual input and exactly what the
 * `search-index:lag` metric observes. `pageIds` must be REAL pages (created
 * via the API first) so the events take the index path, not the
 * missing-source (poison) path.
 *
 * The assertion: after XADD-ing `count` events, the worker's backlog
 * (`pending + lagEntries` from `/health/search-index:lag`) must reach 0
 * within the SLA (default 30 s). Throws with the backlog curve on failure.
 */
export async function runSearchIndexBurst(opts: {
  /** Redis DSN (CI e2e job env / local dev .env; default redis://localhost:6379). */
  redisUrl?: string;
  /** Real page ids to cycle the `page.updated` events over. */
  pageIds: string[];
  workspaceId: string;
  /** Burst size (spec: 1000). */
  count?: number;
  timeoutMs?: number;
  intervalMs?: number;
}): Promise<BurstResult> {
  const { redisUrl, pageIds, workspaceId } = opts;
  const count = opts.count ?? 1_000;
  const timeoutMs = opts.timeoutMs ?? SEARCH_LAG_TIMEOUT_MS;
  const intervalMs = opts.intervalMs ?? SEARCH_POLL_INTERVAL_MS;
  if (pageIds.length === 0) throw new Error("runSearchIndexBurst: pageIds must be non-empty");

  // Lazy import so the rest of the helper (used by the non-burst tests) loads
  // without a Redis dependency at module-eval time.
  const { default: Redis } = await import("ioredis");
  const redis = new Redis(redisUrl ?? process.env.REDIS_URL ?? "redis://localhost:6379", {
    maxRetriesPerRequest: 2,
    lazyConnect: true
  });

  const t0 = Date.now();
  let burstMs = 0;
  try {
    await redis.connect().catch(() => {
      throw new Error(
        `runSearchIndexBurst: cannot connect to Redis (${redisUrl ?? process.env.REDIS_URL ?? "redis://localhost:6379"}) ` +
          `— the burst assertion needs the worker's Redis`,
      );
    });
    const STREAM = "factory:events";
    // One pipeline, `count` XADDs (fast — a single round-trip).
    const pipe = redis.pipeline();
    for (let i = 0; i < count; i += 1) {
      const pageId = pageIds[i % pageIds.length];
      const payload = JSON.stringify({ workspaceId, pageId, fields: ["title"] });
      pipe.xadd(STREAM, "*", "event", "page.updated", "payload", payload, "ts", new Date().toISOString());
    }
    const started = Date.now();
    await pipe.exec();
    burstMs = Date.now() - started;
  } finally {
    redis.disconnect();
  }

  // Drain: poll the worker's lag metric until the backlog is 0.
  const samples: BurstResult["samples"] = [];
  // eslint-disable-next-line no-constant-condition
  while (true) {
    const lag = await getSearchIndexLag();
    samples.push({ tMs: Date.now() - t0, backlog: lag.available ? lag.backlog : -1 });
    if (lag.available && lag.backlog === 0) {
      return { published: count, burstMs, drainMs: Date.now() - t0, samples };
    }
    if (Date.now() - t0 >= timeoutMs) {
      const curve = samples.map((s) => `${s.tMs}ms:${s.backlog}`).join(" ");
      throw new Error(
        `worker backlog did not drain within ${timeoutMs} ms after a ${count}-event burst ` +
          `(search-index:lag samples tMs:backlog = ${curve}; -1 = endpoint unavailable)`,
      );
    }
    await new Promise((r) => setTimeout(r, intervalMs));
  }
}

// ---------------------------------------------------------------------------
// Private-page seam (TASK-0100 Security negative)
// ---------------------------------------------------------------------------

/**
 * Direct-DB write for the ONE field no public API sets: `Page.isPrivate`
 * (UpdatePageDto has no privacy field — recorded quirk; the Phase-9 privacy UI
 * is the intended surface). The spec's negative test needs a private page
 * owned by one member that another member must NEVER find; only a source-row
 * write produces the correct `isPrivate` grant snapshot (the worker stamps it
 * from the row at index time; an API `page.updated` right after re-stamps it).
 *
 * Requires `DATABASE_URL` (CI e2e job env / local dev .env).
 */
export async function setPagePrivateDirectDb(pageId: string, isPrivate: boolean): Promise<boolean> {
  const dsn = process.env.DATABASE_URL;
  if (!dsn) {
    throw new Error(
      "setPagePrivateDirectDb: DATABASE_URL is not set — the private-page " +
        "negative test needs the DB DSN (CI e2e job sets it; local dev .env).",
    );
  }
  const prisma = new PrismaClient({ datasources: { db: { url: dsn } } });
  try {
    const updated = await prisma.page.update({
      where: { id: pageId },
      data: { isPrivate },
      select: { isPrivate: true }
    });
    expect(updated.isPrivate, `setPagePrivateDirectDb(${pageId}): DB write`).toBe(isPrivate);
    return updated.isPrivate;
  } finally {
    await prisma.$disconnect();
  }
}