Building a Video URL Canonicalization Pipeline for Multi-Region Feeds

#php #sqlite #go #webdev

The bug report was one sentence: "Vietnam trending shows the same K-pop video four times." It did. One video, four rows, four URLs that our ingest layer had happily treated as four distinct pieces of content:

https://www.youtube.com/watch?v=kJQP7kiw5Fk&list=RDkJQP7kiw5Fk — pulled from a regional trending feed
https://m.youtube.com/watch?v=kJQP7kiw5Fk — a mobile share link a user submitted
https://youtu.be/kJQP7kiw5Fk?t=42&si=9xQd1Kk2 — same user, different share sheet
https://www.youtube.com/watch?app=desktop&feature=share&v=kJQP7kiw5Fk — a partner API response

At TopVideoHub we aggregate trending video across the Asia-Pacific region in nine languages, and URL-as-identity had been fine for about six months. Once we were pulling from eleven regional feeds plus three partner APIs plus a submissions form, the duplicate rate on raw URL identity sat between 12% and 18% depending on the market — worst in VN, ID and TH, where mobile share links dominate and every one of them carries a fresh si= tracking parameter.

Duplicates are not just cosmetic. They split view counts across rows, they blow up the SQLite FTS5 index, and — the part that actually hurt — they poison BM25 ranking, because four near-identical title rows crowd out everything else for a query. This is the pipeline that replaced it, and the specific things that went wrong while we built it.

Why Naive Normalization Breaks

The first attempt, written in an afternoon, was a single function that lowercased the URL, stripped a denylist of tracking params, and sorted the rest. It made things worse. Here is why, in the order we discovered them:

Lowercasing destroys IDs. YouTube video IDs are 11 characters of base64url. kJQP7kiw5Fk and kjqp7kiw5fk are different videos, and one of them does not exist. Bilibili BV IDs are the same story.
A tracking-param denylist is a losing race. si, feature, app, pp, ab_channel, utm_*, spm_id_from, vd_source — every provider adds new ones and never tells you.
Stripping all params breaks providers where the param is the ID. ?v= on YouTube is the identity. Blanket-stripping the query string turns every watch URL into youtube.com/watch.
parse_url() returns false on some malformed IDN hosts and returns a host you cannot compare on others. Punycode is not optional in this market — you will see 動画.jp style hosts in partner feeds.
Eager percent-decoding changes path structure. rawurldecode() turns %2F into /, which silently creates a new path segment.

The rule we landed on, and the reason this pipeline works at all: normalize structurally, then extract semantically, and never mix the two stages. Stage one knows nothing about video providers and only applies RFC 3986. Stage two knows nothing about URL syntax and only knows what a video ID looks like for one provider.

Stage One — Structural Normalization

This stage is provider-agnostic and deterministic. Same bytes in, same bytes out, no network, no database. It is pure enough that we unit-test it with a 900-line fixture table.

<?php
declare(strict_types=1);

final class UrlNormalizer
{
    private const DEFAULT_PORTS = ['http' => 80, 'https' => 443];

    public function normalize(string $raw): ?string
    {
        // Share sheets and CSV imports smuggle in newlines, tabs and NBSP.
        $raw = trim(preg_replace('/[\x00-\x20\x7F]/u', '', $raw) ?? '');
        if ($raw === '') {
            return null;
        }
        if (!preg_match('#^[a-z][a-z0-9+.\-]*://#i', $raw)) {
            $raw = 'https://' . ltrim($raw, '/');
        }

        $p = parse_url($raw);
        if ($p === false || empty($p['host'])) {
            return null;
        }

        $scheme = strtolower($p['scheme'] ?? 'https');
        if ($scheme !== 'http' && $scheme !== 'https') {
            return null;
        }

        $host = $this->asciiHost($p['host']);
        if ($host === null) {
            return null;   // fail closed; a guessed host becomes a second canonical key
        }

        $port = isset($p['port']) && $p['port'] !== self::DEFAULT_PORTS[$scheme]
            ? ':' . $p['port']
            : '';

        $path  = $this->normalizePath($p['path'] ?? '/');
        $query = isset($p['query']) && $p['query'] !== ''
            ? '?' . $this->normalizeQuery($p['query'])
            : '';

        // Fragment is dropped: no provider we ingest carries identity there.
        return $scheme . '://' . $host . $port . $path . $query;
    }

