~bigbes/sr-ht-spec

ref: 964716696bd588e1fe9d67c5a8b9ff2cd6a08613 sr-ht-spec/cmd/specsrht/main.go -rw-r--r-- 17.5 KiB
96471669 — bigbes feat(cmd): mount the read plane, MCP and GraphQL surfaces 27 days ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
// Command specsrht is the spec.sr.ht daemon, and — invoked under another name
// — the receive hooks of every space it serves.
//
// It runs three things in one process:
//
//   - the hook RPC socket at <repos>/.specsrht/hook.sock, which the pre-receive,
//     update and post-receive hooks of every space call. Validation lives here
//     and not in the hooks because bleve is single-writer and this process holds
//     the index, and because the push path and the API must not be able to
//     disagree about what is valid.
//   - an HTTP listener on -b (default localhost:5091), serving /healthz, the
//     read UI, /mcp for agents and /query for GraphQL — one listener, so one
//     reverse-proxy route covers the service.
//   - the reconciler, at startup and then periodically, repairing the
//     divergence a killed daemon leaves between git refs, Postgres and the
//     index.
//
// # Running as a hook
//
// Every hook a space's repository runs is a symlink to this binary. When
// argv[0] names a hook the process handles that hook and exits, touching
// neither the configuration nor the database; see the hooks package. The
// explicit form `specsrht hook <name>` does the same thing by hand.
//
// # Flags
//
// Parsed by core-go's server.New:
//
//	-b addr   bind address (repeatable); default localhost:5091
//	-d        debug (verbose request logging in core-go)
//	-m addr   Prometheus metrics bind (default random port)
//	-p addr   pprof bind (default random localhost port)
//
// Configuration comes from the shared SourceHut config.ini through core-go's
// fixed search path. Every required key is checked before anything is opened,
// and all missing ones are reported together, so a misconfigured instance
// fails once at startup rather than once per restart.
//
// # Shutdown
//
// SIGINT and SIGTERM both start a warm shutdown. core-go's server.Run only
// handles SIGINT — which is why compare.sr.ht's systemd unit sets
// KillSignal=SIGINT — so this daemon installs a handler that turns a SIGTERM
// into that same SIGINT. A unit for this service therefore needs no
// KillSignal= line: the systemd default works.
package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"github.com/go-chi/chi/v5"
	chimw "github.com/go-chi/chi/v5/middleware"
	_ "github.com/lib/pq" // registers the "postgres" database/sql driver
	"github.com/vaughan0/go-ini"
	"go.bigb.es/auxilia/scribe"

	"sourcecraft.dev/bigbes/sr-ht-core/config"
	coreserver "sourcecraft.dev/bigbes/sr-ht-core/server"

	"sourcecraft.dev/bigbes/sr-ht-spec/core"
	"sourcecraft.dev/bigbes/sr-ht-spec/graph"
	"sourcecraft.dev/bigbes/sr-ht-spec/hooks"
	"sourcecraft.dev/bigbes/sr-ht-spec/mcpsrv"
	"sourcecraft.dev/bigbes/sr-ht-spec/search"
	"sourcecraft.dev/bigbes/sr-ht-spec/service"
	"sourcecraft.dev/bigbes/sr-ht-spec/web"
)

const (
	// serviceName is the SourceHut service identifier and our config section.
	// The ".sr.ht" suffix is what puts us in the nav network list.
	serviceName = "spec.sr.ht"

	// defaultBind is the address core-go binds when no -b is given.
	defaultBind = "localhost:5091"

	// version is reported in the MCP handshake so a client listing several
	// SourceHut MCP endpoints can tell which build it is talking to.
	version = "dev"

	// pingTimeout bounds the startup connectivity check against Postgres.
	pingTimeout = 10 * time.Second

	// shutdownGrace is how long the hook socket is given to finish the calls
	// already in flight once the HTTP listener has drained. A push being
	// validated at that moment finishes rather than being failed closed.
	shutdownGrace = 30 * time.Second
)

func main() {
	// Hook mode first: a hook must not read a config file, open Postgres, or
	// bind anything. It talks to the daemon over a socket and exits.
	if _, _, isHook := hooks.ModeFromArgs(os.Args); isHook {
		os.Exit(hooks.Run(hooks.Runtime{Args: os.Args}))
	}

	log := newLogger()
	slog.SetDefault(log)

	if err := run(log); err != nil {
		// Plain text, not a log record. A startup failure is read by a human
		// on a terminal, and the configuration report is deliberately several
		// lines long — a structured handler would escape it into one.
		fmt.Fprintf(os.Stderr, "spec.sr.ht did not start: %v\n", err)
		os.Exit(1)
	}
}

