Distributing Video Ingest Events Between Services With NATS JetStream

The gap between ingest and everything downstream

Our ingest worker pulls trending video metadata from a handful of upstream APIs, normalizes it, and writes rows into SQLite. That part has never been the problem. The problem is what has to happen after the row lands: a thumbnail gets fetched and re-encoded into three WebP sizes, the FTS5 search index gets rebuilt for that row, per-category counters get recomputed, and the Cloudflare edge cache for the affected watch and category pages gets purged.

For about a year, that fan-out was cron. Four downstream jobs, four schedules, each one polling SQLite for work it hadn't done yet — SELECT id FROM videos WHERE thumb_done = 0 ORDER BY id LIMIT 200. Every consumer owned a boolean column on the videos table. It worked, in the sense that things eventually happened.

Then it stopped working well enough to ignore:

• A latency floor equal to cron granularity. A video ingested at 02:01 got its thumbnail at 02:15 and its search index entry at 02:30. Users hitting a fresh category page saw placeholder tiles.
• Writer-lock contention. SQLite has exactly one writer. Four pollers issuing UPDATE videos SET thumb_done = 1 WHERE id = ? in tight loops fought each other and the ingest worker. WAL mode and a 30s busy_timeout papered over it; they didn't remove it.
• No retry semantics. A job that died halfway through a batch left rows in a state that was neither done nor claimed. We wrote reconciliation scripts. We wrote reconciliation scripts for the reconciliation scripts.
• Adding a consumer meant a schema migration. A fifth downstream task meant a fifth column, a migration across four deployed sites, and a fifth crontab entry.
• No replay. "Re-render every thumbnail ingested last Tuesday" was a bespoke one-off script every single time.

The fix wasn't a smarter cron. It was to stop using the database as a queue and put an actual log in the middle.

Why JetStream and not the usual suspects

I evaluated four options against one hard constraint: whatever we ran had to fit alongside a PHP 8.4 app on modest boxes without becoming its own operational specialty.

• Kafka is the correct answer at a scale we are not at. Even with KRaft removing ZooKeeper, you are running a JVM, tuning heap, and thinking about partition counts before you have written a single consumer. Our peak is on the order of tens of thousands of events per day.
• Redis Streams was tempting because Redis is easy. But consumer groups there put recovery of stuck entries on you — XAUTOCLAIM loops, idle-time thresholds, your own dead-letter logic. That's a queue you assemble rather than a queue you configure.
• RabbitMQ has excellent routing and terrible replay. Once a message is acked it's gone. Half of what I wanted was the ability to rewind.
• NATS with JetStream is a single ~20MB Go binary. Streams are file-backed, consumers have server-side cursors, redelivery and backoff are consumer config rather than application code, and message deduplication is a header. Three nodes gives you R3 replication with no external coordination service.

The deciding factor was honestly the second-order cost. NATS has one config file and a genuinely good CLI. When something goes wrong at 3am I want nats stream report, not a Grafana dashboard I have to build first.

Subject design is the part you can't undo cheaply

Subjects are the routing layer, and unlike the payload schema, they're painful to change once consumers depend on them. We landed on this hierarchy:

video.ingested.<source>.<external_id>
video.updated.<source>.<external_id>
video.removed.<source>.<external_id>
video.enriched.thumbnail.<external_id>
video.enriched.transcript.<external_id>

The rules we settled on after one bad first attempt:

• Subjects name facts, not commands. video.ingested, not video.make_thumbnail. A command subject bakes one consumer's job into the producer's vocabulary; when we added the Cloudflare purge consumer, it subscribed to the same fact with zero producer changes.
• Routing dimensions go in the subject, data goes in the body. source is in the subject because a consumer might legitimately want only YouTube events. title is not, because nobody filters on it.
• Never put something in a subject you can't filter on usefully. Subject tokens can't be range-queried. A timestamp token is dead weight.
• Put the high-cardinality token last so video.ingested.youtube.> is a useful filter and video.ingested.*.* still matches everything.
• One stream, many subjects. Our first attempt used a stream per event type. That multiplied retention config, monitoring surface, and the number of things that can silently fill a disk by five, and bought us nothing.

The stream and one of its consumers, created with the CLI (these commands are idempotent enough to live in a provisioning script):

