feat: add showAt column to user_notification for scheduled notifications

__tests__/notifications.ts

@@ -28,7 +28,7 @@ import { DataSource } from 'typeorm';
 import createOrGetConnection from '../src/db';
 import { usersFixture } from './fixture/user';
 import { notificationV2Fixture } from './fixture/notifications';
-import { subDays } from 'date-fns';
+import { addDays, subDays } from 'date-fns';
 import request from 'supertest';
 import { FastifyInstance } from 'fastify';
 import {
@@ -165,6 +165,32 @@ describe('query notification count', () => {
     expect(res.data).toEqual({ unreadNotificationsCount: 1 });
   });
 
+  it('should not count notifications with future showAt', async () => {
+    loggedUser = '1';
+    const notifs = await con.getRepository(NotificationV2).save([
+      { ...notificationV2Fixture },
+      {
+        ...notificationV2Fixture,
+        uniqueKey: '2',
+      },
+    ]);
+    await con.getRepository(UserNotification).insert([
+      {
+        userId: '1',
+        notificationId: notifs[0].id,
+        createdAt: notificationV2Fixture.createdAt,
+      },
+      {
+        userId: '1',
+        notificationId: notifs[1].id,
+        createdAt: notificationV2Fixture.createdAt,
+        showAt: addDays(new Date(), 1),
+      },
+    ]);
+    const res = await client.query(QUERY());
+    expect(res.data).toEqual({ unreadNotificationsCount: 1 });
+  });
+
   it('should cap the count at UNREAD_NOTIFICATIONS_LIMIT', async () => {
     loggedUser = '1';
     const total = UNREAD_NOTIFICATIONS_LIMIT + 5;
@@ -430,6 +456,40 @@ describe('query notifications', () => {
       date.toISOString(),
     );
   });
+
+  it('should hide notifications with future showAt', async () => {
+    loggedUser = '1';
+    const notifs = await con.getRepository(NotificationV2).save([
+      { ...notificationV2Fixture, title: 'visible' },
+      { ...notificationV2Fixture, uniqueKey: '2', title: 'future' },
+      { ...notificationV2Fixture, uniqueKey: '3', title: 'past showAt' },
+    ]);
+    await con.getRepository(UserNotification).insert([
+      {
+        userId: '1',
+        notificationId: notifs[0].id,
+        createdAt: notificationV2Fixture.createdAt,
+      },
+      {
+        userId: '1',
+        notificationId: notifs[1].id,
+        createdAt: notificationV2Fixture.createdAt,
+        showAt: addDays(new Date(), 1),
+      },
+      {
+        userId: '1',
+        notificationId: notifs[2].id,
+        createdAt: notificationV2Fixture.createdAt,
+        showAt: subDays(new Date(), 1),
+      },
+    ]);
+    const res = await client.query(QUERY);
+    const titles = res.data.notifications.edges.map((e) => e.node.title);
+    expect(titles).toHaveLength(2);
+    expect(titles).toContain('visible');
+    expect(titles).toContain('past showAt');
+    expect(titles).not.toContain('future');
+  });
 });
 
 const prepareNotificationPreferences = async ({
@@ -628,6 +688,38 @@ describe('mutation readNotifications', () => {
       .find({ where: { userId: '2' }, order: { createdAt: 'desc' } });
     res2.map((notification) => expect(notification.readAt).toBeFalsy());
   });
+
+  it('should not mark future showAt notifications as read', async () => {
+    loggedUser = '1';
+    const notifs = await con
+      .getRepository(NotificationV2)
+      .save([
+        { ...notificationV2Fixture },
+        { ...notificationV2Fixture, uniqueKey: '2' },
+      ]);
+    await con.getRepository(UserNotification).insert([
+      {
+        userId: '1',
+        notificationId: notifs[0].id,
+        createdAt: notificationV2Fixture.createdAt,
+      },
+      {
+        userId: '1',
+        notificationId: notifs[1].id,
+        createdAt: notificationV2Fixture.createdAt,
+        showAt: addDays(new Date(), 1),
+      },
+    ]);
+    await client.mutate(QUERY);
+    const userNotifs = await con.getRepository(UserNotification).find({
+      where: { userId: '1' },
+      order: { notificationId: 'ASC' },
+    });
+    const immediate = userNotifs.find((n) => n.notificationId === notifs[0].id);
+    const future = userNotifs.find((n) => n.notificationId === notifs[1].id);
+    expect(immediate.readAt).toBeTruthy();
+    expect(future.readAt).toBeNull();
+  });
 });
 
 describe('mutation muteNotificationPreference', () => {
@@ -1494,6 +1586,46 @@ describe('streamNotificationUsers', () => {
 
     expect(results).toHaveLength(0);
   });
+
+  it('should exclude users whose notification has a future showAt', async () => {
+    const users = [
+      { id: 'user15', name: 'User 15', email: '[email protected]' },
+      { id: 'user16', name: 'User 16', email: '[email protected]' },
+    ];
+
+    await con.getRepository(User).save(users);
+
+    const notif = await con.getRepository(NotificationV2).save({
+      ...notificationV2Fixture,
+      type: NotificationType.ArticleNewComment,
+    });
+
+    await con.getRepository(UserNotification).insert([
+      {
+        userId: 'user15',
+        notificationId: notif.id,
+        public: true,
+        createdAt: notificationV2Fixture.createdAt,
+        showAt: addDays(new Date(), 1),
+      },
+      {
+        userId: 'user16',
+        notificationId: notif.id,
+        public: true,
+        createdAt: notificationV2Fixture.createdAt,
+      },
+    ]);
+
+    const stream = await streamNotificationUsers(
+      con,
+      notif.id,
+      NotificationChannel.InApp,
+    );
+    const results = await streamToArray(stream);
+
+    expect(results).toHaveLength(1);
+    expect(results[0].userId).toBe('user16');
+  });
 });
 
 describe('poll result notifications', () => {

__tests__/notifications/notificationWorkerToWorker.ts

@@ -132,6 +132,7 @@ describe('notificationWorkerToWorker', () => {
         readAt: null,
         userId: '3',
         uniqueKey: null,
+        showAt: null,
       },
       {
         createdAt: notifications[0].createdAt,
@@ -140,6 +141,7 @@ describe('notificationWorkerToWorker', () => {
         readAt: null,
         userId: '4',
         uniqueKey: null,
+        showAt: null,
       },
     ]);
   });
@@ -206,6 +208,35 @@ describe('notificationWorkerToWorker', () => {
     expect(userNotifications.length).toEqual(2);
   });
 
+  it('should set showAt from sendAtMs on notification context', async () => {
+    const sendAtMs = new Date('2025-06-01T12:00:00Z').getTime();
+    const worker = notificationWorkerToWorker({
+      subscription: 'sub',
+      handler: async (message, con) => {
+        const postCtx = await buildPostContext(con, 'p1');
+        const users = await con
+          .getRepository(User)
+          .find({ where: { id: In(['1', '2']) } });
+        return [
+          {
+            type: NotificationType.ArticleUpvoteMilestone,
+            ctx: {
+              ...postCtx,
+              upvoters: users,
+              upvotes: 2,
+              userIds: ['3'],
+              sendAtMs,
+            },
+          },
+        ];
+      },
+    });
+    await worker.handler(message({}), con, null, null);
+    const userNotifications = await con.getRepository(UserNotification).find();
+    expect(userNotifications).toHaveLength(1);
+    expect(userNotifications[0].showAt).toEqual(new Date(sendAtMs));
+  });
+
   it('should handle duplicate notification', async () => {
     await con.getRepository(NotificationV2).save({
       ...notificationV2Fixture,

__tests__/workers/notifications/userBriefReadyNotification.ts

@@ -1,5 +1,6 @@
 import { DataSource } from 'typeorm';
 import { userBriefReadyNotification as worker } from '../../../src/workers/notifications/userBriefReadyNotification';
+import { notificationWorkerToWorker } from '../../../src/workers/notifications';
 import createOrGetConnection from '../../../src/db';
 import { Source, User } from '../../../src/entity';
 import { sourcesFixture, usersFixture } from '../../fixture';
@@ -11,6 +12,8 @@ import type { PubSubSchema } from '../../../src/common';
 import { BriefingModel } from '../../../src/integrations/feed';
 import { NotificationType } from '../../../src/notifications/common';
 import type { NotificationPostContext } from '../../../src/notifications';
+import { UserNotification } from '../../../src/entity/notifications/UserNotification';
+import { Message } from '../../../src/workers/worker';
 
 let con: DataSource;
 
@@ -69,4 +72,52 @@ describe('userBriefReadyNotification worker', () => {
     expect(postContext.userIds).toEqual(['1']);
     expect(postContext.post.id).toEqual(postId);
   });
+
+  it('should pass sendAtMs in context and set showAt on user_notification', async () => {
+    const postId = await generateShortId();
+    const sendAtMs = new Date('2025-06-01T12:00:00Z').getTime();
+
+    const post = con.getRepository(BriefPost).create({
+      id: postId,
+      title: 'Test Brief',
+      content: 'This is a test brief content.',
+      contentHtml: '<p>This is a test brief content.</p>',
+      shortId: postId,
+      authorId: '1',
+      private: true,
+      visible: true,
+    });
+
+    await con.getRepository(BriefPost).save(post);
+
+    const eventData = {
+      payload: {
+        userId: '1',
+        frequency: 'daily',
+        modelName: BriefingModel.Default,
+      },
+      postId,
+      sendAtMs,
+    } as PubSubSchema['api.v1.brief-ready'];
+
+    const result = await invokeTypedNotificationWorker<'api.v1.brief-ready'>(
+      worker,
+      eventData,
+    );
+    expect(result![0].ctx.sendAtMs).toEqual(sendAtMs);
+
+    const fullWorker = notificationWorkerToWorker(worker);
+    const msg: Message = {
+      data: Buffer.from(JSON.stringify(eventData), 'utf-8'),
+      messageId: '1',
+    };
+    await fullWorker.handler(msg, con, null, null);
+
+    const userNotification = await con
+      .getRepository(UserNotification)
+      .findOneBy({ userId: '1' });
+
+    expect(userNotification).not.toBeNull();
+    expect(userNotification!.showAt).toEqual(new Date(sendAtMs));
+  });
 });

__tests__/workers/personalizedDigestEmail.ts

@@ -40,6 +40,7 @@ import { BriefingModel } from '../../src/integrations/feed/types';
 import { BriefPost } from '../../src/entity/posts/BriefPost';
 import { DigestPost } from '../../src/entity/posts/DigestPost';
 import { NotificationV2 } from '../../src/entity/notifications/NotificationV2';
+import { UserNotification } from '../../src/entity/notifications/UserNotification';
 import { NotificationType } from '../../src/notifications/common';
 
 jest.mock('../../src/common', () => ({
@@ -636,6 +637,31 @@ describe('personalizedDigestEmail worker', () => {
     expect(notification).not.toBeNull();
   });
 
+  it('should set showAt on user_notification from emailSendTimestamp', async () => {
+    const personalizedDigest = await con
+      .getRepository(UserPersonalizedDigest)
+      .findOneBy({
+        userId: '1',
+      });
+
+    const dates = getDates(personalizedDigest!, Date.now());
+
+    await expectSuccessfulBackground(worker, {
+      personalizedDigest,
+      ...dates,
+      emailBatchId: 'test-email-batch-id',
+    });
+
+    const userNotification = await con
+      .getRepository(UserNotification)
+      .findOneBy({ userId: '1' });
+
+    expect(userNotification).not.toBeNull();
+    expect(userNotification!.showAt).toEqual(
+      new Date(dates.emailSendTimestamp),
+    );
+  });
+
   it('should still send email after creating DigestPost', async () => {
     const personalizedDigest = await con
       .getRepository(UserPersonalizedDigest)

src/entity/notifications/UserNotification.ts

@@ -52,4 +52,7 @@ export class UserNotification {
 
   @Column({ type: 'text', nullable: true })
   uniqueKey: string | null;
+
+  @Column({ type: 'timestamp', nullable: true, default: null })
+  showAt: Date | null;
 }

src/migration/1772186714109-UserNotificationShowAt.ts

@@ -0,0 +1,17 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class UserNotificationShowAt1772186714109 implements MigrationInterface {
+  name = 'UserNotificationShowAt1772186714109';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "user_notification" ADD "showAt" TIMESTAMP`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "user_notification" DROP COLUMN "showAt"`,
+    );
+  }
+}

src/notifications/common.ts

@@ -466,7 +466,8 @@ export const streamNotificationUsers = (
     .from(UserNotification, 'un')
     .innerJoin('user', 'u', 'un."userId" = u.id')
     .innerJoin(NotificationV2, 'n', 'un."notificationId" = n.id')
-    .where('un."notificationId" = :id', { id });
+    .where('un."notificationId" = :id', { id })
+    .andWhere('(un."showAt" IS NULL OR un."showAt" <= NOW())');
 
   if (channel === NotificationChannel.InApp) {
     query = query
@@ -500,6 +501,7 @@ export const getUnreadNotificationsCount = async (
         .where('un."userId" = :userId', { userId })
         .andWhere('un."public" = true')
         .andWhere('un."readAt" IS NULL')
+        .andWhere('(un."showAt" IS NULL OR un."showAt" <= NOW())')
         .limit(UNREAD_NOTIFICATIONS_LIMIT);
     }, 't')
     .getRawOne<{ count: number }>();