// newLogger builds the process logger. LOG_LEVEL raises or lowers verbosity;
// everything goes to stderr, because a hook's stdout is forwarded to the
// pushing client and this binary is both programs.
func newLogger() *slog.Logger {
	level := new(slog.LevelVar)
	level.Set(parseLevel(os.Getenv("LOG_LEVEL")))
	return slog.New(scribe.NewTintHandler(
		scribe.WithWriter(os.Stderr),
		scribe.WithLevel(level),
	))
}

func parseLevel(s string) slog.Level {
	switch strings.ToLower(strings.TrimSpace(s)) {
	case "debug":
		return slog.LevelDebug
	case "warn", "warning":
		return slog.LevelWarn
	case "error":
		return slog.LevelError
	default:
		return slog.LevelInfo
	}
}

func run(log *slog.Logger) error {
	// LoadConfig never fails on a missing file — it returns a nil ini.File —
	// so validateConfig is what turns an unconfigured instance into one clear
	// message instead of a panic deep inside the first request.
	conf := config.LoadConfig()
	cfg, err := validateConfig(conf)
	if err != nil {
		return err
	}

	pool, err := openDatabase(cfg.ConnectionString)
	if err != nil {
		return err
	}
	defer pool.Close()

	svc, err := service.New(cfg, pool)
	if err != nil {
		return err
	}

	// Refresh every space's hooks before anything can be pushed to it. This is
	// fatal on failure by design: a space whose hooks are missing accepts
	// pushes that are never validated, which is the one outcome the whole
	// receive path exists to prevent. A daemon that will not start is loud; a
	// space quietly accepting malformed documents is not.
	binary, err := os.Executable()
	if err != nil {
		return fmt.Errorf("locate this binary, which every hook symlinks to: %w", err)
	}
	if err := refreshHooks(context.Background(), log, svc, binary); err != nil {
		return err
	}

	hookSrv, err := hooks.NewServer(hooks.Options{
		Backend: svc,
		Socket:  hooks.SocketPath(cfg.Repos),
		Log:     log,
		OnPush:  pushNotifier(log),
	})
	if err != nil {
		return err
	}
	if err := hookSrv.Listen(); err != nil {
		return err
	}

	surf, err := newSurfaces(conf, cfg, svc, version)
	if err != nil {
		return err
	}
	defer surf.Close()

	// server.New parses -b/-d/-m/-p and runs crypto.InitCrypto(conf), whose
	// two required keys validateConfig already checked, so it cannot fatal
	// here for a reason we have not already reported.
	srv := coreserver.New(serviceName, defaultBind, conf, os.Args)
	mountRoutes(srv.AnonRouter(), conf, surf)

	ctx, stop := context.WithCancel(context.Background())
	defer stop()

	served := make(chan error, 1)
	go func() { served <- hookSrv.Serve(ctx) }()
	go svc.RunReconciler(ctx, service.DefaultReconcileInterval, reconcileReporter(log))

	bridgeSIGTERM(log)

	log.Info("spec.sr.ht starting",
		"bind", defaultBind,
		"repos", cfg.Repos,
		"cache", cfg.Cache,
		"origin", cfg.Origin,
		"hook_socket", hookSrv.Socket(),
		"reconcile_interval", service.DefaultReconcileInterval.String(),
	)

	// Blocks until SIGINT — which bridgeSIGTERM makes SIGTERM equivalent to —
	// and then drains the HTTP listeners.
	srv.Run()

	log.Info("draining the hook socket", "grace", shutdownGrace.String())
	stop()
	select {
	case err := <-served:
		if err != nil {
			log.Error("hook socket stopped with an error", scribe.Err(err))
		}
	case <-time.After(shutdownGrace):
		log.Warn("hook socket did not drain in time; closing it")
	}
	if err := hookSrv.Close(); err != nil {
		log.Error("could not close the hook socket", scribe.Err(err))
	}
	log.Info("spec.sr.ht stopped")
	return nil
}

