KinesisMatcher

KinesisMatcher routes Amazon Kinesis Data Streams records to handlers, one record at a time.

A single event can carry many records from one or more streams. The matcher base64 decodes each record's data, parses it as JSON, then works out which of your routes should handle it. Records run in the order they arrive rather than in parallel, because a shard is ordered.

A record written with KPL aggregation holds many messages. The matcher unpacks it, so each message reaches your handler on its own. See Aggregated records.

Install

bash

npm install @latcher/base @latcher/kinesis

@latcher/base is a peer dependency, so install it yourself. It exports Latcher, which every matcher plugs into.

Create the matcher

import { createKinesisMatcher } from '@latcher/kinesis'
import { logInvocation } from './middleware/logInvocation'

const kinesisMatcher = createKinesisMatcher({
  batchItemFailures: true,  // Optional
  middleware: [logInvocation],  // Optional
})

Both options can be left out. createKinesisMatcher() on its own gives you a matcher that fails the whole batch when any record throws.

Options

Option	Type	Required	Default	Description
`batchItemFailures`	`boolean`	No	`false`	Report failed records back to Lambda instead of failing the whole batch. See Failures and retries
`middleware`	`KinesisMiddleware[]`	No	`[]`	Runs for every record this matcher handles, before any route middleware. See Middleware

Register routes

kinesisMatcher.route({
  filters: {
    eventSourceArn: TELEMETRY_STREAM_ARN,
    partitionKey: 'sensor-42',
  },
  dataSchema: ReadingSchema,  // Optional
  middleware: [withDeviceContext],  // Optional
  handler: onReading,
})

filters and handler are the only required keys.

route() returns the matcher, so you can chain registrations.

kinesisMatcher.route(readingRoute).route(statusRoute)

Routes are ranked by how specific they are. Where two overlap, the one matching only a subset of the other is tried first, whatever order you registered them in. Registration order decides the rest, so give each route filters no other route can match. See match order for how ranking works and what it cannot settle.

A record that matches no route throws. With batchItemFailures off that fails the entire batch, including records that would have succeeded. Register a catch-all route filtering only on eventSourceArn if you would rather swallow unknown records, and see nothing matched for what the other matchers do instead.

Filters

Every filter key on one route, showing each form a value can take. All of them are optional, so set the ones that pick out the records you want and leave the rest off.

kinesisMatcher.route({
  filters: {
    eventSourceArn: [TELEMETRY_STREAM_ARN, STATUS_STREAM_ARN],
    partitionKey: 'sensor-42', // Or a pattern: /^sensor-/
    custom: ({ data }) => {
      // Only a custom reaches the decoded data
      if (!isObject(data) || typeof data.celsius !== 'number') return false

      return data.celsius >= ALERT_THRESHOLD
    },
  },
  handler: onReading,
})

Filter	Type	Description
`eventSourceArn`	`FilterStringMatcher`	Matches against the record's stream ARN
`partitionKey`	`FilterStringMatcher`	Matches the partition key the producer wrote the record under
`custom`	`(input: KinesisFilterInput) => boolean \| Promise<boolean>`	Anything the other filters cannot express. Can be async

FilterStringMatcher is string | RegExp | Array<string | RegExp>. See filters for how each form matches, including the * wildcard.

A partition key is whatever the producer chose, so it only tells routes apart if your producers agree on a convention. Prefixing by record type gives you sensor-42 and gateway-7 to filter on, and a route per prefix.

custom sees the data before any schema has run, so narrow it with isObject from @latcher/base rather than reading straight into it. See custom for where it sits in the filter order.

Handler

Handlers take one argument and return nothing.

import { logger } from '@latcher/base'
import type { KinesisRequest, KinesisResponse } from '@latcher/kinesis'

export async function onReading(request: KinesisRequest<Reading>): Promise<KinesisResponse> {
  const { deviceId, celsius } = request.data
  logger.info(`Device ${deviceId} reported ${celsius}C at ${request.sequenceNumber}`)
}

Request object

