Notification grouping job: luồng kỹ thuật và lỗi count không xuống DB

await redisService.hIncrBy(key, "count", 1);
const added = await redisService.sAdd(actorsKey, userId);
if (added === 1) {
  await redisService.hSet(key, {
    authorId,
    postPublicId,
    notificationType,
    targetType,
    lastActorId: userId,
    userId: userId,
    updatedAt: Date.now().toString(),
  });
}

Nếu hash đang ở trạng thái { count: "8" } và {key}:actors đã có user đó, sAdd trả 0, nên hSet không chạy. Count tăng, metadata vẫn thiếu.

if (!data?.authorId) {
  await redisService.del(`${key}:queued`);
  continue;
}

Branch này nên được coi là poison payload. Nếu đã quyết định không xử lý được, cần xóa toàn bộ state liên quan hoặc đưa vào dead-letter/debug key. Chỉ xóa queued marker làm key có cơ hội được enqueue lại nhưng không có cơ chế tự sửa metadata.

model NotificationGroup {
  id          Int              @id @default(autoincrement())
  publicId    String           @unique @default(cuid()) @map("public_id")
  recipientId String           @map("recipient_id")
  type        NotificationType
  targetType  String           @map("target_type")
  targetId    String           @map("target_id")
  actorIds    Json             @map("actor_ids")
  count       Int              @default(1)
  isRead      Boolean          @default(false) @map("is_read")
  lastActorId String           @map("last_actor_id")
  lastEventAt DateTime         @map("last_event_at")

  @@unique([recipientId, type, targetType, targetId], map: "uq_notification_group_target")
  @@index([recipientId, isRead, lastEventAt])
  @@map("notification_groups")
}

Nếu muốn gom theo time window, unique key cần thêm field window bucket, ví dụ groupWindowKey. Với MVP hiện tại, một target một group là cách đơn giản nhất.

async upsertPostNotificationGroup(item: NotificationBatchItem, tx = prisma) {
  const existing = await tx.notificationGroup.findFirst({
    where: {
      recipientId: item.authorId,
      type: item.notificationType,
      targetType: item.targetType,
      targetId: item.postPublicId,
    },
    select: { id: true, actorIds: true, count: true },
  });

  const mergedActorIds = mergeUniqueJsonArray(existing?.actorIds, item.actorIds);

  if (!existing) {
    return tx.notificationGroup.create({ data: {
      recipientId: item.authorId,
      type: item.notificationType,
      targetType: item.targetType,
      targetId: item.postPublicId,
      actorIds: mergedActorIds,
      count: item.count,
      lastActorId: item.lastActorId,
      lastEventAt: new Date(),
      isRead: false,
    }});
  }

  return tx.notificationGroup.update({
    where: { id: existing.id },
    data: {
      actorIds: mergedActorIds,
      count: { increment: item.count },
      lastActorId: item.lastActorId,
      lastEventAt: new Date(),
      isRead: false,
    },
  });
}

Nếu thêm unique composite, có thể dùng upsert theo unique key. Nếu chưa thêm unique, dùng findFirst + update/create trong transaction, nhưng cần chú ý race condition khi nhiều worker cùng xử lý một group.