Scale notification delivery pipeline

app/.graphql/nitro-graphql-client.d.ts

@@ -294,6 +294,7 @@ export type HookEvent =
   | 'NOTIFICATION_SENT'
   | 'NOTIFICATION_DELIVERED'
   | 'NOTIFICATION_FAILED'
+  | 'NOTIFICATION_OPENED'
   | 'NOTIFICATION_CLICKED'
   | 'WORKFLOW_COMPLETED'
   | 'WORKFLOW_FAILED';
@@ -550,6 +551,9 @@ export type Notification = {
   sound?: Maybe<Scalars['String']['output']>;
   badge?: Maybe<Scalars['Int']['output']>;
   status: NotificationStatus;
+  channelType?: Maybe<ChannelType>;
+  channelId?: Maybe<Scalars['ID']['output']>;
+  contactIds?: Maybe<Scalars['JSON']['output']>;
   targetDevices?: Maybe<Scalars['JSON']['output']>;
   platforms?: Maybe<Scalars['JSON']['output']>;
   scheduledAt?: Maybe<Scalars['Timestamp']['output']>;
@@ -580,9 +584,12 @@ export type NotificationAnalytics = {
 
 export type NotificationStatus =
   | 'PENDING'
+  | 'QUEUED'
+  | 'PROCESSING'
   | 'SENT'
   | 'DELIVERED'
   | 'FAILED'
+  | 'PARTIAL'
   | 'SCHEDULED';
 
 export type PageInfo = {

app/.graphql/nitro-graphql-server.d.ts

@@ -355,6 +355,7 @@ export type HookEvent =
   | 'NOTIFICATION_SENT'
   | 'NOTIFICATION_DELIVERED'
   | 'NOTIFICATION_FAILED'
+  | 'NOTIFICATION_OPENED'
   | 'NOTIFICATION_CLICKED'
   | 'WORKFLOW_COMPLETED'
   | 'WORKFLOW_FAILED';
@@ -611,6 +612,9 @@ export interface Notification {
   sound?: Maybe<Scalars['String']['output']>;
   badge?: Maybe<Scalars['Int']['output']>;
   status: NotificationStatus;
+  channelType?: Maybe<ChannelType>;
+  channelId?: Maybe<Scalars['ID']['output']>;
+  contactIds?: Maybe<Scalars['JSON']['output']>;
   targetDevices?: Maybe<Scalars['JSON']['output']>;
   platforms?: Maybe<Scalars['JSON']['output']>;
   scheduledAt?: Maybe<Scalars['Timestamp']['output']>;
@@ -641,9 +645,12 @@ export interface NotificationAnalytics {
 
 export type NotificationStatus =
   | 'PENDING'
+  | 'QUEUED'
+  | 'PROCESSING'
   | 'SENT'
   | 'DELIVERED'
   | 'FAILED'
+  | 'PARTIAL'
   | 'SCHEDULED';
 
 export interface PageInfo {
@@ -1520,6 +1527,9 @@ export type NotificationResolvers<ContextType = H3Event, ParentType extends Reso
   sound?: Resolver<Maybe<ResolversTypes['String']>, ParentType, ContextType>;
   badge?: Resolver<Maybe<ResolversTypes['Int']>, ParentType, ContextType>;
   status?: Resolver<ResolversTypes['NotificationStatus'], ParentType, ContextType>;
+  channelType?: Resolver<Maybe<ResolversTypes['ChannelType']>, ParentType, ContextType>;
+  channelId?: Resolver<Maybe<ResolversTypes['ID']>, ParentType, ContextType>;
+  contactIds?: Resolver<Maybe<ResolversTypes['JSON']>, ParentType, ContextType>;
   targetDevices?: Resolver<Maybe<ResolversTypes['JSON']>, ParentType, ContextType>;
   platforms?: Resolver<Maybe<ResolversTypes['JSON']>, ParentType, ContextType>;
   scheduledAt?: Resolver<Maybe<ResolversTypes['Timestamp']>, ParentType, ContextType>;

app/.graphql/schema.graphql

@@ -273,6 +273,7 @@ enum HookEvent {
   NOTIFICATION_SENT
   NOTIFICATION_DELIVERED
   NOTIFICATION_FAILED
+  NOTIFICATION_OPENED
   NOTIFICATION_CLICKED
   WORKFLOW_COMPLETED
   WORKFLOW_FAILED
@@ -344,6 +345,9 @@ type Notification {
   sound: String
   badge: Int
   status: NotificationStatus!
+  channelType: ChannelType
+  channelId: ID
+  contactIds: JSON
   targetDevices: JSON
   platforms: JSON
   scheduledAt: Timestamp
@@ -373,9 +377,12 @@ type NotificationAnalytics {
 
 enum NotificationStatus {
   PENDING
+  QUEUED
+  PROCESSING
   SENT
   DELIVERED
   FAILED
+  PARTIAL
   SCHEDULED
 }
 

app/server/api/track/click/[id].get.ts

@@ -1,8 +1,5 @@
 import { Buffer } from 'node:buffer'
-import { getDatabase } from '#server/database/connection'
-import { deliveryLog, notification } from '#server/database/schema'
-import { dispatchHooks } from '#server/utils/webhookDispatcher'
-import { eq, sql } from 'drizzle-orm'
+import { addTrackNotificationEventJob } from '#server/queues/tracking.queue'
 import { defineEventHandler, getQuery, getRouterParam, sendRedirect } from 'nitro/h3'
 
 export default defineEventHandler(async (event) => {
@@ -27,49 +24,11 @@ export default defineEventHandler(async (event) => {
 
   if (id) {
     try {
-      const db = getDatabase()
-
-      const rows = await db
-        .select({
-          id: deliveryLog.id,
-          notificationId: deliveryLog.notificationId,
-          clickedAt: deliveryLog.clickedAt,
-        })
-        .from(deliveryLog)
-        .where(eq(deliveryLog.id, id))
-        .limit(1)
-
-      const log = rows[0]
-      if (log && !log.clickedAt) {
-        const now = new Date().toISOString()
-
-        await db
-          .update(deliveryLog)
-          .set({ clickedAt: now, updatedAt: now })
-          .where(eq(deliveryLog.id, id))
-
-        const notifRows = await db
-          .select({ appId: notification.appId })
-          .from(notification)
-          .where(eq(notification.id, log.notificationId))
-          .limit(1)
-
-        if (notifRows[0]) {
-          const { appId } = notifRows[0]
-
-          await db
-            .update(notification)
-            .set({ totalClicked: sql`"totalClicked" + 1`, updatedAt: now })
-            .where(eq(notification.id, log.notificationId))
-
-          await dispatchHooks(appId, 'NOTIFICATION_CLICKED', {
-            notificationId: log.notificationId,
-            deliveryLogId: id,
-            clickedAt: now,
-            destination,
-          })
-        }
-      }
+      await addTrackNotificationEventJob({
+        type: 'click',
+        deliveryLogId: id,
+        destination,
+      })
     }
     catch (err) {
       // Never block the redirect on tracking errors

app/server/api/track/open/[id].get.ts

@@ -1,8 +1,5 @@
 import { Buffer } from 'node:buffer'
-import { getDatabase } from '#server/database/connection'
-import { deliveryLog, notification } from '#server/database/schema'
-import { dispatchHooks } from '#server/utils/webhookDispatcher'
-import { eq, sql } from 'drizzle-orm'
+import { addTrackNotificationEventJob } from '#server/queues/tracking.queue'
 import { defineEventHandler, getRouterParam } from 'nitro/h3'
 
 // 1×1 transparent GIF
@@ -21,52 +18,10 @@ export default defineEventHandler(async (event) => {
   }
 
   try {
-    const db = getDatabase()
-
-    // Fetch delivery log — only update on first open
-    const rows = await db
-      .select({
-        id: deliveryLog.id,
-        notificationId: deliveryLog.notificationId,
-        openedAt: deliveryLog.openedAt,
-      })
-      .from(deliveryLog)
-      .where(eq(deliveryLog.id, id))
-      .limit(1)
-
-    const log = rows[0]
-    if (!log || log.openedAt) {
-      // Unknown ID or already tracked — return pixel silently
-      return PIXEL
-    }
-
-    const now = new Date().toISOString()
-
-    await db
-      .update(deliveryLog)
-      .set({ openedAt: now, updatedAt: now })
-      .where(eq(deliveryLog.id, id))
-
-    const notifRows = await db
-      .select({ appId: notification.appId })
-      .from(notification)
-      .where(eq(notification.id, log.notificationId))
-      .limit(1)
-
-    if (notifRows[0]) {
-      const { appId } = notifRows[0]
-
-      await db
-        .update(notification)
-        .set({ totalOpened: sql`"totalOpened" + 1`, updatedAt: now })
-        .where(eq(notification.id, log.notificationId))
-
-      await dispatchHooks(appId, 'NOTIFICATION_OPENED', {
-        notificationId: log.notificationId,
-        deliveryLogId: id,
-        openedAt: now,
-      })
-    }
+    await addTrackNotificationEventJob({
+      type: 'open',
+      deliveryLogId: id,
+    })
   }
   catch (err) {
     // Never break pixel delivery on tracking errors

app/server/database/connection.ts

@@ -8,7 +8,8 @@ let db: ReturnType<typeof drizzle<typeof schema>> | undefined
 export function getDatabase() {
   if (!db) {
     const connectionString = process.env.DATABASE_URL || 'postgresql://localhost:5432/nitroping'
-    pgClient = postgres(connectionString, { max: 3, idle_timeout: 30 })
+    const maxConnections = Number.parseInt(process.env.DATABASE_POOL_MAX || '20')
+    pgClient = postgres(connectionString, { max: maxConnections, idle_timeout: 30 })
     db = drizzle({ client: pgClient, schema })
   }
   return db

app/server/database/migrations/20260308000000_add_notification_targeting_fields/migration.sql

@@ -0,0 +1,4 @@
+ALTER TABLE "notification"
+  ADD COLUMN IF NOT EXISTS "channelType" "channelType",
+  ADD COLUMN IF NOT EXISTS "channelId" uuid,
+  ADD COLUMN IF NOT EXISTS "contactIds" jsonb;

app/server/database/migrations/20260309000000_expand_notification_status_enum/migration.sql

@@ -0,0 +1,3 @@
+ALTER TYPE "notification_status" ADD VALUE IF NOT EXISTS 'QUEUED';
+ALTER TYPE "notification_status" ADD VALUE IF NOT EXISTS 'PROCESSING';
+ALTER TYPE "notification_status" ADD VALUE IF NOT EXISTS 'PARTIAL';

app/server/database/schema/enums.ts

@@ -10,7 +10,16 @@ export const categoryEnum = pgEnum('category', [
 
 export const platformEnum = pgEnum('platform', ['IOS', 'ANDROID', 'WEB'])
 
-export const notificationStatusEnum = pgEnum('notification_status', ['PENDING', 'SENT', 'DELIVERED', 'FAILED', 'SCHEDULED'])
+export const notificationStatusEnum = pgEnum('notification_status', [
+  'PENDING',
+  'QUEUED',
+  'PROCESSING',
+  'SENT',
+  'DELIVERED',
+  'FAILED',
+  'PARTIAL',
+  'SCHEDULED',
+])
 
 export const deviceStatusEnum = pgEnum('device_status', ['ACTIVE', 'INACTIVE', 'EXPIRED'])
 
@@ -33,6 +42,7 @@ export const hookEventEnum = pgEnum('hookEvent', [
   'NOTIFICATION_SENT',
   'NOTIFICATION_DELIVERED',
   'NOTIFICATION_FAILED',
+  'NOTIFICATION_OPENED',
   'NOTIFICATION_CLICKED',
   'WORKFLOW_COMPLETED',
   'WORKFLOW_FAILED',

app/server/database/schema/notification.ts

@@ -2,7 +2,7 @@ import { index, integer, pgTable, text, uuid } from 'drizzle-orm/pg-core'
 import { createInsertSchema, createSelectSchema } from 'drizzle-zod'
 import { customJsonb, customTimestamp, uuidv7Generator } from '../shared'
 import { app } from './app'
-import { notificationStatusEnum } from './enums'
+import { channelTypeEnum, notificationStatusEnum } from './enums'
 
 export const notification = pgTable('notification', {
   id: uuid().primaryKey().$defaultFn(uuidv7Generator),
@@ -16,6 +16,9 @@ export const notification = pgTable('notification', {
   icon: text(),
   image: text(),
   imageUrl: text(),
+  channelType: channelTypeEnum(),
+  channelId: uuid(),
+  contactIds: customJsonb(),
   targetDevices: customJsonb(),
   platforms: customJsonb(),
   scheduledAt: customTimestamp(),