"""Discovery expansion over the social graph, read from PDSes rather than an index.

A PR author is only counted if the crawl knows the person. Repo ownership plus
Constellation backlinks missed at least one real author (juliet.tngl.sh: a
self-hosted PDS, one repo, one PR, and no pull backlink in Constellation's index).

Follows, knot memberships and spindle memberships are all records a person keeps
in their OWN repository, so reading them needs no index and has no coverage gap.
Everyone those records point at is a Tangled user. Iterate to closure, then the
pull crawl can be re-run over the enlarged set.
"""
import json, os, queue, threading, time, urllib.parse, urllib.request

SD = "/tmp/claude-1000/-home-jmm-Code-blogposts/7ba1f72e-807b-4749-bcd3-a065da793e6d/scratchpad"
UA = "tangled-stats-research/0.5 (permadeath.com; blog post stats)"
COLLECTIONS = ["sh.tangled.graph.follow", "sh.tangled.knot.member",
               "sh.tangled.spindle.member", "sh.tangled.repo.collaborator"]
MAX_ROUNDS = 5


def get(url, tries=2, timeout=20):
    for i in range(tries):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": UA})
            with urllib.request.urlopen(req, timeout=timeout) as f:
                return json.load(f)
        except urllib.error.HTTPError as e:
            if e.code in (400, 404):
                return None
            if i == tries - 1:
                return None
            time.sleep(2)
        except Exception:
            if i == tries - 1:
                return None
            time.sleep(2)
    return None


def par(items, fn, n=16, label="", every=500):
    q = queue.Queue()
    [q.put(i) for i in items]
    lock = threading.Lock()
    res = []
    c = [0]

    def w():
        while True:
            try:
                it = q.get_nowait()
            except queue.Empty:
                return
            try:
                v = fn(it)
                if v is not None:
                    with lock:
                        res.append(v)
            except Exception:
                pass
            finally:
                with lock:
                    c[0] += 1
                    if c[0] % every == 0:
                        print(f"  {label} {c[0]}/{len(items)}", flush=True)
                q.task_done()

    ts = [threading.Thread(target=w, daemon=True) for _ in range(n)]
    [t.start() for t in ts]
    [t.join() for t in ts]
    return res


def pds_of(did):
    doc = get("https://" + did[8:] + "/.well-known/did.json") if did.startswith("did:web:") \
        else get(f"https://plc.directory/{did}")
    if not doc:
        return (did, None)
    for s in doc.get("service", []):
        if s.get("id", "").endswith("atproto_pds"):
            return (did, s["serviceEndpoint"])
    return (did, None)


def list_records(host, did, collection):
    out, cursor = [], None
    for _ in range(500):
        u = (f"{host}/xrpc/com.atproto.repo.listRecords?repo={urllib.parse.quote(did)}"
             f"&collection={collection}&limit=100")
        if cursor:
            u += "&cursor=" + urllib.parse.quote(cursor)
        d = get(u)
        if not d:
            break
        out += d.get("records", [])
        cursor = d.get("cursor")
        if not cursor or not d.get("records"):
            break
    return out


DID_KEYS = ("subject", "member", "did", "collaborator")


def dids_in(value):
    """Any DID this record points at."""
    found = set()

    def walk(v):
        if isinstance(v, str):
            if v.startswith("did:"):
                found.add(v)
            elif v.startswith("at://did:"):
                found.add(v[5:].split("/")[0])
        elif isinstance(v, dict):
            for x in v.values():
                walk(x)
        elif isinstance(v, list):
            for x in v:
                walk(x)

    walk(value)
    return found


def main():
    PM = json.load(open(f"{SD}/pds_map.json"))
    people = set(json.load(open(f"{SD}/closure_people.json"))) if os.path.exists(f"{SD}/closure_people.json") else set(PM)
    people |= set(PM)
    scanned = set(json.load(open(f"{SD}/social_scanned.json"))) if os.path.exists(f"{SD}/social_scanned.json") else set()
    print(f"start: {len(people)} people, {len(scanned)} already scanned", flush=True)

    for rnd in range(1, MAX_ROUNDS + 1):
        todo = [d for d in people if d not in scanned and PM.get(d)]
        print(f"\n=== round {rnd}: scanning {len(todo)} people for social records", flush=True)
        if not todo:
            print("nothing to scan", flush=True)
            break
        lock = threading.Lock()
        found = set()

        def scan(did):
            hits = set()
            for coll in COLLECTIONS:
                for rec in list_records(PM[did], did, coll):
                    hits |= dids_in(rec.get("value") or {})
            with lock:
                found.update(hits)
                scanned.add(did)
            return None

        par(todo, scan, n=16, label="social")
        new = {d for d in found if d.startswith("did:") and d not in people}
        print(f"  DIDs referenced: {len(found)}   NEW people: {len(new)}", flush=True)

        if new:
            print(f"  resolving {len(new)} PDSes", flush=True)
            PM.update(dict(par(sorted(new), pds_of, n=10, label="pds")))
            json.dump(PM, open(f"{SD}/pds_map.json", "w"))
        people |= new
        json.dump(sorted(people), open(f"{SD}/closure_people.json", "w"))
        json.dump(sorted(scanned), open(f"{SD}/social_scanned.json", "w"))
        if not new:
            print("  closed - no new people", flush=True)
            break

    print(f"\nDONE  people={len(people)}", flush=True)


if __name__ == "__main__":
    main()