Field	Type	Description
`data`	`TData`	The record's data, base64 decoded, read as UTF-8 text and JSON parsed. If it is not valid JSON you get the text
`rawData`	`Buffer`	The record's bytes, base64 decoded and left alone
`partitionKey`	`string`	The partition key the producer wrote the record under, which is what Kinesis shards on
`sequenceNumber`	`string`	Where the record sits in its shard. Unique per shard and increasing, so it makes a good idempotency key. Messages from one aggregated record share it
`subSequenceNumber`	`number \| undefined`	Where the message sits inside an aggregated record, counting from 0. `undefined` for a plain record
`approximateArrivalTimestamp`	`number`	When Kinesis accepted the record, as a Unix timestamp in seconds. It carries a fraction, so a `Date` needs `approximateArrivalTimestamp * 1000`
`record`	`KinesisStreamRecord`	The untouched record from AWS, for `eventID`, `invokeIdentityArn` and anything else you need
`context`	`Context`	The Lambda context

A payload that is not UTF-8 text does not survive data. Reading bytes as text swaps anything it cannot read for a replacement character, so gzip, Avro and protobuf come out mangled. Read rawData for those and decode it yourself.

KinesisStreamRecord and Context come from aws-lambda, not from this package.

Response type

KinesisResponse is undefined. There is nothing useful to return from a stream record, so handlers return Promise<KinesisResponse> and the matcher works out what to hand back to Lambda.

Throwing is how you signal failure. See Failures and retries for what that does to the rest of the batch.

Inferred handlers

Nothing to look up and nothing to keep in sync. defineRoute reads the schema and hands your handler a fully typed data, defaults and coercion included, so celsius below is a number and calibrated a boolean without you declaring either.

import { logger } from '@latcher/base'
import { defineRoute } from '@latcher/kinesis'
import { z } from 'zod'

const ReadingSchema = z.object({
  deviceId: z.string(),
  celsius: z.coerce.number(),
  calibrated: z.coerce.boolean().default(false),
})

export const readingRoute = defineRoute({
  filters: { eventSourceArn: TELEMETRY_STREAM_ARN },
  dataSchema: ReadingSchema,
}).handle(async ({ data, partitionKey }) => {
  if (!data.calibrated) return
  logger.info(`Device ${data.deviceId} on ${partitionKey} reported ${data.celsius}C`)
})

kinesisMatcher.route(readingRoute)

Inference pays off most in a Lambda taking several event sources, since you never have to know any of their request shapes. See inferred handlers, where the same queue is written both ways to compare.

Annotated handlers

Annotating the request yourself splits route setup from business logic, using KinesisRequest and your own types.

// handlers/onReading.ts
import { logger } from '@latcher/base'
import type { KinesisRequest, KinesisResponse } from '@latcher/kinesis'
import { z } from 'zod'

export const ReadingSchema = z.object({ deviceId: z.string(), celsius: z.number() })
type Reading = z.infer<typeof ReadingSchema>

export async function onReading(request: KinesisRequest<Reading>): Promise<KinesisResponse> {
  logger.info(`Device ${request.data.deviceId} reported ${request.data.celsius}C`)
}

// kinesis.ts
import { createKinesisMatcher } from '@latcher/kinesis'
import { onReading, ReadingSchema } from './handlers/onReading'

const kinesisMatcher = createKinesisMatcher()

kinesisMatcher.route({
  filters: { eventSourceArn: TELEMETRY_STREAM_ARN },
  dataSchema: ReadingSchema,
  handler: onReading,
})

Derive the type from the schema with z.infer rather than hand-writing an interface that mirrors it. Exporting the schema from the handler file and attaching it to the route in the matcher file keeps the type and the validation from drifting apart. See annotated handlers for the worked version.

Schema validation

dataSchema is the only key that takes a schema, and it is optional.

const ReadingSchema = z.object({
  deviceId: z.string(),
  celsius: z.coerce.number(),
  recordedAt: z.coerce.date(),
})

kinesisMatcher.route({
  filters: { eventSourceArn: TELEMETRY_STREAM_ARN },
  dataSchema: ReadingSchema,
  handler: onReading,
})

Key	Validates
`dataSchema`	The decoded, parsed record data

Any Standard Schema library works. Validation runs after a route has matched, so a record failing its schema throws rather than falling through to the next route. See schema validation for what your handler receives after coercion.

An object schema fails every record whose data is not JSON. A producer writing CSV lines or plain text gets them through as the decoded string, which a z.object() rejects. Validate those with z.string() and parse in the handler, or leave the schema off.

Failures and retries

With batchItemFailures off, which is the default, records run one at a time in the order they arrive and the first throw fails the invocation. Lambda retries the whole batch, and by default keeps retrying that shard until the records succeed or age out, so a record that always throws holds up everything behind it. MaximumRetryAttempts and a failure destination on the event source mapping are how you bound that.

