AWS.Firehose reference

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).

DeliveryStream: 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 },
});

DeliveryStream: 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`),
  })),
});

DeliveryStreamSink

Source: src/AWS/Firehose/DeliveryStreamSink.ts

A batching sink over Firehose PutRecordBatch (500 records / 4 MiB per call).

Each input element is a raw Firehose.Record ({ Data: Uint8Array }), so callers stay in control of encoding and record framing.

Even a 200 response can carry per-record failures (FailedPutCount > 0, per-record ErrorCode — ServiceUnavailableException / internal failure). All of them are transient, so the failed subset is re-submitted in input order on a bounded schedule; exhausting retries fails the sink with a typed BatchRetryExhaustedError carrying the stranded records.

DeliveryStreamSink: Streaming Records

// init — bind the sink (provide AWS.Firehose.DeliveryStreamSinkHttp on the Function)
const sink = yield* AWS.Firehose.DeliveryStreamSink(deliveryStream);

return {
  fetch: Effect.gen(function* () {
    // runtime — stream newline-framed records into Firehose
    yield* Stream.fromIterable(lines).pipe(
      Stream.map((line) => ({
        Data: new TextEncoder().encode(`${line}\n`),
      })),
      Stream.run(sink),
    );
    return HttpServerResponse.json({ ok: true });
  }),
};

ListDeliveryStreams

Source: src/AWS/Firehose/ListDeliveryStreams.ts

Runtime binding for firehose:ListDeliveryStreams.

An account-level binding — call it with no arguments to get a callable that lists delivery stream names in the region (paged via ExclusiveStartDeliveryStreamName + HasMoreDeliveryStreams). Provide the ListDeliveryStreamsHttp layer on the Function to satisfy the binding.

ListDeliveryStreams: Stream Metadata

const listDeliveryStreams = yield* AWS.Firehose.ListDeliveryStreams();

const response = yield* listDeliveryStreams();
const names = response.DeliveryStreamNames;

PutRecord

Source: src/AWS/Firehose/PutRecord.ts

Writes a single data record into a Firehose delivery stream.

Grants firehose:PutRecord on the bound delivery stream. The data blob can be up to 1,000 KiB; Firehose buffers records before delivering them to the destination, so use a delimiter (e.g. \n) to disambiguate records.

PutRecord: Putting Records

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

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

PutRecordBatch

Source: src/AWS/Firehose/PutRecordBatch.ts

Writes multiple data records into a Firehose delivery stream in a single call, achieving higher throughput per producer than single-record puts.

Grants firehose:PutRecordBatch on the bound delivery stream. Each request supports up to 500 records and 4 MiB total (limits are AWS-enforced, not client-enforced). Even a 200 response can carry per-record failures — check FailedPutCount and retry the failed entries from RequestResponses.

PutRecordBatch: Putting Records

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

// runtime
const response = yield* putRecordBatch({
  Records: lines.map((line) => ({
    Data: new TextEncoder().encode(`${line}\n`),
  })),
});
if (response.FailedPutCount > 0) {
  // retry entries whose RequestResponses[i].ErrorCode is set
}