signals: use job queue to handle processing

app/app.go

@@ -137,6 +137,7 @@ type App struct {
 	APIKeyStore   *apikey.Store
 	River         *river.Client[pgx.Tx]
 	RiverUI       *riverui.Server
+	RiverWorkers  *river.Workers
 }
 
 // NewApp constructs a new App and binds the listening socket.

app/initengine.go

@@ -53,6 +53,7 @@ func (app *App) initEngine(ctx context.Context) error {
 		DisableCycle: app.cfg.APIOnly,
 		LogCycles:    app.cfg.LogEngine,
 		River:        app.River,
+		RiverWorkers: app.RiverWorkers,
 	})
 	if err != nil {
 		return errors.Wrap(err, "init engine")

app/initriver.go

@@ -38,13 +38,25 @@ func (r *riverErrs) HandlePanic(ctx context.Context, job *rivertype.JobRow, pani
 	return nil
 }
 
+type noopWorker struct{}
+
+func (noopWorker) Kind() string { return "noop" }
+
 func (app *App) initRiver(ctx context.Context) error {
-	w := river.NewWorkers()
+	app.RiverWorkers = river.NewWorkers()
+
+	// TODO: remove once a worker is added that's not behind a feature flag
+	//
+	// Without this, it will complain about no workers being registered.
+	river.AddWorker(app.RiverWorkers, river.WorkFunc(func(ctx context.Context, j *river.Job[noopWorker]) error {
+		// Do something with the job
+		return nil
+	}))
 
 	var err error
 	app.River, err = river.NewClient(riverpgxv5.New(app.pgx), &river.Config{
 		Logger:  app.Logger,
-		Workers: w,
+		Workers: app.RiverWorkers,
 		Queues: map[string]river.QueueConfig{
 			river.QueueDefault: {MaxWorkers: 100},
 		},

app/initstores.go

@@ -304,7 +304,7 @@ func (app *App) initStores(ctx context.Context) error {
 		return errors.Wrap(err, "init API key store")
 	}
 
-	app.UIKHandler = uik.NewHandler(app.db, app.IntegrationKeyStore, app.AlertStore)
+	app.UIKHandler = uik.NewHandler(app.db, app.IntegrationKeyStore, app.AlertStore, app.River)
 
 	return nil
 }

app/startup.go

@@ -58,6 +58,7 @@ func (app *App) startup(ctx context.Context) error {
 		return err
 	})
 
+	app.initStartup(ctx, "Startup.River", app.initRiver)
 	app.initStartup(ctx, "Startup.DBStores", app.initStores)
 	ctx = app.ConfigStore.Config().Context(ctx)
 
@@ -67,7 +68,6 @@ func (app *App) startup(ctx context.Context) error {
 
 	app.initStartup(ctx, "Startup.Slack", app.initSlack)
 
-	app.initStartup(ctx, "Startup.River", app.initRiver)
 	app.initStartup(ctx, "Startup.Engine", app.initEngine)
 	app.initStartup(ctx, "Startup.Auth", app.initAuth)
 	app.initStartup(ctx, "Startup.GraphQL", app.initGraphQL)

engine/config.go

@@ -35,6 +35,7 @@ type Config struct {
 	SlackStore          *slack.ChannelSender
 	DestRegistry        *nfydest.Registry
 	River               *river.Client[pgx.Tx]
+	RiverWorkers        *river.Workers
 
 	ConfigSource config.Source
 

engine/engine.go

@@ -33,11 +33,6 @@ import (
 	"github.com/target/goalert/util/sqlutil"
 )
 
-type updater interface {
-	Name() string
-	UpdateAll(context.Context) error
-}
-
 // Engine handles automatic escalation of unacknowledged(triggered) alerts, as well as
 // passing to-be-sent notifications to the notification.Sender.
 //
@@ -53,12 +48,15 @@ type Engine struct {
 	triggerCh   chan struct{}
 	runLoopExit chan struct{}
 
-	modules []updater
+	modules []processinglock.Module
 	msg     *message.DB
 
 	a   *alert.Store
 	cfg *Config
 
+	// needed to re-start river during switchover
+	runCtx context.Context
+
 	triggerPauseCh chan *pauseReq
 }
 
@@ -137,7 +135,7 @@ func NewEngine(ctx context.Context, db *sql.DB, c *Config) (*Engine, error) {
 		return nil, errors.Wrap(err, "compatibility backend")
 	}
 
-	p.modules = []updater{
+	p.modules = []processinglock.Module{
 		compatMgr,
 		rotMgr,
 		schedMgr,
@@ -168,6 +166,21 @@ func NewEngine(ctx context.Context, db *sql.DB, c *Config) (*Engine, error) {
 		return nil, errors.Wrap(err, "init backend")
 	}
 
+	args := processinglock.SetupArgs{
+		DB:           db,
+		River:        c.River,
+		Workers:      c.RiverWorkers,
+		ConfigSource: c.ConfigSource,
+	}
+	for _, m := range p.modules {
+		if s, ok := m.(processinglock.Setupable); ok {
+			err = s.Setup(ctx, args)
+			if err != nil {
+				return nil, errors.Wrap(err, "setup module")
+			}
+		}
+	}
+
 	return p, nil
 }
 
@@ -178,7 +191,7 @@ func (p *Engine) AuthLinkURL(ctx context.Context, providerID, subjectID string,
 	return url, err
 }
 
-func (p *Engine) processModule(ctx context.Context, m updater) {
+func (p *Engine) processModule(ctx context.Context, m processinglock.Updatable) {
 	defer recoverPanic(ctx, m.Name())
 	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
@@ -241,6 +254,11 @@ func (p *Engine) Pause(ctx context.Context) error {
 func (p *Engine) _pause(ctx context.Context) error {
 	ch := make(chan error, 1)
 
+	err := p.cfg.River.Stop(ctx)
+	if err != nil {
+		return errors.Wrap(err, "pause river")
+	}
+
 	select {
 	case <-p.shutdownCh:
 		return errors.New("shutting down")
@@ -263,8 +281,17 @@ func (p *Engine) Resume(ctx context.Context) error {
 }
 
 func (p *Engine) _resume(ctx context.Context) error {
-	// nothing to be done `p.mgr.IsPaused` will already
-	// return false
+	if p.cfg.DisableCycle {
+		return nil
+	}
+
+	err := p.cfg.River.QueueResume(ctx, "*", nil)
+	if err != nil {
+		return errors.Wrap(err, "resume river")
+	}
+
+	go p.startRiver()
+
 	return nil
 }
 
@@ -284,9 +311,9 @@ func (p *Engine) Shutdown(ctx context.Context) error {
 
 func (p *Engine) _shutdown(ctx context.Context) (err error) {
 	close(p.shutdownCh)
-	// if !p.cfg.DisableCycle {
-	// 	err = p.cfg.River.Stop(ctx)
-	// }
+	if !p.cfg.DisableCycle {
+		err = p.cfg.River.Stop(ctx)
+	}
 	<-p.runLoopExit
 	return err
 }
@@ -454,8 +481,13 @@ func (p *Engine) processAll(ctx context.Context) bool {
 			return true
 		}
 
+		up, ok := m.(processinglock.Updatable)
+		if !ok {
+			continue
+		}
+
 		start := time.Now()
-		p.processModule(ctx, m)
+		p.processModule(ctx, up)
 		metricModuleDuration.WithLabelValues(m.Name()).Observe(time.Since(start).Seconds())
 	}
 	return false
@@ -528,6 +560,14 @@ func (p *Engine) handlePause(ctx context.Context, respCh chan error) {
 	respCh <- nil
 }
 
+func (p *Engine) startRiver() {
+	ctx := p.runCtx
+	err := p.cfg.River.Start(ctx)
+	if err != nil {
+		log.Log(ctx, errors.Wrap(err, "start river"))
+	}
+}
+
 func (p *Engine) _run(ctx context.Context) error {
 	defer close(p.runLoopExit)
 	ctx = permission.SystemContext(ctx, "Engine")
@@ -549,13 +589,8 @@ func (p *Engine) _run(ctx context.Context) error {
 		}
 	}
 
-	// Disabled until we have our first worker (next PR)
-	// go func() {
-	// 	err := p.cfg.River.Start(ctx)
-	// 	if err != nil {
-	// 		log.Log(ctx, errors.Wrap(err, "start river"))
-	// 	}
-	// }()
+	p.runCtx = ctx
+	go p.startRiver()
 
 	dur := p.cfg.CycleTime
 	if dur == 0 {

engine/processinglock/module.go

@@ -0,0 +1,35 @@
+package processinglock
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/riverqueue/river"
+	"github.com/target/goalert/config"
+)
+
+// Module is a processing lock module.
+type Module interface {
+	Name() string
+}
+
+// Updatable is an interface for types that can be updated.
+type Updatable interface {
+	Module
+	UpdateAll(context.Context) error
+}
+
+// SetupArgs is a struct that contains the arguments for the setup function.
+type SetupArgs struct {
+	DB           *sql.DB
+	River        *river.Client[pgx.Tx]
+	Workers      *river.Workers
+	ConfigSource config.Source
+}
+
+// Setupable is an interface for types that can be set up using the job queue system.
+type Setupable interface {
+	Module
+	Setup(context.Context, SetupArgs) error
+}

engine/signalmgr/queries.sql

@@ -8,6 +8,8 @@ FROM
     pending_signals
 WHERE
     message_id IS NULL
+    AND (sqlc.narg(service_id)::uuid IS NULL
+        OR service_id = @service_id)
 FOR UPDATE
     SKIP LOCKED
 LIMIT 100;
@@ -32,3 +34,19 @@ DELETE FROM pending_signals
 WHERE message_id IS NULL
     AND created_at < NOW() - INTERVAL '1 hour';
 
+-- name: SignalMgrGetScheduled :many
+SELECT
+    count(*),
+    service_id,
+    channel_id
+FROM
+    outgoing_messages
+WHERE
+    message_type = 'signal_message'
+    AND last_status = 'pending'
+    AND (sqlc.narg(service_id)::uuid IS NULL
+        OR service_id = @service_id)
+GROUP BY
+    service_id,
+    channel_id;
+

engine/signalmgr/schedulemessages.go

@@ -0,0 +1,81 @@
+package signalmgr
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/riverqueue/river"
+	"github.com/target/goalert/gadb"
+)
+
+func (db *DB) scheduleMessages(ctx context.Context, serviceID uuid.NullUUID) error {
+	var didWork bool
+	err := db.lock.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
+		q := gadb.New(tx)
+
+		messages, err := q.SignalMgrGetPending(ctx, serviceID)
+		if err != nil {
+			return fmt.Errorf("get pending signals: %w", err)
+		}
+
+		type dest struct {
+			ServiceID uuid.UUID
+			ChannelID uuid.UUID
+		}
+
+		res, err := q.SignalMgrGetScheduled(ctx, serviceID)
+		if err != nil {
+			return fmt.Errorf("get scheduled signals: %w", err)
+		}
+		alreadyScheduled := make(map[dest]struct{}, len(res))
+		for _, r := range res {
+			alreadyScheduled[dest{ServiceID: r.ServiceID.UUID, ChannelID: r.ChannelID.UUID}] = struct{}{}
+		}
+
+		for _, m := range messages {
+			if _, ok := alreadyScheduled[dest{ServiceID: m.ServiceID, ChannelID: m.DestID}]; ok {
+				// Only allow one message per destination, per service, to be scheduled at a time.
+				continue
+			}
+			didWork = true
+			alreadyScheduled[dest{ServiceID: m.ServiceID, ChannelID: m.DestID}] = struct{}{}
+			id := uuid.New()
+			err = q.SignalMgrInsertMessage(ctx, gadb.SignalMgrInsertMessageParams{
+				ID:        id,
+				ServiceID: uuid.NullUUID{Valid: true, UUID: m.ServiceID},
+				ChannelID: uuid.NullUUID{Valid: true, UUID: m.DestID},
+			})
+			if err != nil {
+				return fmt.Errorf("insert message: %w", err)
+			}
+
+			err = q.SignalMgrUpdateSignal(ctx, gadb.SignalMgrUpdateSignalParams{
+				ID:        m.ID,
+				MessageID: uuid.NullUUID{Valid: true, UUID: id},
+			})
+			if err != nil {
+				return fmt.Errorf("update signal: %w", err)
+			}
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	if serviceID.Valid {
+		// only try once for per-service/on-demand updates
+		return nil
+	}
+
+	if didWork {
+		// for global update, we want to keep checking for work until nothing is left.
+		return river.JobSnooze(time.Second)
+	}
+
+	return nil
+}