| author |  Julius Marminge <julius0216@outlook.com> 2026-02-27 20:06:49-08:00 |
| committer |  Julius Marminge <jmarminge@gmail.com> 2026-02-27 21:30:46-08:00 |
| log | b9ff7aa5cdf10c15cecb803f6674279c58ccdb4c |
| tree | fe49eae9988a2db70eb3a60f0db057be214fb9c1 |
| parent | b5188e8ceaee3940d2fb3cd88c8fc2121fef1ae7 |
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 URLs4 files changed, 735 insertions(+), 42 deletions(-)
apps
server
src
orchestration
- wsServer.test.ts
- wsServer.ts
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+485
| ... | @@ -358,6 +358,491 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { | ... | @@ -358,6 +358,491 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { |
| 358 | }), | 358 | }), |
| 359 | ); | 359 | ); |
| 360 | 360 | ||
| 361 | it.effect("does not persist attachment files when projector transaction rolls back", () => | ||
| 362 | Effect.gen(function* () { | ||
| 363 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | ||
| 364 | const eventStore = yield* OrchestrationEventStore; | ||
| 365 | const sql = yield* SqlClient.SqlClient; | ||
| 366 | const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-rollback-")); | ||
| 367 | const now = new Date().toISOString(); | ||
| 368 | |||
| 369 | const serverConfig = { | ||
| 370 | mode: "web", | ||
| 371 | port: 0, | ||
| 372 | host: undefined, | ||
| 373 | cwd: "/tmp/project-attachments", | ||
| 374 | keybindingsConfigPath: path.join(stateDir, "keybindings.json"), | ||
| 375 | stateDir, | ||
| 376 | staticDir: undefined, | ||
| 377 | devUrl: undefined, | ||
| 378 | noBrowser: true, | ||
| 379 | authToken: undefined, | ||
| 380 | autoBootstrapProjectFromCwd: false, | ||
| 381 | logWebSocketEvents: false, | ||
| 382 | } satisfies ServerConfigShape; | ||
| 383 | |||
| 384 | const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) => | ||
| 385 | eventStore | ||
| 386 | .append(event) | ||
| 387 | .pipe( | ||
| 388 | Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), | ||
| 389 | Effect.provideService(ServerConfig, serverConfig), | ||
| 390 | ); | ||
| 391 | |||
| 392 | yield* appendAndProject({ | ||
| 393 | type: "project.created", | ||
| 394 | eventId: EventId.makeUnsafe("evt-rollback-1"), | ||
| 395 | aggregateKind: "project", | ||
| 396 | aggregateId: ProjectId.makeUnsafe("project-rollback"), | ||
| 397 | occurredAt: now, | ||
| 398 | commandId: CommandId.makeUnsafe("cmd-rollback-1"), | ||
| 399 | causationEventId: null, | ||
| 400 | correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"), | ||
| 401 | metadata: {}, | ||
| 402 | payload: { | ||
| 403 | projectId: ProjectId.makeUnsafe("project-rollback"), | ||
| 404 | title: "Project Rollback", | ||
| 405 | workspaceRoot: "/tmp/project-rollback", | ||
| 406 | defaultModel: null, | ||
| 407 | scripts: [], | ||
| 408 | createdAt: now, | ||
| 409 | updatedAt: now, | ||
| 410 | }, | ||
| 411 | }); | ||
| 412 | |||
| 413 | yield* appendAndProject({ | ||
| 414 | type: "thread.created", | ||
| 415 | eventId: EventId.makeUnsafe("evt-rollback-2"), | ||
| 416 | aggregateKind: "thread", | ||
| 417 | aggregateId: ThreadId.makeUnsafe("thread-rollback"), | ||
| 418 | occurredAt: now, | ||
| 419 | commandId: CommandId.makeUnsafe("cmd-rollback-2"), | ||
| 420 | causationEventId: null, | ||
| 421 | correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"), | ||
| 422 | metadata: {}, | ||
| 423 | payload: { | ||
| 424 | threadId: ThreadId.makeUnsafe("thread-rollback"), | ||
| 425 | projectId: ProjectId.makeUnsafe("project-rollback"), | ||
| 426 | title: "Thread Rollback", | ||
| 427 | model: "gpt-5-codex", | ||
| 428 | branch: null, | ||
| 429 | worktreePath: null, | ||
| 430 | createdAt: now, | ||
| 431 | updatedAt: now, | ||
| 432 | }, | ||
| 433 | }); | ||
| 434 | |||
| 435 | yield* sql` | ||
| 436 | CREATE TRIGGER fail_thread_messages_projection_state_update | ||
| 437 | BEFORE UPDATE ON projection_state | ||
| 438 | WHEN NEW.projector = 'projection.thread-messages' | ||
| 439 | BEGIN | ||
| 440 | SELECT RAISE(ABORT, 'forced-projection-state-failure'); | ||
| 441 | END; | ||
| 442 | `; | ||
| 443 | |||
| 444 | const result = yield* Effect.result( | ||
| 445 | appendAndProject({ | ||
| 446 | type: "thread.message-sent", | ||
| 447 | eventId: EventId.makeUnsafe("evt-rollback-3"), | ||
| 448 | aggregateKind: "thread", | ||
| 449 | aggregateId: ThreadId.makeUnsafe("thread-rollback"), | ||
| 450 | occurredAt: now, | ||
| 451 | commandId: CommandId.makeUnsafe("cmd-rollback-3"), | ||
| 452 | causationEventId: null, | ||
| 453 | correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"), | ||
| 454 | metadata: {}, | ||
| 455 | payload: { | ||
| 456 | threadId: ThreadId.makeUnsafe("thread-rollback"), | ||
| 457 | messageId: MessageId.makeUnsafe("message-rollback"), | ||
| 458 | role: "user", | ||
| 459 | text: "Rollback me", | ||
| 460 | attachments: [ | ||
| 461 | { | ||
| 462 | type: "image", | ||
| 463 | name: "rollback.png", | ||
| 464 | mimeType: "image/png", | ||
| 465 | sizeBytes: 5, | ||
| 466 | dataUrl: "data:image/png;base64,SGVsbG8=", | ||
| 467 | }, | ||
| 468 | ], | ||
| 469 | turnId: null, | ||
| 470 | streaming: false, | ||
| 471 | createdAt: now, | ||
| 472 | updatedAt: now, | ||
| 473 | }, | ||
| 474 | }), | ||
| 475 | ); | ||
| 476 | assert.equal(result._tag, "Failure"); | ||
| 477 | |||
| 478 | const rows = yield* sql<{ | ||
| 479 | readonly count: number; | ||
| 480 | }>` | ||
| 481 | SELECT COUNT(*) AS "count" | ||
| 482 | FROM projection_thread_messages | ||
| 483 | WHERE message_id = 'message-rollback' | ||
| 484 | `; | ||
| 485 | assert.equal(rows[0]?.count ?? 0, 0); | ||
| 486 | |||
| 487 | const attachmentPath = path.join( | ||
| 488 | stateDir, | ||
| 489 | "attachments", | ||
| 490 | "thread-rollback", | ||
| 491 | "message-rollback", | ||
| 492 | "0.png", | ||
| 493 | ); | ||
| 494 | assert.equal(fs.existsSync(attachmentPath), false); | ||
| 495 | yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`; | ||
| 496 | fs.rmSync(stateDir, { recursive: true, force: true }); | ||
| 497 | }), | ||
| 498 | ); | ||
| 499 | |||
| 500 | it.effect("removes unreferenced attachment files when a thread is reverted", () => | ||
| 501 | Effect.gen(function* () { | ||
| 502 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | ||
| 503 | const eventStore = yield* OrchestrationEventStore; | ||
| 504 | const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-")); | ||
| 505 | const now = new Date().toISOString(); | ||
| 506 | |||
| 507 | const serverConfig = { | ||
| 508 | mode: "web", | ||
| 509 | port: 0, | ||
| 510 | host: undefined, | ||
| 511 | cwd: "/tmp/project-attachments", | ||
| 512 | keybindingsConfigPath: path.join(stateDir, "keybindings.json"), | ||
| 513 | stateDir, | ||
| 514 | staticDir: undefined, | ||
| 515 | devUrl: undefined, | ||
| 516 | noBrowser: true, | ||
| 517 | authToken: undefined, | ||
| 518 | autoBootstrapProjectFromCwd: false, | ||
| 519 | logWebSocketEvents: false, | ||
| 520 | } satisfies ServerConfigShape; | ||
| 521 | |||
| 522 | const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) => | ||
| 523 | eventStore | ||
| 524 | .append(event) | ||
| 525 | .pipe( | ||
| 526 | Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), | ||
| 527 | Effect.provideService(ServerConfig, serverConfig), | ||
| 528 | ); | ||
| 529 | |||
| 530 | yield* appendAndProject({ | ||
| 531 | type: "project.created", | ||
| 532 | eventId: EventId.makeUnsafe("evt-revert-files-1"), | ||
| 533 | aggregateKind: "project", | ||
| 534 | aggregateId: ProjectId.makeUnsafe("project-revert-files"), | ||
| 535 | occurredAt: now, | ||
| 536 | commandId: CommandId.makeUnsafe("cmd-revert-files-1"), | ||
| 537 | causationEventId: null, | ||
| 538 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-1"), | ||
| 539 | metadata: {}, | ||
| 540 | payload: { | ||
| 541 | projectId: ProjectId.makeUnsafe("project-revert-files"), | ||
| 542 | title: "Project Revert Files", | ||
| 543 | workspaceRoot: "/tmp/project-revert-files", | ||
| 544 | defaultModel: null, | ||
| 545 | scripts: [], | ||
| 546 | createdAt: now, | ||
| 547 | updatedAt: now, | ||
| 548 | }, | ||
| 549 | }); | ||
| 550 | |||
| 551 | yield* appendAndProject({ | ||
| 552 | type: "thread.created", | ||
| 553 | eventId: EventId.makeUnsafe("evt-revert-files-2"), | ||
| 554 | aggregateKind: "thread", | ||
| 555 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 556 | occurredAt: now, | ||
| 557 | commandId: CommandId.makeUnsafe("cmd-revert-files-2"), | ||
| 558 | causationEventId: null, | ||
| 559 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-2"), | ||
| 560 | metadata: {}, | ||
| 561 | payload: { | ||
| 562 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 563 | projectId: ProjectId.makeUnsafe("project-revert-files"), | ||
| 564 | title: "Thread Revert Files", | ||
| 565 | model: "gpt-5-codex", | ||
| 566 | branch: null, | ||
| 567 | worktreePath: null, | ||
| 568 | createdAt: now, | ||
| 569 | updatedAt: now, | ||
| 570 | }, | ||
| 571 | }); | ||
| 572 | |||
| 573 | yield* appendAndProject({ | ||
| 574 | type: "thread.turn-diff-completed", | ||
| 575 | eventId: EventId.makeUnsafe("evt-revert-files-3"), | ||
| 576 | aggregateKind: "thread", | ||
| 577 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 578 | occurredAt: now, | ||
| 579 | commandId: CommandId.makeUnsafe("cmd-revert-files-3"), | ||
| 580 | causationEventId: null, | ||
| 581 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-3"), | ||
| 582 | metadata: {}, | ||
| 583 | payload: { | ||
| 584 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 585 | turnId: TurnId.makeUnsafe("turn-keep"), | ||
| 586 | checkpointTurnCount: 1, | ||
| 587 | checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert-files/turn/1"), | ||
| 588 | status: "ready", | ||
| 589 | files: [], | ||
| 590 | assistantMessageId: MessageId.makeUnsafe("message-keep"), | ||
| 591 | completedAt: now, | ||
| 592 | }, | ||
| 593 | }); | ||
| 594 | |||
| 595 | yield* appendAndProject({ | ||
| 596 | type: "thread.message-sent", | ||
| 597 | eventId: EventId.makeUnsafe("evt-revert-files-4"), | ||
| 598 | aggregateKind: "thread", | ||
| 599 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 600 | occurredAt: now, | ||
| 601 | commandId: CommandId.makeUnsafe("cmd-revert-files-4"), | ||
| 602 | causationEventId: null, | ||
| 603 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-4"), | ||
| 604 | metadata: {}, | ||
| 605 | payload: { | ||
| 606 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 607 | messageId: MessageId.makeUnsafe("message-keep"), | ||
| 608 | role: "assistant", | ||
| 609 | text: "Keep", | ||
| 610 | attachments: [ | ||
| 611 | { | ||
| 612 | type: "image", | ||
| 613 | name: "keep.png", | ||
| 614 | mimeType: "image/png", | ||
| 615 | sizeBytes: 5, | ||
| 616 | dataUrl: "data:image/png;base64,SGVsbG8=", | ||
| 617 | }, | ||
| 618 | ], | ||
| 619 | turnId: TurnId.makeUnsafe("turn-keep"), | ||
| 620 | streaming: false, | ||
| 621 | createdAt: now, | ||
| 622 | updatedAt: now, | ||
| 623 | }, | ||
| 624 | }); | ||
| 625 | |||
| 626 | yield* appendAndProject({ | ||
| 627 | type: "thread.turn-diff-completed", | ||
| 628 | eventId: EventId.makeUnsafe("evt-revert-files-5"), | ||
| 629 | aggregateKind: "thread", | ||
| 630 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 631 | occurredAt: now, | ||
| 632 | commandId: CommandId.makeUnsafe("cmd-revert-files-5"), | ||
| 633 | causationEventId: null, | ||
| 634 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-5"), | ||
| 635 | metadata: {}, | ||
| 636 | payload: { | ||
| 637 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 638 | turnId: TurnId.makeUnsafe("turn-remove"), | ||
| 639 | checkpointTurnCount: 2, | ||
| 640 | checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert-files/turn/2"), | ||
| 641 | status: "ready", | ||
| 642 | files: [], | ||
| 643 | assistantMessageId: MessageId.makeUnsafe("message-remove"), | ||
| 644 | completedAt: now, | ||
| 645 | }, | ||
| 646 | }); | ||
| 647 | |||
| 648 | yield* appendAndProject({ | ||
| 649 | type: "thread.message-sent", | ||
| 650 | eventId: EventId.makeUnsafe("evt-revert-files-6"), | ||
| 651 | aggregateKind: "thread", | ||
| 652 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 653 | occurredAt: now, | ||
| 654 | commandId: CommandId.makeUnsafe("cmd-revert-files-6"), | ||
| 655 | causationEventId: null, | ||
| 656 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-6"), | ||
| 657 | metadata: {}, | ||
| 658 | payload: { | ||
| 659 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 660 | messageId: MessageId.makeUnsafe("message-remove"), | ||
| 661 | role: "assistant", | ||
| 662 | text: "Remove", | ||
| 663 | attachments: [ | ||
| 664 | { | ||
| 665 | type: "image", | ||
| 666 | name: "remove.png", | ||
| 667 | mimeType: "image/png", | ||
| 668 | sizeBytes: 5, | ||
| 669 | dataUrl: "data:image/png;base64,V29ybGQ=", | ||
| 670 | }, | ||
| 671 | ], | ||
| 672 | turnId: TurnId.makeUnsafe("turn-remove"), | ||
| 673 | streaming: false, | ||
| 674 | createdAt: now, | ||
| 675 | updatedAt: now, | ||
| 676 | }, | ||
| 677 | }); | ||
| 678 | |||
| 679 | const keepPath = path.join( | ||
| 680 | stateDir, | ||
| 681 | "attachments", | ||
| 682 | "thread-revert-files", | ||
| 683 | "message-keep", | ||
| 684 | "0.png", | ||
| 685 | ); | ||
| 686 | const removePath = path.join( | ||
| 687 | stateDir, | ||
| 688 | "attachments", | ||
| 689 | "thread-revert-files", | ||
| 690 | "message-remove", | ||
| 691 | "0.png", | ||
| 692 | ); | ||
| 693 | assert.equal(fs.existsSync(keepPath), true); | ||
| 694 | assert.equal(fs.existsSync(removePath), true); | ||
| 695 | |||
| 696 | yield* appendAndProject({ | ||
| 697 | type: "thread.reverted", | ||
| 698 | eventId: EventId.makeUnsafe("evt-revert-files-7"), | ||
| 699 | aggregateKind: "thread", | ||
| 700 | aggregateId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 701 | occurredAt: now, | ||
| 702 | commandId: CommandId.makeUnsafe("cmd-revert-files-7"), | ||
| 703 | causationEventId: null, | ||
| 704 | correlationId: CorrelationId.makeUnsafe("cmd-revert-files-7"), | ||
| 705 | metadata: {}, | ||
| 706 | payload: { | ||
| 707 | threadId: ThreadId.makeUnsafe("thread-revert-files"), | ||
| 708 | turnCount: 1, | ||
| 709 | }, | ||
| 710 | }); | ||
| 711 | |||
| 712 | assert.equal(fs.existsSync(keepPath), true); | ||
| 713 | assert.equal(fs.existsSync(removePath), false); | ||
| 714 | fs.rmSync(stateDir, { recursive: true, force: true }); | ||
| 715 | }), | ||
| 716 | ); | ||
| 717 | |||
| 718 | it.effect("removes thread attachment directory when thread is deleted", () => | ||
| 719 | Effect.gen(function* () { | ||
| 720 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | ||
| 721 | const eventStore = yield* OrchestrationEventStore; | ||
| 722 | const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-")); | ||
| 723 | const now = new Date().toISOString(); | ||
| 724 | |||
| 725 | const serverConfig = { | ||
| 726 | mode: "web", | ||
| 727 | port: 0, | ||
| 728 | host: undefined, | ||
| 729 | cwd: "/tmp/project-attachments", | ||
| 730 | keybindingsConfigPath: path.join(stateDir, "keybindings.json"), | ||
| 731 | stateDir, | ||
| 732 | staticDir: undefined, | ||
| 733 | devUrl: undefined, | ||
| 734 | noBrowser: true, | ||
| 735 | authToken: undefined, | ||
| 736 | autoBootstrapProjectFromCwd: false, | ||
| 737 | logWebSocketEvents: false, | ||
| 738 | } satisfies ServerConfigShape; | ||
| 739 | |||
| 740 | const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) => | ||
| 741 | eventStore | ||
| 742 | .append(event) | ||
| 743 | .pipe( | ||
| 744 | Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), | ||
| 745 | Effect.provideService(ServerConfig, serverConfig), | ||
| 746 | ); | ||
| 747 | |||
| 748 | yield* appendAndProject({ | ||
| 749 | type: "project.created", | ||
| 750 | eventId: EventId.makeUnsafe("evt-delete-files-1"), | ||
| 751 | aggregateKind: "project", | ||
| 752 | aggregateId: ProjectId.makeUnsafe("project-delete-files"), | ||
| 753 | occurredAt: now, | ||
| 754 | commandId: CommandId.makeUnsafe("cmd-delete-files-1"), | ||
| 755 | causationEventId: null, | ||
| 756 | correlationId: CorrelationId.makeUnsafe("cmd-delete-files-1"), | ||
| 757 | metadata: {}, | ||
| 758 | payload: { | ||
| 759 | projectId: ProjectId.makeUnsafe("project-delete-files"), | ||
| 760 | title: "Project Delete Files", | ||
| 761 | workspaceRoot: "/tmp/project-delete-files", | ||
| 762 | defaultModel: null, | ||
| 763 | scripts: [], | ||
| 764 | createdAt: now, | ||
| 765 | updatedAt: now, | ||
| 766 | }, | ||
| 767 | }); | ||
| 768 | |||
| 769 | yield* appendAndProject({ | ||
| 770 | type: "thread.created", | ||
| 771 | eventId: EventId.makeUnsafe("evt-delete-files-2"), | ||
| 772 | aggregateKind: "thread", | ||
| 773 | aggregateId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 774 | occurredAt: now, | ||
| 775 | commandId: CommandId.makeUnsafe("cmd-delete-files-2"), | ||
| 776 | causationEventId: null, | ||
| 777 | correlationId: CorrelationId.makeUnsafe("cmd-delete-files-2"), | ||
| 778 | metadata: {}, | ||
| 779 | payload: { | ||
| 780 | threadId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 781 | projectId: ProjectId.makeUnsafe("project-delete-files"), | ||
| 782 | title: "Thread Delete Files", | ||
| 783 | model: "gpt-5-codex", | ||
| 784 | branch: null, | ||
| 785 | worktreePath: null, | ||
| 786 | createdAt: now, | ||
| 787 | updatedAt: now, | ||
| 788 | }, | ||
| 789 | }); | ||
| 790 | |||
| 791 | yield* appendAndProject({ | ||
| 792 | type: "thread.message-sent", | ||
| 793 | eventId: EventId.makeUnsafe("evt-delete-files-3"), | ||
| 794 | aggregateKind: "thread", | ||
| 795 | aggregateId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 796 | occurredAt: now, | ||
| 797 | commandId: CommandId.makeUnsafe("cmd-delete-files-3"), | ||
| 798 | causationEventId: null, | ||
| 799 | correlationId: CorrelationId.makeUnsafe("cmd-delete-files-3"), | ||
| 800 | metadata: {}, | ||
| 801 | payload: { | ||
| 802 | threadId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 803 | messageId: MessageId.makeUnsafe("message-delete-files"), | ||
| 804 | role: "user", | ||
| 805 | text: "Delete", | ||
| 806 | attachments: [ | ||
| 807 | { | ||
| 808 | type: "image", | ||
| 809 | name: "delete.png", | ||
| 810 | mimeType: "image/png", | ||
| 811 | sizeBytes: 5, | ||
| 812 | dataUrl: "data:image/png;base64,SGVsbG8=", | ||
| 813 | }, | ||
| 814 | ], | ||
| 815 | turnId: null, | ||
| 816 | streaming: false, | ||
| 817 | createdAt: now, | ||
| 818 | updatedAt: now, | ||
| 819 | }, | ||
| 820 | }); | ||
| 821 | |||
| 822 | const threadAttachmentDir = path.join(stateDir, "attachments", "thread-delete-files"); | ||
| 823 | assert.equal(fs.existsSync(threadAttachmentDir), true); | ||
| 824 | |||
| 825 | yield* appendAndProject({ | ||
| 826 | type: "thread.deleted", | ||
| 827 | eventId: EventId.makeUnsafe("evt-delete-files-4"), | ||
| 828 | aggregateKind: "thread", | ||
| 829 | aggregateId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 830 | occurredAt: now, | ||
| 831 | commandId: CommandId.makeUnsafe("cmd-delete-files-4"), | ||
| 832 | causationEventId: null, | ||
| 833 | correlationId: CorrelationId.makeUnsafe("cmd-delete-files-4"), | ||
| 834 | metadata: {}, | ||
| 835 | payload: { | ||
| 836 | threadId: ThreadId.makeUnsafe("thread-delete-files"), | ||
| 837 | deletedAt: now, | ||
| 838 | }, | ||
| 839 | }); | ||
| 840 | |||
| 841 | assert.equal(fs.existsSync(threadAttachmentDir), false); | ||
| 842 | fs.rmSync(stateDir, { recursive: true, force: true }); | ||
| 843 | }), | ||
| 844 | ); | ||
| 845 | |||
| 361 | it.effect("resumes from projector last_applied_sequence without replaying older events", () => | 846 | it.effect("resumes from projector last_applied_sequence without replaying older events", () => |
| 362 | Effect.gen(function* () { | 847 | Effect.gen(function* () { |
| 363 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | 848 | const projectionPipeline = yield* OrchestrationProjectionPipeline; |
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+197-26
| ... | @@ -54,7 +54,10 @@ type ProjectorName = | ... | @@ -54,7 +54,10 @@ type ProjectorName = |
| 54 | 54 | ||
| 55 | interface ProjectorDefinition { | 55 | interface ProjectorDefinition { |
| 56 | readonly name: ProjectorName; | 56 | readonly name: ProjectorName; |
| 57 | readonly apply: (event: OrchestrationEvent) => Effect.Effect<void, ProjectionRepositoryError>; | 57 | readonly apply: ( |
| 58 | event: OrchestrationEvent, | ||
| 59 | attachmentSideEffects: AttachmentSideEffects, | ||
| 60 | ) => Effect.Effect<void, ProjectionRepositoryError>; | ||
| 58 | } | 61 | } |
| 59 | 62 | ||
| 60 | const ATTACHMENTS_ROUTE_PREFIX = "/attachments"; | 63 | const ATTACHMENTS_ROUTE_PREFIX = "/attachments"; |
| ... | @@ -82,6 +85,17 @@ const SAFE_IMAGE_FILE_EXTENSIONS = new Set([ | ... | @@ -82,6 +85,17 @@ const SAFE_IMAGE_FILE_EXTENSIONS = new Set([ |
| 82 | ".ico", | 85 | ".ico", |
| 83 | ]); | 86 | ]); |
| 84 | 87 | ||
| 88 | interface PendingAttachmentWrite { | ||
| 89 | readonly absolutePath: string; | ||
| 90 | readonly bytes: Uint8Array; | ||
| 91 | } | ||
| 92 | |||
| 93 | interface AttachmentSideEffects { | ||
| 94 | readonly writes: Array<PendingAttachmentWrite>; | ||
| 95 | readonly deletedThreadIds: Set<string>; | ||
| 96 | readonly prunedThreadRelativePaths: Map<string, Set<string>>; | ||
| 97 | } | ||
| 98 | |||
| 85 | function parseBase64DataUrl( | 99 | function parseBase64DataUrl( |
| 86 | dataUrl: string, | 100 | dataUrl: string, |
| 87 | ): { readonly mimeType: string; readonly base64: string } | null { | 101 | ): { readonly mimeType: string; readonly base64: string } | null { |
| ... | @@ -119,6 +133,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { | ... | @@ -119,6 +133,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { |
| 119 | readonly threadId: string; | 133 | readonly threadId: string; |
| 120 | readonly messageId: string; | 134 | readonly messageId: string; |
| 121 | readonly attachments: ReadonlyArray<ChatAttachment>; | 135 | readonly attachments: ReadonlyArray<ChatAttachment>; |
| 136 | readonly stageFileWrite: (pendingWrite: PendingAttachmentWrite) => void; | ||
| 122 | }) { | 137 | }) { |
| 123 | if (input.attachments.length === 0) { | 138 | if (input.attachments.length === 0) { |
| 124 | return [] as ReadonlyArray<ChatAttachment>; | 139 | return [] as ReadonlyArray<ChatAttachment>; |
| ... | @@ -129,7 +144,6 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { | ... | @@ -129,7 +144,6 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { |
| 129 | return input.attachments; | 144 | return input.attachments; |
| 130 | } | 145 | } |
| 131 | 146 | ||
| 132 | const fileSystem = yield* FileSystem.FileSystem; | ||
| 133 | const path = yield* Path.Path; | 147 | const path = yield* Path.Path; |
| 134 | const attachmentsRootDir = path.join(serverConfig.value.stateDir, ATTACHMENTS_STATE_SUBDIRECTORY); | 148 | const attachmentsRootDir = path.join(serverConfig.value.stateDir, ATTACHMENTS_STATE_SUBDIRECTORY); |
| 135 | const threadSegment = encodeURIComponent(input.threadId); | 149 | const threadSegment = encodeURIComponent(input.threadId); |
| ... | @@ -138,7 +152,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { | ... | @@ -138,7 +152,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { |
| 138 | return yield* Effect.forEach( | 152 | return yield* Effect.forEach( |
| 139 | input.attachments, | 153 | input.attachments, |
| 140 | (attachment, index) => | 154 | (attachment, index) => |
| 141 | Effect.gen(function* () { | 155 | Effect.sync(() => { |
| 142 | if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) { | 156 | if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) { |
| 143 | return attachment; | 157 | return attachment; |
| 144 | } | 158 | } |
| ... | @@ -162,13 +176,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { | ... | @@ -162,13 +176,7 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { |
| 162 | })}`; | 176 | })}`; |
| 163 | const relativePath = `${threadSegment}/${messageSegment}/${fileName}`; | 177 | const relativePath = `${threadSegment}/${messageSegment}/${fileName}`; |
| 164 | const absolutePath = path.join(attachmentsRootDir, threadSegment, messageSegment, fileName); | 178 | const absolutePath = path.join(attachmentsRootDir, threadSegment, messageSegment, fileName); |
| 165 | yield* fileSystem.makeDirectory(path.dirname(absolutePath), { recursive: true }); | 179 | input.stageFileWrite({ absolutePath, bytes }); |
| 166 | const alreadyExists = yield* fileSystem | ||
| 167 | .exists(absolutePath) | ||
| 168 | .pipe(Effect.catch(() => Effect.succeed(false))); | ||
| 169 | if (!alreadyExists) { | ||
| 170 | yield* fileSystem.writeFile(absolutePath, bytes); | ||
| 171 | } | ||
| 172 | 180 | ||
| 173 | return { | 181 | return { |
| 174 | ...attachment, | 182 | ...attachment, |
| ... | @@ -291,6 +299,130 @@ function retainProjectionActivitiesAfterRevert( | ... | @@ -291,6 +299,130 @@ function retainProjectionActivitiesAfterRevert( |
| 291 | ); | 299 | ); |
| 292 | } | 300 | } |
| 293 | 301 | ||
| 302 | function toAttachmentRelativePath(dataUrl: string): string | null { | ||
| 303 | const prefix = `${ATTACHMENTS_ROUTE_PREFIX}/`; | ||
| 304 | if (!dataUrl.startsWith(prefix)) { | ||
| 305 | return null; | ||
| 306 | } | ||
| 307 | const relativePath = dataUrl.slice(prefix.length).replace(/^[/\\]+/, ""); | ||
| 308 | if (relativePath.length === 0 || relativePath.startsWith("..")) { | ||
| 309 | return null; | ||
| 310 | } | ||
| 311 | return relativePath.replace(/\\/g, "/"); | ||
| 312 | } | ||
| 313 | |||
| 314 | function collectThreadAttachmentRelativePaths( | ||
| 315 | threadId: string, | ||
| 316 | messages: ReadonlyArray<ProjectionThreadMessage>, | ||
| 317 | ): Set<string> { | ||
| 318 | const threadPrefix = `${encodeURIComponent(threadId)}/`; | ||
| 319 | const relativePaths = new Set<string>(); | ||
| 320 | for (const message of messages) { | ||
| 321 | for (const attachment of message.attachments ?? []) { | ||
| 322 | if (attachment.type !== "image") { | ||
| 323 | continue; | ||
| 324 | } | ||
| 325 | const relativePath = toAttachmentRelativePath(attachment.dataUrl); | ||
| 326 | if (!relativePath || !relativePath.startsWith(threadPrefix)) { | ||
| 327 | continue; | ||
| 328 | } | ||
| 329 | const threadRelativePath = relativePath.slice(threadPrefix.length).replace(/^[/\\]+/, ""); | ||
| 330 | if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) { | ||
| 331 | continue; | ||
| 332 | } | ||
| 333 | relativePaths.add(threadRelativePath.replace(/\\/g, "/")); | ||
| 334 | } | ||
| 335 | } | ||
| 336 | return relativePaths; | ||
| 337 | } | ||
| 338 | |||
| 339 | const runAttachmentSideEffects = Effect.fn(function* (input: { | ||
| 340 | readonly sideEffects: AttachmentSideEffects; | ||
| 341 | readonly fileSystem: FileSystem.FileSystem; | ||
| 342 | readonly path: Path.Path; | ||
| 343 | }) { | ||
| 344 | const serverConfig = yield* Effect.serviceOption(ServerConfig); | ||
| 345 | if (Option.isNone(serverConfig)) { | ||
| 346 | return; | ||
| 347 | } | ||
| 348 | |||
| 349 | const attachmentsRootDir = input.path.join( | ||
| 350 | serverConfig.value.stateDir, | ||
| 351 | ATTACHMENTS_STATE_SUBDIRECTORY, | ||
| 352 | ); | ||
| 353 | |||
| 354 | yield* Effect.forEach( | ||
| 355 | input.sideEffects.deletedThreadIds, | ||
| 356 | (threadId) => { | ||
| 357 | const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId)); | ||
| 358 | return input.fileSystem.remove(threadDir, { recursive: true, force: true }); | ||
| 359 | }, | ||
| 360 | { concurrency: 1 }, | ||
| 361 | ); | ||
| 362 | |||
| 363 | yield* Effect.forEach( | ||
| 364 | input.sideEffects.prunedThreadRelativePaths.entries(), | ||
| 365 | ([threadId, keptThreadRelativePaths]) => { | ||
| 366 | if (input.sideEffects.deletedThreadIds.has(threadId)) { | ||
| 367 | return Effect.void; | ||
| 368 | } | ||
| 369 | const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId)); | ||
| 370 | return Effect.gen(function* () { | ||
| 371 | const entries = yield* input.fileSystem | ||
| 372 | .readDirectory(threadDir, { recursive: true }) | ||
| 373 | .pipe(Effect.catch(() => Effect.succeed([] as Array<string>))); | ||
| 374 | yield* Effect.forEach( | ||
| 375 | entries, | ||
| 376 | (entry) => | ||
| 377 | Effect.gen(function* () { | ||
| 378 | const threadRelativePath = entry | ||
| 379 | .replace(/^[/\\]+/, "") | ||
| 380 | .replace(/\\/g, "/"); | ||
| 381 | if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) { | ||
| 382 | return; | ||
| 383 | } | ||
| 384 | |||
| 385 | const absolutePath = input.path.join(threadDir, threadRelativePath); | ||
| 386 | const fileInfo = yield* input.fileSystem | ||
| 387 | .stat(absolutePath) | ||
| 388 | .pipe(Effect.catch(() => Effect.succeed(null))); | ||
| 389 | if (!fileInfo || fileInfo.type !== "File") { | ||
| 390 | return; | ||
| 391 | } | ||
| 392 | |||
| 393 | if (!keptThreadRelativePaths.has(threadRelativePath)) { | ||
| 394 | yield* input.fileSystem.remove(absolutePath, { force: true }); | ||
| 395 | } | ||
| 396 | }), | ||
| 397 | { concurrency: 1 }, | ||
| 398 | ); | ||
| 399 | }); | ||
| 400 | }, | ||
| 401 | { concurrency: 1 }, | ||
| 402 | ); | ||
| 403 | |||
| 404 | const writesByPath = new Map<string, Uint8Array>(); | ||
| 405 | for (const pendingWrite of input.sideEffects.writes) { | ||
| 406 | if (!writesByPath.has(pendingWrite.absolutePath)) { | ||
| 407 | writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes); | ||
| 408 | } | ||
| 409 | } | ||
| 410 | yield* Effect.forEach( | ||
| 411 | writesByPath.entries(), | ||
| 412 | ([absolutePath, bytes]) => | ||
| 413 | Effect.gen(function* () { | ||
| 414 | yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), { recursive: true }); | ||
| 415 | const alreadyExists = yield* input.fileSystem | ||
| 416 | .exists(absolutePath) | ||
| 417 | .pipe(Effect.catch(() => Effect.succeed(false))); | ||
| 418 | if (!alreadyExists) { | ||
| 419 | yield* input.fileSystem.writeFile(absolutePath, bytes); | ||
| 420 | } | ||
| 421 | }), | ||
| 422 | { concurrency: 1 }, | ||
| 423 | ); | ||
| 424 | }); | ||
| 425 | |||
| 294 | const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | 426 | const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 295 | const sql = yield* SqlClient.SqlClient; | 427 | const sql = yield* SqlClient.SqlClient; |
| 296 | const eventStore = yield* OrchestrationEventStore; | 428 | const eventStore = yield* OrchestrationEventStore; |
| ... | @@ -306,7 +438,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -306,7 +438,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 306 | const fileSystem = yield* FileSystem.FileSystem; | 438 | const fileSystem = yield* FileSystem.FileSystem; |
| 307 | const path = yield* Path.Path; | 439 | const path = yield* Path.Path; |
| 308 | 440 | ||
| 309 | const applyProjectsProjection: ProjectorDefinition["apply"] = (event) => | 441 | const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => |
| 310 | Effect.gen(function* () { | 442 | Effect.gen(function* () { |
| 311 | switch (event.type) { | 443 | switch (event.type) { |
| 312 | case "project.created": | 444 | case "project.created": |
| ... | @@ -364,7 +496,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -364,7 +496,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 364 | } | 496 | } |
| 365 | }); | 497 | }); |
| 366 | 498 | ||
| 367 | const applyThreadsProjection: ProjectorDefinition["apply"] = (event) => | 499 | const applyThreadsProjection: ProjectorDefinition["apply"] = (event, attachmentSideEffects) => |
| 368 | Effect.gen(function* () { | 500 | Effect.gen(function* () { |
| 369 | switch (event.type) { | 501 | switch (event.type) { |
| 370 | case "thread.created": | 502 | case "thread.created": |
| ... | @@ -403,6 +535,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -403,6 +535,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 403 | } | 535 | } |
| 404 | 536 | ||
| 405 | case "thread.deleted": { | 537 | case "thread.deleted": { |
| 538 | attachmentSideEffects.deletedThreadIds.add(event.payload.threadId); | ||
| 406 | const existingRow = yield* projectionThreadRepository.getById({ | 539 | const existingRow = yield* projectionThreadRepository.getById({ |
| 407 | threadId: event.payload.threadId, | 540 | threadId: event.payload.threadId, |
| 408 | }); | 541 | }); |
| ... | @@ -482,7 +615,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -482,7 +615,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 482 | } | 615 | } |
| 483 | }); | 616 | }); |
| 484 | 617 | ||
| 485 | const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event) => | 618 | const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event, attachmentSideEffects) => |
| 486 | Effect.gen(function* () { | 619 | Effect.gen(function* () { |
| 487 | switch (event.type) { | 620 | switch (event.type) { |
| 488 | case "thread.message-sent": { | 621 | case "thread.message-sent": { |
| ... | @@ -504,8 +637,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -504,8 +637,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 504 | threadId: event.payload.threadId, | 637 | threadId: event.payload.threadId, |
| 505 | messageId: event.payload.messageId, | 638 | messageId: event.payload.messageId, |
| 506 | attachments: event.payload.attachments, | 639 | attachments: event.payload.attachments, |
| 640 | stageFileWrite: (pendingWrite) => { | ||
| 641 | attachmentSideEffects.writes.push(pendingWrite); | ||
| 642 | }, | ||
| 507 | }).pipe( | 643 | }).pipe( |
| 508 | Effect.provideService(FileSystem.FileSystem, fileSystem), | ||
| 509 | Effect.provideService(Path.Path, path), | 644 | Effect.provideService(Path.Path, path), |
| 510 | Effect.catch((cause) => | 645 | Effect.catch((cause) => |
| 511 | Effect.logWarning("failed to persist message attachments", { | 646 | Effect.logWarning("failed to persist message attachments", { |
| ... | @@ -558,6 +693,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -558,6 +693,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 558 | yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, { | 693 | yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, { |
| 559 | concurrency: 1, | 694 | concurrency: 1, |
| 560 | }).pipe(Effect.asVoid); | 695 | }).pipe(Effect.asVoid); |
| 696 | attachmentSideEffects.prunedThreadRelativePaths.set( | ||
| 697 | event.payload.threadId, | ||
| 698 | collectThreadAttachmentRelativePaths(event.payload.threadId, keptRows), | ||
| 699 | ); | ||
| 561 | return; | 700 | return; |
| 562 | } | 701 | } |
| 563 | 702 | ||
| ... | @@ -566,7 +705,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -566,7 +705,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 566 | } | 705 | } |
| 567 | }); | 706 | }); |
| 568 | 707 | ||
| 569 | const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (event) => | 708 | const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = ( |
| 709 | event, | ||
| 710 | _attachmentSideEffects, | ||
| 711 | ) => | ||
| 570 | Effect.gen(function* () { | 712 | Effect.gen(function* () { |
| 571 | switch (event.type) { | 713 | switch (event.type) { |
| 572 | case "thread.activity-appended": | 714 | case "thread.activity-appended": |
| ... | @@ -614,7 +756,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -614,7 +756,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 614 | } | 756 | } |
| 615 | }); | 757 | }); |
| 616 | 758 | ||
| 617 | const applyThreadSessionsProjection: ProjectorDefinition["apply"] = (event) => | 759 | const applyThreadSessionsProjection: ProjectorDefinition["apply"] = ( |
| 760 | event, | ||
| 761 | _attachmentSideEffects, | ||
| 762 | ) => | ||
| 618 | Effect.gen(function* () { | 763 | Effect.gen(function* () { |
| 619 | if (event.type !== "thread.session-set") { | 764 | if (event.type !== "thread.session-set") { |
| 620 | return; | 765 | return; |
| ... | @@ -633,7 +778,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -633,7 +778,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 633 | }); | 778 | }); |
| 634 | }); | 779 | }); |
| 635 | 780 | ||
| 636 | const applyThreadTurnsProjection: ProjectorDefinition["apply"] = (event) => | 781 | const applyThreadTurnsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => |
| 637 | Effect.gen(function* () { | 782 | Effect.gen(function* () { |
| 638 | switch (event.type) { | 783 | switch (event.type) { |
| 639 | case "thread.turn-start-requested": { | 784 | case "thread.turn-start-requested": { |
| ... | @@ -866,7 +1011,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -866,7 +1011,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 866 | 1011 | ||
| 867 | const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; | 1012 | const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; |
| 868 | 1013 | ||
| 869 | const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = (event) => | 1014 | const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = ( |
| 1015 | event, | ||
| 1016 | _attachmentSideEffects, | ||
| 1017 | ) => | ||
| 870 | Effect.gen(function* () { | 1018 | Effect.gen(function* () { |
| 871 | switch (event.type) { | 1019 | switch (event.type) { |
| 872 | case "thread.activity-appended": { | 1020 | case "thread.activity-appended": { |
| ... | @@ -989,17 +1137,40 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -989,17 +1137,40 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 989 | ]; | 1137 | ]; |
| 990 | 1138 | ||
| 991 | const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) => | 1139 | const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) => |
| 992 | sql.withTransaction( | 1140 | Effect.gen(function* () { |
| 993 | projector.apply(event).pipe( | 1141 | const attachmentSideEffects: AttachmentSideEffects = { |
| 994 | Effect.flatMap(() => | 1142 | writes: [], |
| 995 | projectionStateRepository.upsert({ | 1143 | deletedThreadIds: new Set<string>(), |
| 1144 | prunedThreadRelativePaths: new Map<string, Set<string>>(), | ||
| 1145 | }; | ||
| 1146 | |||
| 1147 | yield* sql.withTransaction( | ||
| 1148 | projector.apply(event, attachmentSideEffects).pipe( | ||
| 1149 | Effect.flatMap(() => | ||
| 1150 | projectionStateRepository.upsert({ | ||
| 1151 | projector: projector.name, | ||
| 1152 | lastAppliedSequence: event.sequence, | ||
| 1153 | updatedAt: event.occurredAt, | ||
| 1154 | }), | ||
| 1155 | ), | ||
| 1156 | ), | ||
| 1157 | ); | ||
| 1158 | |||
| 1159 | yield* runAttachmentSideEffects({ | ||
| 1160 | sideEffects: attachmentSideEffects, | ||
| 1161 | fileSystem, | ||
| 1162 | path, | ||
| 1163 | }).pipe( | ||
| 1164 | Effect.catch((cause) => | ||
| 1165 | Effect.logWarning("failed to apply projected attachment side-effects", { | ||
| 996 | projector: projector.name, | 1166 | projector: projector.name, |
| 997 | lastAppliedSequence: event.sequence, | 1167 | sequence: event.sequence, |
| 998 | updatedAt: event.occurredAt, | 1168 | eventType: event.type, |
| 1169 | cause, | ||
| 999 | }), | 1170 | }), |
| 1000 | ), | 1171 | ), |
| 1001 | ), | 1172 | ); |
| 1002 | ); | 1173 | }); |
| 1003 | 1174 | ||
| 1004 | const bootstrapProjector = (projector: ProjectorDefinition) => | 1175 | const bootstrapProjector = (projector: ProjectorDefinition) => |
| 1005 | projectionStateRepository | 1176 | projectionStateRepository |
apps/server/src/wsServer.test.ts+26
| ... | @@ -487,6 +487,32 @@ describe("WebSocket Server", () => { | ... | @@ -487,6 +487,32 @@ describe("WebSocket Server", () => { |
| 487 | expect(bytes).toEqual(Buffer.from("hello-attachment")); | 487 | expect(bytes).toEqual(Buffer.from("hello-attachment")); |
| 488 | }); | 488 | }); |
| 489 | 489 | ||
| 490 | it("serves persisted attachments for URL-encoded paths", async () => { | ||
| 491 | const stateDir = makeTempDir("t3code-state-attachments-encoded-"); | ||
| 492 | const attachmentPath = path.join( | ||
| 493 | stateDir, | ||
| 494 | "attachments", | ||
| 495 | "thread folder", | ||
| 496 | "message folder", | ||
| 497 | "file name.png", | ||
| 498 | ); | ||
| 499 | fs.mkdirSync(path.dirname(attachmentPath), { recursive: true }); | ||
| 500 | fs.writeFileSync(attachmentPath, Buffer.from("hello-encoded-attachment")); | ||
| 501 | |||
| 502 | server = await createTestServer({ cwd: "/test/project", stateDir }); | ||
| 503 | const addr = server.address(); | ||
| 504 | const port = typeof addr === "object" && addr !== null ? addr.port : 0; | ||
| 505 | expect(port).toBeGreaterThan(0); | ||
| 506 | |||
| 507 | const response = await fetch( | ||
| 508 | `http://127.0.0.1:${port}/attachments/thread%20folder/message%20folder/file%20name.png`, | ||
| 509 | ); | ||
| 510 | expect(response.status).toBe(200); | ||
| 511 | expect(response.headers.get("content-type")).toContain("image/png"); | ||
| 512 | const bytes = Buffer.from(await response.arrayBuffer()); | ||
| 513 | expect(bytes).toEqual(Buffer.from("hello-encoded-attachment")); | ||
| 514 | }); | ||
| 515 | |||
| 490 | it("serves static index for root path", async () => { | 516 | it("serves static index for root path", async () => { |
| 491 | const stateDir = makeTempDir("t3code-state-static-root-"); | 517 | const stateDir = makeTempDir("t3code-state-static-root-"); |
| 492 | const staticDir = makeTempDir("t3code-static-root-"); | 518 | const staticDir = makeTempDir("t3code-static-root-"); |
apps/server/src/wsServer.ts+27-16
| ... | @@ -253,7 +253,14 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< | ... | @@ -253,7 +253,14 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< |
| 253 | 253 | ||
| 254 | if (url.pathname.startsWith(ATTACHMENTS_ROUTE_PREFIX)) { | 254 | if (url.pathname.startsWith(ATTACHMENTS_ROUTE_PREFIX)) { |
| 255 | const attachmentsRoot = path.resolve(path.join(serverConfig.stateDir, "attachments")); | 255 | const attachmentsRoot = path.resolve(path.join(serverConfig.stateDir, "attachments")); |
| 256 | const rawRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length); | 256 | const encodedRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length); |
| 257 | let rawRelativePath: string; | ||
| 258 | try { | ||
| 259 | rawRelativePath = decodeURIComponent(encodedRelativePath); | ||
| 260 | } catch { | ||
| 261 | respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path"); | ||
| 262 | return; | ||
| 263 | } | ||
| 257 | const normalizedRelativePath = path.normalize(rawRelativePath).replace(/^[/\\]+/, ""); | 264 | const normalizedRelativePath = path.normalize(rawRelativePath).replace(/^[/\\]+/, ""); |
| 258 | 265 | ||
| 259 | if ( | 266 | if ( |
| ... | @@ -279,23 +286,27 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< | ... | @@ -279,23 +286,27 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< |
| 279 | return; | 286 | return; |
| 280 | } | 287 | } |
| 281 | 288 | ||
| 282 | const data = yield* fileSystem | 289 | const contentType = Mime.getType(filePath) ?? "application/octet-stream"; |
| 283 | .readFile(filePath) | 290 | res.writeHead(200, { |
| 284 | .pipe(Effect.catch(() => Effect.succeed(null))); | 291 | "Content-Type": contentType, |
| 285 | if (!data) { | 292 | "Cache-Control": "public, max-age=31536000, immutable", |
| 286 | respond(500, { "Content-Type": "text/plain" }, "Internal Server Error"); | 293 | }); |
| 294 | const streamExit = yield* Stream.runForEach(fileSystem.stream(filePath), (chunk) => | ||
| 295 | Effect.sync(() => { | ||
| 296 | if (!res.destroyed) { | ||
| 297 | res.write(chunk); | ||
| 298 | } | ||
| 299 | }), | ||
| 300 | ).pipe(Effect.exit); | ||
| 301 | if (streamExit._tag === "Failure") { | ||
| 302 | if (!res.destroyed) { | ||
| 303 | res.destroy(); | ||
| 304 | } | ||
| 287 | return; | 305 | return; |
| 288 | } | 306 | } |
| 289 | 307 | if (!res.writableEnded) { | |
| 290 | const contentType = Mime.getType(filePath) ?? "application/octet-stream"; | 308 | res.end(); |
| 291 | respond( | 309 | } |
| 292 | 200, | ||
| 293 | { | ||
| 294 | "Content-Type": contentType, | ||
| 295 | "Cache-Control": "public, max-age=31536000, immutable", | ||
| 296 | }, | ||
| 297 | data, | ||
| 298 | ); | ||
| 299 | return; | 310 | return; |
| 300 | } | 311 | } |
| 301 | 312 |