Adding Typo Tolerant Video Title Search With OpenSearch and SQLite FTS5

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

<?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;

#!/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())

<?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',
                ],
            ],
        ];
    }
}

<?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');
    }
}