hyb-oa/server/src/services/notificationDispatcher.js

const { resolveNotificationDomainScope } = require('../constants/notificationTypes');
const { sendInstantMessageByChannel } = require('./notifications/channels');
const { buildDetailUrl } = require('./notifications/detailUrl');
const { matchesNotificationChannel } = require('./notifications/imRouting');

function uniq(values) {
  return Array.from(new Set((values ?? []).filter(Boolean)));
}

function wait(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function normalizeJsonValue(value) {
  if (value == null) {
    return null;
  }

  if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') {
    return value;
  }

  try {
    return JSON.parse(JSON.stringify(value));
  } catch {
    return { value: String(value) };
  }
}

function trimMessage(value, maxLength = 1000) {
  return String(value ?? '').trim().slice(0, maxLength) || null;
}

function buildSkipResult(reason, extra = {}) {
  return {
    deliveredChannels: [],
    skippedReason: reason,
    notificationCount: 0,
    eligibleRecipientCount: 0,
    matchedChannelCount: 0,
    ...extra,
  };
}

async function resolveEligibleRecipients(prisma, recipientEmployeeIds) {
  const recipientIds = uniq(recipientEmployeeIds);
  if (!recipientIds.length) {
    return [];
  }

  return prisma.employee.findMany({
    where: {
      id: { in: recipientIds },
      status: 'active',
      notification_push_enabled: true,
    },
    select: {
      id: true,
      roles: {
        select: {
          role_code: true,
        },
      },
      wecom_identity_binding: {
        select: {
          external_user_id: true,
        },
      },
    },
  });
}

async function persistDeliveryAttempt({
  prisma,
  notificationId,
  channelId,
  attemptNo,
  status,
  resultCode = null,
  errorMessage = null,
  responseBody = null,
  attemptedAt = new Date(),
}) {
  const delivery = await prisma.notificationDelivery.create({
    data: {
      notification_id: notificationId,
      channel_id: channelId,
      status: 'pending',
      attempt_no: attemptNo,
      attempted_at: attemptedAt,
    },
  });

  return prisma.notificationDelivery.update({
    where: { id: delivery.id },
    data: {
      status,
      result_code: trimMessage(resultCode, 64),
      error_message: trimMessage(errorMessage, 2000),
      response_body: normalizeJsonValue(responseBody),
    },
  });
}

