feat(backend): store and query reel embeddings in the data layer

Claude-Session: https://claude.ai/code/session_01MV2z14HLUeGpAd4apqTpCh

db.py

@@ -35,13 +35,17 @@ def _chunked(seq: list, size: int):
         yield seq[i:i + size]
 
 
-def _select_newest_first(columns: str, limit: Optional[int] = None) -> list:
-    """Select rows newest-first, paging past PostgREST's per-request row cap."""
+def _select_newest_first(columns: str, limit: Optional[int] = None, refine=None) -> list:
+    """Select rows newest-first, paging past PostgREST's per-request row cap.
+
+    refine, if given, adds filters to each page's query."""
     rows = []
     while limit is None or len(rows) < limit:
         last = len(rows) + 999 if limit is None else limit - 1
-        page = (get_client().table(TABLE).select(columns)
-                .order("created_at", desc=True).order("id")
+        query = get_client().table(TABLE).select(columns)
+        if refine:
+            query = refine(query)
+        page = (query.order("created_at", desc=True).order("id")
                 .range(len(rows), last).execute().data or [])
         if not page:
             break
@@ -78,7 +82,7 @@ def get_reel_by_url(url: str) -> Optional[dict]:
 
 
 def insert_reel(url, title, raw_transcript, post_caption, extracted_json,
-                status="done", source=None) -> dict:
+                status="done", source=None, embedding=None) -> dict:
     row = {
         "id": str(uuid.uuid4()),
         "url": url,
@@ -90,6 +94,8 @@ def insert_reel(url, title, raw_transcript, post_caption, extracted_json,
         "status": status,
         "source": source,
     }
+    if embedding is not None:
+        row["embedding"] = embedding  # list -> pgvector
     get_client().table(TABLE).insert(row).execute()
     return row_to_record(row)
 
@@ -139,6 +145,15 @@ def reels_for_clustering() -> list:
     return _select_newest_first("id, extracted_json")
 
 
+def reels_missing_embedding() -> list:
+    """Rows with no search embedding yet, as [{id, extracted_json}]."""
+    return _select_newest_first("id, extracted_json", refine=lambda q: q.is_("embedding", "null"))
+
+
+def set_embedding(reel_id: str, embedding: list) -> None:
+    get_client().table(TABLE).update({"embedding": embedding}).eq("id", reel_id).execute()
+
+
 def set_cluster(reel_ids: list, cluster: Optional[str]) -> None:
     for chunk in _chunked(reel_ids, 100):
         get_client().table(TABLE).update({"cluster": cluster}).in_("id", chunk).execute()
@@ -179,9 +194,9 @@ def claim_next_pending() -> Optional[dict]:
 
 
 def update_reel_result(reel_id: str, title: str, raw_transcript, post_caption,
-                       extracted_json, status: str = "done") -> None:
+                       extracted_json, status: str = "done", embedding=None) -> None:
     """Write pipeline results back onto an existing (claimed) row."""
-    get_client().table(TABLE).update({
+    payload = {
         "title": title,
         "raw_transcript": raw_transcript,
         "post_caption": post_caption,
@@ -190,7 +205,10 @@ def update_reel_result(reel_id: str, title: str, raw_transcript, post_caption,
         "error": None,
         "processing_started_at": None,
         "next_attempt_at": None,
-    }).eq("id", reel_id).execute()
+    }
+    if embedding is not None:
+        payload["embedding"] = embedding
+    get_client().table(TABLE).update(payload).eq("id", reel_id).execute()
 
 
 def mark_failed(reel_id: str, error, status: str = "failed") -> None:

test_pipeline.py

@@ -65,6 +65,45 @@ class FakeTable:
         return type("Result", (), {"data": self.data})()
 
 
+class RecordingTable:
+    """Stand-in for a Supabase table query that records payloads and filters."""
+    def __init__(self, rows=None):
+        self.rows = rows or []
+        self.payloads = []
+        self.filters = []
+        self.window = (0, 999)
+
+    def select(self, *args, **kwargs):
+        return self
+
+    def insert(self, payload):
+        self.payloads.append(payload)
+        return self
+
+    def update(self, payload):
+        self.payloads.append(payload)
+        return self
+
+    def eq(self, *args):
+        self.filters.append(("eq", args))
+        return self
+
+    def is_(self, *args):
+        self.filters.append(("is", args))
+        return self
+
+    def order(self, *args, **kwargs):
+        return self
+
+    def range(self, start, end):
+        self.window = (start, end)
+        return self
+
+    def execute(self):
+        start, end = self.window
+        return type("Result", (), {"data": self.rows[start:end + 1]})()
+
+
 @pytest.fixture
 def pipeline(monkeypatch):
     """Mock download, transcription and extraction so the pipeline runs offline."""
@@ -498,3 +537,33 @@ def test_db_queue_metadata_contract(monkeypatch):
     assert db.recover_stale_processing(datetime.now(timezone.utc)) == 1
     assert ("eq", ("status", "processing")) in query.calls
     assert any(kind == "lt" and args[0] == "processing_started_at" for kind, args in query.calls)
+
+
+def test_insert_reel_stores_embedding_only_when_given(monkeypatch):
+    table = RecordingTable()
+    monkeypatch.setattr(db, "get_client", fake_client(table))
+    record = db.insert_reel("u", "t", "tr", "cap", {"core_topic": "x"}, embedding=[0.1, 0.2])
+    db.insert_reel("u2", "t", "tr", "cap", {"core_topic": "x"})
+    assert table.payloads[0]["embedding"] == [0.1, 0.2]
+    assert "embedding" not in table.payloads[1]
+    assert "embedding" not in record  # never leaks into the API record
+
+
+def test_update_reel_result_stores_embedding_only_when_given(monkeypatch):
+    table = RecordingTable()
+    monkeypatch.setattr(db, "get_client", fake_client(table))
+    db.update_reel_result("r1", "t", "tr", "cap", {"core_topic": "x"}, embedding=[0.3])
+    db.update_reel_result("r2", "t", "tr", "cap", {"core_topic": "x"})
+    assert table.payloads[0]["embedding"] == [0.3]
+    assert "embedding" not in table.payloads[1]
+
+
+def test_embedding_queue_helpers(monkeypatch):
+    table = RecordingTable([{"id": "a", "extracted_json": {}}, {"id": "b", "extracted_json": {}}])
+    monkeypatch.setattr(db, "get_client", fake_client(table))
+    assert [row["id"] for row in db.reels_missing_embedding()] == ["a", "b"]
+    assert ("is", ("embedding", "null")) in table.filters
+
+    db.set_embedding("a", [0.5])
+    assert table.payloads == [{"embedding": [0.5]}]
+    assert ("eq", ("id", "a")) in table.filters