reduce write amplification, fix search ranking · zzstoatzz.io/typeahead@3c56b2e

+8 -1

schema.sql

··· 33 33 VALUES ('delete', old.rowid, old.handle, old.display_name); 34 34 END; 35 35 36 - CREATE TRIGGER IF NOT EXISTS actors_au AFTER UPDATE ON actors BEGIN 36 + CREATE TRIGGER IF NOT EXISTS actors_au AFTER UPDATE ON actors 37 + WHEN old.handle <> new.handle OR old.display_name <> new.display_name 38 + BEGIN 37 39 INSERT INTO actors_fts(actors_fts, rowid, handle, display_name) 38 40 VALUES ('delete', old.rowid, old.handle, old.display_name); 39 41 INSERT INTO actors_fts(rowid, handle, display_name) ··· 65 67 domain TEXT PRIMARY KEY, 66 68 hits INTEGER NOT NULL DEFAULT 0 67 69 ); 70 + 71 + CREATE TABLE IF NOT EXISTS tombstones ( 72 + did TEXT PRIMARY KEY, 73 + deleted_at INTEGER NOT NULL 74 + );

+232

scripts/bulk-enrich.py

··· 1 + #!/usr/bin/env -S PYTHONUNBUFFERED=1 uv run --script --quiet 2 + # /// script 3 + # requires-python = ">=3.12" 4 + # dependencies = [] 5 + # /// 6 + """ 7 + bulk enrich actors via bsky getProfiles API. 8 + 9 + pages through all actors by rowid, calls getProfiles in concurrent batches, 10 + writes back handles, avatars, labels, createdAt, associated. 11 + 12 + writes in small batches (50 stmts) with 200ms pauses between to avoid 13 + saturating Turso and blocking production reads. 14 + 15 + usage: 16 + TURSO_URL=... TURSO_AUTH_TOKEN=... ./scripts/bulk-enrich.py 17 + TURSO_URL=... TURSO_AUTH_TOKEN=... ./scripts/bulk-enrich.py --start-rowid 2529 18 + TURSO_URL=... TURSO_AUTH_TOKEN=... ./scripts/bulk-enrich.py --dry-run 19 + """ 20 + 21 + import argparse 22 + import json 23 + import os 24 + import re 25 + import sys 26 + import time 27 + import urllib.request 28 + import urllib.error 29 + from concurrent.futures import ThreadPoolExecutor, as_completed 30 + 31 + BSKY_GET_PROFILES = "https://public.api.bsky.app/xrpc/app.bsky.actor.getProfiles" 32 + PAGE_SIZE = 500 # DIDs per Turso read page 33 + BSKY_CONCURRENCY = 5 # concurrent getProfiles calls 34 + WRITE_BATCH = 50 # stmts per Turso write — keep write lock short 35 + WRITE_PAUSE = 0.2 # seconds between write batches — let reads through 36 + 37 + DIM = "\033[2m" 38 + RESET = "\033[0m" 39 + 40 + 41 + def get_turso_url() -> str: 42 + url = os.environ.get("TURSO_URL", "") 43 + if not url: 44 + print("error: TURSO_URL not set", file=sys.stderr); sys.exit(1) 45 + return url.replace("libsql://", "https://") 46 + 47 + 48 + def get_turso_token() -> str: 49 + token = os.environ.get("TURSO_AUTH_TOKEN", "") 50 + if not token: 51 + print("error: TURSO_AUTH_TOKEN not set", file=sys.stderr); sys.exit(1) 52 + return token 53 + 54 + 55 + def turso_query(sql, args, turso_url, turso_token): 56 + body = json.dumps({"requests": [ 57 + {"type": "execute", "stmt": {"sql": sql, "args": args}}, 58 + {"type": "close"}, 59 + ]}).encode() 60 + req = urllib.request.Request(f"{turso_url}/v3/pipeline", data=body, headers={ 61 + "Authorization": f"Bearer {turso_token}", "Content-Type": "application/json", 62 + }) 63 + with urllib.request.urlopen(req, timeout=30) as resp: 64 + result = json.loads(resp.read()) 65 + res = result["results"][0] 66 + if res.get("type") == "error": 67 + print(f" turso error: {res['error']['message']}", file=sys.stderr) 68 + return [] 69 + cols = [c["name"] for c in res["response"]["result"]["cols"]] 70 + return [{c: (v["value"] if v["type"] != "null" else None) for c, v in zip(cols, row)} 71 + for row in res["response"]["result"]["rows"]] 72 + 73 + 74 + def turso_batch_write(stmts, turso_url, turso_token): 75 + reqs = [{"type": "execute", "stmt": s} for s in stmts] 76 + reqs.append({"type": "close"}) 77 + body = json.dumps({"requests": reqs}).encode() 78 + req = urllib.request.Request(f"{turso_url}/v3/pipeline", data=body, headers={ 79 + "Authorization": f"Bearer {turso_token}", "Content-Type": "application/json", 80 + }) 81 + try: 82 + with urllib.request.urlopen(req, timeout=60) as resp: 83 + json.loads(resp.read()) 84 + return True 85 + except Exception as e: 86 + print(f"\n turso write failed: {e}", file=sys.stderr) 87 + return False 88 + 89 + 90 + def extract_avatar_cid(url): 91 + if not url: return "" 92 + m = re.search(r'/([^/]+?)(?:@[a-z]+)?$', url) 93 + return m.group(1) if m else "" 94 + 95 + 96 + def clean_associated(assoc): 97 + if not assoc or not isinstance(assoc, dict): return "{}" 98 + clean = {k: v for k, v in assoc.items() if v not in (0, False, None)} 99 + return json.dumps(clean) if clean else "{}" 100 + 101 + 102 + def fetch_profiles(dids): 103 + params = "&".join(f"actors={urllib.request.quote(d)}" for d in dids) 104 + req = urllib.request.Request( 105 + f"{BSKY_GET_PROFILES}?{params}", 106 + headers={"User-Agent": "typeahead-enrich/1.0"}, 107 + ) 108 + try: 109 + with urllib.request.urlopen(req, timeout=15) as resp: 110 + return json.loads(resp.read()).get("profiles", []) 111 + except urllib.error.HTTPError as e: 112 + if e.code == 429: return "rate_limited" 113 + return [] 114 + except Exception: 115 + return [] 116 + 117 + 118 + def profile_to_stmt(p): 119 + hide_vals = {"!hide", "!takedown", "!suspend", "spam"} 120 + mod_did = "did:plc:ar7c4by46qjdydhdevvrndac" 121 + hidden = 0 122 + for lbl in (p.get("labels") or []): 123 + if lbl.get("val", "") in hide_vals or (lbl.get("val") == "!no-unauthenticated" and lbl.get("src") == mod_did): 124 + hidden = 1; break 125 + 126 + return { 127 + "sql": """UPDATE actors SET 128 + handle = COALESCE(NULLIF(?2, ''), handle), 129 + display_name = COALESCE(NULLIF(?3, ''), display_name), 130 + avatar_url = COALESCE(NULLIF(?4, ''), avatar_url), 131 + labels = ?5, hidden = ?6, 132 + created_at = COALESCE(NULLIF(?7, ''), created_at), 133 + associated = COALESCE(NULLIF(?8, '{}'), associated), 134 + profile_checked_at = unixepoch() 135 + WHERE did = ?1""", 136 + "args": [ 137 + {"type": "text", "value": p["did"]}, 138 + {"type": "text", "value": p.get("handle", "")}, 139 + {"type": "text", "value": p.get("displayName", "")}, 140 + {"type": "text", "value": extract_avatar_cid(p.get("avatar", ""))}, 141 + {"type": "text", "value": json.dumps(p.get("labels", []))}, 142 + {"type": "integer", "value": str(hidden)}, 143 + {"type": "text", "value": p.get("createdAt", "")}, 144 + {"type": "text", "value": clean_associated(p.get("associated"))}, 145 + ], 146 + } 147 + 148 + 149 + def main(): 150 + parser = argparse.ArgumentParser() 151 + parser.add_argument("--dry-run", action="store_true") 152 + parser.add_argument("--start-rowid", type=int, default=0) 153 + args = parser.parse_args() 154 + 155 + turso_url = get_turso_url() 156 + turso_token = get_turso_token() 157 + 158 + enriched = 0 159 + checked = 0 160 + not_found = 0 161 + t0 = time.time() 162 + last_rowid = args.start_rowid 163 + 164 + print(f"bulk enriching (concurrency={BSKY_CONCURRENCY}, write_batch={WRITE_BATCH}, write_pause={WRITE_PAUSE}s)...") 165 + if args.dry_run: print(" DRY RUN — no writes") 166 + if args.start_rowid: print(f" resuming from rowid {args.start_rowid}") 167 + 168 + while True: 169 + rows = turso_query( 170 + "SELECT rowid, did FROM actors WHERE rowid > ?1 ORDER BY rowid ASC LIMIT ?2", 171 + [{"type": "integer", "value": str(last_rowid)}, {"type": "integer", "value": str(PAGE_SIZE)}], 172 + turso_url, turso_token, 173 + ) 174 + if not rows: 175 + break 176 + 177 + dids = [r["did"] for r in rows] 178 + last_rowid = int(rows[-1]["rowid"]) 179 + 180 + # getProfiles in concurrent batches of 25 181 + batches = [dids[i:i+25] for i in range(0, len(dids), 25)] 182 + pending = [] 183 + 184 + with ThreadPoolExecutor(max_workers=BSKY_CONCURRENCY) as pool: 185 + futures = {pool.submit(fetch_profiles, b): b for b in batches} 186 + for future in as_completed(futures): 187 + batch = futures[future] 188 + result = future.result() 189 + 190 + if result == "rate_limited": 191 + print(f"\n rate limited — pausing 30s...") 192 + time.sleep(30) 193 + result = fetch_profiles(batch) 194 + if result == "rate_limited": 195 + print(" still limited, skipping batch") 196 + checked += len(batch) 197 + continue 198 + 199 + checked += len(batch) 200 + returned = {p["did"] for p in result} 201 + not_found += len(batch) - len(returned) 202 + 203 + for p in result: 204 + pending.append(profile_to_stmt(p)) 205 + enriched += 1 206 + 207 + # write in small batches with pauses between each 208 + if pending and not args.dry_run: 209 + write_t0 = time.time() 210 + write_chunks = 0 211 + for i in range(0, len(pending), WRITE_BATCH): 212 + turso_batch_write(pending[i:i+WRITE_BATCH], turso_url, turso_token) 213 + write_chunks += 1 214 + time.sleep(WRITE_PAUSE) 215 + write_ms = int((time.time() - write_t0) * 1000) 216 + print(f" wrote {len(pending)} stmts in {write_chunks} chunks ({write_ms}ms)") 217 + 218 + elapsed = time.time() - t0 219 + rate = checked / elapsed if elapsed > 0 else 0 220 + tag = "dry" if args.dry_run else "live" 221 + print( 222 + f" [{tag}] checked={checked:,} enriched={enriched:,} " 223 + f"not_found={not_found:,} " 224 + f"{DIM}{rate:.0f} dids/s rowid={last_rowid}{RESET}" 225 + ) 226 + 227 + elapsed = time.time() - t0 228 + print(f"\ndone in {elapsed:.0f}s. enriched={enriched:,}, not_found={not_found:,}, checked={checked:,}") 229 + 230 + 231 + if __name__ == "__main__": 232 + main()

