package notifications_test import ( "context" "encoding/json" "io" "log/slog" "os" "testing" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/require" "git.fabledsword.com/bvandeusen/minstrel/internal/db" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/dbtest" "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" "git.fabledsword.com/bvandeusen/minstrel/internal/notifications" ) func testPool(t *testing.T) *pgxpool.Pool { t.Helper() if testing.Short() { t.Skip("skipping integration test in -short mode") } dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") if dsn == "" { t.Skip("MINSTREL_TEST_DATABASE_URL not set") } if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatalf("migrate: %v", err) } pool, err := pgxpool.New(context.Background(), dsn) if err != nil { t.Fatalf("pool: %v", err) } t.Cleanup(pool.Close) dbtest.ResetDB(t, pool) return pool } func mkUser(t *testing.T, q *dbq.Queries, name string, admin bool) pgtype.UUID { t.Helper() u, err := q.CreateUser(context.Background(), dbq.CreateUserParams{ Username: dbtest.TestUserPrefix + name, PasswordHash: "x", ApiTokenHash: "tok-" + name, // unique per user IsAdmin: admin, }) require.NoError(t, err) return u.ID } func rowsFor(t *testing.T, q *dbq.Queries, user pgtype.UUID) []dbq.ListNotificationsRow { t.Helper() rows, err := q.ListNotifications(context.Background(), dbq.ListNotificationsParams{UserID: user, PageLimit: 100}) require.NoError(t, err) return rows } func count(t *testing.T, payload []byte) int64 { t.Helper() var p struct { Count int64 `json:"count"` } require.NoError(t, json.Unmarshal(payload, &p)) return p.Count } func TestNotify_WritesARowAndNudgesOnlyThatUser(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() alice := mkUser(t, q, "alice", false) bob := mkUser(t, q, "bob", false) bus := eventbus.New() events, unsub := bus.Subscribe(8) defer unsub() n := notifications.New(pool, bus, nil) require.NoError(t, n.Notify(ctx, notifications.KindRequestApproved, notifications.ToUser(alice), map[string]any{"request_id": "r1", "title": "WWW"})) rows := rowsFor(t, q, alice) require.Len(t, rows, 1) require.Equal(t, "request_approved", rows[0].Kind) require.False(t, rows[0].ReadAt.Valid) require.JSONEq(t, `{"request_id":"r1","title":"WWW"}`, string(rows[0].Payload)) require.Empty(t, rowsFor(t, q, bob)) select { case e := <-events: require.Equal(t, notifications.EventCreated, e.Kind) require.Equal(t, uuid.UUID(alice.Bytes).String(), e.UserID) require.Empty(t, e.Data, "the nudge carries no content; the client fetches") case <-time.After(time.Second): t.Fatal("no nudge published") } } func TestNotify_AdminKindReachesAdminsOnly_AndNeverTheExcepted(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() requester := mkUser(t, q, "admin-requester", true) other := mkUser(t, q, "admin-other", true) plain := mkUser(t, q, "plain", false) n := notifications.New(pool, nil, nil) // A plain user named directly must still get nothing: the kind is admin-only. to := notifications.ToAdmins(requester) to.Users = []pgtype.UUID{plain} require.NoError(t, n.Notify(ctx, notifications.KindRequestPending, to, nil)) require.Empty(t, rowsFor(t, q, requester), "the admin who filed it needs no notice") require.Len(t, rowsFor(t, q, other), 1) require.Empty(t, rowsFor(t, q, plain)) } func TestNotify_CoalescedCountAddsUpWhileUnread_AndStartsAfreshOnceRead(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() admin := mkUser(t, q, "admin", true) n := notifications.New(pool, nil, nil) to := notifications.ToUser(admin) require.NoError(t, n.Notify(ctx, notifications.KindTracksMissing, to, map[string]any{"count": 3})) require.NoError(t, n.Notify(ctx, notifications.KindTracksMissing, to, map[string]any{"count": 4})) rows := rowsFor(t, q, admin) require.Len(t, rows, 1, "a burst is one notification") require.Equal(t, int64(7), count(t, rows[0].Payload)) _, err := q.MarkNotificationRead(ctx, dbq.MarkNotificationReadParams{ID: rows[0].ID, UserID: admin}) require.NoError(t, err) require.NoError(t, n.Notify(ctx, notifications.KindTracksMissing, to, map[string]any{"count": 2})) rows = rowsFor(t, q, admin) require.Len(t, rows, 2, "after it is read, the next event is new news") require.Equal(t, int64(2), count(t, rows[0].Payload)) require.False(t, rows[0].ReadAt.Valid) require.True(t, rows[1].ReadAt.Valid) } func TestNotify_CoalescedTotalReplacesWhenEventsStateTheWhole(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() admin := mkUser(t, q, "admin", true) n := notifications.New(pool, nil, nil) // duplicates_found states the pending total each time; it must not sum. require.NoError(t, n.Notify(ctx, notifications.KindDuplicatesFound, notifications.ToUser(admin), map[string]any{"count": 5})) require.NoError(t, n.Notify(ctx, notifications.KindDuplicatesFound, notifications.ToUser(admin), map[string]any{"count": 3})) rows := rowsFor(t, q, admin) require.Len(t, rows, 1) require.Equal(t, int64(3), count(t, rows[0].Payload)) } func TestNotify_RequestKindsNeverCoalesce(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() alice := mkUser(t, q, "alice", false) n := notifications.New(pool, nil, nil) for _, id := range []string{"r1", "r2"} { require.NoError(t, n.Notify(ctx, notifications.KindRequestCompleted, notifications.ToUser(alice), map[string]any{"request_id": id})) } require.Len(t, rowsFor(t, q, alice), 2) } func TestNotify_InboxOffStoresNothing_AndDefaultsApplyWithoutAPref(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() muted := mkUser(t, q, "muted", false) fresh := mkUser(t, q, "fresh", false) require.NoError(t, q.UpsertNotificationPref(ctx, dbq.UpsertNotificationPrefParams{ UserID: muted, Kind: "request_rejected", Inbox: false, Phone: true, Email: true, })) n := notifications.New(pool, nil, nil) to := notifications.Recipients{Users: []pgtype.UUID{muted, fresh}} require.NoError(t, n.Notify(ctx, notifications.KindRequestRejected, to, nil)) require.Empty(t, rowsFor(t, q, muted), "inbox off is off, whatever phone and email say") require.Len(t, rowsFor(t, q, fresh), 1) } func TestMarkRead_ScopedToTheOwner_AndIdempotent(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() alice := mkUser(t, q, "alice", false) mallory := mkUser(t, q, "mallory", false) n := notifications.New(pool, nil, nil) require.NoError(t, n.Notify(ctx, notifications.KindRequestApproved, notifications.ToUser(alice), nil)) id := rowsFor(t, q, alice)[0].ID got, err := q.MarkNotificationRead(ctx, dbq.MarkNotificationReadParams{ID: id, UserID: mallory}) require.NoError(t, err) require.Zero(t, got, "another user's id matches nothing") require.False(t, rowsFor(t, q, alice)[0].ReadAt.Valid) for i := 0; i < 2; i++ { got, err = q.MarkNotificationRead(ctx, dbq.MarkNotificationReadParams{ID: id, UserID: alice}) require.NoError(t, err) require.Equal(t, int64(1), got, "a repeat still matches, so it is not mistaken for 'not yours'") } unread, err := q.CountUnreadNotifications(ctx, alice) require.NoError(t, err) require.Zero(t, unread) } // Every kind the code knows must pass both CHECKs in the real schema. This is // rule 36's guard: a kind added in Go without the migration fails here, not in // production at INSERT time. func TestEveryKindPassesTheSchemaChecks(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() user := mkUser(t, q, "kinds", true) for _, k := range notifications.Kinds() { _, err := q.InsertNotification(ctx, dbq.InsertNotificationParams{UserID: user, Kind: string(k), Payload: []byte(`{}`)}) require.NoError(t, err, "user_notifications rejects %s", k) require.NoError(t, q.UpsertNotificationPref(ctx, dbq.UpsertNotificationPrefParams{ UserID: user, Kind: string(k), Inbox: true, Phone: true, Email: true, }), "user_notification_prefs rejects %s", k) } // And the check is real: an unknown kind is refused. _, err := q.InsertNotification(ctx, dbq.InsertNotificationParams{UserID: user, Kind: "bogus", Payload: []byte(`{}`)}) require.Error(t, err) } func TestRetention_TrimsOldReadRowsAndAncientUnreadOnes(t *testing.T) { pool := testPool(t) q := dbq.New(pool) ctx := context.Background() user := mkUser(t, q, "retention", false) now := time.Now() insert := func(created time.Time, read *time.Time) { var readAt any if read != nil { readAt = *read } _, err := pool.Exec(ctx, `INSERT INTO user_notifications (user_id, kind, created_at, read_at) VALUES ($1, 'request_approved', $2, $3)`, user, created, readAt) require.NoError(t, err) } day := 24 * time.Hour oldRead := now.Add(-100 * day) recentRead := now.Add(-10 * day) insert(now.Add(-120*day), &oldRead) // read 100 days ago: goes insert(now.Add(-20*day), &recentRead) // read 10 days ago: stays insert(now.Add(-400*day), nil) // unread but over a year old: goes insert(now.Add(-200*day), nil) // unread, 200 days: stays r := notifications.NewRetention(pool, slog.New(slog.NewTextHandler(io.Discard, nil))) require.Equal(t, int64(2), r.TrimOnce(ctx, now)) require.Len(t, rowsFor(t, q, user), 2) }