diff --git a/backend/internal/notification/outbox/runtime.go b/backend/internal/notification/outbox/runtime.go index 95cbd507..078f208d 100644 --- a/backend/internal/notification/outbox/runtime.go +++ b/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 diff --git a/backend/internal/notification/outbox/runtime_test.go b/backend/internal/notification/outbox/runtime_test.go index 5c2f9c07..f86792a2 100644 --- a/backend/internal/notification/outbox/runtime_test.go +++ b/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) +} diff --git a/backend/internal/notification/routing/worker.go b/backend/internal/notification/routing/worker.go new file mode 100644 index 00000000..e00a1791 --- /dev/null +++ b/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)) + } +} diff --git a/backend/internal/notification/routing/worker_integration_test.go b/backend/internal/notification/routing/worker_integration_test.go new file mode 100644 index 00000000..dfb1703e --- /dev/null +++ b/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)) +} diff --git a/backend/internal/notification/routing/worker_test.go b/backend/internal/notification/routing/worker_test.go new file mode 100644 index 00000000..c47a7957 --- /dev/null +++ b/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) +}