diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts index 647b7e0daa..ff4cc4ffaa 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts @@ -264,6 +264,38 @@ describe("HyperframesRenderStack — snapshot", () => { expect(v2).not.toContain("PlanS3Uri"); }); + it("sends the same payload from every Lambda task as the SAM template", () => { + // CDK drops null-valued payload fields when it renders the definition (#5259), so compare + // what each deployed task sends rather than the construct source. + const cdkTasks = lambdaTasks(SYNTHED.definition.States); + const samTasks = lambdaTasks(readSamDefinition().States); + expect(Object.keys(cdkTasks).sort()).toEqual(Object.keys(samTasks).sort()); + expect(Object.keys(cdkTasks).length).toBeGreaterThanOrEqual(6); + for (const [taskName, cdkTask] of Object.entries(cdkTasks)) { + expect({ taskName, payload: sentPayload(cdkTask) }).toEqual({ + taskName, + payload: sentPayload(samTasks[taskName]), + }); + } + }); + + it("keeps same-named Lambda tasks in different branches apart", () => { + const task = (format: string) => ({ + Type: "Task", + Resource: "arn:aws:states:::lambda:invoke", + Parameters: { Payload: { Action: "renderChunk", Format: format } }, + }); + const branch = (format: string) => ({ + StartAt: "RenderChunk", + States: { RenderChunk: task(format) }, + }); + const tasks = lambdaTasks({ + Fan: { Type: "Parallel", Branches: [branch("mp4"), branch("webm")] }, + }); + expect(Object.keys(tasks).sort()).toEqual(["Fan/0/RenderChunk", "Fan/1/RenderChunk"]); + expect(sentPayload(tasks["Fan/1/RenderChunk"]).Format).toBe("webm"); + }); + it("materializes null audio in the synthesized CDK v2 assembly payload", () => { const state = requireRecord(SYNTHED.definition.States.AssembleV2, "AssembleV2 state"); const parameters = requireRecordProperty(state, "Parameters", "AssembleV2 parameters"); @@ -318,6 +350,59 @@ function requireRecordProperty( return requireRecord(record[property], label); } +const isLambdaTask = (state: Record): boolean => + state.Type === "Task" && String(state.Resource).endsWith(":states:::lambda:invoke"); + +/** The state maps nested in a Map processor or Parallel branches, keyed by their path. */ +function nestedStates(name: string, state: Record): [string, unknown][] { + const processor = isRecord(state.Iterator) ? state.Iterator : state.ItemProcessor; + const branches: unknown[] = Array.isArray(state.Branches) ? state.Branches : []; + const nested: [string, unknown][] = [ + [`${name}/`, isRecord(processor) && processor.States], + ...branches.map((branch, i): [string, unknown] => [ + `${name}/${i}/`, + isRecord(branch) && branch.States, + ]), + ]; + return nested.filter(([, states]) => isRecord(states)); +} + +/** Every Lambda-invoking task by its full state path, so same-named tasks in two branches stay apart. */ +function lambdaTasks(states: Record, path = ""): Record { + const records = Object.entries(states).filter( + (entry): entry is [string, Record] => isRecord(entry[1]), + ); + return Object.assign( + Object.fromEntries( + records + .filter(([, state]) => isLambdaTask(state)) + .map(([name, state]) => [path + name, state]), + ), + ...records.flatMap(([name, state]) => + nestedStates(path + name, state).map(([nestedPath, nested]) => + lambdaTasks(nested as Record, nestedPath), + ), + ), + ); +} + +/** A task's payload as Step Functions sends it: a `States.StringToJson('null')` field is a null. */ +function sentPayload(state: unknown): Record { + const parameters = requireRecordProperty( + requireRecord(state, "task state"), + "Parameters", + "task parameters", + ); + const payload = requireRecordProperty(parameters, "Payload", "task payload"); + return Object.fromEntries( + Object.entries(payload).map(([key, value]) => + key.endsWith(".$") && value === "States.StringToJson('null')" + ? [key.slice(0, -2), null] + : [key, value], + ), + ); +} + function getV2TaskStates(definition: { States: Record; }): Record<"PlanV2" | "RenderChunkV2" | "AssembleV2", unknown> {