Skip to content
effect-dynamodb

DynamoDB Streams

DynamoDB Streams deliver a time-ordered sequence of item-level changes. effect-dynamodb provides utilities for decoding stream records into typed domain objects.

Understanding the Problem

Section titled “Understanding the Problem”

DynamoDB Stream records contain items in DynamoDB’s native AttributeValue format:

{
  "pk": { "S": "$myapp#v1#employee#employeeid_emp-alice" },
  "sk": { "S": "$myapp#v1#employee" },
  "__edd_e__": { "S": "Employee" },
  "email": { "S": "alice@acme.com" },
  "version": { "N": "1" }
}

Manually decoding this into your domain types is error-prone. effect-dynamodb provides Entity.decodeMarshalledItem for AttributeValue input and Entity.itemSchema for already-unmarshalled input.

Images can also carry two other hidden attributes: __edd_i__, the incarnation token on items of a versioned entity, and __edd_d__, a string set naming unique fields that hold only a decoding default. Decoded models never include them. Entity.decodeMarshalledItem refuses, with a ValidationError, an image of a versioned entity that has __edd_i__ but no version, because its version was removed outside the library.

Setup

Section titled “Setup”

The examples on this page use these models and entities:

class Employee extends Schema.Class<Employee>("Employee")({
  employeeId: Schema.String,
  email: Schema.String,
  displayName: Schema.NonEmptyString,
  department: Schema.String,
}) {}


const EmployeeModel = DynamoModel.configure(Employee, {
  email: { immutable: true },
})


class Task extends Schema.Class<Task>("Task")({
  taskId: Schema.String,
  employeeId: Schema.String,
  title: Schema.NonEmptyString,
  status: Schema.String,
}) {}
const AppSchema = DynamoSchema.make({ name: "myapp", version: 1 })


const EmployeeEntity = Entity.make({
  model: EmployeeModel,
  entityType: "Employee",
  primaryKey: {
    pk: { field: "pk", composite: ["employeeId"] },
    sk: { field: "sk", composite: [] },
  },
  indexes: {
    byDepartment: {
      name: "gsi1",
      pk: { field: "gsi1pk", composite: ["department"] },
      sk: { field: "gsi1sk", composite: ["displayName"] },
    },
  },
  timestamps: true,
  versioned: true,
})


const TaskEntity = Entity.make({
  model: Task,
  entityType: "Task",
  primaryKey: {
    pk: { field: "pk", composite: ["taskId"] },
    sk: { field: "sk", composite: [] },
  },
  indexes: {
    byEmployee: {
      name: "gsi1",
      pk: { field: "gsi1pk", composite: ["employeeId"] },
      sk: { field: "gsi1sk", composite: ["status"] },
    },
  },
  timestamps: true,
})


const MainTable = Table.make({
  schema: AppSchema,
  entities: { EmployeeEntity, TaskEntity },
})

Decoding from AttributeValue Format

Section titled “Decoding from AttributeValue Format”

For Lambda functions consuming DynamoDB Stream events directly, use Entity.decodeMarshalledItem. It unmarshalls the AttributeValue map and decodes it as a table read does, in a single step: field: renames are reversed, sparse maps are rebuilt, and packed fields are decoded by their stored form. It returns an Effect that fails with ValidationError on schema mismatch, or when a compressed value’s codec isn’t registered (the error’s cause is then a CompressorNotRegistered):

const handleEmployeeStream = (event: DynamoDBStreamEvent) =>
  Effect.gen(function* () {
    for (const record of event.Records) {
      if (record.dynamodb?.NewImage) {
        const employee = yield* Entity.decodeMarshalledItem(
          EmployeeEntity,
          record.dynamodb.NewImage,
        )
        yield* Console.log(`Updated: ${(employee as Employee).displayName}`)
      }
    }
  })

Decoding from Unmarshalled Format

Section titled “Decoding from Unmarshalled Format”

