alice.pds.demo.boltless.dev
This repository has no description
4.1 kB
160 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1package main
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "log"
8 "net/http"
9 "strconv"
10 "time"
11
12 "github.com/bluesky-social/indigo/atproto/identity"
13 "github.com/prometheus/client_golang/prometheus"
14 "github.com/prometheus/client_golang/prometheus/promhttp"
15 "github.com/sourcegraph/zoekt"
16 "tangled.org/core/repoident"
17)
18
19type IndexServer struct {
20 cfg *Config
21 dir identity.Directory
22 queue *Queue
23}
24
25func NewIndexServer(cfg *Config) *IndexServer {
26 return &IndexServer{
27 cfg: cfg,
28 dir: baseDir(cfg.PlcUrl),
29 queue: NewQueue(cfg.IndexQueueSize),
30 }
31}
32
33func (s *IndexServer) Run(ctx context.Context) {
34 // // Start a goroutine which updates the queue with commits to index.
35 // go func() {
36 // }()
37
38 for range s.cfg.IndexConcurrency {
39 go s.processQueue(ctx)
40 }
41
42 mux := http.NewServeMux()
43 mux.HandleFunc("/", s.handleHealth)
44 mux.HandleFunc("/debug/metrics", s.handleMetrics)
45 mux.HandleFunc("/debug/queue", s.handleDebugQueue)
46 mux.HandleFunc("/admin/forceIndex", s.handleForceIndex)
47 mux.HandleFunc("/admin/enqueueIndex", s.handleEnqueueIndex)
48 if err := http.ListenAndServe(s.cfg.Listen, mux); err != nil {
49 log.Fatal(err)
50 }
51}
52
53func (s *IndexServer) handleHealth(w http.ResponseWriter, r *http.Request) {
54 // Nothing to do. Just return 200
55}
56
57func (s *IndexServer) handleMetrics(w http.ResponseWriter, r *http.Request) {
58 promhttp.Handler().ServeHTTP(w, r)
59}
60
61type indexRequest struct {
62 Repo repoident.RepoDid `json:"repo"`
63 Branches []zoekt.RepositoryBranch `json:"branches"`
64}
65
66func (s *IndexServer) handleDebugQueue(w http.ResponseWriter, r *http.Request) {
67 for _, req := range s.queue.Snapshot() {
68 fmt.Fprintln(w, req.Repo)
69 for _, b := range req.Branches {
70 fmt.Fprintf(w, "\t%s:\t%s\n", b.Name, b.Version)
71 }
72 }
73}
74
75func (s *IndexServer) handleEnqueueIndex(w http.ResponseWriter, r *http.Request) {
76 route := "enqueueIndex"
77 dec := json.NewDecoder(r.Body)
78 dec.DisallowUnknownFields()
79 var req indexRequest
80 if err := dec.Decode(&req); err != nil {
81 log.Printf("Error decoding index request: %v", err)
82 http.Error(w, "JSON parser error", http.StatusBadRequest)
83 s.incrementRequestsTotal(r.Method, route, http.StatusBadRequest)
84 return
85 }
86
87 if !s.queue.Enqueue(req) {
88 // queue full: reject so the producer retries later
89 http.Error(w, "index queue full", http.StatusServiceUnavailable)
90 s.incrementRequestsTotal(r.Method, route, http.StatusServiceUnavailable)
91 return
92 }
93
94 w.WriteHeader(http.StatusAccepted)
95 s.incrementRequestsTotal(r.Method, route, http.StatusAccepted)
96}
97
98func (s *IndexServer) handleForceIndex(w http.ResponseWriter, r *http.Request) {
99 route := "index"
100 dec := json.NewDecoder(r.Body)
101 dec.DisallowUnknownFields()
102 var req indexRequest
103 if err := dec.Decode(&req); err != nil {
104 log.Printf("Error decoding index request: %v", err)
105 http.Error(w, "JSON parser error", http.StatusBadRequest)
106 return
107 }
108
109 if err := gitIndex(r.Context(), s.cfg, s.dir, req); err != nil {
110 s.respondWithError(w, r.Method, route, err)
111 return
112 }
113
114 w.Header().Set("Content-Type", "application/json")
115 _ = json.NewEncoder(w).Encode(map[string]any{
116 "success": true,
117 })
118
119 s.incrementRequestsTotal(r.Method, route, http.StatusOK)
120}
121
122func (s *IndexServer) respondWithError(w http.ResponseWriter, method, route string, err error) {
123 responseCode := http.StatusInternalServerError
124
125 log.Print(err)
126 s.incrementRequestsTotal(method, route, responseCode)
127
128 w.Header().Set("Content-Type", "application/json")
129 w.WriteHeader(responseCode)
130 response := map[string]any{
131 "Success": false,
132 "Error": err.Error(),
133 }
134
135 _ = json.NewEncoder(w).Encode(response)
136}
137
138func (s *IndexServer) incrementRequestsTotal(method, route string, responseCode int) {
139 requestsTotal.With(prometheus.Labels{"code": strconv.Itoa(responseCode), "method": method, "route": route}).Inc()
140}
141
142func (s *IndexServer) processQueue(ctx context.Context) {
143 for {
144 select {
145 case <-ctx.Done():
146 return
147 default:
148 }
149
150 req, ok := s.queue.Pop()
151 if !ok {
152 time.Sleep(time.Second)
153 continue
154 }
155
156 if err := gitIndex(ctx, s.cfg, s.dir, req); err != nil {
157 log.Printf("indexing repo %s failed: %v", req.Repo, err)
158 }
159 }
160}