Move provider event SQL access into Effect repository

Cursor Agent committed 8 months ago

Commit bce0df01 parent d6e0cd82 files · +231 −62Browse files

Unified · Split

2 files

apps/server/src/persistence/providerEventsRepo.ts +124 −0
apps/server/src/persistenceService.ts +107 −62

apps/server/src/persistence/providerEventsRepo.ts+124−0

@@ -0,0 +1,124 @@
	1	import * as Effect from "effect/Effect";
	2	import * as SqlClient from "effect/unstable/sql/SqlClient";
	3
	4	function toSafeInteger(value: unknown, fallback = 0): number {
	5	if (typeof value === "number" && Number.isFinite(value)) {
	6	return Math.trunc(value);
	7	}
	8	if (typeof value === "bigint") {
	9	return Number(value);
	10	}
	11	return fallback;
	12	}
	13
	14	export interface CompletedProviderItemRow {
	15	item_id: string \| null;
	16	payload_json: string \| null;
	17	}
	18
	19	export const insertProviderEvent = (input: {
	20	id: string;
	21	sessionId: string;
	22	provider: string;
	23	kind: string;
	24	method: string;
	25	runtimeThreadId: string \| null;
	26	turnId: string \| null;
	27	itemId: string \| null;
	28	requestId: string \| null;
	29	requestKind: string \| null;
	30	textDelta: string \| null;
	31	message: string \| null;
	32	payloadJson: string \| null;
	33	createdAt: string;
	34	}): Effect.Effect<number, unknown, SqlClient.SqlClient> =>
	35	Effect.gen(function*() {
	36	const sql = yield* SqlClient.SqlClient;
	37	const rawResult = (yield* sql
	38	.unsafe<{ changes?: number \| bigint }>(
	39	`INSERT OR IGNORE INTO provider_events (
	40	id,
	41	session_id,
	42	provider,
	43	kind,
	44	method,
	45	thread_id,
	46	turn_id,
	47	item_id,
	48	request_id,
	49	request_kind,
	50	text_delta,
	51	message,
	52	payload_json,
	53	created_at
	54	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);`,
	55	[
	56	input.id,
	57	input.sessionId,
	58	input.provider,
	59	input.kind,
	60	input.method,
	61	input.runtimeThreadId,
	62	input.turnId,
	63	input.itemId,
	64	input.requestId,
	65	input.requestKind,
	66	input.textDelta,
	67	input.message,
	68	input.payloadJson,
	69	input.createdAt,
	70	],
	71	)
	72	.raw) as { changes?: number \| bigint };
	73	return toSafeInteger(rawResult.changes, 0);
	74	});
	75
	76	export const listCompletedItemEventsBySessionTurn = (input: {
	77	sessionId: string;
	78	turnId: string;
	79	}): Effect.Effect<CompletedProviderItemRow[], unknown, SqlClient.SqlClient> =>
	80	Effect.gen(function*() {
	81	const sql = yield* SqlClient.SqlClient;
	82	return (yield* sql
	83	.unsafe<CompletedProviderItemRow>(
	84	`SELECT item_id, payload_json
	85	FROM provider_events
	86	WHERE session_id = ? AND turn_id = ? AND method = 'item/completed'
	87	ORDER BY created_at DESC;`,
	88	[input.sessionId, input.turnId],
	89	)
	90	.unprepared) as CompletedProviderItemRow[];
	91	});
	92
	93	export const listCompletedItemEventsByThreadTurn = (input: {
	94	runtimeThreadId: string;
	95	turnId: string;
	96	}): Effect.Effect<CompletedProviderItemRow[], unknown, SqlClient.SqlClient> =>
	97	Effect.gen(function*() {
	98	const sql = yield* SqlClient.SqlClient;
	99	return (yield* sql
	100	.unsafe<CompletedProviderItemRow>(
	101	`SELECT item_id, payload_json
	102	FROM provider_events
	103	WHERE thread_id = ? AND turn_id = ? AND method = 'item/completed'
	104	ORDER BY created_at DESC;`,
	105	[input.runtimeThreadId, input.turnId],
	106	)
	107	.unprepared) as CompletedProviderItemRow[];
	108	});
	109
	110	export const listCompletedItemEventsByTurn = (
	111	turnId: string,
	112	): Effect.Effect<CompletedProviderItemRow[], unknown, SqlClient.SqlClient> =>
	113	Effect.gen(function*() {
	114	const sql = yield* SqlClient.SqlClient;
	115	return (yield* sql
	116	.unsafe<CompletedProviderItemRow>(
	117	`SELECT item_id, payload_json
	118	FROM provider_events
	119	WHERE turn_id = ? AND method = 'item/completed'
	120	ORDER BY created_at DESC;`,
	121	[turnId],
	122	)
	123	.unprepared) as CompletedProviderItemRow[];
	124	});

apps/server/src/persistenceService.ts+107−62

   upsertDocument as upsertDocumentEffect,
   type DocumentRow,
 } from "./persistence/documentsRepo";