    private function asciiHost(string $host): ?string
    {
        $host = rtrim(strtolower($host), '.');
        if ($host === '') {
            return null;
        }
        if (preg_match('/^[a-z0-9.\-]+$/', $host) === 1) {
            return $host;
        }
        $ascii = idn_to_ascii($host, IDNA_NONTRANSITIONAL_TO_ASCII, INTL_IDNA_VARIANT_UTS46);

        return $ascii === false ? null : $ascii;
    }

    private function normalizePath(string $path): string
    {
        $path = $this->fixPercentEncoding($path);
        $out  = [];
        foreach (explode('/', $path) as $seg) {
            if ($seg === '' || $seg === '.') {
                continue;                       // also collapses // runs
            }
            if ($seg === '..') {
                array_pop($out);
                continue;
            }
            $out[] = $seg;
        }

        return $out === [] ? '/' : '/' . implode('/', $out);
    }

    private function normalizeQuery(string $query): string
    {
        $pairs = [];
        foreach (explode('&', $query) as $pair) {
            if ($pair === '') {
                continue;
            }
            [$k, $v] = array_pad(explode('=', $pair, 2), 2, null);
            $pairs[] = [
                $this->fixPercentEncoding($k),
                $v === null ? null : $this->fixPercentEncoding($v),
            ];
        }
        usort($pairs, static fn(array $a, array $b): int => [$a[0], $a[1]] <=> [$b[0], $b[1]]);

        return implode('&', array_map(
            static fn(array $p): string => $p[1] === null ? $p[0] : $p[0] . '=' . $p[1],
            $pairs
        ));
    }

    /** RFC 3986 6.2.2.2: uppercase the hex, decode unreserved octets, touch nothing else. */
    private function fixPercentEncoding(string $s): string
    {
        return preg_replace_callback('/%([0-9A-Fa-f]{2})/', static function (array $m): string {
            $chr = chr((int) hexdec($m[1]));

            return preg_match('/[A-Za-z0-9\-._~]/', $chr) === 1 ? $chr : '%' . strtoupper($m[1]);
        }, $s) ?? $s;
    }
}

Three decisions in there are worth defending:

We sort query parameters but do not remove any. Removal is a semantic decision and belongs in stage two. The stage-one output is a faithful, comparable rendering of the input — it is what we store in the alias table.
asciiHost() fails closed. Our first version returned the original host when idn_to_ascii() returned false. That produced two canonical keys for the same video and was the single hardest bug to reproduce, because it only triggered on a handful of partner rows per week.
Percent-encoding is normalized, not decoded. %E6%97%A5%E6%9C%AC stays as it is; only unreserved ASCII octets collapse. Decoding CJK bytes into a path would change the string length, break comparison against the alias table, and occasionally produce a byte sequence SQLite refuses to index.

Stage Two — Provider Extractors

Now we get to say things like "a YouTube ID is eleven base64url characters." Each extractor owns a set of hosts and returns a CanonicalVideo or null. Nothing in here parses URLs.

<?php
declare(strict_types=1);

final class CanonicalVideo
{
    public function __construct(
        public readonly string $provider,
        public readonly string $id,
        public readonly ?string $part = null,   // bilibili multi-part episodes
    ) {}

    public function key(): string
    {
        return $this->part === null
            ? $this->provider . ':' . $this->id
            : $this->provider . ':' . $this->id . '#' . $this->part;
    }
}

interface VideoIdExtractor
{
    /** @return string[] punycoded, lowercased hosts this extractor owns */
    public function hosts(): array;