If you’ve already unmarshalled the item (e.g., using Marshaller.fromAttributeMap or @aws-sdk/util-dynamodb), use Entity.itemSchema to get a Schema codec for direct decoding. It’s a pure schema: it doesn’t map field: renamed attributes back to their domain names, rebuild sparse maps or decode packed fields. For an entity that uses any of those, pass the unmarshalled item to Entity.decodeMarshalledItem instead, which accepts both forms:

const employeeFromItem = Entity.itemSchema(EmployeeEntity)


const decodeUnmarshalledEmployee = (marshalledImage: Record<string, AttributeValue>) => {
  const unmarshalled = Marshaller.fromAttributeMap(marshalledImage)
  return Schema.decodeUnknownEffect(employeeFromItem)(unmarshalled)
}

Typing the Marshalled Shape

Section titled “Typing the Marshalled Shape”

The Entity.Marshalled<E> type represents the shape of a DynamoDB item in AttributeValue format:

type EmployeeMarshalled = Entity.Marshalled<typeof EmployeeEntity>
// {
//   readonly [x: string]: {
//     readonly S?: string
//     readonly N?: string
//     readonly BOOL?: boolean
//     readonly NULL?: boolean
//     readonly L?: ReadonlyArray<unknown>
//     readonly M?: Record<string, unknown>
//     readonly B?: Uint8Array
//   }
// }

Processing Multiple Entity Types

Section titled “Processing Multiple Entity Types”

When processing streams from a single-table design, use __edd_e__ to discriminate entity types and decode each with Entity.decodeMarshalledItem:

const handleStreamEvent = (event: DynamoDBStreamEvent) =>
  Effect.gen(function* () {
    for (const record of event.Records) {
      const image = record.dynamodb?.NewImage
      if (!image) continue


      const entityType = image["__edd_e__"]?.S
      switch (entityType) {
        case "Employee": {
          const emp = yield* Entity.decodeMarshalledItem(EmployeeEntity, image)
          yield* handleEmployeeChange(emp as Entity.Record<typeof EmployeeEntity>)
          break
        }
        case "Task": {
          const task = yield* Entity.decodeMarshalledItem(TaskEntity, image)
          yield* handleTaskChange(task as Entity.Record<typeof TaskEntity>)
          break
        }
      }
    }
  })


const handleEmployeeChange = (emp: Entity.Record<typeof EmployeeEntity>) =>
  Console.log(`Employee changed: ${emp.displayName} (${emp.email})`)


const handleTaskChange = (task: Entity.Record<typeof TaskEntity>) =>
  Console.log(`Task changed: ${task.title} (${task.status})`)

Binary Values and Packed Storage

Section titled “Binary Values and Packed Storage”

Lambda triggers and EventBridge Pipes deliver stream records as DynamoDB JSON with B and BS values as base64 text, and @aws-sdk/util-dynamodb’s unmarshall leaves them as text. Entity.decodeMarshalledItem decodes them at any depth, so a packed field compressed to B decodes the same from a stream record as from a table read.

For an EventStore stream, stream.decodeStreamImage(image) returns exactly the StreamEvent the read paths return, or Option.none() for an image that isn’t one of the stream’s events:

// One handler for a table's NewImage records. decodeStreamImage returns
// Option.none() for anything that is not one of the stream's events.
const decodeImage = (image: EventStore.StreamImage) =>
  Effect.gen(function* () {
    const event = yield* Ledger.decodeStreamImage(image)
    if (Option.isSome(event)) {
      // Exactly the StreamEvent Ledger.read returns
      return `ledger ${event.value.version}: ${event.value.data._tag}`
    }
    const order = (yield* Entity.decodeMarshalledItem(Orders, image)) as Order
    return `order ${order.orderId}: ${order.lines.length} line(s)`
  })

Compressors come from a Compression service in context (for example Compression.layer([...]) merged into the application layer), as for table reads. See DynamoDB Streams images in the Packed Storage guide.

What’s Next?

Section titled “What’s Next?”