+import {
+  insertProviderEvent as insertProviderEventEffect,
+  listCompletedItemEventsBySessionTurn as listCompletedItemEventsBySessionTurnEffect,
+  listCompletedItemEventsByThreadTurn as listCompletedItemEventsByThreadTurnEffect,
+  listCompletedItemEventsByTurn as listCompletedItemEventsByTurnEffect,
+  type CompletedProviderItemRow,
+} from "./persistence/providerEventsRepo";
 import {
   appendStateEvent as appendStateEventEffect,
   listStateEventsAfterSeq as listStateEventsAfterSeqEffect,
     sessionId?: string;
     runtimeThreadId?: string | null;
   }): string | undefined {
-    const queries: Array<{ sql: string; params: unknown[] }> = [];
+    const rowGroups: CompletedProviderItemRow[][] = [];
     if (input.sessionId) {
-      queries.push({
+      rowGroups.push(
-        sql: `SELECT item_id, payload_json
+        this.runWithEffectSql(
-              FROM provider_events
+          listCompletedItemEventsBySessionTurnEffect({
-              WHERE session_id = ? AND turn_id = ? AND method = 'item/completed'
+            sessionId: input.sessionId,
-              ORDER BY created_at DESC;`,
+            turnId: input.turnId,
-        params: [input.sessionId, input.turnId],
+          }),
-      });
+          () =>
+            this.db.prepare(
+              `SELECT item_id, payload_json
+               FROM provider_events
+               WHERE session_id = ? AND turn_id = ? AND method = 'item/completed'
+               ORDER BY created_at DESC;`,
+            ).all(input.sessionId, input.turnId) as CompletedProviderItemRow[],
+        ),
+      );
     }
     if (input.runtimeThreadId) {
-      queries.push({
+      rowGroups.push(
-        sql: `SELECT item_id, payload_json
+        this.runWithEffectSql(
-              FROM provider_events
+          listCompletedItemEventsByThreadTurnEffect({
-              WHERE thread_id = ? AND turn_id = ? AND method = 'item/completed'
+            runtimeThreadId: input.runtimeThreadId,
-              ORDER BY created_at DESC;`,
+            turnId: input.turnId,
-        params: [input.runtimeThreadId, input.turnId],
+          }),
-      });
+          () =>
+            this.db.prepare(
+              `SELECT item_id, payload_json
+               FROM provider_events
+               WHERE thread_id = ? AND turn_id = ? AND method = 'item/completed'
+               ORDER BY created_at DESC;`,
+            ).all(input.runtimeThreadId, input.turnId) as CompletedProviderItemRow[],
+        ),
+      );
     }
-    queries.push({
+    rowGroups.push(
-      sql: `SELECT item_id, payload_json
+      this.runWithEffectSql(
-            FROM provider_events
+        listCompletedItemEventsByTurnEffect(input.turnId),
-            WHERE turn_id = ? AND method = 'item/completed'
+        () =>
-            ORDER BY created_at DESC;`,
+          this.db.prepare(
-      params: [input.turnId],
+            `SELECT item_id, payload_json
-    });
+             FROM provider_events
+             WHERE turn_id = ? AND method = 'item/completed'
+             ORDER BY created_at DESC;`,
+          ).all(input.turnId) as CompletedProviderItemRow[],
+      ),
+    );
-    for (const query of queries) {
+    for (const rows of rowGroups) {
-      const rows = this.db.prepare(query.sql).all(...query.params) as Array<{
-        item_id: string | null;
-        payload_json: string | null;
-      }>;
       for (const row of rows) {
         const payload = row.payload_json ? this.tryParseJson(row.payload_json) : null;
         const item = asObject(asObject(payload)?.item);
   private insertProviderEvent(event: ProviderEvent): ProviderEventInsertResult {
     const runtimeThreadId = event.threadId ?? parseThreadIdFromEventPayload(event.payload);
     const payloadJson = event.payload === undefined ? null : JSON.stringify(event.payload);
-    const result = this.db
+    const changes = this.runWithEffectSql(
-      .prepare(
+      insertProviderEventEffect({
-        `INSERT OR IGNORE INTO provider_events (
+        id: event.id,
-          id,
+        sessionId: event.sessionId,
-          session_id,
+        provider: event.provider,
-          provider,
+        kind: event.kind,
-          kind,
+        method: event.method,
-          method,
+        runtimeThreadId: runtimeThreadId ?? null,
-          thread_id,
+        turnId: event.turnId ?? null,
-          turn_id,
+        itemId: event.itemId ?? null,
-          item_id,
+        requestId: event.requestId ?? null,
-          request_id,
+        requestKind: event.requestKind ?? null,
-          request_kind,
+        textDelta: event.textDelta ?? null,
-          text_delta,
+        message: event.message ?? null,
-          message,
-          payload_json,
-          created_at
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);`,
-      )
-      .run(
-        event.id,
-        event.sessionId,
-        event.provider,
-        event.kind,
-        event.method,
-        runtimeThreadId,
-        event.turnId ?? null,
-        event.itemId ?? null,
-        event.requestId ?? null,
-        event.requestKind ?? null,
-        event.textDelta ?? null,
-        event.message ?? null,
         payloadJson,
-        event.createdAt,
+        createdAt: event.createdAt,
-      ) as { changes?: number | bigint };
+      }),
+      () => {
+        const result = this.db
+          .prepare(
+            `INSERT OR IGNORE INTO provider_events (
+              id,
+              session_id,
+              provider,
+              kind,
+              method,
+              thread_id,
+              turn_id,
+              item_id,
+              request_id,
+              request_kind,
+              text_delta,
+              message,
+              payload_json,
+              created_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);`,
+          )
+          .run(
+            event.id,
+            event.sessionId,
+            event.provider,
+            event.kind,
+            event.method,
+            runtimeThreadId,
+            event.turnId ?? null,
+            event.itemId ?? null,
+            event.requestId ?? null,
+            event.requestKind ?? null,
+            event.textDelta ?? null,
+            event.message ?? null,
+            payloadJson,
+            event.createdAt,
+          ) as { changes?: number | bigint };
+        return toSafeInteger(result.changes, 0);
+      },
+    );
     return {
-      inserted: toSafeInteger(result.changes, 0) > 0,
+      inserted: changes > 0,
       runtimeThreadId: runtimeThreadId ?? null,
     };
   }

Move provider event SQL access into Effect repository

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code