TableSink

Source: src/AWS/DynamoDB/TableSink.ts

A batching sink over DynamoDB BatchWriteItem (25 write requests / 16 MB per call). Entries the API echoes back in UnprocessedItems (throttling, internal errors) are re-submitted on a bounded schedule; exhausting retries fails the sink with a typed BatchRetryExhaustedError carrying the stranded entries.

Sinks are request-scoped consumers: acquire the sink during the Function’s init, drive it with Stream.run inside a handler, and let it drain fully before the handler returns.

Streaming Writes

Stream Put Requests into a Table

const sink = yield* AWS.DynamoDB.TableSink(table);

yield* Stream.fromIterable(records).pipe(
  Stream.map((record): AWS.DynamoDB.TableSinkEntry => ({
    PutRequest: {
      Item: {
        pk: { S: record.pk },
        sk: { S: record.sk },
      },
    },
  })),
  Stream.run(sink),
);

Stream Delete Requests into a Table

yield* Stream.fromIterable(keys).pipe(
  Stream.map((key): AWS.DynamoDB.TableSinkEntry => ({
    DeleteRequest: {
      Key: {
        pk: { S: key.pk },
        sk: { S: key.sk },
      },
    },
  })),
  Stream.run(sink),
);