1
12import { Inject, Injectable } from "@nestjs/common";
13import type { Request } from "express";
14import { randomUUID } from "node:crypto";
15import { PrismaService } from "../../common/prisma.service.js";
16import { RedisService } from "../../common/redis.service.js";
17import { AuditService } from "../../common/audit.service.js";
18import { jobTraceMetadata } from "../../common/trace-metadata.js";
19import { ApiError } from "../../common/errors.js";
20import { requireUser } from "../../common/types.js";
21import { SearchService } from "../search/search.service.js";
22import { assertOrgAdmin } from "./guards.js";
23
24@Injectable()
25export class OrgSearchService {
26 constructor(
27 @Inject(PrismaService) private readonly prisma: PrismaService,
28 @Inject(RedisService) private readonly redis: RedisService,
29 @Inject(AuditService) private readonly audit: AuditService,
30 @Inject(SearchService) private readonly search: SearchService
31 ) {}
32
33
39 async enqueueReindex(
40 orgId: string,
41 req: Request,
42 workspaceId?: string
43 ): Promise<{ jobId: string; status: "processing"; workspaceId: string | null }> {
44 const userId = requireUser(req);
45 await assertOrgAdmin(this.prisma, orgId, userId);
46
47 if (workspaceId) {
48 const ws = await this.prisma.workspace.findUnique({
49 where: { id: workspaceId },
50 select: { organizationId: true }
51 });
52 if (!ws || ws.organizationId !== orgId) {
53
54 throw ApiError.notFound("Workspace not found in this organization.");
55 }
56 }
57 const jobId = randomUUID();
58 await this.audit.audit(req, req.auth, {
59 action: "search.reindex_requested",
60 resourceType: "organization",
61 resourceId: orgId,
62 metadata: { orgId, workspaceId: workspaceId ?? null, jobId }
63 });
64 const { Queue } = await import("bullmq");
65 const q = new Queue("search-index", {
66 connection: this.redis.getRedis().options as never
67 });
68 try {
69 await q.add(
70 "search-index",
71 { type: "reindex", ...(workspaceId ? { workspaceId } : {}), orgId },
72 { jobId, ...jobTraceMetadata() }
73 );
74 } finally {
75 await q.close();
76 }
77 return { jobId, status: "processing", workspaceId: workspaceId ?? null };
78 }
79
80
99 async health(orgId: string, req: Request) {
100 const userId = requireUser(req);
101 await assertOrgAdmin(this.prisma, orgId, userId);
102 const workspaces = await this.prisma.workspace.findMany({
103 where: { organizationId: orgId },
104 select: { id: true }
105 });
106 const workspaceIds = workspaces.map((w) => w.id);
107
108
109 const [sourceCount, newest] = await Promise.all([
110 this.prisma.searchDocument.count({ where: { workspaceId: { in: workspaceIds } } }),
111 this.prisma.searchDocument.findFirst({
112 where: { workspaceId: { in: workspaceIds } },
113
114
115
116 orderBy: { indexedAt: "desc" },
117 select: { indexedAt: true }
118 })
119 ]);
120 const lastIndexedAt = newest?.indexedAt ?? null;
121 const lagMs = lastIndexedAt ? Date.now() - lastIndexedAt.getTime() : null;
122
123 const lastReconcileRaw = await this.redis
124 .getRedis()
125 .get("search:reconcile:last")
126 .catch(() => null);
127
128
129 let docCount: number | null = null;
130 let healthy = true;
131 let error: string | null = null;
132 try {
133 docCount = await this.search.searchClient().countDocs(workspaceIds.length ? workspaceIds : undefined);
134 } catch (e) {
135 healthy = false;
136
137
138 error =
139 (e as { code?: string })?.code ??
140 (e instanceof Error ? e.name : "SEARCH_UNAVAILABLE");
141 }
142
143 return {
144 healthy,
145 error,
146 workspaces: workspaceIds.length,
147 sourceCount,
148 docCount,
149 drift: docCount !== null ? docCount - sourceCount : null,
150 lagMs,
151 lastIndexedAt: lastIndexedAt ? lastIndexedAt.toISOString() : null,
152 lastReconcileAt: lastReconcileRaw ?? null
153 };
154 }
155}
156