// validateConfig checks every key this daemon needs before anything is opened,
// and reports all of the missing ones at once so an operator fixes the config
// in one pass instead of discovering each gap on a separate restart.
//
// service.LoadConfig owns our own section and collects its own gaps the same
// way; the two lists are merged into one message. The keys checked here are
// the ones core-go itself fatals on, which belong to server.New's contract
// rather than to service/ — duplicating them there would give the instance two
// lists to keep in sync.
func validateConfig(conf ini.File) (service.Config, error) {
	var missing []string
	require := func(section, key, why string) {
		if v, ok := conf.Get(section, key); !ok || strings.TrimSpace(v) == "" {
			missing = append(missing, fmt.Sprintf("[%s] %s — %s", section, key, why))
		}
	}

	// Both are read by crypto.InitCrypto, which server.New calls and which
	// fatals with a terse message when either is absent. The webhook key is
	// required even though v1 emits no webhooks.
	require("sr.ht", "network-key", "fernet key for the unified-login cookie")
	require("webhooks", "private-key", "webhook signing key; crypto.InitCrypto requires it")

	cfg, cfgErr := service.LoadConfig(conf)

	if len(missing) == 0 && cfgErr == nil {
		return cfg, nil
	}

	var b strings.Builder
	b.WriteString("incomplete configuration.")
	if len(missing) > 0 {
		fmt.Fprintf(&b, "\n\nMissing keys the SourceHut runtime requires:\n\t%s",
			strings.Join(missing, "\n\t"))
	}
	if cfgErr != nil {
		fmt.Fprintf(&b, "\n\n%s", cfgErr)
	}
	return service.Config{}, errors.New(b.String())
}

// openDatabase opens the pool the whole daemon shares — request handlers, the
// reconciler and the hook RPC alike — and proves it works before serving.
//
// sql.Open alone connects lazily, so a wrong DSN would first surface as a
// rejected push. Fail-closed makes that safe but not pleasant; failing at
// startup names the problem while somebody is still watching.
func openDatabase(dsn string) (*sql.DB, error) {
	pool, err := sql.Open("postgres", dsn)
	if err != nil {
		return nil, fmt.Errorf("open the database: %w", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), pingTimeout)
	defer cancel()
	if err := pool.PingContext(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("reach the database: %w", err)
	}
	return pool, nil
}

// refreshHooks installs this binary's hooks into every space.
//
// It runs at startup rather than only at space creation so that an upgrade
// which changes the wire protocol, the socket path or the hook set repairs
// every repository by restarting — there is no separate migration step and no
// repository left speaking last week's protocol.
func refreshHooks(ctx context.Context, log *slog.Logger, svc *service.Service, binary string) error {
	spaces, err := svc.ListSpaces(ctx)
	if err != nil {
		return fmt.Errorf("list spaces to refresh their hooks: %w", err)
	}
	for _, sp := range spaces {
		if err := hooks.InstallSpace(svc.ReposRoot(), sp.Ref, hooks.InstallOptions{Binary: binary}); err != nil {
			return fmt.Errorf("install the receive hooks of %s: %w "+
				"(a space whose hooks are missing would accept unvalidated pushes, so this is fatal; "+
				"repair or remove the repository and start again)", sp.Ref, err)
		}
	}
	log.Info("receive hooks refreshed", "spaces", len(spaces), "binary", binary)
	return nil
}

// mountRoutes installs what the daemon serves over HTTP.
func mountRoutes(router chi.Router, conf ini.File, surfaces *surfaces) {
	// server.New already froze the anonymous router for direct middleware
	// registration, so middleware and routes go in together inside a Group —
	// which chi permits on a fresh inline mux sharing the same routing tree.
	router.Group(func(r chi.Router) {
		r.Use(chimw.RealIP)
		r.Use(chimw.Recoverer)
		r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) {
			w.Header().Set("Content-Type", "text/plain; charset=utf-8")
			fmt.Fprint(w, "ok")
		})
	})

	mountWeb(router, conf, surfaces)
}

// surfaces are the three Phase 2 read surfaces, assembled once at startup.
//
// They share one *search.Index deliberately: bleve is single-writer, so a second
// Open on the same directory is not merely wasteful but wrong.
type surfaces struct {
	index *search.Index
	web   *web.Server
	mcp   http.Handler
	gql   http.Handler
}

// newSurfaces opens the index and builds the three read surfaces over it.
//
// All three go through service/ and none of them re-derives addressing, the
// read contract or the project filter — that shared layer is the whole reason
// "read SPEC-0007" cannot mean three different things depending on which door
// you knock on.
func newSurfaces(conf ini.File, cfg service.Config, svc *service.Service, version string) (*surfaces, error) {
	indexPath := filepath.Join(cfg.Cache, "index")
	// A crashed rebuild leaves working directories beside the index; clearing
	// them before Open is always safe, since the index is a pure cache.
	if err := search.CleanStale(indexPath); err != nil {
		return nil, fmt.Errorf("clean stale index working dirs: %w", err)
	}
	index, err := search.Open(indexPath)
	if err != nil {
		return nil, fmt.Errorf("open search index at %s: %w", indexPath, err)
	}

	site, err := web.New(web.Options{
		Conf:     conf,
		Reader:   web.NewReader(svc),
		Searcher: index,
		Resolver: svc.Resolver(),
	})
	if err != nil {
		index.Close()
		return nil, fmt.Errorf("assemble the web UI: %w", err)
	}

	// The origin is the /mcp Host allowlist: the MCP SDK's own DNS-rebinding
	// guard cannot tell a reverse proxy from an attacker (both reach a loopback
	// listener with a non-loopback Host), so it is replaced by a check against
	// this value. Traefik must pass the Host header through or every call 403s.
	mcp, err := mcpsrv.Handler(mcpsrv.Backend{Docs: svc, Index: index}, version, cfg.Origin)
	if err != nil {
		index.Close()
		return nil, fmt.Errorf("assemble the MCP surface: %w", err)
	}

	gql, err := graph.New(graph.Options{
		Reader:   svc,
		Searcher: index,
		Resolver: svc.Resolver(),
	})
	if err != nil {
		index.Close()
		return nil, fmt.Errorf("assemble the GraphQL surface: %w", err)
	}

	return &surfaces{index: index, web: site, mcp: mcp, gql: gql.Handler()}, nil
}

