1import { assert, it } from "@effect/vitest";
2import {
3 EventId,
4 MessageId,
5 ProjectId,
6 ProviderInstanceId,
7 ThreadId,
8 type OrchestrationV2DomainEvent,
9} from "@t3tools/contracts";
10import * as NodeServices from "@effect/platform-node/NodeServices";
11import * as DateTime from "effect/DateTime";
12import * as Deferred from "effect/Deferred";
13import * as Effect from "effect/Effect";
14import * as FileSystem from "effect/FileSystem";
15import * as Fiber from "effect/Fiber";
16import * as Layer from "effect/Layer";
17import * as Path from "effect/Path";
18import * as SqlClient from "effect/sql/SqlClient";
19
20import * as SqlitePersistence from "../persistence/Sqlite.ts";
21import * as ProjectionStore from "./ProjectionStore.ts";
22import * as ProjectStore from "./ProjectStore.ts";
23import * as ThreadSearch from "./ThreadSearch.ts";
24
25const layerTest = Layer.mergeAll(
26 ThreadSearch.layer,
27 ProjectionStore.layer,
28 ProjectStore.layer,
29).pipe(Layer.provideMerge(SqlitePersistence.layerMemory));
30
31const providerInstanceId = ProviderInstanceId.make("codex");
32const at = (minute: number) => DateTime.makeUnsafe(Date.UTC(2026, 8, 27, 0, minute));
33
34const createProject = (projectId: ProjectId) =>
35 Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) =>
36 projects.apply({
37 sequence: 0,
38 eventId: EventId.make(`created:${projectId}`),
39 aggregateKind: "project",
40 aggregateId: projectId,
41 occurredAt: DateTime.formatIso(at(0)),
42 commandId: null,
43 causationEventId: null,
44 correlationId: null,
45 metadata: {},
46 type: "project.created",
47 payload: {
48 projectId,
49 title: projectId,
50 workspaceRoot: `/work/${projectId}`,
51 defaultModelSelection: null,
52 scripts: [],
53 createdAt: DateTime.formatIso(at(0)),
54 updatedAt: DateTime.formatIso(at(0)),
55 },
56 }),
57 );
58
59const thread = (
60 threadId: ThreadId,
61 projectId: ProjectId,
62 overrides: { readonly archivedAt?: DateTime.Utc; readonly deletedAt?: DateTime.Utc } = {},
63): OrchestrationV2DomainEvent => ({
64 id: EventId.make(`created:${threadId}`),
65 type: "thread.created",
66 threadId,
67 providerInstanceId,
68 occurredAt: at(0),
69 payload: {
70 createdBy: "user",
71 creationSource: "web",
72 id: threadId,
73 projectId,
74 title: threadId,
75 providerInstanceId,
76 modelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
77 runtimeMode: "full-access",
78 interactionMode: "default",
79 branch: null,
80 worktreePath: null,
81 activeProviderThreadId: null,
82 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
83 forkedFrom: null,
84 createdAt: at(0),
85 updatedAt: at(0),
86 archivedAt: overrides.archivedAt ?? null,
87 settledOverride: null,
88 settledAt: null,
89 lastVisitedAt: null,
90 deletedAt: overrides.deletedAt ?? null,
91 },
92});
93
94const message = (
95 threadId: ThreadId,
96 id: string,
97 role: "user" | "assistant" | "system",
98 text: string,
99 options: { readonly minute?: number; readonly streaming?: boolean } = {},
100): OrchestrationV2DomainEvent => ({
101 id: EventId.make(`message:${id}`),
102 type: "message.updated",
103 threadId,
104 providerInstanceId,
105 occurredAt: at(options.minute ?? 1),
106 payload: {
107 createdBy: role === "user" ? "user" : "agent",
108 creationSource: role === "user" ? "web" : "provider",
109 id: MessageId.make(id),
110 threadId,
111 runId: null,
112 nodeId: null,
113 role,
114 text,
115 attachments: [],
116 streaming: options.streaming ?? false,
117 createdAt: at(options.minute ?? 1),
118 updatedAt: at(options.minute ?? 1),
119 },
120});
121
122it.layer(layerTest)("ThreadSearch", (it) => {
123 it.effect("returns one finished user or assistant match per active thread", () =>
124 Effect.gen(function* () {
125 const projections = yield* ProjectionStore.ProjectionStoreV2;
126 const search = yield* ThreadSearch.ThreadSearch;
127 const project = ProjectId.make("project:search");
128 const deletedProject = ProjectId.make("project:search-deleted");
129 yield* createProject(project);
130 yield* createProject(deletedProject);
131 yield* Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) =>
132 projects.apply({
133 sequence: 0,
134 eventId: EventId.make("deleted:project"),
135 aggregateKind: "project",
136 aggregateId: deletedProject,
137 occurredAt: DateTime.formatIso(at(2)),
138 commandId: null,
139 causationEventId: null,
140 correlationId: null,
141 metadata: {},
142 type: "project.deleted",
143 payload: { projectId: deletedProject, deletedAt: DateTime.formatIso(at(2)) },
144 }),
145 );
146
147 const both = ThreadId.make("thread:both");
148 const assistantOnly = ThreadId.make("thread:assistant");
149 const events: ReadonlyArray<OrchestrationV2DomainEvent> = [
150 thread(both, project),
151 message(both, "both-assistant", "assistant", "needle from the answer", { minute: 3 }),
152 message(both, "both-user-old", "user", "older needle question", { minute: 1 }),
153 message(both, "both-user-new", "user", "newer needle question", { minute: 2 }),
154 thread(assistantOnly, project),
155 message(assistantOnly, "assistant-only", "assistant", "needle in an answer"),
156 message(assistantOnly, "assistant-streaming", "user", "needle still typing", {
157 streaming: true,
158 }),
159 message(assistantOnly, "assistant-system", "system", "needle system prompt"),
160 thread(ThreadId.make("thread:archived"), project, { archivedAt: at(1) }),
161 message(ThreadId.make("thread:archived"), "archived", "user", "needle archived"),
162 thread(ThreadId.make("thread:deleted"), project, { deletedAt: at(1) }),
163 message(ThreadId.make("thread:deleted"), "deleted", "user", "needle deleted"),
164 thread(ThreadId.make("thread:orphaned"), deletedProject),
165 message(ThreadId.make("thread:orphaned"), "orphaned", "user", "needle orphaned"),
166 ];
167 yield* Effect.forEach(events, projections.apply, { discard: true });
168
169 const result = yield* search.search({ query: "NEEDLE", limit: 20 });
170 assert.deepEqual(
171 result.matches.map((match) => [match.threadId, match.source, match.snippet]),
172 [
173 [both, "user", "newer needle question"],
174 [assistantOnly, "assistant", "needle in an answer"],
175 ],
176 );
177 assert.lengthOf((yield* search.search({ query: "needle", limit: 1 })).matches, 1);
178 // LIKE wildcards in the query match literally.
179 assert.deepEqual((yield* search.search({ query: "ne%le" })).matches, []);
180 }),
181 );
182
183 it.effect("reports an unreadable match as a decode failure", () =>
184 Effect.gen(function* () {
185 const projections = yield* ProjectionStore.ProjectionStoreV2;
186 const search = yield* ThreadSearch.ThreadSearch;
187 const sql = yield* SqlClient.SqlClient;
188 const project = ProjectId.make("project:search-corrupt");
189 const threadId = ThreadId.make("thread:corrupt");
190 yield* createProject(project);
191 yield* Effect.forEach(
192 [thread(threadId, project), message(threadId, "corrupt", "user", "20260927")],
193 projections.apply,
194 { discard: true },
195 );
196 yield* sql`
197 UPDATE orchestration_v2_projection_messages
198 SET payload_json = json_set(payload_json, '$.text', 20260927)
199 WHERE message_id = 'corrupt'
200 `;
201
202 const error = yield* Effect.flip(search.search({ query: "0260927" }));
203 assert.equal(error.operation, "decode");
204 }),
205 );
206});
207
208it.effect("searches on the reader worker while the writer is busy", () =>
209 Effect.gen(function* () {
210 const fileSystem = yield* FileSystem.FileSystem;
211 const path = yield* Path.Path;
212 const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-search-" });
213 const layerFile = Layer.mergeAll(
214 ThreadSearch.layer,
215 ProjectionStore.layer,
216 ProjectStore.layer,
217 ).pipe(
218 Layer.provideMerge(SqlitePersistence.layerFromPath(path.join(directory, "state.sqlite"))),
219 Layer.provide(NodeServices.layer),
220 );
221
222 yield* Effect.gen(function* () {
223 const projections = yield* ProjectionStore.ProjectionStoreV2;
224 const search = yield* ThreadSearch.ThreadSearch;
225 const sql = yield* SqlClient.SqlClient;
226 const project = ProjectId.make("project:search-reader");
227 const threadId = ThreadId.make("thread:reader");
228 yield* createProject(project);
229 yield* Effect.forEach(
230 [thread(threadId, project), message(threadId, "reader", "user", "needle on the reader")],
231 projections.apply,
232 { discard: true },
233 );
234
235 // A write transaction holds the writer connection. A search on the writer would queue
236 // behind it and never finish; one on the reader worker does not wait.
237 const writerHeld = yield* Deferred.make<void>();
238 const releaseWriter = yield* Deferred.make<void>();
239 const writer = yield* Effect.forkChild(
240 sql.withTransaction(
241 Deferred.succeed(writerHeld, undefined).pipe(
242 Effect.andThen(Deferred.await(releaseWriter)),
243 ),
244 ),
245 );
246 yield* Deferred.await(writerHeld);
247 const result = yield* search.search({ query: "needle" });
248 yield* Deferred.succeed(releaseWriter, undefined);
249 yield* Fiber.join(writer);
250
251 assert.deepEqual(
252 result.matches.map((match) => match.threadId),
253 [threadId],
254 );
255 }).pipe(Effect.provide(layerFile));
256 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
257);