    /** @param array<string,string> $query */
    public function extract(string $host, string $path, array $query): ?CanonicalVideo;
}

final class YouTubeExtractor implements VideoIdExtractor
{
    private const ID = '/^[A-Za-z0-9_-]{11}$/';   // case-sensitive, deliberately

    public function hosts(): array
    {
        return [
            'youtube.com', 'www.youtube.com', 'm.youtube.com', 'music.youtube.com',
            'youtu.be', 'youtube-nocookie.com', 'www.youtube-nocookie.com',
        ];
    }

    public function extract(string $host, string $path, array $query): ?CanonicalVideo
    {
        $seg = array_values(array_filter(explode('/', $path), 'strlen'));

        $candidate = match (true) {
            isset($query['v'])                => $query['v'],
            $host === 'youtu.be'              => $seg[0] ?? null,
            in_array($seg[0] ?? '', ['shorts', 'embed', 'live', 'v'], true) => $seg[1] ?? null,
            default                           => null,
        };

        return is_string($candidate) && preg_match(self::ID, $candidate) === 1
            ? new CanonicalVideo('yt', $candidate)
            : null;
    }
}

final class BilibiliExtractor implements VideoIdExtractor
{
    public function hosts(): array
    {
        return ['bilibili.com', 'www.bilibili.com', 'm.bilibili.com'];
    }

    public function extract(string $host, string $path, array $query): ?CanonicalVideo
    {
        if (preg_match('#^/video/(BV[A-Za-z0-9]{10}|av\d+)#', $path, $m) !== 1) {
            return null;
        }
        $id = str_starts_with($m[1], 'av') ? strtolower($m[1]) : $m[1];

        // p=1 is the implicit default; only p>=2 is part of identity.
        $part = isset($query['p']) && ctype_digit($query['p']) && (int) $query['p'] > 1
            ? (string) (int) $query['p']
            : null;

        return new CanonicalVideo('bili', $id, $part);
    }
}

final class CanonicalResolver
{
    /** @var array<string,VideoIdExtractor> */
    private array $byHost = [];

    public function __construct(VideoIdExtractor ...$extractors)
    {
        foreach ($extractors as $e) {
            foreach ($e->hosts() as $h) {
                $this->byHost[$h] = $e;
            }
        }
    }

    public function resolve(string $normalizedUrl): ?CanonicalVideo
    {
        $p    = parse_url($normalizedUrl);
        $host = $p['host'] ?? '';
        $extractor = $this->byHost[$host] ?? null;
        if ($extractor === null) {
            return null;
        }

        $query = [];
        foreach (explode('&', $p['query'] ?? '') as $pair) {
            if ($pair === '') {
                continue;
            }
            [$k, $v] = array_pad(explode('=', $pair, 2), 2, '');
            // Hand-rolled because parse_str() rewrites "a.b" and "a b" to "a_b".
            $query[rawurldecode($k)] = rawurldecode($v);
        }

        return $extractor->extract($host, $p['path'] ?? '/', $query);
    }
}

The part field is the piece people forget. Bilibili multi-part uploads are genuinely different videos sharing a BV ID, and p=1 is the implicit default — so ?p=1 and no p at all must produce the same key, while ?p=2 must not. We shipped without that distinction, merged about 900 legitimately distinct episodes, and had to rebuild from the alias table. Which is exactly why the alias table exists.

The Schema — Aliases Are the Whole Trick

The canonical key is a derived value. Derived values change when your extractors improve, and if the only record of an ingest is the row you overwrote, an extractor bug is unrecoverable. So we store every normalized URL we have ever seen, forever, pointing at exactly one video row.

<?php
declare(strict_types=1);

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