Turn it on and the matcher reports the failure back to Lambda instead.

const kinesisMatcher = createKinesisMatcher({ batchItemFailures: true })

Reporting one failure reports every record after it too. A shard is ordered, so the matcher stops at the first throw and returns that record's sequenceNumber along with the sequenceNumber of every record left in the batch. Lambda redelivers from the failed record onwards.

You also need to set the ReportBatchItemFailures response type on the event source mapping. Without it, AWS ignores what the matcher returns.

Aggregated records

A producer using the Kinesis Producer Library (KPL) can pack many messages into one Kinesis record. Lambda hands that record over whole, so the matcher unpacks it for you. There is nothing to turn on.

Each message gets its own data, rawData and partitionKey. Filters, schemas and middleware run once per message, the same as for a plain record.

Messages from one aggregated record share a sequenceNumber. Use sequenceNumber and subSequenceNumber together as the idempotency key. record is the aggregated record for every message in it.

A throw on one message retries the whole aggregated record. Lambda can only report the aggregated record as failed, not a message inside it. This means the messages before the one that threw run again on the retry.

A record that looks aggregated but fails its checksum, or cannot be read, is routed as a plain record.

Middleware

Matcher and route middleware are both typed KinesisMiddleware, and the chain runs once per record, so a batch of ten records runs it ten times.

import { logger } from '@latcher/base'
import type { KinesisMiddleware } from '@latcher/kinesis'

export const logInvocation: KinesisMiddleware = async (request, next) => {
  logger.info(`Handling record ${request.record.eventID} on ${request.partitionKey}`)
  return next(request)
}

export const withDeviceContext: KinesisMiddleware<Reading> = async (request, next) => {
  logger.info(`Handling device ${request.data.deviceId}`)
  return next(request)
}

const kinesisMatcher = createKinesisMatcher({ middleware: [logInvocation] })

kinesisMatcher.route({
  filters: { eventSourceArn: TELEMETRY_STREAM_ARN },
  dataSchema: ReadingSchema,
  middleware: [withDeviceContext],
  handler: onReading,
})

Route middleware carries the route's data type. A route with a dataSchema needs KinesisMiddleware<Reading>. KinesisMiddleware on its own does not compile there. Matcher middleware takes no type argument, because it runs for every route.

Records are handled one at a time, so appendKeys on the shared logger cannot interleave the way it does on SQS. Keys are cleared per invocation rather than per record, so one you set for a record stays on every record after it in the same batch. See middleware for the execution order and the three levels it attaches at.

Types

All exported from @latcher/kinesis.

Type	Description
`KinesisRequest<TData>`	The handler argument
`KinesisResponse`	Handler return type, `undefined`
`KinesisFilters`	The `filters` object
`KinesisFilterInput`	What `custom` receives
`KinesisRouteDefinition<TData>`	A full route passed to `route()`
`KinesisMatcherOptions`	Options for `createKinesisMatcher`
`KinesisMiddleware<TData>`	Matcher and route middleware

The KinesisMatcher class and the createKinesisMatcher and defineRoute functions come from the same place.

Generic parameters

The three types above that take a parameter take the same one.

Parameter	Types	Default
`TData`	`request.data`	`unknown`

Leave it off and data is unknown, so KinesisRequest on its own is the right annotation for middleware that only reads the partition key or the raw record.

You only need this for annotated handlers. Inference covers it.

Code example

A telemetry stream and a device status stream feeding one Lambda, with a handler for each.

Open a file: index.ts | Kinesis matcher | handlers | schemas

TSkinesis.ts

A record comes from one stream, so the two eventSourceArn filters cannot both match and the order you register them in makes no difference. Put both record shapes on a single stream and it is partitionKey doing that job instead, which needs a prefix per shape.

onReading returns early on a normal reading. A custom comparing celsius does the same job in the matcher, which is worth it when a route is expensive to enter.

index.ts hands the matcher to Latcher, which is what AWS invokes and what every matcher in the Lambda gets registered on. See matchers for how the two levels of matching fit together.

KinesisMatcher ​

Install ​

Create the matcher ​

Options ​

Register routes ​

Filters ​

Handler ​

Request object ​

Response type ​

Inferred handlers ​

Annotated handlers ​

Schema validation ​

Failures and retries ​

Aggregated records ​

Middleware ​

Types ​

Generic parameters ​

Code example ​