func (s *surfaces) Close() error {
	if s == nil || s.index == nil {
		return nil
	}
	return s.index.Close()
}

// mountWeb attaches the three read surfaces.
//
// Order is load-bearing: /mcp and /query are registered before the web UI,
// which mounts at "/" and would otherwise swallow them as document paths —
// "spec" and "query" are legal space names as far as the router is concerned.
//
// Each surface installs its own authentication (they are fail-closed and agree
// on one ACL), so core-go's WithDefaultMiddleware is deliberately not used: its
// auth middleware 401s any un-cookied request, which would also block the agent
// bearer-token path these surfaces exist to serve.
func mountWeb(router chi.Router, _ ini.File, s *surfaces) {
	if s == nil {
		return
	}
	router.Handle("/mcp", s.mcp)
	router.Handle("/query", s.gql)
	router.Mount("/", s.web.Handler())
}

// pushNotifier is what the daemon does when a push lands.
//
// Phase 1 records it and nothing more. Reindexing and advancing the space's
// index rev stamp are Phase 2's, because bleve and the stamp arrive together:
// moving the stamp now, with no index behind it, would assert that the index
// is current and remove the reconciler's only way of noticing that it is not.
func pushNotifier(log *slog.Logger) hooks.PushNotifier {
	return func(_ context.Context, space core.SpaceRef, updates []hooks.RefUpdate) error {
		refs := make([]string, 0, len(updates))
		for _, u := range updates {
			refs = append(refs, u.String())
		}
		log.Info("push landed; reindex pending",
			"space", space.String(),
			"refs", refs,
			"note", "indexing lands in Phase 2; the reconciler reports the staleness until then")
		return nil
	}
}

// reconcileReporter logs the outcome of each reconciler pass. A failure is not
// fatal: the next pass tries again, and a daemon that exits on a transient
// Postgres error takes the push path down with it.
func reconcileReporter(log *slog.Logger) func(*service.ReconcileReport, error) {
	return func(rep *service.ReconcileReport, err error) {
		if err != nil {
			log.Error("reconciler pass failed", scribe.Err(err))
			return
		}
		attrs := []any{
			"spaces", rep.Spaces,
			"repaired", len(rep.Repaired),
			"stale_indexes", len(rep.Reindex),
			"failures", len(rep.Failures),
		}
		for _, r := range rep.Repaired {
			log.Info("reconciler repaired divergence", "repair", fmt.Sprint(r))
		}
		for _, f := range rep.Failures {
			log.Warn("reconciler could not repair", "failure", fmt.Sprint(f))
		}
		log.Info("reconciler pass complete", attrs...)
	}
}

// bridgeSIGTERM makes systemd's default stop signal work.
//
// core-go's server.Run listens for SIGINT only, which is why compare.sr.ht's
// unit carries KillSignal=SIGINT. Rather than require that line here, a
// SIGTERM is turned into the SIGINT server.Run is waiting for. Both signals
// are registered before Run installs its own handler, so a signal arriving in
// the gap is caught rather than killing the process outright.
//
// After the first shutdown signal server.Run calls signal.Reset(os.Interrupt),
// restoring the default disposition — so a second signal terminates
// immediately, which is the documented behaviour and is why this keeps
// forwarding rather than stopping after one.
func bridgeSIGTERM(log *slog.Logger) {
	sig := make(chan os.Signal, 2)
	signal.Notify(sig, syscall.SIGTERM, os.Interrupt)
	go func() {
		for s := range sig {
			if s != syscall.SIGTERM {
				continue
			}
			log.Info("SIGTERM received; starting the warm shutdown core-go waits for SIGINT to begin")
			if err := syscall.Kill(os.Getpid(), syscall.SIGINT); err != nil {
				log.Error("could not raise SIGINT for the warm shutdown", scribe.Err(err))
			}
		}
	}()
}