+79 -14

scripts/cleanup-dead-actors.py

··· 27 27 from concurrent.futures import ThreadPoolExecutor, as_completed 28 28 29 29 BSKY_GET_PROFILES = "https://public.api.bsky.app/xrpc/app.bsky.actor.getProfiles" 30 + BSKY_MOD_DID = "did:plc:ar7c4by46qjdydhdevvrndac" 31 + MOD_HIDE_VALS = {"!hide", "!takedown", "spam"} 30 32 PAGE_SIZE = 500 31 33 BSKY_CONCURRENCY = 5 32 - DELETE_BATCH = 100 33 - DELETE_PAUSE = 0.1 34 + WRITE_BATCH = 25 # max statements per turso batch (keep write lock short) 35 + WRITE_PAUSE = 0.5 # seconds between write batches (let reads through) 34 36 35 37 DIM = "\033[2m" 36 38 RESET = "\033[0m" ··· 38 40 GREEN = "\033[32m" 39 41 40 42 43 + def extract_avatar_cid(url: str) -> str: 44 + """extract CID from bsky avatar URL (mirrors extractAvatarCid in utils.ts)""" 45 + import re 46 + m = re.search(r'/([^/]+?)(?:@[a-z]+)?$', url) 47 + return m.group(1) if m else '' 48 + 49 + 50 + def should_hide(labels: list) -> bool: 51 + """mirrors shouldHide in moderation.ts""" 52 + import datetime 53 + now = datetime.datetime.now(datetime.timezone.utc) 54 + for l in labels: 55 + if l.get("neg"): continue 56 + exp = l.get("exp") 57 + if exp: 58 + try: 59 + if datetime.datetime.fromisoformat(exp.replace("Z", "+00:00")) <= now: 60 + continue 61 + except ValueError: 62 + continue 63 + if l.get("src") == BSKY_MOD_DID and l.get("val") in MOD_HIDE_VALS: 64 + return True 65 + return False 66 + 67 + 68 + def extract_profile_fields(profile: dict) -> dict: 69 + """mirrors extractProfileFields in utils.ts — full profile extraction""" 70 + labels = profile.get("labels", []) 71 + associated = profile.get("associated", {}) 72 + # clean associated: strip zero/false/null values 73 + clean_assoc = {k: v for k, v in (associated or {}).items() 74 + if v not in (0, False, None)} 75 + return { 76 + "handle": profile.get("handle", ""), 77 + "display_name": profile.get("displayName", ""), 78 + "avatar_cid": extract_avatar_cid(profile.get("avatar", "")), 79 + "labels": json.dumps(labels), 80 + "hidden": 1 if should_hide(labels) else 0, 81 + "created_at": profile.get("createdAt", ""), 82 + "associated": json.dumps(clean_assoc) if clean_assoc else "{}", 83 + } 84 + 85 + 41 86 def get_turso_url() -> str: 42 87 url = os.environ.get("TURSO_URL", "") 43 88 if not url: ··· 121 166 last_rowid = args.start_rowid 122 167 123 168 tag = "dry" if args.dry_run else "live" 124 - print(f"cleanup dead actors (concurrency={BSKY_CONCURRENCY}, delete_batch={DELETE_BATCH})") 169 + print(f"cleanup dead actors (concurrency={BSKY_CONCURRENCY}, write_batch={WRITE_BATCH}, write_pause={WRITE_PAUSE}s)") 125 170 if args.dry_run: print(" DRY RUN — no writes") 126 171 if args.limit: print(f" limit: {args.limit:,}") 127 172 if args.start_rowid: print(f" resuming from rowid {args.start_rowid}") ··· 176 221 p = returned[did] 177 222 handle = p.get("handle", "") 178 223 if handle: 179 - to_update.append({"did": did, "handle": handle}) 224 + fields = extract_profile_fields(p) 225 + fields["did"] = did 226 + to_update.append(fields) 180 227 else: 181 228 alive += 1 # deactivated but bsky knows it 182 229 else: ··· 189 236 print(f" {RED}DELETE{RESET} {d['did']} {DIM}{name}{RESET}") 190 237 191 238 for u in to_update: 192 - print(f" {GREEN}UPDATE{RESET} {u['did']} → {u['handle']}") 239 + avatar = "+" if u.get("avatar_cid") else "-" 240 + print(f" {GREEN}UPDATE{RESET} {u['did']} → {u['handle']} avatar={avatar}") 193 241 194 242 if not args.dry_run: 195 - # batch deletes 243 + # all writes go through throttled_write to keep lock hold times short 196 244 if to_delete: 197 245 del_stmts = [ 198 246 {"sql": "DELETE FROM actors WHERE did = ?1", 199 247 "args": [{"type": "text", "value": d["did"]}]} 200 248 for d in to_delete 201 249 ] 202 - for i in range(0, len(del_stmts), DELETE_BATCH): 203 - turso_batch_write(del_stmts[i:i+DELETE_BATCH], turso_url, turso_token) 204 - time.sleep(DELETE_PAUSE) 250 + for i in range(0, len(del_stmts), WRITE_BATCH): 251 + turso_batch_write(del_stmts[i:i+WRITE_BATCH], turso_url, turso_token) 252 + time.sleep(WRITE_PAUSE) 205 253 206 - # batch updates (handle recovery) 207 254 if to_update: 208 255 upd_stmts = [ 209 - {"sql": "UPDATE actors SET handle = ?2 WHERE did = ?1", 210 - "args": [{"type": "text", "value": u["did"]}, 211 - {"type": "text", "value": u["handle"]}]} 256 + {"sql": """UPDATE actors SET 257 + handle = COALESCE(NULLIF(?2, ''), handle), 258 + display_name = COALESCE(NULLIF(?3, ''), display_name), 259 + avatar_url = COALESCE(NULLIF(?4, ''), avatar_url), 260 + labels = ?5, hidden = ?6, 261 + created_at = COALESCE(NULLIF(?7, ''), created_at), 262 + associated = COALESCE(NULLIF(?8, '{}'), associated), 263 + profile_checked_at = unixepoch() 264 + WHERE did = ?1""", 265 + "args": [ 266 + {"type": "text", "value": u["did"]}, 267 + {"type": "text", "value": u["handle"]}, 268 + {"type": "text", "value": u.get("display_name", "")}, 269 + {"type": "text", "value": u.get("avatar_cid", "")}, 270 + {"type": "text", "value": u.get("labels", "[]")}, 271 + {"type": "integer", "value": str(u.get("hidden", 0))}, 272 + {"type": "text", "value": u.get("created_at", "")}, 273 + {"type": "text", "value": u.get("associated", "{}")}, 274 + ]} 212 275 for u in to_update 213 276 ] 214 - turso_batch_write(upd_stmts, turso_url, turso_token) 277 + for i in range(0, len(upd_stmts), WRITE_BATCH): 278 + turso_batch_write(upd_stmts[i:i+WRITE_BATCH], turso_url, turso_token) 279 + time.sleep(WRITE_PAUSE) 215 280 216 281 deleted += len(to_delete) 217 282 updated += len(to_update)

