@makerx/blob-data-streamer

An in-process blob-storage-backed streaming broker — at-least-once, ordered, single-producer-per-topic. Cheap to run, simple to deploy. Azure Blob today; provider-abstracted so S3 can follow.

What it is, in one paragraph

A TypeScript library that turns Azure Blob Storage (S3 to follow) into an in-process, at-least-once, ordered, CDC-style streaming bus with infinite durability and low running cost. The producer is your own process, embedding the library; it holds a blob lease that fences it as the single writer for a topic. Consumers are also embedded; consumer groups share a lease-protected offset blob to elect a single active reader per group. Postgres is an optional add-on — it carries LISTEN/NOTIFY wake-up signals when sub-second latency matters; otherwise consumers poll cheaply via HEAD requests. Sealed segments (older parts of the stream that are immutably archived) are block blobs in both JSONL (verbatim) and Parquet (analytical) variants, both verified via sha256 + row count before being published in the segment index — enabling efficient replay and analytics of the historical stream.

Install

npm install @makerx/blob-data-streamer @azure/storage-blob

@azure/storage-blob is a runtime dependency (the broker uses your BlobServiceClient so you control credentials — connection string, SAS, DefaultAzureCredential, managed identity, workload identity, etc.).

If you opt in to compaction (compaction.enabled: true on Producer.start), also install:

npm install @duckdb/node-api

DuckDB — via the @duckdb/node-api ("Node Neo") client, which ships prebuilt binaries with no node-gyp build (ADR-0021) — converts sealed JSONL segments to Parquet siblings. The library tolerates DuckDB being missing: compaction is off by default, and with compaction on, Parquet conversion degrades to JSONL-only if @duckdb/node-api is absent.

If you want sub-second consumer wake-up via Postgres LISTEN/NOTIFY, install pg and pg-listen. Otherwise consumers poll cheap HEAD requests against the active blob.

Producer quickstart

The producer is embedded in your process — there is no broker service. Producer.start() acquires the lease on current-N.jsonl for (tenant, topic); that lease is what fences this process as the single writer. If another producer holds the lease, start() rejects with LeaseAlreadyHeldError and your orchestrator (Kubernetes, systemd, ACA) decides whether to retry.

import { BlobServiceClient } from '@azure/storage-blob';
import { Producer, azureBlobProvider } from '@makerx/blob-data-streamer';

const serviceClient = BlobServiceClient.fromConnectionString(process.env.AZURE_STORAGE_CONNECTION_STRING!);
await serviceClient.getContainerClient('streams').createIfNotExists();

const storage = azureBlobProvider({ serviceClient, containerName: 'streams' });

const producer = await Producer.start({
  storage,
  tenant: 'acme',
  topic: 'orders',
});

// `send` resolves when the record is durable on Azure Blob (after the
// coalesced AppendBlock that includes it commits). `offset` is the
// broker-assigned position in the topic.
for (let i = 0; i < 5; i++) {
  const { offset } = await producer.send({
    id: `order-${i}`,                                  // your idempotency key
    body: JSON.stringify({ amount: 100 + i }),         // string, opaque to the broker
  });
  console.log(`durable at offset ${offset}`);
}

// Graceful shutdown — flushes any in-flight batch and releases the lease.
await producer.close();

record.id is yours — pick something stable per logical event so consumers can dedupe on it. record.body is an opaque string the broker round-trips; binary payloads should be base64-encoded by you.

Single producer per topic, permanently. This is enforced by the blob lease and is a hard architectural constraint. If you need multiple writers, write to multiple topics and fan-in at the consumer.

Optional Postgres NOTIFY signal layer — wire it when sub-second consumer wake-up matters. Without it, consumers poll cheap HEAD requests:

import { Pool } from 'pg';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

const producer = await Producer.start({
  storage,
  tenant: 'acme',
  topic: 'orders',
  postgres: { pool, leaderId: process.env.HOSTNAME }, // leaderId is diagnostic only
});

If you wire the NOTIFY layer, apply the schema migration once at startup (see applyMigrations).

