Amazon Kinesis Data Streams with Terraform: On-Demand, Encryption and Consumers

terraform {
  required_version = ">= 1.6"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0"
    }
  }
}

provider "aws" {
  region = "eu-central-1"
}

data "aws_caller_identity" "current" {}

variable "producer_role_name" {
  type        = string
  description = "Name of the IAM role that writes document events"
}

variable "consumer_role_name" {
  type        = string
  description = "Name of the IAM role that reads the stream"
}

data "aws_iam_role" "producer" {
  name = var.producer_role_name
}

data "aws_iam_role" "consumer" {
  name = var.consumer_role_name
}

resource "aws_kms_key" "documents" {
  description         = "Kinesis document events"
  enable_key_rotation = true
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid       = "AccountAdmin"
        Effect    = "Allow"
        Principal = { AWS = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:root" }
        Action    = "kms:*"
        Resource  = "*"
      },
      {
        Sid    = "StreamRoles"
        Effect = "Allow"
        Principal = {
          AWS = [data.aws_iam_role.producer.arn, data.aws_iam_role.consumer.arn]
        }
        Action = [
          "kms:Decrypt",
          "kms:DescribeKey",
          "kms:GenerateDataKey",
          "kms:GenerateDataKeyWithoutPlaintext",
        ]
        Resource = "*"
        Condition = {
          StringEquals = {
            "kms:ViaService" = "kinesis.eu-central-1.amazonaws.com"
          }
        }
      },
    ]
  })
}

resource "aws_kms_alias" "documents" {
  name          = "alias/kinesis-documents"
  target_key_id = aws_kms_key.documents.key_id
}

resource "aws_kinesis_stream" "documents" {
  name                     = "document-events"
  retention_period         = 168
  encryption_type          = "KMS"
  kms_key_id               = aws_kms_key.documents.arn
  max_record_size_in_kib   = 10240
  enforce_consumer_deletion = true
  shard_level_metrics = [
    "IncomingBytes",
    "OutgoingBytes",
    "IncomingRecords",
    "IteratorAgeMilliseconds",
  ]

  stream_mode_details {
    stream_mode = "ON_DEMAND"
  }
}

# Dedicated 2 MB/s read pipe for the archiver. Standard and provisioned
# modes allow up to 20 registered consumers on a stream.
resource "aws_kinesis_stream_consumer" "archiver" {
  name       = "archiver"
  stream_arn = aws_kinesis_stream.documents.arn
}

resource "aws_iam_role_policy" "producer_kinesis" {
  name = "document-events-producer"
  role = var.producer_role_name
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect = "Allow"
      Action = [
        "kinesis:PutRecord",
        "kinesis:PutRecords",
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
      ]
      Resource = aws_kinesis_stream.documents.arn
    }]
  })
}

resource "aws_iam_role_policy" "consumer_kinesis" {
  name = "document-events-consumer"
  role = var.consumer_role_name
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "kinesis:DescribeStream",
          "kinesis:DescribeStreamSummary",
          "kinesis:ListShards",
          "kinesis:GetShardIterator",
          "kinesis:GetRecords",
          "kinesis:SubscribeToShard",
          "kinesis:DescribeStreamConsumer",
        ]
        Resource = [
          aws_kinesis_stream.documents.arn,
          aws_kinesis_stream_consumer.archiver.arn,
        ]
      },
    ]
  })
}

resource "aws_cloudwatch_metric_alarm" "iterator_age" {
  alarm_name          = "document-events-iterator-age"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 3
  metric_name         = "GetRecords.IteratorAgeMilliseconds"
  namespace           = "AWS/Kinesis"
  period              = 60
  statistic           = "Maximum"
  threshold           = 60000
  treat_missing_data  = "notBreaching"
  dimensions = {
    StreamName = aws_kinesis_stream.documents.name
  }
}

// documents/documents.go
package documents

import (
	"context"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/pubsub"
)

type DocumentEvent struct {
	DocumentID string `json:"document_id" pubsub-attr:"document_id"`
	Kind       string `json:"kind"`
	Body       string `json:"body"`
}

// Messages with the same DocumentID arrive in publish order.
var Events = pubsub.NewTopic[*DocumentEvent]("document-events", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
	OrderingAttribute: "document_id",
})

var _ = pubsub.NewSubscription(Events, "archive", pubsub.SubscriptionConfig[*DocumentEvent]{
	Handler:     archive,
	AckDeadline: 30 * time.Second,
	RetryPolicy: &pubsub.RetryPolicy{MaxRetries: 8},
})

var _ = pubsub.NewSubscription(Events, "notify", pubsub.SubscriptionConfig[*DocumentEvent]{
	Handler:     notify,
	AckDeadline: 30 * time.Second,
	RetryPolicy: &pubsub.RetryPolicy{MaxRetries: 8},
})

type PublishParams struct {
	DocumentID string `json:"document_id"`
	Kind       string `json:"kind"`
	Body       string `json:"body"`
}

// Called by the editor service after a document save.
//
//encore:api private method=POST path=/documents/events
func Publish(ctx context.Context, p *PublishParams) error {
	if p.DocumentID == "" || p.Kind == "" {
		return &errs.Error{Code: errs.InvalidArgument, Message: "document_id and kind are required"}
	}
	_, err := Events.Publish(ctx, &DocumentEvent{
		DocumentID: p.DocumentID,
		Kind:       p.Kind,
		Body:       p.Body,
	})
	return err
}

func archive(ctx context.Context, ev *DocumentEvent) error {
	// Write the event to long-term storage. Return an error to retry
	// until MaxRetries, after which the message goes to the subscription DLQ.
	return nil
}

func notify(ctx context.Context, ev *DocumentEvent) error {
	return nil
}