import Mime from "@effect/platform-node/Mime";import {  ApprovalRequestId,  type ChatAttachment,  type OrchestrationEvent,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import { Effect, FileSystem, Layer, Option, Path, 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 { ServerConfig } from "../../config.ts";import {  OrchestrationProjectionPipeline,  type OrchestrationProjectionPipelineShape,} from "../Services/ProjectionPipeline.ts";import {  ATTACHMENTS_ROUTE_PREFIX,  IMAGE_EXTENSION_BY_MIME_TYPE,  SAFE_IMAGE_FILE_EXTENSIONS,} from "../../projectFaviconRoute.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,    attachmentSideEffects: AttachmentSideEffects,  ) => Effect.Effect<void, ProjectionRepositoryError>;} interface PendingAttachmentWrite {  readonly absolutePath: string;  readonly bytes: Uint8Array;} interface AttachmentSideEffects {  readonly writes: Array<PendingAttachmentWrite>;  readonly deletedThreadIds: Set<string>;  readonly prunedThreadRelativePaths: Map<string, Set<string>>;} function toSafeThreadAttachmentSegment(threadId: string): string | null {  const segment = encodeURIComponent(threadId);  if (    segment.length === 0 ||    segment === "." ||    segment === ".." ||    segment.includes("/") ||    segment.includes("\\") ||    segment.includes("\0")  ) {    return null;  }  return segment;} function parseBase64DataUrl(  dataUrl: string,): { readonly mimeType: string; readonly base64: string } | null {  const match = /^data:([^;,]+);base64,([a-z0-9+/=\r\n]+)$/i.exec(dataUrl.trim());  if (!match) return null;   const mimeType = match[1]?.trim().toLowerCase();  const base64 = match[2]?.replace(/\s+/g, "");  if (!mimeType || !base64) return null;   return { mimeType, base64 };} function inferImageExtension(attachment: Extract<ChatAttachment, { type: "image" }>): string {  const normalizedMimeType = attachment.mimeType.toLowerCase();  if (normalizedMimeType.startsWith("image/")) {    const fromMime = IMAGE_EXTENSION_BY_MIME_TYPE[normalizedMimeType];    if (fromMime) return fromMime;  }  const ext = Mime.getExtension(attachment.mimeType);  if (ext && SAFE_IMAGE_FILE_EXTENSIONS.has(ext)) return ext;  return ".bin";} const materializeAttachmentsForProjection = Effect.fn(function* (input: {  readonly threadId: string;  readonly messageId: string;  readonly attachments: ReadonlyArray<ChatAttachment>;  readonly stageFileWrite: (pendingWrite: PendingAttachmentWrite) => void;}) {  if (input.attachments.length === 0) return [];   const serverConfig = yield* Effect.serviceOption(ServerConfig);  if (Option.isNone(serverConfig)) return input.attachments;   const path = yield* Path.Path;  const attachmentsRootDir = path.join(serverConfig.value.stateDir, "attachments");  const threadSegment = toSafeThreadAttachmentSegment(input.threadId);  if (!threadSegment) {    yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", {      threadId: input.threadId,      messageId: input.messageId,    });    return input.attachments;  }  const messageSegment = encodeURIComponent(input.messageId);   return yield* Effect.forEach(    input.attachments,    (attachment, index) =>      Effect.sync(() => {        if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) {          return attachment;        }         const parsed = parseBase64DataUrl(attachment.dataUrl);        if (!parsed) {          return attachment;        }        if (!parsed.mimeType.startsWith("image/")) {          return attachment;        }         const bytes = Buffer.from(parsed.base64, "base64");        if (bytes.byteLength === 0) return attachment;         const fileName = `${index}${inferImageExtension({          ...attachment,          mimeType: parsed.mimeType,        })}`;        const uniqueFileName = `${messageSegment}-${fileName}`;        const relativePath = `${threadSegment}/${uniqueFileName}`;        const absolutePath = path.join(attachmentsRootDir, threadSegment, uniqueFileName);        input.stageFileWrite({ absolutePath, bytes });         return {          ...attachment,          dataUrl: `${ATTACHMENTS_ROUTE_PREFIX}/${relativePath}`,        } satisfies ChatAttachment;      }),    { concurrency: 1 },  );}); 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),  );} function toAttachmentRelativePath(dataUrl: string): string | null {  const prefix = `${ATTACHMENTS_ROUTE_PREFIX}/`;  if (!dataUrl.startsWith(prefix)) {    return null;  }  const relativePath = dataUrl.slice(prefix.length).replace(/^[/\\]+/, "");  if (relativePath.length === 0 || relativePath.startsWith("..")) {    return null;  }  return relativePath.replace(/\\/g, "/");} function collectThreadAttachmentRelativePaths(  threadId: string,  messages: ReadonlyArray<ProjectionThreadMessage>,): Set<string> {  const threadSegment = toSafeThreadAttachmentSegment(threadId);  if (!threadSegment) {    return new Set();  }  const threadPrefix = `${threadSegment}/`;  const relativePaths = new Set<string>();  for (const message of messages) {    for (const attachment of message.attachments ?? []) {      if (attachment.type !== "image") {        continue;      }      const relativePath = toAttachmentRelativePath(attachment.dataUrl);      if (!relativePath || !relativePath.startsWith(threadPrefix)) {        continue;      }      const threadRelativePath = relativePath.slice(threadPrefix.length).replace(/^[/\\]+/, "");      if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) {        continue;      }      relativePaths.add(threadRelativePath.replace(/\\/g, "/"));    }  }  return relativePaths;} const runAttachmentSideEffects = Effect.fn(function* (input: {  readonly sideEffects: AttachmentSideEffects;  readonly fileSystem: FileSystem.FileSystem;  readonly path: Path.Path;}) {  const serverConfig = yield* Effect.serviceOption(ServerConfig);  if (Option.isNone(serverConfig)) {    return;  }   const attachmentsRootDir = input.path.join(serverConfig.value.stateDir, "attachments");   yield* Effect.forEach(    input.sideEffects.deletedThreadIds,    (threadId) =>      Effect.gen(function* () {        const threadSegment = toSafeThreadAttachmentSegment(threadId);        if (!threadSegment) {          yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", {            threadId,          });          return;        }        const threadDir = input.path.join(attachmentsRootDir, threadSegment);        yield* input.fileSystem.remove(threadDir, { recursive: true, force: true });      }),    { concurrency: 1 },  );   yield* Effect.forEach(    input.sideEffects.prunedThreadRelativePaths.entries(),    ([threadId, keptThreadRelativePaths]) => {      if (input.sideEffects.deletedThreadIds.has(threadId)) {        return Effect.void;      }      return Effect.gen(function* () {        const threadSegment = toSafeThreadAttachmentSegment(threadId);        if (!threadSegment) {          yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });          return;        }        const threadDir = input.path.join(attachmentsRootDir, threadSegment);        const entries = yield* input.fileSystem          .readDirectory(threadDir, { recursive: true })          .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));        yield* Effect.forEach(          entries,          (entry) =>            Effect.gen(function* () {              const threadRelativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");              if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) {                return;              }               const absolutePath = input.path.join(threadDir, threadRelativePath);              const fileInfo = yield* input.fileSystem                .stat(absolutePath)                .pipe(Effect.catch(() => Effect.succeed(null)));              if (!fileInfo || fileInfo.type !== "File") {                return;              }               if (!keptThreadRelativePaths.has(threadRelativePath)) {                yield* input.fileSystem.remove(absolutePath, { force: true });              }            }),          { concurrency: 1 },        );      });    },    { concurrency: 1 },  );   const writesByPath = new Map<string, Uint8Array>();  for (const pendingWrite of input.sideEffects.writes) {    // Last write wins for a path so updated attachment contents replace stale bytes.    writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);  }  yield* Effect.forEach(    writesByPath.entries(),    ([absolutePath, bytes]) =>      Effect.gen(function* () {        yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), {          recursive: true,        });        yield* input.fileSystem.writeFile(absolutePath, bytes);      }),    { concurrency: 1 },  );}); 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 fileSystem = yield* FileSystem.FileSystem;  const path = yield* Path.Path;   const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) =>    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, attachmentSideEffects) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.created":          yield* projectionThreadRepository.upsert({

Lines 1–500 of 1228. Load more