# One stream captures the whole video.* namespace.
nats stream add VIDEO_EVENTS \
  --subjects "video.>" \
  --storage file \
  --retention limits \
  --discard old \
  --max-age 168h \
  --max-bytes 4GB \
  --max-msg-size 64KB \
  --dupe-window 30m \
  --replicas 3 \
  --defaults

# Each downstream service gets its own durable consumer with its own cursor.
nats consumer add VIDEO_EVENTS thumbnailer \
  --filter "video.ingested.>" \
  --ack explicit \
  --deliver all \
  --max-deliver 5 \
  --wait 60s \
  --max-pending 32 \
  --defaults

nats consumer add VIDEO_EVENTS search-indexer \
  --filter "video.>" \
  --ack explicit \
  --max-deliver 8 \
  --wait 30s \
  --max-pending 500 \
  --defaults

Note --retention limits, not workqueue. WorkQueue retention deletes a message once a consumer acks it, which is exactly wrong when four independent services each need to see every event. Limits retention keeps messages until age or size says otherwise, and each durable consumer tracks its own position. This distinction cost me an afternoon early on when the indexer mysteriously saw only a third of events — the thumbnailer was eating them.

Publishing from PHP without slowing down requests

Our publisher lives in the CLI ingest worker, not in the web request path. That's deliberate. A JetStream publish waits for a PubAck from the server, which is a network round-trip. Behind LiteSpeed, on a page that's supposed to be served from cache anyway, adding a synchronous round-trip to a request handler is how you turn a 40ms response into a 90ms response for no user-visible benefit. If you genuinely must emit from a request, use core NATS fire-and-forget (no ack, no persistence) or write to a local spool the worker drains.

This uses basis-company/nats, which is the PHP client that has actually kept up with JetStream:

<?php
declare(strict_types=1);

namespace DailyWatch\Events;

use Basis\Nats\Client;
use Basis\Nats\Configuration;
use Basis\Nats\Message\Payload;

final class VideoEventPublisher
{
    private const SCHEMA = 1;
    private const STREAM = 'VIDEO_EVENTS';

    public function __construct(private readonly Client $client) {}

    public static function fromEnv(): self
    {
        return new self(new Client(new Configuration([
            'host'    => getenv('NATS_HOST') ?: '127.0.0.1',
            'port'    => (int) (getenv('NATS_PORT') ?: 4222),
            'user'    => getenv('NATS_USER') ?: null,
            'pass'    => getenv('NATS_PASS') ?: null,
            'timeout' => 2.0,
        ])));
    }

    /** @param array{id:int,external_id:string,source:string,revision:int,title:string,duration:int,thumb_url:string} $video */
    public function ingested(array $video): void
    {
        $subject = sprintf('video.ingested.%s.%s', $video['source'], $video['external_id']);

        $body = json_encode([
            'schema'      => self::SCHEMA,
            'video_id'    => $video['id'],
            'external_id' => $video['external_id'],
            'source'      => $video['source'],
            'revision'    => $video['revision'],
            'title'       => $video['title'],
            'duration'    => $video['duration'],
            'thumb_url'   => $video['thumb_url'],
            'occurred_at' => gmdate('c'),
        ], JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES);

        // Server-side dedupe: the same video at the same revision published
        // twice inside the stream's 30m duplicate window is collapsed. This is
        // what makes an ingest-worker crash-and-rerun safe.
        $msgId = hash('sha256', implode('|', [
            $video['source'], $video['external_id'], (string) $video['revision'],
        ]));

        $this->client->getApi()
            ->getStream(self::STREAM)
            ->put($subject, new Payload($body, ['Nats-Msg-Id' => $msgId]));
    }
}

Two things carry most of the weight here.

The revision field is a monotonically increasing integer we bump on every upstream metadata change. It is the thing that makes out-of-order delivery survivable, and I'll come back to it.

The Nats-Msg-Id header is the cheapest reliability win in the whole system. Our ingest worker is not transactional across "write row" and "publish event" — it can crash between them, and on rerun it republishes. With a stable message ID and a 30-minute duplicate window, the server silently swallows the duplicate. We picked 30 minutes because every retry path we have fires within a few minutes; a longer window costs memory in the server's dedupe table for no benefit.

The Go consumer, and why ack timing is the whole game

Thumbnail work is slow and network-bound, so it's a Go service. This uses the newer jetstream package rather than the legacy nats.JetStreamContext:

package main

