| ... | @@ -77,7 +77,7 @@ interface AttachmentSideEffects { | ... | @@ -77,7 +77,7 @@ interface AttachmentSideEffects { |
| 77 | readonly prunedThreadRelativePaths: Map<string, Set<string>>; | 77 | readonly prunedThreadRelativePaths: Map<string, Set<string>>; |
| 78 | } | 78 | } |
| 79 | | 79 | |
| 80 | const materializeAttachmentsForProjection = Effect.fn( | 80 | const materializeAttachmentsForProjection = Effect.fn("materializeAttachmentsForProjection")( |
| 81 | (input: { readonly attachments: ReadonlyArray<ChatAttachment> }) => | 81 | (input: { readonly attachments: ReadonlyArray<ChatAttachment> }) => |
| 82 | Effect.succeed(input.attachments.length === 0 ? [] : input.attachments), | 82 | Effect.succeed(input.attachments.length === 0 ? [] : input.attachments), |
| 83 | ); | 83 | ); |
| ... | @@ -238,124 +238,145 @@ function collectThreadAttachmentRelativePaths( | ... | @@ -238,124 +238,145 @@ function collectThreadAttachmentRelativePaths( |
| 238 | return relativePaths; | 238 | return relativePaths; |
| 239 | } | 239 | } |
| 240 | | 240 | |
| 241 | const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSideEffects) { | 241 | const runAttachmentSideEffects = Effect.fn("runAttachmentSideEffects")(function* ( |
| | 242 | sideEffects: AttachmentSideEffects, |
| | 243 | ) { |
| 242 | const serverConfig = yield* Effect.service(ServerConfig); | 244 | const serverConfig = yield* Effect.service(ServerConfig); |
| 243 | const fileSystem = yield* Effect.service(FileSystem.FileSystem); | 245 | const fileSystem = yield* Effect.service(FileSystem.FileSystem); |
| 244 | const path = yield* Effect.service(Path.Path); | 246 | const path = yield* Effect.service(Path.Path); |
| 245 | | 247 | |
| 246 | const attachmentsRootDir = serverConfig.attachmentsDir; | 248 | const attachmentsRootDir = serverConfig.attachmentsDir; |
| 247 | | 249 | const readAttachmentRootEntries = fileSystem |
| 248 | yield* Effect.forEach( | 250 | .readDirectory(attachmentsRootDir, { recursive: false }) |
| 249 | sideEffects.deletedThreadIds, | 251 | .pipe(Effect.catch(() => Effect.succeed([] as Array<string>))); |
| 250 | (threadId) => | 252 | |
| 251 | Effect.gen(function* () { | 253 | const removeDeletedThreadAttachmentEntry = Effect.fn("removeDeletedThreadAttachmentEntry")( |
| 252 | const threadSegment = toSafeThreadAttachmentSegment(threadId); | 254 | function* (threadSegment: string, entry: string) { |
| 253 | if (!threadSegment) { | 255 | const normalizedEntry = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); |
| 254 | yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", { | 256 | if (normalizedEntry.length === 0 || normalizedEntry.includes("/")) { |
| 255 | threadId, | 257 | return; |
| 256 | }); | 258 | } |
| 257 | return; | 259 | const attachmentId = parseAttachmentIdFromRelativePath(normalizedEntry); |
| 258 | } | 260 | if (!attachmentId) { |
| 259 | const entries = yield* fileSystem | 261 | return; |
| 260 | .readDirectory(attachmentsRootDir, { recursive: false }) | 262 | } |
| 261 | .pipe(Effect.catch(() => Effect.succeed([] as Array<string>))); | 263 | const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId); |
| 262 | yield* Effect.forEach( | 264 | if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) { |
| 263 | entries, | 265 | return; |
| 264 | (entry) => | 266 | } |
| 265 | Effect.gen(function* () { | 267 | yield* fileSystem.remove(path.join(attachmentsRootDir, normalizedEntry), { |
| 266 | const normalizedEntry = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); | 268 | force: true, |
| 267 | if (normalizedEntry.length === 0 || normalizedEntry.includes("/")) { | 269 | }); |
| 268 | return; | 270 | }, |
| 269 | } | | |
| 270 | const attachmentId = parseAttachmentIdFromRelativePath(normalizedEntry); | | |
| 271 | if (!attachmentId) { | | |
| 272 | return; | | |
| 273 | } | | |
| 274 | const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId); | | |
| 275 | if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) { | | |
| 276 | return; | | |
| 277 | } | | |
| 278 | yield* fileSystem.remove(path.join(attachmentsRootDir, normalizedEntry), { | | |
| 279 | force: true, | | |
| 280 | }); | | |
| 281 | }), | | |
| 282 | { concurrency: 1 }, | | |
| 283 | ); | | |
| 284 | }), | | |
| 285 | { concurrency: 1 }, | | |
| 286 | ); | 271 | ); |
| 287 | | 272 | |
| | 273 | const deleteThreadAttachments = Effect.fn("deleteThreadAttachments")(function* ( |
| | 274 | threadId: string, |
| | 275 | ) { |
| | 276 | const threadSegment = toSafeThreadAttachmentSegment(threadId); |
| | 277 | if (!threadSegment) { |
| | 278 | yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", { |
| | 279 | threadId, |
| | 280 | }); |
| | 281 | return; |
| | 282 | } |
| | 283 | |
| | 284 | const entries = yield* readAttachmentRootEntries; |
| | 285 | yield* Effect.forEach( |
| | 286 | entries, |
| | 287 | (entry) => removeDeletedThreadAttachmentEntry(threadSegment, entry), |
| | 288 | { |
| | 289 | concurrency: 1, |
| | 290 | }, |
| | 291 | ); |
| | 292 | }); |
| | 293 | |
| | 294 | const pruneThreadAttachmentEntry = Effect.fn("pruneThreadAttachmentEntry")(function* ( |
| | 295 | threadSegment: string, |
| | 296 | keptThreadRelativePaths: Set<string>, |
| | 297 | entry: string, |
| | 298 | ) { |
| | 299 | const relativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); |
| | 300 | if (relativePath.length === 0 || relativePath.includes("/")) { |
| | 301 | return; |
| | 302 | } |
| | 303 | const attachmentId = parseAttachmentIdFromRelativePath(relativePath); |
| | 304 | if (!attachmentId) { |
| | 305 | return; |
| | 306 | } |
| | 307 | const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId); |
| | 308 | if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) { |
| | 309 | return; |
| | 310 | } |
| | 311 | |
| | 312 | const absolutePath = path.join(attachmentsRootDir, relativePath); |
| | 313 | const fileInfo = yield* fileSystem |
| | 314 | .stat(absolutePath) |
| | 315 | .pipe(Effect.catch(() => Effect.succeed(null))); |
| | 316 | if (!fileInfo || fileInfo.type !== "File") { |
| | 317 | return; |
| | 318 | } |
| | 319 | |
| | 320 | if (!keptThreadRelativePaths.has(relativePath)) { |
| | 321 | yield* fileSystem.remove(absolutePath, { force: true }); |
| | 322 | } |
| | 323 | }); |
| | 324 | |
| | 325 | const pruneThreadAttachments = Effect.fn("pruneThreadAttachments")(function* ( |
| | 326 | threadId: string, |
| | 327 | keptThreadRelativePaths: Set<string>, |
| | 328 | ) { |
| | 329 | if (sideEffects.deletedThreadIds.has(threadId)) { |
| | 330 | return; |
| | 331 | } |
| | 332 | |
| | 333 | const threadSegment = toSafeThreadAttachmentSegment(threadId); |
| | 334 | if (!threadSegment) { |
| | 335 | yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId }); |
| | 336 | return; |
| | 337 | } |
| | 338 | |
| | 339 | const entries = yield* readAttachmentRootEntries; |
| | 340 | yield* Effect.forEach( |
| | 341 | entries, |
| | 342 | (entry) => pruneThreadAttachmentEntry(threadSegment, keptThreadRelativePaths, entry), |
| | 343 | { concurrency: 1 }, |
| | 344 | ); |
| | 345 | }); |
| | 346 | |
| | 347 | yield* Effect.forEach(sideEffects.deletedThreadIds, deleteThreadAttachments, { |
| | 348 | concurrency: 1, |
| | 349 | }); |
| | 350 | |
| 288 | yield* Effect.forEach( | 351 | yield* Effect.forEach( |
| 289 | sideEffects.prunedThreadRelativePaths.entries(), | 352 | sideEffects.prunedThreadRelativePaths.entries(), |
| 290 | ([threadId, keptThreadRelativePaths]) => { | 353 | ([threadId, keptThreadRelativePaths]) => |
| 291 | if (sideEffects.deletedThreadIds.has(threadId)) { | 354 | pruneThreadAttachments(threadId, keptThreadRelativePaths), |
| 292 | return Effect.void; | | |
| 293 | } | | |
| 294 | return Effect.gen(function* () { | | |
| 295 | const threadSegment = toSafeThreadAttachmentSegment(threadId); | | |
| 296 | if (!threadSegment) { | | |
| 297 | yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId }); | | |
| 298 | return; | | |
| 299 | } | | |
| 300 | const entries = yield* fileSystem | | |
| 301 | .readDirectory(attachmentsRootDir, { recursive: false }) | | |
| 302 | .pipe(Effect.catch(() => Effect.succeed([] as Array<string>))); | | |
| 303 | yield* Effect.forEach( | | |
| 304 | entries, | | |
| 305 | (entry) => | | |
| 306 | Effect.gen(function* () { | | |
| 307 | const relativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); | | |
| 308 | if (relativePath.length === 0 || relativePath.includes("/")) { | | |
| 309 | return; | | |
| 310 | } | | |
| 311 | const attachmentId = parseAttachmentIdFromRelativePath(relativePath); | | |
| 312 | if (!attachmentId) { | | |
| 313 | return; | | |
| 314 | } | | |
| 315 | const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId); | | |
| 316 | if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) { | | |
| 317 | return; | | |
| 318 | } | | |
| 319 | | | |
| 320 | const absolutePath = path.join(attachmentsRootDir, relativePath); | | |
| 321 | const fileInfo = yield* fileSystem | | |
| 322 | .stat(absolutePath) | | |
| 323 | .pipe(Effect.catch(() => Effect.succeed(null))); | | |
| 324 | if (!fileInfo || fileInfo.type !== "File") { | | |
| 325 | return; | | |
| 326 | } | | |
| 327 | | | |
| 328 | if (!keptThreadRelativePaths.has(relativePath)) { | | |
| 329 | yield* fileSystem.remove(absolutePath, { force: true }); | | |
| 330 | } | | |
| 331 | }), | | |
| 332 | { concurrency: 1 }, | | |
| 333 | ); | | |
| 334 | }); | | |
| 335 | }, | | |
| 336 | { concurrency: 1 }, | 355 | { concurrency: 1 }, |
| 337 | ); | 356 | ); |
| 338 | }); | 357 | }); |
| 339 | | 358 | |
| 340 | const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | 359 | const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjectionPipeline")( |
| 341 | const sql = yield* SqlClient.SqlClient; | 360 | function* () { |
| 342 | const eventStore = yield* OrchestrationEventStore; | 361 | const sql = yield* SqlClient.SqlClient; |
| 343 | const projectionStateRepository = yield* ProjectionStateRepository; | 362 | const eventStore = yield* OrchestrationEventStore; |
| 344 | const projectionProjectRepository = yield* ProjectionProjectRepository; | 363 | const projectionStateRepository = yield* ProjectionStateRepository; |
| 345 | const projectionThreadRepository = yield* ProjectionThreadRepository; | 364 | const projectionProjectRepository = yield* ProjectionProjectRepository; |
| 346 | const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository; | 365 | const projectionThreadRepository = yield* ProjectionThreadRepository; |
| 347 | const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository; | 366 | const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository; |
| 348 | const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository; | 367 | const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository; |
| 349 | const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; | 368 | const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository; |
| 350 | const projectionTurnRepository = yield* ProjectionTurnRepository; | 369 | const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; |
| 351 | const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository; | 370 | const projectionTurnRepository = yield* ProjectionTurnRepository; |
| 352 | | 371 | const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository; |
| 353 | const fileSystem = yield* FileSystem.FileSystem; | 372 | |
| 354 | const path = yield* Path.Path; | 373 | const fileSystem = yield* FileSystem.FileSystem; |
| 355 | const serverConfig = yield* ServerConfig; | 374 | const path = yield* Path.Path; |
| 356 | | 375 | const serverConfig = yield* ServerConfig; |
| 357 | const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => | 376 | |
| 358 | Effect.gen(function* () { | 377 | const applyProjectsProjection: ProjectorDefinition["apply"] = Effect.fn( |
| | 378 | "applyProjectsProjection", |
| | 379 | )(function* (event, _attachmentSideEffects) { |
| 359 | switch (event.type) { | 380 | switch (event.type) { |
| 360 | case "project.created": | 381 | case "project.created": |
| 361 | yield* projectionProjectRepository.upsert({ | 382 | yield* projectionProjectRepository.upsert({ |
| ... | @@ -412,8 +433,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -412,8 +433,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 412 | } | 433 | } |
| 413 | }); | 434 | }); |
| 414 | | 435 | |
| 415 | const applyThreadsProjection: ProjectorDefinition["apply"] = (event, attachmentSideEffects) => | 436 | const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 416 | Effect.gen(function* () { | 437 | "applyThreadsProjection", |
| | 438 | )(function* (event, attachmentSideEffects) { |
| 417 | switch (event.type) { | 439 | switch (event.type) { |
| 418 | case "thread.created": | 440 | case "thread.created": |
| 419 | yield* projectionThreadRepository.upsert({ | 441 | yield* projectionThreadRepository.upsert({ |
| ... | @@ -597,11 +619,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -597,11 +619,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 597 | } | 619 | } |
| 598 | }); | 620 | }); |
| 599 | | 621 | |
| 600 | const applyThreadMessagesProjection: ProjectorDefinition["apply"] = ( | 622 | const applyThreadMessagesProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 601 | event, | 623 | "applyThreadMessagesProjection", |
| 602 | attachmentSideEffects, | 624 | )(function* (event, attachmentSideEffects) { |
| 603 | ) => | | |
| 604 | Effect.gen(function* () { | | |
| 605 | switch (event.type) { | 625 | switch (event.type) { |
| 606 | case "thread.message-sent": { | 626 | case "thread.message-sent": { |
| 607 | const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ | 627 | const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ |
| ... | @@ -674,11 +694,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -674,11 +694,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 674 | } | 694 | } |
| 675 | }); | 695 | }); |
| 676 | | 696 | |
| 677 | const applyThreadProposedPlansProjection: ProjectorDefinition["apply"] = ( | 697 | const applyThreadProposedPlansProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 678 | event, | 698 | "applyThreadProposedPlansProjection", |
| 679 | _attachmentSideEffects, | 699 | )(function* (event, _attachmentSideEffects) { |
| 680 | ) => | | |
| 681 | Effect.gen(function* () { | | |
| 682 | switch (event.type) { | 700 | switch (event.type) { |
| 683 | case "thread.proposed-plan-upserted": | 701 | case "thread.proposed-plan-upserted": |
| 684 | yield* projectionThreadProposedPlanRepository.upsert({ | 702 | yield* projectionThreadProposedPlanRepository.upsert({ |
| ... | @@ -727,11 +745,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -727,11 +745,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 727 | } | 745 | } |
| 728 | }); | 746 | }); |
| 729 | | 747 | |
| 730 | const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = ( | 748 | const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 731 | event, | 749 | "applyThreadActivitiesProjection", |
| 732 | _attachmentSideEffects, | 750 | )(function* (event, _attachmentSideEffects) { |
| 733 | ) => | | |
| 734 | Effect.gen(function* () { | | |
| 735 | switch (event.type) { | 751 | switch (event.type) { |
| 736 | case "thread.activity-appended": | 752 | case "thread.activity-appended": |
| 737 | yield* projectionThreadActivityRepository.upsert({ | 753 | yield* projectionThreadActivityRepository.upsert({ |
| ... | @@ -781,11 +797,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -781,11 +797,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 781 | } | 797 | } |
| 782 | }); | 798 | }); |
| 783 | | 799 | |
| 784 | const applyThreadSessionsProjection: ProjectorDefinition["apply"] = ( | 800 | const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 785 | event, | 801 | "applyThreadSessionsProjection", |
| 786 | _attachmentSideEffects, | 802 | )(function* (event, _attachmentSideEffects) { |
| 787 | ) => | | |
| 788 | Effect.gen(function* () { | | |
| 789 | if (event.type !== "thread.session-set") { | 803 | if (event.type !== "thread.session-set") { |
| 790 | return; | 804 | return; |
| 791 | } | 805 | } |
| ... | @@ -800,11 +814,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -800,11 +814,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 800 | }); | 814 | }); |
| 801 | }); | 815 | }); |
| 802 | | 816 | |
| 803 | const applyThreadTurnsProjection: ProjectorDefinition["apply"] = ( | 817 | const applyThreadTurnsProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 804 | event, | 818 | "applyThreadTurnsProjection", |
| 805 | _attachmentSideEffects, | 819 | )(function* (event, _attachmentSideEffects) { |
| 806 | ) => | | |
| 807 | Effect.gen(function* () { | | |
| 808 | switch (event.type) { | 820 | switch (event.type) { |
| 809 | case "thread.turn-start-requested": { | 821 | case "thread.turn-start-requested": { |
| 810 | yield* projectionTurnRepository.replacePendingTurnStart({ | 822 | yield* projectionTurnRepository.replacePendingTurnStart({ |
| ... | @@ -1058,13 +1070,11 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -1058,13 +1070,11 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 1058 | } | 1070 | } |
| 1059 | }); | 1071 | }); |
| 1060 | | 1072 | |
| 1061 | const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; | 1073 | const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; |
| 1062 | | 1074 | |
| 1063 | const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = ( | 1075 | const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = Effect.fn( |
| 1064 | event, | 1076 | "applyPendingApprovalsProjection", |
| 1065 | _attachmentSideEffects, | 1077 | )(function* (event, _attachmentSideEffects) { |
| 1066 | ) => | | |
| 1067 | Effect.gen(function* () { | | |
| 1068 | switch (event.type) { | 1078 | switch (event.type) { |
| 1069 | case "thread.activity-appended": { | 1079 | case "thread.activity-appended": { |
| 1070 | const requestId = | 1080 | const requestId = |
| ... | @@ -1150,47 +1160,49 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -1150,47 +1160,49 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 1150 | } | 1160 | } |
| 1151 | }); | 1161 | }); |
| 1152 | | 1162 | |
| 1153 | const projectors: ReadonlyArray<ProjectorDefinition> = [ | 1163 | const projectors: ReadonlyArray<ProjectorDefinition> = [ |
| 1154 | { | 1164 | { |
| 1155 | name: ORCHESTRATION_PROJECTOR_NAMES.projects, | 1165 | name: ORCHESTRATION_PROJECTOR_NAMES.projects, |
| 1156 | apply: applyProjectsProjection, | 1166 | apply: applyProjectsProjection, |
| 1157 | }, | 1167 | }, |
| 1158 | { | 1168 | { |
| 1159 | name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages, | 1169 | name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages, |
| 1160 | apply: applyThreadMessagesProjection, | 1170 | apply: applyThreadMessagesProjection, |
| 1161 | }, | 1171 | }, |
| 1162 | { | 1172 | { |
| 1163 | name: ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans, | 1173 | name: ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans, |
| 1164 | apply: applyThreadProposedPlansProjection, | 1174 | apply: applyThreadProposedPlansProjection, |
| 1165 | }, | 1175 | }, |
| 1166 | { | 1176 | { |
| 1167 | name: ORCHESTRATION_PROJECTOR_NAMES.threadActivities, | 1177 | name: ORCHESTRATION_PROJECTOR_NAMES.threadActivities, |
| 1168 | apply: applyThreadActivitiesProjection, | 1178 | apply: applyThreadActivitiesProjection, |
| 1169 | }, | 1179 | }, |
| 1170 | { | 1180 | { |
| 1171 | name: ORCHESTRATION_PROJECTOR_NAMES.threadSessions, | 1181 | name: ORCHESTRATION_PROJECTOR_NAMES.threadSessions, |
| 1172 | apply: applyThreadSessionsProjection, | 1182 | apply: applyThreadSessionsProjection, |
| 1173 | }, | 1183 | }, |
| 1174 | { | 1184 | { |
| 1175 | name: ORCHESTRATION_PROJECTOR_NAMES.threadTurns, | 1185 | name: ORCHESTRATION_PROJECTOR_NAMES.threadTurns, |
| 1176 | apply: applyThreadTurnsProjection, | 1186 | apply: applyThreadTurnsProjection, |
| 1177 | }, | 1187 | }, |
| 1178 | { | 1188 | { |
| 1179 | name: ORCHESTRATION_PROJECTOR_NAMES.checkpoints, | 1189 | name: ORCHESTRATION_PROJECTOR_NAMES.checkpoints, |
| 1180 | apply: applyCheckpointsProjection, | 1190 | apply: applyCheckpointsProjection, |
| 1181 | }, | 1191 | }, |
| 1182 | { | 1192 | { |
| 1183 | name: ORCHESTRATION_PROJECTOR_NAMES.pendingApprovals, | 1193 | name: ORCHESTRATION_PROJECTOR_NAMES.pendingApprovals, |
| 1184 | apply: applyPendingApprovalsProjection, | 1194 | apply: applyPendingApprovalsProjection, |
| 1185 | }, | 1195 | }, |
| 1186 | { | 1196 | { |
| 1187 | name: ORCHESTRATION_PROJECTOR_NAMES.threads, | 1197 | name: ORCHESTRATION_PROJECTOR_NAMES.threads, |
| 1188 | apply: applyThreadsProjection, | 1198 | apply: applyThreadsProjection, |
| 1189 | }, | 1199 | }, |
| 1190 | ]; | 1200 | ]; |
| 1191 | | 1201 | |
| 1192 | const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) => | 1202 | const runProjectorForEvent = Effect.fn("runProjectorForEvent")(function* ( |
| 1193 | Effect.gen(function* () { | 1203 | projector: ProjectorDefinition, |
| | 1204 | event: OrchestrationEvent, |
| | 1205 | ) { |
| 1194 | const attachmentSideEffects: AttachmentSideEffects = { | 1206 | const attachmentSideEffects: AttachmentSideEffects = { |
| 1195 | deletedThreadIds: new Set<string>(), | 1207 | deletedThreadIds: new Set<string>(), |
| 1196 | prunedThreadRelativePaths: new Map<string, Set<string>>(), | 1208 | prunedThreadRelativePaths: new Map<string, Set<string>>(), |
| ... | @@ -1220,63 +1232,64 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { | ... | @@ -1220,63 +1232,64 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { |
| 1220 | ); | 1232 | ); |
| 1221 | }); | 1233 | }); |
| 1222 | | 1234 | |
| 1223 | const bootstrapProjector = (projector: ProjectorDefinition) => | 1235 | const bootstrapProjector = (projector: ProjectorDefinition) => |
| 1224 | projectionStateRepository | 1236 | projectionStateRepository |
| 1225 | .getByProjector({ | 1237 | .getByProjector({ |
| 1226 | projector: projector.name, | 1238 | projector: projector.name, |
| 1227 | }) | 1239 | }) |
| 1228 | .pipe( | 1240 | .pipe( |
| 1229 | Effect.flatMap((stateRow) => | 1241 | Effect.flatMap((stateRow) => |
| 1230 | Stream.runForEach( | 1242 | Stream.runForEach( |
| 1231 | eventStore.readFromSequence( | 1243 | eventStore.readFromSequence( |
| 1232 | Option.isSome(stateRow) ? stateRow.value.lastAppliedSequence : 0, | 1244 | Option.isSome(stateRow) ? stateRow.value.lastAppliedSequence : 0, |
| | 1245 | ), |
| | 1246 | (event) => runProjectorForEvent(projector, event), |
| 1233 | ), | 1247 | ), |
| 1234 | (event) => runProjectorForEvent(projector, event), | | |
| 1235 | ), | 1248 | ), |
| | 1249 | ); |
| | 1250 | |
| | 1251 | const projectEvent: OrchestrationProjectionPipelineShape["projectEvent"] = (event) => |
| | 1252 | Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), { |
| | 1253 | concurrency: 1, |
| | 1254 | }).pipe( |
| | 1255 | Effect.provideService(FileSystem.FileSystem, fileSystem), |
| | 1256 | Effect.provideService(Path.Path, path), |
| | 1257 | Effect.provideService(ServerConfig, serverConfig), |
| | 1258 | Effect.asVoid, |
| | 1259 | Effect.catchTag("SqlError", (sqlError) => |
| | 1260 | Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)), |
| 1236 | ), | 1261 | ), |
| 1237 | ); | 1262 | ); |
| 1238 | | 1263 | |
| 1239 | const projectEvent: OrchestrationProjectionPipelineShape["projectEvent"] = (event) => | 1264 | const bootstrap: OrchestrationProjectionPipelineShape["bootstrap"] = Effect.forEach( |
| 1240 | Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), { | 1265 | projectors, |
| 1241 | concurrency: 1, | 1266 | bootstrapProjector, |
| 1242 | }).pipe( | 1267 | { concurrency: 1 }, |
| | 1268 | ).pipe( |
| 1243 | Effect.provideService(FileSystem.FileSystem, fileSystem), | 1269 | Effect.provideService(FileSystem.FileSystem, fileSystem), |
| 1244 | Effect.provideService(Path.Path, path), | 1270 | Effect.provideService(Path.Path, path), |
| 1245 | Effect.provideService(ServerConfig, serverConfig), | 1271 | Effect.provideService(ServerConfig, serverConfig), |
| 1246 | Effect.asVoid, | 1272 | Effect.asVoid, |
| | 1273 | Effect.tap(() => |
| | 1274 | Effect.log("orchestration projection pipeline bootstrapped").pipe( |
| | 1275 | Effect.annotateLogs({ projectors: projectors.length }), |
| | 1276 | ), |
| | 1277 | ), |
| 1247 | Effect.catchTag("SqlError", (sqlError) => | 1278 | Effect.catchTag("SqlError", (sqlError) => |
| 1248 | Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)), | 1279 | Effect.fail(toPersistenceSqlError("ProjectionPipeline.bootstrap:query")(sqlError)), |
| 1249 | ), | 1280 | ), |
| 1250 | ); | 1281 | ); |
| 1251 | | 1282 | |
| 1252 | const bootstrap: OrchestrationProjectionPipelineShape["bootstrap"] = Effect.forEach( | 1283 | return { |
| 1253 | projectors, | 1284 | bootstrap, |
| 1254 | bootstrapProjector, | 1285 | projectEvent, |
| 1255 | { concurrency: 1 }, | 1286 | } satisfies OrchestrationProjectionPipelineShape; |
| 1256 | ).pipe( | 1287 | }, |
| 1257 | Effect.provideService(FileSystem.FileSystem, fileSystem), | 1288 | ); |
| 1258 | Effect.provideService(Path.Path, path), | | |
| 1259 | Effect.provideService(ServerConfig, serverConfig), | | |
| 1260 | Effect.asVoid, | | |
| 1261 | Effect.tap(() => | | |
| 1262 | Effect.log("orchestration projection pipeline bootstrapped").pipe( | | |
| 1263 | Effect.annotateLogs({ projectors: projectors.length }), | | |
| 1264 | ), | | |
| 1265 | ), | | |
| 1266 | Effect.catchTag("SqlError", (sqlError) => | | |
| 1267 | Effect.fail(toPersistenceSqlError("ProjectionPipeline.bootstrap:query")(sqlError)), | | |
| 1268 | ), | | |
| 1269 | ); | | |
| 1270 | | | |
| 1271 | return { | | |
| 1272 | bootstrap, | | |
| 1273 | projectEvent, | | |
| 1274 | } satisfies OrchestrationProjectionPipelineShape; | | |
| 1275 | }); | | |
| 1276 | | 1289 | |
| 1277 | export const OrchestrationProjectionPipelineLive = Layer.effect( | 1290 | export const OrchestrationProjectionPipelineLive = Layer.effect( |
| 1278 | OrchestrationProjectionPipeline, | 1291 | OrchestrationProjectionPipeline, |
| 1279 | makeOrchestrationProjectionPipeline, | 1292 | makeOrchestrationProjectionPipeline(), |
| 1280 | ).pipe( | 1293 | ).pipe( |
| 1281 | Layer.provideMerge(NodeServices.layer), | 1294 | Layer.provideMerge(NodeServices.layer), |
| 1282 | Layer.provideMerge(ProjectionProjectRepositoryLive), | 1295 | Layer.provideMerge(ProjectionProjectRepositoryLive), |
Julius Marminge <julius0216@outlook.com> 2026-03-28 22:54:37-07:00
GitHub <noreply@github.com> 2026-03-28 22:54:37-07:00