DEV Community: ahmet gedik

Rebuilding a Video Metadata Service on Litestar Without Losing CJK Search

ahmet gedik — Wed, 19 Aug 2026 08:00:04 +0000

240 ms to return 4 KB of JSON

TopVideoHub aggregates trending video metadata across nine Asia-Pacific regions, and the HTML side of it has never been the problem. PHP 8.4 renders a page, the result lands in a file page cache, LiteSpeed serves it, Cloudflare fronts it. That part is boring in the good way. The endpoints that hurt on TopVideoHub are the ones the browser calls after the page paints:

/api/suggest?q=... — search-as-you-type, mostly Japanese, Korean and Traditional Chinese queries
/api/related?id=... — the strip under the player
/api/trending/{region} — the region switcher on the home page

None of these can live in the HTML page cache. Suggest fires per keystroke with a query I have never seen before. Related is per-video across roughly 180k rows. The region switcher is cacheable at the edge but still needs a cold origin response every few minutes per region.

Each call was a fresh LSAPI request: interpreter bootstrap, autoloader, config parse, a new PDO handle to the SQLite file, PRAGMA setup, and then about 2 ms of actual query work. Sampled from a box in Singapore at 32 concurrent connections: p50 41 ms, p95 180 ms, occasional 240 ms outliers when the LSAPI worker pool churned. Roughly 90% of that number was everything except the query.

So I carved the metadata endpoints out into a separate Python process running Litestar, kept SQLite as the single source of truth, and left PHP owning every byte of HTML. This is the part I got right: it is not a rewrite. It is one process that holds a warm connection and an in-memory cache, sitting behind the same LiteSpeed instance.

Why Litestar and not FastAPI or plain ASGI

I evaluated three options and the deciding factors were unglamorous.

msgspec instead of Pydantic for serialization. Litestar uses msgspec internally for encoding and decoding. For a suggest response of 10 rows with Japanese titles, msgspec's encoder writes raw UTF-8 by default. json_encode() in PHP escapes non-ASCII into six-character sequences unless you pass JSON_UNESCAPED_UNICODE, which turns a 3-byte kana character into 6 bytes on the wire. My suggest payload dropped about 38% just from that, and I had genuinely never noticed the flag was missing in the old endpoint.

Explicit sync_to_thread. SQLite's Python driver is synchronous. Litestar refuses to let you register a sync handler without saying sync_to_thread=True or False — it raises a warning at import time. FastAPI silently offloads def handlers to a threadpool, which is the right default until the day you accidentally put a blocking call in an async def and stall the whole loop. I wanted the failure mode to be loud.

Layered configuration. Cache TTL, cache-control headers and dependencies can be declared on the app, the router, the controller or the handler, and the most specific one wins. That maps directly onto how the site's caching already works (per-route TTLs in PHP config), so I could port the table of TTLs rather than reinvent it.

What I did not pick it for: benchmark charts. On this workload the framework overhead is a rounding error next to the SQLite query. The win came from process reuse, not from routing speed.

The endpoint I was replacing

For reference, here is roughly what the PHP version looked like. It is fine code. It just pays full startup cost on every single keystroke.

<?php
// public/api/suggest.php - the endpoint this post replaces
declare(strict_types=1);

$q = trim((string)($_GET['q'] ?? ''));
if ($q === '' || mb_strlen($q) > 64) {
    http_response_code(400);
    echo '[]';
    exit;
}

$pdo = new PDO('sqlite:' . __DIR__ . '/../../data/videos.db', null, null, [
    PDO::ATTR_ERRMODE            => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
]);
$pdo->exec('PRAGMA query_only = ON');

// NFKC + lowercase, same normalisation the indexer applies
$norm = mb_strtolower(Normalizer::normalize($q, Normalizer::FORM_KC));
$expr = '"' . str_replace('"', '""', $norm) . '"';

$stmt = $pdo->prepare(
    'SELECT v.youtube_id, v.title, v.channel_title
       FROM videos_fts f
       JOIN videos v ON v.id = f.rowid
      WHERE videos_fts MATCH :q
      ORDER BY bm25(videos_fts, 4.0, 1.0), v.view_count DESC
      LIMIT 10'
);
$stmt->execute([':q' => $expr]);

header('Content-Type: application/json; charset=utf-8');
header('Cache-Control: public, max-age=600');
echo json_encode($stmt->fetchAll(), JSON_UNESCAPED_UNICODE);

The schema stays exactly where it is

The Python service opens the same file the PHP cron writer owns, in read-only mode. WAL means readers never block the writer and the writer never blocks readers, which is the only reason this arrangement is safe at all.

PRAGMA journal_mode = WAL;

CREATE TABLE IF NOT EXISTS videos (
  id            INTEGER PRIMARY KEY,
  youtube_id    TEXT NOT NULL UNIQUE,
  title         TEXT NOT NULL,
  title_norm    TEXT NOT NULL,          -- NFKC + lowercased, written by the fetcher
  channel_title TEXT NOT NULL,
  region        TEXT NOT NULL,
  lang          TEXT NOT NULL DEFAULT 'und',
  view_count    INTEGER NOT NULL DEFAULT 0,
  published_at  TEXT NOT NULL,
  fetched_at    TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_videos_region_pub
  ON videos(region, published_at DESC);
CREATE INDEX IF NOT EXISTS idx_videos_recent
  ON videos(published_at DESC, view_count DESC);

-- trigram: the only built-in tokenizer that behaves sanely on CJK
CREATE VIRTUAL TABLE IF NOT EXISTS videos_fts USING fts5(
  title_norm,
  channel_title,
  content='videos',
  content_rowid='id',
  tokenize="trigram remove_diacritics 1"
);

CREATE TRIGGER IF NOT EXISTS videos_ai AFTER INSERT ON videos BEGIN
  INSERT INTO videos_fts(rowid, title_norm, channel_title)
  VALUES (new.id, new.title_norm, new.channel_title);
END;

CREATE TRIGGER IF NOT EXISTS videos_ad AFTER DELETE ON videos BEGIN
  INSERT INTO videos_fts(videos_fts, rowid, title_norm, channel_title)
  VALUES ('delete', old.id, old.title_norm, old.channel_title);
END;

CREATE TRIGGER IF NOT EXISTS videos_au AFTER UPDATE ON videos BEGIN
  INSERT INTO videos_fts(videos_fts, rowid, title_norm, channel_title)
  VALUES ('delete', old.id, old.title_norm, old.channel_title);
  INSERT INTO videos_fts(rowid, title_norm, channel_title)
  VALUES (new.id, new.title_norm, new.channel_title);
END;

Why trigram and not unicode61 or ICU

This is the part that costs people weeks, so it is worth being precise.

unicode61, the FTS5 default, splits on non-alphanumeric codepoints. Japanese and Chinese do not put spaces between words, so a title like a six-character Japanese phrase becomes one token. Searching for a two-character substring of it matches nothing. You can add a prefix index and get prefix matches, which helps only if the user types from the beginning of the whole title. In practice suggest returned zero results for most Japanese queries and I did not notice for a month because the fallback rendered an empty dropdown rather than an error.

The ICU tokenizer does real dictionary-based segmentation and is the correct answer — if you can compile SQLite with ICU. On shared LiteSpeed hosting I cannot, and I am not going to run a custom SQLite build to serve a video site.

The trigram tokenizer indexes every overlapping 3-character window and never tries to find word boundaries at all. That makes it script-agnostic: it works identically for Korean, Japanese, Chinese and Latin text, and it gives you substring matching for free. The costs are real and you should know them going in:

The index is roughly 3–4× larger than a unicode61 index over the same text. On 180k titles that took the FTS table from about 22 MB to about 78 MB. Acceptable.
You cannot set detail='none'. Trigram matching relies on position data to reassemble phrases; strip it and phrase queries stop working.
Queries shorter than 3 characters cannot use the index at all. MATCH on a 2-character string returns nothing, silently.
bm25() over trigrams is noisy — it is ranking character windows, not words. I always add a secondary sort key (view_count DESC) so the ordering is stable and defensible.

That last point matters more than it sounds. Two-character queries are extremely common in Japanese and Chinese search boxes. I needed an explicit fallback path, not a shrug.

The service

One file, no ORM, no migration framework. The writer is PHP; Python only reads.

from __future__ import annotations

import sqlite3
import threading
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

import msgspec
from litestar import Litestar, get
from litestar.config.response_cache import ResponseCacheConfig
from litestar.datastructures import CacheControlHeader, State
from litestar.di import Provide
from litestar.exceptions import NotFoundException
from litestar.params import Parameter

DB_PATH = "/home/tvh/data/videos.db"


class VideoOut(msgspec.Struct, frozen=True):
    id: str
    title: str
    channel: str
    region: str
    views: int
    published_at: str


def _to_video(r: sqlite3.Row) -> VideoOut:
    return VideoOut(
        id=r["youtube_id"],
        title=r["title"],
        channel=r["channel_title"],
        region=r["region"],
        views=r["view_count"],
        published_at=r["published_at"],
    )


class Repo:
    """One read-only connection per worker thread, opened lazily."""

    def __init__(self, path: str) -> None:
        self._path = path
        self._local = threading.local()
        self._opened: list[sqlite3.Connection] = []
        self._lock = threading.Lock()

    def conn(self) -> sqlite3.Connection:
        c = getattr(self._local, "c", None)
        if c is None:
            c = sqlite3.connect(
                f"file:{self._path}?mode=ro", uri=True, check_same_thread=False
            )
            c.row_factory = sqlite3.Row
            c.execute("PRAGMA query_only = ON")
            c.execute("PRAGMA mmap_size = 268435456")
            c.execute("PRAGMA cache_size = -32000")
            self._local.c = c
            with self._lock:
                self._opened.append(c)
        return c

    def close_all(self) -> None:
        with self._lock:
            for c in self._opened:
                c.close()
            self._opened.clear()

    def trending(self, region: str, limit: int) -> list[VideoOut]:
        rows = self.conn().execute(
            """
            SELECT youtube_id, title, channel_title, region, view_count, published_at
              FROM videos
             WHERE region = ?
             ORDER BY published_at DESC
             LIMIT ?
            """,
            (region.upper(), limit),
        ).fetchall()
        return [_to_video(r) for r in rows]

    def suggest(self, match_expr: str, limit: int) -> list[VideoOut]:
        rows = self.conn().execute(
            """
            SELECT v.youtube_id, v.title, v.channel_title, v.region,
                   v.view_count, v.published_at
              FROM videos_fts f
              JOIN videos v ON v.id = f.rowid
             WHERE videos_fts MATCH ?
             ORDER BY bm25(videos_fts, 4.0, 1.0), v.view_count DESC
             LIMIT ?
            """,
            (match_expr, limit),
        ).fetchall()
        return [_to_video(r) for r in rows]


def provide_repo(state: State) -> Repo:
    return state.repo


@get(
    "/api/v1/trending/{region:str}",
    sync_to_thread=True,
    cache=300,
    cache_control=CacheControlHeader(max_age=300, public=True),
)
def trending(
    region: str,
    repo: Repo,
    limit: int = Parameter(default=24, ge=1, le=60),
) -> list[VideoOut]:
    items = repo.trending(region, limit)
    if not items:
        raise NotFoundException(f"no videos indexed for region {region}")
    return items


@get(
    "/api/v1/suggest",
    sync_to_thread=True,
    cache=60,
    cache_control=CacheControlHeader(max_age=60, public=True),
)
def suggest(
    repo: Repo,
    q: str = Parameter(min_length=1, max_length=64),
    limit: int = Parameter(default=10, ge=1, le=25),
) -> list[VideoOut]:
    expr = to_match_expr(q)
    if expr is None:
        return repo.suggest_short(normalize(q), limit)
    return repo.suggest(expr, limit)


@asynccontextmanager
async def db(app: Litestar) -> AsyncGenerator[None, None]:
    app.state.repo = Repo(DB_PATH)
    try:
        yield
    finally:
        app.state.repo.close_all()


app = Litestar(
    route_handlers=[trending, suggest],
    dependencies={"repo": Provide(provide_repo, sync_to_thread=False)},
    lifespan=[db],
    response_cache_config=ResponseCacheConfig(default_expiration=120),
)

Two details in there that cost me time. Provide(provide_repo, sync_to_thread=False) is mandatory for a synchronous dependency — the provider does nothing but return an attribute, so pushing it to a thread would be pure overhead, but Litestar still wants you to say which you meant. And app.state set inside the lifespan context is the same State object injected into provide_repo, so there is no global and no import-time connection.

Run it with litestar --app app:app run --port 8081 while developing, and with a production ASGI server behind that in deployment.

Normalizing the query the same way you normalized the document

The single largest source of "why does this return nothing" bugs in CJK search is asymmetric normalization. Half-width katakana and full-width Latin are different codepoints from their normal-width equivalents, and users type both — Japanese IMEs emit full-width digits and Latin letters constantly. If the indexer applies NFKC and the query path does not, every query containing a full-width character misses.

import re
import unicodedata

_WS = re.compile(r"\s+")


def normalize(q: str) -> str:
    """Must stay byte-identical in behaviour to the PHP indexer."""
    q = unicodedata.normalize("NFKC", q)
    q = q.casefold().strip()
    return _WS.sub(" ", q)


def to_match_expr(q: str) -> str | None:
    """Wrap as an FTS5 string literal, or None if trigram can't serve it."""
    q = normalize(q)
    if len(q) < 3:
        return None
    return '"' + q.replace('"', '""') + '"'

Quoting the whole query as a single FTS5 string literal is not optional. An unquoted MATCH argument is parsed as a query expression, so a user typing AND, *, ^, : or a stray " either changes the semantics or throws a syntax error straight into a 500. Doubling embedded quotes inside the literal is the FTS5 escape rule.

The short-query fallback lives on the repo:

    def suggest_short(self, q: str, limit: int) -> list[VideoOut]:
        """Under 3 chars trigram is useless. Scan the recent partition only."""
        if not q:
            return []
        rows = self.conn().execute(
            """
            SELECT youtube_id, title, channel_title, region, view_count, published_at
              FROM videos
             WHERE published_at > date('now', '-30 day')
               AND title_norm LIKE '%' || ? || '%'
             ORDER BY view_count DESC
             LIMIT ?
            """,
            (q, limit),
        ).fetchall()
        return [_to_video(r) for r in rows]

A leading-wildcard LIKE cannot use an index; this is a scan and I am not pretending otherwise. What makes it acceptable is the 30-day bound, which the published_at DESC index turns into a range of about 41k rows on my dataset — roughly 6 ms warm. If the corpus triples I will bound it harder rather than pretend the query plan improved.

One cross-language wrinkle: PHP's mb_strtolower and Python's str.casefold() are not identical (casefold maps the German sharp s to ss, lowercasing does not). For CJK it makes no difference, and I documented the divergence rather than papering over it.

Sync SQLite inside an async framework

sync_to_thread=True hands the handler to AnyIO's worker thread pool. The default capacity is 40 threads, which is far more concurrency than a single SQLite file wants — 40 threads all mmap-ing and page-faulting is a good way to make p99 worse, not better. I cap it at startup:

import anyio.to_thread

@asynccontextmanager
async def db(app: Litestar) -> AsyncGenerator[None, None]:
    anyio.to_thread.current_default_thread_limiter().total_tokens = 12
    app.state.repo = Repo(DB_PATH)
    try:
        yield
    finally:
        app.state.repo.close_all()

Twelve threads times one connection each is twelve open read handles. Requests beyond that queue in the event loop instead of thrashing the page cache, and the tail latency curve got noticeably flatter.

Caching in three places, not four

The site already had three cache layers (LiteSpeed page cache, a PHP file cache, a PHP data cache). I refused to add a fourth conceptual layer, so the Python service reuses the same TTLs the PHP config already declares:

Litestar response cache (cache=300 on trending, cache=60 on suggest) — in-process, keyed by path plus query string. This is what makes the region switcher effectively free.
Cache-Control on the response — so Cloudflare can hold trending at the edge. Suggest gets a short TTL because query diversity makes edge caching mostly useless anyway.
The SQLite page cache in each connection, which is the layer that actually pays for keeping the process alive.

The in-process cache does mean each ASGI worker holds its own copy. With two workers that is two cold misses per TTL window instead of one. Fine at this scale; if it stops being fine, Litestar's response cache accepts a shared store rather than the default in-memory one, and that is a config change, not a rewrite.

Measuring it honestly

I wrote a small sampler rather than reaching for a load-testing suite, because I wanted latency percentiles against one endpoint at a fixed concurrency, not a report.

package main

import (
    "fmt"
    "io"
    "net/http"
    "os"
    "sort"
    "sync"
    "time"
)

func main() {
    if len(os.Args) < 2 {
        fmt.Println("usage: sampler <url>")
        return
    }
    url := os.Args[1]
    const n, workers = 2000, 32

    lat := make([]time.Duration, n)
    jobs := make(chan int, n)
    client := &http.Client{Timeout: 10 * time.Second}

    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := range jobs {
                start := time.Now()
                resp, err := client.Get(url)
                if err != nil {
                    lat[i] = 10 * time.Second
                    continue
                }
                io.Copy(io.Discard, resp.Body)
                resp.Body.Close()
                lat[i] = time.Since(start)
            }
        }()
    }
    for i := 0; i < n; i++ {
        jobs <- i
    }
    close(jobs)
    wg.Wait()

    sort.Slice(lat, func(a, b int) bool { return lat[a] < lat[b] })
    fmt.Printf("p50=%v p95=%v p99=%v\n", lat[n/2], lat[n*95/100], lat[n*99/100])
}

On a 2-vCPU box, same machine, same database file, /api/suggest with a rotating set of Japanese and Korean queries:

PHP over LSAPI: p50 41 ms, p95 180 ms, p99 244 ms
Litestar, two workers, cache warm: p50 6 ms, p95 19 ms, p99 34 ms

Read that as "process reuse is worth about 35 ms per request here," not "Python beat PHP." The same PHP code behind a persistent worker with a pooled handle would close most of the gap. What I actually bought was a place to keep warm state, plus a search path I can unit-test without booting a web server.

Where it sits in the stack

LiteSpeed proxies /api/v1/* to 127.0.0.1:8081 and serves everything else from PHP. Cloudflare sees one origin and one hostname, so no CORS and no preflight. The old PHP endpoints are still on disk behind a config flag; flipping it back is a one-line change, and I have used that escape hatch twice — once when a botched deploy left the Python process down, once when I broke the normalization function and did not catch it in tests.

What I would keep and what I would skip

Keep: the trigram tokenizer, the shared normalization contract between the indexer and the query path, the explicit thread limiter, and the read-only URI connection flag. Those four things are the whole substance of the migration.

Skip: the instinct to move more endpoints across just because the service exists. Rendering stayed in PHP and should stay there — the page cache already makes it a non-problem, and running two languages against one page is a debugging tax I have no reason to pay. The metadata service earns its keep precisely because it is small, read-only, and has exactly one job.

If you are running SQLite full-text search against CJK content and getting empty result sets, check your tokenizer before you check anything else. It is almost always that.

Distributing Video Ingest Events Between Services With NATS JetStream

ahmet gedik — Wed, 19 Aug 2026 00:00:01 +0000

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.

Building a Video Thumbnail Generator Service with Go and FFmpeg Workers

ahmet gedik — Tue, 18 Aug 2026 16:00:01 +0000

Every video card on our category grids was hotlinking a 1280x720 JPEG from a third-party CDN and then letting CSS scale it down to about 320 device-independent pixels. That is roughly 90 KB of wasted transfer per card, 24 cards per page, across eight regional page variants that each carry their own cache key. Mobile LCP on the busiest category pages sat at 4.1s, and the largest single contributor was an image we did not host, could not resize, and could not re-encode to WebP.

The fix was not clever CSS. It was owning the frame. We built a small Go service that takes a source video (a partner preview MP4, or a poster frame that arrives at the wrong dimensions), pulls a representative frame with FFmpeg, encodes it at three widths in WebP, and writes the result to a content-addressed path the front end links directly. That service now feeds the same multi-region cron that runs TrendVidStream, and the generated files ride the same FTP mirror as the rest of the deploy.

What follows is the part that actually mattered: the FFmpeg invocations, the Go concurrency model that keeps a 2-core build box from melting, and how a stateless Go daemon hands work to a PHP 8.4 + SQLite front end that cannot run a daemon at all.

Why this is not a PHP job

Our front end is PHP 8.4 on LiteSpeed shared hosting with SQLite (FTS5 for search) as the only datastore. It is a genuinely good fit for a read-heavy discovery site: no database server to babysit, page cache on disk, cron jobs pulling regional feeds every 2-7 hours depending on the site.

It is a terrible fit for thumbnail extraction:

Shared hosting caps max_execution_time at 180s. A cold FFmpeg decode of a 4-minute 1080p preview can burn 20-40s. Do 200 of them in one cron tick and you are wearing a hard timeout.
shell_exec is frequently disabled, and when it is not, you get one process per request with no way to bound total concurrency.
There is no shared memory between PHP requests, so two cron ticks racing on the same video ID will happily run FFmpeg twice and write the same file twice.

Go solves the three things PHP cannot here: a long-lived process that owns a bounded worker pool, context deadlines that reliably kill a wedged FFmpeg child, and in-process deduplication so N concurrent requests for one video produce exactly one decode. FFmpeg does the pixels; Go does the traffic control. There is no image processing in the Go code at all, which is deliberate — every attempt I have seen to do frame extraction with a pure-Go decoder ends in a codec matrix nobody wants to own.

The FFmpeg contract

Before any of the Go matters, the FFmpeg command has to be right, because the defaults are wrong for a service.

Four flags do the heavy lifting:

-ss before -i, not after. Placed before the input it becomes input seeking: FFmpeg jumps to the nearest keyframe and starts decoding there. Placed after, it decodes every frame from zero and throws them away. On a 4-minute file seeking to 00:00:30 that is the difference between ~300ms and ~8s.
-nostdin. A daemon-spawned FFmpeg that inherits stdin will silently eat bytes from it and can block forever. This flag has saved me more debugging hours than any other.
-frames:v 1. Stop after one frame. Without it a filter graph that produces frames keeps producing them.
thumbnail=N. This is the filter that makes output not look like garbage. Frame zero on a streaming preview is almost always a black fade-in or a distributor logo. thumbnail=300 buffers 300 frames, computes a histogram for each, and picks the one least like the batch average — i.e. the most visually distinctive frame. It costs one decode pass over those 300 frames and it is worth it every time.

Here is the extraction call, wrapped so the caller gets a deadline and a real error instead of a hang:

package thumb

import (
    "context"
    "fmt"
    "os"
    "os/exec"
    "path/filepath"
    "strconv"
    "strings"
    "syscall"
    "time"
)

// Variant is one output width. We ship 320/640/1280 and let srcset pick.
type Variant struct {
    Width   int
    Quality int // libwebp -quality, 0-100
}

var Variants = []Variant{
    {Width: 320, Quality: 78},
    {Width: 640, Quality: 80},
    {Width: 1280, Quality: 82},
}

// Extract writes one WebP per variant into dstDir, named <base>-<width>.webp.
// seek is where to start looking; the thumbnail filter picks the actual frame.
func Extract(ctx context.Context, src, dstDir, base string, seek time.Duration) ([]string, error) {
    if err := os.MkdirAll(dstDir, 0o755); err != nil {
        return nil, err
    }

    out := make([]string, 0, len(Variants))
    for _, v := range Variants {
        final := filepath.Join(dstDir, fmt.Sprintf("%s-%d.webp", base, v.Width))
        tmp := final + ".tmp"

        args := []string{
            "-nostdin", "-y", "-loglevel", "error",
            "-ss", strconv.FormatFloat(seek.Seconds(), 'f', 3, 64),
            "-i", src,
            "-an", "-sn", "-dn",
            "-map_metadata", "-1",
            "-vf", fmt.Sprintf(
                "thumbnail=300,scale=%d:-2:flags=lanczos,format=yuv420p", v.Width),
            "-frames:v", "1",
            "-c:v", "libwebp",
            "-quality", strconv.Itoa(v.Quality),
            "-compression_level", "6",
            "-preset", "picture",
            "-threads", "1", // pool size controls parallelism, not FFmpeg
            tmp + ".webp",
        }

        cmd := exec.CommandContext(ctx, "ffmpeg", args...)
        // Ask nicely first, then hard-kill. Requires Go 1.20+.
        cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
        cmd.WaitDelay = 5 * time.Second

        var stderr strings.Builder
        cmd.Stderr = &stderr

        if err := cmd.Run(); err != nil {
            os.Remove(tmp + ".webp")
            return nil, fmt.Errorf("ffmpeg %dpx: %w: %s", v.Width, err, strings.TrimSpace(stderr.String()))
        }
        // Rename last so a reader never sees a half-written file.
        if err := os.Rename(tmp+".webp", final); err != nil {
            return nil, err
        }
        out = append(out, final)
    }
    return out, nil
}

Three details in there are load-bearing and easy to miss:

scale=W:-2, not -1. The -2 rounds the computed height to an even number. Odd dimensions break several encoders outright, and you will only find out on the one 1440x1079 source in your corpus.
-threads 1. FFmpeg will happily spawn a thread per core. If you also run 8 workers you get 64 threads fighting over 2 cores and your p99 goes vertical. One thread per process, N processes, N = pool size.
Write to .tmp, then os.Rename. Rename within a filesystem is atomic on Linux, so the web server either serves the complete old file or the complete new one. Without this you will eventually serve a truncated WebP and Chrome will render a grey box.

A bounded pool with request coalescing

The service takes HTTP requests, but it is really a queue with an HTTP face. Two constraints shape it: never run more FFmpeg processes than cores, and never run the same job twice.

The second one matters more than it sounds. Our regional cron jobs overlap — eight regions, staggered by minutes, and the same trending video shows up in five of them. Without coalescing, five simultaneous requests for dQw4w9WgXcQ mean five decodes of the same file, five writes to the same path, and a real chance of two renames interleaving. golang.org/x/sync/singleflight collapses them into one execution whose result is shared by all callers:

package main

import (
    "context"
    "encoding/json"
    "log"
    "net/http"
    "runtime"
    "time"

    "golang.org/x/sync/singleflight"
    "example.com/thumbd/thumb"
)

type Server struct {
    sem    chan struct{}     // bounded FFmpeg concurrency
    group  singleflight.Group // one decode per key, no matter how many callers
    dstDir string
}

func NewServer(dstDir string) *Server {
    n := runtime.NumCPU()
    if n > 4 {
        n = 4 // leave headroom; the box also runs the article pipeline
    }
    return &Server{sem: make(chan struct{}, n), dstDir: dstDir}
}

type request struct {
    ID   string `json:"id"`   // our internal video id, used as the file base
    Src  string `json:"src"`  // path or URL FFmpeg can open
    Seek int    `json:"seek"` // seconds; 0 means let the filter decide
}

func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
    var req request
    if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8<<10)).Decode(&req); err != nil {
        http.Error(w, "bad json", http.StatusBadRequest)
        return
    }
    if req.ID == "" || req.Src == "" {
        http.Error(w, "id and src required", http.StatusBadRequest)
        return
    }

    // Client gets 90s max; the pool wait is inside that budget.
    ctx, cancel := context.WithTimeout(r.Context(), 90*time.Second)
    defer cancel()

    res, err, shared := s.group.Do(req.ID, func() (any, error) {
        select {
        case s.sem <- struct{}{}:
            defer func() { <-s.sem }()
        case <-ctx.Done():
            return nil, ctx.Err()
        }
        return thumb.Extract(ctx, req.Src, s.dstDir, req.ID,
            time.Duration(req.Seek)*time.Second)
    })
    if err != nil {
        log.Printf("extract %s: %v", req.ID, err)
        http.Error(w, "extract failed", http.StatusBadGateway)
        return
    }
    if shared {
        w.Header().Set("X-Coalesced", "1")
    }
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]any{"id": req.ID, "files": res})
}

func main() {
    s := NewServer("/srv/thumbs")
    mux := http.NewServeMux()
    mux.HandleFunc("POST /thumb", s.handle)
    mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
        w.Write([]byte("ok"))
    })
    srv := &http.Server{
        Addr:              "127.0.0.1:8431",
        Handler:           mux,
        ReadHeaderTimeout: 5 * time.Second,
        WriteTimeout:      120 * time.Second, // must exceed the job budget
    }
    log.Fatal(srv.ListenAndServe())
}

A few things I got wrong the first time and would flag for anyone building this:

WriteTimeout must be longer than your job deadline. Mine was 30s against a 90s job. The handler completed successfully and the client got a truncated response, which looks exactly like an FFmpeg failure in the logs. Cost me an afternoon.
Acquire the semaphore inside singleflight.Do, not outside. Outside, every duplicate caller burns a pool slot waiting on a job it is not running.
singleflight is per-process only. If you ever run two instances, you need the filesystem check (below) as the real guard. Coalescing is an optimization, not a correctness mechanism.

Wiring it to PHP and SQLite

The front end does not call the Go service during a page render — that would put an FFmpeg decode on the critical path of an HTTP request, which is the whole problem we were solving. Instead the multi-region cron enqueues, and the renderer reads whatever exists.

The schema is deliberately dumb. One row per video, a status column, and a nullable path. It sits alongside the FTS5 search table but is not part of it — you do not want thumbnail churn triggering FTS index writes.

<?php
declare(strict_types=1);

final class ThumbQueue
{
    private const ENDPOINT = 'http://127.0.0.1:8431/thumb';
    private const TIMEOUT  = 95;

    public function __construct(private readonly PDO $db) {}

    public function migrate(): void
    {
        $this->db->exec(<<<SQL
            CREATE TABLE IF NOT EXISTS thumbs (
                video_id   TEXT PRIMARY KEY,
                src        TEXT NOT NULL,
                status     TEXT NOT NULL DEFAULT 'pending',
                width_max  INTEGER,
                updated_at INTEGER NOT NULL DEFAULT 0,
                attempts   INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS thumbs_status
                ON thumbs(status, attempts, updated_at);
        SQL);
    }

    /** Called by cron. Returns number of videos successfully rendered. */
    public function drain(int $limit = 40): int
    {
        $rows = $this->db->prepare(
            "SELECT video_id, src FROM thumbs
              WHERE status IN ('pending','failed') AND attempts < 3
              ORDER BY attempts ASC, updated_at ASC
              LIMIT :lim"
        );
        $rows->bindValue(':lim', $limit, PDO::PARAM_INT);
        $rows->execute();

        $done = 0;
        foreach ($rows->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $ok = $this->render($row['video_id'], $row['src']);
            $this->db->prepare(
                'UPDATE thumbs
                    SET status = :s, attempts = attempts + 1,
                        width_max = :w, updated_at = :t
                  WHERE video_id = :id'
            )->execute([
                ':s'  => $ok ? 'ready' : 'failed',
                ':w'  => $ok ? 1280 : null,
                ':t'  => time(),
                ':id' => $row['video_id'],
            ]);
            $done += (int) $ok;
        }
        return $done;
    }

    private function render(string $id, string $src): bool
    {
        $ch = curl_init(self::ENDPOINT);
        curl_setopt_array($ch, [
            CURLOPT_POST           => true,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT        => self::TIMEOUT,
            CURLOPT_HTTPHEADER     => ['Content-Type: application/json'],
            CURLOPT_POSTFIELDS     => json_encode(
                ['id' => $id, 'src' => $src, 'seek' => 5],
                JSON_THROW_ON_ERROR
            ),
        ]);
        $body = curl_exec($ch);
        $code = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
        curl_close($ch);

        return $body !== false && $code === 200;
    }
}

