alice.pds.demo.boltless.dev
This repository has no description
8.9 kB
316 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1// similar to zoekt-dynamic-indexserver, but targetting Tangled repos
2package main
3
4import (
5 "bytes"
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "log"
11 "net"
12 "net/http"
13 "os"
14 "os/exec"
15 "os/signal"
16 "strconv"
17 "syscall"
18 "time"
19
20 "github.com/bluesky-social/indigo/atproto/identity"
21 "github.com/bluesky-social/indigo/atproto/syntax"
22 "github.com/carlmjohnson/versioninfo"
23 "github.com/samber/lo"
24 "github.com/sourcegraph/zoekt"
25 "github.com/sourcegraph/zoekt/gitindex"
26 "github.com/sourcegraph/zoekt/index"
27 "github.com/urfave/cli/v3"
28 "tangled.org/core/repoident"
29)
30
31func loggedRun(cmd *exec.Cmd) error {
32 outBuf := &bytes.Buffer{}
33 errBuf := &bytes.Buffer{}
34 cmd.Stdout = outBuf
35 cmd.Stderr = errBuf
36
37 log.Printf("run %v", cmd.Args)
38 if err := cmd.Run(); err != nil {
39 log.Printf("command %s failed: %v\nOUT: %s\nERR: %s",
40 cmd.Args, err, outBuf.String(), errBuf.String())
41 return fmt.Errorf("command %s failed: %v", cmd.Args, err)
42 }
43
44 return nil
45}
46
47// This function is declared as var so that we can stub it in test
48var executeCmd = func(ctx context.Context, name string, arg ...string) error {
49 cmd := exec.CommandContext(ctx, name, arg...)
50 cmd.Stdin = &bytes.Buffer{}
51 err := loggedRun(cmd)
52
53 return err
54}
55
56func main() {
57 if err := run(os.Args); err != nil {
58 log.Fatal(err)
59 }
60}
61
62func run(args []string) error {
63 ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
64 defer cancel()
65
66 app := cli.Command{
67 Name: "zoekt-tngl-indexserver",
68 Usage: "tangled zoekt index server",
69 Version: versioninfo.Short(),
70 }
71 app.Flags = []cli.Flag{
72 &cli.StringFlag{
73 Name: "index-dir",
74 Usage: "directory holding index shards.",
75 Required: true,
76 Sources: cli.EnvVars("TANGLED_ZOEKT_INDEX_DIR"),
77 },
78 &cli.StringFlag{
79 Name: "appview-url",
80 Usage: "appview url. used when debugging",
81 Value: "https://tangled.org",
82 Sources: cli.EnvVars("TANGLED_ZOEKT_APPVIEW_URL"),
83 },
84 }
85 app.Commands = []*cli.Command{
86 {
87 Name: "serve",
88 Usage: "run the index server daemon",
89 Action: runIndexServer,
90 Flags: []cli.Flag{
91 &cli.StringFlag{
92 Name: "plc-url",
93 Usage: "atproto PLC directory.",
94 Value: "https://plc.directory",
95 Sources: cli.EnvVars("TANGLED_ZOEKT_PLC_URL", "TANGLED_PLC_URL", "ATP_PLC_HOST"),
96 },
97 &cli.DurationFlag{
98 Name: "index-timeout",
99 Usage: "kill index job after this much time.",
100 Value: time.Hour,
101 Sources: cli.EnvVars("TANGLED_ZOEKT_INDEX_TIMEOUT"),
102 },
103 &cli.IntFlag{
104 Name: "index-concurrency",
105 Usage: "number of repositories to index concurrently.",
106 Value: 4,
107 Sources: cli.EnvVars("TANGLED_ZOEKT_INDEX_CONCURRENCY"),
108 },
109 &cli.IntFlag{
110 Name: "index-queue-size",
111 Usage: "max index queue size",
112 Value: 100,
113 Sources: cli.EnvVars("TANGLED_ZOEKT_INDEX_QUEUE_SIZE"),
114 },
115 &cli.StringFlag{
116 Name: "listen",
117 Usage: "listen on this address",
118 Value: ":6060",
119 Sources: cli.EnvVars("TANGLED_ZOEKT_SERVER_LISTEN"),
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 },
126 },
127 },
128 {
129 Name: "index",
130 Usage: "manually index a git repository",
131 Action: runIndex,
132 Arguments: []cli.Argument{
133 &cli.StringArg{
134 Name: "git-dir",
135 UsageText: "path to fetched git repository.",
136 },
137 &cli.StringArg{
138 Name: "repo",
139 UsageText: "json-encoded repository information",
140 },
141 },
142 },
143 }
144
145 return app.Run(ctx, args)
146}
147
148type Config struct {
149 // IndexDir is the index directory to use.
150 IndexDir string
151
152 IndexTimeout time.Duration
153
154 // IndexConcurrency is the number of repositories we index at once.
155 IndexConcurrency int
156 IndexQueueSize int
157
158 PlcUrl string
159 AppviewUrl string
160 Listen string
161
162 KnotScheme repoident.SchemePolicy
163}
164
165func createMissingDirectories(cfg *Config) {
166 for _, s := range []string{cfg.IndexDir} {
167 if err := os.MkdirAll(s, 0o755); err != nil {
168 log.Fatalf("MkdirAll %s: %v", s, err)
169 }
170 }
171}
172
173type Repo struct {
174 Did repoident.RepoDid
175 Owner repoident.OwnerDid
176 Slug syntax.RecordKey
177 Knot repoident.KnotURL
178 Branches []zoekt.RepositoryBranch
179}
180
181func (r *Repo) CloneURL() string {
182 return r.Knot.JoinPath(r.Did.String())
183}
184
185func runIndexServer(ctx context.Context, cmd *cli.Command) error {
186 cfg := &Config{
187 IndexDir: cmd.String("index-dir"),
188 IndexTimeout: cmd.Duration("index-timeout"),
189 IndexConcurrency: cmd.Int("index-concurrency"),
190 IndexQueueSize: cmd.Int("index-queue-size"),
191 PlcUrl: cmd.String("plc-url"),
192 AppviewUrl: cmd.String("appview-url"),
193 Listen: cmd.String("listen"),
194 KnotScheme: repoident.SchemeFor(cmd.Bool("allow-http")),
195 }
196 createMissingDirectories(cfg)
197
198 server := NewIndexServer(cfg)
199 go server.Run(ctx)
200
201 <-ctx.Done()
202 return ctx.Err()
203}
204
205// sub-process to index a fetched repository
206func runIndex(ctx context.Context, cmd *cli.Command) error {
207 var (
208 indexDir = cmd.String("index-dir")
209 appviewUrl = cmd.String("appview-url")
210 )
211
212 gitDir := cmd.StringArg("git-dir")
213 if gitDir == "" {
214 return errors.New("git-dir is required.")
215 }
216
217 repoRaw := cmd.StringArg("repo")
218 if repoRaw == "" {
219 return errors.New("repo is required.")
220 }
221
222 var repo Repo
223 if err := json.Unmarshal([]byte(repoRaw), &repo); err != nil {
224 return fmt.Errorf("invalid repo: %w", err)
225 }
226 if repo.Did == "" || repo.Owner == "" || repo.Knot.IsZero() {
227 return fmt.Errorf("repo is missing did, owner, or knot: %q", repoRaw)
228 }
229
230 branches := lo.Map(repo.Branches, func(b zoekt.RepositoryBranch, _ int) string { return b.Name })
231
232 buildOpts := index.Options{}
233 buildOpts.SetDefaults()
234
235 buildOpts.IndexDir = indexDir
236
237 buildOpts.ShardPrefixOverride = repo.Did.String()
238
239 // Tangled templates
240 webUrl := fmt.Sprintf("%s/%s", appviewUrl, repo.Did)
241 buildOpts.RepositoryDescription.CommitURLTemplate = fmt.Sprintf("%s/commit/{{.Version}}", webUrl)
242 buildOpts.RepositoryDescription.FileURLTemplate = fmt.Sprintf("%s/blob/{{.Version}}/{{.Path}}", webUrl)
243 buildOpts.RepositoryDescription.LineFragmentTemplate = "#L{{.LineNumber}}"
244
245 buildOpts.RepositoryDescription.Name = repo.Slug.String()
246 buildOpts.RepositoryDescription.URL = webUrl
247 buildOpts.RepositoryDescription.Metadata = map[string]string{
248 "foo": "bar", // for testing
249 "did": repo.Did.String(),
250 "owner": repo.Owner.String(),
251 "knot": repo.Knot.String(),
252 }
253 // buildOpts.RepositoryDescription.Source = gitDir // configured later in IndexGitRepo
254 buildOpts.RepositoryDescription.Branches = nil
255 buildOpts.RepositoryDescription.SubRepoMap = nil
256 buildOpts.RepositoryDescription.RawConfig = map[string]string{
257 "priority": strconv.FormatFloat(0.0, 'g', -1, 64),
258 "public": marshalBool(true),
259 "fork": marshalBool(false),
260 // Calculate repo rank based on the latest commit date.
261 "latestCommitDate": marshalBool(true),
262 }
263 buildOpts.RepositoryDescription.Rank = 0
264 // buildOpts.RepositoryDescription.IndexOptions = "" // configured later in IndexGitRepo
265 // buildOpts.RepositoryDescription.LatestCommitDate = _ // configured later in IndexGitRepo
266 buildOpts.RepositoryDescription.FileTombstones = nil
267
268 gitOpts := gitindex.Options{
269 RepoDir: gitDir,
270 Submodules: false,
271 Incremental: true,
272 AllowMissingBranch: false,
273 RepoCacheDir: "",
274 BuildOptions: buildOpts,
275 BranchPrefix: "refs/heads/",
276 Branches: branches,
277 DeltaShardNumberFallbackThreshold: 0,
278 }
279 if _, err := gitindex.IndexGitRepo(gitOpts); err != nil {
280 return err
281 }
282 return nil
283}
284
285func marshalBool(b bool) string {
286 if b {
287 return "1"
288 }
289 return "0"
290}
291
292func baseDir(plc string) identity.Directory {
293 base := identity.BaseDirectory{
294 PLCURL: plc,
295 HTTPClient: http.Client{
296 Timeout: time.Second * 10,
297 Transport: &http.Transport{
298 Proxy: http.ProxyFromEnvironment,
299 // would want this around 100ms for services doing lots of handle resolution. Impacts PLC connections as well, but not too bad.
300 IdleConnTimeout: time.Millisecond * 1000,
301 MaxIdleConns: 100,
302 },
303 },
304 Resolver: net.Resolver{
305 Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
306 d := net.Dialer{Timeout: time.Second * 3}
307 return d.DialContext(ctx, network, address)
308 },
309 },
310 TryAuthoritativeDNS: true,
311 // primary Bluesky PDS instance only supports HTTP resolution method
312 SkipDNSDomainSuffixes: []string{".bsky.social"},
313 UserAgent: "indigo-identity/" + versioninfo.Short(),
314 }
315 return identity.NewCacheDirectory(&base, 250_000, time.Hour*24, time.Minute*2, time.Minute*5)
316}