recentlyinteracted

Example: recently-interacted Top-N (Kinesis + Kafka, one pipeline)

A Murmur pipeline that maintains a daily-windowed Top-N "recently interacted entities" sketch fed by two sources at once:

• Kinesis, consumed via an AWS Lambda Kinesis trigger (cmd/lambda). Operationally cheap path for managed analytics ingest (Segment-style event buses, AWS-native event sources). Lambda owns shard polling, fan-out, and partial-batch retry.
• Kafka (or MSK), consumed by a long-running ECS Fargate streaming worker (cmd/worker). Right fit for internal application events / CDC streams.

Both binaries share one pipeline.go definition and write to the same DynamoDB row. The Misra-Gries Combine collapses duplicates correctly across sources, so a single query against cmd/query returns the merged Top-N over all incoming interactions, regardless of which channel they came in on.

Why two sources?

Real systems frequently have one event channel for managed third-party data (Segment → Kinesis is the canonical case) and a separate channel for internal events (a Kafka topic). Murmur's structural-monoid pipeline DSL is execution-mode-agnostic: the same Aggregate(topk.New(K), windowed.Daily(...)) runs unchanged behind both drivers. Single source of truth, two ingest paths.

Schema

type Interaction struct {
    EntityID string  // ranked thing — product ID, content ID, etc.
    UserID   string  // actor (carried through; not part of the agg key)
    Source   string  // "kinesis" or "kafka" — diagnostics only
}

Pipeline (shared between all three binaries)

pipeline.NewPipeline[Interaction, []byte]("recently_interacted").
    Key(func(Interaction) string { return "global" }).
    Value(func(e Interaction) []byte {
        return topk.SingleN(K, e.EntityID, 1)
    }).
    Aggregate(topk.New(K), windowed.Daily(30*24*time.Hour)).
    StoreIn(dynamodb.NewBytesStore(...))

The Lambda binary leaves Source unset (Lambda owns polling). The Kafka worker calls AttachKafkaSource(pipe, cfg) to attach a franz-go source.

K is one number for the whole deployment: Config.ResolveK(), which is TOPK_K in each binary's environment and DefaultK (32) when unset. Both writers and the query server go through it. Sketches sized for different K refuse to merge, and the symptom is an empty or stale Top-N at query time rather than an error at startup — so never hard-code K anywhere.

Run locally

Stand up dependencies:

cd ../..
docker compose up -d kafka dynamodb-local

Create the DDB tables (first time only):

aws --endpoint-url=http://localhost:8000 dynamodb create-table \
    --table-name recently_interacted \
    --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=N \
    --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
    --billing-mode PAY_PER_REQUEST

aws --endpoint-url=http://localhost:8000 dynamodb create-table \
    --table-name recently_interacted_dedup \
    --attribute-definitions AttributeName=pk,AttributeType=S \
    --key-schema AttributeName=pk,KeyType=HASH \
    --billing-mode PAY_PER_REQUEST

Run the Kafka worker:

export DDB_ENDPOINT=http://localhost:8000
export DDB_DEDUP_TABLE=recently_interacted_dedup
go run ./examples/recently-interacted-topk/cmd/worker

The Lambda binary is meant for AWS — locally, you can drive it via sam local invoke with a synthetic Kinesis event payload, or run the test under ./test/... which exercises the same NewKinesisHandler against a fake DynamoDB.

Run the query server in a third terminal:

go run ./examples/recently-interacted-topk/cmd/query

Then query the merged Top-N:

# Last 7 days (merges 7 daily Misra-Gries summaries)
grpcurl -plaintext -d '{"entity":"global","duration_seconds":604800}' \
    localhost:50051 murmur.v1.QueryService/GetWindow

# Last 24 hours
grpcurl -plaintext -d '{"entity":"global","duration_seconds":86400}' \
    localhost:50051 murmur.v1.QueryService/GetWindow

cmd/query configures daily windowing, so Get / GetMany are not the right RPCs here: they read the all-time row at bucket 0, which the windowed writers never touch. The server returns FAILED_PRECONDITION for them rather than an empty result that looks like missing data. Run the pipeline without a window if you want an all-time Get.

The response's data is a serialized Misra-Gries summary (raw bytes); decode via pkg/monoid/sketch/topk.Decode, or use the embedded admin UI which renders the items + counts directly.

Deploy to real AWS

End-to-end Terraform composition lives at terraform/. It composes pipeline-counter (ECS worker + query + state DDB + dedup DDB) with pipeline-lambda-kinesis (Kinesis stream + Lambda + event-source mapping), plus CloudWatch alarms for the Lambda + DDB + Kinesis side. See terraform/README.md for the full runbook.

Production deployment notes

• The Lambda binary builds for provided.al2 / arm64 — see the comment block in cmd/lambda/main.go for the exact go build invocation.
• Configure the event-source mapping with FunctionResponseTypes=["ReportBatchItemFailures"] so failed records (after the in-handler retry budget) are isolated rather than redelivering the whole batch.
• The Kafka worker runs as a standard Murmur ECS Fargate service.
• Both processes use the same DDB_TABLE and DDB_DEDUP_TABLE. The dedup table is strongly recommended in production — Lambda BatchItemFailures may redeliver records adjacent to a failure, and Kafka rebalances may re-emit unacked records. Without dedup, a redelivery spuriously increments the Misra-Gries count.