DeliveryStream

Source: src/AWS/Firehose/DeliveryStream.ts

An Amazon Data Firehose delivery stream that buffers records and delivers them to an S3 bucket.

DeliveryStream owns the stream’s lifecycle and mutable destination configuration (buffering hints, compression, prefixes, tags). The stream is DirectPut by default — producers write with PutRecord / PutRecordBatch — or it can drain an existing Kinesis Data Stream via the source prop. Unless you supply role ARNs, an IAM role is auto-created granting Firehose write access to the destination bucket (and read access to the source stream when one is configured).

Creating Delivery Streams

DirectPut stream delivering to S3

import * as AWS from "alchemy/AWS";

const bucket = yield* AWS.S3.Bucket("DataLake");
const stream = yield* AWS.Firehose.DeliveryStream("Events", {
  destination: {
    bucketArn: bucket.bucketArn,
  },
});

Tuned buffering and compression

const stream = yield* AWS.Firehose.DeliveryStream("Events", {
  destination: {
    bucketArn: bucket.bucketArn,
    prefix: "events/",
    errorOutputPrefix: "errors/",
    bufferingInterval: "1 minute",
    bufferingSizeInMBs: 1,
    compressionFormat: "GZIP",
  },
});

Server-side encryption at rest

const stream = yield* AWS.Firehose.DeliveryStream("Events", {
  destination: { bucketArn: bucket.bucketArn },
  encryption: { keyType: "AWS_OWNED_CMK" },
});

Kinesis Data Stream as source

const source = yield* AWS.Kinesis.Stream("Clickstream");
const stream = yield* AWS.Firehose.DeliveryStream("ClickstreamArchive", {
  source: { kinesisStreamArn: source.streamArn },
  destination: { bucketArn: bucket.bucketArn },
});

Runtime Producers

Bind producer operations in the init phase and use them in runtime handlers. Records are buffered by Firehose and appear in S3 after the buffering interval elapses.

Put a record from a handler

// init
const putRecord = yield* AWS.Firehose.PutRecord(stream);

return {
  fetch: Effect.gen(function* () {
    // runtime
    const response = yield* putRecord({
      Record: { Data: new TextEncoder().encode("hello\n") },
    });
    return HttpServerResponse.json({ recordId: response.RecordId });
  }),
};

Put a batch of records

// init
const putRecordBatch = yield* AWS.Firehose.PutRecordBatch(stream);

// runtime
const response = yield* putRecordBatch({
  Records: lines.map((line) => ({
    Data: new TextEncoder().encode(`${line}\n`),
  })),
});