veryroundbird.house
1package spindle
2
3import (
4 "context"
5 _ "embed"
6 "encoding/json"
7 "fmt"
8 "log/slog"
9 "net/http"
10
11 "github.com/go-chi/chi/v5"
12 "tangled.sh/tangled.sh/core/api/tangled"
13 "tangled.sh/tangled.sh/core/eventconsumer"
14 "tangled.sh/tangled.sh/core/eventconsumer/cursor"
15 "tangled.sh/tangled.sh/core/idresolver"
16 "tangled.sh/tangled.sh/core/jetstream"
17 "tangled.sh/tangled.sh/core/log"
18 "tangled.sh/tangled.sh/core/notifier"
19 "tangled.sh/tangled.sh/core/rbac"
20 "tangled.sh/tangled.sh/core/spindle/config"
21 "tangled.sh/tangled.sh/core/spindle/db"
22 "tangled.sh/tangled.sh/core/spindle/engine"
23 "tangled.sh/tangled.sh/core/spindle/engines/nixery"
24 "tangled.sh/tangled.sh/core/spindle/models"
25 "tangled.sh/tangled.sh/core/spindle/queue"
26 "tangled.sh/tangled.sh/core/spindle/secrets"
27 "tangled.sh/tangled.sh/core/spindle/xrpc"
28 "tangled.sh/tangled.sh/core/xrpc/serviceauth"
29)
30
31//go:embed motd
32var motd []byte
33
34const (
35 rbacDomain = "thisserver"
36)
37
38type Spindle struct {
39 jc *jetstream.JetstreamClient
40 db *db.DB
41 e *rbac.Enforcer
42 l *slog.Logger
43 n *notifier.Notifier
44 engs map[string]models.Engine
45 jq *queue.Queue
46 cfg *config.Config
47 ks *eventconsumer.Consumer
48 res *idresolver.Resolver
49 vault secrets.Manager
50}
51
52func Run(ctx context.Context) error {
53 logger := log.FromContext(ctx)
54
55 cfg, err := config.Load(ctx)
56 if err != nil {
57 return fmt.Errorf("failed to load config: %w", err)
58 }
59
60 d, err := db.Make(cfg.Server.DBPath)
61 if err != nil {
62 return fmt.Errorf("failed to setup db: %w", err)
63 }
64
65 e, err := rbac.NewEnforcer(cfg.Server.DBPath)
66 if err != nil {
67 return fmt.Errorf("failed to setup rbac enforcer: %w", err)
68 }
69 e.E.EnableAutoSave(true)
70
71 n := notifier.New()
72
73 var vault secrets.Manager
74 switch cfg.Server.Secrets.Provider {
75 case "openbao":
76 if cfg.Server.Secrets.OpenBao.ProxyAddr == "" {
77 return fmt.Errorf("openbao proxy address is required when using openbao secrets provider")
78 }
79 vault, err = secrets.NewOpenBaoManager(
80 cfg.Server.Secrets.OpenBao.ProxyAddr,
81 logger,
82 secrets.WithMountPath(cfg.Server.Secrets.OpenBao.Mount),
83 )
84 if err != nil {
85 return fmt.Errorf("failed to setup openbao secrets provider: %w", err)
86 }
87 logger.Info("using openbao secrets provider", "proxy_address", cfg.Server.Secrets.OpenBao.ProxyAddr, "mount", cfg.Server.Secrets.OpenBao.Mount)
88 case "sqlite", "":
89 vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath, secrets.WithTableName("secrets"))
90 if err != nil {
91 return fmt.Errorf("failed to setup sqlite secrets provider: %w", err)
92 }
93 logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath)
94 default:
95 return fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider)
96 }
97
98 nixeryEng, err := nixery.New(ctx, cfg)
99 if err != nil {
100 return err
101 }
102
103 jq := queue.NewQueue(100, 5)
104
105 collections := []string{
106 tangled.SpindleMemberNSID,
107 tangled.RepoNSID,
108 tangled.RepoCollaboratorNSID,
109 }
110 jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, logger, d, true, true)
111 if err != nil {
112 return fmt.Errorf("failed to setup jetstream client: %w", err)
113 }
114 jc.AddDid(cfg.Server.Owner)
115
116 // Check if the spindle knows about any Dids;
117 dids, err := d.GetAllDids()
118 if err != nil {
119 return fmt.Errorf("failed to get all dids: %w", err)
120 }
121 for _, d := range dids {
122 jc.AddDid(d)
123 }
124
125 resolver := idresolver.DefaultResolver()
126
127 spindle := Spindle{
128 jc: jc,
129 e: e,
130 db: d,
131 l: logger,
132 n: &n,
133 engs: map[string]models.Engine{"nixery": nixeryEng},
134 jq: jq,
135 cfg: cfg,
136 res: resolver,
137 vault: vault,
138 }
139
140 err = e.AddSpindle(rbacDomain)
141 if err != nil {
142 return fmt.Errorf("failed to set rbac domain: %w", err)
143 }
144 err = spindle.configureOwner()
145 if err != nil {
146 return err
147 }
148 logger.Info("owner set", "did", cfg.Server.Owner)
149
150 // starts a job queue runner in the background
151 jq.Start()
152 defer jq.Stop()
153
154 // Stop vault token renewal if it implements Stopper
155 if stopper, ok := vault.(secrets.Stopper); ok {
156 defer stopper.Stop()
157 }
158
159 cursorStore, err := cursor.NewSQLiteStore(cfg.Server.DBPath)
160 if err != nil {
161 return fmt.Errorf("failed to setup sqlite3 cursor store: %w", err)
162 }
163
164 err = jc.StartJetstream(ctx, spindle.ingest())
165 if err != nil {
166 return fmt.Errorf("failed to start jetstream consumer: %w", err)
167 }
168
169 // for each incoming sh.tangled.pipeline, we execute
170 // spindle.processPipeline, which in turn enqueues the pipeline
171 // job in the above registered queue.
172 ccfg := eventconsumer.NewConsumerConfig()
173 ccfg.Logger = logger
174 ccfg.Dev = cfg.Server.Dev
175 ccfg.ProcessFunc = spindle.processPipeline
176 ccfg.CursorStore = cursorStore
177 knownKnots, err := d.Knots()
178 if err != nil {
179 return err
180 }
181 for _, knot := range knownKnots {
182 logger.Info("adding source start", "knot", knot)
183 ccfg.Sources[eventconsumer.NewKnotSource(knot)] = struct{}{}
184 }
185 spindle.ks = eventconsumer.NewConsumer(*ccfg)
186
187 go func() {
188 logger.Info("starting knot event consumer")
189 spindle.ks.Start(ctx)
190 }()
191
192 logger.Info("starting spindle server", "address", cfg.Server.ListenAddr)
193 logger.Error("server error", "error", http.ListenAndServe(cfg.Server.ListenAddr, spindle.Router()))
194
195 return nil
196}
197
198func (s *Spindle) Router() http.Handler {
199 mux := chi.NewRouter()
200
201 mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
202 w.Write(motd)
203 })
204 mux.HandleFunc("/events", s.Events)
205 mux.HandleFunc("/owner", func(w http.ResponseWriter, r *http.Request) {
206 w.Write([]byte(s.cfg.Server.Owner))
207 })
208 mux.HandleFunc("/logs/{knot}/{rkey}/{name}", s.Logs)
209
210 mux.Mount("/xrpc", s.XrpcRouter())
211 return mux
212}
213
214func (s *Spindle) XrpcRouter() http.Handler {
215 logger := s.l.With("route", "xrpc")
216
217 serviceAuth := serviceauth.NewServiceAuth(s.l, s.res, s.cfg.Server.Did().String())
218
219 x := xrpc.Xrpc{
220 Logger: logger,
221 Db: s.db,
222 Enforcer: s.e,
223 Engines: s.engs,
224 Config: s.cfg,
225 Resolver: s.res,
226 Vault: s.vault,
227 ServiceAuth: serviceAuth,
228 }
229
230 return x.Router()
231}
232
233func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error {
234 if msg.Nsid == tangled.PipelineNSID {
235 tpl := tangled.Pipeline{}
236 err := json.Unmarshal(msg.EventJson, &tpl)
237 if err != nil {
238 fmt.Println("error unmarshalling", err)
239 return err
240 }
241
242 if tpl.TriggerMetadata == nil {
243 return fmt.Errorf("no trigger metadata found")
244 }
245
246 if tpl.TriggerMetadata.Repo == nil {
247 return fmt.Errorf("no repo data found")
248 }
249
250 if src.Key() != tpl.TriggerMetadata.Repo.Knot {
251 return fmt.Errorf("repo knot does not match event source: %s != %s", src.Key(), tpl.TriggerMetadata.Repo.Knot)
252 }
253
254 // filter by repos
255 _, err = s.db.GetRepo(
256 tpl.TriggerMetadata.Repo.Knot,
257 tpl.TriggerMetadata.Repo.Did,
258 tpl.TriggerMetadata.Repo.Repo,
259 )
260 if err != nil {
261 return err
262 }
263
264 pipelineId := models.PipelineId{
265 Knot: src.Key(),
266 Rkey: msg.Rkey,
267 }
268
269 workflows := make(map[models.Engine][]models.Workflow)
270
271 for _, w := range tpl.Workflows {
272 if w != nil {
273 if _, ok := s.engs[w.Engine]; !ok {
274 err = s.db.StatusFailed(models.WorkflowId{
275 PipelineId: pipelineId,
276 Name: w.Name,
277 }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n)
278 if err != nil {
279 return err
280 }
281
282 continue
283 }
284
285 eng := s.engs[w.Engine]
286
287 if _, ok := workflows[eng]; !ok {
288 workflows[eng] = []models.Workflow{}
289 }
290
291 ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl)
292 if err != nil {
293 return err
294 }
295
296 workflows[eng] = append(workflows[eng], *ewf)
297
298 err = s.db.StatusPending(models.WorkflowId{
299 PipelineId: pipelineId,
300 Name: w.Name,
301 }, s.n)
302 if err != nil {
303 return err
304 }
305 }
306 }
307
308 ok := s.jq.Enqueue(queue.Job{
309 Run: func() error {
310 engine.StartWorkflows(s.l, s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{
311 RepoOwner: tpl.TriggerMetadata.Repo.Did,
312 RepoName: tpl.TriggerMetadata.Repo.Repo,
313 Workflows: workflows,
314 }, pipelineId)
315 return nil
316 },
317 OnFail: func(jobError error) {
318 s.l.Error("pipeline run failed", "error", jobError)
319 },
320 })
321 if ok {
322 s.l.Info("pipeline enqueued successfully", "id", msg.Rkey)
323 } else {
324 s.l.Error("failed to enqueue pipeline: queue is full")
325 }
326 }
327
328 return nil
329}
330
331func (s *Spindle) configureOwner() error {
332 cfgOwner := s.cfg.Server.Owner
333
334 existing, err := s.e.GetSpindleUsersByRole("server:owner", rbacDomain)
335 if err != nil {
336 return err
337 }
338
339 switch len(existing) {
340 case 0:
341 // no owner configured, continue
342 case 1:
343 // find existing owner
344 existingOwner := existing[0]
345
346 // no ownership change, this is okay
347 if existingOwner == s.cfg.Server.Owner {
348 break
349 }
350
351 // remove existing owner
352 err = s.e.RemoveSpindleOwner(rbacDomain, existingOwner)
353 if err != nil {
354 return nil
355 }
356 default:
357 return fmt.Errorf("more than one owner in DB, try deleting %q and starting over", s.cfg.Server.DBPath)
358 }
359
360 return s.e.AddSpindleOwner(rbacDomain, cfgOwner)
361}