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