diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 51c3861a9fca0..129ce850f08f8 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -19,6 +19,7 @@ import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlSchema from "effect/sql/SqlSchema"; import * as Effect from "effect/Effect"; +import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Option from "effect/Option"; @@ -174,8 +175,21 @@ interface ApplicationEventRow { const decodeV2EventJson = Schema.decodeUnknownEffect(OrchestrationV2DomainEventJson); const encodeV2EventJson = Schema.encodeEffect(OrchestrationV2DomainEventJson); const decodeV2StoredEvent = Schema.decodeUnknownEffect(OrchestrationV2StoredEvent); +// Rows carry an event decodeV2EventJson already validated; revalidating it nearly doubled replay cost. +const decodeV2StoredEventHeader = Schema.decodeUnknownEffect( + OrchestrationV2StoredEvent.mapFields(({ event: _event, ...fields }) => fields), +); const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); +// Same result as decodeJson; only malformed input pays for the schema's error. +const parseJson = (json: string) => { + if (typeof json !== "string") return decodeJson(json); + try { + return Effect.succeed(JSON.parse(json)); + } catch { + return decodeJson(json); + } +}; function metadataForV2Event(event: OrchestrationV2DomainEvent): Record { return { @@ -189,11 +203,10 @@ function metadataForV2Event(event: OrchestrationV2DomainEvent): Record) : {}; const event = yield* decodeV2EventJson({ @@ -210,16 +223,14 @@ const rowToV2StoredEvent = Effect.fn("OrchestrationEventStore.rowToV2StoredEvent : { providerInstanceId: values.providerInstanceId }), ...(values.rawEventId === undefined ? {} : { rawEventId: values.rawEventId }), }); - return yield* decodeV2StoredEvent({ + const { sequence, commandId } = yield* decodeV2StoredEventHeader({ sequence: row.sequence, commandId: row.command_id, - event, }); + return { sequence, commandId, event }; }); -const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent")(function* ( - row: ApplicationEventRow, -) { +const rowToProjectEvent = Effect.fnUntraced(function* (row: ApplicationEventRow) { return yield* decodeProjectEvent({ sequence: row.sequence, eventId: row.event_id, @@ -230,8 +241,8 @@ const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent") commandId: row.command_id, causationEventId: row.causation_event_id, correlationId: row.correlation_id, - payload: yield* decodeJson(row.payload_json), - metadata: yield* decodeJson(row.metadata_json), + payload: yield* parseJson(row.payload_json), + metadata: yield* parseJson(row.metadata_json), }); }); @@ -243,11 +254,12 @@ function rowToApplicationStoredEvent( const isKnownProjectEventType = Schema.is(ProjectEventType); -const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) => - skipUnknownEventTypes !== true || - (row.aggregate_kind === "project" - ? isKnownProjectEventType(row.event_type) - : isKnownOrchestrationV2EventType(row.event_type)) +const keepKnownEventTypeRow = (row: ApplicationEventRow) => + ( + row.aggregate_kind === "project" + ? isKnownProjectEventType(row.event_type) + : isKnownOrchestrationV2EventType(row.event_type) + ) ? Effect.succeed(true) : Effect.logWarning("Skipping an application event row with an unknown event type", { aggregateKind: row.aggregate_kind, @@ -574,7 +586,9 @@ const makeEventStore = Effect.gen(function* () { ); }, ).pipe( - Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)), + // Query the next page on the reader while this one decodes. + Stream.bufferArray({ capacity: 1 }), + input?.skipUnknownEventTypes === true ? Stream.filterEffect(keepKnownEventTypeRow) : identity, Stream.mapEffect((row) => rowToV2StoredEvent(row).pipe( Effect.mapError( @@ -582,6 +596,7 @@ const makeEventStore = Effect.gen(function* () { ), ), ), + Stream.withSpan("OrchestrationEventStore.readAgentEvents"), ); }; @@ -699,7 +714,9 @@ const makeEventStore = Effect.gen(function* () { }), ), ).pipe( - Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)), + // Query the next page on the reader while this one decodes. + Stream.bufferArray({ capacity: 1 }), + input.skipUnknownEventTypes === true ? Stream.filterEffect(keepKnownEventTypeRow) : identity, Stream.mapEffect((row) => rowToApplicationStoredEvent(row).pipe( Effect.mapError( @@ -707,6 +724,7 @@ const makeEventStore = Effect.gen(function* () { ), ), ), + Stream.withSpan("OrchestrationEventStore.readApplicationEvents"), ); const streamProjectedApplicationEvents: OrchestrationEventStore["Service"]["streamProjectedApplicationEvents"] =