+20 -6

src/cron.ts

··· 11 11 const cursor = cursorStr ? Number(cursorStr) : 0; 12 12 13 13 const { results } = await db.prepare( 14 - "SELECT rowid, did, handle FROM actors WHERE rowid > ?1 ORDER BY rowid ASC LIMIT 1000" 15 - ).bind(cursor).all<{ rowid: number; did: string; handle: string }>(); 14 + "SELECT rowid, did, handle, hidden, labels FROM actors WHERE rowid > ?1 ORDER BY rowid ASC LIMIT 1000" 15 + ).bind(cursor).all<{ rowid: number; did: string; handle: string; hidden: number; labels: string }>(); 16 16 17 17 if (!results || results.length === 0) { 18 18 // wrapped around — reset cursor for next run ··· 24 24 let checked = 0; 25 25 let changed = 0; 26 26 let deleted = 0; 27 + let skipped = 0; 28 + const t0 = Date.now(); 27 29 28 30 // batch into groups of 25 (getProfiles limit), ~200ms pause between calls 29 31 for (let i = 0; i < results.length; i += 25) { ··· 46 48 const profiles: any[] = data.profiles || []; 47 49 checked += profiles.length; 48 50 51 + // build a lookup of current DB state for this batch 52 + const current = new Map(batch.map((r) => [r.did, r])); 53 + 49 54 const stmts: Stmt[] = []; 50 55 for (const p of profiles) { 51 56 const f = extractProfileFields(p); 57 + const cur = current.get(p.did); 58 + // skip no-op updates — only write if labels or hidden actually changed 59 + if (cur && cur.hidden === f.hidden && cur.labels === f.labels) { skipped++; continue; } 60 + 52 61 stmts.push( 53 62 db.prepare( 54 63 `UPDATE actors SET hidden = ?1, ··· 72 81 const returnedDids = new Set(profiles.map((p: any) => p.did)); 73 82 const deadDids = batch.filter((r) => !returnedDids.has(r.did) && r.handle === ""); 74 83 if (deadDids.length > 0) { 75 - const delStmts: Stmt[] = deadDids.map((r) => 76 - db.prepare("DELETE FROM actors WHERE did = ?1").bind(r.did) 77 - ); 84 + const delStmts: Stmt[] = deadDids.flatMap((r) => [ 85 + db.prepare("DELETE FROM actors WHERE did = ?1").bind(r.did), 86 + db.prepare("INSERT OR REPLACE INTO tombstones (did, deleted_at) VALUES (?1, unixepoch())").bind(r.did), 87 + ]); 78 88 await db.batch(delStmts); 79 89 deleted += deadDids.length; 80 90 await recordActorDelta(db, { actors: -deadDids.length }).catch(() => {}); ··· 87 97 // save cursor at end of this page 88 98 const lastRowid = results[results.length - 1].rowid; 89 99 await env.KV.put("mod_cursor", String(lastRowid)); 100 + 101 + // prune old tombstones (> 7 days) 102 + await db.prepare("DELETE FROM tombstones WHERE deleted_at < unixepoch() - 604800").run(); 103 + 90 104 if (deleted > 0) { 91 105 console.log(JSON.stringify({ event: "mod_cleanup", deleted })); 92 106 } 93 - console.log(JSON.stringify({ event: "moderation_refresh", checked, changed, deleted, cursor: lastRowid })); 107 + console.log(JSON.stringify({ event: "moderation_refresh", checked, changed, skipped, deleted, cursor: lastRowid, ms: Date.now() - t0 })); 94 108 }

+23 -6

src/handlers/admin.ts

··· 33 33 return json({ error: "batch too large (max 10000)" }, 400); 34 34 } 35 35 36 - const stmts = dids.map((did) => 37 - db.prepare("DELETE FROM actors WHERE did = ?1").bind(did) 38 - ); 39 - const batchResults = await db.batch(stmts); 40 - const actualDeletes = batchResults.reduce((s, r) => s + r.meta.changes, 0); 36 + // chunk into batches of 200 with short pauses to let reads through 37 + const CHUNK = 200; 38 + let actualDeletes = 0; 39 + const t0 = Date.now(); 40 + 41 + for (let i = 0; i < dids.length; i += CHUNK) { 42 + if (i > 0) await new Promise((r) => setTimeout(r, 100)); 43 + const chunk = dids.slice(i, i + CHUNK); 44 + const stmts = chunk.flatMap((did) => [ 45 + db.prepare("DELETE FROM actors WHERE did = ?1").bind(did), 46 + db.prepare("INSERT OR REPLACE INTO tombstones (did, deleted_at) VALUES (?1, unixepoch())").bind(did), 47 + ]); 48 + const batchResults = await db.batch(stmts); 49 + actualDeletes += batchResults 50 + .filter((_, j) => j % 2 === 0) 51 + .reduce((s, r) => s + r.meta.changes, 0); 52 + } 53 + 54 + console.log(JSON.stringify({ 55 + event: "admin_delete", requested: dids.length, deleted: actualDeletes, 56 + chunks: Math.ceil(dids.length / CHUNK), ms: Date.now() - t0, 57 + })); 41 58 42 59 if (actualDeletes > 0) { 43 60 ctx.waitUntil(recordActorDelta(db, { actors: -actualDeletes })); 44 61 } 45 62 46 - return json({ ok: true, deleted: dids.length }); 63 + return json({ ok: true, deleted: actualDeletes }); 47 64 } 48 65 49 66 export async function handleCursor(