$db->exec(<<<'SQL'
CREATE TABLE IF NOT EXISTS videos (
    id            INTEGER PRIMARY KEY,
    canonical_key TEXT    NOT NULL,
    provider      TEXT    NOT NULL,
    provider_id   TEXT    NOT NULL,
    canonical_url TEXT    NOT NULL,
    title         TEXT    NOT NULL,
    lang          TEXT    NOT NULL,
    duration_s    INTEGER,
    first_seen    INTEGER NOT NULL,
    last_seen     INTEGER NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS videos_canonical ON videos(canonical_key);

CREATE TABLE IF NOT EXISTS video_url_aliases (
    normalized_url TEXT    PRIMARY KEY,
    video_id       INTEGER NOT NULL REFERENCES videos(id) ON DELETE CASCADE,
    source         TEXT    NOT NULL,
    seen_at        INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS aliases_video ON video_url_aliases(video_id);

-- unicode61 does not segment 日本語 or 한국어 at all; trigram does.
CREATE VIRTUAL TABLE IF NOT EXISTS videos_fts USING fts5(
    title,
    content       = 'videos',
    content_rowid = 'id',
    tokenize      = "trigram case_sensitive 0"
);
SQL);

function ingest(
    PDO $db,
    string $rawUrl,
    array $meta,
    UrlNormalizer $normalizer,
    CanonicalResolver $resolver,
): ?int {
    $normalized = $normalizer->normalize($rawUrl);
    if ($normalized === null) {
        return null;
    }

    // Hot path: 94% of ingests are URLs we have already resolved once.
    $hit = $db->prepare('SELECT video_id FROM video_url_aliases WHERE normalized_url = ?');
    $hit->execute([$normalized]);
    if (($videoId = $hit->fetchColumn()) !== false) {
        $db->prepare('UPDATE videos SET last_seen = ? WHERE id = ?')
           ->execute([time(), $videoId]);

        return (int) $videoId;
    }

    $canonical = $resolver->resolve($normalized);
    if ($canonical === null) {
        // Unknown provider: quarantine for review. Never invent a key from the URL.
        $db->prepare('INSERT OR IGNORE INTO ingest_quarantine (url, seen_at) VALUES (?, ?)')
           ->execute([$normalized, time()]);

        return null;
    }

    $now = time();
    $db->beginTransaction();
    try {
        $db->prepare(<<<'SQL'
            INSERT INTO videos (canonical_key, provider, provider_id, canonical_url,
                                title, lang, duration_s, first_seen, last_seen)
            VALUES (:key, :provider, :pid, :url, :title, :lang, :dur, :now, :now)
            ON CONFLICT(canonical_key) DO UPDATE SET
                last_seen  = excluded.last_seen,
                duration_s = COALESCE(videos.duration_s, excluded.duration_s),
                title      = CASE WHEN length(excluded.title) > length(videos.title)
                                  THEN excluded.title ELSE videos.title END
            SQL)->execute([
                ':key'      => $canonical->key(),
                ':provider' => $canonical->provider,
                ':pid'      => $canonical->id,
                ':url'      => $normalized,
                ':title'    => $meta['title'],
                ':lang'     => $meta['lang'],
                ':dur'      => $meta['duration_s'] ?? null,
                ':now'      => $now,
            ]);

        // lastInsertId() lies after DO UPDATE. Always read the key back.
        $sel = $db->prepare('SELECT id, title FROM videos WHERE canonical_key = ?');
        $sel->execute([$canonical->key()]);
        [$id, $title] = array_values($sel->fetch(PDO::FETCH_ASSOC));

        $db->prepare(<<<'SQL'
            INSERT OR IGNORE INTO video_url_aliases (normalized_url, video_id, source, seen_at)
            VALUES (?, ?, ?, ?)
            SQL)->execute([$normalized, $id, $meta['source'], $now]);

        // External-content FTS5 does not track writes for you.
        $db->prepare("INSERT INTO videos_fts(videos_fts, rowid, title) VALUES('delete', ?, ?)")
           ->execute([$id, $title]);
        $db->prepare('INSERT INTO videos_fts(rowid, title) VALUES (?, ?)')
           ->execute([$id, $title]);

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

    return (int) $id;
}

Two SQLite-specific traps in there cost us a day each. PDO::lastInsertId() returns the previous insert's rowid after an ON CONFLICT DO UPDATE fires, so aliases were being attached to the wrong video. And an external-content FTS5 table has no triggers unless you write them — updating videos.title without the paired 'delete' command leaves stale trigrams in the index, which surfaces as "search finds a title that no longer exists" weeks later.

Shadow Mode Before You Merge Anything

We did not run the new resolver against production rows directly. We ran it in shadow mode for eleven days, writing proposed keys into a canonical_audit table, then audited the clusters it would have created. The signal that catches false merges is cheap: two rows that claim to be the same video but have durations 90 seconds apart are almost certainly not the same video.

#!/usr/bin/env python3
"""Shadow-mode audit: which rows would merge, and which merges look wrong?"""
import sqlite3
import sys
from collections import defaultdict

SUSPECT_DURATION_DELTA = 5  # seconds


def load_clusters(db_path: str) -> dict[str, list[dict]]:
    con = sqlite3.connect(db_path)
    con.row_factory = sqlite3.Row
    rows = con.execute(
        """
        SELECT v.id, v.title, v.duration_s, v.lang,
               a.canonical_key_new, a.normalized_url
        FROM   canonical_audit a
        JOIN   videos v ON v.id = a.video_id
        """
    )
    buckets: dict[str, list[dict]] = defaultdict(list)
    for r in rows:
        buckets[r["canonical_key_new"]].append(dict(r))
    con.close()
    return {k: v for k, v in buckets.items() if len(v) > 1}


def main(db_path: str) -> int:
    clusters = load_clusters(db_path)
    collapsed = sum(len(v) - 1 for v in clusters.values())
    suspect = 0

    for key, members in sorted(clusters.items()):
        durations = [m["duration_s"] for m in members if m["duration_s"]]
        if durations and (max(durations) - min(durations)) > SUSPECT_DURATION_DELTA:
            suspect += 1
            print(f"SUSPECT {key} durations={sorted(durations)}")
            for m in members:
                print(f"    {m['id']:>8}  {m['lang']}  {m['title'][:60]}")

    print(f"\n{len(clusters)} clusters, {collapsed} rows collapse, {suspect} need review")
    return 1 if suspect else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "data/videos.db"))

Over eleven days it flagged 214 suspect clusters. Roughly 180 were the Bilibili p= bug. The rest were live-stream re-uploads that genuinely share an ID on one partner platform, which we now handle with an explicit part derived from the stream start timestamp.

Backfilling 2.3 Million Rows

The PHP path does about 26k rows/second for normalization alone, which sounds fine until you add the per-row SQLite round trip and realize the backfill would run for most of a day while the ingest cron is also trying to write. We ported the hot path to Go for the one-time backfill, validated it against the PHP implementation on a 50k-row sample until the outputs were byte-identical, and then let it run.

package main

import (
    "database/sql"
    "log"
    "runtime"
    "sync"

    _ "modernc.org/sqlite"
)

type job struct {
    id  int64
    url string
}

type result struct {
    id  int64
    key string
}

const dsn = "file:data/videos.db?_pragma=journal_mode(WAL)&_pragma=busy_timeout(10000)"

func main() {
    // Separate handles: one reader cursor + one writer tx on a single conn deadlocks.
    rdb, err := sql.Open("sqlite", dsn+"&mode=ro")
    if err != nil {
        log.Fatal(err)
    }
    defer rdb.Close()

    wdb, err := sql.Open("sqlite", dsn)
    if err != nil {
        log.Fatal(err)
    }
    defer wdb.Close()
    wdb.SetMaxOpenConns(1) // SQLite serializes writers anyway; be explicit about it

    jobs := make(chan job, 8192)
    results := make(chan result, 8192)

    go func() {
        defer close(jobs)
        rows, err := rdb.Query(`SELECT id, canonical_url FROM videos
                                WHERE canonical_key IS NULL ORDER BY id`)
        if err != nil {
            log.Fatal(err)
        }
        defer rows.Close()
        for rows.Next() {
            var j job
            if err := rows.Scan(&j.id, &j.url); err != nil {
                log.Fatal(err)
            }
            jobs <- j
        }
    }()

    var wg sync.WaitGroup
    for i := 0; i < runtime.NumCPU(); i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range jobs {
                // Canonicalize is the Go port of UrlNormalizer + CanonicalResolver.
                if key, ok := Canonicalize(j.url); ok {
                    results <- result{id: j.id, key: key}
                }
            }
        }()
    }
    go func() { wg.Wait(); close(results) }()

    const batchSize = 5000
    n := 0
    tx, err := wdb.Begin()
    if err != nil {
        log.Fatal(err)
    }
    stmt, err := tx.Prepare(`UPDATE videos SET canonical_key = ? WHERE id = ?`)
    if err != nil {
        log.Fatal(err)
    }

    for r := range results {
        if _, err := stmt.Exec(r.key, r.id); err != nil {
            log.Fatalf("row %d: %v", r.id, err)
        }
        if n++; n%batchSize == 0 {
            stmt.Close()
            if err := tx.Commit(); err != nil {
                log.Fatal(err)
            }
            log.Printf("committed %d rows", n)
            if tx, err = wdb.Begin(); err != nil {
                log.Fatal(err)
            }
            if stmt, err = tx.Prepare(`UPDATE videos SET canonical_key = ? WHERE id = ?`); err != nil {
                log.Fatal(err)
            }
        }
    }

    stmt.Close()
    if err := tx.Commit(); err != nil {
        log.Fatal(err)
    }
    log.Printf("done: %d rows", n)
}

