#!/usr/bin/env -S uv run --script --quiet
# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx", "pydantic-settings"]
# ///
"""
Purge bridgy fed vectors from turbopuffer.

The turso purge script cleaned the database, but vectors in turbopuffer
were never removed. This script finds vectors with platform='other' and
empty base_path, resolves their DIDs via plc.directory, and deletes
confirmed bridgy fed vectors from turbopuffer.

Loops until all bridgy vectors are removed (tpuf caps queries at 10k).

Usage:
    ./scripts/purge-bridgyfed-vectors          # dry run (default)
    ./scripts/purge-bridgyfed-vectors --apply  # actually delete
"""

import argparse
import asyncio
import os
import pathlib
import sys
import time

import httpx
from pydantic_settings import BaseSettings, SettingsConfigDict


def log(msg: str) -> None:
    print(msg, flush=True)


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
    )

    turbopuffer_api_key: str
    turbopuffer_namespace: str = "leaflet-search"

    @classmethod
    def settings_customise_sources(cls, settings_cls, **kwargs):
        """Dotenv file wins over environment variables."""
        return (
            kwargs["init_settings"],
            kwargs["dotenv_settings"],
            kwargs["env_settings"],
            kwargs["file_secret_settings"],
        )


async def resolve_pds(
    client: httpx.AsyncClient, did: str, semaphore: asyncio.Semaphore
) -> str | None:
    """Get PDS endpoint from PLC directory (or did:web .well-known)."""
    async with semaphore:
        try:
            if did.startswith("did:web:"):
                domain = did[len("did:web:"):]
                url = f"https://{domain}/.well-known/did.json"
            else:
                url = f"https://plc.directory/{did}"
            resp = await client.get(url, timeout=30)
            resp.raise_for_status()
            data = resp.json()
            for service in data.get("service", []):
                if service.get("type") == "AtprotoPersonalDataServer":
                    return service["serviceEndpoint"]
        except Exception as e:
            print(f"  lookup failed for {did}: {e}", file=sys.stderr, flush=True)
    return None


def tpuf_export_dids(client: httpx.Client, settings: Settings) -> list[dict]:
    """Page the ENTIRE namespace by id (same pattern as scripts/build-atlas).

    A filtered top_k query maxes at 10k rows and gives no way to see the
    tail — the 2026-06-10 re-sweep declared victory after one clean 10k page
    while ~2.4k bridgy/banned vectors sat beyond it (caught because the atlas
    rebuild, which pages everything, still showed donmai/drivepatents).
    """
    url = f"https://api.turbopuffer.com/v2/namespaces/{settings.turbopuffer_namespace}/query"
    all_rows: list[dict] = []
    last_id = None
    while True:
        body: dict = {
            "rank_by": ["id", "asc"],
            "limit": 10000,
            "include_attributes": ["did"],
        }
        if last_id is not None:
            body["filters"] = ["id", "Gt", last_id]
        resp = client.post(
            url,
            headers={
                "Authorization": f"Bearer {settings.turbopuffer_api_key}",
                "Content-Type": "application/json",
            },
            json=body,
            timeout=120,
        )
        resp.raise_for_status()
        rows = resp.json().get("rows", [])
        if not rows:
            break
        all_rows.extend(rows)
        last_id = rows[-1]["id"]
        log(f"  fetched {len(all_rows)} vectors so far...")
        if len(rows) < 10000:
            break
    return all_rows


def tpuf_delete(client: httpx.Client, settings: Settings, ids: list[str]) -> None:
    """Delete vectors by ID from turbopuffer."""
    url = f"https://api.turbopuffer.com/v2/namespaces/{settings.turbopuffer_namespace}"
    resp = client.post(
        url,
        headers={
            "Authorization": f"Bearer {settings.turbopuffer_api_key}",
            "Content-Type": "application/json",
        },
        json={"deletes": ids},
        timeout=60,
    )
    resp.raise_for_status()


async def resolve_new_dids(
    dids: list[str], cache: dict[str, bool]
) -> dict[str, bool]:
    """Resolve DIDs not already in cache. Returns updated cache."""
    new_dids = [d for d in dids if d not in cache]
    if not new_dids:
        return cache

    log(f"  resolving {len(new_dids)} new DIDs (20 concurrent)...")
    plc_sem = asyncio.Semaphore(20)

    async with httpx.AsyncClient() as client:

        async def resolve_one(did: str) -> tuple[str, bool]:
            pds = await resolve_pds(client, did, plc_sem)
            is_bridgy = pds is not None and "brid.gy" in pds
            return did, is_bridgy

        results = await asyncio.gather(*[resolve_one(d) for d in new_dids])

    for did, is_bridgy in results:
        cache[did] = is_bridgy

    bridgy = sum(1 for _, b in results if b)
    log(f"  {bridgy} bridgy, {len(new_dids) - bridgy} non-bridgy")
    return cache


async def main():
    parser = argparse.ArgumentParser(
        description="Purge bridgy fed vectors from turbopuffer"
    )
    parser.add_argument(
        "--apply", action="store_true", help="Actually delete (default is dry run)"
    )
    args = parser.parse_args()

    try:
        settings = Settings()  # type: ignore
    except Exception as e:
        print(f"error loading settings: {e}", file=sys.stderr)
        print("required env vars: TURBOPUFFER_API_KEY", file=sys.stderr)
        sys.exit(1)

    if not args.apply:
        log("=== DRY RUN (pass --apply to delete) ===\n")

    # banned bulk-archive bots — purged unconditionally. Single source of truth:
    # /banned-dids.txt at the repo root (shared with policy.zig + ingester).
    # registry of who/why/evidence: docs/exclusions.md
    banned_path = pathlib.Path(__file__).resolve().parent.parent / "banned-dids.txt"
    banned = {
        did
        for line in banned_path.read_text().splitlines()
        if (did := line.split("#", 1)[0].strip())
    }

    sync_client = httpx.Client()
    log("exporting full namespace (id-keyed pages)...")
    rows = tpuf_export_dids(sync_client, settings)
    log(f"  {len(rows)} vectors total")

    did_to_ids: dict[str, list[str]] = {}
    for row in rows:
        did_to_ids.setdefault(row.get("did") or "(missing)", []).append(row["id"])
    log(f"  {len(did_to_ids)} distinct DIDs")

    pds_cache: dict[str, bool] = {}
    resolvable = [d for d in did_to_ids if d.startswith("did:")]
    pds_cache = await resolve_new_dids(resolvable, pds_cache)

    ids_to_delete = []
    bad_dids = 0
    for did, ids in did_to_ids.items():
        if did in banned or pds_cache.get(did, False):
            ids_to_delete.extend(ids)
            bad_dids += 1

    log(f"\n{bad_dids} banned/bridgy DIDs, {len(ids_to_delete)} vectors to delete")

    if not ids_to_delete:
        sync_client.close()
        log("\nnothing to purge!")
        return

    if not args.apply:
        sync_client.close()
        log(f"\nwould delete {len(ids_to_delete)} vectors from {bad_dids} DIDs")
        return

    total_deleted = 0
    for i in range(0, len(ids_to_delete), 500):
        batch = ids_to_delete[i : i + 500]
        tpuf_delete(sync_client, settings, batch)
        total_deleted += len(batch)
        log(f"  deleted {len(batch)} vectors (total: {total_deleted})")

    sync_client.close()
    log(f"\ndone! deleted {total_deleted} vectors from {bad_dids} DIDs")


if __name__ == "__main__":
    asyncio.run(main())
