zhangshun/mydreamjob-server/src/main/java/com/mydreamjob/notification/NotificationService.java

package com.mydreamjob.notification;

import com.mydreamjob.auth.User;
import com.mydreamjob.auth.UserRepository;
import com.mydreamjob.common.exception.BusinessException;
import com.mydreamjob.common.exception.ErrorCode;
import com.mydreamjob.company.CompanyRepository;
import com.mydreamjob.company.VerifiedCompanyGuard;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Slf4j
@Service
@RequiredArgsConstructor
public class NotificationService {

    private final NotificationRepository notificationRepository;
    private final PushProvider pushProvider;
    private final NotificationPublisher notificationPublisher;
    private final UserRepository userRepository;
    private final CompanyRepository companyRepository;

    private static final int MAX_PAGE_SIZE = 50;

    /** 创建通知并触发推送 */
    @Transactional
    public Notification create(Long userId, String type, String title, String content) {
        return create(userId, type, title, content, null, null, null);
    }

    /**
     * 创建通知（带业务关联和幂等键）
     */
    @Transactional
    public Notification create(Long userId, String type, String title, String content,
                               String businessType, Long businessId, String idempotencyKey) {
        if (idempotencyKey != null && notificationRepository.existsByIdempotencyKey(idempotencyKey)) {
            log.info("Duplicate notification skipped: key={}", idempotencyKey);
            return null;
        }
        Notification n = Notification.builder()
                .userId(userId).type(type).title(title).content(content)
                .businessType(businessType).businessId(businessId)
                .idempotencyKey(idempotencyKey).read(false)
                .build();
        try {
            n = notificationRepository.save(n);
        } catch (org.springframework.dao.DataIntegrityViolationException e) {
            log.info("Idempotency key conflict: {}", idempotencyKey);
            return null;
        }
        try {
            pushProvider.send(userId, title, content);
        } catch (Exception e) {
            log.warn("Push failed for notification id={}: {}", n.getId(), e.getMessage());
        }
        return n;
    }

    /**
     * 发布通知事件到 RabbitMQ（异步落库）
     */
    public void publishEvent(Long userId, String type, String title, String content,
                             String businessType, Long businessId, String idempotencyKey) {
        notificationPublisher.publish(NotificationEvent.builder()
                .userId(userId).type(type).title(title).content(content)
                .businessType(businessType).businessId(businessId)
                .idempotencyKey(idempotencyKey)
                .build());
    }

    /** 分页查询用户通知 */
    public Page<Notification> list(Long userId, int page, int size) {
        requireVerifiedIfCompany(userId);
        if (page < 0 || size < 1 || size > MAX_PAGE_SIZE) throw new BusinessException(ErrorCode.BAD_REQUEST);
        return notificationRepository.findByUserIdOrderByCreatedAtDesc(userId, PageRequest.of(page, size));
    }

    /** 标记单条已读（幂等：已读也返回成功） */
    @Transactional
    public void markRead(Long userId, Long notificationId) {
        requireVerifiedIfCompany(userId);
        Notification n = notificationRepository.findById(notificationId)
                .orElseThrow(() -> new BusinessException(ErrorCode.NOTIFICATION_NOT_FOUND));
        if (!n.getUserId().equals(userId)) throw new BusinessException(ErrorCode.FORBIDDEN);
        if (!Boolean.TRUE.equals(n.getRead())) {
            n.setRead(true);
            n.setReadAt(LocalDateTime.now());
            notificationRepository.save(n);
        }
    }

    /** 标记全部已读（幂等） */
    @Transactional
    public int markAllRead(Long userId) {
        requireVerifiedIfCompany(userId);
        int count = 0;
        int pageIdx = 0;
        boolean hasMore = true;
        while (hasMore) {
            Page<Notification> unread = notificationRepository
                    .findByUserIdOrderByCreatedAtDesc(userId, PageRequest.of(pageIdx, 100));
            if (unread.isEmpty()) { hasMore = false; break; }
            boolean anyUpdated = false;
            for (Notification n : unread.getContent()) {
                if (!Boolean.TRUE.equals(n.getRead())) {
                    n.setRead(true);
                    n.setReadAt(LocalDateTime.now());
                    notificationRepository.save(n);
                    count++;
                    anyUpdated = true;
                }
            }
            hasMore = !unread.isLast();
            pageIdx++;
        }
        return count;
    }

    /** 未读数 */
    public long unreadCount(Long userId) {
        requireVerifiedIfCompany(userId);
        return notificationRepository.countUnreadByUserId(userId);
    }

    private void requireVerifiedIfCompany(Long userId) {
        User actor = userRepository.findByIdWithRoles(userId).orElse(null);
        VerifiedCompanyGuard.requireVerifiedIfCompany(actor,
                () -> companyRepository.findByOwnerUserId(userId).orElse(null));
    }

    public Map<String, Object> toView(Notification n) {
        Map<String, Object> v = new HashMap<>();
        v.put("id", n.getId());
        v.put("type", n.getType());
        v.put("title", n.getTitle());
        v.put("content", n.getContent());
        v.put("read", n.getRead());
        v.put("readAt", n.getReadAt());
        v.put("createdAt", n.getCreatedAt());
        return v;
    }
}