From b9ff7aa5cdf10c15cecb803f6674279c58ccdb4c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 27 Feb 2026 20:06:49 -0800 Subject: [PATCH] Make attachment writes transactional and decode attachment URLs - Stage attachment file writes until projector transactions commit, then apply write/prune/delete side effects - Remove orphaned attachment files on thread revert and delete thread attachment dirs on thread deletion - Serve URL-encoded attachment paths in `wsServer` and stream file responses - Add regression tests for rollback safety, revert/delete cleanup, and encoded attachment URLs --- .../Layers/ProjectionPipeline.test.ts | 485 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 223 +++++++- apps/server/src/wsServer.test.ts | 26 + apps/server/src/wsServer.ts | 45 +- 4 files changed, 736 insertions(+), 43 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 64aa510e19337efc60e8e6d6377f14f04807f147..b929cbb5cb2e79f1f9b4d23bf95ae4a86cc0e277 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -358,6 +358,491 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("does not persist attachment files when projector transaction rolls back", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-rollback-")); + const now = new Date().toISOString(); + + const serverConfig = { + mode: "web", + port: 0, + host: undefined, + cwd: "/tmp/project-attachments", + keybindingsConfigPath: path.join(stateDir, "keybindings.json"), + stateDir, + staticDir: undefined, + devUrl: undefined, + noBrowser: true, + authToken: undefined, + autoBootstrapProjectFromCwd: false, + logWebSocketEvents: false, + } satisfies ServerConfigShape; + + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe( + Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), + Effect.provideService(ServerConfig, serverConfig), + ); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-rollback-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-rollback"), + title: "Project Rollback", + workspaceRoot: "/tmp/project-rollback", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-rollback-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-rollback"), + projectId: ProjectId.makeUnsafe("project-rollback"), + title: "Thread Rollback", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* sql` + CREATE TRIGGER fail_thread_messages_projection_state_update + BEFORE UPDATE ON projection_state + WHEN NEW.projector = 'projection.thread-messages' + BEGIN + SELECT RAISE(ABORT, 'forced-projection-state-failure'); + END; + `; + + const result = yield* Effect.result( + appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-rollback-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-rollback"), + messageId: MessageId.makeUnsafe("message-rollback"), + role: "user", + text: "Rollback me", + attachments: [ + { + type: "image", + name: "rollback.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }), + ); + assert.equal(result._tag, "Failure"); + + const rows = yield* sql<{ + readonly count: number; + }>` + SELECT COUNT(*) AS "count" + FROM projection_thread_messages + WHERE message_id = 'message-rollback' + `; + assert.equal(rows[0]?.count ?? 0, 0); + + const attachmentPath = path.join( + stateDir, + "attachments", + "thread-rollback", + "message-rollback", + "0.png", + ); + assert.equal(fs.existsSync(attachmentPath), false); + yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`; + fs.rmSync(stateDir, { recursive: true, force: true }); + }), + ); + + it.effect("removes unreferenced attachment files when a thread is reverted", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-")); + const now = new Date().toISOString(); + + const serverConfig = { + mode: "web", + port: 0, + host: undefined, + cwd: "/tmp/project-attachments", + keybindingsConfigPath: path.join(stateDir, "keybindings.json"), + stateDir, + staticDir: undefined, + devUrl: undefined, + noBrowser: true, + authToken: undefined, + autoBootstrapProjectFromCwd: false, + logWebSocketEvents: false, + } satisfies ServerConfigShape; + + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe( + Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), + Effect.provideService(ServerConfig, serverConfig), + ); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-revert-files-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-revert-files"), + title: "Project Revert Files", + workspaceRoot: "/tmp/project-revert-files", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-revert-files-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + projectId: ProjectId.makeUnsafe("project-revert-files"), + title: "Thread Revert Files", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.makeUnsafe("evt-revert-files-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + turnId: TurnId.makeUnsafe("turn-keep"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert-files/turn/1"), + status: "ready", + files: [], + assistantMessageId: MessageId.makeUnsafe("message-keep"), + completedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-revert-files-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-4"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + messageId: MessageId.makeUnsafe("message-keep"), + role: "assistant", + text: "Keep", + attachments: [ + { + type: "image", + name: "keep.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: TurnId.makeUnsafe("turn-keep"), + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.makeUnsafe("evt-revert-files-5"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-5"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-5"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + turnId: TurnId.makeUnsafe("turn-remove"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert-files/turn/2"), + status: "ready", + files: [], + assistantMessageId: MessageId.makeUnsafe("message-remove"), + completedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-revert-files-6"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-6"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-6"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + messageId: MessageId.makeUnsafe("message-remove"), + role: "assistant", + text: "Remove", + attachments: [ + { + type: "image", + name: "remove.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,V29ybGQ=", + }, + ], + turnId: TurnId.makeUnsafe("turn-remove"), + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + const keepPath = path.join( + stateDir, + "attachments", + "thread-revert-files", + "message-keep", + "0.png", + ); + const removePath = path.join( + stateDir, + "attachments", + "thread-revert-files", + "message-remove", + "0.png", + ); + assert.equal(fs.existsSync(keepPath), true); + assert.equal(fs.existsSync(removePath), true); + + yield* appendAndProject({ + type: "thread.reverted", + eventId: EventId.makeUnsafe("evt-revert-files-7"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-revert-files-7"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-files-7"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert-files"), + turnCount: 1, + }, + }); + + assert.equal(fs.existsSync(keepPath), true); + assert.equal(fs.existsSync(removePath), false); + fs.rmSync(stateDir, { recursive: true, force: true }); + }), + ); + + it.effect("removes thread attachment directory when thread is deleted", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-")); + const now = new Date().toISOString(); + + const serverConfig = { + mode: "web", + port: 0, + host: undefined, + cwd: "/tmp/project-attachments", + keybindingsConfigPath: path.join(stateDir, "keybindings.json"), + stateDir, + staticDir: undefined, + devUrl: undefined, + noBrowser: true, + authToken: undefined, + autoBootstrapProjectFromCwd: false, + logWebSocketEvents: false, + } satisfies ServerConfigShape; + + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe( + Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), + Effect.provideService(ServerConfig, serverConfig), + ); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-delete-files-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-delete-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-delete-files-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-delete-files-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-delete-files"), + title: "Project Delete Files", + workspaceRoot: "/tmp/project-delete-files", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-delete-files-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-delete-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-delete-files-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-delete-files-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-delete-files"), + projectId: ProjectId.makeUnsafe("project-delete-files"), + title: "Thread Delete Files", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-delete-files-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-delete-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-delete-files-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-delete-files-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-delete-files"), + messageId: MessageId.makeUnsafe("message-delete-files"), + role: "user", + text: "Delete", + attachments: [ + { + type: "image", + name: "delete.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + const threadAttachmentDir = path.join(stateDir, "attachments", "thread-delete-files"); + assert.equal(fs.existsSync(threadAttachmentDir), true); + + yield* appendAndProject({ + type: "thread.deleted", + eventId: EventId.makeUnsafe("evt-delete-files-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-delete-files"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-delete-files-4"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-delete-files-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-delete-files"), + deletedAt: now, + }, + }); + + assert.equal(fs.existsSync(threadAttachmentDir), false); + fs.rmSync(stateDir, { recursive: true, force: true }); + }), + ); + it.effect("resumes from projector last_applied_sequence without replaying older events", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index fd6deebb5a7a126beae68da10e8a408dde50b542..27fd81eba39cc04c1429e523c14274a157a8fef2 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -54,7 +54,10 @@ type ProjectorName = interface ProjectorDefinition { readonly name: ProjectorName; - readonly apply: (event: OrchestrationEvent) => Effect.Effect; + readonly apply: ( + event: OrchestrationEvent, + attachmentSideEffects: AttachmentSideEffects, + ) => Effect.Effect; } const ATTACHMENTS_ROUTE_PREFIX = "/attachments"; @@ -82,6 +85,17 @@ const SAFE_IMAGE_FILE_EXTENSIONS = new Set([ ".ico", ]); +interface PendingAttachmentWrite { + readonly absolutePath: string; + readonly bytes: Uint8Array; +} + +interface AttachmentSideEffects { + readonly writes: Array; + readonly deletedThreadIds: Set; + readonly prunedThreadRelativePaths: Map>; +} + function parseBase64DataUrl( dataUrl: string, ): { readonly mimeType: string; readonly base64: string } | null { @@ -119,6 +133,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { readonly threadId: string; readonly messageId: string; readonly attachments: ReadonlyArray; + readonly stageFileWrite: (pendingWrite: PendingAttachmentWrite) => void; }) { if (input.attachments.length === 0) { return [] as ReadonlyArray; @@ -129,7 +144,6 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { return input.attachments; } - const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const attachmentsRootDir = path.join(serverConfig.value.stateDir, ATTACHMENTS_STATE_SUBDIRECTORY); const threadSegment = encodeURIComponent(input.threadId); @@ -138,7 +152,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { return yield* Effect.forEach( input.attachments, (attachment, index) => - Effect.gen(function* () { + Effect.sync(() => { if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) { return attachment; } @@ -162,13 +176,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { })}`; const relativePath = `${threadSegment}/${messageSegment}/${fileName}`; const absolutePath = path.join(attachmentsRootDir, threadSegment, messageSegment, fileName); - yield* fileSystem.makeDirectory(path.dirname(absolutePath), { recursive: true }); - const alreadyExists = yield* fileSystem - .exists(absolutePath) - .pipe(Effect.catch(() => Effect.succeed(false))); - if (!alreadyExists) { - yield* fileSystem.writeFile(absolutePath, bytes); - } + input.stageFileWrite({ absolutePath, bytes }); return { ...attachment, @@ -291,6 +299,130 @@ function retainProjectionActivitiesAfterRevert( ); } +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, +): Set { + const threadPrefix = `${encodeURIComponent(threadId)}/`; + const relativePaths = new Set(); + 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_STATE_SUBDIRECTORY, + ); + + yield* Effect.forEach( + input.sideEffects.deletedThreadIds, + (threadId) => { + const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId)); + return 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; + } + const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId)); + return Effect.gen(function* () { + const entries = yield* input.fileSystem + .readDirectory(threadDir, { recursive: true }) + .pipe(Effect.catch(() => Effect.succeed([] as Array))); + 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(); + for (const pendingWrite of input.sideEffects.writes) { + if (!writesByPath.has(pendingWrite.absolutePath)) { + 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 }); + const alreadyExists = yield* input.fileSystem + .exists(absolutePath) + .pipe(Effect.catch(() => Effect.succeed(false))); + if (!alreadyExists) { + yield* input.fileSystem.writeFile(absolutePath, bytes); + } + }), + { concurrency: 1 }, + ); +}); + const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const eventStore = yield* OrchestrationEventStore; @@ -306,7 +438,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const applyProjectsProjection: ProjectorDefinition["apply"] = (event) => + const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => Effect.gen(function* () { switch (event.type) { case "project.created": @@ -364,7 +496,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); - const applyThreadsProjection: ProjectorDefinition["apply"] = (event) => + const applyThreadsProjection: ProjectorDefinition["apply"] = (event, attachmentSideEffects) => Effect.gen(function* () { switch (event.type) { case "thread.created": @@ -403,6 +535,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } case "thread.deleted": { + attachmentSideEffects.deletedThreadIds.add(event.payload.threadId); const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, }); @@ -482,7 +615,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); - const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event) => + const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event, attachmentSideEffects) => Effect.gen(function* () { switch (event.type) { case "thread.message-sent": { @@ -504,8 +637,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { threadId: event.payload.threadId, messageId: event.payload.messageId, attachments: event.payload.attachments, + stageFileWrite: (pendingWrite) => { + attachmentSideEffects.writes.push(pendingWrite); + }, }).pipe( - Effect.provideService(FileSystem.FileSystem, fileSystem), Effect.provideService(Path.Path, path), Effect.catch((cause) => Effect.logWarning("failed to persist message attachments", { @@ -558,6 +693,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, { concurrency: 1, }).pipe(Effect.asVoid); + attachmentSideEffects.prunedThreadRelativePaths.set( + event.payload.threadId, + collectThreadAttachmentRelativePaths(event.payload.threadId, keptRows), + ); return; } @@ -566,7 +705,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); - const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (event) => + const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = ( + event, + _attachmentSideEffects, + ) => Effect.gen(function* () { switch (event.type) { case "thread.activity-appended": @@ -614,7 +756,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); - const applyThreadSessionsProjection: ProjectorDefinition["apply"] = (event) => + const applyThreadSessionsProjection: ProjectorDefinition["apply"] = ( + event, + _attachmentSideEffects, + ) => Effect.gen(function* () { if (event.type !== "thread.session-set") { return; @@ -633,7 +778,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { }); }); - const applyThreadTurnsProjection: ProjectorDefinition["apply"] = (event) => + const applyThreadTurnsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => Effect.gen(function* () { switch (event.type) { case "thread.turn-start-requested": { @@ -866,7 +1011,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; - const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = (event) => + const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = ( + event, + _attachmentSideEffects, + ) => Effect.gen(function* () { switch (event.type) { case "thread.activity-appended": { @@ -989,17 +1137,40 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { ]; const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) => - sql.withTransaction( - projector.apply(event).pipe( - Effect.flatMap(() => - projectionStateRepository.upsert({ + Effect.gen(function* () { + const attachmentSideEffects: AttachmentSideEffects = { + writes: [], + deletedThreadIds: new Set(), + prunedThreadRelativePaths: new Map>(), + }; + + yield* sql.withTransaction( + projector.apply(event, attachmentSideEffects).pipe( + Effect.flatMap(() => + projectionStateRepository.upsert({ + projector: projector.name, + lastAppliedSequence: event.sequence, + updatedAt: event.occurredAt, + }), + ), + ), + ); + + yield* runAttachmentSideEffects({ + sideEffects: attachmentSideEffects, + fileSystem, + path, + }).pipe( + Effect.catch((cause) => + Effect.logWarning("failed to apply projected attachment side-effects", { projector: projector.name, - lastAppliedSequence: event.sequence, - updatedAt: event.occurredAt, + sequence: event.sequence, + eventType: event.type, + cause, }), ), - ), - ); + ); + }); const bootstrapProjector = (projector: ProjectorDefinition) => projectionStateRepository diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 07ef0ab7db98f56172b4a1c8b49815a65bd54301..c56ed157a06a5f1e706f13b184d45dfb7189d380 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -487,6 +487,32 @@ describe("WebSocket Server", () => { expect(bytes).toEqual(Buffer.from("hello-attachment")); }); + it("serves persisted attachments for URL-encoded paths", async () => { + const stateDir = makeTempDir("t3code-state-attachments-encoded-"); + const attachmentPath = path.join( + stateDir, + "attachments", + "thread folder", + "message folder", + "file name.png", + ); + fs.mkdirSync(path.dirname(attachmentPath), { recursive: true }); + fs.writeFileSync(attachmentPath, Buffer.from("hello-encoded-attachment")); + + server = await createTestServer({ cwd: "/test/project", stateDir }); + const addr = server.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + expect(port).toBeGreaterThan(0); + + const response = await fetch( + `http://127.0.0.1:${port}/attachments/thread%20folder/message%20folder/file%20name.png`, + ); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("image/png"); + const bytes = Buffer.from(await response.arrayBuffer()); + expect(bytes).toEqual(Buffer.from("hello-encoded-attachment")); + }); + it("serves static index for root path", async () => { const stateDir = makeTempDir("t3code-state-static-root-"); const staticDir = makeTempDir("t3code-static-root-"); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index e65f1379a02cfccd0b40af8d9174b2b5e1dda188..16f035645ab6d2fc52b38f25179c6aaa741479b3 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -253,7 +253,14 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< if (url.pathname.startsWith(ATTACHMENTS_ROUTE_PREFIX)) { const attachmentsRoot = path.resolve(path.join(serverConfig.stateDir, "attachments")); - const rawRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length); + const encodedRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length); + let rawRelativePath: string; + try { + rawRelativePath = decodeURIComponent(encodedRelativePath); + } catch { + respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path"); + return; + } const normalizedRelativePath = path.normalize(rawRelativePath).replace(/^[/\\]+/, ""); if ( @@ -279,23 +286,27 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< return; } - const data = yield* fileSystem - .readFile(filePath) - .pipe(Effect.catch(() => Effect.succeed(null))); - if (!data) { - respond(500, { "Content-Type": "text/plain" }, "Internal Server Error"); - return; - } - const contentType = Mime.getType(filePath) ?? "application/octet-stream"; - respond( - 200, - { - "Content-Type": contentType, - "Cache-Control": "public, max-age=31536000, immutable", - }, - data, - ); + res.writeHead(200, { + "Content-Type": contentType, + "Cache-Control": "public, max-age=31536000, immutable", + }); + const streamExit = yield* Stream.runForEach(fileSystem.stream(filePath), (chunk) => + Effect.sync(() => { + if (!res.destroyed) { + res.write(chunk); + } + }), + ).pipe(Effect.exit); + if (streamExit._tag === "Failure") { + if (!res.destroyed) { + res.destroy(); + } + return; + } + if (!res.writableEnded) { + res.end(); + } return; } -- 2.54.0