On the render side the template asks one question: is status = 'ready'? If yes it emits a srcset over our own three widths; if no it falls back to the old remote URL. That fallback is the reason we could ship this incrementally instead of as a big-bang migration — the site worked identically at 0% coverage and at 100%, and we watched the ratio climb across four sites over about a week.

One SQLite note that bit us: with several cron jobs writing at once you want PRAGMA journal_mode=WAL and a busy_timeout of a few seconds, or the queue update will throw SQLITE_BUSY right after a successful render and the video gets re-decoded next tick.

Backfilling 40,000 rows without a thundering herd

The steady-state cron handles a few dozen new videos per tick. The initial backfill was a different animal, and PHP's 180s wall meant it had to run from the build box. A short Python script was the pragmatic answer — stdlib only, so it runs anywhere:

#!/usr/bin/env python3
"""Backfill thumbnails by walking the SQLite queue and posting to thumbd."""
import json
import sqlite3
import sys
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

ENDPOINT = "http://127.0.0.1:8431/thumb"
WORKERS = 6          # > server pool: keeps the pool saturated, no more
DB = sys.argv[1] if len(sys.argv) > 1 else "data/site.db"


def post(video_id: str, src: str) -> tuple[str, bool]:
    payload = json.dumps({"id": video_id, "src": src, "seek": 5}).encode()
    req = urllib.request.Request(
        ENDPOINT, data=payload, headers={"Content-Type": "application/json"}
    )
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=120) as resp:
                return video_id, resp.status == 200
        except (urllib.error.URLError, TimeoutError):
            time.sleep(2 ** attempt)  # 1s, 2s, 4s
    return video_id, False


def main() -> int:
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA busy_timeout = 5000")
    rows = conn.execute(
        "SELECT video_id, src FROM thumbs WHERE status != 'ready' AND attempts < 3"
    ).fetchall()
    print(f"{len(rows)} pending", flush=True)

    ok = bad = 0
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        futures = {pool.submit(post, vid, src): vid for vid, src in rows}
        for fut in as_completed(futures):
            vid, success = fut.result()
            conn.execute(
                "UPDATE thumbs SET status = ?, attempts = attempts + 1,"
                " updated_at = ? WHERE video_id = ?",
                ("ready" if success else "failed", int(time.time()), vid),
            )
            ok, bad = ok + success, bad + (not success)
            if (ok + bad) % 250 == 0:
                conn.commit()
                print(f"ok={ok} failed={bad}", flush=True)
    conn.commit()
    print(f"done ok={ok} failed={bad}")
    return 0 if bad == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())

Six client threads against a four-slot server pool is intentional. You want the pool queue shallow but never empty; pushing client concurrency to 30 just moves the wait from the semaphore to the HTTP layer, where the timeout is less forgiving.

Shipping the output over FTP

Here is the constraint that shapes the deploy: our four sites run on shared LiteSpeed hosting with FTP access and no ability to run a persistent process. The Go service therefore never runs on the sites. It runs on the build box, and the WebP files it produces are just static assets that ride the same lftp mirror as PHP templates and CSS.

That turns out to be an advantage:

The static binary is built once with CGO_ENABLED=0 GOOS=linux go build -trimpath. No runtime, no extensions, no PHP version coupling.
Thumbnails are content-addressed by video ID and width, so lftp mirror --only-newer --parallel=4 uploads only what changed. A typical tick pushes a few hundred KB.
Regional sites share the generated files. Eight regions do not mean eight renders — the region only affects which videos get requested, never how they are encoded.
If the generator is down, deploys still work. Nothing on the hosting side depends on it being alive.

The one gotcha worth repeating: make sure the host list file used by your deploy script has Unix line endings. A trailing \r gets appended to the remote path and lftp will cheerfully create a directory with a carriage return in its name, then report success. cat -A on the config file is a five-second check that has saved me twice.

What actually changed

Measured on the same set of category pages, before and after, mobile throttled:

Thumbnail payload per grid page: ~2.1 MB → ~310 KB. Three widths in srcset means phones fetch the 320px variant, which averages 11 KB at quality 78.
Mobile LCP: 4.1s → 1.9s. Most of that is not the byte count — it is that the image now comes from the same origin with a long Cache-Control, instead of a third-party host requiring a fresh DNS lookup and TLS handshake.
Generation throughput: about 26 videos/minute on 2 cores, three variants each. FFmpeg spends roughly 70% of that in the thumbnail=300 decode pass, which is the price of not shipping black frames.
Failure rate settled near 1.4%, almost entirely sources that 404 or have no video stream. Three attempts then permanent failed, and the template falls back to the remote URL.

If you build one of these, the ordering that saved me time was: get the FFmpeg command right in a shell first, then wrap it, then add concurrency, then add the queue. Every bug I chased in the Go layer for the first two days turned out to be an FFmpeg flag I had not verified by hand.

Conclusion

The service is under 400 lines of Go and it is boring by design: FFmpeg does the decoding, a buffered channel does the throttling, singleflight does the deduplication, and os.Rename does the atomicity. The interesting engineering was not in the code — it was in accepting that the PHP front end should stay a pure reader, and that anything with an unbounded runtime belongs in a separate process with its own deadline.

The pattern generalizes past thumbnails. Any time you have a fast, cacheable read path and a slow, resource-hungry transform, splitting them at a queue boundary lets each side use the right tool. Ours happens to be PHP on one side and Go on the other, connected by a SQLite table and an FTP mirror, which is not an architecture anyone would draw on a whiteboard — but it has been running unattended for months across four sites and eight regions, and the grid pages load in under two seconds.

Rotating JWT Refresh Tokens Without Logging Out Every Video Client

ahmet gedik — Tue, 18 Aug 2026 07:00:03 +0000

Our video API had a bug that only showed up in the German traffic logs. A user opens the ViralVidVault Android app, the access token expires mid-scroll, the app fires a refresh, and simultaneously a background prefetch worker fires its own refresh with the same token. Two requests, one token. Our old implementation invalidated the refresh token on first use and issued a new one — so whichever request arrived second got a 401, the client cleared its session, and the user was logged out while watching a video. We saw this roughly 40 times a day, almost always on mobile networks where retries and connection races are common.

The naive fix is to stop rotating refresh tokens. That is the wrong fix: a long-lived non-rotating refresh token stolen from local storage grants an attacker indefinite access with no way to detect it. Rotation is what makes theft detectable. This article covers how we implemented rotation with reuse detection on ViralVidVault, running PHP 8.4 against SQLite in WAL mode behind LiteSpeed, including the grace-window mechanism that killed the race condition and the GDPR constraints that shaped the token table schema.

Why Rotation Changes the Threat Model

A refresh token is a bearer credential with a long lifetime. If it never rotates, these two worlds are indistinguishable from the server's perspective:

A legitimate client refreshing every 15 minutes for 30 days.
An attacker who exfiltrated the token on day 1 and has been refreshing every 15 minutes since.

Rotation introduces a serial number. Each refresh consumes token N and issues token N+1, both belonging to the same family (a chain rooted at the original login). The rule is: any attempt to use a consumed token is evidence of duplication. Either the attacker used the stolen token first (and the legitimate client now presents a stale one), or the legitimate client used it first (and the attacker now presents a stale one). Either way, the server sees a token that was already spent, and the correct response is to revoke the entire family and force re-authentication.

This is the reuse-detection pattern described in OAuth 2.0 Security Best Current Practice (RFC 9700). It converts a silent, permanent compromise into a loud, bounded one: the attacker gets at most one refresh cycle before the family dies.

The cost is exactly the bug we hit. Legitimate clients also produce duplicate refreshes — from network retries, from concurrent tabs, from a mobile OS killing and restoring a process mid-request. If every duplicate triggers family revocation, you have built a logout cannon.

Schema Design Under GDPR

Before the code, the storage. Refresh tokens tie directly to an identifiable user, so under GDPR they are personal data and everything attached to them inherits that classification. Two decisions followed from that:

Store hashes, not tokens. The database holds SHA-256 of the token, never the token itself. A database dump — from a backup leak, from an SQL injection — yields nothing usable. We use SHA-256 rather than bcrypt/argon2 here because refresh tokens are 256 bits of CSPRNG output, not user-chosen passwords: there is no dictionary to attack, so the slow-hash cost buys nothing and would add latency to every refresh.

Do not store IP addresses in the token table. Our first draft logged ip_address per token for anomaly detection. Under Article 6 that needs a legitimate-interest assessment, and it means refresh tokens become part of any Article 15 subject access request. We replaced it with a coarse client_region (two-letter country, derived at issue time from the Cloudflare CF-IPCountry header) and a user_agent_hash. Region granularity is enough to notice a Berlin session suddenly refreshing from another continent, without retaining an identifier that pinpoints a household.

CREATE TABLE refresh_tokens (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    family_id       TEXT    NOT NULL,
    token_hash      TEXT    NOT NULL UNIQUE,
    user_id         INTEGER NOT NULL,
    issued_at       INTEGER NOT NULL,
    expires_at      INTEGER NOT NULL,
    consumed_at     INTEGER,
    replaced_by     TEXT,
    client_region   TEXT,
    user_agent_hash TEXT,
    revoked_at      INTEGER,
    revoke_reason   TEXT
);

CREATE INDEX idx_rt_family  ON refresh_tokens(family_id);
CREATE INDEX idx_rt_user    ON refresh_tokens(user_id, expires_at);
CREATE INDEX idx_rt_cleanup ON refresh_tokens(expires_at);

CREATE TABLE token_families (
    family_id     TEXT PRIMARY KEY,
    user_id       INTEGER NOT NULL,
    created_at    INTEGER NOT NULL,
    revoked_at    INTEGER,
    revoke_reason TEXT
);

The token_families table exists so revocation is a single-row write. Revoking a family by updating every member row is O(chain length) and, worse, races with an in-flight refresh appending a new row to that same chain. One row, one write, checked on every refresh.

On SQLite specifically: WAL mode is not optional for this workload. Refresh requests are write transactions, and in rollback-journal mode a single writer blocks all readers — meaning every video-metadata read on the site would stall behind token writes. With WAL, readers proceed against the last committed snapshot while a writer appends. We also set busy_timeout so concurrent writers back off instead of returning SQLITE_BUSY immediately.

<?php
declare(strict_types=1);

final class TokenStore
{
    public function __construct(private \PDO $db) {}

    public static function open(string $path): self
    {
        $db = new \PDO('sqlite:' . $path, null, null, [
            \PDO::ATTR_ERRMODE            => \PDO::ERRMODE_EXCEPTION,
            \PDO::ATTR_DEFAULT_FETCH_MODE => \PDO::FETCH_ASSOC,
        ]);

        $db->exec('PRAGMA journal_mode = WAL');
        $db->exec('PRAGMA synchronous = NORMAL');
        $db->exec('PRAGMA busy_timeout = 5000');
        $db->exec('PRAGMA foreign_keys = ON');

        return new self($db);
    }

    public function pdo(): \PDO
    {
        return $this->db;
    }
}

synchronous = NORMAL is safe under WAL — a power loss can lose the last commits but cannot corrupt the database. For refresh tokens, losing the final few milliseconds of writes means at worst a client re-authenticates.

The Grace Window

Here is the mechanism that fixed our logout bug.

When a refresh token is consumed, we do not delete it. We stamp consumed_at and record replaced_by — the hash of the successor token. If a consumed token is presented again, we look at how long ago it was consumed:

Within the grace window (we use 30 seconds): this is almost certainly a retry or a concurrent request from the same client. Return the successor token that was already issued, rather than minting another. The family stays alive. Crucially, we return the same successor, not a new one — otherwise two racing requests each get a different valid token and the chain forks.
Outside the grace window: a token consumed 10 minutes ago being replayed now is not a network retry. Revoke the whole family.

The grace window is a security trade-off, and it should be stated plainly: an attacker who steals a refresh token and replays it within 30 seconds of legitimate use gets the same successor the real client got, and reuse detection does not fire. In exchange, every honest retry stops logging people out. Thirty seconds is longer than any realistic mobile retry chain and short enough that an attacker would need to be replaying in near-real-time — which implies they already have live access to the device or the transport, a strictly worse compromise that token rotation was never going to solve.

We additionally tighten it: within the grace window we require the user_agent_hash to match the one recorded at issue time. A retry from the same client has the same UA; a replay from an attacker's tooling usually does not.

Implementing the Refresh Endpoint

The whole operation must be atomic. The check-then-write sequence — read the token row, verify it is unconsumed, mark it consumed, insert the successor — is a classic TOCTOU if another request interleaves. In SQLite we wrap it in BEGIN IMMEDIATE, which acquires the write lock at transaction start rather than at first write, so two concurrent refreshes serialize instead of one discovering a conflict at commit time.

<?php
declare(strict_types=1);

final class RefreshTokenService
{
    private const GRACE_SECONDS      = 30;
    private const REFRESH_TTL        = 60 * 60 * 24 * 30; // 30 days
    private const ACCESS_TTL         = 900;               // 15 minutes

    public function __construct(
        private \PDO $db,
        private AccessTokenIssuer $access,
        private \Closure $now,
    ) {}

    /**
     * @return array{access_token:string,refresh_token:string,expires_in:int}
     * @throws TokenRejected
     */
    public function rotate(string $presented, string $uaHash, string $region): array
    {
        $hash = hash('sha256', $presented);
        $now  = ($this->now)();

        $this->db->beginTransaction();          // PDO maps this to BEGIN
        $this->db->exec('ROLLBACK; BEGIN IMMEDIATE');

        try {
            $row = $this->fetchToken($hash);

            if ($row === null) {
                throw new TokenRejected('unknown_token');
            }

            if ($this->familyRevoked($row['family_id'])) {
                throw new TokenRejected('family_revoked');
            }

            if ($row['revoked_at'] !== null) {
                throw new TokenRejected('token_revoked');
            }

            if ($row['expires_at'] <= $now) {
                throw new TokenRejected('token_expired');
            }

            if ($row['consumed_at'] !== null) {
                $result = $this->handleReuse($row, $uaHash, $now);
                $this->db->commit();
                return $result;
            }

            $result = $this->issueSuccessor($row, $uaHash, $region, $now);
            $this->db->commit();
            return $result;
        } catch (\Throwable $e) {
            if ($this->db->inTransaction()) {
                $this->db->rollBack();
            }
            throw $e;
        }
    }

    private function handleReuse(array $row, string $uaHash, int $now): array
    {
        $age       = $now - (int) $row['consumed_at'];
        $sameAgent = hash_equals((string) $row['user_agent_hash'], $uaHash);

        if ($age > self::GRACE_SECONDS || !$sameAgent) {
            $this->revokeFamily($row['family_id'], 'reuse_detected', $now);
            throw new TokenRejected('reuse_detected');
        }

        // In-grace retry: hand back the successor that was already minted.
        $successor = $this->fetchToken((string) $row['replaced_by']);

        if ($successor === null || $successor['revoked_at'] !== null) {
            throw new TokenRejected('successor_unavailable');
        }

        // We cannot return the plaintext successor from its hash, so the
        // successor's plaintext is cached for exactly the grace window.
        $plain = $this->graceCache->get((string) $row['replaced_by']);

        if ($plain === null) {
            throw new TokenRejected('grace_expired');
        }

        return [
            'access_token'  => $this->access->issue((int) $row['user_id']),
            'refresh_token' => $plain,
            'expires_in'    => self::ACCESS_TTL,
        ];
    }

    private function issueSuccessor(array $row, string $uaHash, string $region, int $now): array
    {
        $plain   = bin2hex(random_bytes(32));
        $newHash = hash('sha256', $plain);

        $this->db->prepare(
            'UPDATE refresh_tokens
                SET consumed_at = ?, replaced_by = ?
              WHERE token_hash = ? AND consumed_at IS NULL'
        )->execute([$now, $newHash, $row['token_hash']]);

        $this->db->prepare(
            'INSERT INTO refresh_tokens
                 (family_id, token_hash, user_id, issued_at, expires_at,
                  client_region, user_agent_hash)
             VALUES (?, ?, ?, ?, ?, ?, ?)'
        )->execute([
            $row['family_id'], $newHash, $row['user_id'],
            $now, $now + self::REFRESH_TTL, $region, $uaHash,
        ]);

        $this->graceCache->put($newHash, $plain, self::GRACE_SECONDS);

        return [
            'access_token'  => $this->access->issue((int) $row['user_id']),
            'refresh_token' => $plain,
            'expires_in'    => self::ACCESS_TTL,
        ];
    }

    private function revokeFamily(string $familyId, string $reason, int $now): void
    {
        $this->db->prepare(
            'UPDATE token_families SET revoked_at = ?, revoke_reason = ?
              WHERE family_id = ? AND revoked_at IS NULL'
        )->execute([$now, $reason, $familyId]);
    }
}

A note on $this->graceCache: the grace path needs to return a plaintext token, but the database only stores hashes — by design. So the successor's plaintext lives in a short-TTL cache (APCu on a single node; Redis or a Cloudflare Workers KV namespace if you run multiple origins) keyed by its hash, expiring exactly at the grace boundary. This is the one place plaintext refresh tokens exist server-side, which is why the TTL is measured in seconds and the store is memory-only, never on disk.

If you find that unacceptable, the alternative is to skip returning a token on the grace path and return only a fresh access token with a signal telling the client to keep its current refresh token. That is cleaner cryptographically but requires client cooperation, and we had shipped mobile clients we could not change.

Access Token Issuance

The access token itself is a short-lived JWT. Two details matter for a video API.

First, use EdDSA (Ed25519) or ES256 rather than HS256 if anything other than your auth service needs to verify tokens. We verify JWTs at the edge in a Cloudflare Worker before requests ever reach the origin — a symmetric key would mean shipping the signing secret to the edge, where a Worker compromise becomes a token-forgery capability. With Ed25519 the Worker holds only the public key.

Second, include a fid (family ID) claim. When a family is revoked, the edge can drop access tokens from that family immediately instead of waiting up to 15 minutes for natural expiry, by checking a small revocation set in KV.

import time
import json
import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


