@@ -20,6 +20,8 @@ import (
2020
2121 "net/http"
2222
23+ "riverqueue.com/riverui"
24+
2325 "github.com/dennisme/grex/internal/api"
2426 "github.com/dennisme/grex/internal/buildinfo"
2527 "github.com/dennisme/grex/internal/config"
@@ -31,6 +33,33 @@ import (
3133 "github.com/dennisme/grex/internal/ui"
3234)
3335
36+ // mountRiverUI wraps River's own job/queue UI (riverqueue.com/riverui) at
37+ // /riverui, reusing the same River client the purge job already runs
38+ // (persistence.NewPurgeClient) — no second River client. No riverui-
39+ // specific auth added: it mounts on uiMux, the same mux server.New wraps
40+ // once with mTLS + SPIFFE role mapping when ui_tls.client_ca_file is
41+ // configured (see docs/admin/authentication.md), so this route inherits
42+ // that access control for free. That's a different axis from the
43+ // browser-login OIDC flow that's still unbuilt (issue #11) — this isn't
44+ // blocked on that, and doesn't need riverui's own
45+ // RIVER_BASIC_AUTH_USER/PASS mechanism, which would be a second,
46+ // inconsistent auth story to rip out later. Caller must still call
47+ // Start(ctx) on the returned handler before it's fully functional (caching
48+ // and background query support), same two-step shape as riverui's own
49+ // example.
50+ func mountRiverUI (uiMux * http.ServeMux , purgeClient * river.Client [pgx.Tx ], logger * slog.Logger ) (* riverui.Handler , error ) {
51+ handler , err := riverui .NewHandler (& riverui.HandlerOpts {
52+ Logger : logger ,
53+ Prefix : "/riverui" ,
54+ Endpoints : riverui .NewEndpoints (purgeClient , nil ),
55+ })
56+ if err != nil {
57+ return nil , fmt .Errorf ("river ui: %w" , err )
58+ }
59+ uiMux .Handle ("/riverui/" , handler )
60+ return handler , nil
61+ }
62+
3463const shutdownGrace = 10 * time .Second
3564
3665// persistenceFlushInterval is how often dirty agents are saved to the
@@ -107,6 +136,7 @@ func run(args []string) error {
107136 var dirtyTracker * persistence.DirtyTracker
108137 var store persistence.StateStore
109138 var maxConcurrentWrites int
139+ var purgeClient * river.Client [pgx.Tx ]
110140 if cfg .Database .Host != "" {
111141 dsn := fmt .Sprintf ("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s" ,
112142 cfg .Database .Host , cfg .Database .Port , cfg .Database .User , cfg .Database .Password ,
@@ -127,6 +157,16 @@ func run(args []string) error {
127157 // persistence.PoolCollector's doc comment on why this defaults to the
128158 // grex process's own CPU count, not the database's actual capacity).
129159 maxConcurrentWrites = int (pool .Config ().MaxConns )
160+
161+ // Constructed here (not deferred to the goroutine-launching block
162+ // below) so mountRiverUI can mount its handler on uiMux before
163+ // srv.Start() begins serving it — registering a new route on a mux
164+ // already being served concurrently is a race. Only Start(ctx),
165+ // which needs the shutdown context, waits until after srv.Start().
166+ purgeClient , err = persistence .NewPurgeClient (pool , cfg .Fleet .SoftDeleteDuration , events , logger )
167+ if err != nil {
168+ return fmt .Errorf ("purge client: %w" , err )
169+ }
130170 }
131171
132172 registry := fleet .New (fleet.Config {
@@ -150,6 +190,14 @@ func run(args []string) error {
150190 }
151191 uiHandler .Mount (uiMux )
152192
193+ var riverUIHandler * riverui.Handler
194+ if dbPool != nil {
195+ riverUIHandler , err = mountRiverUI (uiMux , purgeClient , logger )
196+ if err != nil {
197+ return err
198+ }
199+ }
200+
153201 srv := server .New (cfg , logger ,
154202 server.OpAMP {Handler : handler , ConnContext : connCtx },
155203 server.UI {Handler : uiMux },
@@ -162,21 +210,19 @@ func run(args []string) error {
162210 ctx , stop := signal .NotifyContext (context .Background (), syscall .SIGINT , syscall .SIGTERM )
163211 defer stop ()
164212 go registry .Run (ctx )
165- var purgeClient * river.Client [pgx.Tx ]
166213 if dbPool != nil {
167214 flusher := persistence .NewFlusher (registry , dirtyTracker , store , persistenceFlushInterval , logger , maxConcurrentWrites , events )
168215 go flusher .Run (ctx )
169216
170217 snapshotter := persistence .NewSessionSnapshotter (registry , store , sessionSnapshotInterval , logger , maxConcurrentWrites , events )
171218 go snapshotter .Run (ctx )
172219
173- purgeClient , err = persistence .NewPurgeClient (dbPool , cfg .Fleet .SoftDeleteDuration , events , logger )
174- if err != nil {
175- return fmt .Errorf ("purge client: %w" , err )
176- }
177220 if err := purgeClient .Start (ctx ); err != nil {
178221 return fmt .Errorf ("start purge client: %w" , err )
179222 }
223+ if err := riverUIHandler .Start (ctx ); err != nil {
224+ return fmt .Errorf ("start river ui: %w" , err )
225+ }
180226 defer func () {
181227 stopCtx , cancel := context .WithTimeout (context .Background (), shutdownGrace )
182228 defer cancel ()
0 commit comments