feat(search): validate the index before the alias points at it

The old sync created a collection, imported batches without reading a
single import response, and swapped the alias regardless. A partial
import published a half-empty index, and two overlapping DAG runs could
prune each other's collections.

Publication now checks every import response and the final document
count before upserting the alias, and holds a session-scoped advisory
lock across the read and the publish so concurrent runs serialise.
Cleanup keeps the previous collection as a rollback pointer and is
best-effort: an uncertain alias response must never delete what might
still be live.

Also parses the Typesense URL properly instead of splitting on colons,
which mangled any host carrying a scheme and a default port.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

pipeline/scripts/sync_typesense.py

@@ -11,6 +11,10 @@ Usage:
 from __future__ import annotations
 
 import argparse
+import json
+import logging
+import re
+import uuid
 import os
 import sys
 import time
@@ -112,63 +116,78 @@ def build_document(row: dict) -> dict:
     return doc
 
 
+def publish_collection(client, rows: list[dict]) -> str:
+    """Validate a new collection before moving the alias; keep rollback data.
+
+    Caller holds the database advisory lock across reading and publication so
+    overlapping school-data DAGs cannot publish or prune each other's work.
+    Failed drafts are left for the next successful publication to prune: a lost
+    alias-update response must never cause deletion of a potentially live index.
+    """
+    if not rows or len({r["urn"] for r in rows}) != len(rows):
+        raise ValueError("Search source must contain nonempty, unique school URNs")
+    name = f"schools_{int(time.time())}_{uuid.uuid4().hex[:12]}"
+    try:
+        previous = client.aliases["schools"].retrieve()["collection_name"]
+    except typesense.exceptions.ObjectNotFound:
+        previous = None
+    client.collections.create({**COLLECTION_SCHEMA, "name": name})
+    for i in range(0, len(rows), 500):
+        batch = [build_document(r) for r in rows[i:i + 500]]
+        results = client.collections[name].documents.import_(batch, {"action": "upsert"})
+        if isinstance(results, str):
+            results = [json.loads(line) for line in results.splitlines() if line.strip()]
+        if len(results) != len(batch) or any(r.get("success") is not True for r in results):
+            raise ValueError("Search import failed; live alias unchanged")
+    if client.collections[name].retrieve()["num_documents"] != len(rows):
+        raise ValueError("Search document count mismatch; live alias unchanged")
+    client.aliases.upsert("schools", {"collection_name": name})
+
+    # Retention is best-effort and must not make successful publication fail.
+    try:
+        keep = {name, previous}
+        keep.update(a["collection_name"] for a in client.aliases.retrieve()["aliases"])
+        for collection in client.collections.retrieve():
+            old = collection["name"]
+            if old not in keep and re.fullmatch(r"schools_\d+(?:_[0-9a-f]+)?", old):
+                client.collections[old].delete()
+    except Exception:
+        logging.getLogger(__name__).exception("Search published, but old collection cleanup failed")
+    return name
+
+
 def sync(typesense_url: str, api_key: str):
+    from urllib.parse import urlparse
+    url = urlparse(typesense_url)
     client = typesense.Client({
-        "nodes": [{"host": typesense_url.split("//")[-1].split(":")[0],
-                    "port": typesense_url.split(":")[-1],
-                    "protocol": "http"}],
+        "nodes": [{"host": url.hostname, "port": str(url.port or 8108),
+                   "protocol": url.scheme}],
         "api_key": api_key,
         "connection_timeout_seconds": 10,
     })
-
-    # Create timestamped collection for zero-downtime swap
-    ts = int(time.time())
-    collection_name = f"schools_{ts}"
-
-    print(f"Creating collection: {collection_name}")
-    schema = {**COLLECTION_SCHEMA, "name": collection_name}
-    client.collections.create(schema)
-
-    # Fetch data from marts — join fact_performance if it exists
     conn = get_db_connection()
-    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
-        # Check whether the merged fact table exists
-        cur.execute("""
-            SELECT table_name FROM information_schema.tables
-            WHERE table_schema = 'marts' AND table_name = 'fact_performance'
-        """)
-        has_fact_performance = cur.fetchone() is not None
-
-        query = QUERY_BASE
-        if has_fact_performance:
-            query = query.replace(
-                "l.longitude as lng",
-                "l.longitude as lng,\n        p.rwm_expected_pct,\n        p.progress_8_score",
-            )
-            query += QUERY_PERFORMANCE_JOIN
-
-        cur.execute(query)
-        rows = cur.fetchall()
-    conn.close()
-
-    print(f"Indexing {len(rows)} schools...")
-
-    # Batch import
-    batch_size = 500
-    for i in range(0, len(rows), batch_size):
-        batch = [build_document(r) for r in rows[i : i + batch_size]]
-        client.collections[collection_name].documents.import_(batch, {"action": "upsert"})
-        print(f"  Indexed {min(i + batch_size, len(rows))}/{len(rows)}")
-
-    # Swap alias
-    print("Swapping alias 'schools' → new collection")
     try:
-        client.aliases.upsert("schools", {"collection_name": collection_name})
-    except Exception:
-        # If alias doesn't exist yet, create it
-        client.aliases.upsert("schools", {"collection_name": collection_name})
-
-    print("Done.")
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            # Session-scoped lock is released even on errors when conn closes.
+            cur.execute("SELECT pg_advisory_lock(731042019)")
+            cur.execute("""
+                SELECT table_name FROM information_schema.tables
+                WHERE table_schema = 'marts' AND table_name = 'fact_performance'
+            """)
+            has_fact_performance = cur.fetchone() is not None
+            query = QUERY_BASE
+            if has_fact_performance:
+                query = query.replace(
+                    "l.longitude as lng",
+                    "l.longitude as lng, p.rwm_expected_pct, p.progress_8_score",
+                )
+                query += QUERY_PERFORMANCE_JOIN
+            cur.execute(query)
+            rows = cur.fetchall()
+        name = publish_collection(client, rows)
+        print(f"Published {len(rows)} schools in {name}")
+    finally:
+        conn.close()
 
 
 def main():