alice.pds.demo.boltless.dev
This repository has no description
5.5 kB
212 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1package main
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "log"
9 "net/http"
10 "regexp"
11 "strconv"
12 "strings"
13 "time"
14
15 "github.com/bluesky-social/indigo/atproto/identity"
16 "github.com/go-git/go-git/v5/plumbing"
17 "github.com/prometheus/client_golang/prometheus"
18 "github.com/prometheus/client_golang/prometheus/promhttp"
19 "github.com/samber/lo"
20 "tangled.org/core/repoident"
21)
22
23type IndexServer struct {
24 cfg *Config
25 dir identity.Directory
26 queue *Queue
27}
28
29func NewIndexServer(cfg *Config) *IndexServer {
30 return &IndexServer{
31 cfg: cfg,
32 dir: baseDir(cfg.PlcUrl),
33 queue: NewQueue(cfg.IndexQueueSize),
34 }
35}
36
37func (s *IndexServer) Run(ctx context.Context) {
38 // // Start a goroutine which updates the queue with commits to index.
39 // go func() {
40 // }()
41
42 for range s.cfg.IndexConcurrency {
43 go s.processQueue(ctx)
44 }
45
46 mux := http.NewServeMux()
47 mux.HandleFunc("/", s.handleHealth)
48 mux.HandleFunc("/debug/metrics", s.handleMetrics)
49 mux.HandleFunc("/debug/queue", s.handleDebugQueue)
50 mux.HandleFunc("/admin/forceIndex", s.handleForceIndex)
51 mux.HandleFunc("/admin/enqueueIndex", s.handleEnqueueIndex)
52 if err := http.ListenAndServe(s.cfg.Listen, mux); err != nil {
53 log.Fatal(err)
54 }
55}
56
57func (s *IndexServer) handleHealth(w http.ResponseWriter, r *http.Request) {
58 // Nothing to do. Just return 200
59}
60
61func (s *IndexServer) handleMetrics(w http.ResponseWriter, r *http.Request) {
62 promhttp.Handler().ServeHTTP(w, r)
63}
64
65type branchName string
66
67func (b branchName) Ref() string {
68 return lo.Ternary(b == "HEAD", "HEAD", "refs/heads/"+string(b))
69}
70
71func (b *branchName) UnmarshalText(text []byte) error {
72 name := branchName(text)
73 if strings.HasPrefix(string(name), "refs/") {
74 return fmt.Errorf("branch %q must be a short name, without the refs/ prefix", name)
75 }
76 if err := plumbing.ReferenceName(name.Ref()).Validate(); err != nil {
77 return fmt.Errorf("branch %q isn't a valid ref: %w", name, err)
78 }
79 *b = name
80 return nil
81}
82
83type objectID string
84
85var objectIDPattern = regexp.MustCompile(`^([0-9a-fA-F]{40}|[0-9a-fA-F]{64})$`)
86
87func (o *objectID) UnmarshalText(text []byte) error {
88 if !objectIDPattern.Match(text) {
89 return fmt.Errorf("%q isn't a sha1 or sha256 object id", text)
90 }
91 *o = objectID(text)
92 return nil
93}
94
95type indexBranch struct {
96 Name branchName `json:"name"`
97 Version objectID `json:"version"`
98}
99
100type indexRequest struct {
101 Repo repoident.RepoDid `json:"repo"`
102 Branches []indexBranch `json:"branches"`
103}
104
105func decodeIndexRequest(r *http.Request) (indexRequest, error) {
106 dec := json.NewDecoder(r.Body)
107 dec.DisallowUnknownFields()
108 var req indexRequest
109 if err := dec.Decode(&req); err != nil {
110 return indexRequest{}, err
111 }
112 if req.Repo == "" {
113 return indexRequest{}, errors.New("index request has no repo did")
114 }
115 if len(req.Branches) == 0 {
116 return indexRequest{}, fmt.Errorf("index request for %s has no branches", req.Repo)
117 }
118 return req, nil
119}
120
121func (s *IndexServer) handleDebugQueue(w http.ResponseWriter, r *http.Request) {
122 for _, req := range s.queue.Snapshot() {
123 fmt.Fprintln(w, req.Repo)
124 for _, b := range req.Branches {
125 fmt.Fprintf(w, "\t%s:\t%s\n", b.Name, b.Version)
126 }
127 }
128}
129
130func (s *IndexServer) handleEnqueueIndex(w http.ResponseWriter, r *http.Request) {
131 route := "enqueueIndex"
132 req, err := decodeIndexRequest(r)
133 if err != nil {
134 log.Printf("Error decoding index request: %v", err)
135 http.Error(w, "JSON parser error", http.StatusBadRequest)
136 s.incrementRequestsTotal(r.Method, route, http.StatusBadRequest)
137 return
138 }
139
140 if !s.queue.Enqueue(req) {
141 // queue full: reject so the producer retries later
142 http.Error(w, "index queue full", http.StatusServiceUnavailable)
143 s.incrementRequestsTotal(r.Method, route, http.StatusServiceUnavailable)
144 return
145 }
146
147 w.WriteHeader(http.StatusAccepted)
148 s.incrementRequestsTotal(r.Method, route, http.StatusAccepted)
149}
150
151func (s *IndexServer) handleForceIndex(w http.ResponseWriter, r *http.Request) {
152 route := "index"
153 req, err := decodeIndexRequest(r)
154 if err != nil {
155 log.Printf("Error decoding index request: %v", err)
156 http.Error(w, "JSON parser error", http.StatusBadRequest)
157 s.incrementRequestsTotal(r.Method, route, http.StatusBadRequest)
158 return
159 }
160
161 if err := gitIndex(r.Context(), s.cfg, s.dir, req); err != nil {
162 s.respondWithError(w, r.Method, route, err)
163 return
164 }
165
166 w.Header().Set("Content-Type", "application/json")
167 _ = json.NewEncoder(w).Encode(map[string]any{
168 "success": true,
169 })
170
171 s.incrementRequestsTotal(r.Method, route, http.StatusOK)
172}
173
174func (s *IndexServer) respondWithError(w http.ResponseWriter, method, route string, err error) {
175 responseCode := http.StatusInternalServerError
176
177 log.Print(err)
178 s.incrementRequestsTotal(method, route, responseCode)
179
180 w.Header().Set("Content-Type", "application/json")
181 w.WriteHeader(responseCode)
182 response := map[string]any{
183 "Success": false,
184 "Error": err.Error(),
185 }
186
187 _ = json.NewEncoder(w).Encode(response)
188}
189
190func (s *IndexServer) incrementRequestsTotal(method, route string, responseCode int) {
191 requestsTotal.With(prometheus.Labels{"code": strconv.Itoa(responseCode), "method": method, "route": route}).Inc()
192}
193
194func (s *IndexServer) processQueue(ctx context.Context) {
195 for {
196 select {
197 case <-ctx.Done():
198 return
199 default:
200 }
201
202 req, ok := s.queue.Pop()
203 if !ok {
204 time.Sleep(time.Second)
205 continue
206 }
207
208 if err := gitIndex(ctx, s.cfg, s.dir, req); err != nil {
209 log.Printf("indexing repo %s failed: %v", req.Repo, err)
210 }
211 }
212}