The backfill wrote the whole set in 4 minutes 12 seconds. The unique index on canonical_key was created afterwards, not before — with the index in place, the same run took over an hour because every batch was fighting page splits.

What Broke After Deploy

None of this is the interesting part. The interesting part is what merging 300k rows does to everything downstream:

Roughly 41k watch-page slugs became 301 redirects overnight. We keep a permanent slug_redirects table rather than letting them 404, because Google had already indexed a lot of them and the market we care about is one where organic search is most of our traffic.
Cloudflare kept serving the old pages. Purge-by-prefix on the watch path, then a full LiteSpeed page-cache clear on origin. A cache purge that only hits one layer of a three-layer stack is worse than no purge — you get inconsistent pages depending on which edge POP a user lands on.
Vary: Accept-Language interacted badly with the merge. Merged rows inherited the title from whichever alias had the longest string, which for CJK titles is almost always the shortest semantic title, because CJK conveys more per character. We now pick the title by matching the row's lang to the request locale and fall back to character-count-weighted length.
FTS5 trigram tokenizer needs at least three characters to match. After dedup, search felt better on long queries and identical on two-character CJK queries, which is a limitation of the tokenizer, not the pipeline. We prefix-pad short queries against a separate bigram column.

The numbers, four weeks after: duplicate rate 14.2% → 0.3%, videos row count down 13%, FTS index 22% smaller on disk, search p95 down from 41ms to 26ms. Normalization itself costs about 38µs at p99 per URL, which is noise next to a single SQLite read.

Conclusion

If you take two things from this, take these. First, keep structural normalization and semantic extraction in separate, separately-tested stages — the moment one function does both, every provider quirk you add makes the RFC-compliant part slightly less compliant. Second, store every URL you have ever seen in an alias table and treat the canonical key as disposable. Extractors are wrong on their first version and their fifth version, and the only difference between a bad afternoon and an unrecoverable data loss is whether you kept the inputs.

Run it in shadow mode for a week before you merge a single row. The 214 clusters that audit flagged would all have been silent, permanent data corruption.