import (
    "context"
    "encoding/json"
    "errors"
    "log"
    "os/signal"
    "syscall"
    "time"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

type VideoIngested struct {
    Schema     int    `json:"schema"`
    VideoID    int64  `json:"video_id"`
    ExternalID string `json:"external_id"`
    Source     string `json:"source"`
    Revision   int    `json:"revision"`
    ThumbURL   string `json:"thumb_url"`
}

var errPermanent = errors.New("permanent failure")

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    nc, err := nats.Connect(nats.DefaultURL,
        nats.MaxReconnects(-1),
        nats.ReconnectWait(time.Second),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Drain()

    js, err := jetstream.New(nc)
    if err != nil {
        log.Fatal(err)
    }

    cons, err := js.CreateOrUpdateConsumer(ctx, "VIDEO_EVENTS", jetstream.ConsumerConfig{
        Durable:       "thumbnailer",
        FilterSubject: "video.ingested.>",
        AckPolicy:     jetstream.AckExplicitPolicy,
        AckWait:       60 * time.Second,
        MaxDeliver:    5,
        MaxAckPending: 32, // bounds in-flight work: this is our concurrency dial
        BackOff: []time.Duration{
            2 * time.Second, 10 * time.Second, time.Minute, 5 * time.Minute,
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    cc, err := cons.Consume(func(msg jetstream.Msg) {
        var ev VideoIngested
        if err := json.Unmarshal(msg.Data(), &ev); err != nil {
            // Malformed bytes will never parse. Don't burn four more deliveries.
            _ = msg.Term()
            return
        }

        done := heartbeat(msg, 20*time.Second)
        defer done()

        switch err := renderThumbnail(ctx, ev); {
        case err == nil:
            _ = msg.Ack()
        case errors.Is(err, errPermanent):
            log.Printf("term %s: %v", msg.Subject(), err)
            _ = msg.Term()
        default:
            log.Printf("nak %s: %v", msg.Subject(), err)
            _ = msg.Nak()
        }
    })
    if err != nil {
        log.Fatal(err)
    }
    defer cc.Stop()

    <-ctx.Done()
}

// heartbeat keeps extending the ack deadline while slow work is in flight,
// so a genuinely slow job is never mistaken for a dead consumer.
func heartbeat(msg jetstream.Msg, every time.Duration) func() {
    stop := make(chan struct{})
    go func() {
        t := time.NewTicker(every)
        defer t.Stop()
        for {
            select {
            case <-t.C:
                _ = msg.InProgress()
            case <-stop:
                return
            }
        }
    }()
    return func() { close(stop) }
}

func renderThumbnail(ctx context.Context, ev VideoIngested) error {
    // Fetch ev.ThumbURL, re-encode to WebP at 3 widths, write to object storage.
    // Wrap 404/410 from upstream as errPermanent so the message is terminated
    // rather than retried five times against a URL that will never exist.
    return nil
}

The three-way Ack / Nak / Term split is the part people skip, and it's the part that determines whether your consumer is self-healing or a redelivery storm. Nak says try again; Term says this will never succeed, stop. Treating a 404 thumbnail URL as retryable meant every dead upstream video generated five deliveries with backoff before falling off — thousands of pointless HTTP requests a week.

MaxAckPending: 32 is doing double duty as flow control. The server will not deliver a 33rd unacked message to this consumer, so a slow thumbnailer simply lets the backlog sit in the stream rather than blowing up its own memory. That's backpressure you get for free, and it's the single biggest reason I stopped hand-rolling worker pools.

Ordering, and the fact that you probably don't need it

JetStream gives you total order within a stream, but the moment you run more than one consumer instance, or use Nak with backoff, delivery order stops matching publish order. A video.updated event can land before the video.ingested it followed.

We don't need global ordering. We need per-video correctness, which is a much weaker requirement, and the revision field satisfies it: any consumer that mutates state compares the event's revision against what's already stored and drops anything stale. Last-write-wins with an explicit version, rather than implicit ordering.

The search indexer is where this matters most, because FTS5 external-content tables need a delete-then-insert to update a row, and doing that twice out of order corrupts your ranking:

import asyncio
import json
import sqlite3

import nats
from nats.errors import TimeoutError as NatsTimeout

DB = "/var/www/dailywatch/data/videos.db"
BATCH = 100


def index_batch(conn: sqlite3.Connection, events: list[dict]) -> None:
    """One transaction, one writer. SQLite is much happier this way."""
    with conn:
        for ev in events:
            row = conn.execute(
                "SELECT rowid, revision FROM videos WHERE id = ?", (ev["video_id"],)
            ).fetchone()
            if row is None:
                continue  # row deleted since publish; nothing to index
            if row[1] > ev["revision"]:
                continue  # stale event, a newer revision already landed
            conn.execute("DELETE FROM videos_fts WHERE rowid = ?", (row[0],))
            conn.execute(
                "INSERT INTO videos_fts(rowid, title, channel, tags) VALUES (?, ?, ?, ?)",
                (row[0], ev["title"], ev.get("channel", ""), " ".join(ev.get("tags", []))),
            )


async def main() -> None:
    nc = await nats.connect("nats://127.0.0.1:4222", max_reconnect_attempts=-1)
    js = nc.jetstream()
    sub = await js.pull_subscribe(
        "video.>", durable="search-indexer", stream="VIDEO_EVENTS"
    )

    conn = sqlite3.connect(DB, timeout=30)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")

    while True:
        try:
            msgs = await sub.fetch(BATCH, timeout=5)
        except NatsTimeout:
            continue

        good, events = [], []
        for m in msgs:
            try:
                events.append(json.loads(m.data))
                good.append(m)
            except json.JSONDecodeError:
                await m.term()

        try:
            index_batch(conn, events)
        except sqlite3.Error as exc:
            print(f"batch failed, redelivering {len(good)} msgs: {exc}")
            for m in good:
                await m.nak(delay=5)
            continue

        # Ack only after the transaction has committed. Never before.
        for m in good:
            await m.ack()


asyncio.run(main())

Batching is not premature optimization here — it's the whole point. Acking 100 messages after one committed transaction is dramatically cheaper than 100 transactions, and it turns SQLite's single-writer constraint from a bottleneck into a non-issue. The ordering rule is absolute: commit, then ack. Acking first converts at-least-once delivery into at-most-once, and you will lose index updates on the next crash.

Operating it

A few things I wish someone had told me before rather than after.

• Set --max-age on every stream at creation time. Our one real incident was a disk filling up because a stream had max-bytes but no age limit and a backfill job published two million events in an hour. --discard old plus a real age limit means the stream sheds history instead of rejecting publishes.
• Alert on num_redelivered and num_ack_pending, not just num_pending. A backlog that's draining is fine. A backlog where the same messages are being redelivered means a consumer is failing in a loop, and num_pending alone won't tell you.
• Tap the max-deliveries advisory. JetStream publishes an advisory when a message exhausts MaxDeliver. That's your dead-letter feed and it costs one subscription.
• Version your payloads from event one. The schema integer is three bytes of JSON and it means a consumer can reject or adapt rather than crash on a field it doesn't recognize.

# Where is everything, right now
nats stream report
nats consumer report VIDEO_EVENTS

# Dead-letter feed: messages that exhausted MaxDeliver on any consumer
nats sub '$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>'

# Replay: rewind one consumer to a timestamp without touching the others
nats consumer rm VIDEO_EVENTS thumbnailer -f
nats consumer add VIDEO_EVENTS thumbnailer \
  --filter 'video.ingested.>' --ack explicit \
  --deliver-policy by-start-time --start-time '2026-08-01T00:00:00Z' \
  --max-deliver 5 --wait 60s --max-pending 32 --defaults

That last one deserves emphasis, because it's the capability that justified the whole migration. When we shipped a bad WebP encoding setting, fixing every affected thumbnail was: delete the durable consumer, recreate it with a start time, let it re-consume. No custom backfill script, no WHERE created_at BETWEEN query, no risk of touching rows the bug never affected. The other three consumers didn't notice.

What I'd do differently

Start with limits retention and explicit age and size caps. Keep one stream per bounded namespace rather than one per event type. Put the version field in from the first publish. And resist the urge to make subjects into commands — the day you add a consumer nobody planned for, fact-shaped subjects are what let you do it without redeploying the producer.

What we ended up with is smaller than the cron system it replaced. There are no status columns on the videos table, no reconciliation scripts, no per-consumer migrations. Ingest publishes a fact and stops caring. Median time from ingest to a fully rendered, indexed, cache-purged video page went from roughly 18 minutes to under 4 seconds on DailyWatch, and the part I appreciate most is that adding a fifth consumer is now a config line and a binary, not a schema change deployed across four sites.

JetStream isn't the right tool at every scale. But if you're at the point where your database has grown a status column per background job, it's worth the afternoon.