Consumer quickstart

The consumer is also embedded in your process. Consumer.start() acquires the lease on .consumer_groups/<group>.offset; within a consumer group exactly one consumer is active at a time, and the others (standbys started elsewhere) block waiting on the lease. When the active consumer's process dies, the lease expires and a standby takes over from the last committed offset.

import { BlobServiceClient } from '@azure/storage-blob';
import { Consumer, azureBlobProvider } from '@makerx/blob-data-streamer';

const serviceClient = BlobServiceClient.fromConnectionString(process.env.AZURE_STORAGE_CONNECTION_STRING!);
const storage = azureBlobProvider({ serviceClient, containerName: 'streams' });

const consumer = await Consumer.start({
  storage,
  tenant: 'acme',
  topic: 'orders',
  consumerGroup: 'fulfilment',     // change this string to fan out a parallel projection
  handler: async (record) => {
    const order = JSON.parse(record.body);

    // 1. await EVERY downstream side-effect to durable completion BEFORE returning.
    //    The broker commits offsets after handler resolves; if you return early
    //    and crash, the offset advances past a record whose side-effect was lost.
    // 2. Be idempotent on record.id — up to `commitEveryN - 1` records can be
    //    re-dispatched per consumer restart. UPSERT, outbox, idempotency table,
    //    pick one.
    await upsertOrderInDb(record.id, order);
  },
});

// On shutdown — flushes pending offset commit + releases the lease.
process.on('SIGTERM', async () => {
  await consumer.close();
  process.exit(0);
});

At-least-once contract — the broker guarantees every produced record is dispatched to the handler at least once. Duplicates can happen on consumer restart, lease handover, or replay. Your handler MUST be idempotent on record.id. Full contract on the ConsumerHandler JSDoc in the source.

Reading the full stream externally — Consumer.getStreamManifest() returns the ordered list of files (sealed Parquet/JSONL segments + the live tail), each with a read SAS URL, so a BI tool or bulk job can read the whole stream without running a consumer. Lease-free; no consumer group required.

const manifest = await Consumer.getStreamManifest({ storage, tenant: 'acme', topic: 'orders' });
for (const f of manifest.files) {
  // f.format ('parquet' | 'jsonl'), f.url (read SAS or null), f.startOffset / f.endOffset, f.volatile (live tail may still grow)
  console.log(f.format, f.url);
}

Pass format: 'parquet-only' for an analytics-only view, or best-effort fromTimestamp / toTimestamp to narrow by producer emission time — the result is a file-level superset, so filter records by ts after reading. See ADR-0022.

Polling vs NOTIFY:

• Polling (default) — consumer issues a cheap HEAD on current-N.jsonl every pollIntervalMs (default 1 000 ms). Latency floor ≈ pollIntervalMs / 2. No Postgres needed.
• NOTIFY — wire Postgres + apply migrations + pass postgres.connection as either a connection string or a pg.ClientConfig. p50 latency drops to ~150 ms. Falls back to polling automatically if Postgres reconnects or NOTIFY is lost.

// Static credentials (local dev / containers):
const consumer = await Consumer.start({
  storage,
  tenant: 'acme',
  topic: 'orders',
  consumerGroup: 'fulfilment',
  postgres: { connection: process.env.DATABASE_URL! },
  handler,
});

// Managed identity (Azure AAD) — pass a ClientConfig with a token-refreshing
// password function. `pg-listen` (used internally) keeps a dedicated long-lived
// connection for LISTEN, so this is NOT a `pg.Pool`.
const credential = new DefaultAzureCredential();
const consumer = await Consumer.start({
  storage,
  tenant: 'acme',
  topic: 'orders',
  consumerGroup: 'fulfilment',
  postgres: {
    connection: {
      host: 'mydb.postgres.database.azure.com',
      user: 'mydb-app',
      database: 'mydb',
      password: async () =>
        (await credential.getToken('https://ossrdbms-aad.database.windows.net')).token,
      ssl: { rejectUnauthorized: true },
    },
  },
  handler,
});

