You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

472 lines
25 KiB

package com.mydreamjob.messaging;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mydreamjob.auth.User;
import com.mydreamjob.auth.UserRepository;
import com.mydreamjob.audit.AuditService;
import com.mydreamjob.audit.AuditContext;
import com.mydreamjob.audit.SecurityAuditService;
import com.mydreamjob.bidirectionalmatch.ApplicationMatchRecordRepository;
import com.mydreamjob.common.exception.BusinessException;
import com.mydreamjob.common.exception.ErrorCode;
import com.mydreamjob.company.Company;
import com.mydreamjob.company.CompanyRepository;
import com.mydreamjob.company.VerifiedCompanyGuard;
import com.mydreamjob.messaging.dto.CreateConversationRequest;
import com.mydreamjob.messaging.dto.MessagePageResponse;
import com.mydreamjob.messaging.dto.MessageResponse;
import com.mydreamjob.messaging.dto.SendMessageRequest;
import com.mydreamjob.safety.BlockRelationRepository;
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 org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import java.util.*;
import java.time.Clock;
import java.time.LocalDateTime;
@Slf4j
@Service
@RequiredArgsConstructor
public class MessagingService {
private final ConversationRepository conversationRepository;
private final ConversationMemberRepository memberRepository;
private final MessageRepository messageRepository;
private final MessageReadRepository readRepository;
private final UserRepository userRepository;
private final CompanyRepository companyRepository;
private final ApplicationMatchRecordRepository applicationMatchRecordRepository;
private final BlockRelationRepository blockRelationRepository;
private final SecurityAuditService securityAuditService;
private final AuditService auditService;
private final com.mydreamjob.candidatescore.refresh.CandidateScoringRefreshRequestService scoringRefreshRequestService;
private final Clock clock;
private final MessageViewMapper messageViewMapper;
private final ObjectMapper objectMapper;
private final com.mydreamjob.messaging.realtime.ChatEventService chatEvents;
private final ChatResumeSnapshotRepository resumeSnapshots;
private final ConversationSummaryService conversationSummaries;
private static final int MAX_PAGE_SIZE = 50;
private static final int MAX_CONTENT_LENGTH = 2000;
private static final String RECALLED_STATUS = "RECALLED";
private static final String SENT_STATUS = "SENT";
private static final int RECALL_WINDOW_MINUTES = 2;
private void assertMember(Long conversationId, Long userId) {
if (!memberRepository.existsByConversationIdAndUserId(conversationId, userId))
throw new BusinessException(ErrorCode.FORBIDDEN);
}
/** 统一会话存在性检查:不存在返回 404/CONVERSATION_NOT_FOUND,再进入成员校验 */
private Conversation requireConversation(Long conversationId) {
return conversationRepository.findById(conversationId)
.orElseThrow(() -> new BusinessException(ErrorCode.CONVERSATION_NOT_FOUND));
}
public List<Map<String, Object>> listConversations(Long userId) {
requireVerifiedIfCompany(userId);
List<ConversationMember> memberships = memberRepository.findByUserId(userId);
if (memberships.isEmpty()) return List.of();
List<Long> convIds = memberships.stream().map(ConversationMember::getConversationId).toList();
List<Conversation> convs = conversationRepository.findAllById(convIds);
Map<Long, Long> unreadByConversation = new HashMap<>();
messageRepository.countUnreadByConversationIds(convIds, userId).forEach(row ->
unreadByConversation.put(row.getConversationId(), row.getUnreadCount()));
convs.sort((a, b) -> b.getUpdatedAt().compareTo(a.getUpdatedAt()));
List<Map<String, Object>> result = new ArrayList<>();
for (Conversation c : convs) {
Map<String, Object> v = toView(c, userId);
v.put("unreadCount", unreadByConversation.getOrDefault(c.getId(), 0L));
result.add(v);
}
conversationSummaries.enrich(result,userId);
return result;
}
public Map<String, Object> getConversation(Long conversationId, Long userId) {
requireVerifiedIfCompany(userId);
Conversation c = requireConversation(conversationId);
assertMember(conversationId, userId);
Map<String, Object> v = toView(c, userId);
v.put("unreadCount", countUnread(conversationId, userId));
conversationSummaries.enrich(java.util.List.of(v),userId);
return v;
}
@Transactional
public Map<String, Object> createOrGetDirectConversation(Long userId, CreateConversationRequest req) {
requireVerifiedIfCompany(userId);
Long otherId = req.getParticipantUserId();
if (userId.equals(otherId)) throw new BusinessException(ErrorCode.BAD_REQUEST);
if (!userRepository.existsById(otherId)) throw new BusinessException(ErrorCode.USER_NOT_FOUND);
// 已有 DIRECT 会话 → 返回原会话(历史会话保持可访问,不删除、不失效)
List<ConversationMember> my = memberRepository.findByUserId(userId);
for (ConversationMember m : my) {
if (memberRepository.existsByConversationIdAndUserId(m.getConversationId(), otherId)) {
Conversation existing = conversationRepository.findById(m.getConversationId()).orElse(null);
if (existing != null && "DIRECT".equals(existing.getType())) return toView(existing, userId);
}
}
// 新建会话:任意方向存在屏蔽 → 禁止新建 DIRECT 会话
assertNotBlockedBetween(userId, otherId);
// 新建会话:企业-求职者组合必须先存在双向匹配成功的投递,防止绕过匹配解锁
assertCompanyCandidatePairHasMatchedApplication(userId, otherId);
Conversation c = conversationRepository.save(Conversation.builder().createdBy(userId).type("DIRECT").build());
memberRepository.save(ConversationMember.builder().conversationId(c.getId()).userId(userId).build());
memberRepository.save(ConversationMember.builder().conversationId(c.getId()).userId(otherId).build());
return toView(c);
}
/**
* 防绕过守卫:仅当双方为"企业账号 + 求职者账号"且确实要新建会话时,要求存在一条
* 同时归属双方且状态为 MATCHED 的投递;否则 403。
* 学生-学生、企业-企业及其他组合保持原语义;无法判定身份时(上游已校验存在性)防御性放行。
*/
private void assertCompanyCandidatePairHasMatchedApplication(Long userId, Long otherId) {
User me = userRepository.findById(userId).orElse(null);
User other = userRepository.findById(otherId).orElse(null);
if (me == null || other == null) return;
boolean meCompany = hasRole(me, User.Role.COMPANY);
boolean meStudent = hasRole(me, User.Role.STUDENT);
boolean otherCompany = hasRole(other, User.Role.COMPANY);
boolean otherStudent = hasRole(other, User.Role.STUDENT);
boolean companyCandidatePair = (meCompany && otherStudent) || (meStudent && otherCompany);
if (!companyCandidatePair) return;
Long companyUserId = meCompany ? userId : otherId;
Long studentUserId = meCompany ? otherId : userId;
Long companyId = companyRepository.findByOwnerUserId(companyUserId)
.map(Company::getId).orElse(null);
if (companyId == null
|| !applicationMatchRecordRepository.existsMatchedApplication(companyId, studentUserId)) {
throw new BusinessException(ErrorCode.MATCH_REQUIRED_FOR_CONVERSATION);
}
}
private boolean hasRole(User user, User.Role role) {
return user.getRoles() != null && user.getRoles().stream().anyMatch(r -> r == role);
}
private void requireVerifiedIfCompany(Long userId) {
User actor = userRepository.findByIdWithRoles(userId).orElse(null);
VerifiedCompanyGuard.requireVerifiedIfCompany(actor,
() -> companyRepository.findByOwnerUserId(userId).orElse(null));
}
// ── V33 屏蔽控制:任一方向存在屏蔽即禁止站内信交互 ──
private boolean isBlocked(Long aId, Long bId) {
return blockRelationRepository.existsByBlockerUserIdAndBlockedUserId(aId, bId)
|| blockRelationRepository.existsByBlockerUserIdAndBlockedUserId(bId, aId);
}
/**
* 对外统一返回 403/2002 + 「message cannot be sent」,不泄露是谁屏蔽了谁;
* 内部由审计记录 BLOCKED_SEND_PROHIBITED。屏蔽期间发送失败的消息直接丢弃,不补发。
*/
private void assertNotBlockedBetween(Long aId, Long bId) {
if (isBlocked(aId, bId)) {
AuditContext ctx = currentAuditContext();
// 拒绝型安全审计以独立事务落库,不受本次 403 回滚影响
securityAuditService.logRejected(aId, "BLOCKED_SEND_PROHIBITED", "USER", bId,
"Message send prohibited between users " + aId + " and " + bId, ctx);
log.info("Blocked send prohibited: reporter={} other={}", aId, bId);
throw new BusinessException(ErrorCode.FORBIDDEN, "message cannot be sent");
}
}
private AuditContext currentAuditContext() {
ServletRequestAttributes attrs =
(ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
if (attrs != null) return AuditContext.from(attrs.getRequest());
return AuditContext.builder().build();
}
@Transactional
public Map<String, Object> sendMessage(Long userId, SendMessageRequest req) {
requireVerifiedIfCompany(userId);
requireConversation(req.getConversationId());
assertMember(req.getConversationId(), userId);
conversationRepository.findLockedById(req.getConversationId());
if (req.getContent().length() > MAX_CONTENT_LENGTH) throw new BusinessException(ErrorCode.BAD_REQUEST);
// V33 屏蔽控制:会话任一成员存在任一方向屏蔽 → 禁止发送(对外统一 403/2002)
List<ConversationMember> members = memberRepository.findByConversationId(req.getConversationId());
for (ConversationMember m : members) {
if (!m.getUserId().equals(userId)) {
assertNotBlockedBetween(userId, m.getUserId());
}
}
if (req.getClientMessageId() != null) {
if (!req.getClientMessageId().matches("[A-Za-z0-9_-]{1,64}"))
throw new BusinessException(ErrorCode.BAD_REQUEST);
var existing = messageRepository.findByConversationIdAndSenderUserIdAndClientMessageId(
req.getConversationId(), userId, req.getClientMessageId());
if (existing.isPresent()) {
if ("RESUME".equals(existing.get().getMessageType()))
throw new BusinessException(ErrorCode.BAD_REQUEST, "clientMessageId belongs to a resume message");
if (!Objects.equals(existing.get().getRequestContentHash(), contentHash(req.getContent())))
throw new BusinessException(ErrorCode.BAD_REQUEST, "clientMessageId already used");
return toView(existing.get(), userId);
}
}
Message msg = messageRepository.save(Message.builder()
.conversationId(req.getConversationId()).senderUserId(userId).content(req.getContent())
.clientMessageId(req.getClientMessageId())
.requestContentHash(req.getClientMessageId() == null ? null : contentHash(req.getContent())).build());
chatEvents.record("MESSAGE_CREATED", req.getConversationId(), userId, msg.getId());
conversationRepository.findById(req.getConversationId()).ifPresent(c -> {
c.setUpdatedAt(msg.getCreatedAt()); conversationRepository.save(c);
});
// R01-D4-1:成功发送后,若发送者为学生则登记评分刷新(ACTIVITY 维度按发送消息计数)。
// 消息撤回不登记(公式只统计发送消息总数)。
userRepository.findById(userId).ifPresent(u -> {
if (u.getRoles() != null && u.getRoles().stream()
.anyMatch(r -> r == com.mydreamjob.auth.User.Role.STUDENT)) {
scoringRefreshRequestService.requestRefresh(userId, LocalDateTime.now(clock));
}
});
return toView(msg, userId);
}
public MessagePageResponse listMessages(Long conversationId, Long userId, int page, int size) {
requireVerifiedIfCompany(userId);
Conversation conversation = requireConversation(conversationId);
assertMember(conversationId, userId);
if (page < 0 || size < 1 || size > MAX_PAGE_SIZE) throw new BusinessException(ErrorCode.BAD_REQUEST);
Page<Message> result = messageRepository.findByConversationIdOrderByCreatedAtDesc(
conversationId, PageRequest.of(page, size));
List<Long> messageIds = result.getContent().stream().map(Message::getId).toList();
Set<Long> readIds = messageIds.isEmpty() ? Set.of() : readRepository
.findByMessageIdInAndReaderUserId(messageIds, userId).stream()
.map(MessageRead::getMessageId).collect(java.util.stream.Collectors.toSet());
boolean talentInvitation = "TALENT_INVITATION".equals(conversation.getType());
List<MessageResponse> items = result.getContent().stream().map(m -> {
MessageResponse response = messageViewMapper.toResponse(m, userId, talentInvitation);
response.setIsRead(readIds.contains(m.getId()));
return response;
}).toList();
return MessagePageResponse.builder()
.items(items)
.total(result.getTotalElements())
.page(result.getNumber())
.size(result.getSize())
.totalPages(result.getTotalPages())
.build();
}
@Transactional
public void markRead(Long conversationId, Long userId) {
requireVerifiedIfCompany(userId);
requireConversation(conversationId);
assertMember(conversationId, userId);
conversationRepository.findLockedById(conversationId);
List<Message> msgs = messageRepository
.findByConversationIdOrderByCreatedAtDesc(conversationId, PageRequest.of(0, 500)).getContent();
List<Long> msgIds = msgs.stream()
.filter(m -> !RECALLED_STATUS.equals(m.getStatus()))
.map(Message::getId).toList();
if (msgIds.isEmpty()) return;
List<MessageRead> existing = readRepository.findByMessageIdInAndReaderUserId(msgIds, userId);
Set<Long> existingIds = new HashSet<>();
for (MessageRead r : existing) existingIds.add(r.getMessageId());
for (Long mid : msgIds) {
if (!existingIds.contains(mid)) {
readRepository.save(MessageRead.builder().messageId(mid).readerUserId(userId).build());
chatEvents.record("MESSAGE_READ", conversationId, userId, mid);
}
}
}
private long countUnread(Long convId, Long userId) {
List<Message> msgs = messageRepository
.findByConversationIdOrderByCreatedAtDesc(convId, PageRequest.of(0, 500)).getContent();
if (msgs.isEmpty()) return 0;
List<Long> msgIds = msgs.stream()
.filter(m -> !RECALLED_STATUS.equals(m.getStatus()))
.map(Message::getId).toList();
if (msgIds.isEmpty()) return 0;
return msgIds.size() - readRepository.findByMessageIdInAndReaderUserId(msgIds, userId).size();
}
/** 游标只来自补拉响应,不能直接用实时事件 ID 跳过尚未收到的事件。 */
public Map<String, Object> replayEvents(Long conversationId, Long userId, long after, int size) {
requireVerifiedIfCompany(userId);
requireConversation(conversationId);
assertMember(conversationId, userId);
if (after < 0 || size < 1 || size > 100) throw new BusinessException(ErrorCode.BAD_REQUEST);
return chatEvents.replay(conversationId, userId, after, size);
}
/**
* 在事务内执行消息撤回。调用者身份、会话成员关系和原发送者关系均由服务端校验。
* capturedNow 由入口统一捕获,便于边界测试,也避免同一请求混用多个时间。
*/
@Transactional
public MessageResponse recallMessage(Long currentUserId, Long conversationId, Long messageId,
AuditContext auditContext, LocalDateTime capturedNow) {
requireVerifiedIfCompany(currentUserId);
Conversation conversation = requireConversation(conversationId);
assertMember(conversationId, currentUserId);
conversationRepository.findLockedById(conversationId);
User currentUser = userRepository.findById(currentUserId).orElse(null);
if (currentUser != null && hasRole(currentUser, User.Role.ADMIN)) {
throw new BusinessException(ErrorCode.FORBIDDEN);
}
Message message = messageRepository.findById(messageId)
.orElseThrow(() -> new BusinessException(ErrorCode.MESSAGE_NOT_FOUND));
if (!Objects.equals(message.getConversationId(), conversationId)) {
throw new BusinessException(ErrorCode.FORBIDDEN);
}
if (!Objects.equals(message.getSenderUserId(), currentUserId)) {
throw new BusinessException(ErrorCode.FORBIDDEN);
}
String status = message.getStatus() == null ? SENT_STATUS : message.getStatus();
if (RECALLED_STATUS.equals(status)) {
return toMessageResponse(message, currentUserId, conversation);
}
LocalDateTime createdAt = message.getCreatedAt();
if (createdAt == null || capturedNow == null
|| capturedNow.isAfter(createdAt.plusMinutes(RECALL_WINDOW_MINUTES))) {
throw new BusinessException(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED);
}
Long expectedVersion = message.getVersion() == null ? 0L : message.getVersion();
int updated = messageRepository.recallIfCurrent(messageId, conversationId, currentUserId,
expectedVersion, MessageViewMapper.RECALLED_PLACEHOLDER, capturedNow);
if (updated != 1) {
Message latest = messageRepository.findCurrentByIdForRecall(messageId).orElse(null);
if (latest != null && RECALLED_STATUS.equals(latest.getStatus())) {
return toMessageResponse(latest, currentUserId, conversation);
}
throw new BusinessException(ErrorCode.MESSAGE_RECALL_CONCURRENT_UPDATE);
}
if ("RESUME".equals(message.getMessageType())) resumeSnapshots.purge(messageId);
auditService.log(currentUserId, "MESSAGE_RECALLED", "MESSAGE", messageId,
recallAuditDetails(messageId, conversationId, conversation.getType(), auditContext),
recallAuditContext(auditContext));
Message recalled = messageRepository.findById(messageId).orElse(message);
if ("RESUME".equals(recalled.getMessageType())) {
recalled.setResumeRevoked(true); recalled.setResumeFilename(null);
recalled.setResumeContentType(null); recalled.setResumeSize(null);
recalled = messageRepository.saveAndFlush(recalled);
}
chatEvents.record("MESSAGE_RECALLED", conversationId, currentUserId, messageId);
return toMessageResponse(recalled, currentUserId, conversation);
}
/** 生产入口使用 Spring Clock 捕获一次当前时间。 */
@Transactional
public MessageResponse recallMessage(Long currentUserId, Long conversationId, Long messageId,
AuditContext auditContext) {
Clock effectiveClock = clock == null ? Clock.systemDefaultZone() : clock;
return recallMessage(currentUserId, conversationId, messageId, auditContext,
LocalDateTime.now(effectiveClock));
}
private MessageResponse toMessageResponse(Message message, Long currentUserId, Conversation conversation) {
MessageResponse response = messageViewMapper.toResponse(message, currentUserId,
conversation != null && "TALENT_INVITATION".equals(conversation.getType()));
if (response != null) {
response.setIsRead(readRepository.existsByMessageIdAndReaderUserId(message.getId(), currentUserId));
}
return response;
}
private AuditContext recallAuditContext(AuditContext source) {
return AuditContext.builder()
.requestId(source == null ? null : source.getRequestId())
.build();
}
private String recallAuditDetails(Long messageId, Long conversationId, String conversationType,
AuditContext context) {
String safeConversationType = "DIRECT".equals(conversationType)
|| "TALENT_INVITATION".equals(conversationType)
? conversationType : "UNKNOWN";
Map<String, Object> details = new LinkedHashMap<>();
details.put("messageId", messageId);
details.put("conversationId", conversationId);
details.put("conversationType", safeConversationType);
details.put("status", "RECALLED");
details.put("requestId", context == null ? null : context.getRequestId());
try {
return objectMapper.writeValueAsString(details);
} catch (JsonProcessingException ex) {
throw new IllegalStateException("message recall audit serialization failed");
}
}
public Map<String, Object> toView(Conversation c) {
Map<String, Object> v = new HashMap<>();
v.put("id", c.getId()); v.put("title", c.getTitle()); v.put("type", c.getType());
v.put("createdBy", c.getCreatedBy()); v.put("createdAt", c.getCreatedAt()); v.put("updatedAt", c.getUpdatedAt());
return v;
}
/** V42 会话隐私映射:TALENT_INVITATION 不返回 createdBy(其中可能是另一方学生 ID)。 */
public Map<String, Object> toView(Conversation c, Long currentUserId) {
if (c == null || !"TALENT_INVITATION".equals(c.getType())) return toView(c);
Map<String, Object> v = new HashMap<>();
v.put("id", c.getId()); v.put("title", c.getTitle()); v.put("type", c.getType());
v.put("participantScope", "CURRENT_USER_AND_OTHER_PARTICIPANT");
v.put("createdAt", c.getCreatedAt()); v.put("updatedAt", c.getUpdatedAt());
return v;
}
public Map<String, Object> toView(Message m) {
return toLegacyView(messageViewMapper.toResponse(m, null, false));
}
/** V42 消息隐私映射:TALENT_INVITATION 仅返回相对发送者身份。 */
public Map<String, Object> toView(Message m, Long currentUserId) {
if (m == null) return null;
return toView(m, currentUserId, isTalentInvitationConversation(m.getConversationId()));
}
private Map<String, Object> toView(Message m, Long currentUserId, boolean talentInvitation) {
return toLegacyView(messageViewMapper.toResponse(m, currentUserId, talentInvitation));
}
private Map<String, Object> toLegacyView(MessageResponse response) {
Map<String, Object> v = new HashMap<>();
if (response == null) return v;
v.put("id", response.getId());
if (response.getClientMessageId() != null) v.put("clientMessageId", response.getClientMessageId());
if (response.getVersion() != null) v.put("version", response.getVersion());
v.put("conversationId", response.getConversationId());
if (response.getSenderUserId() != null) v.put("senderUserId", response.getSenderUserId());
if (response.getSenderScope() != null) v.put("senderScope", response.getSenderScope());
v.put("content", response.getContent());
v.put("status", response.getStatus());
if (response.getRecalledAt() != null) v.put("recalledAt", response.getRecalledAt());
v.put("createdAt", response.getCreatedAt());
if (response.getMessageType() != null) v.put("messageType", response.getMessageType());
if (response.getResume() != null) v.put("resume", response.getResume());
if (response.getIsRead() != null) v.put("isRead", response.getIsRead());
return v;
}
private String contentHash(String content) {
try {
return java.util.HexFormat.of().formatHex(java.security.MessageDigest.getInstance("SHA-256")
.digest(content.getBytes(java.nio.charset.StandardCharsets.UTF_8)));
} catch (java.security.NoSuchAlgorithmException impossible) {
throw new IllegalStateException("SHA-256 unavailable");
}
}
private boolean isTalentInvitationConversation(Long conversationId) {
return conversationId != null && conversationRepository.findById(conversationId)
.map(c -> "TALENT_INVITATION".equals(c.getType())).orElse(false);
}
}