alice.pds.demo.boltless.dev
This repository has no description
cmd/prefill-zoekt,zoekt-tngl-indexserver: type repoDIDs & knot URLs via repoident
Lewis: May this revision serve well! <did:plc:3fwecdnvtcscjnrx2p4n7alz>
author
did:plc:3fwecdnvtcscjnrx2p4n7a…
committer
Tangled
date
(Jul 30, 2026, 3:00 PM UTC)
commit
9f81b7fa
9f81b7fa6889326f8ea5724a2244266006b261b5
parent
d78be590
d78be590071fbdbea4719a940a9cdfcf37fef35c
change-id
mtukulyp
mtukulypkozwtqxqsspmlzulmrsomwyp
did:plc:3fwecdnvtcscjnrx2p4n7a…
committer
+118
-86
+2
.gitignore
+2
.gitignore
+77
-60
cmd/prefill-zoekt/main.go
+77
-60
cmd/prefill-zoekt/main.go
···
17
17
"fmt"
18
18
"log"
19
19
"net/http"
20
+
"net/url"
20
21
"os"
21
22
"os/exec"
22
23
"strings"
···
25
26
26
27
"github.com/bluesky-social/indigo/atproto/identity"
27
28
"github.com/bluesky-social/indigo/atproto/syntax"
29
+
"github.com/samber/lo"
28
30
"github.com/sourcegraph/zoekt"
31
+
"tangled.org/core/repoident"
29
32
)
30
33
31
34
func main() {
32
35
reposPath := flag.String("repos", "REPOS", "path to repos list file (one DID per line)")
33
-
server := flag.String("server", "http://localhost:6060", "indexserver base url")
36
+
serverUrl := flag.String("server", "http://localhost:6060", "indexserver base url")
34
37
plc := flag.String("plc", "https://plc.directory", "atproto PLC directory url")
35
38
concurrency := flag.Int("concurrency", 5, "number of repos to process in parallel")
39
+
allowHttp := flag.Bool("allow-http", false, "accept repo DIDs whose knot service endpoint is plaintext http, and skip TLS verification when reading HEAD")
36
40
flag.Parse()
37
41
42
+
server, err := url.Parse(*serverUrl)
43
+
if err != nil {
44
+
log.Fatalf("parsing -server %q: %v", *serverUrl, err)
45
+
}
46
+
if (server.Scheme != "http" && server.Scheme != "https") || server.Host == "" {
47
+
log.Fatalf("-server %q must be an http or https URL with a host", *serverUrl)
48
+
}
49
+
38
50
data, err := os.ReadFile(*reposPath)
39
51
if err != nil {
40
52
log.Fatalf("reading %s: %v", *reposPath, err)
···
47
59
var wg sync.WaitGroup
48
60
sem := make(chan struct{}, *concurrency)
49
61
50
-
for i, line := range strings.Split(string(data), "\n") {
51
-
did := strings.TrimSpace(line)
52
-
if did == "" {
53
-
continue
62
+
lo.ForEach(strings.Split(string(data), "\n"), func(line string, i int) {
63
+
raw := strings.TrimSpace(line)
64
+
if raw == "" {
65
+
return
54
66
}
55
67
56
68
wg.Add(1)
57
69
sem <- struct{}{}
58
-
go func(i int, did string) {
70
+
go func() {
59
71
defer wg.Done()
60
72
defer func() { <-sem }()
61
73
62
-
knot, err := resolveKnot(ctx, &dir, did)
74
+
head, knot, err := prefillRepo(ctx, &dir, server, raw, *allowHttp)
63
75
if err != nil {
64
-
log.Printf("line %d: %s: resolving knot: %v", i+1, did, err)
76
+
log.Printf("line %d: %s: %v", i+1, raw, err)
65
77
fail.Add(1)
66
78
return
67
79
}
68
-
69
-
branch, sha, err := resolveHead(knot, did)
70
-
if err != nil {
71
-
log.Printf("line %d: %s: resolving HEAD: %v", i+1, did, err)
72
-
fail.Add(1)
73
-
return
74
-
}
75
-
76
-
if err := enqueue(*server, did, branch, sha); err != nil {
77
-
log.Printf("line %d: %s: enqueue: %v", i+1, did, err)
78
-
fail.Add(1)
79
-
return
80
-
}
81
-
log.Printf("line %d: %s: enqueued %s@%s (knot=%s)", i+1, did, branch, sha, knot)
80
+
log.Printf("line %d: %s: enqueued %s@%s (knot=%s)", i+1, raw, head.Name, head.Version, knot)
82
81
ok.Add(1)
83
-
}(i, did)
84
-
}
82
+
}()
83
+
})
85
84
86
85
wg.Wait()
87
86
fmt.Printf("done: %d enqueued, %d failed\n", ok.Load(), fail.Load())
88
87
}
89
88
90
-
func resolveKnot(ctx context.Context, dir identity.Directory, did string) (string, error) {
91
-
d, err := syntax.ParseDID(did)
89
+
func prefillRepo(ctx context.Context, dir identity.Directory, server *url.URL, raw string, allowHTTP bool) (zoekt.RepositoryBranch, repoident.KnotURL, error) {
90
+
var knot repoident.KnotURL
91
+
92
+
repoDid, err := repoident.NewRepoDid(raw)
92
93
if err != nil {
93
-
return "", err
94
+
return zoekt.RepositoryBranch{}, knot, err
95
+
}
96
+
97
+
ident, err := dir.LookupDID(ctx, syntax.DID(repoDid))
98
+
if err != nil {
99
+
return zoekt.RepositoryBranch{}, knot, fmt.Errorf("resolving repo DID: %w", err)
100
+
}
101
+
102
+
knot, err = repoident.KnotURLFromIdentity(ident, repoident.SchemeFor(allowHTTP))
103
+
if err != nil {
104
+
return zoekt.RepositoryBranch{}, knot, fmt.Errorf("resolving knot: %w", err)
94
105
}
95
-
ident, err := dir.LookupDID(ctx, d)
106
+
107
+
head, err := resolveHead(knot, repoDid, allowHTTP)
96
108
if err != nil {
97
-
return "", err
109
+
return head, knot, fmt.Errorf("resolving HEAD: %w", err)
98
110
}
99
-
knot := ident.PDSEndpoint()
100
-
if knot == "" {
101
-
return "", fmt.Errorf("no PDS endpoint in DID document")
111
+
112
+
if err := enqueue(server, repoDid, head); err != nil {
113
+
return head, knot, fmt.Errorf("enqueue: %w", err)
102
114
}
103
-
return knot, nil
115
+
return head, knot, nil
104
116
}
105
117
106
-
func resolveHead(knot, did string) (branch, sha string, err error) {
107
-
url := strings.TrimRight(knot, "/") + "/" + did
108
-
out, err := exec.Command(
109
-
"git",
110
-
"-c", "http.sslVerify=false",
111
-
"ls-remote", "--symref", url, "HEAD",
112
-
).Output()
118
+
func resolveHead(knot repoident.KnotURL, repoDid repoident.RepoDid, allowHTTP bool) (zoekt.RepositoryBranch, error) {
119
+
remote := knot.JoinPath(repoDid.String())
120
+
args := append(
121
+
lo.Ternary(allowHTTP, []string{"-c", "http.sslVerify=false"}, nil),
122
+
"ls-remote", "--symref", remote, "HEAD",
123
+
)
124
+
out, err := exec.Command("git", args...).Output()
113
125
if err != nil {
114
-
return "", "", fmt.Errorf("git ls-remote --symref %s HEAD: %w", url, err)
115
-
}
116
-
for line := range strings.SplitSeq(string(out), "\n") {
117
-
fields := strings.Fields(line)
118
-
if len(fields) < 2 {
119
-
continue
120
-
}
121
-
switch {
122
-
case fields[0] == "ref:":
123
-
branch = strings.TrimPrefix(fields[1], "refs/heads/")
124
-
case fields[1] == "HEAD":
125
-
sha = fields[0]
126
-
}
126
+
return zoekt.RepositoryBranch{}, fmt.Errorf("git ls-remote --symref %s HEAD: %w", remote, err)
127
127
}
128
-
if branch == "" || sha == "" {
129
-
return "", "", fmt.Errorf("could not resolve HEAD (branch=%q sha=%q)", branch, sha)
128
+
head := lo.Reduce(
129
+
strings.Split(string(out), "\n"),
130
+
func(head zoekt.RepositoryBranch, line string, _ int) zoekt.RepositoryBranch {
131
+
fields := strings.Fields(line)
132
+
if len(fields) < 2 {
133
+
return head
134
+
}
135
+
switch {
136
+
case fields[0] == "ref:":
137
+
head.Name = strings.TrimPrefix(fields[1], "refs/heads/")
138
+
case fields[1] == "HEAD":
139
+
head.Version = fields[0]
140
+
}
141
+
return head
142
+
},
143
+
zoekt.RepositoryBranch{},
144
+
)
145
+
if head.Name == "" || head.Version == "" {
146
+
return zoekt.RepositoryBranch{}, fmt.Errorf("couldn't resolve HEAD (branch=%q sha=%q)", head.Name, head.Version)
130
147
}
131
-
return branch, sha, nil
148
+
return head, nil
132
149
}
133
150
134
-
func enqueue(server, did, branch, sha string) error {
151
+
func enqueue(server *url.URL, repoDid repoident.RepoDid, head zoekt.RepositoryBranch) error {
135
152
body, err := json.Marshal(map[string]any{
136
-
"repo": did,
137
-
"branches": []zoekt.RepositoryBranch{{Name: branch, Version: sha}},
153
+
"repo": repoDid.String(),
154
+
"branches": []zoekt.RepositoryBranch{head},
138
155
})
139
156
if err != nil {
140
157
return err
141
158
}
142
159
143
-
resp, err := http.Post(strings.TrimRight(server, "/")+"/admin/enqueueIndex",
160
+
resp, err := http.Post(server.JoinPath("admin", "enqueueIndex").String(),
144
161
"application/json", bytes.NewReader(body))
145
162
if err != nil {
146
163
return err
+13
-11
cmd/zoekt-tngl-indexserver/index.go
+13
-11
cmd/zoekt-tngl-indexserver/index.go
···
12
12
13
13
"github.com/bluesky-social/indigo/atproto/identity"
14
14
"github.com/bluesky-social/indigo/atproto/syntax"
15
-
indigoxrpc "github.com/bluesky-social/indigo/xrpc"
16
15
"github.com/sourcegraph/zoekt"
17
-
"tangled.org/core/api/tangled"
16
+
"tangled.org/core/repoident"
17
+
"tangled.org/core/repoverify"
18
18
)
19
19
20
20
// 1 MB; match https://sourcegraph.sourcegraph.com/r/github.com/sourcegraph/sourcegraph/-/blob/cmd/searcher/internal/search/store.go?L32
···
24
24
ctx, cancel := context.WithTimeout(ctx, cfg.IndexTimeout)
25
25
defer cancel()
26
26
27
-
repo, err := loadRepo(ctx, dir, req.Repo)
27
+
repo, err := loadRepo(ctx, cfg, dir, req.Repo)
28
28
if err != nil {
29
-
return nil
29
+
return fmt.Errorf("loading repo %s: %w", req.Repo, err)
30
30
}
31
31
repo.Branches = req.Branches
32
32
···
47
47
return nil
48
48
}
49
49
50
-
func loadRepo(ctx context.Context, dir identity.Directory, repoDID syntax.DID) (*Repo, error) {
51
-
ident, err := dir.LookupDID(ctx, repoDID)
50
+
func loadRepo(ctx context.Context, cfg *Config, dir identity.Directory, repoDID repoident.RepoDid) (*Repo, error) {
51
+
ident, err := dir.LookupDID(ctx, syntax.DID(repoDID))
52
52
if err != nil {
53
53
return nil, err
54
54
}
55
55
56
-
knot := ident.PDSEndpoint()
56
+
knot, err := repoident.KnotURLFromIdentity(ident, cfg.KnotScheme)
57
+
if err != nil {
58
+
return nil, fmt.Errorf("repoDid %s: %w", repoDID, err)
59
+
}
57
60
58
-
xrpcc := &indigoxrpc.Client{Host: knot}
59
-
out, err := tangled.RepoDescribeRepo(ctx, xrpcc, repoDID.String())
61
+
described, err := repoverify.Describe(ctx, nil, knot, repoDID)
60
62
if err != nil {
61
63
return nil, err
62
64
}
63
65
64
66
return &Repo{
65
67
Did: repoDID,
66
-
Owner: syntax.DID(out.OwnerDid),
67
-
Slug: syntax.RecordKey(out.Rkey),
68
+
Owner: described.OwnerDid,
69
+
Slug: described.Rkey,
68
70
Knot: knot,
69
71
}, nil
70
72
}
+19
-9
cmd/zoekt-tngl-indexserver/main.go
+19
-9
cmd/zoekt-tngl-indexserver/main.go
···
20
20
"github.com/bluesky-social/indigo/atproto/identity"
21
21
"github.com/bluesky-social/indigo/atproto/syntax"
22
22
"github.com/carlmjohnson/versioninfo"
23
+
"github.com/samber/lo"
23
24
"github.com/sourcegraph/zoekt"
24
25
"github.com/sourcegraph/zoekt/gitindex"
25
26
"github.com/sourcegraph/zoekt/index"
26
27
"github.com/urfave/cli/v3"
28
+
"tangled.org/core/repoident"
27
29
)
28
30
29
31
func loggedRun(cmd *exec.Cmd) error {
···
116
118
Value: ":6060",
117
119
Sources: cli.EnvVars("TANGLED_ZOEKT_SERVER_LISTEN"),
118
120
},
121
+
&cli.BoolFlag{
122
+
Name: "allow-http",
123
+
Usage: "accept repo DIDs whose knot service endpoint is plaintext http.",
124
+
Sources: cli.EnvVars("TANGLED_ZOEKT_ALLOW_HTTP"),
125
+
},
119
126
},
120
127
},
121
128
{
···
151
158
PlcUrl string
152
159
AppviewUrl string
153
160
Listen string
161
+
162
+
KnotScheme repoident.SchemePolicy
154
163
}
155
164
156
165
func createMissingDirectories(cfg *Config) {
···
162
171
}
163
172
164
173
type Repo struct {
165
-
Did syntax.DID // repo DID
166
-
Owner syntax.DID
174
+
Did repoident.RepoDid
175
+
Owner repoident.OwnerDid
167
176
Slug syntax.RecordKey
168
-
Knot string // knot service url derived from #atproto_pds service endpoint
177
+
Knot repoident.KnotURL
169
178
Branches []zoekt.RepositoryBranch
170
179
}
171
180
172
181
func (r *Repo) CloneURL() string {
173
-
return r.Knot + "/" + r.Did.String()
182
+
return r.Knot.JoinPath(r.Did.String())
174
183
}
175
184
176
185
func runIndexServer(ctx context.Context, cmd *cli.Command) error {
···
182
191
PlcUrl: cmd.String("plc-url"),
183
192
AppviewUrl: cmd.String("appview-url"),
184
193
Listen: cmd.String("listen"),
194
+
KnotScheme: repoident.SchemeFor(cmd.Bool("allow-http")),
185
195
}
186
196
createMissingDirectories(cfg)
187
197
···
213
223
if err := json.Unmarshal([]byte(repoRaw), &repo); err != nil {
214
224
return fmt.Errorf("invalid repo: %w", err)
215
225
}
216
-
217
-
var branches []string
218
-
for _, b := range repo.Branches {
219
-
branches = append(branches, b.Name)
226
+
if repo.Did == "" || repo.Owner == "" || repo.Knot.IsZero() {
227
+
return fmt.Errorf("repo is missing did, owner, or knot: %q", repoRaw)
220
228
}
229
+
230
+
branches := lo.Map(repo.Branches, func(b zoekt.RepositoryBranch, _ int) string { return b.Name })
221
231
222
232
buildOpts := index.Options{}
223
233
buildOpts.SetDefaults()
···
238
248
"foo": "bar", // for testing
239
249
"did": repo.Did.String(),
240
250
"owner": repo.Owner.String(),
241
-
"knot": repo.Knot,
251
+
"knot": repo.Knot.String(),
242
252
}
243
253
// buildOpts.RepositoryDescription.Source = gitDir // configured later in IndexGitRepo
244
254
buildOpts.RepositoryDescription.Branches = nil
+4
-4
cmd/zoekt-tngl-indexserver/queue.go
+4
-4
cmd/zoekt-tngl-indexserver/queue.go
···
3
3
import (
4
4
"sync"
5
5
6
-
"github.com/bluesky-social/indigo/atproto/syntax"
6
+
"tangled.org/core/repoident"
7
7
)
8
8
9
9
// deduplicating index work queue
10
10
type Queue struct {
11
11
mu sync.Mutex
12
-
order []syntax.DID
13
-
pending map[syntax.DID]indexRequest
12
+
order []repoident.RepoDid
13
+
pending map[repoident.RepoDid]indexRequest
14
14
size int
15
15
}
16
16
17
17
func NewQueue(size int) *Queue {
18
18
return &Queue{
19
-
pending: make(map[syntax.DID]indexRequest),
19
+
pending: make(map[repoident.RepoDid]indexRequest),
20
20
size: size,
21
21
}
22
22
}
+2
-2
cmd/zoekt-tngl-indexserver/server.go
+2
-2
cmd/zoekt-tngl-indexserver/server.go
···
10
10
"time"
11
11
12
12
"github.com/bluesky-social/indigo/atproto/identity"
13
-
"github.com/bluesky-social/indigo/atproto/syntax"
14
13
"github.com/prometheus/client_golang/prometheus"
15
14
"github.com/prometheus/client_golang/prometheus/promhttp"
16
15
"github.com/sourcegraph/zoekt"
16
+
"tangled.org/core/repoident"
17
17
)
18
18
19
19
type IndexServer struct {
···
59
59
}
60
60
61
61
type indexRequest struct {
62
-
Repo syntax.DID `json:"repo"`
62
+
Repo repoident.RepoDid `json:"repo"`
63
63
Branches []zoekt.RepositoryBranch `json:"branches"`
64
64
}
65
65
+1
docker-compose.yml
+1
docker-compose.yml
···
301
301
TANGLED_ZOEKT_INDEX_DIR: /data/index
302
302
TANGLED_ZOEKT_PLC_URL: https://plc.tngl.boltless.dev
303
303
TANGLED_ZOEKT_APPVIEW_URL: http://127.0.0.1:3000
304
+
TANGLED_ZOEKT_ALLOW_HTTP: "true"
304
305
volumes:
305
306
- zoekt-index:/data/index
306
307
- ./localinfra/certs/root.crt:/etc/ssl/certs/caddy.crt:ro