We recommend you stay at a minimum on the broker's default pollIntervalMs (1 000 ms). Aggressive polling creates HEAD/range-read contention with producer writes — measured to inflate p99 by 5× at 4 k evt/s. Running costs also increase the more frequently you poll.

Applying the Postgres schema once

Only needed when you use the NOTIFY signal layer. Run once at startup (idempotent — subsequent calls no-op):

import { Pool } from 'pg';
import { applyMigrations } from '@makerx/blob-data-streamer';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
await applyMigrations({ pool });
// ... now safe to construct producers/consumers with `postgres: { pool }`.

For managed identity / AAD, build the pool with password: async () => token and pass that pool through — the library borrows a client and inherits the auth. See the applyMigrations JSDoc for the AAD example. schemaName defaults to streams; pass a different one to both applyMigrations and the postgres.schemaName option on Producer.start if you need to avoid a collision.

node-pg-migrate is an optional peer dependency — install it only if you call applyMigrations:

npm install node-pg-migrate

When to use it / when not

This library targets SMB-to-mid-rate workloads where Kafka is overkill or unaffordable. Real-Azure-measured envelope on a 2 vCPU / 4 GiB ACA Consumption replica in australiaeast against Standard_LRS storage:

| Band | Aggregate rate | sendToAck p50 / p99 | Polling e2e p99 | NOTIFY e2e p50 / p99 | | --- | --- | --- | --- | --- | | SMB | ≤ 500 evt/s | < 50 ms / < 500 ms | < 2 s | < 500 ms / < 2 s | | Mid-rate | 500 – 20 000 evt/s | < 50 ms / < 500 ms | < 2 s | n/a (polling wins above 4 k/s) | | Analytics-class | > 20 000 evt/s | < 500 ms / < 3 s | < 10 s | n/a |

Single 2 vCPU host caps at ~30–50 k evt/s aggregate (event-loop CPU bound, not Azure-throttling). Above that, scale out via separate containers per producer+consumer pair for the full envelope.

Use when:

• Aggregate < 20 k evt/s and you want Kafka-style semantics without cost and overhead of operating Kafka.
• You need single-producer-per-topic at the application layer.
• At-least-once + idempotent handlers fit your downstream model.
• Azure Blob is your durable substrate of choice (S3 backend is on the roadmap).

Do not use when:

• You need exactly-once cross-topic transactional semantics.
• You need multi-producer-per-topic fan-in (architecturally out of scope).
• You need GDPR record-level deletion of historic events (append-only by design — crypto-shredding is the recommended workaround, not yet shipped).
• You need a Kafka wire-protocol — this is a bespoke library, not a Kafka clone.

Production checklist

1. Build your BlobServiceClient with managed identity wherever possible — the broker never sees the credential, so token rotation and IAM policies live in your hands.
2. Pick a deployment shape based on your aggregate rate. Recommended starting point: SMB scale runs in-process; mid-rate N≥5 runs single-container role-split via child_process.spawn; above ~8 k/s/pair or > 20 k/s aggregate, scale out across containers.
3. Default pollIntervalMs (1 000 ms) is correct for almost every workload. Aggressive polling creates HEAD/range-read contention with producer writes. Leave it alone unless you need < 500 ms freshness and Postgres NOTIFY is not an option.
4. Opt in to compaction (Producer.start({ compaction: { enabled: true } })) when you want sealed JSONL + Parquet archive on Azure for replay and analytics. Off by default so SDK callers don't get a background setInterval worker they didn't ask for.
5. Make your handler idempotent on record.id. UPSERT, outbox, idempotency table — pick one. The broker re-dispatches up to commitEveryN - 1 records on consumer restart.
6. Wire OTel. initOtel() sets up traces + metrics via OTLP gRPC; the library emits producer.append_retry_total, consumer.dispatch.duration_ms, pending_seals_depth, lease_renewal_error_total, and more.

License

Copyright (c) 2026 MakerX Pty Ltd.

UNLICENSED — closed-source, all rights reserved. Distributed via public npm for convenience; see LICENSE. Licensing enquiries: [email protected].