async function dispatchNotificationBatch({
  prisma,
  config,
  logger,
  requestId,
  notifications,
  type,
  title,
  body,
  targetType,
  targetId,
  targetPath = null,
  actorId = null,
  maxAttempts = 3,
  retryDelayMs = 5000,
  sendMessage = sendInstantMessageByChannel,
}) {
  const notificationItems = Array.isArray(notifications) ? notifications.filter(Boolean) : [];
  if (!notificationItems.length) {
    return buildSkipResult('no_notifications');
  }

  if (!config?.imNotifierEnabled) {
    logger?.info?.(
      {
        requestId,
        actorId,
        notificationType: type,
        notificationCount: notificationItems.length,
        skippedReason: 'im_notifier_disabled',
      },
      'Notification dispatch skipped',
    );
    return buildSkipResult('im_notifier_disabled', {
      notificationCount: notificationItems.length,
    });
  }

  const anchorNotification = notificationItems[0];
  const recipientIds = uniq(
    notificationItems.map((item) => item.recipient_employee_id ?? item.recipientEmployeeId ?? null),
  );
  const eligibleRecipients = await resolveEligibleRecipients(prisma, recipientIds);
  if (!eligibleRecipients.length) {
    logger?.info?.(
      {
        requestId,
        actorId,
        notificationType: type,
        notificationCount: notificationItems.length,
        skippedReason: 'no_eligible_recipients',
      },
      'Notification dispatch skipped',
    );
    return buildSkipResult('no_eligible_recipients', {
      notificationCount: notificationItems.length,
    });
  }

  const recipientRoleCodes = uniq(
    eligibleRecipients.flatMap((recipient) => recipient.roles.map((role) => role.role_code)),
  );
  const recipientExternalUserIds = uniq(
    eligibleRecipients.map((recipient) => recipient.wecom_identity_binding?.external_user_id ?? null),
  );
  const domainScope = resolveNotificationDomainScope(type, targetType ?? anchorNotification.target_type ?? null);
  if (!domainScope) {
    logger?.warn?.(
      {
        requestId,
        actorId,
        notificationType: type,
        notificationCount: notificationItems.length,
        skippedReason: 'domain_scope_unresolved',
      },
      'Notification dispatch skipped',
    );
    return buildSkipResult('domain_scope_unresolved', {
      notificationCount: notificationItems.length,
      eligibleRecipientCount: eligibleRecipients.length,
    });
  }

  const channels = await prisma.notificationChannel.findMany({
    where: {
      enabled: true,
      channel_type: {
        in: ['wecom', 'wechat_work'],
      },
    },
    orderBy: [{ channel_type: 'asc' }, { name: 'asc' }],
  });
  const matchedChannels = channels.filter((channel) =>
    matchesNotificationChannel(channel, {
      notificationType: type,
      domainScope,
      recipientRoleCodes,
    }),
  );
  if (!matchedChannels.length) {
    logger?.info?.(
      {
        requestId,
        actorId,
        notificationType: type,
        notificationCount: notificationItems.length,
        eligibleRecipientCount: eligibleRecipients.length,
        skippedReason: 'no_matching_channel',
      },
      'Notification dispatch skipped',
    );
    return buildSkipResult('no_matching_channel', {
      notificationCount: notificationItems.length,
      eligibleRecipientCount: eligibleRecipients.length,
    });
  }

  const detailUrl = buildDetailUrl(config, targetPath);
  const deliveredChannels = [];
  for (const channel of matchedChannels) {
    let channelResult = null;

    for (let attemptNo = 1; attemptNo <= maxAttempts; attemptNo += 1) {
      let result;
      try {
        result = await sendMessage({
          config,
          channel,
          payload: {
            title,
            body,
          targetType,
          targetId,
          detailUrl,
          recipientEmployeeIds: recipientIds,
          recipientExternalUserIds,
        },
        prisma,
      });
      } catch (error) {
        result = {
          ok: false,
          retryable: true,
          code: 'dispatcher_unhandled_error',
          message: String(error?.message ?? error ?? 'dispatcher_unhandled_error'),
          response: null,
        };
      }

      const deliveryStatus = result.ok ? 'success' : 'failed';
      await persistDeliveryAttempt({
        prisma,
        notificationId: anchorNotification.id,
        channelId: channel.id,
        attemptNo,
        status: deliveryStatus,
        resultCode: result.code,
        errorMessage: result.ok ? null : result.message,
        responseBody: result.response ?? null,
      });

      logger?.info?.(
        {
          requestId,
          actorId,
          notificationType: type,
          channelId: channel.id,
          attemptNo,
          notificationCount: notificationItems.length,
          eligibleRecipientCount: eligibleRecipients.length,
          resultCode: result.code,
          ok: result.ok,
        },
        'Notification delivery attempt finished',
      );

      channelResult = {
        channelId: channel.id,
        channelName: channel.name,
        status: deliveryStatus,
        attemptNo,
        resultCode: result.code,
        errorMessage: result.ok ? null : trimMessage(result.message, 2000),
      };

      if (result.ok || !result.retryable || attemptNo >= maxAttempts) {
        break;
      }

      await wait(retryDelayMs);
    }

    deliveredChannels.push(channelResult);
  }

  return {
    deliveredChannels,
    skippedReason: null,
    notificationCount: notificationItems.length,
    eligibleRecipientCount: eligibleRecipients.length,
    matchedChannelCount: matchedChannels.length,
  };
}

async function flushNotificationDispatchTasks({
  prisma,
  config,
  logger,
  requestId,
  tasks,
}) {
  const results = [];
  for (const task of tasks ?? []) {
    try {
      results.push(
        await dispatchNotificationBatch({
          prisma,
          config,
          logger,
          requestId,
          ...task,
        }),
      );
    } catch (error) {
      logger?.error?.(
        {
          err: error,
          requestId,
          notificationType: task?.type ?? null,
          targetType: task?.targetType ?? null,
          targetId: task?.targetId ?? null,
        },
        'Notification dispatch failed',
      );
      results.push(
        buildSkipResult('dispatch_failed', {
          notificationCount: Array.isArray(task?.notifications) ? task.notifications.length : 0,
        }),
      );
    }
  }

  return results;
}

module.exports = {
  dispatchNotificationBatch,
  flushNotificationDispatchTasks,
  __testables: {
    buildSkipResult,
    normalizeJsonValue,
    resolveEligibleRecipients,
    trimMessage,
    wait,
  },
};