KinesisStream

import { App } from '@anvil-cloud/sdk';
import * as anvil from '@anvil-cloud/sdk';

export default new App({
  run(ctx) {
    const indexer = new anvil.aws.Lambda('indexer', {
      runtime: 'nodejs24.x',
      handler: 'functions/indexer/index.handler',
    });

    const events = new anvil.aws.KinesisStream('events', {
      consumers: [
        {
          consumerName: 'indexer',
          lambda: { arn: indexer.arn, lambdaRoleArn: indexer.roleArn },
          grantRead: true,
          filter: { tables: ['orders', 'app.customers'] },
        },
      ],
    });

    ctx.export('streamArn', events.arn);
  },
});

import anvil_cloud as anvil

def infra(ctx: anvil.Context):
    indexer = anvil.aws.Lambda("indexer",
        runtime="nodejs24.x",
        handler="functions/indexer/index.handler",
    )

    events = anvil.aws.KinesisStream("events",
        consumers=[{
            "consumer_name": "indexer",
            # "lambda_" (trailing underscore) because "lambda" is a Python keyword.
            "lambda_": {"arn": indexer.arn, "lambda_role_arn": indexer.role_arn},
            "grant_read": True,
            "filter": {"tables": ["orders", "app.customers"]},
        }],
    )

    ctx.export("stream_arn", events.arn)

anvil.App(run=infra)

package main

import (
    "github.com/anvil-cloud/anvil/sdk/go/anvil"
    anvilaws "github.com/anvil-cloud/anvil/sdk/go/anvil/aws"
    "github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
    anvil.Run(anvil.AppConfig{
        Run: func(ctx *anvil.Context) error {
            indexer, err := anvilaws.NewLambda(ctx.PulumiCtx(), "indexer", &anvilaws.LambdaArgs{
                Runtime: anvilaws.LambdaRuntime_Nodejs24_x,
                Handler: pulumi.String("functions/indexer/index.handler"),
            })
            if err != nil {
                return err
            }

            events, err := anvilaws.NewKinesisStream(ctx.PulumiCtx(), "events", &anvilaws.KinesisStreamArgs{
                Consumers: anvilaws.KinesisConsumerArgsArray{
                    &anvilaws.KinesisConsumerArgsArgs{
                        ConsumerName: "indexer",
                        Lambda: &anvilaws.KinesisLambdaConsumerArgsArgs{
                            Arn:           indexer.Arn,
                            LambdaRoleArn: indexer.RoleArn,
                        },
                        GrantRead: pulumi.Bool(true),
                        Filter: &anvilaws.KinesisConsumerFilterArgsArgs{
                            Tables: pulumi.StringArray{pulumi.String("orders"), pulumi.String("app.customers")},
                        },
                    },
                },
            })
            if err != nil {
                return err
            }

            ctx.Export("streamArn", events.Arn)
            return nil
        },
    })
}