import { ApprovalRequestId, type OrchestrationEvent } from "@t3tools/contracts";import { Effect, Layer, Option, Stream } from "effect";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";import { type ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";import {  type ProjectionThreadMessage,  ProjectionThreadMessageRepository,} from "../../persistence/Services/ProjectionThreadMessages.ts";import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";import { type ProjectionTurn, ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts";import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts";import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts";import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts";import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";import {  OrchestrationProjectionPipeline,  type OrchestrationProjectionPipelineShape,} from "../Services/ProjectionPipeline.ts"; export const ORCHESTRATION_PROJECTOR_NAMES = {  projects: "projection.projects",  threads: "projection.threads",  threadMessages: "projection.thread-messages",  threadActivities: "projection.thread-activities",  threadSessions: "projection.thread-sessions",  threadTurns: "projection.thread-turns",  checkpoints: "projection.checkpoints",  pendingApprovals: "projection.pending-approvals",} as const; type ProjectorName =  (typeof ORCHESTRATION_PROJECTOR_NAMES)[keyof typeof ORCHESTRATION_PROJECTOR_NAMES]; interface ProjectorDefinition {  readonly name: ProjectorName;  readonly apply: (event: OrchestrationEvent) => Effect.Effect<void, ProjectionRepositoryError>;} function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {  if (typeof payload !== "object" || payload === null) {    return null;  }  const requestId = (payload as Record<string, unknown>).requestId;  return typeof requestId === "string" ? ApprovalRequestId.makeUnsafe(requestId) : null;} function retainProjectionMessagesAfterRevert(  messages: ReadonlyArray<ProjectionThreadMessage>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadMessage> {  const retainedMessageIds = new Set<string>();  const retainedTurnIds = new Set<string>();  const keptTurns = turns.filter(    (turn) => turn.turnId !== null && turn.checkpointTurnCount !== null && turn.checkpointTurnCount <= turnCount,  );  for (const turn of keptTurns) {    if (turn.turnId !== null) {      retainedTurnIds.add(turn.turnId);    }    if (turn.pendingMessageId !== null) {      retainedMessageIds.add(turn.pendingMessageId);    }    if (turn.assistantMessageId !== null) {      retainedMessageIds.add(turn.assistantMessageId);    }  }   for (const message of messages) {    if (message.role === "system") {      retainedMessageIds.add(message.messageId);      continue;    }    if (message.turnId !== null && retainedTurnIds.has(message.turnId)) {      retainedMessageIds.add(message.messageId);    }  }   const retainedUserCount = messages.filter(    (message) => message.role === "user" && retainedMessageIds.has(message.messageId),  ).length;  const missingUserCount = Math.max(0, turnCount - retainedUserCount);  if (missingUserCount > 0) {    const fallbackUserMessages = messages      .filter(        (message) =>          message.role === "user" &&          !retainedMessageIds.has(message.messageId) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) ||          left.messageId.localeCompare(right.messageId),      )      .slice(0, missingUserCount);    for (const message of fallbackUserMessages) {      retainedMessageIds.add(message.messageId);    }  }   const retainedAssistantCount = messages.filter(    (message) => message.role === "assistant" && retainedMessageIds.has(message.messageId),  ).length;  const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);  if (missingAssistantCount > 0) {    const fallbackAssistantMessages = messages      .filter(        (message) =>          message.role === "assistant" &&          !retainedMessageIds.has(message.messageId) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) ||          left.messageId.localeCompare(right.messageId),      )      .slice(0, missingAssistantCount);    for (const message of fallbackAssistantMessages) {      retainedMessageIds.add(message.messageId);    }  }   return messages.filter((message) => retainedMessageIds.has(message.messageId));} function retainProjectionActivitiesAfterRevert(  activities: ReadonlyArray<ProjectionThreadActivity>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadActivity> {  const retainedTurnIds = new Set<string>(    turns      .filter(        (turn) =>          turn.turnId !== null &&          turn.checkpointTurnCount !== null &&          turn.checkpointTurnCount <= turnCount,      )      .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),  );  return activities.filter(    (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId),  );} const makeOrchestrationProjectionPipeline = Effect.gen(function* () {  const sql = yield* SqlClient.SqlClient;  const eventStore = yield* OrchestrationEventStore;  const projectionStateRepository = yield* ProjectionStateRepository;  const projectionProjectRepository = yield* ProjectionProjectRepository;  const projectionThreadRepository = yield* ProjectionThreadRepository;  const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;  const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;  const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;  const projectionTurnRepository = yield* ProjectionTurnRepository;  const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository;   const applyProjectsProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      switch (event.type) {        case "project.created":          yield* projectionProjectRepository.upsert({            projectId: event.payload.projectId,            title: event.payload.title,            workspaceRoot: event.payload.workspaceRoot,            defaultModel: event.payload.defaultModel,            scripts: event.payload.scripts,            createdAt: event.payload.createdAt,            updatedAt: event.payload.updatedAt,            deletedAt: null,          });          return;         case "project.meta-updated": {          const existingRow = yield* projectionProjectRepository.getById({            projectId: event.payload.projectId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionProjectRepository.upsert({            ...existingRow.value,            ...(event.payload.title !== undefined ? { title: event.payload.title } : {}),            ...(event.payload.workspaceRoot !== undefined              ? { workspaceRoot: event.payload.workspaceRoot }              : {}),            ...(event.payload.defaultModel !== undefined              ? { defaultModel: event.payload.defaultModel }              : {}),            ...(event.payload.scripts !== undefined ? { scripts: event.payload.scripts } : {}),            updatedAt: event.payload.updatedAt,          });          return;        }         case "project.deleted": {          const existingRow = yield* projectionProjectRepository.getById({            projectId: event.payload.projectId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionProjectRepository.upsert({            ...existingRow.value,            deletedAt: event.payload.deletedAt,            updatedAt: event.payload.deletedAt,          });          return;        }         default:          return;      }    });   const applyThreadsProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.created":          yield* projectionThreadRepository.upsert({            threadId: event.payload.threadId,            projectId: event.payload.projectId,            title: event.payload.title,            model: event.payload.model,            branch: event.payload.branch,            worktreePath: event.payload.worktreePath,            latestTurnId: null,            createdAt: event.payload.createdAt,            updatedAt: event.payload.updatedAt,            deletedAt: null,          });          return;         case "thread.meta-updated": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            ...(event.payload.title !== undefined ? { title: event.payload.title } : {}),            ...(event.payload.model !== undefined ? { model: event.payload.model } : {}),            ...(event.payload.branch !== undefined ? { branch: event.payload.branch } : {}),            ...(event.payload.worktreePath !== undefined              ? { worktreePath: event.payload.worktreePath }              : {}),            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.deleted": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            deletedAt: event.payload.deletedAt,            updatedAt: event.payload.deletedAt,          });          return;        }         case "thread.message-sent":        case "thread.activity-appended": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            updatedAt: event.occurredAt,          });          return;        }         case "thread.session-set": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            latestTurnId: event.payload.session.activeTurnId,            updatedAt: event.occurredAt,          });          return;        }         case "thread.turn-diff-completed": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            latestTurnId: event.payload.turnId,            updatedAt: event.occurredAt,          });          return;        }         case "thread.reverted": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            latestTurnId: null,            updatedAt: event.occurredAt,          });          return;        }         default:          return;      }    });   const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.message-sent": {          const existingRows = yield* projectionThreadMessageRepository.listByThreadId({            threadId: event.payload.threadId,          });          const existingMessage = existingRows.find(            (row) => row.messageId === event.payload.messageId,          );          const nextText =            existingMessage && event.payload.streaming              ? `${existingMessage.text}${event.payload.text}`              : existingMessage && event.payload.text.length === 0                ? existingMessage.text                : event.payload.text;          yield* projectionThreadMessageRepository.upsert({            messageId: event.payload.messageId,            threadId: event.payload.threadId,            turnId: event.payload.turnId,            role: event.payload.role,            text: nextText,            isStreaming: event.payload.streaming,            createdAt: existingMessage?.createdAt ?? event.payload.createdAt,            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.reverted": {          const existingRows = yield* projectionThreadMessageRepository.listByThreadId({            threadId: event.payload.threadId,          });          if (existingRows.length === 0) {            return;          }           const existingTurns = yield* projectionTurnRepository.listByThreadId({            threadId: event.payload.threadId,          });          const keptRows = retainProjectionMessagesAfterRevert(            existingRows,            existingTurns,            event.payload.turnCount,          );          if (keptRows.length === existingRows.length) {            return;          }           yield* projectionThreadMessageRepository.deleteByThreadId({            threadId: event.payload.threadId,          });          yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, {            concurrency: 1,          }).pipe(Effect.asVoid);          return;        }         default:          return;      }    });   const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.activity-appended":          yield* projectionThreadActivityRepository.upsert({            activityId: event.payload.activity.id,            threadId: event.payload.threadId,            turnId: event.payload.activity.turnId,            tone: event.payload.activity.tone,            kind: event.payload.activity.kind,            summary: event.payload.activity.summary,            payload: event.payload.activity.payload,            createdAt: event.payload.activity.createdAt,          });          return;         case "thread.reverted": {          const existingRows = yield* projectionThreadActivityRepository.listByThreadId({            threadId: event.payload.threadId,          });          if (existingRows.length === 0) {            return;          }          const existingTurns = yield* projectionTurnRepository.listByThreadId({            threadId: event.payload.threadId,          });          const keptRows = retainProjectionActivitiesAfterRevert(            existingRows,            existingTurns,            event.payload.turnCount,          );          if (keptRows.length === existingRows.length) {            return;          }          yield* projectionThreadActivityRepository.deleteByThreadId({            threadId: event.payload.threadId,          });          yield* Effect.forEach(keptRows, projectionThreadActivityRepository.upsert, {            concurrency: 1,          }).pipe(Effect.asVoid);          return;        }         default:          return;      }    });   const applyThreadSessionsProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      if (event.type !== "thread.session-set") {        return;      }      yield* projectionThreadSessionRepository.upsert({        threadId: event.payload.threadId,        status: event.payload.session.status,        providerName: event.payload.session.providerName,        providerSessionId: event.payload.session.providerSessionId,        providerThreadId: event.payload.session.providerThreadId,        activeTurnId: event.payload.session.activeTurnId,        lastError: event.payload.session.lastError,        updatedAt: event.payload.session.updatedAt,      });    });   const applyThreadTurnsProjection: ProjectorDefinition["apply"] = (event) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.turn-start-requested": {          yield* projectionTurnRepository.replacePendingTurnStart({            threadId: event.payload.threadId,            messageId: event.payload.messageId,            requestedAt: event.payload.createdAt,          });          return;        }         case "thread.session-set": {          const turnId = event.payload.session.activeTurnId;          if (turnId === null || event.payload.session.status !== "running") {            return;          }           const existingTurn = yield* projectionTurnRepository.getByTurnId({            threadId: event.payload.threadId,            turnId,          });          const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({            threadId: event.payload.threadId,          });

Lines 1–500 of 898. Load more