server.go
1package bridge
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "log/slog"
8 "net/http"
9 "os"
10 "path/filepath"
11 "sort"
12 "strings"
13 "sync"
14 "time"
15
16 "git.kilimanjaro.io/ci/internal/eventlog"
17 "git.kilimanjaro.io/ci/internal/ident"
18 "git.kilimanjaro.io/ci/internal/lex"
19)
20
21// Server ingests the spool into the event log and serves the stream.
22type Server struct {
23 Log *eventlog.Log
24 SpoolDir string
25 // ForgeHost names repos: did:web:<ForgeHost>:<repo>.
26 ForgeHost string
27 // OwnerDid is the owner (and, for lack of per-user DIDs, the committer)
28 // of every repo. Default: did:web:<ForgeHost>.
29 OwnerDid string
30 // Token, if set, is required as a bearer token on /events and /ingest.
31 Token string
32 Logger *slog.Logger
33
34 mu sync.Mutex
35 poke chan struct{}
36}
37
38func (s *Server) owner() string {
39 if s.OwnerDid != "" {
40 return s.OwnerDid
41 }
42 return ident.ServiceDID(s.ForgeHost)
43}
44
45func (s *Server) logger() *slog.Logger {
46 if s.Logger != nil {
47 return s.Logger
48 }
49 return slog.Default()
50}
51
52// Ingest appends the events of every spooled push, oldest first, and deletes
53// each file once its events are in the log. A file that can't be parsed is
54// moved aside (<name>.bad) so it can't block the spool.
55func (s *Server) Ingest() (int, error) {
56 s.mu.Lock()
57 defer s.mu.Unlock()
58 files, err := filepath.Glob(filepath.Join(s.SpoolDir, "*.json"))
59 if err != nil {
60 return 0, err
61 }
62 sort.Strings(files)
63 n := 0
64 for _, f := range files {
65 b, err := os.ReadFile(f)
66 if err != nil {
67 return n, err
68 }
69 var p Push
70 if err := json.Unmarshal(b, &p); err != nil || p.Repo == "" {
71 s.logger().Error("bad spool file", "file", f, "err", err)
72 _ = os.Rename(f, f+".bad")
73 continue
74 }
75 owner := s.owner()
76 for _, u := range p.Updates {
77 u.LexiconTypeID = lex.GitRefUpdateNSID
78 u.Repo = ident.RepoDID(s.ForgeHost, p.Repo)
79 u.OwnerDid = &owner
80 u.CommitterDid = owner
81 if u.Meta == nil {
82 u.Meta = &lex.GitRefUpdateMeta{CommitCount: &lex.GitRefUpdateCommitCount{}}
83 }
84 if _, err := s.Log.Append(lex.GitRefUpdateNSID, "", u); err != nil {
85 return n, err
86 }
87 n++
88 s.logger().Info("ref update", "repo", p.Repo, "ref", u.Ref, "new", u.NewSha)
89 }
90 if err := os.Remove(f); err != nil {
91 return n, err
92 }
93 }
94 return n, nil
95}
96
97// Run ingests on start, whenever /ingest is poked, and every interval.
98func (s *Server) Run(ctx context.Context, interval time.Duration) {
99 s.mu.Lock()
100 if s.poke == nil {
101 s.poke = make(chan struct{}, 1)
102 }
103 poke := s.poke
104 s.mu.Unlock()
105 for {
106 if _, err := s.Ingest(); err != nil {
107 s.logger().Error("ingest", "err", err)
108 }
109 select {
110 case <-ctx.Done():
111 return
112 case <-poke:
113 case <-time.After(interval):
114 }
115 }
116}
117
118// Handler serves:
119//
120// GET /events the knot-compatible event stream
121// POST /ingest a hook's poke: ingest the spool now
122// GET /.well-known/did.json the bridge's did:web document
123// GET /healthz 200 when the log is readable
124func (s *Server) Handler() http.Handler {
125 mux := http.NewServeMux()
126 mux.Handle("GET /events", eventlog.RequireToken(s.Token, s.Log))
127 mux.Handle("POST /ingest", eventlog.RequireToken(s.Token, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
128 s.mu.Lock()
129 if s.poke == nil {
130 s.poke = make(chan struct{}, 1)
131 }
132 select {
133 case s.poke <- struct{}{}:
134 default:
135 }
136 s.mu.Unlock()
137 w.WriteHeader(http.StatusAccepted)
138 })))
139 mux.HandleFunc("GET /.well-known/did.json", func(w http.ResponseWriter, r *http.Request) {
140 w.Header().Set("Content-Type", "application/did+json")
141 _ = json.NewEncoder(w).Encode(map[string]any{
142 "@context": []string{"https://www.w3.org/ns/did/v1"},
143 "id": ident.ServiceDID(s.ForgeHost),
144 "service": []map[string]string{{
145 "id": "#tangled_knot", "type": "TangledKnot", "serviceEndpoint": "https://" + s.ForgeHost,
146 }},
147 })
148 })
149 mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
150 if _, err := s.Log.List(0, 1); err != nil {
151 http.Error(w, err.Error(), http.StatusServiceUnavailable)
152 return
153 }
154 fmt.Fprintln(w, "ok")
155 })
156 return mux
157}
158
159// Notify is the hook side: spool a push, then poke the server (best
160// effort). Errors are returned for logging; the hook still exits 0.
161func Notify(ctx context.Context, spoolDir, server, token string, p Push) error {
162 if len(p.Updates) == 0 {
163 return nil
164 }
165 if _, err := Spool(spoolDir, p); err != nil {
166 return fmt.Errorf("spool: %w", err)
167 }
168 if server == "" {
169 return nil
170 }
171 ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
172 defer cancel()
173 req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(server, "/")+"/ingest", nil)
174 if err != nil {
175 return err
176 }
177 if token != "" {
178 req.Header.Set("Authorization", "Bearer "+token)
179 }
180 resp, err := http.DefaultClient.Do(req)
181 if err != nil {
182 return nil // spooled; the server picks it up on its next pass
183 }
184 resp.Body.Close()
185 return nil
186}
