From 9f81b7fa6889326f8ea5724a2244266006b261b5 Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 30 Jul 2026 11:59:23 +0300 Subject: [PATCH] cmd/prefill-zoekt,zoekt-tngl-indexserver: type repoDIDs & knot URLs via repoident Lewis: May this revision serve well! --- .gitignore | 2 + cmd/prefill-zoekt/main.go | 137 +++++++++++++++------------ cmd/zoekt-tngl-indexserver/index.go | 24 ++--- cmd/zoekt-tngl-indexserver/main.go | 28 ++++-- cmd/zoekt-tngl-indexserver/queue.go | 8 +- cmd/zoekt-tngl-indexserver/server.go | 4 +- docker-compose.yml | 1 + 7 files changed, 118 insertions(+), 86 deletions(-) diff --git a/.gitignore b/.gitignore index 79760d1a..2c808aad 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,8 @@ tmp appview/pages/static/* spindle/spindle spindle/spindle-microvm-run +/prefill-zoekt +/zoekt-tngl-indexserver result !.gitkeep !appview/pages/static/topbar-search.js diff --git a/cmd/prefill-zoekt/main.go b/cmd/prefill-zoekt/main.go index 4e0cdfa7..d70d3b3e 100644 --- a/cmd/prefill-zoekt/main.go +++ b/cmd/prefill-zoekt/main.go @@ -17,6 +17,7 @@ import ( "fmt" "log" "net/http" + "net/url" "os" "os/exec" "strings" @@ -25,16 +26,27 @@ import ( "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/samber/lo" "github.com/sourcegraph/zoekt" + "tangled.org/core/repoident" ) func main() { reposPath := flag.String("repos", "REPOS", "path to repos list file (one DID per line)") - server := flag.String("server", "http://localhost:6060", "indexserver base url") + serverUrl := flag.String("server", "http://localhost:6060", "indexserver base url") plc := flag.String("plc", "https://plc.directory", "atproto PLC directory url") concurrency := flag.Int("concurrency", 5, "number of repos to process in parallel") + allowHttp := flag.Bool("allow-http", false, "accept repo DIDs whose knot service endpoint is plaintext http, and skip TLS verification when reading HEAD") flag.Parse() + server, err := url.Parse(*serverUrl) + if err != nil { + log.Fatalf("parsing -server %q: %v", *serverUrl, err) + } + if (server.Scheme != "http" && server.Scheme != "https") || server.Host == "" { + log.Fatalf("-server %q must be an http or https URL with a host", *serverUrl) + } + data, err := os.ReadFile(*reposPath) if err != nil { log.Fatalf("reading %s: %v", *reposPath, err) @@ -47,100 +59,105 @@ func main() { var wg sync.WaitGroup sem := make(chan struct{}, *concurrency) - for i, line := range strings.Split(string(data), "\n") { - did := strings.TrimSpace(line) - if did == "" { - continue + lo.ForEach(strings.Split(string(data), "\n"), func(line string, i int) { + raw := strings.TrimSpace(line) + if raw == "" { + return } wg.Add(1) sem <- struct{}{} - go func(i int, did string) { + go func() { defer wg.Done() defer func() { <-sem }() - knot, err := resolveKnot(ctx, &dir, did) + head, knot, err := prefillRepo(ctx, &dir, server, raw, *allowHttp) if err != nil { - log.Printf("line %d: %s: resolving knot: %v", i+1, did, err) + log.Printf("line %d: %s: %v", i+1, raw, err) fail.Add(1) return } - - branch, sha, err := resolveHead(knot, did) - if err != nil { - log.Printf("line %d: %s: resolving HEAD: %v", i+1, did, err) - fail.Add(1) - return - } - - if err := enqueue(*server, did, branch, sha); err != nil { - log.Printf("line %d: %s: enqueue: %v", i+1, did, err) - fail.Add(1) - return - } - log.Printf("line %d: %s: enqueued %s@%s (knot=%s)", i+1, did, branch, sha, knot) + log.Printf("line %d: %s: enqueued %s@%s (knot=%s)", i+1, raw, head.Name, head.Version, knot) ok.Add(1) - }(i, did) - } + }() + }) wg.Wait() fmt.Printf("done: %d enqueued, %d failed\n", ok.Load(), fail.Load()) } -func resolveKnot(ctx context.Context, dir identity.Directory, did string) (string, error) { - d, err := syntax.ParseDID(did) +func prefillRepo(ctx context.Context, dir identity.Directory, server *url.URL, raw string, allowHTTP bool) (zoekt.RepositoryBranch, repoident.KnotURL, error) { + var knot repoident.KnotURL + + repoDid, err := repoident.NewRepoDid(raw) if err != nil { - return "", err + return zoekt.RepositoryBranch{}, knot, err } - ident, err := dir.LookupDID(ctx, d) + + ident, err := dir.LookupDID(ctx, syntax.DID(repoDid)) if err != nil { - return "", err + return zoekt.RepositoryBranch{}, knot, fmt.Errorf("resolving repo DID: %w", err) } - knot := ident.PDSEndpoint() - if knot == "" { - return "", fmt.Errorf("no PDS endpoint in DID document") + + knot, err = repoident.KnotURLFromIdentity(ident, repoident.SchemeFor(allowHTTP)) + if err != nil { + return zoekt.RepositoryBranch{}, knot, fmt.Errorf("resolving knot: %w", err) } - return knot, nil -} -func resolveHead(knot, did string) (branch, sha string, err error) { - url := strings.TrimRight(knot, "/") + "/" + did - out, err := exec.Command( - "git", - "-c", "http.sslVerify=false", - "ls-remote", "--symref", url, "HEAD", - ).Output() + head, err := resolveHead(knot, repoDid, allowHTTP) if err != nil { - return "", "", fmt.Errorf("git ls-remote --symref %s HEAD: %w", url, err) + return head, knot, fmt.Errorf("resolving HEAD: %w", err) } - for line := range strings.SplitSeq(string(out), "\n") { - fields := strings.Fields(line) - if len(fields) < 2 { - continue - } - switch { - case fields[0] == "ref:": - branch = strings.TrimPrefix(fields[1], "refs/heads/") - case fields[1] == "HEAD": - sha = fields[0] - } + + if err := enqueue(server, repoDid, head); err != nil { + return head, knot, fmt.Errorf("enqueue: %w", err) } - if branch == "" || sha == "" { - return "", "", fmt.Errorf("could not resolve HEAD (branch=%q sha=%q)", branch, sha) + return head, knot, nil +} + +func resolveHead(knot repoident.KnotURL, repoDid repoident.RepoDid, allowHTTP bool) (zoekt.RepositoryBranch, error) { + remote := knot.JoinPath(repoDid.String()) + args := append( + lo.Ternary(allowHTTP, []string{"-c", "http.sslVerify=false"}, nil), + "ls-remote", "--symref", remote, "HEAD", + ) + out, err := exec.Command("git", args...).Output() + if err != nil { + return zoekt.RepositoryBranch{}, fmt.Errorf("git ls-remote --symref %s HEAD: %w", remote, err) + } + head := lo.Reduce( + strings.Split(string(out), "\n"), + func(head zoekt.RepositoryBranch, line string, _ int) zoekt.RepositoryBranch { + fields := strings.Fields(line) + if len(fields) < 2 { + return head + } + switch { + case fields[0] == "ref:": + head.Name = strings.TrimPrefix(fields[1], "refs/heads/") + case fields[1] == "HEAD": + head.Version = fields[0] + } + return head + }, + zoekt.RepositoryBranch{}, + ) + if head.Name == "" || head.Version == "" { + return zoekt.RepositoryBranch{}, fmt.Errorf("couldn't resolve HEAD (branch=%q sha=%q)", head.Name, head.Version) } - return branch, sha, nil + return head, nil } -func enqueue(server, did, branch, sha string) error { +func enqueue(server *url.URL, repoDid repoident.RepoDid, head zoekt.RepositoryBranch) error { body, err := json.Marshal(map[string]any{ - "repo": did, - "branches": []zoekt.RepositoryBranch{{Name: branch, Version: sha}}, + "repo": repoDid.String(), + "branches": []zoekt.RepositoryBranch{head}, }) if err != nil { return err } - resp, err := http.Post(strings.TrimRight(server, "/")+"/admin/enqueueIndex", + resp, err := http.Post(server.JoinPath("admin", "enqueueIndex").String(), "application/json", bytes.NewReader(body)) if err != nil { return err diff --git a/cmd/zoekt-tngl-indexserver/index.go b/cmd/zoekt-tngl-indexserver/index.go index e139612d..da04667e 100644 --- a/cmd/zoekt-tngl-indexserver/index.go +++ b/cmd/zoekt-tngl-indexserver/index.go @@ -12,9 +12,9 @@ import ( "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" - indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/sourcegraph/zoekt" - "tangled.org/core/api/tangled" + "tangled.org/core/repoident" + "tangled.org/core/repoverify" ) // 1 MB; match https://sourcegraph.sourcegraph.com/r/github.com/sourcegraph/sourcegraph/-/blob/cmd/searcher/internal/search/store.go?L32 @@ -24,9 +24,9 @@ func gitIndex(ctx context.Context, cfg *Config, dir identity.Directory, req inde ctx, cancel := context.WithTimeout(ctx, cfg.IndexTimeout) defer cancel() - repo, err := loadRepo(ctx, dir, req.Repo) + repo, err := loadRepo(ctx, cfg, dir, req.Repo) if err != nil { - return nil + return fmt.Errorf("loading repo %s: %w", req.Repo, err) } repo.Branches = req.Branches @@ -47,24 +47,26 @@ func gitIndex(ctx context.Context, cfg *Config, dir identity.Directory, req inde return nil } -func loadRepo(ctx context.Context, dir identity.Directory, repoDID syntax.DID) (*Repo, error) { - ident, err := dir.LookupDID(ctx, repoDID) +func loadRepo(ctx context.Context, cfg *Config, dir identity.Directory, repoDID repoident.RepoDid) (*Repo, error) { + ident, err := dir.LookupDID(ctx, syntax.DID(repoDID)) if err != nil { return nil, err } - knot := ident.PDSEndpoint() + knot, err := repoident.KnotURLFromIdentity(ident, cfg.KnotScheme) + if err != nil { + return nil, fmt.Errorf("repoDid %s: %w", repoDID, err) + } - xrpcc := &indigoxrpc.Client{Host: knot} - out, err := tangled.RepoDescribeRepo(ctx, xrpcc, repoDID.String()) + described, err := repoverify.Describe(ctx, nil, knot, repoDID) if err != nil { return nil, err } return &Repo{ Did: repoDID, - Owner: syntax.DID(out.OwnerDid), - Slug: syntax.RecordKey(out.Rkey), + Owner: described.OwnerDid, + Slug: described.Rkey, Knot: knot, }, nil } diff --git a/cmd/zoekt-tngl-indexserver/main.go b/cmd/zoekt-tngl-indexserver/main.go index 4afc5f80..7ae942ce 100644 --- a/cmd/zoekt-tngl-indexserver/main.go +++ b/cmd/zoekt-tngl-indexserver/main.go @@ -20,10 +20,12 @@ import ( "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/carlmjohnson/versioninfo" + "github.com/samber/lo" "github.com/sourcegraph/zoekt" "github.com/sourcegraph/zoekt/gitindex" "github.com/sourcegraph/zoekt/index" "github.com/urfave/cli/v3" + "tangled.org/core/repoident" ) func loggedRun(cmd *exec.Cmd) error { @@ -116,6 +118,11 @@ func run(args []string) error { Value: ":6060", Sources: cli.EnvVars("TANGLED_ZOEKT_SERVER_LISTEN"), }, + &cli.BoolFlag{ + Name: "allow-http", + Usage: "accept repo DIDs whose knot service endpoint is plaintext http.", + Sources: cli.EnvVars("TANGLED_ZOEKT_ALLOW_HTTP"), + }, }, }, { @@ -151,6 +158,8 @@ type Config struct { PlcUrl string AppviewUrl string Listen string + + KnotScheme repoident.SchemePolicy } func createMissingDirectories(cfg *Config) { @@ -162,15 +171,15 @@ func createMissingDirectories(cfg *Config) { } type Repo struct { - Did syntax.DID // repo DID - Owner syntax.DID + Did repoident.RepoDid + Owner repoident.OwnerDid Slug syntax.RecordKey - Knot string // knot service url derived from #atproto_pds service endpoint + Knot repoident.KnotURL Branches []zoekt.RepositoryBranch } func (r *Repo) CloneURL() string { - return r.Knot + "/" + r.Did.String() + return r.Knot.JoinPath(r.Did.String()) } func runIndexServer(ctx context.Context, cmd *cli.Command) error { @@ -182,6 +191,7 @@ func runIndexServer(ctx context.Context, cmd *cli.Command) error { PlcUrl: cmd.String("plc-url"), AppviewUrl: cmd.String("appview-url"), Listen: cmd.String("listen"), + KnotScheme: repoident.SchemeFor(cmd.Bool("allow-http")), } createMissingDirectories(cfg) @@ -213,12 +223,12 @@ func runIndex(ctx context.Context, cmd *cli.Command) error { if err := json.Unmarshal([]byte(repoRaw), &repo); err != nil { return fmt.Errorf("invalid repo: %w", err) } - - var branches []string - for _, b := range repo.Branches { - branches = append(branches, b.Name) + if repo.Did == "" || repo.Owner == "" || repo.Knot.IsZero() { + return fmt.Errorf("repo is missing did, owner, or knot: %q", repoRaw) } + branches := lo.Map(repo.Branches, func(b zoekt.RepositoryBranch, _ int) string { return b.Name }) + buildOpts := index.Options{} buildOpts.SetDefaults() @@ -238,7 +248,7 @@ func runIndex(ctx context.Context, cmd *cli.Command) error { "foo": "bar", // for testing "did": repo.Did.String(), "owner": repo.Owner.String(), - "knot": repo.Knot, + "knot": repo.Knot.String(), } // buildOpts.RepositoryDescription.Source = gitDir // configured later in IndexGitRepo buildOpts.RepositoryDescription.Branches = nil diff --git a/cmd/zoekt-tngl-indexserver/queue.go b/cmd/zoekt-tngl-indexserver/queue.go index f4201170..1ab3ea33 100644 --- a/cmd/zoekt-tngl-indexserver/queue.go +++ b/cmd/zoekt-tngl-indexserver/queue.go @@ -3,20 +3,20 @@ package main import ( "sync" - "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/repoident" ) // deduplicating index work queue type Queue struct { mu sync.Mutex - order []syntax.DID - pending map[syntax.DID]indexRequest + order []repoident.RepoDid + pending map[repoident.RepoDid]indexRequest size int } func NewQueue(size int) *Queue { return &Queue{ - pending: make(map[syntax.DID]indexRequest), + pending: make(map[repoident.RepoDid]indexRequest), size: size, } } diff --git a/cmd/zoekt-tngl-indexserver/server.go b/cmd/zoekt-tngl-indexserver/server.go index 76ed614d..de983555 100644 --- a/cmd/zoekt-tngl-indexserver/server.go +++ b/cmd/zoekt-tngl-indexserver/server.go @@ -10,10 +10,10 @@ import ( "time" "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/sourcegraph/zoekt" + "tangled.org/core/repoident" ) type IndexServer struct { @@ -59,7 +59,7 @@ func (s *IndexServer) handleMetrics(w http.ResponseWriter, r *http.Request) { } type indexRequest struct { - Repo syntax.DID `json:"repo"` + Repo repoident.RepoDid `json:"repo"` Branches []zoekt.RepositoryBranch `json:"branches"` } diff --git a/docker-compose.yml b/docker-compose.yml index 79766ef9..c5ab813a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -301,6 +301,7 @@ services: TANGLED_ZOEKT_INDEX_DIR: /data/index TANGLED_ZOEKT_PLC_URL: https://plc.tngl.boltless.dev TANGLED_ZOEKT_APPVIEW_URL: http://127.0.0.1:3000 + TANGLED_ZOEKT_ALLOW_HTTP: "true" volumes: - zoekt-index:/data/index - ./localinfra/certs/root.crt:/etc/ssl/certs/caddy.crt:ro -- 2.51.2