R50 通知推送：dispatcher 投递服务、业务事件接入与员工推送开关

2026-04-05 11:21:39 +08:00
@@ -3,10 +3,10 @@ const { z } = require("zod");
 const { AppError } = require("../utils/AppError");
 const { buildAuditLogData } = require("../utils/audit");
 const { sendData } = require("../utils/reply");
-const { createNotifications } = require("../services/notifications");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
+const { createNotificationsWithRecords } = require("../services/notifications");
 const { sendNotificationEmailSafe } = require("../services/email");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
+const { flushNotificationDispatchTasks } = require("../services/notificationDispatcher");

 const UUID_SCHEMA = z.string().uuid();
 const NOTE_SCHEMA = z.string().trim().max(500).nullable().optional();
@@ -204,6 +204,7 @@ async function assetMovementsRoutes(fastify) {
      const body = parseBody(ASSIGN_BODY_SCHEMA, request.body);
      const note = normalizeNote(body.note);
      const emailTasks = [];
+      const dispatchTasks = [];

      const result = await fastify.prisma.$transaction(async (tx) => {
        const asset = await ensureAssetExists(tx, assetId);
@@ -261,7 +262,7 @@ async function assetMovementsRoutes(fastify) {
        const recipients = uniq([employee.id]);
        const title = `【资产领用】资产卡片 ${asset.asset_no}`;
        const bodyText = `资产卡片 ${asset.asset_no} 已分配给你，请在资产台账中查看。`;
-        await createNotifications(tx, {
+        const notifications = await createNotificationsWithRecords(tx, {
          recipientEmployeeIds: recipients,
          type: NOTIFICATION_TYPES.assetAssigned,
          title,
@@ -269,18 +270,14 @@ async function assetMovementsRoutes(fastify) {
          targetType: "asset_card",
          targetId: asset.id,
        });
-        await enqueueImNotifications(tx, {
-          config: fastify.config,
-          logger: fastify.log,
-          recipientEmployeeIds: recipients,
+        dispatchTasks.push({
+          notifications,
          type: NOTIFICATION_TYPES.assetAssigned,
          title,
          body: bodyText,
          targetType: "asset_card",
          targetId: asset.id,
          targetPath: "/assets",
-          entityType: "asset_card",
-          entityId: asset.id,
          actorId: request.employee.id,
        });
        emailTasks.push({
@@ -303,6 +300,13 @@ async function assetMovementsRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, {
        assetId: result.updatedAsset.id,
@@ -326,6 +330,7 @@ async function assetMovementsRoutes(fastify) {
      const body = parseBody(RETURN_BODY_SCHEMA, request.body);
      const note = normalizeNote(body.note);
      const emailTasks = [];
+      const dispatchTasks = [];

      const result = await fastify.prisma.$transaction(async (tx) => {
        const asset = await ensureAssetExists(tx, assetId);
@@ -368,7 +373,7 @@ async function assetMovementsRoutes(fastify) {
        const recipients = uniq([asset.assignee_employee_id]);
        const title = `【资产归还】资产卡片 ${asset.asset_no}`;
        const bodyText = `资产卡片 ${asset.asset_no} 已完成归还。`;
-        await createNotifications(tx, {
+        const notifications = await createNotificationsWithRecords(tx, {
          recipientEmployeeIds: recipients,
          type: NOTIFICATION_TYPES.assetReturned,
          title,
@@ -376,18 +381,14 @@ async function assetMovementsRoutes(fastify) {
          targetType: "asset_card",
          targetId: asset.id,
        });
-        await enqueueImNotifications(tx, {
-          config: fastify.config,
-          logger: fastify.log,
-          recipientEmployeeIds: recipients,
+        dispatchTasks.push({
+          notifications,
          type: NOTIFICATION_TYPES.assetReturned,
          title,
          body: bodyText,
          targetType: "asset_card",
          targetId: asset.id,
          targetPath: "/assets",
-          entityType: "asset_card",
-          entityId: asset.id,
          actorId: request.employee.id,
        });
        emailTasks.push({
@@ -410,6 +411,13 @@ async function assetMovementsRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, {
        assetId: result.updatedAsset.id,
@@ -436,6 +444,7 @@ async function assetMovementsRoutes(fastify) {
      const hasLocationField = Object.prototype.hasOwnProperty.call(body, "toLocation");
      const nextLocation = hasLocationField ? normalizeLocation(body.toLocation) : undefined;
      const emailTasks = [];
+      const dispatchTasks = [];

      const result = await fastify.prisma.$transaction(async (tx) => {
        const asset = await ensureAssetExists(tx, assetId);
@@ -531,7 +540,7 @@ async function assetMovementsRoutes(fastify) {
        ]);
        const title = `【资产调拨】资产卡片 ${asset.asset_no}`;
        const bodyText = `资产卡片 ${asset.asset_no} 已完成调拨，请在资产台账中查看最新归属与位置。`;
-        await createNotifications(tx, {
+        const notifications = await createNotificationsWithRecords(tx, {
          recipientEmployeeIds: recipients,
          type: NOTIFICATION_TYPES.assetTransferred,
          title,
@@ -539,18 +548,14 @@ async function assetMovementsRoutes(fastify) {
          targetType: "asset_card",
          targetId: asset.id,
        });
-        await enqueueImNotifications(tx, {
-          config: fastify.config,
-          logger: fastify.log,
-          recipientEmployeeIds: recipients,
+        dispatchTasks.push({
+          notifications,
          type: NOTIFICATION_TYPES.assetTransferred,
          title,
          body: bodyText,
          targetType: "asset_card",
          targetId: asset.id,
          targetPath: "/assets",
-          entityType: "asset_card",
-          entityId: asset.id,
          actorId: request.employee.id,
        });
        emailTasks.push({
@@ -573,6 +578,13 @@ async function assetMovementsRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, {
        assetId: result.updatedAsset.id,
@@ -3,11 +3,11 @@ const { z } = require("zod");
 const { AppError } = require("../utils/AppError");
 const { buildAuditLogData } = require("../utils/audit");
 const { issueDocumentNumber } = require("../services/documentSequence");
-const { createNotifications, resolveRecipientsForAssignee } = require("../services/notifications");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
+const { createNotificationsWithRecords, resolveRecipientsForAssignee } = require("../services/notifications");
 const { sendNotificationEmailSafe } = require("../services/email");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
 const { sendData } = require("../utils/reply");
+const { flushNotificationDispatchTasks } = require("../services/notificationDispatcher");

 const UUID_SCHEMA = z.string().uuid();
 const INVENTORY_SCOPE_TYPES = ["all", "legal_entity", "cost_center", "location"];
@@ -517,6 +517,7 @@ async function inventoryChecksRoutes(fastify) {
    async (request, reply) => {
      const inventoryCheckId = parseUuid("id", request.params?.id);
      const emailTasks = [];
+      const dispatchTasks = [];
      const inventoryCheck = await fastify.prisma.$transaction(async (tx) => {
        const current = await ensureInventoryCheck(tx, inventoryCheckId);
        ensureCheckStatus(current, "counting");
@@ -581,7 +582,7 @@ async function inventoryChecksRoutes(fastify) {
        const title = `【盘点完成】盘点单 ${updated.check_no}`;
        const bodyText = `盘点单 ${updated.check_no} 已完成，请在资产盘点中查看结果。`;

-        await createNotifications(tx, {
+        const notifications = await createNotificationsWithRecords(tx, {
          recipientEmployeeIds: recipients,
          type: NOTIFICATION_TYPES.inventoryCheckCompleted,
          title,
@@ -589,19 +590,14 @@ async function inventoryChecksRoutes(fastify) {
          targetType: "inventory_check",
          targetId: updated.id,
        });
-
-        await enqueueImNotifications(tx, {
-          config: fastify.config,
-          logger: fastify.log,
-          recipientEmployeeIds: recipients,
+        dispatchTasks.push({
+          notifications,
          type: NOTIFICATION_TYPES.inventoryCheckCompleted,
          title,
          body: bodyText,
          targetType: "inventory_check",
          targetId: updated.id,
          targetPath: "/inventory-checks",
-          entityType: "inventory_check",
-          entityId: updated.id,
          actorId: request.employee.id,
        });

@@ -625,6 +621,13 @@ async function inventoryChecksRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, { inventoryCheck: mapInventoryCheck(inventoryCheck) });
    },
@@ -1,5 +1,33 @@
+const { z } = require("zod");
+
+const { AppError } = require("../utils/AppError");
+const { buildAuditLogData } = require("../utils/audit");
 const { sendData } = require("../utils/reply");

+const NOTIFICATION_PREFERENCES_SCHEMA = z
+  .object({
+    notificationPushEnabled: z.boolean(),
+  })
+  .strict();
+
+function buildValidationError(details) {
+  return new AppError({
+    code: "VALID_001",
+    message: "参数校验失败",
+    statusCode: 400,
+    details,
+  });
+}
+
+function parseNotificationPreferences(body) {
+  const parsed = NOTIFICATION_PREFERENCES_SCHEMA.safeParse(body ?? {});
+  if (!parsed.success) {
+    throw buildValidationError(parsed.error.flatten());
+  }
+
+  return parsed.data;
+}
+
 async function meRoutes(fastify) {
  fastify.get(
    "/me",
@@ -18,6 +46,7 @@ async function meRoutes(fastify) {
          department_id: true,
          direct_manager_id: true,
          status: true,
+          notification_push_enabled: true,
          roles: { select: { role_code: true } },
          wecom_identity_binding: { select: { external_user_id: true, created_at: true } },
        },
@@ -33,12 +62,49 @@ async function meRoutes(fastify) {
        departmentId: employee?.department_id ?? "",
        directManagerId: employee?.direct_manager_id ?? null,
        status: employee?.status ?? "active",
+        notificationPushEnabled: employee?.notification_push_enabled ?? true,
        roleCodes,
        wecomBinding: employee?.wecom_identity_binding ? { externalUserId: employee.wecom_identity_binding.external_user_id, createdAt: employee.wecom_identity_binding.created_at?.toISOString?.() ?? null } : null,
      });
    },
  );
+
+  fastify.patch(
+    "/me/notification-preferences",
+    {
+      preHandler: fastify.authenticate,
+    },
+    async (request, reply) => {
+      const body = parseNotificationPreferences(request.body);
+      const employee = await fastify.prisma.employee.update({
+        where: { id: request.employee.id },
+        data: {
+          notification_push_enabled: body.notificationPushEnabled,
+          updated_by: request.employee.id,
+        },
+        select: {
+          id: true,
+          notification_push_enabled: true,
+        },
+      });
+
+      await fastify.prisma.auditLog.create({
+        data: buildAuditLogData(request, {
+          actorId: request.employee.id,
+          action: "EMPLOYEE_NOTIFICATION_PREFERENCE_UPDATED",
+          targetType: "employee",
+          targetId: employee.id,
+          extra: {
+            notificationPushEnabled: employee.notification_push_enabled,
+          },
+        }),
+      });
+
+      return sendData(reply, request, {
+        notificationPushEnabled: employee.notification_push_enabled,
+      });
+    },
+  );
 }

 module.exports = meRoutes;
-
@@ -19,6 +19,13 @@ const LIST_QUERY_SCHEMA = z.object({
  pageSize: z.coerce.number().int().min(1).max(100).optional(),
  unreadOnly: z.enum(['0', '1', 'true', 'false']).optional(),
 });
+const DELIVERY_STATUS_VALUES = Object.freeze(['pending', 'success', 'failed']);
+const DELIVERY_QUERY_SCHEMA = z.object({
+  notificationId: z.string().uuid().optional(),
+  channelId: z.string().uuid().optional(),
+  status: z.enum(DELIVERY_STATUS_VALUES).optional(),
+  limit: z.coerce.number().int().min(1).max(200).optional(),
+});

 const CHANNEL_ID_SCHEMA = z.string().uuid();
 const ROLE_CODE_VALUES = Object.freeze(Object.keys(ROLE_PERMISSIONS));
@@ -141,6 +148,20 @@ function parseListQuery(query) {
  };
 }

+function parseDeliveryQuery(query) {
+  const parsed = DELIVERY_QUERY_SCHEMA.safeParse(query ?? {});
+  if (!parsed.success) {
+    throw buildValidationError(parsed.error.flatten());
+  }
+
+  return {
+    notificationId: parsed.data.notificationId ?? null,
+    channelId: parsed.data.channelId ?? null,
+    status: parsed.data.status ?? null,
+    limit: parsed.data.limit ?? 200,
+  };
+}
+
 function parseNotificationId(value) {
  const parsed = CHANNEL_ID_SCHEMA.safeParse(value);
  if (!parsed.success) {
@@ -249,6 +270,54 @@ function mapNotificationChannel(record, options = {}) {
  };
 }

+function mapNotificationDeliverySummary(record) {
+  return {
+    notificationId: record.notification_id,
+    channelId: record.channel_id,
+    channelName: record.channel?.name ?? '-',
+    channelType: record.channel?.channel_type ?? 'wecom',
+    notificationType: record.notification?.type ?? '',
+    notificationTitle: record.notification?.title ?? '',
+    status: record.status,
+    attemptCount: 1,
+    latestAttemptAt: record.attempted_at?.toISOString?.() ?? null,
+    resultCode: record.result_code ?? null,
+    errorMessage: record.error_message ?? null,
+    responseBody: record.response_body ?? null,
+    notificationCreatedAt: record.notification?.created_at?.toISOString?.() ?? null,
+  };
+}
+
+function summarizeDeliveryAttempts(items, query) {
+  const summaries = [];
+  const summaryMap = new Map();
+  const skippedKeys = new Set();
+
+  for (const item of items) {
+    const key = `${item.notification_id}:${item.channel_id}`;
+    if (skippedKeys.has(key)) {
+      continue;
+    }
+
+    const current = summaryMap.get(key);
+    if (!current) {
+      if (query.status && item.status !== query.status) {
+        skippedKeys.add(key);
+        continue;
+      }
+
+      const summary = mapNotificationDeliverySummary(item);
+      summaryMap.set(key, summary);
+      summaries.push(summary);
+      continue;
+    }
+
+    current.attemptCount += 1;
+  }
+
+  return summaries;
+}
+
 function handleNotificationChannelWriteError(error) {
  if (error instanceof AppError) {
    throw error;
@@ -412,6 +481,51 @@ async function notificationsRoutes(fastify) {
    },
  );

+  fastify.get(
+    '/notifications/deliveries',
+    {
+      preHandler: fastify.requirePermission('notification.channel.manage'),
+      schema: {
+        tags: ['notifications'],
+        summary: 'List Notification Deliveries',
+      },
+    },
+    async (request, reply) => {
+      const query = parseDeliveryQuery(request.query);
+      const where = {
+        ...(query.notificationId ? { notification_id: query.notificationId } : {}),
+        ...(query.channelId ? { channel_id: query.channelId } : {}),
+      };
+
+      const items = await fastify.prisma.notificationDelivery.findMany({
+        where,
+        orderBy: [{ attempted_at: 'desc' }, { attempt_no: 'desc' }],
+        take: query.limit,
+        include: {
+          channel: {
+            select: {
+              id: true,
+              channel_type: true,
+              name: true,
+            },
+          },
+          notification: {
+            select: {
+              id: true,
+              type: true,
+              title: true,
+              created_at: true,
+            },
+          },
+        },
+      });
+
+      return sendData(reply, request, {
+        items: summarizeDeliveryAttempts(items, query),
+      });
+    },
+  );
+
  fastify.get(
    '/notifications/channels',
    {
@@ -26,14 +26,14 @@ const {
 } = require("../services/integrations/erpProcurement");
 const { reserveBudgetForEntity, releaseReservedBudgetIfAny } = require("../services/purchaseBudget");
 const { sendNotificationEmailSafe } = require("../services/email");
-const { createNotifications, resolveRecipientsForAssignee } = require("../services/notifications");
+const { createNotificationsWithRecords, resolveRecipientsForAssignee } = require("../services/notifications");
 const {
  buildPendingTaskCopy,
  buildWorkflowEntityMeta,
 } = require("../services/notifications/workflowCopy");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
 const { openReadStream, StorageNotFoundError, writeBuffer } = require("../services/storage");
+const { dispatchNotificationBatch } = require("../services/notificationDispatcher");

 const UUID_SCHEMA = z.string().uuid();
 const DATE_ONLY_SCHEMA = z.string().regex(/^\d{4}-\d{2}-\d{2}$/);
@@ -896,8 +896,8 @@ function buildSkippedNotificationResult(reason) {
  return { skipped: true, reason };
 }

-function buildSkippedImResult(reason) {
-  return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: reason };
+function buildSkippedDispatchResult(reason) {
+  return { deliveredChannels: [], skippedReason: reason, notificationCount: 0, eligibleRecipientCount: 0, matchedChannelCount: 0 };
 }

 function buildSkippedEmailResult(reason, failed = 0) {
@@ -934,8 +934,8 @@ async function finalizePurchaseSubmitSideEffects({
  actorId,
  loadSummary = loadPurchaseSubmitSummary,
  resolveRecipients = resolveRecipientsForAssignee,
-  createAppNotifications = createNotifications,
-  enqueueIm = enqueueImNotifications,
+  createAppNotifications = createNotificationsWithRecords,
+  dispatchNotifications = dispatchNotificationBatch,
  sendEmailSafe = sendNotificationEmailSafe,
 }) {
  let copy = buildPendingTaskCopy(
@@ -986,7 +986,7 @@ async function finalizePurchaseSubmitSideEffects({
    return {
      recipientEmployeeIds: [],
      notification: buildSkippedNotificationResult("recipient_resolution_failed"),
-      im: buildSkippedImResult("recipient_resolution_failed"),
+      dispatch: buildSkippedDispatchResult("recipient_resolution_failed"),
      email: buildSkippedEmailResult("recipient_resolution_failed"),
    };
  }
@@ -995,14 +995,15 @@ async function finalizePurchaseSubmitSideEffects({
    return {
      recipientEmployeeIds: [],
      notification: buildSkippedNotificationResult("no_recipients"),
-      im: buildSkippedImResult("no_recipients"),
+      dispatch: buildSkippedDispatchResult("no_recipients"),
      email: buildSkippedEmailResult("no_recipients"),
    };
  }

  let notification = buildSkippedNotificationResult("not_attempted");
+  let notificationRecords = [];
  try {
-    notification = await createAppNotifications(prisma, {
+    notificationRecords = await createAppNotifications(prisma, {
      recipientEmployeeIds,
      type: NOTIFICATION_TYPES.workflowTaskAssigned,
      title,
@@ -1010,6 +1011,7 @@ async function finalizePurchaseSubmitSideEffects({
      targetType: "workflow_task",
      targetId: workflowTaskId,
    });
+    notification = { count: notificationRecords.length };
  } catch (error) {
    logger?.error?.(
      {
@@ -1024,20 +1026,20 @@ async function finalizePurchaseSubmitSideEffects({
    notification = buildSkippedNotificationResult("notification_failed");
  }

-  let im = buildSkippedImResult("not_attempted");
+  let dispatch = buildSkippedDispatchResult("not_attempted");
  try {
-    im = await enqueueIm(prisma, {
+    dispatch = await dispatchNotifications({
+      prisma,
      config,
      logger,
-      recipientEmployeeIds,
+      requestId,
+      notifications: notificationRecords,
      type: NOTIFICATION_TYPES.workflowTaskAssigned,
      title,
      body,
      targetType: "workflow_task",
      targetId: workflowTaskId,
      targetPath,
-      entityType: "purchase_request",
-      entityId: purchaseRequestId,
      actorId,
    });
  } catch (error) {
@@ -1049,9 +1051,9 @@ async function finalizePurchaseSubmitSideEffects({
        workflowTaskId,
        recipients: recipientEmployeeIds.length,
      },
-      "Failed to enqueue purchase submit IM notifications",
+      "Failed to dispatch purchase submit notifications",
    );
-    im = buildSkippedImResult("enqueue_im_failed");
+    dispatch = buildSkippedDispatchResult("dispatch_failed");
  }

  let email = buildSkippedEmailResult("not_attempted");
@@ -1085,7 +1087,7 @@ async function finalizePurchaseSubmitSideEffects({
  return {
    recipientEmployeeIds,
    notification,
-    im,
+    dispatch,
    email,
  };
 }
@@ -5,17 +5,17 @@ const { buildAuditLogData } = require("../utils/audit");
 const { currentDateString, parseDateOnlyString } = require("../utils/date");
 const { findActiveDelegationForDelegate } = require("../utils/delegation");
 const {
-  createNotifications,
+  createNotificationsWithRecords,
  resolveRecipientsForAssignee,
 } = require("../services/notifications");
 const {
  buildPendingTaskCopy,
  buildWorkflowEntityMeta,
 } = require("../services/notifications/workflowCopy");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
 const { sendNotificationEmailSafe } = require("../services/email");
 const { enqueueReimbursementJournalExport } = require("../services/integrations/erpJournal");
+const { dispatchNotificationBatch, flushNotificationDispatchTasks } = require("../services/notificationDispatcher");
 const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
 const { ensureCutoverAllowed } = require("../services/cutover");
 const { resolveApplicableBudgetPoolByContext } = require("../services/budgetPool");
@@ -450,8 +450,8 @@ function buildSkippedNotificationResult(reason) {
  return { skipped: true, reason };
 }

-function buildSkippedImResult(reason) {
-  return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: reason };
+function buildSkippedDispatchResult(reason) {
+  return { deliveredChannels: [], skippedReason: reason, notificationCount: 0, eligibleRecipientCount: 0, matchedChannelCount: 0 };
 }

 function buildSkippedEmailResult(reason, failed = 0) {
@@ -488,8 +488,8 @@ async function finalizeReimbursementSubmitSideEffects({
  actorId,
  loadSummary = loadReimbursementSubmitSummary,
  resolveRecipients = resolveRecipientsForAssignee,
-  createAppNotifications = createNotifications,
-  enqueueIm = enqueueImNotifications,
+  createAppNotifications = createNotificationsWithRecords,
+  dispatchNotifications = dispatchNotificationBatch,
  sendEmailSafe = sendNotificationEmailSafe,
 }) {
  let copy = buildPendingTaskCopy(
@@ -540,7 +540,7 @@ async function finalizeReimbursementSubmitSideEffects({
    return {
      recipientEmployeeIds: [],
      notification: buildSkippedNotificationResult("recipient_resolution_failed"),
-      im: buildSkippedImResult("recipient_resolution_failed"),
+      dispatch: buildSkippedDispatchResult("recipient_resolution_failed"),
      email: buildSkippedEmailResult("recipient_resolution_failed"),
    };
  }
@@ -549,14 +549,15 @@ async function finalizeReimbursementSubmitSideEffects({
    return {
      recipientEmployeeIds: [],
      notification: buildSkippedNotificationResult("no_recipients"),
-      im: buildSkippedImResult("no_recipients"),
+      dispatch: buildSkippedDispatchResult("no_recipients"),
      email: buildSkippedEmailResult("no_recipients"),
    };
  }

  let notification = buildSkippedNotificationResult("not_attempted");
+  let notificationRecords = [];
  try {
-    notification = await createAppNotifications(prisma, {
+    notificationRecords = await createAppNotifications(prisma, {
      recipientEmployeeIds,
      type: NOTIFICATION_TYPES.workflowTaskAssigned,
      title,
@@ -564,6 +565,7 @@ async function finalizeReimbursementSubmitSideEffects({
      targetType: "workflow_task",
      targetId: workflowTaskId,
    });
+    notification = { count: notificationRecords.length };
  } catch (error) {
    logger?.error?.(
      {
@@ -578,20 +580,20 @@ async function finalizeReimbursementSubmitSideEffects({
    notification = buildSkippedNotificationResult("notification_failed");
  }

-  let im = buildSkippedImResult("not_attempted");
+  let dispatch = buildSkippedDispatchResult("not_attempted");
  try {
-    im = await enqueueIm(prisma, {
+    dispatch = await dispatchNotifications({
+      prisma,
      config,
      logger,
-      recipientEmployeeIds,
+      requestId,
+      notifications: notificationRecords,
      type: NOTIFICATION_TYPES.workflowTaskAssigned,
      title,
      body,
      targetType: "workflow_task",
      targetId: workflowTaskId,
      targetPath,
-      entityType: "reimbursement",
-      entityId: reimbursementId,
      actorId,
    });
  } catch (error) {
@@ -603,9 +605,9 @@ async function finalizeReimbursementSubmitSideEffects({
        workflowTaskId,
        recipients: recipientEmployeeIds.length,
      },
-      "Failed to enqueue reimbursement submit IM notifications",
+      "Failed to dispatch reimbursement submit notifications",
    );
-    im = buildSkippedImResult("enqueue_im_failed");
+    dispatch = buildSkippedDispatchResult("dispatch_failed");
  }

  let email = buildSkippedEmailResult("not_attempted");
@@ -639,7 +641,7 @@ async function finalizeReimbursementSubmitSideEffects({
  return {
    recipientEmployeeIds,
    notification,
-    im,
+    dispatch,
    email,
  };
 }
@@ -2042,6 +2044,7 @@ async function reimbursementRoutes(fastify) {
      const payInput = parsePayBody(request.body);
      let response = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const reimbursementRows = await tx.$queryRaw`
@@ -2196,7 +2199,7 @@ async function reimbursementRoutes(fastify) {
            const recipients = Array.from(
              new Set([reimbursement.owner_id, reimbursement.submitter_id].filter(Boolean)),
            );
-            await createNotifications(tx, {
+            const notifications = await createNotificationsWithRecords(tx, {
              recipientEmployeeIds: recipients,
              type: NOTIFICATION_TYPES.reimbursementPaid,
              title: `【已付款】报销单 ${reimbursement.form_number ?? reimbursementId}`,
@@ -2204,19 +2207,14 @@ async function reimbursementRoutes(fastify) {
              targetType: "reimbursement_form",
              targetId: reimbursementId,
            });
-
-            await enqueueImNotifications(tx, {
-              config: fastify.config,
-              logger: fastify.log,
-              recipientEmployeeIds: recipients,
+            dispatchTasks.push({
+              notifications,
              type: NOTIFICATION_TYPES.reimbursementPaid,
              title: `【已付款】报销单 ${reimbursement.form_number ?? reimbursementId}`,
              body: `报销单 ${reimbursement.form_number ?? reimbursementId} 已完成付款。`,
              targetType: "reimbursement_form",
              targetId: reimbursementId,
              targetPath: `/reimbursements/${reimbursementId}`,
-              entityType: "reimbursement",
-              entityId: reimbursementId,
              actorId: request.employee.id,
            });

@@ -2276,6 +2274,13 @@ async function reimbursementRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, response);
    },
@@ -9,12 +9,11 @@ const { resolveApplicableBudgetPoolByContext } = require("../services/budgetPool
 const { reserveBudgetForEntity, releaseReservedBudgetIfAny } = require("../services/purchaseBudget");
 const { parseWorkflowSteps, resolveActionableStep } = require("../utils/workflow");
 const { sendNotificationEmailSafe } = require("../services/email");
-const { createNotifications, resolveRecipientsForAssignee } = require("../services/notifications");
+const { createNotificationsWithRecords, resolveRecipientsForAssignee } = require("../services/notifications");
 const {
  buildPendingTaskCopy,
  buildWorkflowEntityMeta,
 } = require("../services/notifications/workflowCopy");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
 const { buildTravelPaymentTrack } = require("../services/travelPayments");
 const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
 const { ensureCutoverAllowed } = require("../services/cutover");
@@ -24,6 +23,7 @@ const {
  resolveEntityPrecheck,
 } = require("../services/feeControl");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
+const { flushNotificationDispatchTasks } = require("../services/notificationDispatcher");

 const UUID_SCHEMA = z.string().uuid();
 const BOOLEAN_QUERY_SCHEMA = z.enum(["0", "1", "true", "false"]).optional();
@@ -929,6 +929,7 @@ async function createTravelRequestsRoutes(fastify) {
      const travelRequestId = parseTravelRequestId(request.params?.id);
      let response = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const travelRequest = await loadTravelRequestForUpdate(tx, travelRequestId);
@@ -1113,7 +1114,7 @@ async function createTravelRequestsRoutes(fastify) {
        );
        const { title, body } = pendingCopy;

-        await createNotifications(tx, {
+        const notifications = await createNotificationsWithRecords(tx, {
          recipientEmployeeIds: recipients,
          type: NOTIFICATION_TYPES.workflowTaskAssigned,
          title,
@@ -1121,19 +1122,14 @@ async function createTravelRequestsRoutes(fastify) {
          targetType: "workflow_task",
          targetId: workflowTask.id,
        });
-
-        await enqueueImNotifications(tx, {
-          config: fastify.config,
-          logger: fastify.log,
-          recipientEmployeeIds: recipients,
+        dispatchTasks.push({
+          notifications,
          type: NOTIFICATION_TYPES.workflowTaskAssigned,
          title,
          body,
          targetType: "workflow_task",
          targetId: workflowTask.id,
          targetPath: `/workflow/tasks/${workflowTask.id}`,
-          entityType: "travel_request",
-          entityId: travelRequestId,
          actorId: request.employee.id,
        });

@@ -1184,6 +1180,13 @@ async function createTravelRequestsRoutes(fastify) {
          ...emailTask,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, response);
    },
@@ -13,7 +13,7 @@ const {
  loadPurchaseRequestErpContext,
 } = require("../services/integrations/erpProcurement");
 const {
-  createNotifications,
+  createNotificationsWithRecords,
  resolveRecipientsForAssignee,
 } = require("../services/notifications");
 const {
@@ -24,9 +24,9 @@ const {
  buildWorkflowEntityMeta,
  normalizeWorkflowEntityType,
 } = require("../services/notifications/workflowCopy");
-const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
 const { sendNotificationEmailSafe } = require("../services/email");
 const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
+const { flushNotificationDispatchTasks } = require("../services/notificationDispatcher");
 const {
  findActiveDelegationForDelegate,
  listActiveDelegationPrincipals,
@@ -577,10 +577,9 @@ function resolveWorkflowOutcomeNotificationType(entityType, outcome) {

 async function queueTaskAssignedNotifications({
  tx,
-  config,
-  logger,
  actorId,
  emailTasks,
+  dispatchTasks,
  recipientEmployeeIds,
  title,
  body,
@@ -593,7 +592,7 @@ async function queueTaskAssignedNotifications({
    return;
  }

-  await createNotifications(tx, {
+  const notifications = await createNotificationsWithRecords(tx, {
    recipientEmployeeIds: recipients,
    type: NOTIFICATION_TYPES.workflowTaskAssigned,
    title,
@@ -601,19 +600,14 @@ async function queueTaskAssignedNotifications({
    targetType: 'workflow_task',
    targetId: taskId,
  });
-
-  await enqueueImNotifications(tx, {
-    config,
-    logger,
-    recipientEmployeeIds: recipients,
+  dispatchTasks.push({
+    notifications,
    type: NOTIFICATION_TYPES.workflowTaskAssigned,
    title,
    body,
    targetType: 'workflow_task',
    targetId: taskId,
    targetPath: `/workflow/tasks/${taskId}`,
-    entityType,
-    entityId,
    actorId,
  });

@@ -628,10 +622,9 @@ async function queueTaskAssignedNotifications({

 async function queueWorkflowOutcomeNotifications({
  tx,
-  config,
-  logger,
  actorId,
  emailTasks,
+  dispatchTasks,
  workflowInstance,
  outcome,
  comment = null,
@@ -656,7 +649,7 @@ async function queueWorkflowOutcomeNotifications({
        ? buildRejectedCopy(entityMeta, comment)
        : buildReturnedCopy(entityMeta, comment);

-  await createNotifications(tx, {
+  const notifications = await createNotificationsWithRecords(tx, {
    recipientEmployeeIds: recipients,
    type: notificationType,
    title: copy.title,
@@ -664,19 +657,14 @@ async function queueWorkflowOutcomeNotifications({
    targetType: entityMeta.targetType,
    targetId: workflowInstance.entity_id,
  });
-
-  await enqueueImNotifications(tx, {
-    config,
-    logger,
-    recipientEmployeeIds: recipients,
+  dispatchTasks.push({
+    notifications,
    type: notificationType,
    title: copy.title,
    body: copy.body,
    targetType: entityMeta.targetType,
    targetId: workflowInstance.entity_id,
    targetPath: entityMeta.detailPath,
-    entityType: normalizedEntityType,
-    entityId: workflowInstance.entity_id,
    actorId,
  });

@@ -1355,6 +1343,7 @@ async function workflowRoutes(fastify) {
      const taskId = parseTaskId(request.params?.taskId);
      let result = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const task = await lockWorkflowTask(tx, taskId);
@@ -1458,10 +1447,9 @@ async function workflowRoutes(fastify) {

            await queueTaskAssignedNotifications({
              tx,
-              config: fastify.config,
-              logger: fastify.log,
              actorId: request.employee.id,
              emailTasks,
+              dispatchTasks,
              recipientEmployeeIds: recipients,
              title: pendingCopy.title,
              body: pendingCopy.body,
@@ -1613,10 +1601,9 @@ async function workflowRoutes(fastify) {
        try {
          await queueWorkflowOutcomeNotifications({
            tx,
-            config: fastify.config,
-            logger: fastify.log,
            actorId: request.employee.id,
            emailTasks,
+            dispatchTasks,
            workflowInstance,
            outcome: 'approved',
          });
@@ -1647,6 +1634,13 @@ async function workflowRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, result);
    },
@@ -1662,6 +1656,7 @@ async function workflowRoutes(fastify) {
      const comment = parseRequiredComment(request.body);
      let result = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const task = await lockWorkflowTask(tx, taskId);
@@ -1797,10 +1792,9 @@ async function workflowRoutes(fastify) {
        try {
          await queueWorkflowOutcomeNotifications({
            tx,
-            config: fastify.config,
-            logger: fastify.log,
            actorId: request.employee.id,
            emailTasks,
+            dispatchTasks,
            workflowInstance,
            outcome: 'rejected',
            comment,
@@ -1831,6 +1825,13 @@ async function workflowRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, result);
    },
@@ -1846,6 +1847,7 @@ async function workflowRoutes(fastify) {
      const comment = parseRequiredComment(request.body);
      let result = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const task = await lockWorkflowTask(tx, taskId);
@@ -1981,10 +1983,9 @@ async function workflowRoutes(fastify) {
        try {
          await queueWorkflowOutcomeNotifications({
            tx,
-            config: fastify.config,
-            logger: fastify.log,
            actorId: request.employee.id,
            emailTasks,
+            dispatchTasks,
            workflowInstance,
            outcome: 'returned',
            comment,
@@ -2015,6 +2016,13 @@ async function workflowRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, result);
    },
@@ -2030,6 +2038,7 @@ async function workflowRoutes(fastify) {
      const { toApproverId, reason } = parseTransferBody(request.body);
      let result = null;
      const emailTasks = [];
+      const dispatchTasks = [];

      await fastify.prisma.$transaction(async (tx) => {
        const task = await lockWorkflowTask(tx, taskId);
@@ -2121,10 +2130,9 @@ async function workflowRoutes(fastify) {
          const pendingCopy = buildPendingTaskCopy(entityMeta, { transfer: true });
          await queueTaskAssignedNotifications({
            tx,
-            config: fastify.config,
-            logger: fastify.log,
            actorId: request.employee.id,
            emailTasks,
+            dispatchTasks,
            recipientEmployeeIds: [toApproverId],
            title: pendingCopy.title,
            body: pendingCopy.body,
@@ -2160,6 +2168,13 @@ async function workflowRoutes(fastify) {
          ...task,
        });
      }
+      await flushNotificationDispatchTasks({
+        prisma: fastify.prisma,
+        config: fastify.config,
+        logger: fastify.log,
+        requestId: request.id,
+        tasks: dispatchTasks,
+      });

      return sendData(reply, request, result);
    },