def _b64(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def issue_access_token(
    private_key: Ed25519PrivateKey,
    key_id: str,
    user_id: int,
    family_id: str,
    ttl: int = 900,
) -> str:
    now = int(time.time())

    header = {"alg": "EdDSA", "typ": "JWT", "kid": key_id}
    payload = {
        "sub": str(user_id),
        "fid": family_id,
        "iat": now,
        "nbf": now,
        "exp": now + ttl,
        "iss": "https://viralvidvault.com",
        "aud": "vvv-video-api",
    }

    signing_input = "{}.{}".format(
        _b64(json.dumps(header, separators=(",", ":")).encode()),
        _b64(json.dumps(payload, separators=(",", ":")).encode()),
    )
    signature = private_key.sign(signing_input.encode())

    return f"{signing_input}.{_b64(signature)}"

Keep the payload minimal. Every claim is bytes on every request, and for a video API that means bytes on every thumbnail-metadata call and every playback-position update. It also means anything you put in there is readable by anyone who intercepts the token — no email addresses, no display names, no region data. Under GDPR, a JWT carrying personal data is personal data in transit and in every client-side log that captures request headers.

Client-Side Refresh Coordination

Server-side grace handles duplicates, but the better outcome is not producing them. Clients should coalesce concurrent refreshes into a single in-flight request.

package auth

import (
    "context"
    "sync"
    "time"
)

type TokenSet struct {
    Access    string
    Refresh   string
    ExpiresAt time.Time
}

type Client struct {
    mu      sync.Mutex
    current TokenSet
    flight  *refreshCall
    refresh func(ctx context.Context, token string) (TokenSet, error)
}

type refreshCall struct {
    done chan struct{}
    set  TokenSet
    err  error
}

// Token returns a valid access token, coalescing concurrent refreshes
// so that only one network round trip happens per expiry.
func (c *Client) Token(ctx context.Context) (string, error) {
    c.mu.Lock()

    if time.Until(c.current.ExpiresAt) > 30*time.Second {
        tok := c.current.Access
        c.mu.Unlock()
        return tok, nil
    }

    if c.flight != nil {
        call := c.flight
        c.mu.Unlock()
        select {
        case <-call.done:
            return call.set.Access, call.err
        case <-ctx.Done():
            return "", ctx.Err()
        }
    }

    call := &refreshCall{done: make(chan struct{})}
    c.flight = call
    stale := c.current.Refresh
    c.mu.Unlock()

    call.set, call.err = c.refresh(context.WithoutCancel(ctx), stale)

    c.mu.Lock()
    if call.err == nil {
        c.current = call.set
    }
    c.flight = nil
    c.mu.Unlock()

    close(call.done)
    return call.set.Access, call.err
}

Three things worth copying from this:

The refresh triggers 30 seconds before expiry, not after a 401. Refreshing reactively means at least one failed request per cycle, and on a video app that is a visible stall.
context.WithoutCancel on the refresh call. If the caller that happened to win the race gets cancelled — user navigates away mid-request — the refresh must still complete, because other goroutines are waiting on it and, more importantly, because a cancelled-mid-flight rotation leaves the client holding a token the server has already consumed.
Waiters block on a channel rather than each issuing their own request. This is the single highest-value change we made; it eliminated roughly 90% of duplicate refreshes before they reached the server.

Cleanup and Monitoring

Rotation generates rows. At 15-minute access tokens and a 30-day refresh TTL, an active daily user produces dozens of rows per day. Two jobs keep it bounded:

Expired-token purge. Delete rows where expires_at < now - 86400. Keeping one extra day lets you investigate a reuse alert after the fact. Run it from cron, not from the request path.
Revoked-family purge. Delete families revoked more than 30 days ago, plus their tokens. Under GDPR storage limitation, retaining revoked credentials indefinitely has no justification.

After large deletes on SQLite, run PRAGMA incremental_vacuum — configure auto_vacuum = INCREMENTAL at database creation so you can reclaim pages without the full-rewrite stall of VACUUM.

For monitoring, the metric that actually matters is the reuse-detection rate. A healthy system sits near zero. We alert when it exceeds 0.1% of refreshes over an hour. In our experience every spike has had a mundane cause — a client build with broken coalescing, a load balancer replaying requests — but the one time it does not, it is a credential-theft signal you want within the hour, not at the next quarterly review.

Also track grace-window hits separately from reuse detections. A rising grace-hit rate means client coalescing is degrading, which is a bug worth fixing even though users never notice it.

What We Would Do Differently

If we were starting over: bind refresh tokens to a client-held key (DPoP, RFC 9449) from day one. Rotation plus reuse detection makes theft detectable; proof-of-possession makes a stolen token useless, because the attacker cannot produce the signature over the request that the token requires. Rotation is then a defence-in-depth layer rather than the primary control, and the grace-window trade-off stops mattering because a replayed token without the private key fails regardless.

We did not do this because our oldest mobile clients had no secure key storage path we trusted, and shipping a breaking auth change to installed apps is a months-long migration. That is a real constraint, not an excuse — but if your clients are new, skip our intermediate step.

Conclusion

The distance between "rotate refresh tokens" as advice and rotation that does not log people out is mostly concurrency handling. The pieces that made it work for us: a family ID so revocation is one write, consumed_at and replaced_by so a spent token still carries the information needed to answer a retry, a short grace window with a user-agent check, BEGIN IMMEDIATE so the check-and-rotate is genuinely atomic under WAL, and client-side coalescing to stop most duplicates ever reaching the server. Store hashes, keep the JWT payload empty of anything identifying, and put a purge job on a cron.

Our German logout reports went to zero and stayed there, and we kept the property that mattered: if someone steals a refresh token, we find out within one refresh cycle.

Postgres LISTEN/NOTIFY for Real-Time Video Metadata Invalidation at Scale

ahmet gedik — Tue, 18 Aug 2026 00:00:02 +0000

Our trending video ingest runs every four hours across nine Asia-Pacific regions — US, GB, JP, KR, TW, SG, VN, TH, HK. Each run rewrites view counts, titles, and channel metadata for roughly 40,000 rows. The problem was never the write. The problem was that a Japanese video whose title got corrected from a mojibake'd ã‚¢ãƒ‹ãƒ¡ to アニメ would keep serving the broken title from three separate cache layers for up to six hours, because nothing downstream knew the row had changed.

We were doing what most people do: TTL everything and hope. Watch pages cached six hours, category listings three hours, the CJK search index rebuilt on a cron. When a correction landed, the only honest answer to "when will users see this?" was "eventually." For a site serving Korean and Traditional Chinese queries where a single tokenization fix changes whether a video is findable at all, "eventually" is a bug. This is how we replaced blind TTLs with Postgres LISTEN/NOTIFY on the metadata master, and what broke along the way. If you want to see the read side of this in production, it's TopVideoHub.

Why the Metadata Master Is Postgres and the Edges Are Not

Worth being explicit about the architecture, because it explains why LISTEN/NOTIFY earns its place here.

Each regional edge node runs PHP 8.4 behind LiteSpeed, with a local SQLite database using FTS5 and a custom CJK tokenizer for search. SQLite on the edge is not a compromise — it is the whole point. A search query for 아이돌 무대 hits a local file on the same box, no network round trip, and FTS5 with a bigram tokenizer handles CJK segmentation without a separate search cluster. Cloudflare sits in front, caching HTML by URL.

But SQLite has no pub/sub. It has no way to tell anyone that a row changed. So the ingest pipeline writes to a central Postgres instance — the metadata master — and edges pull from it. The question is how edges learn when to pull.

The options we considered:

Polling a updated_at cursor. Works, but at nine regions × poll interval you either burn queries or accept latency. We started here.
A message broker (NATS, Redis Streams, Kafka). Correct at large scale, but it means another stateful service to run, monitor, and pay for, on a system whose entire appeal is that it has few moving parts.
LISTEN/NOTIFY. The database we already have, already holds the truth, and already knows the exact instant a row changed. No new infrastructure.

For our write volume — bursty, ~40k rows every four hours, near-zero between runs — LISTEN/NOTIFY was the right size of hammer. It would not be at 10k writes/sec. I'll cover where it falls over later.

The Trigger That Emits Change Events

The naive version of this is a trigger that fires pg_notify per row. That version will hurt you, and I'll show why in a moment, but start with it because the mechanics are clearer.

CREATE TABLE video_metadata (
  video_id      TEXT PRIMARY KEY,
  region        TEXT NOT NULL,
  title         TEXT NOT NULL,
  channel_id    TEXT NOT NULL,
  view_count    BIGINT NOT NULL DEFAULT 0,
  lang          TEXT NOT NULL DEFAULT 'en',
  updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE OR REPLACE FUNCTION notify_video_change() RETURNS trigger AS $$
DECLARE
  payload JSON;
BEGIN
  -- Only notify on changes that actually affect rendered output.
  -- view_count churns constantly and nobody cares about exact numbers.
  IF TG_OP = 'UPDATE'
     AND NEW.title = OLD.title
     AND NEW.channel_id = OLD.channel_id
     AND NEW.lang = OLD.lang THEN
    RETURN NEW;
  END IF;

  payload := json_build_object(
    'op',       TG_OP,
    'video_id', NEW.video_id,
    'region',   NEW.region,
    'lang',     NEW.lang,
    'reindex',  (TG_OP = 'INSERT' OR NEW.title IS DISTINCT FROM OLD.title)
  );

  PERFORM pg_notify('video_meta', payload::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER video_metadata_notify
  AFTER INSERT OR UPDATE ON video_metadata
  FOR EACH ROW EXECUTE FUNCTION notify_video_change();

Two things in there matter more than they look.

First, the early RETURN NEW when only view_count changed. Our ingest updates view counts on every row, every run. Without that guard, a routine refresh emits 40,000 notifications and invalidates every cached page on the site for no user-visible reason. Filtering at the trigger — not at the consumer — is what keeps this viable.

Second, reindex is computed in the trigger rather than inferred downstream. Only a title change requires an FTS5 rebuild for that row, and FTS5 rebuilds with a CJK tokenizer are the expensive part. Telling the consumer exactly what work is needed beats making it guess.

The Payload Size Trap

pg_notify payloads are capped at 8000 bytes. Exceed it and the transaction fails, not just the notification. That means an oversized payload rolls back the write that triggered it.

We hit this the obvious way: someone suggested including the full row in the payload to save the consumer a query. Fine for English titles. A Japanese title with an emoji-laden channel description encodes to several times the bytes you'd estimate from character count, since UTF-8 spends 3 bytes per CJK codepoint and 4 per emoji. The payload passed 8000 bytes on a long Vietnamese description with tone marks and took the ingest run down with it.

The fix is the discipline you should adopt from the start: notifications carry identity, never content. The payload says what changed; the consumer queries for the current value. This is also more correct — if three updates land before the consumer catches up, fetching current state gives you the latest, whereas replaying three content payloads makes you apply stale data in order.

The Listener

Here's the consumer that runs on each edge node. Python, because the ingest tooling is already Python and psycopg's async notification support is the cleanest of the drivers I've used.

import json
import select
import sqlite3
import time
import logging

import psycopg

log = logging.getLogger("meta-listener")

BATCH_WINDOW = 2.0      # seconds to coalesce before flushing
MAX_BATCH = 500


def drain(conn, deadline):
    """Collect notifications until deadline, deduped by video_id."""
    pending = {}
    while time.monotonic() < deadline and len(pending) < MAX_BATCH:
        timeout = max(0.0, deadline - time.monotonic())
        if not select.select([conn], [], [], timeout)[0]:
            continue
        conn.execute("SELECT 1")  # force libpq to consume input
        for note in conn.notifies(timeout=0):
            try:
                msg = json.loads(note.payload)
            except json.JSONDecodeError:
                log.warning("bad payload: %r", note.payload[:120])
                continue
            prev = pending.get(msg["video_id"])
            # sticky reindex: if any event asked for it, we reindex
            msg["reindex"] = msg["reindex"] or (prev or {}).get("reindex", False)
            pending[msg["video_id"]] = msg
    return list(pending.values())


def run(dsn, sqlite_path, region):
    backoff = 1
    while True:
        try:
            with psycopg.connect(dsn, autocommit=True) as conn:
                conn.execute("LISTEN video_meta")
                log.info("listening on video_meta")
                backoff = 1
                resync(conn, sqlite_path, region)  # catch up on missed window
                while True:
                    batch = drain(conn, time.monotonic() + BATCH_WINDOW)
                    if batch:
                        apply_batch(conn, sqlite_path, region, batch)
        except (psycopg.OperationalError, psycopg.InterfaceError) as exc:
            log.error("connection lost: %s; retry in %ss", exc, backoff)
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)

The resync call on every reconnect is not optional. This is the single most important thing to understand about LISTEN/NOTIFY, and I'll give it its own section.

Notifications Are Not a Queue

If your listener is disconnected when NOTIFY fires, that notification is gone. Permanently. There is no replay, no offset, no acknowledgment. Postgres delivers to currently-connected listeners and forgets.

That means every deploy, every network blip, every Postgres restart, every connection reaped by a proxy's idle timeout is a hole in your data. On a system where the edges are in Singapore and Tokyo and the master is somewhere else, connections drop more than you'd like.

The correct mental model: NOTIFY is a latency optimization on top of polling, not a replacement for it. You still need a cursor-based catch-up. NOTIFY just makes the common case fast.

def resync(pg, sqlite_path, region):
    """Cursor-based catch-up. Runs on connect and on a slow timer."""
    db = sqlite3.connect(sqlite_path)
    row = db.execute(
        "SELECT value FROM sync_state WHERE key = 'meta_cursor'"
    ).fetchone()
    cursor = row[0] if row else "1970-01-01T00:00:00Z"

    # Overlap by 30s: updated_at is set at statement time, but the row becomes
    # visible at commit time. A long transaction can commit rows whose
    # updated_at is older than a cursor we already advanced past.
    rows = pg.execute(
        """
        SELECT video_id, title, channel_id, lang, view_count, updated_at
          FROM video_metadata
         WHERE region = %s
           AND updated_at > %s::timestamptz - interval '30 seconds'
         ORDER BY updated_at
         LIMIT 5000
        """,
        (region, cursor),
    ).fetchall()

    if not rows:
        return 0

    with db:
        for vid, title, chan, lang, views, updated in rows:
            db.execute(
                """INSERT INTO videos (video_id, title, channel_id, lang, view_count)
                   VALUES (?, ?, ?, ?, ?)
                   ON CONFLICT(video_id) DO UPDATE SET
                     title = excluded.title,
                     channel_id = excluded.channel_id,
                     lang = excluded.lang,
                     view_count = excluded.view_count""",
                (vid, title, chan, lang, views),
            )
            db.execute("DELETE FROM videos_fts WHERE video_id = ?", (vid,))
            db.execute(
                "INSERT INTO videos_fts (video_id, title_tok) VALUES (?, ?)",
                (vid, tokenize_cjk(title, lang)),
            )
        db.execute(
            """INSERT INTO sync_state (key, value) VALUES ('meta_cursor', ?)
               ON CONFLICT(key) DO UPDATE SET value = excluded.value""",
            (rows[-1][5].isoformat(),),
        )
    log.info("resync applied %d rows", len(rows))
    return len(rows)

The 30-second overlap deserves emphasis. updated_at DEFAULT now() uses transaction start time, and rows become visible to other sessions only at commit. A transaction that starts at 10:00:00 and commits at 10:00:20 writes rows stamped 10:00:00 that no one could see until 10:00:20. If your listener advanced its cursor to 10:00:10 in the meantime, those rows are invisible forever. Overlapping means you reprocess a few rows — harmless, since the upsert is idempotent — instead of silently losing them. If you want to eliminate the window rather than paper over it, use a monotonic sequence assigned at commit time, but the overlap is far simpler and has been sufficient for us.

We also run resync on a 10-minute timer regardless of connection health. Belt and braces.

Wiring It to PHP and Cloudflare

Once the local SQLite is current, two caches still hold stale HTML: LiteSpeed's page cache and Cloudflare's edge cache. The listener shells out to a small PHP invalidator rather than reimplementing the cache-key logic in Python — the URL structure lives in the PHP router and I'm not maintaining two copies of it.

<?php
declare(strict_types=1);

final class MetadataInvalidator
{
    private const CF_API = 'https://api.cloudflare.com/client/v4/zones/%s/purge_cache';
    private const CF_BATCH = 30;

    public function __construct(
        private readonly string $zoneId,
        private readonly string $apiToken,
        private readonly string $baseUrl,
        private readonly string $lscacheDir,
    ) {}

    /** @param list<array{video_id:string, channel_id:string}> $changes */
    public function invalidate(array $changes): void
    {
        $urls = [];
        $channels = [];

        foreach ($changes as $c) {
            $urls[] = sprintf('%s/watch/%s', $this->baseUrl, $c['video_id']);
            $channels[$c['channel_id']] = true;
        }
        foreach (array_keys($channels) as $channelId) {
            $urls[] = sprintf('%s/channel/%s', $this->baseUrl, $channelId);
        }

        $this->purgeLiteSpeed($urls);

        foreach (array_chunk(array_unique($urls), self::CF_BATCH) as $chunk) {
            $this->purgeCloudflare($chunk);
        }
    }

    private function purgeLiteSpeed(array $urls): void
    {
        foreach ($urls as $url) {
            $key = md5(parse_url($url, PHP_URL_PATH) ?? '');
            $file = sprintf('%s/%s/%s.html', $this->lscacheDir, substr($key, 0, 2), $key);
            if (is_file($file)) {
                @unlink($file);
            }
        }
    }

    private function purgeCloudflare(array $chunk): void
    {
        $ch = curl_init(sprintf(self::CF_API, $this->zoneId));
        curl_setopt_array($ch, [
            CURLOPT_POST           => true,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT        => 10,
            CURLOPT_HTTPHEADER     => [
                'Authorization: Bearer ' . $this->apiToken,
                'Content-Type: application/json',
            ],
            CURLOPT_POSTFIELDS => json_encode(['files' => array_values($chunk)], JSON_THROW_ON_ERROR),
        ]);

        $body = curl_exec($ch);
        $code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        curl_close($ch);

        if ($code !== 200) {
            error_log(sprintf('[invalidator] CF purge failed http=%d body=%s', $code, substr((string) $body, 0, 300)));
        }
    }
}

A few notes from operating this:

Cloudflare's purge-by-URL endpoint takes 30 URLs per call on the plan we're on, hence the chunking. Sending more gets the whole batch rejected, not truncated.
Purge by URL, not by tag or prefix, unless you're on Enterprise. Tag-based purging is the feature you want and the one you probably can't buy.
Failed purges are not fatal. Worst case the TTL expires normally, which is exactly where we were before. Log and move on; do not retry aggressively into a rate limit.
The LiteSpeed cache path layout depends on your config. Verify yours instead of copying that md5-prefix scheme blindly; ours is a two-character shard directory.

Coalescing Is Where the Wins Are

The 2-second batch window in drain() is the difference between this design working and this design being a self-inflicted denial of service.

During a regional ingest run, a popular Japanese channel might have 40 videos updated within the same second. Without coalescing that is 40 SQLite transactions, 40 FTS5 rebuilds, and — worse — 40 separate Cloudflare API calls that include the same /channel/{id} URL 40 times.

With coalescing, the dictionary in drain() dedupes by video_id, and invalidate() dedupes channel URLs via the associative array. One batch, one transaction, one or two API calls.

Measured on our four-hour ingest for the JP region:

Before: ~1,100 notifications → ~1,100 purge requests, several minutes of sustained API traffic, frequent 429s.
After: ~1,100 notifications → 3 batches → 6 purge API calls. Total invalidation wall time under 4 seconds.

The reindex flag being sticky across coalesced events matters too. If the first event for a video was a channel change (reindex: false) and the second was a title fix (reindex: true), naive last-write-wins on the dict would be fine — but reverse the order and you'd drop the reindex. Hence the explicit or when merging.

Where This Design Stops Working

Being honest about the limits, because this pattern gets recommended past its range:

NOTIFY serializes on commit. All notifications from a transaction are delivered at commit, and Postgres takes a lock on the notification queue. At high concurrent write rates this becomes a measurable contention point. Our write pattern is bursty-but-single-writer, so we never see it. A multi-tenant OLTP system with hundreds of concurrent committers would.
The 8GB queue limit. If a listener connection is alive but not draining, the queue grows, and at 8GB (max_notify_queue_size in recent versions) committing transactions start failing. A hung listener can take down writes. Add a monitor on pg_notification_queue_usage().
No fan-out control. Every listener on a channel gets every message. We filter by region in the consumer, which means the Tokyo node receives and discards notifications for São Paulo. At nine regions that's fine. At ninety, use per-region channel names (video_meta_jp) and LISTEN selectively.
Connection pooling breaks it. PgBouncer in transaction mode will silently destroy LISTEN — the session-level state does not survive the connection being handed to another client. Use session mode, or a direct connection for the listener. This costs people days.

Conclusion

The change in user-visible behavior is the part worth stating plainly: a title correction that used to take up to six hours to reach a reader in Seoul now takes under five seconds end to end, including the FTS5 reindex that makes the corrected title searchable.

What made it work was not the LISTEN/NOTIFY mechanism itself, which is about fifteen lines of SQL. It was the three things around it: filtering in the trigger so routine view-count churn does not invalidate the world, coalescing in a short window so bursts collapse into single operations, and — most importantly — keeping a cursor-based resync underneath, because notifications are fire-and-forget and any design that treats them as a durable queue will lose data the first time a connection drops.

If you're running TTL-based invalidation today and the staleness is starting to hurt, this is a low-infrastructure step up. Just build the polling fallback first and add NOTIFY on top of it, not the other way around.

Building a GraphQL Video Discovery API with Strawberry and FastAPI over SQLite

ahmet gedik — Mon, 17 Aug 2026 17:00:02 +0000

Our Android client painted one search screen with four HTTP requests: GET /api/search?q=, then GET /api/channels?ids=, then GET /api/categories, then a per-video GET /api/video/{id} as soon as the user tapped a card. Every one of those endpoints returned the whole row out of SQLite — 41 columns for a video — and the client rendered six of them. On a throttled 3G profile the screen took 2.4 s at p95 to first meaningful paint, and about 180 KB of JSON crossed the wire to draw something that genuinely needs 20 KB. That is the problem I set out to fix at DailyWatch, and the fix was a read-only GraphQL sidecar built with Strawberry and FastAPI, pointed at the same SQLite file the PHP site already reads.

This is the write-up of what that actually took: schema design against an FTS5 index, keyset pagination over bm25 scores, killing the N+1 with DataLoader, cost limits, and the part nobody blogs about — making a POST-shaped protocol cacheable at the CDN edge.

Four round trips, six fields used

The REST surface grew the way REST surfaces grow. /api/search was written first and returned video rows. Then the client needed channel names next to each result, so we added /api/channels?ids=a,b,c — which is a hand-rolled DataLoader with no schema, no batching contract, and a URL length limit. Then a category filter needed the category list. The endpoints were individually fine; the composition was the problem.

The concrete costs, measured rather than guessed:

Over-fetching. The videos table carries description text, raw region codes, fetch timestamps, and moderation flags. The card UI needs id, title, duration, view count, thumbnail, and channel title.
Latency chaining. Request 2 cannot start until request 1 returns the channel IDs. Two serial RTTs on a 250 ms mobile connection is half a second before any pixel moves.
Cache fragmentation. Four responses, four TTLs, four invalidation paths. When the cron job refreshed trending videos, one of the four consistently went stale in a way that made the UI look broken.
No contract. Adding a field to the search response broke an older client that did strict JSON decoding. There was no way to know which fields were actually in use.

The site itself is a PHP 8.4 monolith served through LiteSpeed with a page cache in front, and it is genuinely fast — server-rendered HTML, no hydration, SQLite reads in single-digit milliseconds. Nothing about that needed to change. The apps were the ones paying for a data model shaped around HTML pages.

Why a Python sidecar and not GraphQL inside the PHP app

I considered bolting a GraphQL layer onto the monolith. There are competent PHP implementations. I chose a separate service anyway, and the reasoning was mostly operational rather than aesthetic:

The read model is already a file. SQLite means a second process can open the same database read-only, in WAL mode, with zero coordination and zero additional infrastructure. No replica, no connection pool, no network hop to the data.
Blast radius. If the GraphQL service falls over, the website keeps serving HTML. Only the apps degrade, and they degrade to a cached view. A GraphQL layer inside the monolith shares a fatal error handler with the thing that pays the bills.
Strawberry is code-first with real type hints. The schema is generated from Python types, so pyright catches a resolver returning the wrong shape at build time. An SDL file in a repo drifts from its resolvers within about two sprints; a dataclass cannot.
Async fits the access pattern. A GraphQL request here is many small indexed reads, not one big query. That maps cleanly onto batching.

The honest cost of the sidecar: one more deploy target, one more origin behind Cloudflare, one more thing to monitor, and a hard invariant that this service never writes. I enforce that last one with PRAGMA query_only = 1 on every connection rather than trusting myself.

The first thing that had to be shared, not forked, was full-text query construction. Here is the PHP side that already existed — the FTS5 tokenizer sanitizer, plus the query it feeds:

<?php
declare(strict_types=1);

final class FtsQuery
{
    private const int MAX_TOKENS = 8;

    /** Turn arbitrary user input into a safe FTS5 MATCH expression. */
    public static function fromUserInput(string $raw): ?string
    {
        // FTS5 reads " * : ^ - AND OR NOT as syntax. Users type them as text.
        $clean = preg_replace('/[^\p{L}\p{N}\s]+/u', ' ', mb_strtolower(trim($raw))) ?? '';
        $tokens = array_slice(
            array_values(array_filter(explode(' ', $clean), strlen(...))),
            0,
            self::MAX_TOKENS
        );
        if ($tokens === []) {
            return null;
        }

        $last = array_key_last($tokens);
        foreach ($tokens as $i => $t) {
            // Prefix-match only the trailing token: 'coffee gr' -> "coffee" AND "gr"*
            $tokens[$i] = ($i === $last && mb_strlen($t) >= 2)
                ? '"' . $t . '"*'
                : '"' . $t . '"';
        }
        return implode(' AND ', $tokens);
    }
}

$pdo = new PDO('sqlite:' . __DIR__ . '/../data/app.db', options: [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
]);
$pdo->exec('PRAGMA query_only = 1');

$match = FtsQuery::fromUserInput($_GET['q'] ?? '');
if ($match === null) {
    echo json_encode(['items' => []]);
    exit;
}

// Title weighted 10x, channel title 3x, description 1x. bm25 is negative: lower = better.
$stmt = $pdo->prepare(<<<SQL
    SELECT v.id, v.title, v.duration_seconds, v.view_count,
           bm25(videos_fts, 10.0, 3.0, 1.0) AS score
    FROM videos_fts
    JOIN videos v ON v.rowid = videos_fts.rowid
    WHERE videos_fts MATCH :m
    ORDER BY score, v.rowid
    LIMIT 20
SQL);
$stmt->execute([':m' => $match]);
echo json_encode(['items' => $stmt->fetchAll()], JSON_UNESCAPED_UNICODE);

That sanitizer had to be reimplemented in Python byte-for-byte, and getting it wrong means the app and the website return different results for the same query. I will come back to how I pinned that down.

The schema is the contract, not the table

The temptation with code-first GraphQL is to mirror your tables. Resist it. The schema should describe what clients need; the repository layer translates. Two concrete decisions here: thumbnail is a computed field with a width argument rather than four stored URL columns, and channel is a real object resolved lazily so a query that does not ask for it costs nothing.

# schema.py
from __future__ import annotations

from datetime import datetime

import strawberry
from strawberry.types import Info


@strawberry.type
class Channel:
    id: str
    title: str
    subscriber_count: int | None
    thumbnail_url: str | None


@strawberry.type
class Video:
    id: str
    title: str
    published_at: datetime
    duration_seconds: int
    view_count: int
    region: str
    # Never leaves the process; used only to feed the loader.
    channel_id: strawberry.Private[str]

    @strawberry.field(description='Uploading channel. Batched across the whole request.')
    async def channel(self, info: Info) -> Channel | None:
        return await info.context['loaders']['channel'].load(self.channel_id)

    @strawberry.field(description='Thumbnail URL sized for the requested render width.')
    def thumbnail(self, width: int = 320) -> str:
        size = 'maxresdefault' if width > 480 else 'hqdefault'
        return f'https://i.ytimg.com/vi/{self.id}/{size}.jpg'


@strawberry.type
class VideoEdge:
    cursor: str
    node: Video


@strawberry.type
class VideoConnection:
    edges: list[VideoEdge]
    has_next_page: bool


@strawberry.type
class Query:
    @strawberry.field
    async def search(
        self,
        info: Info,
        q: str,
        first: int = 20,
        after: str | None = None,
        region: str | None = None,
    ) -> VideoConnection:
        first = max(1, min(first, 50))
        return await info.context['repo'].search(q=q, first=first, after=after, region=region)

    @strawberry.field
    async def channel(self, info: Info, id: str) -> Channel | None:
        return await info.context['loaders']['channel'].load(id)

strawberry.Private[str] is doing real work: it keeps channel_id on the Python object for the resolver while excluding it from the generated schema. Clients cannot query it, so I am free to change how channels are keyed later.

Capping first in the resolver rather than trusting the client is not paranoia. It is the difference between a request that reads 20 rows and one that reads 10,000 because someone typo'd a zero.

Making FTS5 behave under async

SQLite is synchronous. aiosqlite hides that behind a dedicated thread per connection, which means one connection serializes its reads. That is fine — the reads are 0.2–4 ms — but it shapes the deployment: four uvicorn workers, one connection each, four concurrent readers against a WAL file. Do not build a 50-connection pool and expect 50× throughput from one disk.

The second decision is pagination. LIMIT ? OFFSET ? re-scans everything before the offset, so page 40 costs forty times page 1. Keyset pagination on (score, id) costs the same on every page. SQLite's row-value comparison (3.15+) makes the predicate readable.

# repo.py
from __future__ import annotations

import base64
import re

import aiosqlite

from schema import Channel, Video, VideoConnection, VideoEdge

MAX_TOKENS = 8
SEARCH_CEILING = 500  # deepest reachable result; documented, not silently truncated
_NOISE = re.compile(r'[\W_]+', re.UNICODE)


def fts_query(raw: str) -> str | None:
    """Port of FtsQuery::fromUserInput. Must stay byte-identical to the PHP version."""
    tokens = [t for t in _NOISE.sub(' ', raw.lower()).split() if t][:MAX_TOKENS]
    if not tokens:
        return None
    last = len(tokens) - 1
    parts = [
        f'"{t}"*' if (i == last and len(t) >= 2) else f'"{t}"'
        for i, t in enumerate(tokens)
    ]
    return ' AND '.join(parts)


def _encode(score: float, vid: str) -> str:
    return base64.urlsafe_b64encode(f'{score:.6f}|{vid}'.encode()).decode()


def _decode(cursor: str) -> tuple[float, str]:
    score, vid = base64.urlsafe_b64decode(cursor.encode()).decode().split('|', 1)
    return float(score), vid


SEARCH_SQL = '''
WITH ranked AS (
    SELECT v.id, v.title, v.published_at, v.duration_seconds,
           v.view_count, v.region, v.channel_id,
           bm25(videos_fts, 10.0, 3.0, 1.0) AS score
    FROM videos_fts
    JOIN videos v ON v.rowid = videos_fts.rowid
    WHERE videos_fts MATCH :m
      AND (:region IS NULL OR v.region = :region)
    ORDER BY score, v.id
    LIMIT :ceiling
)
SELECT * FROM ranked
WHERE (:has_cursor = 0) OR ((score, id) > (:c_score, :c_id))
ORDER BY score, id
LIMIT :lim
'''


class Repo:
    def __init__(self, uri: str) -> None:
        self._uri = uri
        self._conn: aiosqlite.Connection | None = None

    async def connect(self) -> None:
        self._conn = await aiosqlite.connect(self._uri, uri=True)
        self._conn.row_factory = aiosqlite.Row
        await self._conn.execute('PRAGMA query_only = 1')
        await self._conn.execute('PRAGMA cache_size = -32000')      # 32 MB page cache
        await self._conn.execute('PRAGMA mmap_size = 268435456')    # 256 MB mmap

    async def close(self) -> None:
        if self._conn is not None:
            await self._conn.close()

    async def search(self, q: str, first: int, after: str | None, region: str | None) -> VideoConnection:
        match = fts_query(q)
        if match is None or self._conn is None:
            return VideoConnection(edges=[], has_next_page=False)

        c_score, c_id = _decode(after) if after else (0.0, '')
        rows = await self._conn.execute_fetchall(SEARCH_SQL, {
            'm': match,
            'region': region,
            'ceiling': SEARCH_CEILING,
            'has_cursor': 1 if after else 0,
            'c_score': c_score,
            'c_id': c_id,
            'lim': first + 1,   # one extra row tells us has_next_page for free
        })

        has_next = len(rows) > first
        edges = [
            VideoEdge(
                cursor=_encode(r['score'], r['id']),
                node=Video(
                    id=r['id'],
                    title=r['title'],
                    published_at=r['published_at'],
                    duration_seconds=r['duration_seconds'],
                    view_count=r['view_count'],
                    region=r['region'],
                    channel_id=r['channel_id'],
                ),
            )
            for r in rows[:first]
        ]
        return VideoConnection(edges=edges, has_next_page=has_next)

    async def channels_by_ids(self, ids: list[str]) -> list[Channel | None]:
        """DataLoader batch function. MUST return one slot per key, in key order."""
        if self._conn is None:
            return [None] * len(ids)
        placeholders = ','.join('?' * len(ids))
        rows = await self._conn.execute_fetchall(
            f'SELECT id, title, subscriber_count, thumbnail_url '
            f'FROM channels WHERE id IN ({placeholders})',
            ids,
        )
        found = {r['id']: Channel(**dict(r)) for r in rows}
        return [found.get(i) for i in ids]

A few details worth calling out, because each of them cost me an afternoon:

bm25() is negative, and more negative means a better match. Sorting ascending is correct. Sorting descending gives you the worst results first and looks superficially plausible in a demo.
Ties need a tiebreaker. Two videos with identical scores and no secondary sort key will swap order between queries, and keyset pagination will then skip or duplicate rows. ORDER BY score, id fixes it.
The SEARCH_CEILING is a deliberate limit. The CTE materializes matches before the keyset filter, so unbounded deep paging would degrade. Five hundred results is more than any human pages through; I document it rather than pretending it does not exist.
Fetch first + 1 rows to compute has_next_page without a second COUNT(*) over the FTS index.

Killing the N+1 with a per-request DataLoader

Twenty search results reference up to twenty distinct channels. Naively, Video.channel fires twenty separate SELECT ... WHERE id = ? queries. Each one is fast; twenty of them behind a thread-serialized connection is not.

strawberry.dataloader.DataLoader collects the keys requested within a single tick of the event loop and calls your batch function once. The two rules that matter: the batch function must return exactly one slot per key in the same order, and the loader must be created per request, never at module scope. A module-level loader caches results across users forever, which is a stale-data bug and, if your schema ever grows an authenticated field, a data-leak bug.

# app.py
import hashlib
import json
from contextlib import asynccontextmanager
from pathlib import Path

import strawberry
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from graphql.validation import NoSchemaIntrospectionCustomRule
from strawberry.dataloader import DataLoader
from strawberry.extensions import (
    AddValidationRules,
    MaxTokensLimiter,
    ParserCache,
    QueryDepthLimiter,
    ValidationCache,
)
from strawberry.fastapi import GraphQLRouter

from repo import Repo
from schema import Query

DB_URI = 'file:/var/www/data/app.db?mode=ro'
repo = Repo(DB_URI)

schema = strawberry.Schema(
    query=Query,
    extensions=[
        ParserCache(maxsize=256),        # skip re-parsing the same document
        ValidationCache(maxsize=256),    # skip re-validating it too
        QueryDepthLimiter(max_depth=8),
        MaxTokensLimiter(max_token_count=1200),
        AddValidationRules([NoSchemaIntrospectionCustomRule]),  # prod only
    ],
)


async def build_context(request: Request) -> dict:
    return {
        'request': request,
        'repo': repo,
        'loaders': {
            # Fresh per request. max_batch_size stays under SQLite's parameter limit.
            'channel': DataLoader(load_fn=repo.channels_by_ids, max_batch_size=200),
        },
    }


@asynccontextmanager
async def lifespan(app: FastAPI):
    await repo.connect()
    yield
    await repo.close()


app = FastAPI(lifespan=lifespan)
app.include_router(
    GraphQLRouter(schema, context_getter=build_context, graphql_ide=None),
    prefix='/graphql',
)

# --- Persisted queries over GET, so the CDN can cache them ---------------
_raw = json.loads(Path('persisted-queries.json').read_text())
PERSISTED: dict[str, str] = {}
for doc_id, document in _raw.items():
    digest = hashlib.sha256(document.encode()).hexdigest()
    if digest != doc_id:
        raise RuntimeError(f'persisted query {doc_id} does not match its own hash')
    PERSISTED[doc_id] = document


@app.get('/gql')
async def persisted_get(request: Request) -> JSONResponse:
    document = PERSISTED.get(request.query_params.get('id', ''))
    if document is None:
        return JSONResponse({'errors': [{'message': 'PersistedQueryNotFound'}]}, status_code=400)

    raw_vars = request.query_params.get('v')
    try:
        variables = json.loads(raw_vars) if raw_vars else {}
    except json.JSONDecodeError:
        return JSONResponse({'errors': [{'message': 'BadVariables'}]}, status_code=400)

    result = await schema.execute(
        document,
        variable_values=variables,
        context_value=await build_context(request),
    )
    if result.errors:
        return JSONResponse(
            {'data': result.data, 'errors': [e.formatted for e in result.errors]},
            status_code=200,
            headers={'Cache-Control': 'no-store'},
        )
    return JSONResponse(
        {'data': result.data},
        headers={
            'Cache-Control': 'public, s-maxage=300, stale-while-revalidate=600',
            'Vary': 'Accept-Encoding',
        },
    )

Cost limits before you need them

A public GraphQL endpoint is an open invitation to write an expensive query. Four cheap defences, in the order I would add them:

Depth limiting. QueryDepthLimiter(max_depth=8) stops the classic channel { videos { channel { videos { ... } } } } recursion. Our schema's legitimate deepest path is 5.
Token limiting. MaxTokensLimiter rejects absurdly large documents before the parser does real work. Cheaper than any post-parse analysis.
Argument caps in resolvers. Every list field clamps first. This is the one that actually bounds database work, and no generic extension can do it for you.
Introspection off in production. NoSchemaIntrospectionCustomRule as a validation rule. Leave it enabled in staging, where your tooling needs it.

Parser and validation caches are not security features, but they matter here: with persisted queries the document set is small and fixed, so both caches hit essentially 100% of the time and take parsing off the hot path entirely.

Caching a POST protocol at the edge

This is the part that made the whole project worth it. GraphQL over POST is uncacheable by any CDN, which is a real regression when your site already gets a fat cache hit ratio on HTML through LiteSpeed and Cloudflare.

Persisted queries solve it. At build time, a script extracts every GraphQL document from the client, hashes each one with SHA-256, and writes persisted-queries.json. The client ships hashes, not documents. At runtime the app issues GET /gql?id=<sha256>&v=<url-encoded-json> — a plain, idempotent, fully-qualified GET that Cloudflare will cache on the URL. Three things fall out of this:

The request is smaller. A 900-byte query document becomes a 64-character hash.
The endpoint is an allowlist. An unknown hash is a 400. Arbitrary queries are impossible in production, which makes the cost limits above a second line of defence rather than the only one.
The edge does the work. Search results with s-maxage=300 plus stale-while-revalidate=600 mean repeated queries never touch Python. Our persisted GET hit ratio settled around 70%, and the trending query — identical for every user in a region — sits far higher.

I keep /graphql (POST, ad-hoc, introspection enabled) available in staging only. Production exposes /gql alone. Errors always get no-store; caching a transient failure for five minutes is a self-inflicted outage.

Pinning the tokenizer with a parity test

Back to the shared logic. Two implementations of the same sanitizer will diverge, and search that disagrees between the website and the app is a bug users report as "the app is broken" with no further detail. So the Python version is tested against the PHP version directly, by running it.

# tests/test_tokenizer_parity.py
import json
import subprocess

import pytest

from repo import fts_query

CASES = [
    'coffee grinder',
    'C++ tutorial',
    "l'ete a Paris",
    'naive_bayes',
    '   ',
    'a',
    'one two three four five six seven eight nine ten',
    '"drop" OR 1=1',
]


def php_fts(raw: str) -> str | None:
    proc = subprocess.run(
        ['php', 'tools/fts_query.php', raw],
        capture_output=True, text=True, check=True,
    )
    return json.loads(proc.stdout)


@pytest.mark.parametrize('raw', CASES)
def test_python_matches_php(raw: str) -> None:
    assert fts_query(raw) == php_fts(raw)

The first version of the Python tokenizer used re.sub(r'[^\w\s]', ' ', ...), which keeps underscores because \w includes them, while the PHP [^\p{L}\p{N}\s] class strips them. So naive_bayes became one token in Python and two in PHP — different result sets for the same input. The parity test caught it in the first run. Ten lines of test for a class of bug that would otherwise have been found by a user.

What actually changed

Numbers from our hardware, which is one modest box, so treat them as directional:

Search screen: 4 requests to 1.
Payload for 20 results with channel data: ~180 KB down to ~24 KB uncompressed.
p95 time to first meaningful paint on a throttled 3G profile: 2.4 s to ~1.1 s.
Server time for the search query: 6–12 ms warm, dominated by FTS5, not by GraphQL execution.
Channel lookups per search request: 20 down to 1.

And the costs, stated plainly, because they are real:

HTTP status codes stop meaning much. A GraphQL response with errors is still a 200 with an errors array. Every dashboard and alert built on status codes needs rewriting against the response body.
Partial data is now a client concern. data.search.edges[3].channel can be null because one channel row is missing. The client must handle it; REST hid this behind a whole-response failure.
Schema evolution is forever. Removing a field means proving no deployed client asks for it. Persisted queries help enormously — the allowlist is the usage inventory.
Logs get worse. POST /graphql 200 tells you nothing. Log the operation name and the persisted-query id, or you are debugging blind.

I would not do this for a server-rendered site with no API clients. The PHP monolith stays exactly as it was, because for HTML it is the right tool and the numbers say so. GraphQL earned its place here for precisely one reason: multiple clients with different data shapes, hitting the same read model, over slow connections.

Conclusion

The pieces that mattered, in order of impact: a per-request DataLoader, keyset pagination on (score, id) instead of OFFSET, argument caps in resolvers, and persisted queries over GET so the CDN can absorb the repeat traffic. Strawberry contributed the part I care about most — the schema is Python types, so the type checker enforces the contract instead of a hand-maintained SDL file drifting away from reality.

If you are starting this on top of SQLite, do the parity test first. Everything else you can refactor later; a search tokenizer that quietly disagrees with itself will burn a week before you find it.

Semantic Search Over Video Metadata With pgvector and OpenAI Embeddings

ahmet gedik — Mon, 17 Aug 2026 05:00:02 +0000

A visitor typed space documentary with good narration into the search box and got an empty result page. The catalog had 41 videos that matched that description almost perfectly — Cosmos re-uploads, BBC Earth clips, a dozen astrophotography timelapses with narration credits in the description. Not one of them shared a meaningful token with the query. SQLite FTS5 did exactly what a lexical index is built to do, and still returned nothing.

That blank page is the reason this post exists. I run TrendVidStream, a multi-region streaming discovery site: PHP 8.4 on shared LiteSpeed hosting, a single SQLite file as the primary store, an FTS5 external-content table for search, a cron that pulls trending metadata from 8 regions on a staggered schedule, and deploys that happen over FTP with lftp mirror. No Composer on the remote, no Docker, no queue broker. Adding a vector index to that stack is less a database decision than a logistics problem.

Why FTS5 falls over on discovery queries

FTS5 is genuinely good. BM25 ranking, prefix queries, snippet() highlighting, zero operational cost, and it lives inside the same file I already deploy. It handles the query class "user knows the title" better than any vector index I tested. The failures are narrow and predictable:

Vocabulary mismatch. space documentary vs. a title reading The Universe In 4K — Full Episode. Zero token overlap, zero results.
Intent queries. something to fall asleep to, videos like Kurzgesagt, short funny clips. These describe a property of the content, not its wording.
Cross-language drift. With 8 regions in the pipeline, roughly a third of my rows have titles in Japanese, Korean, German, or Portuguese while a large slice of queries arrive in English. FTS5's default tokenizer will not bridge that.
Truncated descriptions. YouTube descriptions are half boilerplate. The signal is in the first two sentences; BM25 sees the whole soup, including "subscribe" and affiliate links, and weights accordingly.

What I wanted was to keep FTS5 for what it is good at and add a second retriever for everything else.

The deploy constraints shaped the architecture

I checked phpinfo() on all four hosts. Two of them do not have pdo_pgsql compiled in, and shared hosting is not where you go to request extensions. That killed the obvious design where PHP talks to Postgres directly.

So the split ended up like this:

A small VPS runs Postgres 17 with pgvector 0.8. This is the only new piece of infrastructure.
A Python ingest worker on the VPS reads video metadata, builds a document per video, embeds it, and upserts into Postgres. It runs on the same 2h/4h/7h stagger as the site crons so it never fights them for API quota.
A Go query service on the same VPS exposes one endpoint: embed the query, run the ANN search, return IDs and scores. It caches query embeddings, because the embedding API call — not Postgres — is the latency floor.
PHP 8.4 on each site calls that endpoint with a 250 ms timeout, merges the returned IDs with local FTS5 results, and hydrates rows from SQLite. If the call fails, the page renders FTS5-only results and nobody notices.

The FTP deploy never touches the vector path. public/ and app/ ship as before; the search service is deployed with git pull && systemctl restart. That separation was worth more to me than any latency I gave up to the extra hop.

Document construction beats model selection

I burned two days tuning ef_search and comparing text-embedding-3-small against 3-large before I accepted the boring truth: what you feed the embedder matters far more than which embedder you use.

My first version embedded title + " " + description. Recall was mediocre, and the failure mode was obvious once I looked at neighbors — videos clustered by description boilerplate, not by content. Three unrelated gaming clips came out as near-duplicates because all three descriptions ended with the same 200-word sponsor block.

The rules that actually moved the numbers:

Strip URLs, hashtag walls, "subscribe to my channel", and social handles before embedding.
Truncate the description to ~600 characters. The tail is almost never signal.
Include channel name and category as labeled fields. Channel: NASA is a strong semantic prior.
Include at most 8 tags. Beyond that, tags are keyword spam and they blur the vector.
Use a stable labeled template, not concatenation. Title: ...\nChannel: ... outperformed raw concatenation on my eval set, consistently.
Do not embed view counts, IDs, durations, or region codes. Those belong in SQL predicates, not in the vector.

import hashlib
import re
import time

import psycopg
from openai import OpenAI

DOC_TEMPLATE_VERSION = "v3"
MODEL = "text-embedding-3-small"
DIMS = 512
BATCH = 128

BOILERPLATE = re.compile(
    r"(subscribe to (my|our) channel|follow me on \w+|https?://\S+|#\w+)",
    re.IGNORECASE,
)
client = OpenAI()


def build_doc(row: dict) -> str:
    desc = BOILERPLATE.sub(" ", row.get("description") or "")
    desc = re.sub(r"\s+", " ", desc).strip()[:600]
    tags = [t for t in (row.get("tags") or "").split(",") if t][:8]

    parts = [
        f"Title: {row['title']}",
        f"Channel: {row['channel_title']}",
        f"Category: {row['category_name']}",
        f"Tags: {', '.join(tags)}" if tags else "",
        f"About: {desc}" if desc else "",
    ]
    return "\n".join(p for p in parts if p)


def doc_hash(doc: str) -> str:
    key = f"{DOC_TEMPLATE_VERSION}|{MODEL}|{DIMS}|{doc}"
    return hashlib.sha256(key.encode("utf-8")).hexdigest()


def embed_batch(docs: list[str]) -> list[list[float]]:
    for attempt in range(5):
        try:
            resp = client.embeddings.create(model=MODEL, input=docs, dimensions=DIMS)
            return [d.embedding for d in resp.data]
        except Exception:
            if attempt == 4:
                raise
            time.sleep(2 ** attempt)  # 429 and 5xx both land here
    raise RuntimeError("unreachable")


def to_literal(vec: list[float]) -> str:
    return "[" + ",".join(f"{v:.6f}" for v in vec) + "]"


def sync(conn: psycopg.Connection, rows: list[dict]) -> int:
    with conn.cursor() as cur:
        cur.execute("SELECT video_id, doc_hash FROM video_embeddings")
        known = dict(cur.fetchall())

    pending = []
    for row in rows:
        doc = build_doc(row)
        h = doc_hash(doc)
        if known.get(row["video_id"]) != h:
            pending.append((row["video_id"], doc, h, row["regions"]))

    written = 0
    for i in range(0, len(pending), BATCH):
        chunk = pending[i:i + BATCH]
        vectors = embed_batch([c[1] for c in chunk])
        params = [
            (c[0], c[2], to_literal(vec), c[3])
            for c, vec in zip(chunk, vectors)
        ]
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO video_embeddings
                    (video_id, doc_hash, embedding, regions, updated_at)
                VALUES (%s, %s, %s::halfvec, %s, now())
                ON CONFLICT (video_id) DO UPDATE SET
                    doc_hash  = EXCLUDED.doc_hash,
                    embedding = EXCLUDED.embedding,
                    regions   = EXCLUDED.regions,
                    updated_at = now()
                """,
                params,
            )
        conn.commit()
        written += len(chunk)
    return written

The hash is the load-bearing part. Video metadata churns — titles get edited, descriptions get sponsor blocks appended — but on a typical 2-hour cycle only ~3% of rows actually change. Hashing the rendered document rather than the source row means cosmetic upstream changes cost nothing, and putting the template version and model into the hash means that when I change either, everything re-embeds on the next run without me writing a migration. I learned that one the hard way after silently mixing v1 and v2 documents in the same index for a week and wondering why recall had gotten worse.

Schema, dimensions, and the index

I use dimensions=512 instead of the native 1536. text-embedding-3-small is Matryoshka-trained, so you can request a truncated embedding directly from the API and it stays useful. On my 120-query eval set the drop from 1536 to 512 cost about 1.5 points of recall@10 and cut index size by two-thirds. Combined with halfvec (16-bit floats) that is a 6x reduction against vector(1536) — 61k rows fit in roughly 60 MB of index instead of 380 MB, which means it stays in the page cache of a small VPS.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE video_embeddings (
    video_id     text          PRIMARY KEY,
    doc_hash     text          NOT NULL,
    embedding    halfvec(512)  NOT NULL,
    regions      text[]        NOT NULL DEFAULT '{}',
    published_at timestamptz,
    updated_at   timestamptz   NOT NULL DEFAULT now()
);

SET maintenance_work_mem = '2GB';

CREATE INDEX video_embeddings_hnsw
    ON video_embeddings
    USING hnsw (embedding halfvec_cosine_ops)
    WITH (m = 16, ef_construction = 96);

CREATE INDEX video_embeddings_regions
    ON video_embeddings USING gin (regions);

-- query side
SET LOCAL hnsw.ef_search = 100;
SET LOCAL hnsw.iterative_scan = 'relaxed_order';

SELECT video_id,
       1 - (embedding <=> $1::halfvec) AS score
FROM video_embeddings
WHERE regions && ARRAY[$2]::text[]
  AND published_at > now() - interval '18 months'
ORDER BY embedding <=> $1::halfvec
LIMIT 40;

Three notes on that, each of which cost me an evening:

maintenance_work_mem decides whether your build takes 4 minutes or 40. If HNSW construction spills to disk, pgvector logs it. Watch for that line.
The ORDER BY must use the distance operator directly against a parameter. Wrap it in a CTE, sort by an alias, or apply any expression around it, and the planner quietly drops to a sequential scan. EXPLAIN is not optional here; check for Index Scan using video_embeddings_hnsw.
Region filtering is where naive HNSW goes wrong. A single-region site filters out ~85% of rows. With a plain ANN scan, Postgres fetches 40 neighbors, throws away 34 of them post-filter, and hands you 6 results. pgvector 0.8's iterative scan fixes this by re-entering the index until it has enough rows that survive the filter. Set hnsw.iterative_scan to relaxed_order for top-k search; strict_order is slower and I could not measure a quality difference at k=40.

The Go query service

Query-side latency is dominated by the embedding API call, not by Postgres. The ANN scan over 61k rows runs in 3–7 ms. The embedding round trip is 90–140 ms at p50. Since search queries follow a brutal power law — my top 200 queries account for over 60% of traffic — an in-process cache erases most of that.

package main

import (
    "context"
    "encoding/json"
    "net/http"
    "strconv"
    "strings"
    "sync"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

type Hit struct {
    VideoID string  `json:"video_id"`
    Score   float64 `json:"score"`
}

type cached struct {
    vec []float32
    exp time.Time
}

type Server struct {
    db    *pgxpool.Pool
    embed func(context.Context, string) ([]float32, error)

    mu    sync.RWMutex
    cache map[string]cached
}

func (s *Server) queryVector(ctx context.Context, q string) ([]float32, error) {
    key := strings.ToLower(strings.TrimSpace(q))

    s.mu.RLock()
    c, ok := s.cache[key]
    s.mu.RUnlock()
    if ok && time.Now().Before(c.exp) {
        return c.vec, nil
    }

    vec, err := s.embed(ctx, key)
    if err != nil {
        return nil, err
    }
    s.mu.Lock()
    s.cache[key] = cached{vec: vec, exp: time.Now().Add(6 * time.Hour)}
    s.mu.Unlock()
    return vec, nil
}

func literal(vec []float32) string {
    var b strings.Builder
    b.WriteByte('[')
    for i, v := range vec {
        if i > 0 {
            b.WriteByte(',')
        }
        b.WriteString(strconv.FormatFloat(float64(v), 'f', 6, 32))
    }
    b.WriteByte(']')
    return b.String()
}

// pool config sets hnsw.ef_search + iterative_scan in AfterConnect,
// so every pooled session already has the right GUCs.
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 200*time.Millisecond)
    defer cancel()

    q := r.URL.Query()
    if strings.TrimSpace(q.Get("q")) == "" {
        http.Error(w, "missing q", http.StatusBadRequest)
        return
    }
    k, _ := strconv.Atoi(q.Get("k"))
    if k <= 0 || k > 100 {
        k = 40
    }

    vec, err := s.queryVector(ctx, q.Get("q"))
    if err != nil {
        http.Error(w, "embed failed", http.StatusBadGateway)
        return
    }

    rows, err := s.db.Query(ctx, `
        SELECT video_id, 1 - (embedding <=> $1::halfvec) AS score
        FROM video_embeddings
        WHERE regions && ARRAY[$2]::text[]
        ORDER BY embedding <=> $1::halfvec
        LIMIT $3`, literal(vec), q.Get("region"), k)
    if err != nil {
        http.Error(w, "query failed", http.StatusBadGateway)
        return
    }
    defer rows.Close()

    hits := make([]Hit, 0, k)
    for rows.Next() {
        var h Hit
        if err := rows.Scan(&h.VideoID, &h.Score); err != nil {
            http.Error(w, "scan failed", http.StatusInternalServerError)
            return
        }
        hits = append(hits, h)
    }

    w.Header().Set("Content-Type", "application/json")
    _ = json.NewEncoder(w).Encode(map[string]any{"hits": hits})
}

The map-with-mutex cache is deliberately dumb. It is bounded in practice by query diversity and a 6-hour TTL, and it holds about 40k entries at 512 float32s each — roughly 80 MB worst case. If that ever became a problem I would swap in an LRU, but measuring first told me it would not.

Hybrid ranking in PHP 8.4

Vector-only search regressed on exact-title queries, which are the ones users are most annoyed to see fail. The fix is Reciprocal Rank Fusion: merge the two ranked lists by 1 / (k + rank) with k = 60. It needs no score normalization — which matters, because BM25 scores and cosine similarities are not on comparable scales and any attempt to normalize them across queries is guesswork.

<?php
declare(strict_types=1);

final class HybridVideoSearch
{
    private const int RRF_K = 60;
    private const int TIMEOUT_MS = 250;

    public function __construct(
        private readonly PDO $sqlite,
        private readonly string $vectorEndpoint,
        private readonly string $region,
    ) {}

    /** @return list<array{video_id:string, score:float, sources:list<string>}> */
    public function search(string $query, int $limit = 20): array
    {
        $lists = [
            'fts' => $this->fts5($query, 40),
            'vec' => $this->semantic($query, 40), // [] when the service is slow or down
        ];

        $scores = [];
        foreach ($lists as $source => $ids) {
            foreach ($ids as $rank => $id) {
                $scores[$id]['score'] ??= 0.0;
                $scores[$id]['score'] += 1 / (self::RRF_K + $rank + 1);
                $scores[$id]['sources'][] = $source;
            }
        }
        uasort($scores, static fn (array $a, array $b): int => $b['score'] <=> $a['score']);

        $out = [];
        foreach (array_slice($scores, 0, $limit, preserve_keys: true) as $id => $row) {
            $out[] = [
                'video_id' => (string) $id,
                'score'    => $row['score'],
                'sources'  => $row['sources'],
            ];
        }
        return $out;
    }

    /** @return list<string> */
    private function fts5(string $query, int $limit): array
    {
        $terms = preg_split('/[^\p{L}\p{N}]+/u', $query, -1, PREG_SPLIT_NO_EMPTY) ?: [];
        if ($terms === []) {
            return [];
        }
        // quote every term so FTS5 operators in user input cannot reach the parser
        $match = implode(' OR ', array_map(
            static fn (string $t): string => '"' . $t . '"*',
            $terms,
        ));

        $st = $this->sqlite->prepare(
            'SELECT video_id FROM videos_fts
             WHERE videos_fts MATCH :m
             ORDER BY bm25(videos_fts, 8.0, 1.0)
             LIMIT :l'
        );
        $st->bindValue(':m', $match);
        $st->bindValue(':l', $limit, PDO::PARAM_INT);
        $st->execute();

        return $st->fetchAll(PDO::FETCH_COLUMN);
    }

    /** @return list<string> */
    private function semantic(string $query, int $limit): array
    {
        $url = $this->vectorEndpoint . '?' . http_build_query([
            'q'      => $query,
            'region' => $this->region,
            'k'      => $limit,
        ]);

        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER    => true,
            CURLOPT_TIMEOUT_MS        => self::TIMEOUT_MS,
            CURLOPT_CONNECTTIMEOUT_MS => 80,
        ]);
        $raw = curl_exec($ch);
        $ok  = $raw !== false && curl_getinfo($ch, CURLINFO_RESPONSE_CODE) === 200;
        curl_close($ch);

        if (!$ok) {
            return []; // degrade to FTS5-only, no error page
        }

        try {
            $decoded = json_decode((string) $raw, true, 512, JSON_THROW_ON_ERROR);
        } catch (JsonException) {
            return [];
        }
        return array_column($decoded['hits'] ?? [], 'video_id');
    }
}

The bm25(videos_fts, 8.0, 1.0) weights matter: column 1 is the title, column 2 the description, and weighting the title 8x pushed a lot of junk out of the lexical list before fusion ever saw it. Also note the escaping in fts5() — unquoted user input goes straight to the FTS5 query parser, where a stray " or NEAR is a syntax error and a 500 page.

What the numbers actually looked like

I built a 120-query eval set by hand: real queries from the site's search log, each with 1–5 video IDs I judged relevant. Small, biased toward the cases I already knew were broken, and absolutely not a benchmark — but it is my traffic, which is the only distribution I care about.

FTS5 only — recall@10 0.41, MRR 0.29
Vector only — recall@10 0.68, MRR 0.51
RRF hybrid — recall@10 0.79, MRR 0.61

Cost was the surprise. Backfilling 61,000 unique videos at ~180 tokens per document is roughly 11M tokens — about $0.22 one-time with text-embedding-3-small. Incremental churn runs ~900 documents a day, so ongoing embedding cost is well under a dollar a month. Query embeddings are ~8 tokens each with a 60–70% cache hit rate; they round to zero. The real cost is the VPS.

End-to-end p95 from PHP, measured at the controller: 210 ms on a cache miss, 34 ms on a hit. The 250 ms timeout fires on roughly 0.4% of requests, and those users get FTS5 results rather than an error.

Things I would do differently

Start at 512 dimensions. I backfilled at 1536 first, then re-embedded everything a week later. That is entirely avoidable planning.
Version the document template from day one. Mixing template versions in one index is a silent quality bug with no error message attached.
Keep FTS5. It is not a fallback I tolerate; it is half the ranking signal and it is the half that handles exact lookups.
Test a multilingual open-weights model. With a third of the catalog in non-English titles, running something like bge-m3 locally on the same VPS may beat the API on quality for my specific mix, and removes the network call entirely.

Conclusion

The part I underestimated was document construction. Choosing pgvector over a dedicated vector database, tuning m and ef_search, picking between HNSW and IVFFlat — all of that moved recall by single-digit percentages. Stripping boilerplate from descriptions and adding labeled channel and category fields moved it by twenty points. If you are adding semantic search to an existing catalog, spend your first day printing out the exact strings you are about to embed and reading them. You will find sponsor blocks, and you will find that removing them was the whole job.

And keep the lexical index. Hybrid retrieval on a small, honest eval set beat every single-retriever configuration I tried, and it degrades gracefully when the network does what networks do.

Hardening a PHP Video Admin Panel with Signed Double-Submit CSRF Cookies

ahmet gedik — Sun, 16 Aug 2026 22:00:01 +0000

The request we never made

Last quarter I ran a scheduled audit of the moderation log on our admin panel and found 41 region_block actions I could not account for. No admin session was active at those timestamps. The IP in the audit row was a real editor's IP. The user agent was a real browser.

The cause turned out to be embarrassingly small. Our CSRF guard was one line: if (!empty($_POST) && !hash_equals($_SESSION['csrf'] ?? '', $_POST['csrf'] ?? '')) { http_response_code(403); exit; }

Read it carefully. The guard only fires when $_POST is non-empty. Our older HTML forms posted application/x-www-form-urlencoded, so they populated $_POST and got checked. But every endpoint we had added in the previous six months — bulk approve, bulk region-block, the trend-score recompute trigger — was called from JavaScript with Content-Type: application/json. PHP does not populate $_POST for JSON bodies. $_POST was empty. The guard was skipped. Those endpoints had been completely unprotected since the day they shipped, and a page on any domain could fire them at a logged-in editor's browser.

I rebuilt the whole thing around signed double-submit cookies rather than patching the conditional, and this post is the implementation that now runs in production on ViralVidVault — PHP 8.4, SQLite in WAL mode, LiteSpeed, with Cloudflare Workers in front. The code below is what is actually deployed, with the secret handling and route names changed.

Why we dropped session-stored tokens

The obvious fix is $_SESSION['csrf'] done properly: enforce on every unsafe method, compare with hash_equals, done. That is a correct design and for a lot of apps it is the right one. We moved off it for four specific reasons.

PHP's session lock serializes an admin's own tabs. The default file handler holds an exclusive lock from session_start() to script end. A bulk purge over 6,000 rows takes 8–12 seconds on our box. During that window every other tab that admin has open blocks on session_start(). Editors thought the panel had crashed. Moving sessions into SQLite made it worse, not better — the same lock, now competing with the WAL writer.
Four sites, one codebase, no shared state. We deploy the same tree to four domains. Anything that needs server-side state at request time is one more thing that has to be identical, migrated, and backed up on all of them.
Cached HTML is a token leak. We run a three-layer cache (LiteSpeed page cache, a PHP file page cache, then a data cache). Admin routes are excluded, but "excluded by configuration" is a promise, not a guarantee. A token that is only valid for one session and cryptographically bound to it fails closed if a page ever leaks into a shared cache.
Data minimisation. Under GDPR the CSRF cookie is a strictly necessary cookie — it is exempt from consent under the ePrivacy "strictly necessary" carve-out, which means it must genuinely be strictly necessary and nothing more. A stateless token that carries only an expiry, a nonce and a MAC is far easier to document in a records-of-processing entry than "an opaque identifier joined server-side to a session table."

What double-submit actually buys you

The classic naive version: generate a random value, put it in a cookie, also put it in the form. On POST, check that the two match. The reasoning is that an attacker's page can cause the cookie to be sent (cookies travel cross-site) but cannot read it, so it cannot put the matching value in the body.

That reasoning has a hole, and it is the reason plain double-submit is no longer recommended on its own. The attacker does not need to read your cookie. They need to write one. Cookies do not respect the same-origin policy for writes the way storage does:

A compromised or attacker-controlled subdomain can set a cookie with Domain=example.com that the apex will send.
A MITM on any plaintext HTTP connection to any host in the registrable domain can inject a Set-Cookie that the browser will honour on HTTPS requests, because cookies are not isolated by scheme.
An old, forgotten staging host under the same domain is enough.

Once the attacker can write the cookie, they set it to attacker-chosen-value, put attacker-chosen-value in their forged form, and the equality check passes. This is cookie tossing, and it defeats unsigned double-submit completely.

The fix is that the token has to be unforgeable, not merely matching. Sign it with a server-side secret and verify the signature independently of the equality check. An attacker who can write cookies still cannot produce a value that carries a valid MAC. Two further hardening steps:

Use the __Host- cookie prefix. Browsers refuse a __Host--prefixed cookie unless it is Secure, has Path=/, and has no Domain attribute. No Domain means no subdomain can set it for the apex. This single prefix kills most of the cookie-tossing surface at the browser level.
Bind the token to the session. Include a hash of the session identifier in the signed payload. A valid token minted for admin A is then useless in admin B's browser, and a token captured from a logged-out session is useless after re-login.

What it does not buy you, and I want to be blunt about this: none of it survives XSS. If an attacker runs script on your origin they read the token out of the DOM and post whatever they like. CSRF defence and XSS defence are separate problems, and a strict CSP is the other half of the job.

HttpOnly, and why we do not let JavaScript read the cookie

Most double-submit write-ups set the cookie without HttpOnly so document.cookie can read it back for fetch. We do not. The cookie is HttpOnly; the server renders the same token into the page as a <meta> tag, and JavaScript reads it from there.

This is still double-submit — the browser sends the cookie automatically, the client sends the token explicitly, the server compares them — but the token never lives in a JS-readable cookie. It also plays better with a strict CSP: the meta tag is part of the document that the server controlled, and there is no code path in which a third-party script pulls a security token out of document.cookie.

Token format

The token is four dot-separated fields — v1, the unix expiry, a base64url nonce, and a base64url MAC — where the MAC is HMAC-SHA256(secret, "v1|" + expiry + "|" + nonce + "|" + sessionBinding).

sessionBinding is hash('sha256', $sessionId . $userId). It is inside the MAC input but not transmitted, so the token leaks nothing about the session. Verification recomputes it from the current request. Expiry is 2 hours with 60 seconds of leeway for clock skew between the app host and whatever generated the page.

The PHP 8.4 implementation

<?php
declare(strict_types=1);

namespace App\Security;

final class Csrf
{
    private const string VERSION  = 'v1';
    private const string COOKIE   = '__Host-csrf';
    private const string HEADER   = 'HTTP_X_CSRF_TOKEN';
    private const int    TTL      = 7200;
    private const int    LEEWAY   = 60;
    private const int    NONCE_LEN = 24;

    public function __construct(
        private readonly string $secret,
        private readonly bool $secureContext = true,
    ) {
        if (strlen($this->secret) < 32) {
            throw new \InvalidArgumentException('CSRF secret must be >= 32 bytes');
        }
    }

    public function issue(string $binding, ?int $now = null): string
    {
        $now ??= time();
        $expiry = $now + self::TTL;
        $nonce  = $this->b64url(random_bytes(self::NONCE_LEN));
        $mac    = $this->sign($expiry, $nonce, $binding);

        return self::VERSION . '.' . $expiry . '.' . $nonce . '.' . $mac;
    }

    /** Returns null on success, or a machine-readable failure reason. */
    public function verify(string $submitted, string $fromCookie, string $binding, ?int $now = null): ?string
    {
        $now ??= time();

        if ($submitted === '' || $fromCookie === '') {
            return 'missing';
        }
        // Compare the two copies first: cheap, constant-time, catches the common case.
        if (!hash_equals($fromCookie, $submitted)) {
            return 'mismatch';
        }

        $parts = explode('.', $submitted);
        if (count($parts) !== 4 || $parts[0] !== self::VERSION) {
            return 'malformed';
        }
        [, $expiryRaw, $nonce, $mac] = $parts;

        if (!ctype_digit($expiryRaw)) {
            return 'malformed';
        }
        $expiry = (int) $expiryRaw;

        // Signature check BEFORE expiry check, so a forged token never
        // reaches logic that branches on attacker-controlled values.
        if (!hash_equals($this->sign($expiry, $nonce, $binding), $mac)) {
            return 'bad_signature';
        }
        if ($now > $expiry + self::LEEWAY) {
            return 'expired';
        }

        return null;
    }

    public function sendCookie(string $token): void
    {
        setcookie($this->cookieName(), $token, [
            'expires'  => 0,           // session cookie; token carries its own expiry
            'path'     => '/',
            'secure'   => $this->secureContext,
            'httponly' => true,
            'samesite' => 'Lax',
        ]);
    }

    public function fromRequest(): array
    {
        $cookie = (string) ($_COOKIE[$this->cookieName()] ?? '');

        $submitted = (string) ($_POST['_csrf'] ?? '');
        if ($submitted === '') {
            $submitted = (string) ($_SERVER[self::HEADER] ?? '');
        }
        if ($submitted === '' && str_contains((string) ($_SERVER['CONTENT_TYPE'] ?? ''), 'application/json')) {
            $body = json_decode((string) file_get_contents('php://input'), true);
            $submitted = is_array($body) ? (string) ($body['_csrf'] ?? '') : '';
        }

        return [$submitted, $cookie];
    }

    public function cookieName(): string
    {
        // __Host- requires Secure; on plain-HTTP local dev the browser drops it silently.
        return $this->secureContext ? self::COOKIE : 'dev-csrf';
    }

    private function sign(int $expiry, string $nonce, string $binding): string
    {
        $payload = self::VERSION . '|' . $expiry . '|' . $nonce . '|' . $binding;

        return $this->b64url(hash_hmac('sha256', $payload, $this->secret, true));
    }

    private function b64url(string $raw): string
    {
        return rtrim(strtr(base64_encode($raw), '+/', '-_'), '=');
    }
}

Three details worth calling out:

verify() checks the signature before the expiry. If you check expiry first you are branching on an integer an attacker fully controls, and you end up with timing and logic differences between "expired" and "forged" that leak information.
The cookie is a session cookie (expires => 0). The lifetime that matters lives inside the signed payload, where the client cannot extend it.
fromRequest() deliberately accepts the token from a form field, a header, or a JSON body. That is the whole point of the incident that started this: the guard must work regardless of how the endpoint is called.

Wiring it into the router

The rule is: every state-changing method is checked, and the check is opt-out by explicit allowlist, not opt-in. If a new endpoint forgets to declare itself, it gets protected, not skipped.

<?php
declare(strict_types=1);

use App\Security\Csrf;

const SAFE_METHODS = ['GET', 'HEAD', 'OPTIONS', 'TRACE'];

// Endpoints authenticated by something other than a cookie (signed webhook, API key).
const CSRF_EXEMPT = [
    '/hooks/indexnow',
    '/hooks/cf-purge',
];

function csrfGuard(Csrf $csrf, string $method, string $path, string $binding): void
{
    if (in_array($method, SAFE_METHODS, true) || in_array($path, CSRF_EXEMPT, true)) {
        return;
    }

    // Guard against the confusing failure mode: an upload over post_max_size
    // arrives with empty $_POST and $_FILES and looks exactly like a CSRF failure.
    $declared = (int) ($_SERVER['CONTENT_LENGTH'] ?? 0);
    $limit    = (int) parseBytes(ini_get('post_max_size') ?: '8M');
    if ($declared > 0 && $limit > 0 && $declared > $limit) {
        http_response_code(413);
        header('Content-Type: application/json');
        echo json_encode(['error' => 'payload_too_large', 'limit' => $limit]);
        exit;
    }

    // Defence in depth: the Origin header is set by the browser on every
    // cross-site POST and cannot be forged by page script.
    $origin = $_SERVER['HTTP_ORIGIN'] ?? null;
    if ($origin !== null && !isAllowedOrigin($origin)) {
        rejectCsrf('bad_origin');
    }

    [$submitted, $cookie] = $csrf->fromRequest();
    $reason = $csrf->verify($submitted, $cookie, $binding);

    if ($reason !== null) {
        rejectCsrf($reason);
    }
}

function rejectCsrf(string $reason): never
{
    // 419 is non-standard but unambiguous for the client: token problem, retry after refresh.
    http_response_code($reason === 'expired' ? 419 : 403);
    header('Content-Type: application/json');
    header('Cache-Control: no-store');
    csrfAudit($reason);                       // rate-limited; see the WAL note below
    echo json_encode(['error' => 'csrf', 'reason' => $reason]);
    exit;
}

function isAllowedOrigin(string $origin): bool
{
    $host = parse_url($origin, PHP_URL_HOST);

    return is_string($host) && hash_equals($_SERVER['HTTP_HOST'] ?? '', $host);
}

function parseBytes(string $val): int
{
    $val  = trim($val);
    $unit = strtolower($val[strlen($val) - 1] ?? '');
    $num  = (int) $val;

    return match ($unit) {
        'g' => $num * 1024 ** 3,
        'm' => $num * 1024 ** 2,
        'k' => $num * 1024,
        default => $num,
    };
}

// ---------------------------------------------------------------------------
// Render side. The same token goes into three places on every admin page:
// the cookie, a meta tag for fetch(), and a hidden field for the no-JS forms.
// ---------------------------------------------------------------------------

$binding = hash('sha256', session_id() . '|' . $currentAdminId);
$token   = $csrf->issue($binding);
$csrf->sendCookie($token);

// in the layout <head>
printf('<meta name="csrf-token" content="%s">', htmlspecialchars($token, ENT_QUOTES, 'UTF-8'));

// in each form
printf('<input type="hidden" name="_csrf" value="%s">', htmlspecialchars($token, ENT_QUOTES, 'UTF-8'));

Do not rotate the token on every request. We tried it for two days. Editors keep six or seven tabs open on the moderation queue, and per-request rotation means the five older tabs are all holding dead tokens; the first bulk action after switching tabs fails. Rotate on login, on logout, and on privilege change. Let the TTL handle the rest.

The client side

The client needs to attach the token to every unsafe request and recover cleanly when it expires — a moderator who left a tab open over lunch should not lose a queue of selections.

let csrfToken = document.querySelector('meta[name="csrf-token"]')?.content ?? '';

async function refreshCsrf() {
  const res = await fetch('/admin/csrf', { credentials: 'same-origin' });
  if (!res.ok) throw new Error('csrf refresh failed');
  csrfToken = (await res.json()).token;
  return csrfToken;
}

export async function apiPost(url, payload, { retried = false } = {}) {
  const res = await fetch(url, {
    method: 'POST',
    credentials: 'same-origin',
    headers: {
      'Content-Type': 'application/json',
      'X-CSRF-Token': csrfToken,
    },
    body: JSON.stringify(payload),
  });

  if (res.status === 419 && !retried) {
    await refreshCsrf();
    return apiPost(url, payload, { retried: true });
  }
  if (res.status === 403) {
    const { reason } = await res.json().catch(() => ({ reason: 'unknown' }));
    throw new Error(`request rejected: ${reason}`);
  }
  return res.json();
}

The /admin/csrf endpoint is a GET that requires an authenticated session, sets a fresh cookie, and returns the token. It must send Cache-Control: no-store — a cached token refresh endpoint would be a genuinely funny bug to debug.

Retry exactly once. An unbounded retry loop against a server that has decided your session is dead is a self-inflicted denial of service, and I have watched a colleague ship one.

Testing the attacks, not just the happy path

Unit tests that assert "valid token passes, missing token fails" are close to worthless here. The interesting cases are the forgeries. This harness runs against a throwaway instance in CI:

import time
import requests

BASE = "http://127.0.0.1:8080"
TARGET = f"{BASE}/admin/videos/bulk"
PAYLOAD = {"action": "region_block", "ids": [1, 2, 3]}


def logged_in_session():
    s = requests.Session()
    s.post(f"{BASE}/admin/login", data={"user": "ci", "pass": "ci-password"})
    page = s.get(f"{BASE}/admin/queue").text
    token = page.split('name="csrf-token" content="')[1].split('"')[0]
    return s, token


def post(session, token=None, header=True, cookie=None):
    headers = {"X-CSRF-Token": token} if (token and header) else {}
    cookies = {"dev-csrf": cookie} if cookie else None
    return session.post(TARGET, json=PAYLOAD, headers=headers, cookies=cookies)


def test_valid_request_passes():
    s, token = logged_in_session()
    assert post(s, token).status_code == 200


def test_json_body_without_token_is_rejected():
    # The original bug: JSON body left $_POST empty and skipped the guard.
    s, _ = logged_in_session()
    assert post(s).status_code == 403


def test_cookie_tossing_is_rejected():
    # Attacker controls both copies but cannot produce a valid MAC.
    s, _ = logged_in_session()
    forged = "v1.%d.AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.ZZZZZZZZ" % (int(time.time()) + 3600)
    r = post(s, forged, cookie=forged)
    assert r.status_code == 403
    assert r.json()["reason"] == "bad_signature"


def test_truncated_mac_is_rejected():
    s, token = logged_in_session()
    v, exp, nonce, mac = token.split(".")
    clipped = ".".join([v, exp, nonce, mac[:8]])
    assert post(s, clipped, cookie=clipped).json()["reason"] == "bad_signature"


def test_token_from_another_session_is_rejected():
    victim, _ = logged_in_session()
    attacker, stolen = logged_in_session()
    r = post(victim, stolen, cookie=stolen)
    assert r.status_code == 403
    assert r.json()["reason"] == "bad_signature"   # binding differs -> MAC differs


def test_expired_token_returns_419():
    s, token = logged_in_session()
    v, _, nonce, mac = token.split(".")
    stale = ".".join([v, str(int(time.time()) - 99999), nonce, mac])
    assert post(s, stale, cookie=stale).status_code in (403, 419)

The cross-session test is the one I would keep if I could keep only one. It is the test that fails the moment somebody "simplifies" the binding out of the signature input because it made a refactor awkward.

LiteSpeed and Cloudflare

Two infrastructure notes that cost me time.

First, admin routes must be excluded from every cache layer, and on LiteSpeed the exclusion has to be inside a LiteSpeed-specific block — Apache's mod_rewrite parses commas inside [E=...] flags as flag separators and returns a 500:

<IfModule LiteSpeed>
    CacheEngine on
    RewriteEngine On
    RewriteRule ^admin - [E=Cache-Control:no-cache]
    RewriteCond %{HTTP_COOKIE} (^|;\s*)__Host-adm=
    RewriteRule .* - [E=Cache-Control:no-cache]
</IfModule>

The second rule is the belt-and-braces one: if the admin session cookie is present at all, nothing about that response is cacheable, whatever route it is.

Second, if you terminate at a Cloudflare Worker, make sure the Worker forwards the Cookie header untouched and does not apply a "Cache Everything" page rule to admin paths. A Worker that constructs a new Request and forgets to copy headers will strip your X-CSRF-Token, and the resulting 403 looks exactly like an attack in your logs.

Failure modes we hit in production

__Host- on local HTTP. The browser drops the cookie silently — no console warning that anyone notices. Everything 403s locally and works in staging. Hence the cookieName() switch on scheme.
Upload over post_max_size. PHP delivers an empty $_POST and $_FILES, so the token vanishes and the failure reports as CSRF. Editors uploading thumbnails saw "security error" and assumed they had been logged out. The CONTENT_LENGTH pre-check in the guard turns that into an honest 413.
Audit-log write amplification. Logging every failure straight to SQLite was fine until a misconfigured crawler hammered a POST endpoint and we took thousands of writes per minute against a WAL that also serves the trending-video ingest. Failures are now aggregated in APCu and flushed once a minute.
Clock skew. Two hosts drifted about 40 seconds apart and produced sporadic expired rejections at the boundary. The 60-second leeway plus NTP fixed it. Do not set leeway to zero to look rigorous.
The exempt list is the real risk. Every entry in CSRF_EXEMPT is a route where you have promised the authentication is not cookie-based. We review that list on every release, and each entry has a comment naming the mechanism that replaces the token.

Conclusion

The vulnerability that started this was not a weak algorithm. It was a guard that only ran for one of the two ways our own code called our own endpoints. That is the normal shape of a CSRF bug: not a broken primitive, a hole in the coverage.

So the design goals, in order, are coverage first and cryptography second. Check every unsafe method by default with an explicit, reviewed exemption list. Read the token from a form field, a header, or a JSON body, because your own frontend will use all three eventually. Then make the token unforgeable with an HMAC over an expiry, a nonce, and a session binding, ship it under a __Host- prefix, and verify the signature before you branch on anything the client sent. The stateless part is a bonus: no session locking, no shared server state, and a cookie that is genuinely strictly necessary and therefore trivially defensible to a regulator.

Test the forgeries, not the happy path. And write down why every exempt route is exempt — that list is where the next incident is hiding.

Why We Moved Video Metadata Primary Keys From Bigserial To UUID v7

ahmet gedik — Sun, 16 Aug 2026 14:00:01 +0000

Our nightly ingest window is 90 minutes. Nine Asia-Pacific regions, roughly a dozen upstream feeds each, ~180,000 rows of video metadata upserted into a single Postgres table that already held 41 million rows. Last November that job started running 70 minutes, then 88, then it blew through the window and collided with the morning cache warm. Nothing had changed in the query plan. What had changed was the size of the primary key index relative to shared_buffers: 2.9 GB of B-tree holding nothing but random 128-bit values, at roughly 65% page fill, thrashing every single insert against a different page.

That table backs TopVideoHub, a multi-language trending-video aggregator built on PHP 8.4 with a Postgres system of record and per-region SQLite FTS5 edge indexes for CJK search. This post is what we actually did about it — moving from random UUID v4 to RFC 9562 UUID v7 — including the generator code in three languages, the numbers we measured, and the four things that bit us.

Why v4 was the wrong default for append-heavy metadata

We picked UUID v4 for the usual reasons. Video rows are created by workers in three regions, ids need to be generatable client-side before the row exists (we build the FTS payload and the CDN purge list in the same transaction), and we did not want a central sequence in the hot path. All of that is still true. Randomness was never the requirement — decentralized generation was, and we conflated the two.

The cost of full randomness in a B-tree is specific and measurable:

Page splits everywhere. Every insert lands in an unpredictable leaf page. Postgres splits a full leaf 50/50, and those halves rarely fill again. Steady-state fill factor on our v4 PK index sat around 65%.
Full-page writes in WAL. After a checkpoint, the first write to any page writes the whole page into WAL. Random inserts touch thousands of distinct pages per batch instead of one hot page, so the WAL volume per ingest was dominated by FPWs.
Zero cache locality. The working set of a random index is the whole index. On a 16 GB box with 4 GB shared_buffers, a 2.9 GB index that is only ever touched randomly evicts everything else.
No usable ordering. ORDER BY id was meaningless, so every feed query needed a separate published_at index and an OFFSET-based paginator.

UUID v7 fixes the first three by construction and hands you the fourth for free.

What UUID v7 actually encodes

RFC 9562 (May 2024) standardized v7 as a 128-bit layout:

bits 0–47 — Unix timestamp in milliseconds, big-endian, unsigned
bits 48–51 — version, always 0111
bits 52–63 — rand_a, 12 bits the RFC lets you use as a sub-millisecond counter
bits 64–65 — variant, always 10
bits 66–127 — rand_b, 62 random bits

The important consequence: because Postgres compares uuid values as a plain 16-byte memcmp, and SQLite compares BLOB the same way, byte order equals time order. Sorting by the key sorts by creation time in both stores, with no extra column and no extra index.

The 12-bit counter is what makes ordering stable within a millisecond. If you leave rand_a random, ids created in the same millisecond shuffle relative to each other, which breaks keyset pagination in exactly the way that is hardest to reproduce in staging. Uniqueness never depends on the counter — that is the job of the 62 random bits — so the counter can wrap or repeat without any practical collision risk.

Generating v7 in PHP 8.4

ramsey/uuid 4.7+ has Uuid::uuid7() and symfony/uid has UuidV7, and both are fine choices. We ship our own for two reasons: we want the raw 16-byte form without a value object allocation per row (we generate ~180k per night inside a tight loop), and we want to control the monotonic counter so it survives a backwards clock step.

<?php
declare(strict_types=1);

final class UuidV7
{
    private static int $lastMs = 0;
    private static int $counter = 0;

    /** Returns the raw 16 bytes, not a formatted string. */
    public static function binary(): string
    {
        $ms = (int) (microtime(true) * 1000);

        if ($ms === self::$lastMs) {
            self::$counter++;
            if (self::$counter > 0x0FFF) {
                // >4096 ids in one ms: borrow the next millisecond
                self::$lastMs = ++$ms;
                self::$counter = 0;
            }
        } elseif ($ms < self::$lastMs) {
            // NTP stepped us backwards: hold the line, keep counting
            $ms = self::$lastMs;
            self::$counter++;
        } else {
            self::$lastMs = $ms;
            // seed low so a busy ms has headroom before it borrows
            self::$counter = random_int(0, 0x00FF);
        }

        // 48-bit ts | 4-bit version (0111) | 12-bit counter
        $hi = pack('J', ($ms << 16) | 0x7000 | (self::$counter & 0x0FFF));

        $lo = random_bytes(8);
        $lo[0] = chr((ord($lo[0]) & 0x3F) | 0x80); // variant 10

        return $hi . $lo;
    }

    public static function format(string $bin): string
    {
        $h = bin2hex($bin);
        return substr($h, 0, 8) . '-' . substr($h, 8, 4) . '-'
             . substr($h, 12, 4) . '-' . substr($h, 16, 4) . '-'
             . substr($h, 20, 12);
    }

    public static function createdAt(string $bin): float
    {
        ['hi' => $hi] = unpack('Jhi', substr($bin, 0, 8));
        return ($hi >> 16) / 1000.0;
    }
}

One note on the pack('J', ...) line, since it looks like it should overflow. Current epoch milliseconds are about 1.77e12, which is under 2^41; shifting left 16 gives a value under 2^57, comfortably inside PHP's signed 64-bit int. This code stops being correct in the year 10889, which we have accepted as technical debt.

The counter is static, which means it is per-process. Under LiteSpeed's PHP SAPI each worker holds its own, and workers on different machines obviously do not coordinate. That is fine: two workers can produce ids in the same millisecond with the same counter value, and the 62 random bits still separate them. The counter only guarantees ordering within a generator, which is the property keyset pagination actually needs.

Making Postgres agree

Postgres 18 ships uuidv7() and uuid_extract_timestamp() in core, which is the whole story if you are already there. We are on 17 in production, so the defaults come from a SQL function. This is the widely-circulated overlay() trick, and it is worth reading closely because it is doing bit surgery on the output of gen_random_uuid():

CREATE OR REPLACE FUNCTION uuid_generate_v7() RETURNS uuid AS $$
  SELECT encode(
    set_bit(
      set_bit(
        overlay(
          uuid_send(gen_random_uuid())
          PLACING substring(
            int8send((extract(epoch FROM clock_timestamp()) * 1000)::bigint)
            FROM 3
          )
          FROM 1 FOR 6
        ),
        52, 1
      ),
      53, 1
    ),
    'hex')::uuid;
$$ LANGUAGE sql VOLATILE;

CREATE TABLE videos (
    id            uuid PRIMARY KEY DEFAULT uuid_generate_v7(),
    provider      text        NOT NULL,
    provider_ref  text        NOT NULL,
    region        char(2)     NOT NULL,
    title         text        NOT NULL,
    title_lang    text        NOT NULL DEFAULT 'und',
    duration_s    integer     NOT NULL,
    published_at  timestamptz NOT NULL,
    ingested_at   timestamptz NOT NULL DEFAULT now(),
    stats         jsonb       NOT NULL DEFAULT '{}'::jsonb,
    UNIQUE (provider, provider_ref)
) WITH (fillfactor = 90);

-- PG < 18: pull the timestamp back out of the key
CREATE OR REPLACE FUNCTION uuid_v7_ts(u uuid) RETURNS timestamptz AS $$
  SELECT to_timestamp(
    ('x0000' || substring(u::text, 1, 8) || substring(u::text, 10, 4))::bit(64)::bigint
    / 1000.0
  );
$$ LANGUAGE sql IMMUTABLE STRICT;

The database default is a safety net, not the main path — 99% of our ids arrive from the application because we need them before the INSERT. But having a correct default means a psql session, a migration, or an ops script never accidentally introduces a v4 into a table whose whole design assumes v7.

Store it as uuid, never as text. The uuid type is 16 bytes and compares with memcmp; the canonical string form is 37 bytes with a varlena header and compares with collation-aware machinery. On 41 million rows that difference alone was over 800 MB of heap.

Time-range scans without a timestamp index

Because the leading 48 bits are a timestamp, a range of ids is a range of time. You can build the bounds without any function call on the column, which keeps the plan on a plain index scan:

-- everything ingested in the last 6 hours, straight off the PK
SELECT id, title, region
FROM videos
WHERE id >= (
        lpad(to_hex((extract(epoch FROM now() - interval '6 hours') * 1000)::bigint), 12, '0')
        || '7000-8000-000000000000'
      )::uuid
ORDER BY id DESC
LIMIT 200;

That is not a replacement for published_at — upstream publish time and our ingest time are different things, and we index both. But it removed an entire class of "what did the 04:00 run actually touch" operational query from needing an index at all, and it made keyset pagination trivial: WHERE id < :cursor ORDER BY id DESC LIMIT 40, with the cursor being the last id on the previous page. No OFFSET, no ties, no second sort column.

Bulk ingest from the Go workers

Our regional fetchers are Go binaries that normalize provider payloads and hand batches to Postgres over COPY. Generating ids in the worker means the FTS payload and the Cloudflare purge list can be built in the same pass, before the transaction commits.

package ingest

import (
    "context"
    "crypto/rand"
    "encoding/binary"
    "sync"
    "time"

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
)

type V7Gen struct {
    mu     sync.Mutex
    lastMs int64
    seq    uint16
}

func (g *V7Gen) Next() [16]byte {
    g.mu.Lock()
    ms := time.Now().UnixMilli()
    switch {
    case ms == g.lastMs:
        g.seq++
        if g.seq > 0x0FFF {
            g.lastMs++
            ms, g.seq = g.lastMs, 0
        }
    case ms < g.lastMs:
        ms = g.lastMs
        g.seq++
    default:
        g.lastMs, g.seq = ms, 0
    }
    seq := g.seq & 0x0FFF
    g.mu.Unlock()

    var id [16]byte
    binary.BigEndian.PutUint64(id[0:8], uint64(ms)<<16|0x7000|uint64(seq))
    rand.Read(id[8:16])
    id[8] = id[8]&0x3F | 0x80
    return id
}

type Video struct {
    Provider, Ref, Region, Title, Lang string
    Duration                          int32
    PublishedAt                       time.Time
    ID                                [16]byte
}

func Bulk(ctx context.Context, pool *pgxpool.Pool, g *V7Gen, rows []Video) (int64, error) {
    src := pgx.CopyFromSlice(len(rows), func(i int) ([]any, error) {
        rows[i].ID = g.Next()
        r := rows[i]
        return []any{r.ID, r.Provider, r.Ref, r.Region, r.Title, r.Lang, r.Duration, r.PublishedAt}, nil
    })

    return pool.CopyFrom(ctx,
        pgx.Identifier{"videos_stage"},
        []string{"id", "provider", "provider_ref", "region", "title", "title_lang", "duration_s", "published_at"},
        src)
}

We copy into an unlogged videos_stage table and then do a single INSERT ... SELECT ... ON CONFLICT (provider, provider_ref) DO UPDATE. Because the staged ids are already in ascending order, the merge into the real table appends to the right edge of the PK index instead of scattering.

The numbers

Measured on our primary — 4 vCPU, 16 GB RAM, NVMe, Postgres 17.4, shared_buffers = 4GB, 41.3M rows — comparing the v4 table against the rebuilt v7 table with identical data:

PK index size: 2.91 GB → 1.68 GB (−42%). Most of that is fill factor: the v7 index settles near 90% occupancy because it only ever appends.
COPY + merge of a 180k-row batch: 41.2s → 12.6s.
WAL generated per nightly ingest: 6.2 GB → 1.9 GB. Fewer distinct pages touched after each checkpoint means far fewer full-page writes.
PK index buffer hit ratio during ingest: 71% → 99.3%.
VACUUM on the videos table: 14 min → 5 min, mostly because there is less index to scan.

The end-to-end result was the one we cared about: the ingest window went from 88 minutes back to 31, and the morning cache warm stopped colliding with it.

One honest caveat on these numbers. A rebuilt index always looks good against a production index that has been churning for two years; some of that 42% is "we ran REINDEX", not "we switched to v7". Three months later the v7 index is at 1.79 GB and the fill factor has held, which is the part that actually proves the point — the v4 index was back over 2.4 GB three months after its last rebuild.

Feeding the SQLite FTS5 edge indexes

Each region gets a read-only SQLite file shipped to the edge, holding a trimmed projection of the video table plus an FTS5 index. CJK is the reason we use SQLite here at all: with a trigram tokenizer we get usable substring matching for Japanese, Korean and Chinese titles without a word segmenter, which Postgres full-text would need.

The same 16 bytes carry across. SQLite stores them as a BLOB and orders them with memcmp, so the cursor pagination logic is literally the same query shape on both sides:

<?php
declare(strict_types=1);

$edge = new PDO('sqlite:' . $path, options: [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
$edge->exec('PRAGMA journal_mode = WAL');
$edge->exec('PRAGMA synchronous = NORMAL');

$edge->exec('CREATE TABLE IF NOT EXISTS videos (
    rowid  INTEGER PRIMARY KEY,
    uuid   BLOB NOT NULL UNIQUE,
    title  TEXT NOT NULL,
    lang   TEXT NOT NULL,
    ts_ms  INTEGER NOT NULL
)');

$edge->exec("CREATE VIRTUAL TABLE IF NOT EXISTS videos_fts USING fts5(
    title,
    lang UNINDEXED,
    content='videos',
    content_rowid='rowid',
    tokenize='trigram case_sensitive 0'
)");

$ins = $edge->prepare(
    'INSERT INTO videos (uuid, title, lang, ts_ms) VALUES (:u, :t, :l, :ts)
     ON CONFLICT(uuid) DO UPDATE SET title = excluded.title, lang = excluded.lang'
);

$edge->beginTransaction();
foreach ($batch as $row) {
    $bin = $row['id']; // raw 16 bytes straight from Postgres
    $ins->bindValue(':u', $bin, PDO::PARAM_LOB);
    $ins->bindValue(':t', $row['title']);
    $ins->bindValue(':l', $row['title_lang']);
    $ins->bindValue(':ts', (int) (UuidV7::createdAt($bin) * 1000), PDO::PARAM_INT);
    $ins->execute();
}
$edge->commit();

// keyset page, newest first — identical shape to the Postgres query
$page = $edge->prepare(
    'SELECT uuid, title FROM videos WHERE uuid < :cursor ORDER BY uuid DESC LIMIT 40'
);

The ts_ms column is redundant — it is derivable from the uuid — but SQLite has no cheap way to slice a BLOB prefix into an integer in an index, so we materialize it for range filters. That is a deliberate 8 bytes per row.

Backfilling 41 million rows

The migration ran as an online dual-write: add id_v7 uuid, backfill in batches, swap the constraint, drop the old column. The interesting part is what value to assign to a historical row.

Our first attempt used clock_timestamp() during the backfill, which gave every legacy row an id timestamped to migration night. Technically valid; operationally useless, because it destroyed the one property we had just paid for. The fix was to derive the timestamp from published_at:

import os
import struct
from uuid import UUID

import psycopg

BATCH = 20_000


def uuid7_at(ms: int, seq: int) -> UUID:
    hi = (ms << 16) | 0x7000 | (seq & 0x0FFF)
    lo = bytearray(os.urandom(8))
    lo[0] = (lo[0] & 0x3F) | 0x80
    return UUID(bytes=struct.pack('>Q', hi) + bytes(lo))


with psycopg.connect(os.environ['PG_DSN']) as conn:
    total = 0
    while True:
        with conn.cursor() as cur:
            cur.execute(
                '''SELECT legacy_id, (EXTRACT(EPOCH FROM published_at) * 1000)::bigint
                   FROM videos
                   WHERE id_v7 IS NULL
                   ORDER BY published_at
                   LIMIT %s''',
                (BATCH,),
            )
            rows = cur.fetchall()
            if not rows:
                break

            payload = [
                (uuid7_at(int(ms), i), legacy_id)
                for i, (legacy_id, ms) in enumerate(rows)
            ]
            cur.executemany(
                'UPDATE videos SET id_v7 = %s WHERE legacy_id = %s', payload
            )
        conn.commit()
        total += len(rows)
        print(f'{total} rows backfilled', flush=True)

Batches are ordered by published_at and the enumerate index becomes the counter, so ids come out monotonic across the whole backfill. The counter wraps every 4096 rows within a batch of 20,000 — that is intentional and harmless, since uniqueness lives in the 62 random bits and rows sharing a millisecond have no meaningful order anyway.

The backfill took eleven hours in 20k-row batches with a two-second sleep between them, throttled to keep replica lag under five seconds. We did not attempt to preserve any relationship between old and new ids; instead we kept legacy_id as a plain indexed column for six months so old cached URLs and partner integrations kept resolving.

Four things that bit us

A v7 id publishes its own creation time. Anyone holding an id knows to the millisecond when the row was created. For public video pages we route on slugs and never expose the uuid, so the Cloudflare cache key and the LiteSpeed page cache path contain no ids at all. But we had one internal admin endpoint leaking ids into an HTML data attribute, which meant a scraper could have reconstructed our exact ingest schedule. Fixed, but it is exactly the kind of thing that does not show up in a schema review.

Never use a v7 id as a secret. Only 62 bits are random and the rest is guessable. Session tokens, password reset links, unsubscribe links — those stay v4, or better, random_bytes(32). Sequential-ish keys and unguessable tokens are different jobs.

ORDER BY id is ingest order, not publish order. A video published in 2019 that we first saw last week sorts as last week. We hit this the day after the cutover when the "newest" feed for one region filled with old catalog rows that had just been backfilled. Feeds sort by published_at; only internal tooling sorts by id.

The right edge of the index is now a hot spot. Every insert targets the same rightmost leaf page, so at high write concurrency you trade random-page thrash for buffer contention on one page. At our peak of about 2,000 inserts/sec this is nowhere near a problem, and the WAL savings dwarf it. If you are writing tens of thousands of rows per second from many connections, benchmark it rather than assuming — that is the one workload where v4's randomness is genuinely an advantage.

Conclusion

UUID v7 is not a clever trick, it is a correction of a mismatch: we needed decentralized id generation and accidentally bought full randomness along with it. Putting a millisecond timestamp in the leading 48 bits gives back everything a B-tree wants — append-only inserts, high fill factor, cache locality, cheap WAL — while keeping the property we actually chose UUIDs for.

If you are on Postgres 18, uuidv7() is already there and the migration is a DEFAULT change plus a backfill. On 17 or older, the SQL function above plus an application-side generator gets you the same behavior. Either way, derive backfilled ids from a real event timestamp rather than migration time, keep the id out of anything user-facing, and measure your index fill factor three months later rather than the day of the rebuild. That last number is the one that tells you whether it worked.

Using etcd for Distributed Configuration of Video Region Routers

ahmet gedik — Sun, 16 Aug 2026 04:00:02 +0000

Our region router used to live in a PHP array. config/regions.php mapped a two-letter country code to an upstream video index shard, a CDN hostname, and a set of category weights. It worked fine when there was one origin box. It stopped working the day we had four, spread across three continents, and a Cloudflare Worker in front deciding which one to hit.

The failure mode was mundane and expensive. Someone (me) shifted BR from the us-east shard to a new sa-east shard, deployed via FTP to three of the four hosts, and forgot the fourth. For nine hours, roughly a quarter of Brazilian traffic on DailyWatch hit an origin that had no Portuguese-language trending rows cached, served a cold SQLite FTS5 query on every request, and got a 4-second TTFB. Nobody paged us because nothing was down. The graphs just sagged.

The fix was not "deploy more carefully." The fix was to stop treating routing config as code. Routing config changes on a different clock than code does — code ships weekly, region routing changes when a shard gets hot at 3am. Those need different substrates. This is what pushing region routing into etcd looked like, including the parts that were annoying.

Why etcd and Not Redis, Consul, or a Database Table

I evaluated four options and the choice came down to one property: watch semantics with a resumable revision.

A regions table in SQLite/MySQL. Requires polling. Polling at 5s across N origins is fine, but there's no way to know you missed an update, and there's no atomic multi-key write. Rejected.
Redis pub/sub. Fire-and-forget. If your subscriber is reconnecting during the publish, the message is gone and you're silently stale — exactly the failure we were trying to eliminate. Redis Streams fixes this but then you're building a changelog by hand.
Consul KV. Genuinely fine. Blocking queries give you long-poll watches with an index. I'd have picked it if we were already running Consul for service discovery. We weren't, and Consul's agent-per-node model is more operational surface than a config store needs.
etcd. Watches are streamed over gRPC and carry a global monotonic revision. If your watcher dies, you reconnect with start_revision = last_seen + 1 and etcd replays everything you missed — or tells you explicitly that history was compacted away, at which point you know to do a full resync. That explicit "you are now stale" signal is the whole ballgame.

etcd also gives you transactions (compare-and-swap across multiple keys) and leases, both of which turned out to matter more than I expected.

The cost: etcd is a Raft cluster. Three nodes minimum for any real availability, five if you want to survive two failures. It is latency-sensitive to disk fsync. Do not put it on the same burstable-IO volume as your database. This is a real operational commitment, not a docker run.

Key Layout Is the Actual Design Work

etcd is a flat sorted byte-keyspace with range queries. There are no directories, only prefix scans. So the key layout is your schema, and getting it wrong means either fetching too much or making N round trips.

What we landed on:

/dw/routing/v1/region/AU    -> {"shard":"ap-southeast","cdn":"cdn-ap.","weight":1.0,"tier":"a"}
/dw/routing/v1/region/BR    -> {"shard":"sa-east",...}
/dw/routing/v1/region/_default -> {"shard":"us-east",...}
/dw/routing/v1/shard/us-east   -> {"origin":"10.4.1.20","healthy":true,"max_rps":800}
/dw/routing/v1/shard/ap-southeast -> {...}
/dw/routing/v1/epoch        -> "5121"
/dw/routing/v1/lock/rebalance -> (lease-held, ephemeral)

Four rules that came out of doing this badly first:

Version the prefix (v1). When the value schema changes incompatibly, you write v2 alongside, migrate readers, then delete v1. Trying to do in-place schema evolution on a live config store is how you get a split-brain fleet.
Split region → shard from shard → origin. Originally I embedded the origin IP in each region value. Moving one shard's origin meant rewriting 40 keys in a transaction. Now it's one key.
Keep values small. etcd's default --max-request-bytes is 1.5 MiB and every value is replicated through Raft and held in memory. Region config is hundreds of bytes. If you're tempted to put a blob in etcd, put it in object storage and put the URL in etcd.
One key for the epoch. A monotonically increasing counter bumped on every routing change. Cheap way for anything — a health check, a log line, a debug header — to answer "which config generation is this box on?" without diffing the whole map.

The Watcher Sidecar

PHP-FPM under LiteSpeed is a process-per-request model. There is no long-lived process to hold a gRPC watch stream, and no shared memory you'd want to write from a request handler. So the watcher is a separate long-running daemon that maintains a local snapshot file, and PHP reads the snapshot. The daemon is the only thing that talks to etcd.

Go, because the official etcd client is Go and the watch reconnection logic is already correct there:

package main

import (
    "context"
    "encoding/json"
    "log"
    "os"
    "path/filepath"
    "time"

    clientv3 "go.etcd.io/etcd/client/v3"
)

const prefix = "/dw/routing/v1/"

type snapshot struct {
    Revision int64             `json:"revision"`
    Written  int64             `json:"written_at"`
    Keys     map[string]string `json:"keys"`
}

func main() {
    out := os.Getenv("ROUTING_SNAPSHOT")
    if out == "" {
        out = "/var/lib/dailywatch/routing.json"
    }

    cli, err := clientv3.New(clientv3.Config{
        Endpoints:            []string{"10.4.0.11:2379", "10.4.0.12:2379", "10.4.0.13:2379"},
        DialTimeout:          5 * time.Second,
        DialKeepAliveTime:    30 * time.Second,
        DialKeepAliveTimeout: 5 * time.Second,
    })
    if err != nil {
        log.Fatalf("etcd dial: %v", err)
    }
    defer cli.Close()

    state := snapshot{Keys: map[string]string{}}

    for {
        rev, err := resync(cli, &state)
        if err != nil {
            log.Printf("resync failed, retrying: %v", err)
            time.Sleep(2 * time.Second)
            continue
        }
        if err := write(out, state); err != nil {
            log.Printf("snapshot write: %v", err)
        }
        log.Printf("synced %d keys at revision %d", len(state.Keys), rev)

        if err := follow(cli, &state, out); err != nil {
            // Compaction or stream death: fall back to a full resync.
            log.Printf("watch ended: %v", err)
        }
    }
}

func resync(cli *clientv3.Client, s *snapshot) (int64, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    resp, err := cli.Get(ctx, prefix, clientv3.WithPrefix())
    if err != nil {
        return 0, err
    }
    s.Keys = make(map[string]string, len(resp.Kvs))
    for _, kv := range resp.Kvs {
        s.Keys[string(kv.Key)] = string(kv.Value)
    }
    s.Revision = resp.Header.Revision
    return s.Revision, nil
}

func follow(cli *clientv3.Client, s *snapshot, out string) error {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    ch := cli.Watch(ctx, prefix,
        clientv3.WithPrefix(),
        clientv3.WithRev(s.Revision+1),
        clientv3.WithPrevKV(),
    )

    for wr := range ch {
        if err := wr.Err(); err != nil {
            return err // includes ErrCompacted -> caller does a full resync
        }
        for _, ev := range wr.Events {
            k := string(ev.Kv.Key)
            switch ev.Type {
            case clientv3.EventTypePut:
                s.Keys[k] = string(ev.Kv.Value)
            case clientv3.EventTypeDelete:
                delete(s.Keys, k)
            }
        }
        s.Revision = wr.Header.Revision
        if err := write(out, *s); err != nil {
            log.Printf("snapshot write: %v", err)
        }
    }
    return nil
}

// write is atomic: temp file in the same directory, then rename.
func write(path string, s snapshot) error {
    s.Written = time.Now().Unix()
    buf, err := json.Marshal(s)
    if err != nil {
        return err
    }
    tmp, err := os.CreateTemp(filepath.Dir(path), ".routing-*")
    if err != nil {
        return err
    }
    defer os.Remove(tmp.Name())

    if _, err := tmp.Write(buf); err != nil {
        tmp.Close()
        return err
    }
    if err := tmp.Sync(); err != nil {
        tmp.Close()
        return err
    }
    if err := tmp.Close(); err != nil {
        return err
    }
    if err := os.Chmod(tmp.Name(), 0o644); err != nil {
        return err
    }
    return os.Rename(tmp.Name(), path)
}

Three details that are not decoration:

WithRev(s.Revision+1). Without this you start watching from now and silently lose every event between your Get and your Watch. That window is small and it will bite you exactly once, in production, at a bad time.
wr.Err() returning ErrCompacted. etcd garbage-collects old revisions. If your watcher was down longer than the compaction window, history is gone. The only correct response is a full resync — which is why follow returns the error up to the retry loop rather than trying to be clever.
Write temp file, fsync, then rename. rename(2) within a filesystem is atomic. PHP-FPM workers reading concurrently will see either the whole old file or the whole new file, never a truncated one. Writing in place would eventually hand some request a half-written JSON document.

Reading the Snapshot from PHP Without Melting the Disk

The PHP side never talks to etcd. It reads a local file, and it does so through APCu with a stat-based freshness check, so the common path is a memory lookup plus one stat().

<?php
declare(strict_types=1);

final class RegionRouter
{
    private const SNAPSHOT = '/var/lib/dailywatch/routing.json';
    private const CACHE_KEY = 'routing:v1';
    private const STALE_AFTER = 900; // seconds

    /** @var array{revision:int,written_at:int,keys:array<string,string>}|null */
    private static ?array $memo = null;

    public function route(string $country): RouteDecision
    {
        $snap = $this->snapshot();
        $cc = strtoupper(substr($country, 0, 2));

        $raw = $snap['keys']['/dw/routing/v1/region/' . $cc]
            ?? $snap['keys']['/dw/routing/v1/region/_default']
            ?? null;

        if ($raw === null) {
            return RouteDecision::fallback('no-default-key');
        }

        $region = json_decode($raw, true, 8, JSON_THROW_ON_ERROR);
        $shardRaw = $snap['keys']['/dw/routing/v1/shard/' . $region['shard']] ?? null;

        if ($shardRaw === null) {
            // Dangling reference: region points at a shard that no longer exists.
            return RouteDecision::fallback('dangling-shard:' . $region['shard']);
        }

        $shard = json_decode($shardRaw, true, 8, JSON_THROW_ON_ERROR);
        if (($shard['healthy'] ?? false) !== true) {
            return RouteDecision::fallback('shard-unhealthy:' . $region['shard']);
        }

        return new RouteDecision(
            origin:   $shard['origin'],
            cdnHost:  $region['cdn'],
            weight:   (float) ($region['weight'] ?? 1.0),
            revision: $snap['revision'],
        );
    }

    /** @return array{revision:int,written_at:int,keys:array<string,string>} */
    private function snapshot(): array
    {
        clearstatcache(true, self::SNAPSHOT);
        $mtime = @filemtime(self::SNAPSHOT);

        if ($mtime === false) {
            return self::$memo ?? ['revision' => 0, 'written_at' => 0, 'keys' => []];
        }

        if (self::$memo !== null && self::$memo['written_at'] >= $mtime) {
            return self::$memo;
        }

        $cached = apcu_fetch(self::CACHE_KEY, $ok);
        if ($ok && is_array($cached) && $cached['written_at'] >= $mtime) {
            return self::$memo = $cached;
        }

        $body = @file_get_contents(self::SNAPSHOT);
        if ($body === false) {
            return self::$memo ?? ['revision' => 0, 'written_at' => 0, 'keys' => []];
        }

        /** @var array{revision:int,written_at:int,keys:array<string,string>} $decoded */
        $decoded = json_decode($body, true, 16, JSON_THROW_ON_ERROR);

        if (time() - $decoded['written_at'] > self::STALE_AFTER) {
            error_log(sprintf(
                'routing snapshot stale by %ds (rev %d)',
                time() - $decoded['written_at'],
                $decoded['revision']
            ));
        }

        apcu_store(self::CACHE_KEY, $decoded, 60);
        return self::$memo = $decoded;
    }
}

final class RouteDecision
{
    public function __construct(
        public readonly string $origin,
        public readonly string $cdnHost,
        public readonly float $weight,
        public readonly int $revision,
        public readonly ?string $degradedReason = null,
    ) {}

    public static function fallback(string $reason): self
    {
        return new self('127.0.0.1', '', 1.0, 0, $reason);
    }
}

The written_at >= $mtime comparison rather than == matters: the daemon can write twice within the same second, and mtime granularity on some filesystems will not distinguish them. Using >= means a stale memo can survive at most until the next second boundary, which for routing config is fine.

The part I'd emphasize to anyone building this: the failure path is the product. RouteDecision::fallback() gets hit when the snapshot is missing, when a region points at a deleted shard, when a shard is marked unhealthy. Each carries a distinct reason string, which we emit as a X-DW-Route-Degraded header on internal requests and count in a Prometheus gauge. When something is wrong you want to know which wrongness, not that routing "failed."

Writing Config Safely with Transactions and Leases

Reads are the easy half. Writes are where you can break the whole fleet in one command, so the control-plane tooling does three things: validates before writing, writes atomically, and holds a lock so two operators can't rebalance simultaneously.

#!/usr/bin/env python3
"""Rebalance regions between shards. Validates, locks, writes atomically."""
import json
import sys

import etcd3  # pip install etcd3-py-client

PREFIX = "/dw/routing/v1/"
LOCK_KEY = PREFIX + "lock/rebalance"
EPOCH_KEY = PREFIX + "epoch"


def load_all(client):
    out = {}
    for value, meta in client.get_prefix(PREFIX):
        out[meta.key.decode()] = value.decode()
    return out


def validate(keys):
    """Every region must point at a shard that exists. A default must exist."""
    errors = []
    shards = {
        k.rsplit("/", 1)[1]
        for k in keys
        if k.startswith(PREFIX + "shard/")
    }
    if PREFIX + "region/_default" not in keys:
        errors.append("missing region/_default")

    for key, raw in keys.items():
        if not key.startswith(PREFIX + "region/"):
            continue
        try:
            region = json.loads(raw)
        except json.JSONDecodeError as exc:
            errors.append(f"{key}: invalid json: {exc}")
            continue
        target = region.get("shard")
        if target not in shards:
            errors.append(f"{key}: points at unknown shard {target!r}")
        if not 0.0 < float(region.get("weight", 1.0)) <= 10.0:
            errors.append(f"{key}: weight out of range")
    return errors


def rebalance(client, moves):
    """moves: {'BR': 'sa-east', 'CL': 'sa-east'}"""
    current = load_all(client)

    proposed = dict(current)
    for cc, shard in moves.items():
        key = f"{PREFIX}region/{cc}"
        if key not in proposed:
            raise SystemExit(f"unknown region {cc}")
        region = json.loads(proposed[key])
        region["shard"] = shard
        proposed[key] = json.dumps(region, separators=(",", ":"), sort_keys=True)

    errors = validate(proposed)
    if errors:
        for err in errors:
            print(f"INVALID: {err}", file=sys.stderr)
        raise SystemExit(1)

    epoch = int(current.get(EPOCH_KEY, "0"))

    # Lock so two operators cannot interleave rebalances.
    lease = client.lease(30)
    got = client.transaction(
        compare=[client.transactions.version(LOCK_KEY) == 0],
        success=[client.transactions.put(LOCK_KEY, b"held", lease=lease)],
        failure=[],
    )
    if not got[0]:
        raise SystemExit("rebalance already in progress")

    try:
        changed = [f"{PREFIX}region/{cc}" for cc in moves]
        ok, _ = client.transaction(
            # Refuse if anyone modified these keys since we read them.
            compare=[
                client.transactions.value(EPOCH_KEY) == str(epoch).encode()
            ],
            success=[
                client.transactions.put(k, proposed[k].encode()) for k in changed
            ] + [
                client.transactions.put(EPOCH_KEY, str(epoch + 1).encode())
            ],
            failure=[],
        )
        if not ok:
            raise SystemExit("epoch changed under us; re-run")
        print(f"moved {len(moves)} regions, epoch {epoch} -> {epoch + 1}")
    finally:
        lease.revoke()  # releases the lock even if we blew up above


if __name__ == "__main__":
    rebalance(etcd3.client(host="10.4.0.11", port=2379), {"BR": "sa-east"})

What each mechanism buys:

The epoch compare-and-swap turns the write into optimistic concurrency control. Read epoch, compute changes, write only if epoch is unchanged. Two concurrent rebalances: one wins, one gets told to retry against fresh state. No lost updates.
The lease-backed lock is belt-and-braces on top. Leases expire — if the operator's laptop closes mid-run, the lock evaporates after 30 seconds rather than wedging the system forever. This is the thing you cannot build correctly with a plain key.
Validation runs against the proposed full state, not the diff. A move is invalid if it creates a dangling reference, and you can only see that by checking the whole map. This one check would have caught the original BR incident before it shipped.

Operational Things That Actually Bit Us

Compaction defaults will surprise you. etcd retains all revisions until compacted. Run with --auto-compaction-retention=1h (or --auto-compaction-mode=revision --auto-compaction-retention=10000). Without it, the backend DB grows until it hits --quota-backend-bytes (2 GiB default) and the cluster goes read-only with mvcc: database space exceeded. Recovering means compact, then etcdctl defrag, then etcdctl alarm disarm. Do this in a staging cluster once so your hands know the sequence.

Defrag is blocking, per member. etcdctl defrag locks the member while it runs. Do it one member at a time with --endpoints pointed at a single node, never --cluster.

Watch out for the Cloudflare layer disagreeing with etcd. We push a subset of routing state to a Cloudflare Worker KV namespace so edge decisions don't require an origin round trip, and KV is eventually consistent with up to a ~60s propagation delay. For the first month, etcd and edge KV would briefly disagree after a rebalance and requests would ping-pong. The fix was to make the origin authoritative and idempotent: if the edge routes a request to the wrong origin, that origin serves it correctly anyway rather than redirecting. Never let two config planes argue in the request path.

Give the watcher a dead-man's switch. The daemon writes written_at on every snapshot flush, including no-op flushes on a 60-second timer. A separate check alerts if written_at falls more than five minutes behind wall clock. Silent staleness is the exact failure we started with, and a watcher that dies quietly reproduces it perfectly.

Don't put secrets in etcd because it's convenient. Values are encrypted in transit with TLS but at rest they're plaintext in the bbolt file unless you've set up encryption providers. Routing config is not secret. API keys are. Keep them separate.

Client TLS is not optional. etcd with no auth on a private network is one misconfigured security group away from anyone rewriting your routing table. Use --client-cert-auth with per-service certs and RBAC roles scoped to the prefix each service actually needs — the watcher gets read on /dw/routing/v1/, the rebalance tool gets write.

What Changed

Routing changes went from a 20-minute FTP deploy across four hosts, with a real chance of partial application, to a single transaction that every origin picks up in under a second. The epoch counter appears in our health endpoint, so "are all four boxes on the same config?" is one curl per host instead of a diff of four files.

The honest accounting: we added a three-node Raft cluster to operate, a Go daemon per origin, and a class of failure (stale snapshot) that didn't exist before. For a two-server setup this is clearly not worth it — a config file in git and a deploy script is fine, and I'd say so. It became worth it at the point where the number of places config had to land exceeded the number of places a human could reliably check.

If you're heading this way, the two things I'd insist on: version your key prefix from day one, and build the degraded-path reason codes before you build the happy path. The happy path works on the first try. The degraded path is what you'll be reading at 3am.

Adding Typo Tolerant Video Title Search With OpenSearch and SQLite FTS5

ahmet gedik — Sat, 15 Aug 2026 19:00:01 +0000

Our search log had one row I could not stop staring at: strangr things sesaon 5 trailr, submitted 412 times in a single month across eight regions, returning exactly zero results. Not a thin result set — zero. The correctly spelled version of the same query returned 340 videos with a 71% click-through rate. Somewhere in that gap sat a few thousand sessions a month that landed on an empty state and bounced.

I run TrendVidStream, a streaming-discovery site that ingests trending video metadata from eight regions on a staggered cron and serves it from a single SQLite file on shared LiteSpeed hosting, deployed by FTP. Search was SQLite FTS5 and had been for two years. FTS5 is genuinely excellent at what it does. What it does not do is edit distance, and video search is drowning in edit distance: mobile keyboards, transliterated titles, and half the world typing English show names phonetically.

This is what I actually built, what it cost, and the three things I got wrong first.

What FTS5 gives you and where it stops

Here is the search path we shipped for two years. It is not bad code. It is just structurally incapable of the thing I needed.

<?php
declare(strict_types=1);

// app/Search/Fts5Search.php — the version that ran for two years.
final class Fts5Search
{
    public function __construct(private readonly PDO $db) {}

    /** @return list<array{video_id:string,title:string,score:float}> */
    public function search(string $q, int $limit = 40): array
    {
        // FTS5 MATCH is its own grammar; a stray quote kills the statement.
        $terms = preg_split('/\s+/u', trim($q), -1, PREG_SPLIT_NO_EMPTY) ?: [];
        if ($terms === []) {
            return [];
        }

        $match = implode(' ', array_map(
            static fn (string $t): string => '"' . str_replace('"', '""', $t) . '"*',
            $terms
        ));

        $stmt = $this->db->prepare(
            'SELECT v.video_id, v.title, bm25(video_fts, 8.0, 1.0) AS score
               FROM video_fts
               JOIN videos v ON v.rowid = video_fts.rowid
              WHERE video_fts MATCH :m
           ORDER BY score
              LIMIT :lim'
        );
        $stmt->bindValue(':m', $match);
        $stmt->bindValue(':lim', $limit, PDO::PARAM_INT);
        $stmt->execute();

        return $stmt->fetchAll(PDO::FETCH_ASSOC);
    }
}

// search('season') -> 340 rows in 9ms
// search('sesaon') -> 0 rows. Every time. Forever.

The trailing * gives prefix matching, so seas finds season. That covers people who stop typing. It does nothing for people who type the wrong characters.

I tried the two obvious escapes before renting anything:

The trigram tokenizer. SQLite ships one, and it enables substring matching and LIKE acceleration. It is not fuzzy matching. season produces the trigrams sea eas aso son; sesaon produces ses esa sao aon. Overlap: zero. A transposition destroys every trigram that spans it, which is exactly the typo class I needed to survive.
spellfix1. This is the right tool in the SQLite family — it does edit distance and phonetic matching. It is also a loadable extension, and our hosting does not permit sqlite3_load_extension. Dead end on shared hosting, and I was not moving four sites to a VPS to fix search.

So the choice narrowed to: put a real search engine next to the app, or accept the zero-result rate. I went with OpenSearch, single node, and kept FTS5 in place as the fallback. That second half turned out to matter more than the first.

Sizing it before spending money

A thing I have learned the expensive way: measure the corpus before you pick the instance.

1.24M video rows across eight regions, deduplicated to 890K unique videos
Average title length 61 characters; channel names average 18
Only title, channel, regions, category_id, view_count, published_at need to be searchable or filterable
Everything else — descriptions, thumbnails, duration — stays in SQLite and gets joined back after we have IDs

That last bullet is the one that keeps the bill small. OpenSearch is an ID resolver here, not a document store. It answers "which 40 video IDs, in what order" and then PHP hydrates those 40 rows from SQLite by primary key, which SQLite does in well under a millisecond. The index ends up around 340MB with one shard and zero replicas, which fits comfortably on a 2GB VPS running one node.

Network placement matters more than instance size. Our web hosts and the OpenSearch node are in the same European metro; cross-continent would have added 90ms to every keystroke on the typeahead endpoint. Access is HTTPS with basic auth plus an IP allowlist for the four origin IPs. There is no public route to port 9200, and there never will be.

The mapping is the whole design

Everything about typo tolerance is decided at index time. Fuzziness at query time only works if the analyzed tokens are shaped to make it work.

<?php
declare(strict_types=1);

// tools/search/create_index.php — run manually on every mapping change.
// Versioned index name + alias, so a reindex is an atomic pointer swap.
$version = 'videos_v3';

$body = [
    'settings' => [
        'index' => [
            'number_of_shards'   => 1,
            'number_of_replicas' => 0,
            'max_ngram_diff'     => 13,
            'refresh_interval'   => '30s', // we are not a live feed
        ],
        'analysis' => [
            'normalizer' => [
                'raw_lower' => [
                    'type'   => 'custom',
                    'filter' => ['lowercase', 'asciifolding'],
                ],
            ],
            'filter' => [
                'edge_2_15' => ['type' => 'edge_ngram', 'min_gram' => 2, 'max_gram' => 15],
            ],
            'analyzer' => [
                'title_folded' => [
                    'tokenizer' => 'standard',
                    'filter'    => ['lowercase', 'asciifolding'],
                ],
                'title_prefix' => [
                    'tokenizer' => 'standard',
                    'filter'    => ['lowercase', 'asciifolding', 'edge_2_15'],
                ],
            ],
        ],
    ],
    'mappings' => [
        'properties' => [
            'title' => [
                'type'     => 'text',
                'analyzer' => 'title_folded',
                'fields'   => [
                    // typeahead only; search_analyzer must NOT re-ngram the query
                    'prefix' => [
                        'type'            => 'text',
                        'analyzer'        => 'title_prefix',
                        'search_analyzer' => 'title_folded',
                    ],
                    'raw' => ['type' => 'keyword', 'normalizer' => 'raw_lower'],
                ],
            ],
            'channel' => [
                'type'     => 'text',
                'analyzer' => 'title_folded',
                'fields'   => ['raw' => ['type' => 'keyword', 'normalizer' => 'raw_lower']],
            ],
            'regions'      => ['type' => 'keyword'],
            'category_id'  => ['type' => 'keyword'],
            'view_count'   => ['type' => 'long'],
            'published_at' => ['type' => 'date'],
        ],
    ],
];

$ch = curl_init(getenv('OS_HOST') . '/' . $version);
curl_setopt_array($ch, [
    CURLOPT_CUSTOMREQUEST  => 'PUT',
    CURLOPT_POSTFIELDS     => json_encode($body, JSON_THROW_ON_ERROR),
    CURLOPT_HTTPHEADER     => ['Content-Type: application/json'],
    CURLOPT_USERPWD        => getenv('OS_USER') . ':' . getenv('OS_PASS'),
    CURLOPT_RETURNTRANSFER => true,
]);

echo curl_exec($ch), PHP_EOL;

Three decisions in there are worth defending:

asciifolding on both the analyzer and the keyword normalizer. Our regions include titles with é, ü, ı, and ñ. Users type the ASCII form roughly 80% of the time. Folding at index time makes pokemon find Pokémon without any fuzziness budget being spent on the accent.
A separate prefix subfield with a different search_analyzer. If you edge-ngram the query as well as the document, sea expands to s, se, sea and matches essentially everything. Index-side ngrams, query-side plain tokens. This bites people constantly.
Never run fuzzy queries against the ngram field. Fuzziness over edge-ngrams is a combinatorial explosion and it will find you in production, not in your test set. The ngram field is for typeahead prefix matching only; typo tolerance lives on the plain title field.

The alias is the other half. The index is videos_v3; the app only ever talks to the alias videos. Because our deploy is FTP file sync with no atomic release directory, I cannot coordinate "new code + new index" as one switch. An alias means the reindex finishes, the alias flips in one API call, and the old index gets deleted a day later if nothing screams.

Bulk indexing out of SQLite without blocking the site

The indexer runs as the last step of the existing multi-region cron, after the fetch job has written new videos. It is Python because opensearch-py has a bulk helper that handles chunking and retries and I did not want to reimplement that in PHP.

The critical detail is the read-only URI connection. The cron and the live site share one SQLite file; an indexer that takes a write lock for six minutes is an outage.

#!/usr/bin/env python3
"""tools/search/reindex.py — incremental sync SQLite -> OpenSearch.

Runs as the final step of the regional fetch cron. Read-only on SQLite:
the live site keeps serving while this runs.
"""
import os
import sqlite3
import sys
from opensearchpy import OpenSearch
from opensearchpy.helpers import bulk

DB = os.environ['SQLITE_PATH']
ALIAS = 'videos'
CHECKPOINT = os.path.join(os.path.dirname(DB), '.search_checkpoint')


def read_checkpoint() -> int:
    try:
        with open(CHECKPOINT) as fh:
            return int(fh.read().strip())
    except (OSError, ValueError):
        return 0


def connect_sqlite() -> sqlite3.Connection:
    # mode=ro is the point: no write lock, no contention with the web tier.
    conn = sqlite3.connect(f'file:{DB}?mode=ro', uri=True, timeout=15)
    conn.row_factory = sqlite3.Row
    return conn


def docs(conn: sqlite3.Connection, since: int):
    rows = conn.execute(
        '''SELECT v.video_id, v.title, v.channel_title, v.category_id,
                  v.view_count, v.published_at, v.updated_at,
                  group_concat(DISTINCT r.region_code) AS regions
             FROM videos v
             LEFT JOIN video_regions r ON r.video_id = v.video_id
            WHERE v.updated_at > ?
            GROUP BY v.video_id
            ORDER BY v.updated_at ASC''',
        (since,),
    )

    newest = since
    for row in rows:
        newest = max(newest, row['updated_at'])
        yield {
            '_op_type': 'index',
            '_index': ALIAS,
            '_id': row['video_id'],       # idempotent: reruns overwrite, never duplicate
            '_source': {
                'title': row['title'],
                'channel': row['channel_title'] or '',
                'category_id': str(row['category_id'] or ''),
                'regions': (row['regions'] or '').split(',') if row['regions'] else [],
                'view_count': row['view_count'] or 0,
                'published_at': row['published_at'],
            },
        }

    yield {'_checkpoint': newest}  # sentinel, filtered below


def main() -> int:
    client = OpenSearch(
        hosts=[os.environ['OS_HOST']],
        http_auth=(os.environ['OS_USER'], os.environ['OS_PASS']),
        use_ssl=True,
        timeout=30,
        max_retries=3,
        retry_on_timeout=True,
    )

    since = read_checkpoint()
    newest = since
    conn = connect_sqlite()

    def stream():
        nonlocal newest
        for doc in docs(conn, since):
            if '_checkpoint' in doc:
                newest = doc['_checkpoint']
                continue
            yield doc

    ok, errors = bulk(client, stream(), chunk_size=1000, raise_on_error=False)
    print(f'indexed={ok} errors={len(errors)} since={since} -> {newest}')

    if errors:
        for err in errors[:5]:
            print(err, file=sys.stderr)
        return 1  # do NOT advance the checkpoint on partial failure

    with open(CHECKPOINT, 'w') as fh:
        fh.write(str(newest))
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

Two properties I care about here. First, _id is the video ID, so the operation is idempotent — a cron that runs twice because of an overlapping schedule produces the same index, not duplicates. Second, the checkpoint only advances on a clean run. A partial failure means the next run reprocesses the overlap, which is cheap and correct. The opposite (advance always, lose documents on error) is the bug you find three months later when a video nobody searches for turns out to be missing.

A full rebuild of 890K documents takes about 4 minutes. Incremental runs after a regional fetch touch 2K–15K documents and finish in under 10 seconds.

The query, in layers

Here is where typo tolerance actually happens. This is a bool with several should clauses, each doing one job, scored together.

<?php
declare(strict_types=1);

// app/Search/OpenSearchQuery.php
final class OpenSearchQuery
{
    public static function build(string $q, ?string $region, int $limit): array
    {
        $filter = [];
        if ($region !== null) {
            $filter[] = ['term' => ['regions' => $region]];
        }

        return [
            'size'    => $limit,
            '_source' => false,          // we only want IDs; SQLite hydrates the rest
            'query'   => [
                'function_score' => [
                    'query' => [
                        'bool' => [
                            'filter'               => $filter,
                            'minimum_should_match' => 1,
                            'should' => [
                                // 1. exact normalized title — always wins
                                ['term' => ['title.raw' => ['value' => mb_strtolower($q), 'boost' => 12.0]]],

                                // 2. phrase match, order matters
                                ['match_phrase' => ['title' => ['query' => $q, 'boost' => 6.0, 'slop' => 1]]],

                                // 3. the typo-tolerant leg
                                ['multi_match' => [
                                    'query'                => $q,
                                    'fields'               => ['title^3', 'channel'],
                                    'type'                 => 'best_fields',
                                    'fuzziness'            => 'AUTO:4,7',
                                    'prefix_length'        => 1,
                                    'max_expansions'       => 50,
                                    'minimum_should_match' => '70%',
                                    'boost'                => 2.0,
                                ]],

                                // 4. typeahead prefix, low boost so it never outranks a real match
                                ['match' => ['title.prefix' => ['query' => $q, 'boost' => 0.6]]],
                            ],
                        ],
                    ],
                    'functions' => [
                        ['field_value_factor' => [
                            'field'    => 'view_count',
                            'modifier' => 'log1p',
                            'factor'   => 0.15,
                            'missing'  => 1,
                        ]],
                        ['gauss' => ['published_at' => [
                            'origin' => 'now', 'scale' => '21d', 'offset' => '3d', 'decay' => 0.5,
                        ]]],
                    ],
                    'score_mode' => 'sum',
                    'boost_mode' => 'multiply',
                ],
            ],
        ];
    }
}

The parameters that took the most tuning:

fuzziness: AUTO:4,7 means terms under 4 characters get no fuzziness, 4–6 characters get edit distance 1, and 7+ get edit distance 2. The default AUTO starts fuzzing at 3 characters, which made the match she, tha, and toe. Video titles are full of short function words. Raising the floor to 4 removed an entire class of nonsense results.
prefix_length: 1 requires the first character to be exact. Users mistype the middle and end of words far more than the first character, and this single setting cut fuzzy query latency by roughly 60% because it prunes the term dictionary scan immediately.
minimum_should_match: '70%' on the fuzzy leg. Without it, a five-word query matched documents sharing one fuzzy token, and the result page filled with garbage that scored just high enough to display. Requiring most of the query to land is what makes fuzzy results feel intentional rather than random.
max_expansions: 50 caps how many index terms each fuzzy term expands to. Uncapped fuzzy matching on a large title corpus is how you discover what a slow query log looks like.
The recency gauss is site-specific. We are a trending-video site; a match from three weeks ago is worth less than a match from yesterday, and boost_mode: multiply lets relevance stay the primary signal with freshness as a modifier rather than an override.

On strangr things sesaon 5 trailr: clause 1 and 2 contribute nothing, clause 3 does all the work, and the query now returns 287 results with the right trailer at position 2.

Never let search take the site down

This is the part I would keep even if I ripped out OpenSearch tomorrow. FTS5 stayed. It is now the fallback behind a circuit breaker, and the search page is structurally incapable of returning a 500 because the search node is unreachable.

<?php
declare(strict_types=1);

// app/Search/SearchFacade.php
final class SearchFacade
{
    private const FAIL_THRESHOLD = 3;
    private const OPEN_SECONDS   = 60;
    private const TIMEOUT_MS     = 250;

    public function __construct(
        private readonly PDO $db,
        private readonly Fts5Search $fallback,
        private readonly string $host,
        private readonly string $auth,
    ) {}

    public function search(string $q, ?string $region, int $limit = 40): array
    {
        if ($this->breakerOpen()) {
            return $this->fallback->search($q, $limit);
        }

        try {
            $ids = $this->queryOpenSearch($q, $region, $limit);
            $this->recordSuccess();
            return $ids === [] ? [] : $this->hydrate($ids);
        } catch (Throwable $e) {
            $this->recordFailure();
            error_log('search fallback: ' . $e->getMessage());
            return $this->fallback->search($q, $limit);
        }
    }

    /** @return list<string> */
    private function queryOpenSearch(string $q, ?string $region, int $limit): array
    {
        $ch = curl_init($this->host . '/videos/_search');
        curl_setopt_array($ch, [
            CURLOPT_POST           => true,
            CURLOPT_POSTFIELDS     => json_encode(
                OpenSearchQuery::build($q, $region, $limit),
                JSON_THROW_ON_ERROR
            ),
            CURLOPT_HTTPHEADER     => ['Content-Type: application/json'],
            CURLOPT_USERPWD        => $this->auth,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT_MS     => self::TIMEOUT_MS,
            CURLOPT_CONNECTTIMEOUT_MS => 120,
        ]);

        $raw  = curl_exec($ch);
        $code = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
        curl_close($ch);

        if ($raw === false || $code !== 200) {
            throw new RuntimeException('opensearch http ' . $code);
        }

        $body = json_decode($raw, true, 512, JSON_THROW_ON_ERROR);

        return array_map(
            static fn (array $hit): string => (string) $hit['_id'],
            $body['hits']['hits'] ?? []
        );
    }

    /** Reorder SQLite rows to match OpenSearch relevance order. */
    private function hydrate(array $ids): array
    {
        $in   = implode(',', array_fill(0, count($ids), '?'));
        $stmt = $this->db->prepare(
            "SELECT video_id, title, thumbnail, duration, channel_title, published_at
               FROM videos WHERE video_id IN ($in)"
        );
        $stmt->execute($ids);

        $byId = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $byId[$row['video_id']] = $row;
        }

        $out = [];
        foreach ($ids as $id) {
            if (isset($byId[$id])) {
                $out[] = $byId[$id];   // preserve relevance order
            }
        }
        return $out;
    }

    private function breakerOpen(): bool
    {
        $row = $this->db->query(
            'SELECT fails, opened_at FROM search_breaker WHERE id = 1'
        )->fetch(PDO::FETCH_ASSOC);

        if (!$row || (int) $row['fails'] < self::FAIL_THRESHOLD) {
            return false;
        }
        return (time() - (int) $row['opened_at']) < self::OPEN_SECONDS;
    }

    private function recordFailure(): void
    {
        $this->db->exec(
            'INSERT INTO search_breaker (id, fails, opened_at) VALUES (1, 1, ' . time() . ')
             ON CONFLICT(id) DO UPDATE SET fails = fails + 1, opened_at = ' . time()
        );
    }

    private function recordSuccess(): void
    {
        $this->db->exec('UPDATE search_breaker SET fails = 0 WHERE id = 1');
    }
}

The breaker state lives in SQLite because shared hosting gives us no APCu and no Redis. It is one row and two integers, and the write only happens on state transitions, so it costs nothing in the normal path. The 250ms timeout is deliberately aggressive: if OpenSearch cannot answer in 250ms, FTS5 can answer in 9ms, and a slightly worse result now beats a better result after a spinner.

The FTP deploy story shakes out cleanly with this shape. The application ships as plain files with no Composer vendor tree — the OpenSearch client is 200 lines of cURL. Host and credentials live in the per-site env file that the deploy script already manages. If I push new code that expects a mapping the index does not have yet, the queries fail, the breaker opens, and the site quietly serves FTS5 results until I fix it. That is exactly the failure mode I want from a dumb file-sync deploy.

Six weeks of numbers

Zero-result rate on search: 11.4% → 2.1%
Search-result click-through: 43% → 58%
p95 search latency: 9ms (FTS5) → 38ms (OpenSearch)
Sessions that ran a second search after an empty first one: down 64%
Infrastructure cost: $12/month for the node

Search got four times slower and materially better. I would make that trade again — 38ms is invisible next to the video thumbnails loading on the same page, and an empty result set is not.

Three things I would do differently

Alias from day one. I indexed straight into videos for the first two weeks, then had to take a five-minute search outage to reindex under a new mapping. The alias indirection costs one API call and buys zero-downtime reindexing forever.
Do not fuzzy-match channel names as aggressively as titles. I originally boosted channel at the same weight as title, and fuzzy channel matches hijacked queries — searching for a show pulled up every video from a channel whose name was one edit away. Dropping channel to unboosted inside best_fields fixed it.
Log queries with their result count before you need the data. The only reason I could justify this project was two years of search_log rows with a result_count column. Without it I would have been arguing from intuition, and I would have been guessing at which typo classes actually mattered.

If you are sitting on FTS5 and watching a zero-result rate you cannot explain, the honest answer is that no amount of tokenizer tuning gets you edit distance. But you do not have to throw FTS5 away to get it — one small node, a versioned index behind an alias, and a circuit breaker in front turn your existing full-text search into a fallback that costs nothing and saves you on the day the search node reboots.

Building a GraphQL Video Discovery API with Strawberry and FastAPI on SQLite

ahmet gedik — Sat, 15 Aug 2026 07:00:01 +0000

Eleven variants of /trending and none of them fit

We ingest viral video signals from 14 European markets and republish ranked lists — per country, per category, per 6-hour velocity window. The public site is PHP 8.4 on LiteSpeed with SQLite in WAL mode behind Cloudflare Workers, and that stack is genuinely fine for serving HTML. What broke was everything that was not HTML.

Over eighteen months we accumulated /api/trending, /api/trending/compact, /api/trending/embed, /api/market/{cc}/trending, plus a v2 of three of those. Each one existed because some consumer needed three fields fewer or one join more than the last. The embed widget pulled 38 KB of JSON to render six thumbnails and a title. The internal trend dashboard fired nine sequential requests to build one view.

The thing that actually forced a decision was neither of those. Every new column on the video record meant auditing eleven serializers to confirm we had not just leaked a region code derived from a request IP into a public payload. Since ViralVidVault sells itself on GDPR-compliant analytics, an accidental field in an accidental endpoint is not a papercut, it is the whole product's credibility. Eleven hand-written serializers is eleven places to get that wrong.

So we put a read-only GraphQL API in front of the same SQLite file, written in Python with Strawberry and FastAPI. One schema, one place where a field becomes public, one audit surface. Below is the design that survived six months in production, plus the three things that bit us.

Why a second runtime instead of a PHP GraphQL server

webonyx/graphql-php is a solid library and staying single-runtime would have been the boring, correct-sounding choice. We went Python anyway, for reasons that were specific rather than ideological:

The ranking code was already Python. Velocity scoring, decay curves, and the market normalisation live in a Python batch job. Reusing those dataclasses as the schema source of truth removed a translation layer.
DataLoader wants an event loop. Batching nested resolvers is the single biggest performance lever in GraphQL, and doing it well in synchronous PHP-FPM means either fibers or a lot of manual pre-fetching.
Code-first typing. Strawberry builds the schema from annotated Python classes. mypy catches a resolver returning the wrong shape before the SDL is ever generated.
Process isolation as a safety property. The Python service opens the database mode=ro and sets PRAGMA query_only. It is structurally incapable of corrupting the site's data, which made the deploy risk close to zero.

The cost is real: two runtimes, two deploy pipelines, two dependency audits. We made that acceptable by scoping hard. The GraphQL service is read-only, stateless, and serves no HTML. If it falls over, the site does not notice. That is the only reason a second runtime is worth it.

Model the queries, not the tables

The most common GraphQL failure I see is a schema that is a 1:1 mirror of the database, which gives you an ORM over HTTP and all of the coupling that implies. We started from the four things consumers actually ask for: a ranked page of videos for a market, one video by id, the channel behind a video, and a velocity time series.

Two deliberate omissions shaped everything after. There is no viewer type and no user type — the API is anonymous, so there is no session to model. And Channel.subscriberCount is exposed as a bucket string, not an integer, because exact counts on small channels combined with a market filter get uncomfortably close to identifying a specific creator's performance data.

# schema.py
from __future__ import annotations

from datetime import datetime
from enum import Enum

import strawberry


@strawberry.enum
class Market(Enum):
    DE = 'DE'
    FR = 'FR'
    ES = 'ES'
    IT = 'IT'
    NL = 'NL'
    PL = 'PL'
    SE = 'SE'
    GB = 'GB'


@strawberry.enum
class Window(Enum):
    H6 = 'h6'
    H24 = 'h24'
    D7 = 'd7'


@strawberry.type
class VelocityPoint:
    at: datetime
    views: int
    delta_per_hour: float


@strawberry.type
class Channel:
    id: strawberry.ID
    name: str
    market: Market
    subscriber_bucket: str  # '10k-50k', never an exact figure


@strawberry.type
class Video:
    id: strawberry.ID
    title: str
    published_at: datetime
    duration_s: int
    thumb: str
    score: float
    # Private fields exist on the Python object but never reach the SDL.
    channel_id: strawberry.Private[str]

    @strawberry.field
    async def channel(self, info: strawberry.Info) -> Channel | None:
        return await info.context['loaders'].channel.load(self.channel_id)

    @strawberry.field
    async def velocity(
        self, info: strawberry.Info, window: Window = Window.H24
    ) -> list[VelocityPoint]:
        return await info.context['loaders'].velocity.load((str(self.id), window.value))


@strawberry.type
class VideoEdge:
    cursor: str
    node: Video


@strawberry.type
class PageInfo:
    has_next_page: bool
    end_cursor: str | None


@strawberry.type
class VideoConnection:
    edges: list[VideoEdge]
    page_info: PageInfo
    rank_epoch: int  # which ranking snapshot this page was cut from


@strawberry.type
class Query:
    @strawberry.field(description='Ranked viral videos for one market.')
    async def trending(
        self,
        info: strawberry.Info,
        market: Market,
        window: Window = Window.H24,
        first: int = 25,
        after: str | None = None,
    ) -> VideoConnection:
        return await trending_page(
            info.context['db'], market.value, window.value, min(first, 100), after
        )

    @strawberry.field
    async def video(self, info: strawberry.Info, id: strawberry.ID) -> Video | None:
        return await info.context['loaders'].video.load(str(id))

strawberry.Private is doing quiet but important work here. channel_id is needed by the resolver and must never appear in the public schema. Declaring it private means the leak is impossible rather than merely unlikely, which is the whole reason we moved off hand-written serializers.

Wiring Strawberry into FastAPI over a read-only SQLite pool

SQLite in WAL mode allows one writer and many concurrent readers without the readers blocking. That is exactly the shape of this workload: a PHP cron writes ranking snapshots every few hours, and the API only ever reads.

A few things are not obvious the first time:

journal_mode = WAL is a persistent property of the database file, set once by the writer. A read-only connection cannot set it and will error if it tries.
Opening a connection and running pragmas costs roughly 0.4 ms. Trivial once, wasteful at a few hundred requests per second, so we keep a small fixed pool.
Long-lived read connections pin the WAL. SQLite starts an implicit read transaction on the first query and holds that snapshot until the transaction ends. A pooled connection that never commits or rolls back keeps an old snapshot alive, which blocks WAL checkpointing and lets the -wal file grow without bound. We learned this when a 40 MB database grew a 900 MB WAL over a weekend. The fix is one line in the pool's release path.

# db.py
import asyncio
from contextlib import asynccontextmanager

import aiosqlite

DB_URI = 'file:/var/www/vvv/data/vvv.sqlite?mode=ro'


class ReadPool:
    def __init__(self, uri: str, size: int = 8) -> None:
        self._uri = uri
        self._size = size
        self._q: asyncio.Queue = asyncio.Queue(maxsize=size)

    async def open(self) -> None:
        for _ in range(self._size):
            conn = await aiosqlite.connect(self._uri, uri=True)
            conn.row_factory = aiosqlite.Row
            await conn.execute('PRAGMA query_only = ON')
            await conn.execute('PRAGMA busy_timeout = 2000')
            await conn.execute('PRAGMA cache_size = -32000')      # 32 MB page cache
            await conn.execute('PRAGMA mmap_size = 268435456')    # 256 MB mmap
            self._q.put_nowait(conn)

    @asynccontextmanager
    async def acquire(self):
        conn = await self._q.get()
        try:
            yield conn
        finally:
            # Ends the implicit read transaction. Without this the connection
            # pins an old WAL snapshot and checkpointing stalls forever.
            await conn.rollback()
            self._q.put_nowait(conn)

    async def close(self) -> None:
        while not self._q.empty():
            await self._q.get_nowait().close()


# main.py
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

from .loaders import Loaders
from .schema import schema

pool = ReadPool(DB_URI)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await pool.open()
    yield
    await pool.close()


async def get_context():
    # One connection and one fresh loader set per HTTP request.
    async with pool.acquire() as conn:
        yield {'db': conn, 'loaders': Loaders(conn)}


app = FastAPI(lifespan=lifespan)
app.include_router(
    GraphQLRouter(schema, context_getter=get_context, graphql_ide=None),
    prefix='/graphql',
)

context_getter is resolved as a normal FastAPI dependency, so an async generator works and the connection is returned to the pool when the response finishes. Building Loaders per request rather than per process matters: DataLoader keeps a per-instance memo cache, and a process-lifetime cache would happily serve last Tuesday's ranking.

The N+1 that WAL hides from you

SQLite is fast enough locally to disguise bad query patterns. A page of 200 videos issuing 200 single-row channel lookups at ~30 µs each is 6 ms, which nobody notices in a flame graph. Then the dashboard asks for a 24-point velocity series per video and the same shape costs 200 range scans, and now you are at 340 ms and blaming Python.

DataLoader collapses both into one query per field per request. Two rules matter: the loader must return results in the same order as the requested keys, and the batch must be capped below SQLite's bound-parameter limit (32766 in modern builds, but 200 is a saner batch anyway).

This file also holds the pagination logic, because the cursor and the batch loader share the same snapshot concept.

# loaders.py
import base64
import json

from strawberry.dataloader import DataLoader

from .schema import Channel, Market, Video, VideoConnection, VideoEdge, PageInfo


def encode_cursor(score: float, video_id: str, epoch: int) -> str:
    raw = json.dumps([round(score, 6), video_id, epoch], separators=(',', ':'))
    return base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=')


def decode_cursor(cur: str) -> tuple[float, str, int]:
    pad = '=' * (-len(cur) % 4)
    score, video_id, epoch = json.loads(base64.urlsafe_b64decode(cur + pad))
    return float(score), str(video_id), int(epoch)


async def trending_page(conn, market: str, window: str, first: int, after: str | None):
    keyset, params = '', [market, window]

    if after:
        score, vid, epoch = decode_cursor(after)
        # Pin the caller to the snapshot they started paging on.
        keyset = 'AND (r.score < ? OR (r.score = ? AND r.video_id > ?))'
    else:
        row = await (await conn.execute(
            'SELECT MAX(rank_epoch) AS e FROM rank_snapshot WHERE market = ? AND window = ?',
            (market, window),
        )).fetchone()
        epoch = row['e']

    params.append(epoch)
    if after:
        params += [score, score, vid]
    params.append(first + 1)

    sql = f'''
        SELECT v.id, v.title, v.published_at, v.duration_s, v.thumb, v.channel_id, r.score
          FROM rank_snapshot r
          JOIN video v ON v.id = r.video_id
         WHERE r.market = ? AND r.window = ? AND r.rank_epoch = ? {keyset}
         ORDER BY r.score DESC, r.video_id ASC
         LIMIT ?
    '''
    rows = await (await conn.execute(sql, params)).fetchall()

    has_next = len(rows) > first
    rows = rows[:first]
    edges = [
        VideoEdge(
            cursor=encode_cursor(r['score'], r['id'], epoch),
            node=Video(
                id=r['id'], title=r['title'], published_at=r['published_at'],
                duration_s=r['duration_s'], thumb=r['thumb'],
                score=r['score'], channel_id=r['channel_id'],
            ),
        )
        for r in rows
    ]
    return VideoConnection(
        edges=edges,
        page_info=PageInfo(
            has_next_page=has_next,
            end_cursor=edges[-1].cursor if edges else None,
        ),
        rank_epoch=epoch,
    )


async def load_channels(conn, ids: list[str]) -> list[Channel | None]:
    placeholders = ','.join('?' * len(ids))
    rows = await (await conn.execute(
        f'SELECT id, name, market, subscriber_bucket FROM channel WHERE id IN ({placeholders})',
        ids,
    )).fetchall()
    by_id = {
        r['id']: Channel(
            id=r['id'], name=r['name'],
            market=Market(r['market']), subscriber_bucket=r['subscriber_bucket'],
        )
        for r in rows
    }
    # Order must mirror the requested keys, gaps become None.
    return [by_id.get(i) for i in ids]


class Loaders:
    def __init__(self, conn) -> None:
        self.channel = DataLoader(
            load_fn=lambda keys: load_channels(conn, keys), max_batch_size=200
        )
        self.velocity = DataLoader(
            load_fn=lambda keys: load_velocity(conn, keys), max_batch_size=200
        )
        self.video = DataLoader(load_fn=lambda keys: load_videos(conn, keys))

Why the cursor carries the epoch

OFFSET pagination against a table that re-ranks on a cron is quietly broken. A reader on page 3 when the 06:00 job lands will see items they already saw and miss items entirely, because the offsets now point at a different ordering. Keyset pagination on (score DESC, video_id ASC) fixes duplicates within a stable ordering, but not the re-rank itself.

So the cursor encodes the rank_epoch it was cut from, and subsequent pages are served from that same snapshot. A paging session stays internally consistent even if the ranking changes underneath it, and VideoConnection.rank_epoch tells the client which snapshot it is looking at so it can decide whether to restart from page one.

The index that makes all of this a single scan:

CREATE INDEX idx_rank_page ON rank_snapshot(market, window, rank_epoch, score DESC, video_id);
Check it with EXPLAIN QUERY PLAN; you want SEARCH r USING INDEX idx_rank_page and no USE TEMP B-TREE FOR ORDER BY.

The write side stays in PHP

SQLite tolerates exactly one writer, so we kept the writer where it already lived. The cron computes scores in PHP 8.4, inserts a whole new epoch in one transaction, and prunes old ones — keeping two epochs alive so readers mid-page do not fall off a cliff.

<?php
declare(strict_types=1);

final class RankSnapshotWriter
{
    public function __construct(private readonly PDO $db) {}

    /** @param array<string,float> $scored video_id => score */
    public function publish(string $market, string $window, array $scored): int
    {
        $epoch = time();
        $this->db->beginTransaction();

        try {
            $ins = $this->db->prepare(
                'INSERT INTO rank_snapshot (market, window, rank_epoch, video_id, score)
                 VALUES (?, ?, ?, ?, ?)'
            );
            foreach ($scored as $videoId => $score) {
                $ins->execute([$market, $window, $epoch, $videoId, $score]);
            }

            // Keep the live epoch plus the previous one, so in-flight cursors resolve.
            $prune = $this->db->prepare(<<<'SQL'
                DELETE FROM rank_snapshot
                 WHERE market = ? AND window = ?
                   AND rank_epoch NOT IN (
                       SELECT rank_epoch FROM (
                           SELECT DISTINCT rank_epoch FROM rank_snapshot
                            WHERE market = ? AND window = ?
                            ORDER BY rank_epoch DESC LIMIT 2
                       )
                   )
                SQL);
            $prune->execute([$market, $window, $market, $window]);

            $this->db->commit();
        } catch (Throwable $e) {
            $this->db->rollBack();
            throw $e;
        }

        return $epoch;
    }
}

$pdo = new PDO('sqlite:/var/www/vvv/data/vvv.sqlite', options: [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
]);
$pdo->exec('PRAGMA journal_mode = WAL');
$pdo->exec('PRAGMA synchronous = NORMAL');
$pdo->exec('PRAGMA wal_autocheckpoint = 512');

// PHP 8.4: new expressions no longer need wrapping parentheses.
$epoch = new RankSnapshotWriter($pdo)->publish('DE', 'h24', $scores);
fprintf(STDERR, 'published epoch %d (%d rows)%s', $epoch, count($scores), PHP_EOL);

Because the insert is one transaction, readers either see the whole new epoch or none of it. There is no window where a page is half old and half new. That is the property that makes the epoch-pinned cursor honest.

GraphQL is a denial-of-service footgun by default

A public GraphQL endpoint lets a stranger write your queries. Three controls, in the order they earn their keep:

Static cost analysis before execution. Depth limits alone are not enough — trending(first: 100) { velocity } is shallow and expensive.
Persisted queries in production. Clients send a SHA-256 hash of a known document, not arbitrary text. This turns the whole class of hostile-query problems into a lookup miss, and as a bonus makes the request a cacheable GET.
Introspection off in production. Not security, but it removes the free schema map.

# security.py
import hashlib
from pathlib import Path

from fastapi import HTTPException
from graphql import GraphQLError
from graphql.validation import ValidationRule
from strawberry.extensions import AddValidationRules

FIELD_COST = {'trending': 2, 'velocity': 5, 'channel': 1, 'video': 1}
MAX_COST = 1000


class CostLimit(ValidationRule):
    def __init__(self, context) -> None:
        super().__init__(context)
        self.cost = 0

    def enter_field(self, node, *_args) -> None:
        base = FIELD_COST.get(node.name.value, 0)
        if not base:
            return
        first = next((a for a in node.arguments if a.name.value == 'first'), None)
        n = int(getattr(first.value, 'value', 1)) if first is not None else 1
        self.cost += base * max(n, 1)
        if self.cost > MAX_COST:
            self.report_error(
                GraphQLError(f'query cost {self.cost} exceeds limit {MAX_COST}', node)
            )


# Built at boot from a directory of .graphql files shipped with the clients.
PERSISTED = {
    hashlib.sha256(p.read_bytes()).hexdigest(): p.read_text()
    for p in Path('persisted').glob('*.graphql')
}


def resolve_persisted(digest: str) -> str:
    try:
        return PERSISTED[digest]
    except KeyError:
        raise HTTPException(status_code=400, detail='unknown persisted query')


# schema.py
# schema = strawberry.Schema(
#     query=Query,
#     extensions=[AddValidationRules([CostLimit])],
#     config=StrawberryConfig(...),
# )

We expose GET /graphql/q/{digest}?vars=... alongside the POST endpoint. A Cloudflare Worker rewrites it, adds Cache-Control: public, max-age=300, stale-while-revalidate=900, and the edge absorbs the repeat traffic. Persisted GETs currently run around an 87% edge hit ratio for the embed widget, which is the single biggest win in this whole project and has nothing to do with GraphQL being GraphQL.

What is deliberately absent from the schema

The GDPR posture is mostly about what does not exist:

No viewer, no user, no session. The API is anonymous and sets no cookies, so there is nothing to consent to.
Request IPs are never resolved to a region inside this service. Market is an explicit argument supplied by the caller.
Aggregate market statistics suppress any bucket with fewer than 50 events, so a thin slice cannot be narrowed to one creator or one viewer.
Logs record the operation name and the persisted-query digest, never the variables. Variables can carry a video id, and a video id plus a timestamp plus an IP in an access log is personal data by any reasonable reading.

Having one schema instead of eleven serializers is what makes this reviewable. Adding a field is a diff in one file, and the reviewer's only question is whether that field should be public.

What it cost and what it bought

On a 2 vCPU box, a 50-video connection with channel and a 24-point velocity series per node runs p50 3.1 ms and p95 11 ms — dominated by JSON serialisation, not SQLite. The embed payload went from 38 KB to 2.8 KB because the widget now asks for four fields. Eleven REST endpoints are down to two routes.

Things I would skip if starting again: subscriptions (nobody asked, and polling a 5-minute-cached GET is cheaper), and federation (one schema, one team, no seams to stitch). Things I would do sooner: the cursor epoch, and the rollback() in the pool release. Both were incidents before they were design decisions.

The general lesson is unglamorous. GraphQL did not make the API fast — keyset pagination, a covering index, and DataLoader did. What GraphQL bought was a single typed boundary where a field becomes public, which for a product built on a privacy promise turned out to be worth more than the bytes we saved.