diff --git a/backend/internal/notification/routing/fixture_integration_test.go b/backend/internal/notification/routing/fixture_integration_test.go new file mode 100644 index 00000000..52cc9e67 --- /dev/null +++ b/backend/internal/notification/routing/fixture_integration_test.go @@ -0,0 +1,201 @@ +//go:build integration + +package routing_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/prometheus/client_golang/prometheus" + "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/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" +) + +// idleAdapter satisfies the runtime's adapter slot; nothing is ever sent +// because the River client in these tests is never started. +type idleAdapter struct{} + +func (idleAdapter) Send(context.Context, outbox.Command) (outbox.Result, error) { + return outbox.Result{}, nil +} + +// plannerFixture is one org with one asset at one location, plus helpers to +// add alarms, moves and subscribers and to count what planning wrote. +type plannerFixture struct { + db *testutil.TestDB + orgID int + assetID int + locationID int + scanPoint int + seq int +} + +func newPlannerFixture(t *testing.T, db *testutil.TestDB, orgID int, key string) *plannerFixture { + t.Helper() + ctx := context.Background() + f := &plannerFixture{db: db, orgID: orgID} + f.assetID = testutil.CreateTestAsset(t, db.AdminPool, orgID, "ASSET-"+key).ID + + require.NoError(t, db.AdminPool.QueryRow(ctx, ` + INSERT INTO trakrf.locations (org_id, external_key, name) VALUES ($1, $2, $3) RETURNING id`, + orgID, "LOC-"+key, "Dock "+key).Scan(&f.locationID)) + + topic := "trakrf.id/" + key + "/reads" + dev, err := db.Store.CreateScanDevice(ctx, orgID, scandevice.CreateScanDeviceRequest{ + Name: "Reader " + key, Type: scandevice.DeviceTypeCS463, PublishTopic: &topic, + }) + require.NoError(t, err) + 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`, + orgID, dev.ID).Scan(&f.scanPoint)) + return f +} + +func newPlannerDB(t *testing.T) (*testutil.TestDB, *plannerFixture) { + t.Helper() + db := testutil.SetupTestDBFull(t) + orgID := testutil.CreateTestAccount(t, db.AdminPool) + return db, newPlannerFixture(t, db, orgID, "A") +} + +// alarm inserts an alarm at the fixture's location with no notification +// event, and returns its id. +func (f *plannerFixture) alarm(t *testing.T) int { + t.Helper() + return f.alarmAt(t, &f.locationID) +} + +// alarmAt inserts an alarm at the given location, or at none. +func (f *plannerFixture) alarmAt(t *testing.T, locationID *int) int { + t.Helper() + var id int + require.NoError(t, f.db.AdminPool.QueryRow(context.Background(), ` + INSERT INTO trakrf.alarm_events (org_id, asset_id, scan_point_id, location_id, epc) + VALUES ($1, $2, $3, $4, 'E2801160600002') RETURNING id`, + f.orgID, f.assetID, f.scanPoint, locationID).Scan(&id)) + return id +} + +// move inserts a location change for the asset into the fixture's location. +func (f *plannerFixture) move(t *testing.T) int { + t.Helper() + f.seq++ + at := time.Now().UTC().Add(time.Duration(f.seq) * time.Minute).Truncate(time.Microsecond) + return testutil.CreateTestLocationChange(t, f.db, f.orgID, f.assetID, f.locationID, "manual_save", nil, at) +} + +// subscribe adds a recipient with the given contact details (empty means +// none) and subscribes it to the asset on channel. It returns the recipient id. +func (f *plannerFixture) subscribe(t *testing.T, channel, email, phone string) int { + t.Helper() + ctx := context.Background() + req := notificationrecipient.CreateRecipientRequest{Name: fmt.Sprintf("Person %d", f.seq)} + f.seq++ + if email != "" { + req.Email = &email + } + if phone != "" { + req.Phone = &phone + } + r, err := f.db.Store.CreateNotificationRecipient(ctx, f.orgID, req) + require.NoError(t, err) + _, _, err = f.db.Store.CreateAssetNotificationSubscription(ctx, f.orgID, f.assetID, r.ID, channel) + require.NoError(t, err) + return r.ID +} + +// subscribeEmails adds n email subscribers. +func (f *plannerFixture) subscribeEmails(t *testing.T, n int) { + t.Helper() + for i := 0; i < n; i++ { + f.subscribe(t, notificationrecipient.ChannelEmail, fmt.Sprintf("person%d-%d@acme.test", i, f.seq), "") + } +} + +// planner builds a DeliveryPlanner on the real store and the given inserter. +func (f *plannerFixture) planner(t *testing.T, jobs outbox.JobInserter) *routing.DeliveryPlanner { + t.Helper() + renderer, err := routing.NewTemplateRenderer() + require.NoError(t, err) + metrics, err := routing.NewMetrics(prometheus.NewRegistry()) + require.NoError(t, err) + outboxMetrics, err := outbox.NewMetrics(prometheus.NewRegistry()) + require.NoError(t, err) + p, err := routing.NewDeliveryPlanner(routing.Deps{ + Store: f.db.Store, Jobs: jobs, Renderer: renderer, + Metrics: metrics, OutboxMetrics: outboxMetrics, PhoneRegion: "US", + }) + require.NoError(t, err) + return p +} + +// realJobs returns an unstarted River client: it inserts jobs but never runs +// them. +func (f *plannerFixture) realJobs(t *testing.T) *river.Client[pgx.Tx] { + t.Helper() + metrics, err := outbox.NewMetrics(prometheus.NewRegistry()) + require.NoError(t, err) + rt, err := outbox.NewRuntime(f.db.AppPool, idleAdapter{}, f.db.Store, metrics) + require.NoError(t, err) + return rt.Client +} + +func (f *plannerFixture) count(t *testing.T, query string, args ...any) int { + t.Helper() + var n int + require.NoError(t, f.db.AdminPool.QueryRow(context.Background(), query, args...).Scan(&n)) + return n +} + +// events counts this org's notification events from one source. +func (f *plannerFixture) events(t *testing.T, source notificationevent.Source) int { + t.Helper() + return f.count(t, `SELECT count(*) FROM trakrf.notification_events WHERE org_id = $1 AND source = $2`, + f.orgID, string(source)) +} + +// deliveries counts this org's notification deliveries. +func (f *plannerFixture) deliveries(t *testing.T) int { + t.Helper() + return f.count(t, `SELECT count(*) FROM trakrf.notification_deliveries WHERE org_id = $1`, f.orgID) +} + +// jobs counts River jobs of one kind. +func (f *plannerFixture) jobs(t *testing.T, kind string) int { + t.Helper() + return f.count(t, `SELECT count(*) FROM trakrf.river_job WHERE kind = $1`, kind) +} + +// assertNothingWritten checks the three tables are empty for this org. +func (f *plannerFixture) assertNothingWritten(t *testing.T) { + t.Helper() + require.Zero(t, f.events(t, notificationevent.SourceAlarm)+f.events(t, notificationevent.SourceLocationChange), "events") + require.Zero(t, f.deliveries(t), "deliveries") + require.Zero(t, f.jobs(t, outbox.DeliveryJobArgs{}.Kind()), "delivery jobs") +} + +// failingJobs delegates to inner but fails the failOn-th InsertTx. +type failingJobs struct { + inner outbox.JobInserter + failOn int + calls int +} + +func (f *failingJobs) InsertTx(ctx context.Context, tx pgx.Tx, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error) { + f.calls++ + if f.calls == f.failOn { + return nil, fmt.Errorf("injected insert failure") + } + return f.inner.InsertTx(ctx, tx, args, opts) +} diff --git a/backend/internal/notification/routing/planner_integration_test.go b/backend/internal/notification/routing/planner_integration_test.go new file mode 100644 index 00000000..49300a68 --- /dev/null +++ b/backend/internal/notification/routing/planner_integration_test.go @@ -0,0 +1,281 @@ +//go:build integration + +package routing_test + +import ( + "context" + "strconv" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/trakrf/platform/backend/internal/models/notificationdelivery" + "github.com/trakrf/platform/backend/internal/models/notificationevent" + "github.com/trakrf/platform/backend/internal/models/notificationrecipient" + "github.com/trakrf/platform/backend/internal/notification/outbox" + "github.com/trakrf/platform/backend/internal/notification/routing" +) + +const deliveryKind = "notification_delivery" + +func alarmTrigger(f *plannerFixture, id int) routing.Trigger { + return routing.Trigger{OrgID: f.orgID, Source: notificationevent.SourceAlarm, SourceID: id} +} + +func moveTrigger(f *plannerFixture, id int) routing.Trigger { + return routing.Trigger{OrgID: f.orgID, Source: notificationevent.SourceLocationChange, SourceID: id} +} + +// assertScheduled checks one event of source with n deliveries and n jobs, +// each delivery tied to the event, a recipient and the asset. +func assertScheduled(t *testing.T, f *plannerFixture, source notificationevent.Source, sourceColumn string, sourceID, n int) { + t.Helper() + ctx := context.Background() + require.Equal(t, 1, f.events(t, source)) + + var eventID int + var gotSourceID *int + require.NoError(t, f.db.AdminPool.QueryRow(ctx, + `SELECT id, `+sourceColumn+` FROM trakrf.notification_events WHERE org_id = $1 AND source = $2`, + f.orgID, string(source)).Scan(&eventID, &gotSourceID)) + require.NotNil(t, gotSourceID) + require.Equal(t, sourceID, *gotSourceID) + + require.Equal(t, n, f.deliveries(t)) + require.Equal(t, n, f.count(t, ` + SELECT count(*) FROM trakrf.notification_deliveries + WHERE org_id = $1 AND event_id = $2 AND recipient_id IS NOT NULL AND asset_id = $3 + AND payload->>'source' = $4 AND river_job_id IS NOT NULL`, + f.orgID, itoa(eventID), f.assetID, string(source))) + require.Equal(t, n, f.jobs(t, deliveryKind)) +} + +func itoa(n int) string { + return strconv.Itoa(n) +} + +func TestPlan_AlarmTriggerSchedulesDeliveries(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + alarmID := f.alarm(t) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + require.Len(t, res.Recipients, 3) + for _, r := range res.Recipients { + require.Equal(t, routing.OutcomeEnqueued, r.Outcome) + require.NotEmpty(t, r.DeliveryID) + } + assertScheduled(t, f, notificationevent.SourceAlarm, "alarm_event_id", alarmID, 3) +} + +func TestPlan_LocationChangeTriggerSchedulesDeliveries(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + moveID := f.move(t) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), moveTrigger(f, moveID)) + require.NoError(t, err) + require.Len(t, res.Recipients, 3) + assertScheduled(t, f, notificationevent.SourceLocationChange, "location_change_id", moveID, 3) +} + +func TestPlan_GateReadAsBothRegistersTwoEvents(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + alarmID, moveID := f.alarm(t), f.move(t) + p := f.planner(t, f.realJobs(t)) + + _, err := p.Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + _, err = p.Plan(context.Background(), moveTrigger(f, moveID)) + require.NoError(t, err) + + require.Equal(t, 1, f.events(t, notificationevent.SourceAlarm)) + require.Equal(t, 1, f.events(t, notificationevent.SourceLocationChange)) + require.Equal(t, 6, f.deliveries(t)) + require.Equal(t, 6, f.count(t, `SELECT count(DISTINCT delivery_id) FROM trakrf.notification_deliveries WHERE org_id = $1`, f.orgID)) + require.Equal(t, 6, f.jobs(t, deliveryKind)) +} + +func TestPlan_TwiceAddsNothing(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + alarmID := f.alarm(t) + p := f.planner(t, f.realJobs(t)) + + _, err := p.Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + res, err := p.Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, routing.OutcomeDuplicate, res.Outcome) + assertScheduled(t, f, notificationevent.SourceAlarm, "alarm_event_id", alarmID, 3) +} + +func TestPlan_LateSubscriberGetsNothingOnReplay(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 2) + alarmID := f.alarm(t) + p := f.planner(t, f.realJobs(t)) + + _, err := p.Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + f.subscribe(t, notificationrecipient.ChannelEmail, "late@acme.test", "") + res, err := p.Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, routing.OutcomeDuplicate, res.Outcome) + assertScheduled(t, f, notificationevent.SourceAlarm, "alarm_event_id", alarmID, 2) +} + +func TestPlan_ConcurrentPlansScheduleOnce(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + alarmID := f.alarm(t) + p := f.planner(t, f.realJobs(t)) + + start := make(chan struct{}) + errs := make([]error, 2) + var wg sync.WaitGroup + for i := range errs { + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, errs[i] = p.Plan(context.Background(), alarmTrigger(f, alarmID)) + }() + } + close(start) + wg.Wait() + + require.NoError(t, errs[0]) + require.NoError(t, errs[1]) + assertScheduled(t, f, notificationevent.SourceAlarm, "alarm_event_id", alarmID, 3) +} + +func TestPlan_SoftDeletedAssetRegistersEventOnly(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 2) + alarmID := f.alarm(t) + _, err := f.db.AdminPool.Exec(context.Background(), `UPDATE trakrf.assets SET deleted_at = now() WHERE id = $1`, f.assetID) + require.NoError(t, err) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, routing.OutcomeSkippedAssetDeleted, res.Outcome) + require.Equal(t, 1, f.events(t, notificationevent.SourceAlarm)) + require.Zero(t, f.deliveries(t)) + require.Zero(t, f.jobs(t, deliveryKind)) +} + +func TestPlan_ZeroSubscribersRegistersEventOnly(t *testing.T) { + _, f := newPlannerDB(t) + alarmID := f.alarm(t) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, routing.OutcomeSkippedNoSubscribers, res.Outcome) + require.Equal(t, 1, f.events(t, notificationevent.SourceAlarm)) + require.Zero(t, f.deliveries(t)) + require.Zero(t, f.jobs(t, deliveryKind)) +} + +func TestPlan_UnentitledOrgWritesNothing(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 2) + alarmID := f.alarm(t) + _, err := f.db.AdminPool.Exec(context.Background(), + `UPDATE trakrf.organizations SET subscription_enabled = false WHERE id = $1`, f.orgID) + require.NoError(t, err) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, routing.OutcomeSkippedUnentitled, res.Outcome) + f.assertNothingWritten(t) +} + +func TestPlan_FailureOnSecondJobWritesNothing(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 3) + alarmID := f.alarm(t) + jobs := &failingJobs{inner: f.realJobs(t), failOn: 2} + + _, err := f.planner(t, jobs).Plan(context.Background(), alarmTrigger(f, alarmID)) + + require.Error(t, err) + require.NotErrorIs(t, err, routing.ErrInvalidTrigger) + require.GreaterOrEqual(t, jobs.calls, 2) + f.assertNothingWritten(t) +} + +func TestPlan_OtherOrgSourceIsInvalid(t *testing.T) { + db, f := newPlannerDB(t) + f.subscribeEmails(t, 2) + + var otherOrg int + require.NoError(t, db.AdminPool.QueryRow(context.Background(), ` + INSERT INTO trakrf.organizations (name, identifier, is_active) VALUES ('Other', 'other-org', true) RETURNING id`).Scan(&otherOrg)) + other := newPlannerFixture(t, db, otherOrg, "B") + foreignAlarm := other.alarm(t) + + _, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, foreignAlarm)) + + require.ErrorIs(t, err, routing.ErrInvalidTrigger) + f.assertNothingWritten(t) + other.assertNothingWritten(t) +} + +func TestPlan_SuppressionMatchesAcrossPhoneFormats(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribe(t, notificationrecipient.ChannelSMS, "", "(555) 123-4567") + alarmID := f.alarm(t) + _, err := f.db.AdminPool.Exec(context.Background(), ` + INSERT INTO trakrf.notification_suppressions (org_id, channel, address, reason) + VALUES ($1, 'sms', '+15551234567', 'sms_stop')`, f.orgID) + require.NoError(t, err) + + res, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Len(t, res.Recipients, 1) + require.Equal(t, notificationdelivery.Channel("sms"), res.Recipients[0].Channel) + require.Equal(t, routing.OutcomeSkippedOptedOut, res.Recipients[0].Outcome) + require.Equal(t, 1, f.events(t, notificationevent.SourceAlarm)) + require.Zero(t, f.count(t, `SELECT count(*) FROM trakrf.notification_deliveries WHERE org_id = $1 AND channel = 'sms'`, f.orgID)) + require.Zero(t, f.jobs(t, deliveryKind)) +} + +func TestPlan_CancelledContextWritesNothing(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 2) + alarmID := f.alarm(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := f.planner(t, f.realJobs(t)).Plan(ctx, alarmTrigger(f, alarmID)) + + require.Error(t, err) + require.NotErrorIs(t, err, routing.ErrInvalidTrigger) + f.assertNothingWritten(t) +} + +func TestPlan_AlarmWithoutLocationSaysUnknown(t *testing.T) { + _, f := newPlannerDB(t) + f.subscribeEmails(t, 1) + // A move always has a destination; an alarm may have none. + alarmID := f.alarmAt(t, nil) + + _, err := f.planner(t, f.realJobs(t)).Plan(context.Background(), alarmTrigger(f, alarmID)) + require.NoError(t, err) + + require.Equal(t, 1, f.count(t, ` + SELECT count(*) FROM trakrf.notification_deliveries + WHERE org_id = $1 AND payload->>'body' LIKE '%an unknown location%'`, f.orgID)) +} + +var _ outbox.JobInserter = (*failingJobs)(nil)