import { OrchestrationCheckpointFile, OrchestrationReadModel, ProjectScript, type OrchestrationCheckpointSummary, type OrchestrationMessage, type OrchestrationProject, type OrchestrationSession, type OrchestrationThread, type OrchestrationThreadActivity,} from "@t3tools/contracts";import { Effect, Layer, Schema, Struct } from "effect";import * as SqlClient from "effect/unstable/sql/SqlClient";import * as SqlSchema from "effect/unstable/sql/SqlSchema"; import { isPersistenceError, toPersistenceDecodeError, toPersistenceSqlError, type ProjectionRepositoryError,} from "../../persistence/Errors.ts";import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts";import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts";import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts";import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";import { ProjectionSnapshotQuery, type ProjectionSnapshotQueryShape,} from "../Services/ProjectionSnapshotQuery.ts"; const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel);const ProjectionProjectDbRowSchema = ProjectionProject.mapFields( Struct.assign({ scripts: Schema.fromJsonString(Schema.Array(ProjectScript)), }),);const ProjectionThreadDbRowSchema = ProjectionThread;const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields( Struct.assign({ isStreaming: Schema.Number, }),);const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ payload: Schema.fromJsonString(Schema.Unknown), }),);const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields( Struct.assign({ files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)), }),);const ProjectionStateDbRowSchema = ProjectionState; const REQUIRED_SNAPSHOT_PROJECTORS = [ ORCHESTRATION_PROJECTOR_NAMES.projects, ORCHESTRATION_PROJECTOR_NAMES.threads, ORCHESTRATION_PROJECTOR_NAMES.threadMessages, ORCHESTRATION_PROJECTOR_NAMES.threadActivities, ORCHESTRATION_PROJECTOR_NAMES.threadSessions, ORCHESTRATION_PROJECTOR_NAMES.checkpoints,] as const; function maxIso(left: string | null, right: string): string { if (left === null) { return right; } return left > right ? left : right;} function computeSnapshotSequence( stateRows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionStateDbRowSchema>>,): number { if (stateRows.length === 0) { return 0; } const sequenceByProjector = new Map( stateRows.map((row) => [row.projector, row.lastAppliedSequence] as const), ); let minSequence = Number.POSITIVE_INFINITY; for (const projector of REQUIRED_SNAPSHOT_PROJECTORS) { const sequence = sequenceByProjector.get(projector); if (sequence === undefined) { return 0; } if (sequence < minSequence) { minSequence = sequence; } } return Number.isFinite(minSequence) ? minSequence : 0;} function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): ProjectionRepositoryError => Schema.isSchemaError(cause) ? toPersistenceDecodeError(decodeOperation)(cause) : toPersistenceSqlError(sqlOperation)(cause);} const makeProjectionSnapshotQuery = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const listProjectRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionProjectDbRowSchema, execute: () => sql` SELECT project_id AS "projectId", title, workspace_root AS "workspaceRoot", default_model AS "defaultModel", scripts_json AS "scripts", created_at AS "createdAt", updated_at AS "updatedAt", deleted_at AS "deletedAt" FROM projection_projects ORDER BY created_at ASC, project_id ASC `, }); const listThreadRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadDbRowSchema, execute: () => sql` SELECT thread_id AS "threadId", project_id AS "projectId", title, model, branch, worktree_path AS "worktreePath", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", deleted_at AS "deletedAt" FROM projection_threads ORDER BY created_at ASC, thread_id ASC `, }); const listThreadMessageRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadMessageDbRowSchema, execute: () => sql` SELECT message_id AS "messageId", thread_id AS "threadId", turn_id AS "turnId", role, text, is_streaming AS "isStreaming", created_at AS "createdAt", updated_at AS "updatedAt" FROM projection_thread_messages ORDER BY thread_id ASC, created_at ASC, message_id ASC `, }); const listThreadActivityRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadActivityDbRowSchema, execute: () => sql` SELECT activity_id AS "activityId", thread_id AS "threadId", turn_id AS "turnId", tone, kind, summary, payload_json AS "payload", created_at AS "createdAt" FROM projection_thread_activities ORDER BY thread_id ASC, created_at ASC, activity_id ASC `, }); const listThreadSessionRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadSessionDbRowSchema, execute: () => sql` SELECT thread_id AS "threadId", status, provider_name AS "providerName", provider_session_id AS "providerSessionId", provider_thread_id AS "providerThreadId", active_turn_id AS "activeTurnId", last_error AS "lastError", updated_at AS "updatedAt" FROM projection_thread_sessions ORDER BY thread_id ASC `, }); const listCheckpointRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionCheckpointDbRowSchema, execute: () => sql` SELECT thread_id AS "threadId", turn_id AS "turnId", checkpoint_turn_count AS "checkpointTurnCount", checkpoint_ref AS "checkpointRef", checkpoint_status AS "status", checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", completed_at AS "completedAt" FROM projection_turns WHERE checkpoint_turn_count IS NOT NULL ORDER BY thread_id ASC, checkpoint_turn_count ASC `, }); const listProjectionStateRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionStateDbRowSchema, execute: () => sql` SELECT projector, last_applied_sequence AS "lastAppliedSequence", updated_at AS "updatedAt" FROM projection_state `, }); const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () => sql .withTransaction( Effect.gen(function* () { const [ projectRows, threadRows, messageRows, activityRows, sessionRows, checkpointRows, stateRows, ] = yield* Effect.all([ listProjectRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listProjects:query", "ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows", ), ), ), listThreadRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listThreads:query", "ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows", ), ), ), listThreadMessageRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query", "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows", ), ), ), listThreadActivityRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query", "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows", ), ), ), listThreadSessionRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query", "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows", ), ), ), listCheckpointRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query", "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows", ), ), ), listProjectionStateRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getSnapshot:listProjectionState:query", "ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows", ), ), ), ]); const messagesByThread = new Map<string, Array<OrchestrationMessage>>(); const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>(); const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>(); const sessionsByThread = new Map<string, OrchestrationSession>(); let updatedAt: string | null = null; for (const row of projectRows) { updatedAt = maxIso(updatedAt, row.updatedAt); } for (const row of threadRows) { updatedAt = maxIso(updatedAt, row.updatedAt); } for (const row of stateRows) { updatedAt = maxIso(updatedAt, row.updatedAt); } for (const row of messageRows) { updatedAt = maxIso(updatedAt, row.updatedAt); const threadMessages = messagesByThread.get(row.threadId) ?? []; threadMessages.push({ id: row.messageId, role: row.role, text: row.text, turnId: row.turnId, streaming: row.isStreaming === 1, createdAt: row.createdAt, updatedAt: row.updatedAt, }); messagesByThread.set(row.threadId, threadMessages); } for (const row of activityRows) { updatedAt = maxIso(updatedAt, row.createdAt); const threadActivities = activitiesByThread.get(row.threadId) ?? []; threadActivities.push({ id: row.activityId, tone: row.tone, kind: row.kind, summary: row.summary, payload: row.payload, turnId: row.turnId, createdAt: row.createdAt, }); activitiesByThread.set(row.threadId, threadActivities); } for (const row of checkpointRows) { updatedAt = maxIso(updatedAt, row.completedAt); const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? []; threadCheckpoints.push({ turnId: row.turnId, checkpointTurnCount: row.checkpointTurnCount, checkpointRef: row.checkpointRef, status: row.status, files: row.files, assistantMessageId: row.assistantMessageId, completedAt: row.completedAt, }); checkpointsByThread.set(row.threadId, threadCheckpoints); } for (const row of sessionRows) { updatedAt = maxIso(updatedAt, row.updatedAt); sessionsByThread.set(row.threadId, { threadId: row.threadId, status: row.status, providerName: row.providerName, providerSessionId: row.providerSessionId, providerThreadId: row.providerThreadId, activeTurnId: row.activeTurnId, lastError: row.lastError, updatedAt: row.updatedAt, }); } const projects: Array<OrchestrationProject> = projectRows.map((row) => ({ id: row.projectId, title: row.title, workspaceRoot: row.workspaceRoot, defaultModel: row.defaultModel, scripts: row.scripts, createdAt: row.createdAt, updatedAt: row.updatedAt, deletedAt: row.deletedAt, })); const threads: Array<OrchestrationThread> = threadRows.map((row) => ({ id: row.threadId, projectId: row.projectId, title: row.title, model: row.model, branch: row.branch, worktreePath: row.worktreePath, latestTurnId: row.latestTurnId, createdAt: row.createdAt, updatedAt: row.updatedAt, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], activities: activitiesByThread.get(row.threadId) ?? [], checkpoints: checkpointsByThread.get(row.threadId) ?? [], session: sessionsByThread.get(row.threadId) ?? null, })); const snapshot = { snapshotSequence: computeSnapshotSequence(stateRows), projects, threads, updatedAt: updatedAt ?? new Date(0).toISOString(), }; return yield* decodeReadModel(snapshot).pipe( Effect.mapError( toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"), ), ); }), ) .pipe( Effect.mapError((error) => { if (isPersistenceError(error)) { return error; } return toPersistenceSqlError("ProjectionSnapshotQuery.getSnapshot:query")(error); }), ); return { getSnapshot, } satisfies ProjectionSnapshotQueryShape;}); export const OrchestrationProjectionSnapshotQueryLive = Layer.effect( ProjectionSnapshotQuery, makeProjectionSnapshotQuery,);
import {
OrchestrationCheckpointFile,
OrchestrationReadModel,
ProjectScript,
type OrchestrationCheckpointSummary,
type OrchestrationMessage,
type OrchestrationProject,
type OrchestrationSession,
type OrchestrationThread,
type OrchestrationThreadActivity,
} from "@t3tools/contracts";
import { Effect, Layer, Schema, Struct } from "effect";
import * as SqlClient from "effect/unstable/sql/SqlClient";
import * as SqlSchema from "effect/unstable/sql/SqlSchema";
import {
isPersistenceError,
toPersistenceDecodeError,
toPersistenceSqlError,
type ProjectionRepositoryError,
} from "../../persistence/Errors.ts";
import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts";
import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts";
import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";
import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";
import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts";
import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
import {
ProjectionSnapshotQuery,
type ProjectionSnapshotQueryShape,
} from "../Services/ProjectionSnapshotQuery.ts";
const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel);
const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(
Struct.assign({
scripts: Schema.fromJsonString(Schema.Array(ProjectScript)),
}),
);
const ProjectionThreadDbRowSchema = ProjectionThread;
const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(
Struct.assign({
isStreaming: Schema.Number,
}),
);
const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
Struct.assign({
payload: Schema.fromJsonString(Schema.Unknown),
}),
);
const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(
Struct.assign({
files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
}),
);
const ProjectionStateDbRowSchema = ProjectionState;
const REQUIRED_SNAPSHOT_PROJECTORS = [
ORCHESTRATION_PROJECTOR_NAMES.projects,
ORCHESTRATION_PROJECTOR_NAMES.threads,
ORCHESTRATION_PROJECTOR_NAMES.threadMessages,
ORCHESTRATION_PROJECTOR_NAMES.threadActivities,
ORCHESTRATION_PROJECTOR_NAMES.threadSessions,
ORCHESTRATION_PROJECTOR_NAMES.checkpoints,
] as const;
function maxIso(left: string | null, right: string): string {
if (left === null) {
return right;
}
return left > right ? left : right;
}
function computeSnapshotSequence(
stateRows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionStateDbRowSchema>>,
): number {
if (stateRows.length === 0) {
return 0;
}
const sequenceByProjector = new Map(
stateRows.map((row) => [row.projector, row.lastAppliedSequence] as const),
);
let minSequence = Number.POSITIVE_INFINITY;
for (const projector of REQUIRED_SNAPSHOT_PROJECTORS) {
const sequence = sequenceByProjector.get(projector);
if (sequence === undefined) {
return 0;
}
if (sequence < minSequence) {
minSequence = sequence;
}
}
return Number.isFinite(minSequence) ? minSequence : 0;
}
function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
return (cause: unknown): ProjectionRepositoryError =>
Schema.isSchemaError(cause)
? toPersistenceDecodeError(decodeOperation)(cause)
: toPersistenceSqlError(sqlOperation)(cause);
}
const makeProjectionSnapshotQuery = Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
const listProjectRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionProjectDbRowSchema,
execute: () =>
sql`
SELECT
project_id AS "projectId",
title,
workspace_root AS "workspaceRoot",
default_model AS "defaultModel",
scripts_json AS "scripts",
created_at AS "createdAt",
updated_at AS "updatedAt",
deleted_at AS "deletedAt"
FROM projection_projects
ORDER BY created_at ASC, project_id ASC
`,
});
const listThreadRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
project_id AS "projectId",
title,
model,
branch,
worktree_path AS "worktreePath",
latest_turn_id AS "latestTurnId",
created_at AS "createdAt",
updated_at AS "updatedAt",
deleted_at AS "deletedAt"
FROM projection_threads
ORDER BY created_at ASC, thread_id ASC
`,
});
const listThreadMessageRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadMessageDbRowSchema,
execute: () =>
sql`
SELECT
message_id AS "messageId",
thread_id AS "threadId",
turn_id AS "turnId",
role,
text,
is_streaming AS "isStreaming",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_messages
ORDER BY thread_id ASC, created_at ASC, message_id ASC
`,
});
const listThreadActivityRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadActivityDbRowSchema,
execute: () =>
sql`
SELECT
activity_id AS "activityId",
thread_id AS "threadId",
turn_id AS "turnId",
tone,
kind,
summary,
payload_json AS "payload",
created_at AS "createdAt"
FROM projection_thread_activities
ORDER BY thread_id ASC, created_at ASC, activity_id ASC
`,
});
const listThreadSessionRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadSessionDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
status,
provider_name AS "providerName",
provider_session_id AS "providerSessionId",
provider_thread_id AS "providerThreadId",
active_turn_id AS "activeTurnId",
last_error AS "lastError",
updated_at AS "updatedAt"
FROM projection_thread_sessions
ORDER BY thread_id ASC
`,
});
const listCheckpointRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionCheckpointDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
turn_id AS "turnId",
checkpoint_turn_count AS "checkpointTurnCount",
checkpoint_ref AS "checkpointRef",
checkpoint_status AS "status",
checkpoint_files_json AS "files",
assistant_message_id AS "assistantMessageId",
completed_at AS "completedAt"
FROM projection_turns
WHERE checkpoint_turn_count IS NOT NULL
ORDER BY thread_id ASC, checkpoint_turn_count ASC
`,
});
const listProjectionStateRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionStateDbRowSchema,
execute: () =>
sql`
SELECT
projector,
last_applied_sequence AS "lastAppliedSequence",
updated_at AS "updatedAt"
FROM projection_state
`,
});
const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>
sql
.withTransaction(
Effect.gen(function* () {
const [
projectRows,
threadRows,
messageRows,
activityRows,
sessionRows,
checkpointRows,
stateRows,
] = yield* Effect.all([
listProjectRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listProjects:query",
"ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",
),
),
),
listThreadRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreads:query",
"ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
),
),
),
listThreadMessageRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",
),
),
),
listThreadActivityRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",
),
),
),
listThreadSessionRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",
),
),
),
listCheckpointRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",
"ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",
),
),
),
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listProjectionState:query",
"ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows",
),
),
),
]);
const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
const sessionsByThread = new Map<string, OrchestrationSession>();
let updatedAt: string | null = null;
for (const row of projectRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of threadRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of stateRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of messageRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
const threadMessages = messagesByThread.get(row.threadId) ?? [];
threadMessages.push({
id: row.messageId,
role: row.role,
text: row.text,
turnId: row.turnId,
streaming: row.isStreaming === 1,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
});
messagesByThread.set(row.threadId, threadMessages);
}
for (const row of activityRows) {
updatedAt = maxIso(updatedAt, row.createdAt);
const threadActivities = activitiesByThread.get(row.threadId) ?? [];
threadActivities.push({
id: row.activityId,
tone: row.tone,
kind: row.kind,
summary: row.summary,
payload: row.payload,
turnId: row.turnId,
createdAt: row.createdAt,
});
activitiesByThread.set(row.threadId, threadActivities);
}
for (const row of checkpointRows) {
updatedAt = maxIso(updatedAt, row.completedAt);
const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
threadCheckpoints.push({
turnId: row.turnId,
checkpointTurnCount: row.checkpointTurnCount,
checkpointRef: row.checkpointRef,
status: row.status,
files: row.files,
assistantMessageId: row.assistantMessageId,
completedAt: row.completedAt,
});
checkpointsByThread.set(row.threadId, threadCheckpoints);
}
for (const row of sessionRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
sessionsByThread.set(row.threadId, {
threadId: row.threadId,
status: row.status,
providerName: row.providerName,
providerSessionId: row.providerSessionId,
providerThreadId: row.providerThreadId,
activeTurnId: row.activeTurnId,
lastError: row.lastError,
updatedAt: row.updatedAt,
});
}
const projects: Array<OrchestrationProject> = projectRows.map((row) => ({
id: row.projectId,
title: row.title,
workspaceRoot: row.workspaceRoot,
defaultModel: row.defaultModel,
scripts: row.scripts,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
deletedAt: row.deletedAt,
}));
const threads: Array<OrchestrationThread> = threadRows.map((row) => ({
id: row.threadId,
projectId: row.projectId,
title: row.title,
model: row.model,
branch: row.branch,
worktreePath: row.worktreePath,
latestTurnId: row.latestTurnId,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
deletedAt: row.deletedAt,
messages: messagesByThread.get(row.threadId) ?? [],
activities: activitiesByThread.get(row.threadId) ?? [],
checkpoints: checkpointsByThread.get(row.threadId) ?? [],
session: sessionsByThread.get(row.threadId) ?? null,
}));
const snapshot = {
snapshotSequence: computeSnapshotSequence(stateRows),
projects,
threads,
updatedAt: updatedAt ?? new Date(0).toISOString(),
};
return yield* decodeReadModel(snapshot).pipe(
Effect.mapError(
toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"),
),
);
}),
)
.pipe(
Effect.mapError((error) => {
if (isPersistenceError(error)) {
return error;
}
return toPersistenceSqlError("ProjectionSnapshotQuery.getSnapshot:query")(error);
}),
);
return {
getSnapshot,
} satisfies ProjectionSnapshotQueryShape;
});
export const OrchestrationProjectionSnapshotQueryLive = Layer.effect(
ProjectionSnapshotQuery,
makeProjectionSnapshotQuery,
);