| ... | @@ -1255,6 +1255,77 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-atta | ... | @@ -1255,6 +1255,77 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-atta |
| 1255 | ); | 1255 | ); |
| 1256 | | 1256 | |
| 1257 | it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { | 1257 | it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { |
| | 1258 | it.effect("replays a bootstrap backlog larger than the event store default limit", () => |
| | 1259 | Effect.gen(function* () { |
| | 1260 | const projectionPipeline = yield* OrchestrationProjectionPipeline; |
| | 1261 | const eventStore = yield* OrchestrationEventStore; |
| | 1262 | const sql = yield* SqlClient.SqlClient; |
| | 1263 | const now = "2026-01-01T00:00:00.000Z"; |
| | 1264 | const projectId = ProjectId.make("project-bootstrap-backlog"); |
| | 1265 | |
| | 1266 | const sequenceRows = yield* sql<{ readonly maxSequence: number | null }>` |
| | 1267 | SELECT MAX(sequence) AS "maxSequence" FROM orchestration_events |
| | 1268 | `; |
| | 1269 | const sequenceBeforeBacklog = sequenceRows[0]?.maxSequence ?? 0; |
| | 1270 | const appendedEvents = yield* Effect.forEach( |
| | 1271 | Array.from({ length: 1_001 }, (_, index) => index), |
| | 1272 | (index) => { |
| | 1273 | const eventId = EventId.make(`evt-bootstrap-backlog-${index}`); |
| | 1274 | const commandId = CommandId.make(`cmd-bootstrap-backlog-${index}`); |
| | 1275 | return eventStore.append({ |
| | 1276 | type: "project.created", |
| | 1277 | eventId, |
| | 1278 | aggregateKind: "project", |
| | 1279 | aggregateId: projectId, |
| | 1280 | occurredAt: now, |
| | 1281 | commandId, |
| | 1282 | causationEventId: null, |
| | 1283 | correlationId: CorrelationId.make(commandId), |
| | 1284 | metadata: {}, |
| | 1285 | payload: { |
| | 1286 | projectId, |
| | 1287 | title: `Bootstrap backlog ${index}`, |
| | 1288 | workspaceRoot: "/tmp/project-bootstrap-backlog", |
| | 1289 | defaultModelSelection: null, |
| | 1290 | scripts: [], |
| | 1291 | createdAt: now, |
| | 1292 | updatedAt: now, |
| | 1293 | }, |
| | 1294 | }); |
| | 1295 | }, |
| | 1296 | ); |
| | 1297 | const lastSequence = appendedEvents[appendedEvents.length - 1]!.sequence; |
| | 1298 | |
| | 1299 | yield* Effect.forEach( |
| | 1300 | Object.values(ORCHESTRATION_PROJECTOR_NAMES), |
| | 1301 | (projector) => { |
| | 1302 | const lastAppliedSequence = |
| | 1303 | projector === ORCHESTRATION_PROJECTOR_NAMES.projects |
| | 1304 | ? sequenceBeforeBacklog |
| | 1305 | : lastSequence; |
| | 1306 | return sql` |
| | 1307 | INSERT INTO projection_state (projector, last_applied_sequence, updated_at) |
| | 1308 | VALUES (${projector}, ${lastAppliedSequence}, ${now}) |
| | 1309 | ON CONFLICT (projector) |
| | 1310 | DO UPDATE SET |
| | 1311 | last_applied_sequence = excluded.last_applied_sequence, |
| | 1312 | updated_at = excluded.updated_at |
| | 1313 | `; |
| | 1314 | }, |
| | 1315 | { discard: true }, |
| | 1316 | ); |
| | 1317 | |
| | 1318 | yield* projectionPipeline.bootstrap; |
| | 1319 | |
| | 1320 | const stateRows = yield* sql<{ readonly lastAppliedSequence: number }>` |
| | 1321 | SELECT last_applied_sequence AS "lastAppliedSequence" |
| | 1322 | FROM projection_state |
| | 1323 | WHERE projector = ${ORCHESTRATION_PROJECTOR_NAMES.projects} |
| | 1324 | `; |
| | 1325 | assert.deepEqual(stateRows, [{ lastAppliedSequence: lastSequence }]); |
| | 1326 | }), |
| | 1327 | ); |
| | 1328 | |
| 1258 | it.effect("resumes from projector last_applied_sequence without replaying older events", () => | 1329 | it.effect("resumes from projector last_applied_sequence without replaying older events", () => |
| 1259 | Effect.gen(function* () { | 1330 | Effect.gen(function* () { |
| 1260 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | 1331 | const projectionPipeline = yield* OrchestrationProjectionPipeline; |
Tradi3 <56069280+krutftw@users.noreply.github.com> 2026-08-27 17:58:24+08:00
GitHub <noreply@github.com> 2026-08-27 02:58:24-07:00