coverage — ~bigbes/sr-ht-spec — 64cae3af — cmd/specsrht/main.go

Coverage
8.8% 21/239 statements
Δ
-0.1
Blob
6058741
Uncovered L128-L133L140-L155L159-L165L182-L185L187-L226L228-L236L238-L239L265-L276L278-L287L289-L318L325-L331L333-L352L387-L434L436-L440L490-L501L510-L520L522-L523L527-L537L540L567-L577L579-L588L594-L598L609-L622L629-L639L641-L645L647L650-L654L676-L683L725-L730L736-L740L743-L746L748-L767L773-L775L777-L780L787-L805L821-L827L829-L832
1 // Command specsrht is the spec.sr.ht daemon, and — invoked under another name
2 // — the receive hooks of every space it serves.
3 //
4 // It runs three things in one process:
5 //
6 // - the hook RPC socket at <repos>/.specsrht/hook.sock, which the pre-receive,
7 // update and post-receive hooks of every space call. Validation lives here
8 // and not in the hooks because bleve is single-writer and this process holds
9 // the index, and because the push path and the API must not be able to
10 // disagree about what is valid.
11 // - an HTTP listener on -b (default localhost:5091), serving /healthz, the
12 // read UI, /mcp for agents and /query for GraphQL — one listener, so one
13 // reverse-proxy route covers the service.
14 // - the reconciler, at startup and then periodically, repairing the
15 // divergence a killed daemon leaves between git refs, Postgres and the
16 // index.
17 //
18 // # Running as a hook
19 //
20 // Every hook a space's repository runs is a symlink to this binary. When
21 // argv[0] names a hook the process handles that hook and exits, touching
22 // neither the configuration nor the database; see the hooks package. The
23 // explicit form `specsrht hook <name>` does the same thing by hand.
24 //
25 // # Admin commands
26 //
27 // Two subcommands run and exit without binding anything, so they are safe to
28 // invoke while the daemon is up:
29 //
30 // specsrht space create ~owner/name | specsrht space list
31 // specsrht doc propose ~owner/space <file>... [--as path] [--title t]
32 //
33 // The first is the only entry point a space has, and a deployment without one
34 // is inert: there is nothing to hold documents. The second opens a proposal
35 // from files on this host, for when the documents and the operator are already
36 // here and a bearer token would be ceremony.
37 //
38 // There is no `specsrht token`. It minted spec's own agent credential; agent
39 // credentials are tokens.sr.ht working tokens now, so they are minted there —
40 // its /tokens page, or POST /exchange with a parent token — and carry the
41 // spec:propose grant to write and spec:read to read.
42 //
43 // # Flags
44 //
45 // Parsed by core-go's server.New:
46 //
47 // -b addr bind address (repeatable); default localhost:5091
48 // -d debug: verbose request logging in core-go, and — read straight out
49 // of the argument vector by sr-ht-ecore's logging.Defaults, before
50 // server.New parses anything — debug verbosity for this daemon's own
51 // logger from its first line. $LOG_LEVEL does the same for one run,
52 // and [spec.sr.ht] log-level is the instance's persistent setting.
53 // -m addr Prometheus metrics bind (default random port)
54 // -p addr pprof bind (default random localhost port)
55 //
56 // Configuration comes from the shared SourceHut config.ini through core-go's
57 // fixed search path. Every required key is checked before anything is opened,
58 // and all missing ones are reported together, so a misconfigured instance
59 // fails once at startup rather than once per restart.
60 //
61 // # Shutdown
62 //
63 // SIGINT and SIGTERM both start a warm shutdown. core-go's server.Run only
64 // handles SIGINT — which is why compare.sr.ht's systemd unit sets
65 // KillSignal=SIGINT — so this daemon installs a handler that turns a SIGTERM
66 // into that same SIGINT. A unit for this service therefore needs no
67 // KillSignal= line: the systemd default works.
68 package main
69
70 import (
71 "context"
72 "database/sql"
73 "errors"
74 "fmt"
75 "log/slog"
76 "net/http"
77 "os"
78 "os/signal"
79 "path/filepath"
80 "strings"
81 "syscall"
82 "time"
83
84 "github.com/go-chi/chi/v5"
85 chimw "github.com/go-chi/chi/v5/middleware"
86 _ "github.com/lib/pq" // registers the "postgres" database/sql driver
87 "github.com/vaughan0/go-ini"
88 "go.bigb.es/auxilia/scribe"
89 "sourcecraft.dev/bigbes/sr-ht-ecore/apimeta"
90 "sourcecraft.dev/bigbes/sr-ht-ecore/logging"
91
92 "sourcecraft.dev/bigbes/sr-ht-core/config"
93 "sourcecraft.dev/bigbes/sr-ht-core/database"
94 coreserver "sourcecraft.dev/bigbes/sr-ht-core/server"
95 "sourcecraft.dev/bigbes/sr-ht-core/webhooks"
96
97 "sourcecraft.dev/bigbes/sr-ht-spec/api"
98 "sourcecraft.dev/bigbes/sr-ht-spec/core"
99 "sourcecraft.dev/bigbes/sr-ht-spec/graph"
100 "sourcecraft.dev/bigbes/sr-ht-spec/hooks"
101 "sourcecraft.dev/bigbes/sr-ht-spec/mcpsrv"
102 "sourcecraft.dev/bigbes/sr-ht-spec/search"
103 "sourcecraft.dev/bigbes/sr-ht-spec/service"
104 "sourcecraft.dev/bigbes/sr-ht-spec/web"
105 )
106
107 const (
108 // serviceName is the SourceHut service identifier and our config section.
109 // The ".sr.ht" suffix is what puts us in the nav network list.
110 serviceName = "spec.sr.ht"
111
112 // defaultBind is the address core-go binds when no -b is given.
113 defaultBind = "localhost:5091"
114
115 // version is reported in the MCP handshake so a client listing several
116 // SourceHut MCP endpoints can tell which build it is talking to.
117 version = "dev"
118
119 // pingTimeout bounds the startup connectivity check against Postgres.
120 pingTimeout = 10 * time.Second
121
122 // shutdownGrace is how long the hook socket is given to finish the calls
123 // already in flight once the HTTP listener has drained. A push being
124 // validated at that moment finishes rather than being failed closed.
125 shutdownGrace = 30 * time.Second
126 )
127
128 0 func main() {
129 0 // Hook mode first: a hook must not read a config file, open Postgres, or
130 0 // bind anything. It talks to the daemon over a socket and exits.
131 0 if _, _, isHook := hooks.ModeFromArgs(os.Args); isHook {
132 0 os.Exit(hooks.Run(hooks.Runtime{Args: os.Args}))
133 0 }
134
135 // The config is read before the logger so that [spec.sr.ht] log-level is in
136 // force for the first line this process writes. It cannot fail: core-go's
137 // LoadConfig answers a nil ini.File for an instance with no config file at
138 // all, and validateConfig — which runs later, in run — is what turns that
139 // into one message an operator can act on.
140 0 conf := config.LoadConfig()
141 0 log := installLogger(conf)
142 0
143 0 // Admin subcommands run and exit without binding anything, so they are safe
144 0 // to invoke while the daemon holds the hook socket.
145 0 if len(os.Args) > 1 {
146 0 admin := map[string]func([]string) error{
147 0 "space": runSpace,
148 0 "doc": runDoc,
149 0 }
150 0 if cmd := os.Args[1]; admin[cmd] != nil {
151 0 if err := admin[cmd](os.Args[2:]); err != nil {
152 0 fmt.Fprintf(os.Stderr, "specsrht %s: %v\n", cmd, err)
153 0 os.Exit(1)
154 0 }
155 0 return
156 }
157 }
158
159 0 if err := run(conf, log); err != nil {
160 0 // Plain text, not a log record. A startup failure is read by a human
161 0 // on a terminal, and the configuration report is deliberately several
162 0 // lines long — a structured handler would escape it into one.
163 0 fmt.Fprintf(os.Stderr, "spec.sr.ht did not start: %v\n", err)
164 0 os.Exit(1)
165 0 }
166 }
167
168 // runSpace is the space administration command:
169 //
170 // specsrht space create ~owner/name
171 // specsrht space list
172 //
173 // Spaces have no other entry point. The read plane only reads, and the write
174 // plane is the proposal API, which operates on documents inside a space that
175 // already exists — so without this, a freshly deployed instance has no way to
176 // hold anything at all.
177 //
178 // It installs the receive hooks itself rather than leaving them to the daemon's
179 // startup refresh: a space created while the daemon is running would otherwise
180 // accept unvalidated pushes until the next restart, which is exactly the
181 // fail-open the receive path exists to prevent.
182 0 func runSpace(args []string) error {
183 0 if len(args) == 0 {
184 0 return errors.New("usage: specsrht space create ~owner/name | specsrht space list")
185 0 }
186
187 0 conf := config.LoadConfig()
188 0 cfg, err := validateConfig(conf)
189 0 if err != nil {
190 0 return err
191 0 }
192 0 pool, err := openDatabase(cfg.ConnectionString)
193 0 if err != nil {
194 0 return err
195 0 }
196 0 defer pool.Close()
197 0
198 0 svc, err := service.New(cfg, pool)
199 0 if err != nil {
200 0 return err
201 0 }
202 0 ctx := context.Background()
203 0
204 0 switch args[0] {
205 0 case "create":
206 0 if len(args) != 2 {
207 0 return errors.New("usage: specsrht space create ~owner/name")
208 0 }
209 0 ref, err := core.ParseSpaceRef(args[1])
210 0 if err != nil {
211 0 return fmt.Errorf("parse %q: %w", args[1], err)
212 0 }
213 0 if _, err := svc.CreateSpace(ctx, ref); err != nil {
214 0 return err
215 0 }
216 0 binary, err := os.Executable()
217 0 if err != nil {
218 0 return fmt.Errorf("locate this binary, which every hook symlinks to: %w", err)
219 0 }
220 0 if err := hooks.InstallSpace(cfg.Repos, ref, hooks.InstallOptions{Binary: binary}); err != nil {
221 0 return fmt.Errorf("install receive hooks for %s: %w", ref, err)
222 0 }
223 0 fmt.Printf("created %s\n repo: %s\n clone: git clone %s\n",
224 0 ref, filepath.Join(cfg.Repos, "~"+ref.Owner, ref.Name),
225 0 filepath.Join(cfg.Repos, "~"+ref.Owner, ref.Name))
226 0 return nil
227
228 0 case "list":
229 0 spaces, err := svc.ListSpaces(ctx)
230 0 if err != nil {
231 0 return err
232 0 }
233 0 for _, sp := range spaces {
234 0 fmt.Println(sp.Ref)
235 0 }
236 0 return nil
237
238 0 default:
239 0 return fmt.Errorf("unknown subcommand %q: want create or list", args[0])
240 }
241 }
242
243 // installLogger builds the process logger and makes it slog's default.
244 //
245 // The policy — the verbosity, the source positions, the colour decision and the
246 // set of attribute keys that must never reach a log file — is sr-ht-ecore's
247 // logging.Defaults, because none of it is spec.sr.ht's to decide. -d, $LOG_LEVEL
248 // and [spec.sr.ht] log-level are how an operator addresses every daemon on this
249 // instance, and what must be redacted (the unified-login cookie, tokens.sr.ht
250 // working tokens, the Authorization header they travel in) is a fact about the
251 // instance rather than about this service. The local mask list is gone; the
252 // network-key and private-key entries it contributed are in the shared one, and
253 // so are the dsn and connection-string names it did not have.
254 //
255 // The handler stays here, and it is scribe's, because ecore deliberately builds
256 // none. Everything goes to stderr: a hook's stdout is forwarded to the pushing
257 // client, and this binary is both programs.
258 //
259 // logging.Install is what makes the library packages loggable at all. The read
260 // plane, the credential resolver and sr-ht-ecore's panic middleware all log
261 // through the default logger and none of them takes a *slog.Logger — a
262 // middleware in another module cannot be handed this one, and without the
263 // SetDefault its panic reports would come out of Go's plain stderr handler with
264 // none of the masking applied.
265 0 func installLogger(conf ini.File) *slog.Logger {
266 0 opts := logging.Defaults(conf, serviceName)
267 0 return logging.Install(scribe.NewTintHandler(
268 0 scribe.WithWriter(os.Stderr),
269 0 scribe.WithLevel(opts.Level),
270 0 scribe.WithSource(opts.AddSource),
271 0 scribe.WithTimeFormat(opts.TimeFormat),
272 0 scribe.WithNoColor(!opts.Color),
273 0 scribe.WithMaskKeys(opts.MaskKeys...),
274 0 scribe.WithMask(opts.MaskPattern, opts.MaskReplacement),
275 0 ))
276 0 }
277
278 0 func run(conf ini.File, log *slog.Logger) error {
279 0 // The config was read in main, before the logger, so that a verbosity written
280 0 // in config.ini is in force for the first line this process writes. It never
281 0 // fails on a missing file — core-go's LoadConfig returns a nil ini.File — so
282 0 // validateConfig is what turns an unconfigured instance into one clear
283 0 // message instead of a panic deep inside the first request.
284 0 cfg, err := validateConfig(conf)
285 0 if err != nil {
286 0 return err
287 0 }
288
289 0 pool, err := openDatabase(cfg.ConnectionString)
290 0 if err != nil {
291 0 return err
292 0 }
293 0 defer pool.Close()
294 0
295 0 // WithInstanceTokens builds the agent credential plane every surface shares.
296 0 // It is required, not offered: spec.sr.ht mints no credential of its own any
297 0 // more, so an instance with no [tokens.sr.ht] section could authenticate no
298 0 // agent at all, over HTTP or over `git push`. service.New fails here rather
299 0 // than letting the daemon come up and refuse every agent one request at a
300 0 // time.
301 0 //
302 0 // /query holds a second plane the resolver knows nothing about — the
303 0 // meta.sr.ht personal access token newMetaPlane builds — and that one is
304 0 // deliberately not here, because a plane on the resolver is a plane on every
305 0 // surface. It is built beside the graph.New call in newSurfaces instead.
306 0 svc, err := service.New(cfg, pool, service.WithInstanceTokens(conf))
307 0 if err != nil {
308 0 return err
309 0 }
310 0 log.Info("agent credential plane", "tokens.sr.ht", svc.Resolver().HasInstancePlane())
311 0
312 0 // Seed the owner's user row before serving. core-go's auth.Middleware looks
313 0 // a request's username up in the "user" table and, on a miss, calls out to
314 0 // meta.sr.ht — seeding the single owner up front keeps that lookup local, and
315 0 // it gives the webhook engine the user_id it scopes subscriptions by.
316 0 if err := svc.EnsureOwnerUser(context.Background()); err != nil {
317 0 return fmt.Errorf("seed the owner user row: %w", err)
318 0 }
319
320 // Refresh every space's hooks before anything can be pushed to it. This is
321 // fatal on failure by design: a space whose hooks are missing accepts
322 // pushes that are never validated, which is the one outcome the whole
323 // receive path exists to prevent. A daemon that will not start is loud; a
324 // space quietly accepting malformed documents is not.
325 0 binary, err := os.Executable()
326 0 if err != nil {
327 0 return fmt.Errorf("locate this binary, which every hook symlinks to: %w", err)
328 0 }
329 0 if err := refreshHooks(context.Background(), log, svc, binary); err != nil {
330 0 return err
331 0 }
332
333 0 surf, err := newSurfaces(conf, cfg, svc, version)
334 0 if err != nil {
335 0 return err
336 0 }
337 0 defer surf.Close()
338 0
339 0 // After the surfaces, because the push notifier reindexes through the same
340 0 // index they read from — one bleve writer, held here.
341 0 hookSrv, err := hooks.NewServer(hooks.Options{
342 0 Backend: svc,
343 0 Socket: hooks.SocketPath(cfg.Repos),
344 0 Log: log,
345 0 OnPush: pushNotifier(log, svc, surf.index),
346 0 })
347 0 if err != nil {
348 0 return err
349 0 }
350 0 if err := hookSrv.Listen(); err != nil {
351 0 return err
352 0 }
353
354 // server.New parses -b/-d/-m/-p and runs crypto.InitCrypto(conf), whose
355 // two required keys validateConfig already checked, so it cannot fatal
356 // here for a reason we have not already reported. It must run before
357 // mountRoutes: apimeta.Handler reads the webhook public key InitCrypto
358 // establishes, once, when the handler is built.
359 //
360 // WithDefaultMiddleware is here for the database pool, the redis client and
361 // the email queue that WithQueues hands the webhook delivery worker — not
362 // for the authenticated router it decorates, which now carries no routes at
363 // all. There is deliberately no WithSchema: it would mount /query on that
364 // authenticated router, behind core-go's auth.Middleware, which speaks
365 // meta.sr.ht's OAuth vocabulary and 401s the tokens.sr.ht one every other
366 // surface of this service accepts. /query takes both — a working token and a
367 // meta PAT — and serving two planes from one endpoint is exactly what that
368 // middleware cannot do. It is mounted on the anonymous router by mountRoutes
369 // instead, with graph's own credential middleware in front of it, and
370 // api-meta.json is served there too because core-go serves that file only for
371 // the schemas it hosts itself.
372 //
373 // MaxComplexity is the one thing WithSchema set that still has to be set,
374 // and its reader has nothing to do with serving /query: the webhook delivery
375 // worker runs a subscriber's stored query through corewebhooks.Exec, which
376 // reads the bound off this field — through the context WithQueues gives it,
377 // which is the one context in this daemon that still carries core-go's
378 // server — and refuses everything above it. Zero does not mean "no limit"
379 // there; it means every delivery fails, logged and not raised. Measured, by
380 // removing this line: "operation has complexity 2, which exceeds the maximum
381 // of 0" and no delivery.
382 //
383 // Server.Schema is deliberately not set. WithSchema assigns it, but nothing
384 // in core-go reads it back — the delivery worker executes against the schema
385 // it was handed in NewQueue, and the resolvers that used to reach for it
386 // through the server context now hold their own.
387 0 limit, err := maxComplexity(conf)
388 0 if err != nil {
389 0 return fmt.Errorf("[%s::api] max-complexity: %w", serviceName, err)
390 0 }
391 0 webhookQueue := webhooks.NewQueue(surf.gql.Schema(), conf)
392 0 srv := coreserver.New(serviceName, defaultBind, conf, os.Args).
393 0 WithDefaultMiddleware()
394 0 srv.MaxComplexity = limit
395 0 srv.WithQueues(webhookQueue.Queue)
396 0 mountRoutes(srv.AnonRouter(), conf, pool, surf)
397 0
398 0 // Now that the webhook queue is started (WithQueues gave its worker the
399 0 // server+database+config context), install the sink so proposal lifecycle
400 0 // events fire deliveries. The owner user id is valid — EnsureOwnerUser ran
401 0 // above.
402 0 svc.SetEventSink(newWebhookEventSink(webhookQueue, svc.OwnerUserID(), cfg.Instance.OwnerName, log))
403 0
404 0 ctx, stop := context.WithCancel(context.Background())
405 0 defer stop()
406 0
407 0 served := make(chan error, 1)
408 0 go func() { served <- hookSrv.Serve(ctx) }()
409 0 go svc.RunReconciler(ctx, service.DefaultReconcileInterval, reconcileReporter(log))
410 0
411 0 bridgeSIGTERM(log)
412 0
413 0 log.Info("spec.sr.ht starting",
414 0 "bind", defaultBind,
415 0 "repos", cfg.Repos,
416 0 "cache", cfg.Cache,
417 0 "origin", cfg.Origin,
418 0 "hook_socket", hookSrv.Socket(),
419 0 "reconcile_interval", service.DefaultReconcileInterval.String(),
420 0 )
421 0
422 0 // Blocks until SIGINT — which bridgeSIGTERM makes SIGTERM equivalent to —
423 0 // and then drains the HTTP listeners.
424 0 srv.Run()
425 0
426 0 log.Info("draining the hook socket", "grace", shutdownGrace.String())
427 0 stop()
428 0 select {
429 0 case err := <-served:
430 0 if err != nil {
431 0 log.Error("hook socket stopped with an error", scribe.Err(err))
432 0 }
433 0 case <-time.After(shutdownGrace):
434 0 log.Warn("hook socket did not drain in time; closing it")
435 }
436 0 if err := hookSrv.Close(); err != nil {
437 0 log.Error("could not close the hook socket", scribe.Err(err))
438 0 }
439 0 log.Info("spec.sr.ht stopped")
440 0 return nil
441 }
442
443 // validateConfig checks every key this daemon needs before anything is opened,
444 // and reports all of the missing ones at once so an operator fixes the config
445 // in one pass instead of discovering each gap on a separate restart.
446 //
447 // service.LoadConfig owns our own section and collects its own gaps the same
448 // way; the two lists are merged into one message. The keys checked here are
449 // the ones core-go itself fatals on, which belong to server.New's contract
450 // rather than to service/ — duplicating them there would give the instance two
451 // lists to keep in sync.
452 12 func validateConfig(conf ini.File) (service.Config, error) {
453 12 var missing []string
454 24 require := func(section, key, why string) {
455 24 if v, ok := conf.Get(section, key); !ok || strings.TrimSpace(v) == "" {
456 5 missing = append(missing, fmt.Sprintf("[%s] %s — %s", section, key, why))
457 5 }
458 }
459
460 // Both are read by crypto.InitCrypto, which server.New calls and which
461 // fatals with a terse message when either is absent. The webhook key is
462 // required even though v1 emits no webhooks.
463 12 require("sr.ht", "network-key", "fernet key for the unified-login cookie")
464 12 require("webhooks", "private-key", "webhook signing key; crypto.InitCrypto requires it")
465 12
466 12 cfg, cfgErr := service.LoadConfig(conf)
467 12
468 12 if len(missing) == 0 && cfgErr == nil {
469 1 return cfg, nil
470 1 }
471
472 11 var b strings.Builder
473 11 b.WriteString("incomplete configuration.")
474 11 if len(missing) > 0 {
475 4 fmt.Fprintf(&b, "\n\nMissing keys the SourceHut runtime requires:\n\t%s",
476 4 strings.Join(missing, "\n\t"))
477 4 }
478 11 if cfgErr != nil {
479 8 fmt.Fprintf(&b, "\n\n%s", cfgErr)
480 8 }
481 11 return service.Config{}, errors.New(b.String())
482 }
483
484 // openDatabase opens the pool the whole daemon shares — request handlers, the
485 // reconciler and the hook RPC alike — and proves it works before serving.
486 //
487 // sql.Open alone connects lazily, so a wrong DSN would first surface as a
488 // rejected push. Fail-closed makes that safe but not pleasant; failing at
489 // startup names the problem while somebody is still watching.
490 0 func openDatabase(dsn string) (*sql.DB, error) {
491 0 pool, err := sql.Open("postgres", dsn)
492 0 if err != nil {
493 0 return nil, fmt.Errorf("open the database: %w", err)
494 0 }
495 0 ctx, cancel := context.WithTimeout(context.Background(), pingTimeout)
496 0 defer cancel()
497 0 if err := pool.PingContext(ctx); err != nil {
498 0 pool.Close()
499 0 return nil, fmt.Errorf("reach the database: %w", err)
500 0 }
501 0 return pool, nil
502 }
503
504 // refreshHooks installs this binary's hooks into every space.
505 //
506 // It runs at startup rather than only at space creation so that an upgrade
507 // which changes the wire protocol, the socket path or the hook set repairs
508 // every repository by restarting — there is no separate migration step and no
509 // repository left speaking last week's protocol.
510 0 func refreshHooks(ctx context.Context, log *slog.Logger, svc *service.Service, binary string) error {
511 0 spaces, err := svc.ListSpaces(ctx)
512 0 if err != nil {
513 0 return fmt.Errorf("list spaces to refresh their hooks: %w", err)
514 0 }
515 0 for _, sp := range spaces {
516 0 if err := hooks.InstallSpace(svc.ReposRoot(), sp.Ref, hooks.InstallOptions{Binary: binary}); err != nil {
517 0 return fmt.Errorf("install the receive hooks of %s: %w "+
518 0 "(a space whose hooks are missing would accept unvalidated pushes, so this is fatal; "+
519 0 "repair or remove the repository and start again)", sp.Ref, err)
520 0 }
521 }
522 0 log.Info("receive hooks refreshed", "spaces", len(spaces), "binary", binary)
523 0 return nil
524 }
525
526 // mountRoutes installs what the daemon serves over HTTP.
527 0 func mountRoutes(router chi.Router, conf ini.File, pool *sql.DB, surfaces *surfaces) {
528 0 // server.New already froze the anonymous router for direct middleware
529 0 // registration, so middleware and routes go in together inside a Group —
530 0 // which chi permits on a fresh inline mux sharing the same routing tree.
531 0 router.Group(func(r chi.Router) {
532 0 r.Use(chimw.RealIP)
533 0 r.Use(chimw.Recoverer)
534 0 r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) {
535 0 w.Header().Set("Content-Type", "text/plain; charset=utf-8")
536 0 fmt.Fprint(w, "ok")
537 0 })
538 })
539
540 0 mountWeb(router, conf, pool, surfaces)
541 }
542
543 // surfaces are the three Phase 2 read surfaces, assembled once at startup.
544 //
545 // They share one *search.Index deliberately: bleve is single-writer, so a second
546 // Open on the same directory is not merely wasteful but wrong.
547 type surfaces struct {
548 index *search.Index
549 web *web.Server
550 mcp http.Handler
551 api http.Handler
552
553 // gql is the /query endpoint. Like /mcp and /api it carries its own
554 // credential middleware and is mounted on the anonymous router; unlike them
555 // it also owns the executable schema, which is handed to the webhook queue
556 // so a subscription's stored query is executed at delivery time against
557 // exactly the schema its author wrote it for.
558 gql *graph.Server
559 }
560
561 // newSurfaces opens the index and builds the three read surfaces over it.
562 //
563 // All three go through service/ and none of them re-derives addressing, the
564 // read contract or the project filter — that shared layer is the whole reason
565 // "read SPEC-0007" cannot mean three different things depending on which door
566 // you knock on.
567 0 func newSurfaces(conf ini.File, cfg service.Config, svc *service.Service, version string) (*surfaces, error) {
568 0 indexPath := filepath.Join(cfg.Cache, "index")
569 0 // A crashed rebuild leaves working directories beside the index; clearing
570 0 // them before Open is always safe, since the index is a pure cache.
571 0 if err := search.CleanStale(indexPath); err != nil {
572 0 return nil, fmt.Errorf("clean stale index working dirs: %w", err)
573 0 }
574 0 index, err := search.Open(indexPath)
575 0 if err != nil {
576 0 return nil, fmt.Errorf("open search index at %s: %w", indexPath, err)
577 0 }
578
579 0 site, err := web.New(web.Options{
580 0 Conf: conf,
581 0 Reader: web.NewReader(svc),
582 0 Searcher: index,
583 0 Resolver: svc.Resolver(),
584 0 })
585 0 if err != nil {
586 0 index.Close()
587 0 return nil, fmt.Errorf("assemble the web UI: %w", err)
588 0 }
589
590 // The origin is the /mcp Host allowlist: the MCP SDK's own DNS-rebinding
591 // guard cannot tell a reverse proxy from an attacker (both reach a loopback
592 // listener with a non-loopback Host), so it is replaced by a check against
593 // this value. Traefik must pass the Host header through or every call 403s.
594 0 mcp, err := mcpsrv.Handler(mcpsrv.Backend{Docs: svc, Index: index, Write: svc}, version, cfg.Origin)
595 0 if err != nil {
596 0 index.Close()
597 0 return nil, fmt.Errorf("assemble the MCP surface: %w", err)
598 0 }
599 // spec_propose resolves the acting agent from the bearer token on the tool
600 // call, so /mcp needs the principal middleware the read tools never did.
601 // It sets an anonymous principal when there is no token, which service.Propose
602 // refuses — the ACL stays in service/, this only populates the identity.
603 //
604 // mcpsrv.Gate sits inside that middleware and closes the read plane: the read
605 // tools (spec_search/spec_read/spec_list) enforced nothing on their own, so it
606 // applies the owner+agents ACL to the whole surface — the same one graph's
607 // /query and the web UI apply. spec_propose stays fail-closed in service/ too;
608 // the gate just makes the read tools match.
609 0 mcp = svc.Resolver().Middleware()(mcpsrv.Gate(mcp))
610 0
611 0 // The one credential plane that is not shared. /query is federated into
612 0 // api.sr.ht, which forwards a single client Authorization header to every
613 0 // service a query touches, so it has to take the meta.sr.ht personal access
614 0 // token that header carries; the three surfaces above must not, because a PAT
615 0 // would then be a way around the tokens.sr.ht grant they require. Building it
616 0 // beside the one endpoint that gets it is what keeps that true — see
617 0 // newMetaPlane.
618 0 metaPlane, err := newMetaPlane(svc.Resolver().Owner())
619 0 if err != nil {
620 0 index.Close()
621 0 return nil, err
622 0 }
623
624 // /query, with the credential plane /mcp and /api use — the resolver is the
625 // one svc holds, so a token that reads through one surface reads through all
626 // three — plus the meta plane above, which this surface alone holds.
627 // graph.Server installs that middleware itself, which is why it is mounted on
628 // the anonymous router below and not through core-go's WithSchema.
629 0 gql, err := graph.New(graph.Options{
630 0 Reader: svc,
631 0 Searcher: index,
632 0 Proposals: graph.NewProposals(svc),
633 0 Resolver: svc.Resolver(),
634 0 Meta: metaPlane,
635 0 })
636 0 if err != nil {
637 0 index.Close()
638 0 return nil, fmt.Errorf("assemble the GraphQL surface: %w", err)
639 0 }
640
641 0 rest, err := api.New(api.Options{Writer: svc, Resolver: svc.Resolver()})
642 0 if err != nil {
643 0 index.Close()
644 0 return nil, fmt.Errorf("assemble the REST write surface: %w", err)
645 0 }
646
647 0 return &surfaces{index: index, web: site, mcp: mcp, api: rest.Handler(), gql: gql}, nil
648 }
649
650 0 func (s *surfaces) Close() error {
651 0 if s == nil || s.index == nil {
652 0 return nil
653 0 }
654 0 return s.index.Close()
655 }
656
657 // mountWeb attaches the anonymous-router HTTP surfaces: the web UI, the MCP
658 // endpoint, the REST write plane and /query.
659 //
660 // All four keep spec's own authentication (tokens.sr.ht working tokens,
661 // anonymous-capable reads, login redirects for the browser), which is why they
662 // are here rather than behind core-go's 401-by-default auth. /query used to be
663 // the exception, served by core-go's server.WithSchema on the authenticated
664 // router; it authenticated with meta's OAuth vocabulary there, which is not the
665 // vocabulary the other three speak, and a caller holding a working token that
666 // works everywhere else on this service was refused by the one surface meant to
667 // be the instance-native read plane.
668 //
669 // "mcp", "api" and "query" are all legal space names as far as the router is
670 // concerned, so each of them is a path the web UI's "/" mount could plausibly
671 // serve as a document. It does not: chi resolves by trie specificity and not by
672 // registration order, which was measured rather than assumed — a preceding
673 // comment here asserted the opposite, and moving mountGraphQL after the "/"
674 // mount changes no route. The registration order below is for reading, not for
675 // routing.
676 0 func mountWeb(router chi.Router, conf ini.File, pool *sql.DB, s *surfaces) {
677 0 if s == nil {
678 0 return
679 0 }
680 0 router.Handle("/mcp", s.mcp)
681 0 router.Mount("/api", s.api)
682 0 mountGraphQL(router, conf, pool, s.gql)
683 0 router.Mount("/", s.web.Handler())
684 }
685
686 // mountGraphQL installs /query and the api-meta.json beside it.
687 //
688 // The endpoint carries its own credential middleware, so it goes here on the
689 // anonymous router rather than through core-go's server.WithSchema. It does need
690 // two things from the router that /mcp and /api do not:
691 //
692 // - core-go's config and database middleware. The webhook management resolvers
693 // open transactions through core-go's database context, and
694 // WithDefaultMiddleware installs that on the authenticated router only —
695 // which this is not.
696 // - api-meta.json, at core-go's own path. meta.sr.ht fetches that file from
697 // every service it discovers to build /oauth2/personal-token, core-go serves
698 // it only for the schemas it hosts itself, and this service now hosts its
699 // own. A 404 there is a broken personal-token page for the whole instance.
700 //
701 // The middleware goes on a Group and not on router, because chi refuses a Use
702 // once any route exists on a mux, and this router already has some. The Group is
703 // a fresh inline mux over the same routing tree, which is where middleware and
704 // routes can still be attached together.
705 //
706 // api-meta.json is outside that Group deliberately: it is a static document that
707 // reads neither the config nor the database, and giving it a database
708 // transaction's worth of setup for every poll from meta would be work done for
709 // nobody.
710 1 func mountGraphQL(router chi.Router, conf ini.File, pool *sql.DB, gql http.Handler) {
711 1 router.Group(func(r chi.Router) {
712 1 r.Use(config.Middleware(conf, serviceName))
713 1 r.Use(database.Middleware(pool))
714 1 r.Handle(queryRoute, gql)
715 1 })
716 1 router.Get(apimeta.Path, apimeta.Handler(apiScopes...))
717 }
718
719 // pushNotifier is what the daemon does when a push lands.
720 //
721 // Phase 1 records it and nothing more. Reindexing and advancing the space's
722 // index rev stamp are Phase 2's, because bleve and the stamp arrive together:
723 // moving the stamp now, with no index behind it, would assert that the index
724 // is current and remove the reconciler's only way of noticing that it is not.
725 0 func pushNotifier(log *slog.Logger, svc *service.Service, index *search.Index) hooks.PushNotifier {
726 0 return func(ctx context.Context, space core.SpaceRef, updates []hooks.RefUpdate) error {
727 0 refs := make([]string, 0, len(updates))
728 0 for _, u := range updates {
729 0 refs = append(refs, u.String())
730 0 }
731
732 // A push that touches only proposal branches changes nothing the index
733 // holds: the index carries the approved revision, and proposal content
734 // is deliberately not searchable — surfacing unreviewed text in search
735 // is the same leak as serving it from the read plane.
736 0 approved := false
737 0 for _, u := range updates {
738 0 if !strings.HasPrefix(u.Ref, "refs/heads/"+core.ProposalPrefix) {
739 0 approved = true
740 0 break
741 }
742 }
743 0 if !approved {
744 0 log.Info("push landed; no reindex needed", "space", space.String(), "refs", refs)
745 0 return nil
746 0 }
747
748 0 sp, err := svc.OpenSpace(ctx, space)
749 0 if err != nil {
750 0 return fmt.Errorf("open %s to reindex: %w", space, err)
751 0 }
752 0 rev, err := svc.ResolveRev(ctx, sp, service.ApprovedRev)
753 0 if err != nil {
754 0 return fmt.Errorf("resolve the approved head of %s: %w", space, err)
755 0 }
756 0 arc, bodies, err := svc.Archive(ctx, sp, service.ApprovedRev)
757 0 if err != nil {
758 0 return fmt.Errorf("read %s at %s: %w", space, rev, err)
759 0 }
760 0 docs, err := search.Extract(arc, bodies)
761 0 if err != nil {
762 0 return fmt.Errorf("project %s for indexing: %w", space, err)
763 0 }
764 0 stats, err := index.RebuildSpace(ctx, space, docs)
765 0 if err != nil {
766 0 return fmt.Errorf("reindex %s: %w", space, err)
767 0 }
768
769 // The stamp goes last and only on success. Written earlier it would
770 // assert the index reflects a revision it does not, which is precisely
771 // the staleness the reconciler exists to detect — and it would detect
772 // nothing.
773 0 if _, err := svc.Store().SetIndexStamp(ctx, sp.ID, rev); err != nil {
774 0 return fmt.Errorf("stamp the index for %s at %s: %w", space, rev, err)
775 0 }
776
777 0 log.Info("push landed; space reindexed",
778 0 "space", space.String(), "refs", refs, "rev", rev,
779 0 "indexed", stats.Indexed, "deleted", stats.Deleted, "took", stats.Took.String())
780 0 return nil
781 }
782 }
783
784 // reconcileReporter logs the outcome of each reconciler pass. A failure is not
785 // fatal: the next pass tries again, and a daemon that exits on a transient
786 // Postgres error takes the push path down with it.
787 0 func reconcileReporter(log *slog.Logger) func(*service.ReconcileReport, error) {
788 0 return func(rep *service.ReconcileReport, err error) {
789 0 if err != nil {
790 0 log.Error("reconciler pass failed", scribe.Err(err))
791 0 return
792 0 }
793 0 attrs := []any{
794 0 "spaces", rep.Spaces,
795 0 "repaired", len(rep.Repaired),
796 0 "stale_indexes", len(rep.Reindex),
797 0 "failures", len(rep.Failures),
798 0 }
799 0 for _, r := range rep.Repaired {
800 0 log.Info("reconciler repaired divergence", "repair", fmt.Sprint(r))
801 0 }
802 0 for _, f := range rep.Failures {
803 0 log.Warn("reconciler could not repair", "failure", fmt.Sprint(f))
804 0 }
805 0 log.Info("reconciler pass complete", attrs...)
806 }
807 }
808
809 // bridgeSIGTERM makes systemd's default stop signal work.
810 //
811 // core-go's server.Run listens for SIGINT only, which is why compare.sr.ht's
812 // unit carries KillSignal=SIGINT. Rather than require that line here, a
813 // SIGTERM is turned into the SIGINT server.Run is waiting for. Both signals
814 // are registered before Run installs its own handler, so a signal arriving in
815 // the gap is caught rather than killing the process outright.
816 //
817 // After the first shutdown signal server.Run calls signal.Reset(os.Interrupt),
818 // restoring the default disposition — so a second signal terminates
819 // immediately, which is the documented behaviour and is why this keeps
820 // forwarding rather than stopping after one.
821 0 func bridgeSIGTERM(log *slog.Logger) {
822 0 sig := make(chan os.Signal, 2)
823 0 signal.Notify(sig, syscall.SIGTERM, os.Interrupt)
824 0 go func() {
825 0 for s := range sig {
826 0 if s != syscall.SIGTERM {
827 0 continue
828 }
829 0 log.Info("SIGTERM received; starting the warm shutdown core-go waits for SIGINT to begin")
830 0 if err := syscall.Kill(os.Getpid(), syscall.SIGINT); err != nil {
831 0 log.Error("could not raise SIGINT for the warm shutdown", scribe.Err(err))
832 0 }
833 }
834 }()
835 }