org-admin-health.integration.test.ts

org-admin-health.integration.test.tstests/integration/org-admin-health.integration.test.tsraw ← integration19,451 B← Back to the app
1/**
2 * T-176 / T-177 / T-178 — org-admin HEALTH tabs (jobs + search + notifications).
3 *
4 * The three new org-admin routes are all `assertOrgAdmin`-gated server-side
5 * (unknown org → 404 no-leak; org MEMBER/GUEST/non-member → 403
6 * `requires_org_admin`). This spec proves the guard matrix on all three, the
7 * happy-path shape of each, the audited requeue mutation, and the
8 * "search down → healthy:false (200, never 500)" contract.
9 *
10 *   GET  /orgs/:orgId/admin/jobs             — per-queue counts (cluster scope)
11 *   GET  /orgs/:orgId/admin/jobs/failed      — recent failed jobs (DLQ view)
12 *   POST /orgs/:orgId/admin/jobs/requeue     — re-run a failed job (+audit)
13 *   GET  /orgs/:orgId/admin/search/health    — doc count / lag / last reconcile
14 *   GET  /orgs/:orgId/admin/notifications/health — in-app + email delivery
15 *
16 * Search is proven against the dev/CI OpenSearch the API already targets
17 * (SEARCH_URL, default localhost:9200) — the house pattern
18 * (search.integration.test.ts). When the stack is absent (a Docker-less
19 * authoring host) the whole suite self-SKIPS (a top-level stack gate), it
20 * never fails. On the dev stack the `factory-content` alias already resolves
21 * → we write through it (never touch it); on a fresh CI stack we create the
22 * plain index. Either way we seed 2 docs with a run-unique workspaceId and
23 * countDocs is TENANT-FILTERED → docCount === sourceCount === 2 exactly.
24 */
25import { afterAll, beforeAll, describe, expect, it } from "vitest";
26import { execSync } from "node:child_process";
27import { Queue, Worker } from "bullmq";
28import Redis from "ioredis";
29import { SearchClient, contentMapping } from "@factory/search";
30import { ErrorCodes } from "@factory/shared";
31import { getRedisUrl } from "./helpers/containers";
32import { createTestApp, type TestAppHandle } from "./helpers/api";
33import { authHeaders, createWorkspace, createUser } from "./helpers/seed";
34import type { PrismaClient } from "@prisma/client";
35 
36/** The OpenSearch the API talks to (API's SEARCH_URL default). House pattern:
37 *  the full dev/CI stack's search engine, NOT a container this spec controls
38 *  (the host can be disk-constrained — a fresh container would crash). */
39const SEARCH_URL = (process.env.SEARCH_URL ?? "http://localhost:9200").replace(/\/+$/, "");
40 
41function dockerAvailable(): boolean {
42  try {
43    execSync("docker info", { stdio: "ignore", timeout: 8_000 });
44    return true;
45  } catch {
46    return false;
47  }
48}
49 
50async function opensearchUp(url: string): Promise<boolean> {
51  try {
52    const res = await fetch(`${url}/_cluster/health`, { signal: AbortSignal.timeout(5_000) });
53    return res.ok;
54  } catch {
55    return false;
56  }
57}
58 
59/** Collection-time stack gate (Docker + OpenSearch) — self-skip when the dev
60 *  stack is absent (a Docker-less authoring host), never a fake pass. */
61const stackReady = dockerAvailable() && (await opensearchUp(SEARCH_URL));
62 
63let app: TestAppHandle;
64// The BullMQ Redis + queue handles are created in beforeAll (NOT at module
65// import time): createTestApp() sets process.env.REDIS_URL, so a module-level
66// `new Queue` would fall back to a local :6379 and hit a DIFFERENT Redis than
67// the app (the app's queues live in the testcontainers Redis).
68let redis: Redis;
69let qEmail: Queue;
70/** Direct client on the SAME OpenSearch the API targets (SEARCH_URL). */
71let os: SearchClient;
72let createdIndex = false;
73const json = async (res: Response): Promise<Record<string, unknown>> =>
74  (await res.json().catch(() => ({}))) as Record<string, unknown>;
75 
76const reasonOf = (body: Record<string, unknown>): string | undefined =>
77  ((body.error as { details?: Array<{ reason?: string }> })?.details?.[0]?.reason);
78 
79async function makeOrg(tok: string): Promise<string> {
80  const res = await app.request("POST", "/api/v1/orgs", {
81    headers: authHeaders(tok),
82    body: { name: `Health Org ${Date.now()}-${Math.random().toString(36).slice(2, 6)}` }
83  });
84  const body = (await json(res)) as { id?: string; organization?: { id?: string } };
85  const orgId = body.id ?? body.organization?.id;
86  if (!orgId) throw new Error(`makeOrg failed: ${res.status} ${await res.text()}`);
87  return orgId;
88}
89 
90async function addOrgMember(prisma: PrismaClient, orgId: string, userId: string, role: "ADMIN" | "MEMBER" | "GUEST") {
91  await prisma.organizationMember.upsert({
92    where: { organizationId_userId: { organizationId: orgId, userId } },
93    update: { role },
94    create: { organizationId: orgId, userId, role }
95  });
96}
97 
98const RUN = Date.now().toString(36) + Math.random().toString(36).slice(2, 6);
99 
100let owner: Awaited<ReturnType<typeof createUser>>;
101let admin: Awaited<ReturnType<typeof createUser>>;
102let member: Awaited<ReturnType<typeof createUser>>;
103let orgId = "";
104let wsId = "";
105let failedJobId = ""; // the UUID BullMQ assigned (the `email:{id}` dedup key is rejected by BullMQ 5.x — see the T-178 note)
106let seedNotifId = "";
107 
108// A whole doc (type page) — chunk docs are excluded from countDocs.
109function pageDoc(workspaceId: string, id: string): Record<string, unknown> {
110  return {
111    docId: `page:${id}`,
112    type: "page",
113    id,
114    workspaceId,
115    title: `Health page ${id}`,
116    body: "health doc body",
117    grants: { public: false, workspace: true, user: false, teams: [], users: [] },
118    archived: false,
119    url: `/w/${workspaceId}/page/${id}`,
120    createdAt: new Date().toISOString(),
121    updatedAt: new Date().toISOString()
122  };
123}
124 
125describe.skipIf(!stackReady)("T-176/177/178: org-admin health tabs", () => {
126  beforeAll(async () => {
127    // The API targets the dev/CI OpenSearch at SEARCH_URL (default
128    // localhost:9200) — the house pattern (search.integration.test.ts). Our
129    // direct client targets the same instance.
130    os = new SearchClient({ baseUrl: SEARCH_URL });
131    app = await createTestApp();
132    // Queue handles on the SAME Redis the app uses (REDIS_URL is now set).
133    const url = await getRedisUrl();
134    redis = new Redis(url, { maxRetriesPerRequest: null });
135    qEmail = new Queue("email", { connection: redis });
136    owner = await createUser(app.request, { name: "H Owner" });
137    admin = await createUser(app.request, { name: "H Admin" });
138    member = await createUser(app.request, { name: "H Member" });
139    orgId = await makeOrg(owner.accessToken!);
140    await addOrgMember(app.prisma, orgId, admin.userId, "ADMIN");
141    await addOrgMember(app.prisma, orgId, member.userId, "MEMBER");
142    const ws = await createWorkspace(app.request, authHeaders(owner.accessToken!), `H WS ${RUN}`);
143    wsId = ws.workspaceId;
144    await app.prisma.workspace.update({ where: { id: wsId }, data: { organizationId: orgId } });
145 
146    // Index bootstrap (house pattern, search.integration.test.ts): on the DEV
147    // stack the `factory-content` alias already resolves (to factory-content-vN)
148    // → write through it, NEVER touch the alias. On a FRESH CI stack it is
149    // absent → create the plain index with the canonical mapping. Either way we
150    // seed with a run-unique workspaceId, and countDocs is TENANT-FILTERED, so
151    // docCount is exactly the docs WE seed regardless of the shared index's
152    // other content.
153    if (!(await os.indexExists("factory-content"))) {
154      const map = contentMapping();
155      await os.createIndexIfMissing("factory-content", {
156        settings: map.settings as Record<string, unknown>,
157        mappings: map.mappings as Record<string, unknown>
158      });
159      createdIndex = true;
160    }
161 
162    // A real FAILED email job — a short-lived failing worker consumes it
163    // (attempts:1 → failed). The email queue has no other integration writer,
164    // so this is interference-free and realistic. The job's payload mirrors the
165    // worker's `buildInstantEmailData` output (`notificationIds: [id]`); the
166    // service correlates on the PAYLOAD, and BullMQ assigns a UUID id (the
167    // worker's `email:{id}` dedup key is rejected by BullMQ 5.x — T-178 note).
168    const notif = await app.prisma.notification.create({
169      data: {
170        workspaceId: wsId,
171        userId: member.userId,
172        type: "COMMENT",
173        channel: "IN_APP",
174        title: "Health fail probe"
175      }
176    });
177    seedNotifId = notif.id;
178    const w = new Worker("email", () => Promise.reject(new Error("boom-t178")), {
179      connection: redis,
180      concurrency: 1
181    });
182    const job = await qEmail.add("email-notification", { notificationIds: [notif.id] }, {
183      attempts: 1
184    });
185    await w.waitUntilReady();
186    // Poll for the failed state (a throwaway worker is the only thing that can
187    // move a waiting job to failed — moveToFailed requires the active list).
188    const deadline = Date.now() + 15_000;
189    while (Date.now() < deadline) {
190      const state = await (await qEmail.getJob(job.id!))?.getState();
191      if (state === "failed") break;
192      if (state === "completed") throw new Error(`seed job ${job.id} completed, expected failed`);
193      await new Promise((r) => setTimeout(r, 50));
194    }
195    expect(await (await qEmail.getJob(job.id!))?.getState()).toBe("failed");
196    await w.close();
197    failedJobId = job.id!;
198  }, 180_000);
199 
200  afterAll(async () => {
201    await qEmail?.close();
202    await redis?.quit().catch(() => undefined);
203    // Index cleanup: a FRESH-stack plain index (we created it) is dropped
204    // whole; a LIVE dev-stack index is only pruned of THIS run's tenant docs
205    // (delete-by-query on our unique workspaceId) — the shared alias is never
206    // touched.
207    if (createdIndex) {
208      await fetch(`${SEARCH_URL}/factory-content?expand_wildcards=all`, { method: "DELETE" }).catch(
209        () => undefined
210      );
211    } else if (wsId) {
212      await fetch(`${SEARCH_URL}/factory-content/_delete_by_query`, {
213        method: "POST",
214        body: JSON.stringify({ query: { term: { workspaceId: wsId } } }),
215        headers: { "content-type": "application/json" }
216      }).catch(() => undefined);
217    }
218    // Remove the org (cascades its workspaces / notifications / search docs).
219    if (app && orgId) {
220      const orgWorkspaces = await app.prisma.workspace.findMany({
221        where: { organizationId: orgId },
222        select: { id: true }
223      });
224      await app.prisma.organization.deleteMany({ where: { id: orgId } });
225      await app.prisma.workspace.deleteMany({ where: { id: { in: orgWorkspaces.map((w) => w.id) } } });
226    }
227    await app?.close();
228  }, 30_000);
229 
230  it("guard matrix: owner+admin pass; member/non-member → 403; unknown org → 404 (all 3 routes)", async () => {
231    const paths = ["/jobs", "/search/health", "/notifications/health"];
232    for (const p of paths) {
233      const ok = await app.request("GET", `/api/v1/orgs/${orgId}/admin${p}`, {
234        headers: authHeaders(owner.accessToken!)
235      });
236      expect(ok.status).toBe(200);
237      const okA = await app.request("GET", `/api/v1/orgs/${orgId}/admin${p}`, {
238        headers: authHeaders(admin.accessToken!)
239      });
240      expect(okA.status).toBe(200);
241      const denied = await app.request("GET", `/api/v1/orgs/${orgId}/admin${p}`, {
242        headers: authHeaders(member.accessToken!)
243      });
244      expect(denied.status).toBe(403);
245      expect(reasonOf(await json(denied))).toBe("requires_org_admin");
246    }
247    // non-member (no org membership at all) → 403 on one route
248    const outsider = await createUser(app.request, { name: "H Outsider" });
249    const no403 = await app.request("GET", `/api/v1/orgs/${orgId}/admin/jobs`, {
250      headers: authHeaders(outsider.accessToken!)
251    });
252    expect(no403.status).toBe(403);
253    expect(reasonOf(await json(no403))).toBe("requires_org_admin");
254    // unknown org → 404 (no-leak)
255    const unknown = await app.request("GET", `/api/v1/orgs/does-not-exist/admin/jobs`, {
256      headers: authHeaders(owner.accessToken!)
257    });
258    expect(unknown.status).toBe(404);
259  });
260 
261  it("GET jobs: cluster scope, 7 queues each with 4 state counters", async () => {
262    const res = await app.request("GET", `/api/v1/orgs/${orgId}/admin/jobs`, {
263      headers: authHeaders(owner.accessToken!)
264    });
265    expect(res.status).toBe(200);
266    const body = (await json(res)) as {
267      scope: string;
268      queues: Array<{ name: string; waiting: number; active: number; completed: number; failed: number }>;
269    };
270    expect(body.scope).toBe("cluster");
271    expect(body.queues.map((q) => q.name).sort()).toEqual(
272      ["ai", "automations", "email", "files", "maintenance", "search-index", "webhooks"]
273    );
274    for (const q of body.queues) {
275      for (const k of ["waiting", "active", "completed", "failed"]) {
276        expect(typeof q[k]).toBe("number");
277      }
278    }
279  });
280 
281  it("GET search/health: sourceCount, docCount (delta = seeded), lag + lastIndexed", async () => {
282    // Seed 2 mirror rows + 2 whole docs in the controlled index.
283    const docA = `pgA-${RUN}`;
284    const docB = `pgB-${RUN}`;
285    await app.prisma.searchDocument.createMany({
286      data: [
287        {
288          workspaceId: wsId,
289          type: "page",
290          refId: docA,
291          title: `A`,
292          permissions: { public: false, workspace: true } as object
293        },
294        {
295          workspaceId: wsId,
296          type: "page",
297          refId: docB,
298          title: `B`,
299          permissions: { public: false, workspace: true } as object
300        }
301      ]
302    });
303    await os.indexDocs([
304      pageDoc(wsId, docA) as never,
305      pageDoc(wsId, docB) as never
306    ]);
307    // Force a refresh so the just-indexed docs are visible to _search (OpenSearch
308    // auto-refreshes only every ~1s; a deterministic count needs it immediate).
309    await os.refresh();
310 
311    const res = await app.request("GET", `/api/v1/orgs/${orgId}/admin/search/health`, {
312      headers: authHeaders(owner.accessToken!)
313    });
314    expect(res.status).toBe(200);
315    const body = (await json(res)) as {
316      healthy: boolean;
317      error: string | null;
318      workspaces: number;
319      sourceCount: number;
320      docCount: number | null;
321      drift: number | null;
322      lagMs: number | null;
323      lastIndexedAt: string | null;
324      lastReconcileAt: string | null;
325    };
326    expect(body.healthy).toBe(true);
327    expect(body.error).toBeNull();
328    expect(body.workspaces).toBe(1);
329    expect(body.sourceCount).toBe(2);
330    // countDocs is TENANT-FILTERED (terms workspaceId=[wsId]) and our wsId is
331    // run-unique, so the count is exactly the 2 whole docs we seeded — even on
332    // the shared dev index that holds other tenants' docs.
333    expect(body.docCount).toBe(2);
334    expect(body.drift).toBe(0);
335    expect(body.lastIndexedAt).toBeTruthy();
336    expect(typeof body.lagMs).toBe("number");
337    expect(body.lagMs as number).toBeGreaterThanOrEqual(0);
338  });
339 
340  it("GET search/health with OpenSearch down → healthy:false, 200 (never 500)", async () => {
341    // Same module instance the booted AppModule uses (the relative path
342    // resolves to the file the app's SearchModule imports — the audit-completeness
343    // spec uses this exact `app.get(Service)` + relative-import pattern).
344    const { SearchService } = await import(
345      /* webpackIgnore: true */ "../../apps/api/src/modules/search/search.service.ts"
346    );
347    const ss = app.get!(SearchService) as {
348      setClientForTesting: (c: unknown) => void;
349    } | null;
350    expect(ss).toBeTruthy();
351    // A client whose every call throws SEARCH_UNAVAILABLE (a dead endpoint).
352    const dead = new SearchClient({
353      baseUrl: "http://127.0.0.1:1",
354      fetchImpl: (async () => {
355        throw new TypeError("fetch failed");
356      }) as typeof fetch
357    });
358    ss!.setClientForTesting(dead);
359    try {
360      const res = await app.request("GET", `/api/v1/orgs/${orgId}/admin/search/health`, {
361        headers: authHeaders(owner.accessToken!)
362      });
363      expect(res.status).toBe(200);
364      const body = (await json(res)) as { healthy: boolean; error: string; docCount: number | null };
365      expect(body.healthy).toBe(false);
366      expect(body.error).toBe(ErrorCodes.SEARCH_UNAVAILABLE);
367      expect(body.docCount).toBeNull();
368    } finally {
369      ss!.setClientForTesting(null); // restore the real client
370    }
371  });
372 
373  it("GET notifications/health: in-app volume + email failure correlated to org notif", async () => {
374    const res = await app.request("GET", `/api/v1/orgs/${orgId}/admin/notifications/health`, {
375      headers: authHeaders(owner.accessToken!)
376    });
377    expect(res.status).toBe(200);
378    const body = (await json(res)) as {
379      workspaces: number;
380      inAppTotal: number;
381      inAppUnread: number;
382      emailQueued: number;
383      emailFailed: number;
384      failedJobs: Array<{ notificationId: string; jobId: string; failedReason: string | null; attemptsMade: number }>;
385    };
386    expect(body.workspaces).toBe(1);
387    expect(body.inAppTotal).toBeGreaterThanOrEqual(1);
388    expect(body.inAppUnread).toBeGreaterThanOrEqual(1);
389    expect(typeof body.emailQueued).toBe("number");
390    expect(body.emailFailed).toBeGreaterThanOrEqual(1);
391    // The email queue is SHARED (a prior run may have left failed jobs), so
392    // locate OUR job by its payload notificationId (the service correlates on
393    // the payload, not the jobId) — never assume it is first.
394    const f = body.failedJobs.find((j) => j.notificationId === seedNotifId);
395    expect(f).toBeTruthy();
396    expect(f!.jobId).toBe(failedJobId); // the UUID BullMQ assigned
397    expect(f!.notificationId).toBe(seedNotifId);
398    expect(f!.failedReason).toContain("boom-t178");
399    expect(f!.attemptsMade).toBe(1);
400  });
401 
402  it("GET jobs/failed (queue=email) lists the failed email job", async () => {
403    const res = await app.request("GET", `/api/v1/orgs/${orgId}/admin/jobs/failed?queue=email`, {
404      headers: authHeaders(owner.accessToken!)
405    });
406    expect(res.status).toBe(200);
407    const body = (await json(res)) as { scope: string; jobs: Array<{ queue: string; jobId: string }> };
408    expect(body.scope).toBe("cluster");
409    expect(body.jobs.some((j) => j.queue === "email" && j.jobId === failedJobId)).toBe(true);
410  });
411 
412  it("POST jobs/requeue: non-admin 403; admin 200 + job leaves failed + audit row", async () => {
413    // non-admin → 403
414    const denied = await app.request("POST", `/api/v1/orgs/${orgId}/admin/jobs/requeue`, {
415      headers: authHeaders(member.accessToken!),
416      body: { queue: "email", jobId: failedJobId }
417    });
418    expect(denied.status).toBe(403);
419    expect(reasonOf(await json(denied))).toBe("requires_org_admin");
420 
421    // admin → 200; the job leaves the failed state (back to waiting).
422    const ok = await app.request("POST", `/api/v1/orgs/${orgId}/admin/jobs/requeue`, {
423      headers: authHeaders(admin.accessToken!),
424      body: { queue: "email", jobId: failedJobId }
425    });
426    expect(ok.status).toBe(200);
427    expect((await json(ok))).toMatchObject({ queued: true, queue: "email", jobId: failedJobId });
428    const state = await (await qEmail.getJob(failedJobId))?.getState();
429    expect(state).not.toBe("failed");
430 
431    // The requeue is a sensitive mutation → audited. Org routes are
432    // workspace-independent, so filter by action + the org in metadata (the
433    // house pattern the reindex spec uses; the row's resourceType is "queue").
434    const audits = await app.prisma.auditLog.findMany({
435      where: { action: "jobs.failed_job_requeued", metadata: { path: ["orgId"], equals: orgId } }
436    });
437    expect(audits.length).toBe(1);
438    expect(audits[0]!.metadata).toMatchObject({ queue: "email", jobId: failedJobId });
439  });
440});
441