Wire the store, lease table and limiter into crossbar

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

cmd/crossbar/main.go

@@ -17,7 +17,10 @@ import (
 	"git.wntrmute.dev/kyle/crossbar/internal/admin"
 	"git.wntrmute.dev/kyle/crossbar/internal/config"
 	"git.wntrmute.dev/kyle/crossbar/internal/health"
+	"git.wntrmute.dev/kyle/crossbar/internal/lease"
+	"git.wntrmute.dev/kyle/crossbar/internal/limiter"
 	"git.wntrmute.dev/kyle/crossbar/internal/proxy"
+	"git.wntrmute.dev/kyle/crossbar/internal/store"
 )
 
 func main() {
@@ -38,6 +41,12 @@ func run() error {
 
 	log := slog.New(slog.NewTextHandler(os.Stderr, nil))
 
+	st, err := store.Open(cfg.DB)
+	if err != nil {
+		return err
+	}
+	defer st.Close()
+
 	baseURLs := make(map[string]string, len(cfg.Hosts))
 	for name, host := range cfg.Hosts {
 		baseURLs[name] = host.BaseURL
@@ -48,9 +57,79 @@ func run() error {
 	defer stop()
 	go table.Run(ctx)
 
+	hosts := proxy.HostView(table, cfg)
+	lim := limiter.New()
+	for name, h := range cfg.Hosts {
+		for model, m := range h.Models {
+			lim.Configure(name, model, m.Parallel, cfg.QueueMax)
+		}
+	}
+
+	leases, err := lease.New(st, hosts, proxy.Chooser(cfg, table, lim), cfg.LeaseIdle.Duration)
+	if err != nil {
+		return err
+	}
+	for name, rt := range cfg.Routes {
+		leases.Candidates(name, rt.Hosts)
+	}
+
 	mux := http.NewServeMux()
-	mux.Handle("/_crossbar/", admin.Handler(cfg, table, nil, nil, nil, nil))
-	mux.Handle("/", proxy.New(cfg, table, nil, nil, nil, log))
+	mux.Handle("/_crossbar/", admin.Handler(cfg, table, leases, lim, st, hosts))
+	mux.Handle("/", proxy.New(cfg, table, leases, lim, st, log))
+
+	// Background maintenance until ctx is done. Errors are logged, never fatal.
+	go func() {
+		ticker := time.NewTicker(time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				leases.ExpireIdle(time.Now())
+			}
+		}
+	}()
+
+	go func() {
+		ticker := time.NewTicker(time.Hour)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				n, err := st.Prune(time.Now(), cfg.Retention.Duration)
+				if err != nil {
+					log.Error("prune", "err", err)
+					continue
+				}
+				log.Info("pruned request rows", "rows", n)
+			}
+		}
+	}()
+
+	go func() {
+		ticker := time.NewTicker(cfg.PollInterval.Duration)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				for name, s := range table.All() {
+					if err := st.RecordHostHealth(store.HostHealth{
+						TS:      time.Now(),
+						Host:    name,
+						Healthy: s.Healthy,
+						Loaded:  s.Loaded,
+					}); err != nil {
+						log.Error("record host health", "host", name, "err", err)
+					}
+				}
+			}
+		}
+	}()
 
 	srv := &http.Server{
 		Addr:              cfg.Listen,