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