feat: run delivery planning from the trigger job

backend/internal/notification/outbox/runtime.go

@@ -25,9 +25,12 @@ type Runtime struct {
 // *pgxpool.Pool backing Storage — Storage's own PgxPool interface exists for
 // pgxmock compatibility and is not sufficient here, since riverpgxv5.New
 // requires the concrete pgxpool type for its own maintenance queries).
-func NewRuntime(pool *pgxpool.Pool, adapter ChannelAdapter, store deliveryWorkerStore, metrics *Metrics) (*Runtime, error) {
+func NewRuntime(pool *pgxpool.Pool, adapter ChannelAdapter, store deliveryWorkerStore, metrics *Metrics, register ...func(*river.Workers)) (*Runtime, error) {
 	workers := river.NewWorkers()
 	river.AddWorker(workers, NewDeliveryWorker(store, adapter, metrics))
+	for _, add := range register {
+		add(workers)
+	}
 
 	client, err := river.NewClient(riverpgxv5.New(pool), &river.Config{
 		Queues: map[string]river.QueueConfig{
@@ -43,6 +46,17 @@ func NewRuntime(pool *pgxpool.Pool, adapter ChannelAdapter, store deliveryWorker
 	return &Runtime{Client: client}, nil
 }
 
+// NewInserter builds an insert-only River client: no queues and no workers,
+// so it never claims a job. With no worker bundle River does not check that an
+// inserted kind has a worker, so it can queue any job kind.
+func NewInserter(pool *pgxpool.Pool) (*river.Client[pgx.Tx], error) {
+	client, err := river.NewClient(riverpgxv5.New(pool), &river.Config{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to construct river inserter: %w", err)
+	}
+	return client, nil
+}
+
 func (r *Runtime) Start(ctx context.Context) error {
 	if r == nil || r.Client == nil {
 		return nil

backend/internal/notification/outbox/runtime_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/riverqueue/river"
 	"github.com/stretchr/testify/require"
 
 	"github.com/trakrf/platform/backend/internal/notification/outbox"
@@ -26,3 +27,17 @@ func TestRuntime_StartStop_NoPanicNoLeak(t *testing.T) {
 	require.NoError(t, runtime.Start(ctx))
 	require.NoError(t, runtime.Stop(ctx))
 }
+
+func TestNewRuntime_RegistersExtraWorkers(t *testing.T) {
+	db := testutil.SetupTestDBFull(t)
+	metrics, err := outbox.NewMetrics(prometheus.NewRegistry())
+	require.NoError(t, err)
+
+	calls := 0
+	_, err = outbox.NewRuntime(db.AppPool, fakeAdapter{}, db.Store, metrics, func(w *river.Workers) {
+		calls++
+		require.NotNil(t, w)
+	})
+	require.NoError(t, err)
+	require.Equal(t, 1, calls)
+}

backend/internal/notification/routing/worker.go

@@ -0,0 +1,46 @@
+package routing
+
+import (
+	"context"
+	"errors"
+
+	"github.com/riverqueue/river"
+)
+
+// TriggerPlanner plans the deliveries for one trigger. *DeliveryPlanner
+// satisfies it.
+type TriggerPlanner interface {
+	Plan(ctx context.Context, t Trigger) (Result, error)
+}
+
+var _ TriggerPlanner = (*DeliveryPlanner)(nil)
+
+// TriggerWorker runs the planner for each queued trigger job.
+type TriggerWorker struct {
+	river.WorkerDefaults[TriggerJobArgs]
+	planner TriggerPlanner
+}
+
+func NewTriggerWorker(p TriggerPlanner) *TriggerWorker {
+	return &TriggerWorker{planner: p}
+}
+
+// Work cancels a job whose trigger can never be planned and returns any other
+// error so River retries it.
+func (w *TriggerWorker) Work(ctx context.Context, job *river.Job[TriggerJobArgs]) error {
+	if _, err := w.planner.Plan(ctx, job.Args.Trigger()); err != nil {
+		if errors.Is(err, ErrInvalidTrigger) {
+			return river.JobCancel(err)
+		}
+		return err
+	}
+	return nil
+}
+
+// RegisterTriggerWorker returns a hook for outbox.NewRuntime that adds the
+// trigger worker to the runtime's workers.
+func RegisterTriggerWorker(p TriggerPlanner) func(*river.Workers) {
+	return func(w *river.Workers) {
+		river.AddWorker(w, NewTriggerWorker(p))
+	}
+}

backend/internal/notification/routing/worker_integration_test.go

@@ -0,0 +1,139 @@
+//go:build integration
+
+package routing_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	"github.com/trakrf/platform/backend/internal/models/notificationevent"
+	"github.com/trakrf/platform/backend/internal/models/notificationrecipient"
+	"github.com/trakrf/platform/backend/internal/models/scandevice"
+	"github.com/trakrf/platform/backend/internal/notification/outbox"
+	"github.com/trakrf/platform/backend/internal/notification/routing"
+	"github.com/trakrf/platform/backend/internal/testutil"
+)
+
+// workerRecordingAdapter notes every command it is asked to send.
+type workerRecordingAdapter struct {
+	mu   sync.Mutex
+	cmds []outbox.Command
+}
+
+func (a *workerRecordingAdapter) Send(_ context.Context, cmd outbox.Command) (outbox.Result, error) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.cmds = append(a.cmds, cmd)
+	return outbox.Result{ProviderMessageID: "worker-test"}, nil
+}
+
+// workerFixture is one org with an asset that has an alarm and two email
+// subscribers.
+type workerFixture struct {
+	orgID, alarmID, subscribers int
+}
+
+func newWorkerFixture(t *testing.T, db *testutil.TestDB) workerFixture {
+	t.Helper()
+	ctx := context.Background()
+	f := workerFixture{orgID: testutil.CreateTestAccount(t, db.AdminPool), subscribers: 2}
+	asset := testutil.CreateTestAsset(t, db.AdminPool, f.orgID, "ASSET-WORKER")
+
+	topic := "trakrf.id/worker/reads"
+	dev, err := db.Store.CreateScanDevice(ctx, f.orgID, scandevice.CreateScanDeviceRequest{
+		Name: "Reader worker", Type: scandevice.DeviceTypeCS463, PublishTopic: &topic,
+	})
+	require.NoError(t, err)
+	var scanPointID int
+	require.NoError(t, db.AdminPool.QueryRow(ctx,
+		`SELECT id FROM trakrf.scan_points WHERE org_id = $1 AND scan_device_id = $2 AND antenna_port = 1`,
+		f.orgID, dev.ID).Scan(&scanPointID))
+	require.NoError(t, db.AdminPool.QueryRow(ctx, `
+		INSERT INTO trakrf.alarm_events (org_id, asset_id, scan_point_id, epc)
+		VALUES ($1, $2, $3, 'E2801160600002') RETURNING id`,
+		f.orgID, asset.ID, scanPointID).Scan(&f.alarmID))
+
+	for _, email := range []string{"worker-a@acme.test", "worker-b@acme.test"} {
+		e := email
+		r, err := db.Store.CreateNotificationRecipient(ctx, f.orgID, notificationrecipient.CreateRecipientRequest{Name: e, Email: &e})
+		require.NoError(t, err)
+		_, _, err = db.Store.CreateAssetNotificationSubscription(ctx, f.orgID, asset.ID, r.ID, notificationrecipient.ChannelEmail)
+		require.NoError(t, err)
+	}
+	return f
+}
+
+func TestNewInserter_InsertsTriggerJob(t *testing.T) {
+	ctx := context.Background()
+	db := testutil.SetupTestDBFull(t)
+	orgID := testutil.CreateTestAccount(t, db.AdminPool)
+
+	inserter, err := outbox.NewInserter(db.AppPool)
+	require.NoError(t, err)
+
+	trigger := routing.Trigger{OrgID: orgID, Source: notificationevent.SourceAlarm, SourceID: 42}
+	require.NoError(t, db.Store.WithOrgTx(ctx, orgID, func(tx pgx.Tx) error {
+		inserted, err := routing.EnqueueTriggerTx(ctx, tx, inserter, trigger)
+		require.True(t, inserted)
+		return err
+	}))
+
+	var n int
+	require.NoError(t, db.AdminPool.QueryRow(ctx,
+		`SELECT count(*) FROM trakrf.river_job WHERE kind = $1`, routing.TriggerJobArgs{}.Kind()).Scan(&n))
+	require.Equal(t, 1, n)
+}
+
+func TestTriggerWorker_EndToEnd(t *testing.T) {
+	ctx := context.Background()
+	db := testutil.SetupTestDBFull(t)
+	f := newWorkerFixture(t, db)
+
+	registry := prometheus.NewRegistry()
+	outboxMetrics, err := outbox.NewMetrics(registry)
+	require.NoError(t, err)
+	routingMetrics, err := routing.NewMetrics(registry)
+	require.NoError(t, err)
+	renderer, err := routing.NewTemplateRenderer()
+	require.NoError(t, err)
+
+	inserter, err := outbox.NewInserter(db.AppPool)
+	require.NoError(t, err)
+	planner, err := routing.NewDeliveryPlanner(routing.Deps{
+		Store: db.Store, Jobs: inserter, Renderer: renderer,
+		Metrics: routingMetrics, OutboxMetrics: outboxMetrics,
+		PhoneRegion: "US", Log: zerolog.Nop(),
+	})
+	require.NoError(t, err)
+
+	adapter := &workerRecordingAdapter{}
+	runtime, err := outbox.NewRuntime(db.AppPool, adapter, db.Store, outboxMetrics, routing.RegisterTriggerWorker(planner))
+	require.NoError(t, err)
+
+	trigger := routing.Trigger{OrgID: f.orgID, Source: notificationevent.SourceAlarm, SourceID: f.alarmID}
+	require.NoError(t, db.Store.WithOrgTx(ctx, f.orgID, func(tx pgx.Tx) error {
+		_, err := routing.EnqueueTriggerTx(ctx, tx, inserter, trigger)
+		return err
+	}))
+
+	require.NoError(t, runtime.Start(ctx))
+	t.Cleanup(func() { _ = runtime.Stop(context.Background()) })
+
+	count := func(table string) int {
+		var n int
+		require.NoError(t, db.AdminPool.QueryRow(ctx, `SELECT count(*) FROM trakrf.`+table+` WHERE org_id = $1`, f.orgID).Scan(&n))
+		return n
+	}
+	require.Eventually(t, func() bool {
+		return count("notification_events") == 1 && count("notification_deliveries") == f.subscribers
+	}, 10*time.Second, 100*time.Millisecond)
+
+	require.NoError(t, runtime.Stop(ctx))
+}

backend/internal/notification/routing/worker_test.go

@@ -0,0 +1,57 @@
+package routing_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+	"github.com/stretchr/testify/require"
+
+	"github.com/trakrf/platform/backend/internal/models/notificationevent"
+	"github.com/trakrf/platform/backend/internal/notification/routing"
+)
+
+type fakeTriggerPlanner struct {
+	got routing.Trigger
+	err error
+}
+
+func (f *fakeTriggerPlanner) Plan(_ context.Context, t routing.Trigger) (routing.Result, error) {
+	f.got = t
+	return routing.Result{}, f.err
+}
+
+func triggerJob(orgID int, source string, sourceID int) *river.Job[routing.TriggerJobArgs] {
+	return &river.Job[routing.TriggerJobArgs]{
+		Args: routing.TriggerJobArgs{OrgID: orgID, Source: source, SourceID: sourceID},
+	}
+}
+
+func TestWork_InvalidTriggerIsCancelled(t *testing.T) {
+	planner := &fakeTriggerPlanner{err: fmt.Errorf("plan: %w", routing.ErrInvalidTrigger)}
+	err := routing.NewTriggerWorker(planner).Work(context.Background(), triggerJob(1, "alarm", 2))
+
+	var cancel *rivertype.JobCancelError
+	require.True(t, errors.As(err, &cancel), "got %v", err)
+}
+
+func TestWork_OtherErrorIsRetried(t *testing.T) {
+	boom := errors.New("database hiccup")
+	planner := &fakeTriggerPlanner{err: boom}
+	err := routing.NewTriggerWorker(planner).Work(context.Background(), triggerJob(1, "alarm", 2))
+
+	require.ErrorIs(t, err, boom)
+	var cancel *rivertype.JobCancelError
+	require.False(t, errors.As(err, &cancel))
+}
+
+func TestWork_PassesTrigger(t *testing.T) {
+	planner := &fakeTriggerPlanner{}
+	err := routing.NewTriggerWorker(planner).Work(context.Background(), triggerJob(1, "location_change", 9))
+
+	require.NoError(t, err)
+	require.Equal(t, routing.Trigger{OrgID: 1, Source: notificationevent.SourceLocationChange, SourceID: 9}, planner.got)
+}