Open History to see the last commit
1import * as Arr from "effect/Array";
2import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Result from "effect/Result";
8import * as Schema from "effect/Schema";
9import * as Struct from "effect/Struct";
10import * as SqlClient from "effect/sql/SqlClient";
11import * as SqlSchema from "effect/sql/SqlSchema";
12
13import {
14 AgentSessionImportSource,
15 IsoDateTime,
16 ProviderInstanceId,
17 RuntimeMode,
18 ThreadId,
19} from "@t3tools/contracts";
20
21import {
22 PersistenceDecodeError,
23 type PersistenceErrorCorrelation,
24 PersistenceSqlError,
25 type ProviderSessionRuntimeRepositoryError,
26} from "./Errors.ts";
27
28const ProviderSessionRuntimeStatus = Schema.Literals(["starting", "running", "stopped", "error"]);
29
30export const ProviderSessionRuntime = Schema.Struct({
31 threadId: ThreadId,
32 providerName: Schema.String,
33 providerInstanceId: Schema.NullOr(ProviderInstanceId),
34 adapterKey: Schema.String,
35 runtimeMode: RuntimeMode,
36 status: ProviderSessionRuntimeStatus,
37 lastSeenAt: IsoDateTime,
38 resumeCursor: Schema.NullOr(Schema.Unknown),
39 runtimePayload: Schema.NullOr(Schema.Unknown),
40});
41export type ProviderSessionRuntime = typeof ProviderSessionRuntime.Type;
42
43export const GetProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId });
44export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInput.Type;
45
46export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId });
47export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type;
48
49export const RecordImportedTranscriptInput = Schema.Struct({
50 threadId: ThreadId,
51 source: AgentSessionImportSource,
52});
53export type RecordImportedTranscriptInput = typeof RecordImportedTranscriptInput.Type;
54
55export interface ProviderSessionRuntimeUpsertOptions {
56 readonly onConflict?: "update" | "ignore";
57}
58
59export class ProviderSessionRuntimeRepository extends Context.Service<
60 ProviderSessionRuntimeRepository,
61 {
62 readonly upsert: (
63 runtime: ProviderSessionRuntime,
64 options?: ProviderSessionRuntimeUpsertOptions,
65 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
66
67 readonly recordImportedTranscript: (
68 input: RecordImportedTranscriptInput,
69 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
70
71 readonly getByThreadId: (
72 input: GetProviderSessionRuntimeInput,
73 ) => Effect.Effect<
74 Option.Option<ProviderSessionRuntime>,
75 ProviderSessionRuntimeRepositoryError
76 >;
77
78 readonly list: (options?: {
79 readonly excludeStopped?: boolean;
80 }) => Effect.Effect<
81 ReadonlyArray<ProviderSessionRuntime>,
82 ProviderSessionRuntimeRepositoryError
83 >;
84
85 readonly deleteByThreadId: (
86 input: DeleteProviderSessionRuntimeInput,
87 ) => Effect.Effect<void, ProviderSessionRuntimeRepositoryError>;
88 }
89>()("t3/persistence/ProviderSessionRuntime/ProviderSessionRuntimeRepository") {}
90
91const ProviderSessionRuntimeDbRowSchema = ProviderSessionRuntime.mapFields(
92 Struct.assign({
93 resumeCursor: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)),
94 runtimePayload: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)),
95 }),
96);
97
98const ProviderSessionRuntimeRawDbRowSchema = Schema.Struct({
99 threadId: Schema.String,
100 providerName: Schema.Unknown,
101 providerInstanceId: Schema.Unknown,
102 adapterKey: Schema.Unknown,
103 runtimeMode: Schema.Unknown,
104 status: Schema.Unknown,
105 lastSeenAt: Schema.Unknown,
106 resumeCursor: Schema.Unknown,
107 runtimePayload: Schema.Unknown,
108});
109
110const decodeRuntimeRow = Schema.decodeUnknownEffect(ProviderSessionRuntimeDbRowSchema);
111
112const GetRuntimeRequestSchema = Schema.Struct({
113 threadId: ThreadId,
114});
115
116const DeleteRuntimeRequestSchema = GetRuntimeRequestSchema;
117
118const RecordImportedTranscriptRequestSchema = RecordImportedTranscriptInput.mapFields(
119 Struct.assign({ source: Schema.fromJsonString(AgentSessionImportSource) }),
120);
121
122function toPersistenceSqlOrDecodeError(
123 sqlOperation: string,
124 decodeOperation: string,
125 correlation?: PersistenceErrorCorrelation,
126) {
127 return (cause: unknown): ProviderSessionRuntimeRepositoryError =>
128 Schema.isSchemaError(cause)
129 ? PersistenceDecodeError.fromSchemaError(decodeOperation, cause, correlation)
130 : new PersistenceSqlError({
131 operation: sqlOperation,
132 ...(correlation === undefined ? {} : { correlation }),
133 cause,
134 });
135}
136
137/** @public */
138export const make = Effect.gen(function* () {
139 const sql = yield* SqlClient.SqlClient;
140
141 // Runtime payloads may be stale; only recordImportedTranscript can change source records.
142 const upsertRuntimeRow = SqlSchema.void({
143 Request: ProviderSessionRuntimeDbRowSchema,
144 execute: (runtime) =>
145 sql`
146 INSERT INTO provider_session_runtime (
147 thread_id,
148 provider_name,
149 provider_instance_id,
150 adapter_key,
151 runtime_mode,
152 status,
153 last_seen_at,
154 resume_cursor_json,
155 runtime_payload_json
156 )
157 VALUES (
158 ${runtime.threadId},
159 ${runtime.providerName},
160 ${runtime.providerInstanceId},
161 ${runtime.adapterKey},
162 ${runtime.runtimeMode},
163 ${runtime.status},
164 ${runtime.lastSeenAt},
165 ${runtime.resumeCursor},
166 CASE
167 WHEN json_type(${runtime.runtimePayload}) = 'object'
168 THEN json_remove(${runtime.runtimePayload}, '$.importedTranscripts')
169 ELSE ${runtime.runtimePayload}
170 END
171 )
172 ON CONFLICT (thread_id)
173 DO UPDATE SET
174 provider_name = excluded.provider_name,
175 provider_instance_id = excluded.provider_instance_id,
176 adapter_key = excluded.adapter_key,
177 runtime_mode = excluded.runtime_mode,
178 status = excluded.status,
179 last_seen_at = excluded.last_seen_at,
180 resume_cursor_json = excluded.resume_cursor_json,
181 runtime_payload_json = CASE
182 WHEN json_type(
183 CASE
184 WHEN json_valid(provider_session_runtime.runtime_payload_json)
185 THEN provider_session_runtime.runtime_payload_json
186 ELSE '{}'
187 END,
188 '$.importedTranscripts'
189 ) IS NOT NULL
190 THEN json_set(
191 CASE
192 WHEN json_type(excluded.runtime_payload_json) = 'object'
193 THEN excluded.runtime_payload_json
194 ELSE '{}'
195 END,
196 '$.importedTranscripts',
197 json_extract(provider_session_runtime.runtime_payload_json, '$.importedTranscripts')
198 )
199 ELSE excluded.runtime_payload_json
200 END
201 `,
202 });
203
204 const insertRuntimeRow = SqlSchema.void({
205 Request: ProviderSessionRuntimeDbRowSchema,
206 execute: (runtime) =>
207 sql`
208 INSERT INTO provider_session_runtime (
209 thread_id,
210 provider_name,
211 provider_instance_id,
212 adapter_key,
213 runtime_mode,
214 status,
215 last_seen_at,
216 resume_cursor_json,
217 runtime_payload_json
218 )
219 VALUES (
220 ${runtime.threadId},
221 ${runtime.providerName},
222 ${runtime.providerInstanceId},
223 ${runtime.adapterKey},
224 ${runtime.runtimeMode},
225 ${runtime.status},
226 ${runtime.lastSeenAt},
227 ${runtime.resumeCursor},
228 CASE
229 WHEN json_type(${runtime.runtimePayload}) = 'object'
230 THEN json_remove(${runtime.runtimePayload}, '$.importedTranscripts')
231 ELSE ${runtime.runtimePayload}
232 END
233 )
234 ON CONFLICT (thread_id) DO NOTHING
235 `,
236 });
237
238 const recordImportedTranscriptRow = SqlSchema.void({
239 Request: RecordImportedTranscriptRequestSchema,
240 execute: ({ threadId, source }) =>
241 sql`
242 WITH current_runtime AS (
243 SELECT CASE
244 WHEN json_valid(runtime_payload_json) THEN CASE
245 WHEN json_type(runtime_payload_json) = 'object' THEN runtime_payload_json
246 ELSE '{}'
247 END
248 ELSE '{}'
249 END AS payload
250 FROM provider_session_runtime
251 WHERE thread_id = ${threadId}
252 )
253 UPDATE provider_session_runtime
254 SET runtime_payload_json = (
255 SELECT json_set(
256 payload,
257 '$.importedTranscripts',
258 json((
259 SELECT json_group_array(json(value))
260 FROM (
261 SELECT value
262 FROM json_each(CASE
263 WHEN json_type(payload, '$.importedTranscripts') = 'array'
264 THEN json_extract(payload, '$.importedTranscripts')
265 ELSE '[]'
266 END)
267 WHERE CASE
268 WHEN type = 'object' THEN
269 json_extract(value, '$.providerInstanceId')
270 IS NOT json_extract(${source}, '$.providerInstanceId')
271 OR json_extract(value, '$.filePath') IS NOT json_extract(${source}, '$.filePath')
272 ELSE 0
273 END
274 UNION ALL
275 SELECT ${source} AS value
276 )
277 ))
278 )
279 FROM current_runtime
280 )
281 WHERE thread_id = ${threadId}
282 `,
283 });
284
285 const getRuntimeRowByThreadId = SqlSchema.findOneOption({
286 Request: GetRuntimeRequestSchema,
287 Result: ProviderSessionRuntimeRawDbRowSchema,
288 execute: ({ threadId }) =>
289 sql`
290 SELECT
291 thread_id AS "threadId",
292 provider_name AS "providerName",
293 provider_instance_id AS "providerInstanceId",
294 adapter_key AS "adapterKey",
295 runtime_mode AS "runtimeMode",
296 status,
297 last_seen_at AS "lastSeenAt",
298 resume_cursor_json AS "resumeCursor",
299 runtime_payload_json AS "runtimePayload"
300 FROM provider_session_runtime
301 WHERE thread_id = ${threadId}
302 `,
303 });
304
305 const listRuntimeRows = SqlSchema.findAll({
306 Request: Schema.Struct({ excludeStopped: Schema.Boolean }),
307 Result: ProviderSessionRuntimeRawDbRowSchema,
308 execute: ({ excludeStopped }) =>
309 sql`
310 SELECT
311 thread_id AS "threadId",
312 provider_name AS "providerName",
313 provider_instance_id AS "providerInstanceId",
314 adapter_key AS "adapterKey",
315 runtime_mode AS "runtimeMode",
316 status,
317 last_seen_at AS "lastSeenAt",
318 resume_cursor_json AS "resumeCursor",
319 runtime_payload_json AS "runtimePayload"
320 FROM provider_session_runtime
321 ${excludeStopped ? sql`WHERE status != 'stopped'` : sql``}
322 ORDER BY last_seen_at ASC, thread_id ASC
323 `,
324 });
325
326 const deleteRuntimeByThreadId = SqlSchema.void({
327 Request: DeleteRuntimeRequestSchema,
328 execute: ({ threadId }) =>
329 sql`
330 DELETE FROM provider_session_runtime
331 WHERE thread_id = ${threadId}
332 `,
333 });
334
335 const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) =>
336 (options?.onConflict === "ignore" ? insertRuntimeRow(runtime) : upsertRuntimeRow(runtime)).pipe(
337 Effect.mapError(
338 toPersistenceSqlOrDecodeError(
339 "ProviderSessionRuntimeRepository.upsert:query",
340 "ProviderSessionRuntimeRepository.upsert:encodeRequest",
341 { threadId: runtime.threadId },
342 ),
343 ),
344 );
345
346 const recordImportedTranscript: ProviderSessionRuntimeRepository["Service"]["recordImportedTranscript"] =
347 (input) =>
348 recordImportedTranscriptRow(input).pipe(
349 Effect.mapError(
350 toPersistenceSqlOrDecodeError(
351 "ProviderSessionRuntimeRepository.recordImportedTranscript:query",
352 "ProviderSessionRuntimeRepository.recordImportedTranscript:encodeRequest",
353 { threadId: input.threadId },
354 ),
355 ),
356 );
357
358 const getByThreadId: ProviderSessionRuntimeRepository["Service"]["getByThreadId"] = (input) =>
359 getRuntimeRowByThreadId(input).pipe(
360 NodeSqliteClient.readOnly,
361 Effect.mapError(
362 toPersistenceSqlOrDecodeError(
363 "ProviderSessionRuntimeRepository.getByThreadId:query",
364 "ProviderSessionRuntimeRepository.getByThreadId:decodeRow",
365 { threadId: input.threadId },
366 ),
367 ),
368 Effect.flatMap((runtimeRowOption) =>
369 Option.match(runtimeRowOption, {
370 onNone: () => Effect.succeedNone,
371 onSome: (row) =>
372 decodeRuntimeRow(row).pipe(
373 Effect.mapError((cause) =>
374 PersistenceDecodeError.fromSchemaError(
375 "ProviderSessionRuntimeRepository.getByThreadId:decodeRow",
376 cause,
377 { threadId: input.threadId },
378 ),
379 ),
380 Effect.asSome,
381 ),
382 }),
383 ),
384 );
385
386 const list: ProviderSessionRuntimeRepository["Service"]["list"] = (options) =>
387 listRuntimeRows({ excludeStopped: options?.excludeStopped === true }).pipe(
388 NodeSqliteClient.readOnly,
389 Effect.mapError(
390 toPersistenceSqlOrDecodeError(
391 "ProviderSessionRuntimeRepository.list:query",
392 "ProviderSessionRuntimeRepository.list:decodeRows",
393 ),
394 ),
395 Effect.flatMap((rows) =>
396 Effect.forEach(rows, (row) =>
397 decodeRuntimeRow(row).pipe(
398 Effect.asSome,
399 Effect.catch((cause) =>
400 Effect.logWarning("provider.session.runtime.row-skipped", {
401 threadId: row.threadId,
402 error: PersistenceDecodeError.fromSchemaError(
403 "ProviderSessionRuntimeRepository.list:decodeRows",
404 cause,
405 { threadId: row.threadId },
406 ).message,
407 }).pipe(Effect.as(Option.none<ProviderSessionRuntime>())),
408 ),
409 ),
410 ),
411 ),
412 Effect.map((decoded) =>
413 Arr.filterMap(decoded, (row) =>
414 Option.isSome(row) ? Result.succeed(row.value) : Result.failVoid,
415 ),
416 ),
417 );
418
419 const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = (
420 input,
421 ) =>
422 deleteRuntimeByThreadId(input).pipe(
423 Effect.mapError(
424 (cause) =>
425 new PersistenceSqlError({
426 operation: "ProviderSessionRuntimeRepository.deleteByThreadId:query",
427 correlation: { threadId: input.threadId },
428 cause,
429 }),
430 ),
431 );
432
433 return {
434 upsert,
435 recordImportedTranscript,
436 getByThreadId,
437 list,
438 deleteByThreadId,
439 } satisfies ProviderSessionRuntimeRepository["Service"];
440});
441
442export const layer = Layer.effect(ProviderSessionRuntimeRepository, make);