+42 -20

src/handlers/search.ts

··· 41 41 42 42 const t0 = Date.now(); 43 43 44 + // 3-tier ranking: exact handle → handle prefix → FTS prefix 44 45 const ftsQuery = `"${term}"*`; 45 - const { results } = await db.prepare( 46 - `SELECT a.did, a.handle, a.display_name, a.avatar_url, a.labels, a.created_at, a.associated 47 - FROM actors_fts 48 - JOIN actors a ON a.rowid = actors_fts.rowid 49 - WHERE actors_fts MATCH ?1 AND a.handle != '' AND a.hidden = 0 50 - ORDER BY rank 51 - LIMIT ?2` 52 - ) 53 - .bind(ftsQuery, limit) 54 - .all<ActorRow>(); 46 + const [exactRes, prefixRes, ftsRes] = await db.batch([ 47 + db.prepare( 48 + `SELECT did, handle, display_name, avatar_url, labels, created_at, associated 49 + FROM actors WHERE handle = ?1 COLLATE NOCASE AND hidden = 0 LIMIT 1` 50 + ).bind(term), 51 + db.prepare( 52 + `SELECT did, handle, display_name, avatar_url, labels, created_at, associated 53 + FROM actors WHERE handle LIKE ?1 COLLATE NOCASE AND handle != ?2 COLLATE NOCASE AND hidden = 0 54 + ORDER BY length(handle) LIMIT ?3` 55 + ).bind(term + '%', term, limit), 56 + db.prepare( 57 + `SELECT a.did, a.handle, a.display_name, a.avatar_url, a.labels, a.created_at, a.associated 58 + FROM actors_fts 59 + JOIN actors a ON a.rowid = actors_fts.rowid 60 + WHERE actors_fts MATCH ?1 AND a.handle != '' AND a.hidden = 0 61 + ORDER BY rank 62 + LIMIT ?2` 63 + ).bind(ftsQuery, limit), 64 + ], "read"); 55 65 56 - const actors = (results || []) 57 - .map((r) => ({ 58 - did: r.did, 59 - handle: r.handle, 60 - ...(r.display_name ? { displayName: r.display_name } : {}), 61 - ...(r.avatar_url ? { avatar: avatarUrl(r.did, r.avatar_url) } : {}), 62 - ...(r.associated && r.associated !== '{}' ? { associated: JSON.parse(r.associated) } : {}), 63 - labels: JSON.parse(r.labels || '[]'), 64 - ...(r.created_at ? { createdAt: r.created_at } : {}), 65 - })); 66 + const seen = new Set<string>(); 67 + const merged: ActorRow[] = []; 68 + for (const row of [ 69 + ...((exactRes.results as ActorRow[]) || []), 70 + ...((prefixRes.results as ActorRow[]) || []), 71 + ...((ftsRes.results as ActorRow[]) || []), 72 + ]) { 73 + if (seen.has(row.did)) continue; 74 + seen.add(row.did); 75 + merged.push(row); 76 + if (merged.length >= limit) break; 77 + } 78 + 79 + const actors = merged.map((r) => ({ 80 + did: r.did, 81 + handle: r.handle, 82 + ...(r.display_name ? { displayName: r.display_name } : {}), 83 + ...(r.avatar_url ? { avatar: avatarUrl(r.did, r.avatar_url) } : {}), 84 + ...(r.associated && r.associated !== '{}' ? { associated: JSON.parse(r.associated) } : {}), 85 + labels: JSON.parse(r.labels || '[]'), 86 + ...(r.created_at ? { createdAt: r.created_at } : {}), 87 + })); 66 88 67 89 // --- backfill: remove this block once at parity with Bluesky --- 68 90 const hasGaps = actors.length < limit || actors.some((a) => !a.avatar);

Configure Feed

Configure Feed