feat(aihr): add M1 broadcast targeting and mandatory reads

This commit is contained in:
2026-07-23 12:11:41 +08:00
parent cd94f4fe9f
commit f6609808df
14 changed files with 925 additions and 69 deletions
@@ -8,6 +8,7 @@ import lombok.RequiredArgsConstructor;
import org.dromara.aihr.broadcast.AihrBroadcastDto.AdminBroadcastListResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastDetailResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastListResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastTargetOptions;
import org.dromara.aihr.broadcast.AihrBroadcastDto.PublishRequest;
import org.dromara.aihr.broadcast.AihrBroadcastDto.PublishResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.UnreadCountResponse;
@@ -40,9 +41,10 @@ public class AihrBroadcastController {
@GetMapping("/messages")
public R<BroadcastListResponse> messages(
@RequestParam(required = false) Integer pageNum,
@RequestParam(required = false) Integer pageSize
@RequestParam(required = false) Integer pageSize,
@RequestParam(required = false) String scope
) {
return R.ok(broadcastService.messages(pageNum, pageSize));
return R.ok(broadcastService.messages(pageNum, pageSize, scope));
}
@SaCheckRole(value = {TenantConstants.SUPER_ADMIN_ROLE_KEY, HR_OPERATOR_ROLE}, mode = SaMode.OR)
@@ -55,6 +57,12 @@ public class AihrBroadcastController {
return R.ok(broadcastService.adminMessages(pageNum, pageSize, status));
}
@SaCheckRole(value = {TenantConstants.SUPER_ADMIN_ROLE_KEY, HR_OPERATOR_ROLE}, mode = SaMode.OR)
@GetMapping("/admin/target-options")
public R<BroadcastTargetOptions> targetOptions() {
return R.ok(broadcastService.targetOptions());
}
@GetMapping("/messages/{id}")
public R<BroadcastDetailResponse> message(@PathVariable Long id) {
return R.ok(broadcastService.message(id));
@@ -1,5 +1,6 @@
package org.dromara.aihr.broadcast;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.Pattern;
import jakarta.validation.constraints.Size;
@@ -26,8 +27,14 @@ public final class AihrBroadcastDto {
Long id,
String title,
boolean read,
String publishedAt
String publishedAt,
boolean requiredRead,
boolean targeted,
String targetReason
) {
public BroadcastListItem(Long id, String title, boolean read, String publishedAt) {
this(id, title, read, publishedAt, false, false, null);
}
}
public record BroadcastDetailResponse(
@@ -35,8 +42,14 @@ public final class AihrBroadcastDto {
String title,
String content,
boolean read,
String publishedAt
String publishedAt,
boolean requiredRead,
boolean targeted,
String targetReason
) {
public BroadcastDetailResponse(Long id, String title, String content, boolean read, String publishedAt) {
this(id, title, content, read, publishedAt, false, false, null);
}
}
public record AdminBroadcastListResponse(
@@ -58,8 +71,56 @@ public final class AihrBroadcastDto {
String publishedAt,
Long withdrawnBy,
String withdrawnAt,
String withdrawReason
String withdrawReason,
boolean requiredRead,
long targetRuleCount,
long matchedRecipientCount,
long unmatchedTargetCount
) {
public AdminBroadcastListItem(
Long id,
String title,
String content,
String status,
int currentVersion,
Long publishedBy,
String publishedAt,
Long withdrawnBy,
String withdrawnAt,
String withdrawReason
) {
this(id, title, content, status, currentVersion, publishedBy, publishedAt,
withdrawnBy, withdrawnAt, withdrawReason, false, 0, 0, 0);
}
}
public record BroadcastTargetOption(String value, String label) {
}
public record BroadcastTargetOptions(
List<BroadcastTargetOption> projects,
List<BroadcastTargetOption> positionNames,
List<BroadcastTargetOption> positionLevels
) {
}
public record BroadcastTargetRequest(
@Size(max = 50, message = "定向项目不能超过 50 个")
List<@NotBlank(message = "项目编码不能为空") @Size(max = 50, message = "项目编码不能超过 50 个字符") String> projectCodes,
@Size(max = 50, message = "定向岗位不能超过 50 个")
List<@NotBlank(message = "岗位不能为空") @Size(max = 100, message = "岗位不能超过 100 个字符") String> positionNames,
@Size(max = 20, message = "定向职级不能超过 20 个")
List<@NotBlank(message = "职级不能为空") @Size(max = 30, message = "职级不能超过 30 个字符") String> positionLevels
) {
public BroadcastTargetRequest {
projectCodes = projectCodes == null ? List.of() : List.copyOf(projectCodes);
positionNames = positionNames == null ? List.of() : List.copyOf(positionNames);
positionLevels = positionLevels == null ? List.of() : List.copyOf(positionLevels);
}
public static BroadcastTargetRequest empty() {
return new BroadcastTargetRequest(List.of(), List.of(), List.of());
}
}
public record PublishRequest(
@@ -75,11 +136,19 @@ public final class AihrBroadcastDto {
String content,
@NotBlank(message = "当前租户不能为空")
@Size(max = 64, message = "当前租户不能超过 64 个字符")
String expectedTenantId
String expectedTenantId,
Boolean requiredRead,
@Valid BroadcastTargetRequest targets
) {
public PublishRequest(String requestId, String title, String content, String expectedTenantId) {
this(requestId, title, content, expectedTenantId, false, BroadcastTargetRequest.empty());
}
}
public record PublishResponse(Long id, int version) {
public record PublishResponse(Long id, int version, long matchedRecipientCount, long unmatchedTargetCount) {
public PublishResponse(Long id, int version) {
this(id, version, 0, 0);
}
}
public record WithdrawRequest(
@@ -6,6 +6,9 @@ import org.dromara.aihr.broadcast.AihrBroadcastDto.AdminBroadcastListResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastDetailResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastListItem;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastListResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastTargetOption;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastTargetOptions;
import org.dromara.aihr.broadcast.AihrBroadcastDto.BroadcastTargetRequest;
import org.dromara.aihr.broadcast.AihrBroadcastDto.PublishRequest;
import org.dromara.aihr.broadcast.AihrBroadcastDto.PublishResponse;
import org.dromara.aihr.broadcast.AihrBroadcastDto.UnreadCountResponse;
@@ -30,9 +33,13 @@ import java.sql.Statement;
import java.sql.Timestamp;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HexFormat;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.regex.Pattern;
@Service
@@ -65,27 +72,45 @@ public class AihrBroadcastService {
}
public BroadcastListResponse messages(Integer rawPageNum, Integer rawPageSize) {
return messages(rawPageNum, rawPageSize, null);
}
public BroadcastListResponse messages(Integer rawPageNum, Integer rawPageSize, String rawScope) {
AihrKnowledgePrincipal principal = principalResolver.current();
requireEmployeeAudience(principal);
Page page = page(rawPageNum, rawPageSize);
Long total = jdbcTemplate.queryForObject("""
select count(*)
from aihr_broadcast_message m
where m.tenant_id = ? and m.status = 'PUBLISHED'
""", Long.class, principal.tenantId());
BroadcastMessageScope scope = messageScope(rawScope);
EmployeeMessageQuery employeeMessages = employeeMessageQuery(principal, scope);
Long total = jdbcTemplate.queryForObject("select count(*) " + employeeMessages.fromWhere(),
Long.class, employeeMessages.parameters().toArray());
List<Object> parameters = new ArrayList<>(employeeMessages.parameters());
parameters.add(page.pageSize());
parameters.add(page.offset());
List<BroadcastListItem> rows = jdbcTemplate.query("""
select m.id, m.title, m.published_time,
case when r.id is null then 0 else 1 end as read_flag
from aihr_broadcast_message m
left join aihr_broadcast_read r
on r.tenant_id = m.tenant_id and r.message_id = m.id and r.user_id = ?
where m.tenant_id = ? and m.status = 'PUBLISHED'
order by m.published_time desc, m.id desc
case when r.id is null then 0 else 1 end as read_flag,
case when m.required_read = 1 then 1 else 0 end as required_read,
case when tr.user_id is null then 0 else 1 end as targeted_flag,
tr.target_reason
""" + employeeMessages.fromWhere() + """
order by case
when m.required_read = 1
and r.id is null
and (not exists (
select 1 from aihr_broadcast_target_rule required_rule
where required_rule.tenant_id = m.tenant_id
and required_rule.message_id = m.id
) or tr.user_id is not null) then 0
when tr.user_id is not null then 1
else 2
end,
m.published_time desc, m.id desc
limit ? offset ?
""", (rs, rowNum) -> new BroadcastListItem(
rs.getLong("id"), rs.getString("title"), rs.getBoolean("read_flag"),
format(rs.getTimestamp("published_time"))
), principal.userId(), principal.tenantId(), page.pageSize(), page.offset());
format(rs.getTimestamp("published_time")), rs.getBoolean("required_read"),
rs.getBoolean("targeted_flag"), rs.getString("target_reason")
), parameters.toArray());
return new BroadcastListResponse(total == null ? 0 : total, page.pageNum(), page.pageSize(), rows);
}
@@ -108,7 +133,16 @@ public class AihrBroadcastService {
parameters.add(page.offset());
List<AdminBroadcastListItem> rows = jdbcTemplate.query("""
select m.id, m.title, m.content, m.status, 1 as current_version,
m.published_by, m.published_time, m.withdrawn_by, m.withdrawn_time, m.withdraw_reason
m.published_by, m.published_time, m.withdrawn_by, m.withdrawn_time, m.withdraw_reason,
case when m.required_read = 1 then 1 else 0 end as required_read,
(select count(*) from aihr_broadcast_target_rule target_rule
where target_rule.tenant_id = m.tenant_id and target_rule.message_id = m.id) as target_rule_count,
(select count(distinct recipient.user_id) from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = m.tenant_id and recipient.message_id = m.id
and recipient.target_status = 'MATCHED') as matched_recipient_count,
(select count(*) from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = m.tenant_id and recipient.message_id = m.id
and recipient.target_status = 'UNMATCHED') as unmatched_target_count
from aihr_broadcast_message m
where m.tenant_id = ?""" + statusFilter + """
order by m.published_time desc, m.id desc
@@ -117,27 +151,81 @@ public class AihrBroadcastService {
rs.getLong("id"), rs.getString("title"), rs.getString("content"), rs.getString("status"),
rs.getInt("current_version"), rs.getLong("published_by"), format(rs.getTimestamp("published_time")),
rs.getObject("withdrawn_by", Long.class), format(rs.getTimestamp("withdrawn_time")),
rs.getString("withdraw_reason")
rs.getString("withdraw_reason"), rs.getBoolean("required_read"), rs.getLong("target_rule_count"),
rs.getLong("matched_recipient_count"), rs.getLong("unmatched_target_count")
), parameters.toArray());
return new AdminBroadcastListResponse(tenantId, total == null ? 0 : total, page.pageNum(), page.pageSize(), rows);
}
public BroadcastTargetOptions targetOptions() {
AihrKnowledgePrincipal principal = principalResolver.current();
String tenantId = adminScopeTenantId(principal);
List<BroadcastTargetOption> projects = jdbcTemplate.query("""
select project_code as value,
coalesce(nullif(max(project_name), ''), project_code) as label
from aihr_org_snapshot
where binary tenant_id = binary ?
and employment_status = 'active'
and project_code is not null and project_code <> ''
group by project_code
order by label, value
limit 200
""", (rs, rowNum) -> new BroadcastTargetOption(
rs.getString("value"), rs.getString("label")
), tenantId);
List<BroadcastTargetOption> positions = jdbcTemplate.query("""
select distinct position_name as value, position_name as label
from aihr_org_snapshot
where binary tenant_id = binary ?
and employment_status = 'active'
and position_name is not null and position_name <> ''
order by label
limit 200
""", (rs, rowNum) -> new BroadcastTargetOption(
rs.getString("value"), rs.getString("label")
), tenantId);
List<BroadcastTargetOption> levels = jdbcTemplate.query("""
select distinct position_level as value, position_level as label
from aihr_org_snapshot
where binary tenant_id = binary ?
and employment_status = 'active'
and position_level is not null and position_level <> ''
order by label
limit 50
""", (rs, rowNum) -> new BroadcastTargetOption(
rs.getString("value"), rs.getString("label")
), tenantId);
return new BroadcastTargetOptions(projects, positions, levels);
}
public BroadcastDetailResponse message(Long rawMessageId) {
AihrKnowledgePrincipal principal = principalResolver.current();
requireEmployeeAudience(principal);
long messageId = messageId(rawMessageId);
List<BroadcastDetailResponse> rows = jdbcTemplate.query("""
select m.id, m.title, m.content, m.published_time,
case when r.id is null then 0 else 1 end as read_flag
case when r.id is null then 0 else 1 end as read_flag,
case when m.required_read = 1 then 1 else 0 end as required_read,
case when tr.user_id is null then 0 else 1 end as targeted_flag,
tr.target_reason
from aihr_broadcast_message m
left join aihr_broadcast_read r
on r.tenant_id = m.tenant_id and r.message_id = m.id and r.user_id = ?
left join (
select recipient.tenant_id, recipient.message_id, recipient.user_id,
min(recipient.target_reason) as target_reason
from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = ? and recipient.user_id = ?
and recipient.target_status = 'MATCHED'
group by recipient.tenant_id, recipient.message_id, recipient.user_id
) tr on tr.tenant_id = m.tenant_id and tr.message_id = m.id
where m.tenant_id = ? and m.id = ? and m.status = 'PUBLISHED'
limit 1
""", (rs, rowNum) -> new BroadcastDetailResponse(
rs.getLong("id"), rs.getString("title"), rs.getString("content"), rs.getBoolean("read_flag"),
format(rs.getTimestamp("published_time"))
), principal.userId(), principal.tenantId(), messageId);
format(rs.getTimestamp("published_time")), rs.getBoolean("required_read"),
rs.getBoolean("targeted_flag"), rs.getString("target_reason")
), principal.userId(), principal.tenantId(), principal.userId(), principal.tenantId(), messageId);
if (rows.isEmpty()) {
throw unavailable();
}
@@ -199,26 +287,31 @@ public class AihrBroadcastService {
String requestKey = requestKey(request == null ? null : request.requestId());
String title = required(request == null ? null : request.title(), "消息标题", 200);
String content = required(request == null ? null : request.content(), "消息正文", 10_000);
boolean requiredRead = request != null && Boolean.TRUE.equals(request.requiredRead());
TargetSelection targetSelection = targetSelection(request == null ? null : request.targets());
String requestHash = requestHash(title, content);
String targetPayloadHash = targetPayloadHash(requiredRead, targetSelection);
PublishReplay existing = findPublishReplay(tenantId, principal.userId(), requestKey);
if (existing != null) {
return replayOrConflict(existing, title, content);
return replayOrConflict(existing, title, content, requiredRead, targetSelection, targetPayloadHash);
}
KeyHolder keyHolder = new GeneratedKeyHolder();
try {
jdbcTemplate.update(connection -> {
PreparedStatement statement = connection.prepareStatement("""
insert into aihr_broadcast_message
(tenant_id, title, content, status, published_by, publish_request_key, publish_request_hash,
published_time, create_time, update_time)
values (?, ?, ?, 'PUBLISHED', ?, ?, ?, now(), now(), now())
(tenant_id, title, content, required_read, status, published_by, publish_request_key, publish_request_hash,
target_payload_hash, published_time, create_time, update_time)
values (?, ?, ?, ?, 'PUBLISHED', ?, ?, ?, ?, now(), now(), now())
""", Statement.RETURN_GENERATED_KEYS);
statement.setString(1, tenantId);
statement.setString(2, title);
statement.setString(3, content);
statement.setLong(4, principal.userId());
statement.setString(5, requestKey);
statement.setString(6, requestHash);
statement.setBoolean(4, requiredRead);
statement.setLong(5, principal.userId());
statement.setString(6, requestKey);
statement.setString(7, requestHash);
statement.setString(8, targetPayloadHash);
return statement;
}, keyHolder);
} catch (DuplicateKeyException duplicate) {
@@ -227,7 +320,7 @@ public class AihrBroadcastService {
// concurrent request that just won the unique-key race.
PublishReplay raced = findPublishReplayForUpdate(tenantId, principal.userId(), requestKey);
if (raced != null) {
return replayOrConflict(raced, title, content);
return replayOrConflict(raced, title, content, requiredRead, targetSelection, targetPayloadHash);
}
throw duplicate;
}
@@ -245,7 +338,8 @@ public class AihrBroadcastService {
if (snapshots != 1) {
throw new ServiceException("保存消息版本失败");
}
return new PublishResponse(id.longValue(), 1);
TargetAudit targetAudit = persistTargetSnapshot(tenantId, id.longValue(), targetSelection);
return new PublishResponse(id.longValue(), 1, targetAudit.matchedRecipientCount(), targetAudit.unmatchedTargetCount());
}
@Transactional(rollbackFor = Exception.class)
@@ -280,6 +374,58 @@ public class AihrBroadcastService {
}
}
private static EmployeeMessageQuery employeeMessageQuery(
AihrKnowledgePrincipal principal,
BroadcastMessageScope scope
) {
List<Object> parameters = new ArrayList<>();
parameters.add(principal.userId());
parameters.add(principal.tenantId());
parameters.add(principal.userId());
parameters.add(principal.tenantId());
String scopeFilter = switch (scope) {
case ALL -> "";
case RELATED -> " and tr.user_id is not null";
case REQUIRED_UNREAD -> """
and m.required_read = 1
and r.id is null
and (not exists (
select 1 from aihr_broadcast_target_rule required_rule
where required_rule.tenant_id = m.tenant_id
and required_rule.message_id = m.id
) or tr.user_id is not null)
""";
};
return new EmployeeMessageQuery("""
from aihr_broadcast_message m
left join aihr_broadcast_read r
on r.tenant_id = m.tenant_id and r.message_id = m.id and r.user_id = ?
left join (
select recipient.tenant_id, recipient.message_id, recipient.user_id,
min(recipient.target_reason) as target_reason
from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = ? and recipient.user_id = ?
and recipient.target_status = 'MATCHED'
group by recipient.tenant_id, recipient.message_id, recipient.user_id
) tr on tr.tenant_id = m.tenant_id and tr.message_id = m.id
where m.tenant_id = ? and m.status = 'PUBLISHED'
""" + scopeFilter, parameters);
}
private static BroadcastMessageScope messageScope(String rawScope) {
String scope = rawScope == null ? "" : rawScope.trim().toUpperCase(Locale.ROOT);
if (scope.isEmpty() || "ALL".equals(scope)) {
return BroadcastMessageScope.ALL;
}
if ("RELATED".equals(scope)) {
return BroadcastMessageScope.RELATED;
}
if ("REQUIRED_UNREAD".equals(scope)) {
return BroadcastMessageScope.REQUIRED_UNREAD;
}
throw new ServiceException("scope 只支持 ALL、RELATED 或 REQUIRED_UNREAD", HttpStatus.BAD_REQUEST);
}
private static Page page(Integer rawPageNum, Integer rawPageSize) {
int pageNum = rawPageNum == null ? 1 : rawPageNum;
int pageSize = rawPageSize == null ? DEFAULT_PAGE_SIZE : rawPageSize;
@@ -302,24 +448,171 @@ public class AihrBroadcastService {
private PublishReplay findPublishReplay(String tenantId, Long publishedBy, String requestKey, boolean currentRead) {
List<PublishReplay> rows = jdbcTemplate.query("""
select id, title, content
from aihr_broadcast_message
where tenant_id = ? and published_by = ? and publish_request_key = ?
select m.id, m.title, m.content, m.target_payload_hash,
(select count(distinct recipient.user_id) from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = m.tenant_id and recipient.message_id = m.id
and recipient.target_status = 'MATCHED') as matched_recipient_count,
(select count(*) from aihr_broadcast_target_recipient recipient
where recipient.tenant_id = m.tenant_id and recipient.message_id = m.id
and recipient.target_status = 'UNMATCHED') as unmatched_target_count
from aihr_broadcast_message m
where m.tenant_id = ? and m.published_by = ? and m.publish_request_key = ?
limit 1
""" + (currentRead ? " for update" : ""), (rs, rowNum) -> new PublishReplay(
rs.getLong("id"), rs.getString("title"), rs.getString("content")),
rs.getLong("id"), rs.getString("title"), rs.getString("content"), rs.getString("target_payload_hash"),
rs.getLong("matched_recipient_count"), rs.getLong("unmatched_target_count")),
tenantId, publishedBy, requestKey);
return rows.isEmpty() ? null : rows.get(0);
}
private static PublishResponse replayOrConflict(PublishReplay existing, String title, String content) {
// Compare the immutable payload itself, not just its digest. This preserves safe
// replay semantics for rows created by earlier hash formats and cannot be confused
// by delimiter collisions in user-entered text.
if (!title.equals(existing.title()) || !content.equals(existing.content())) {
private static PublishResponse replayOrConflict(
PublishReplay existing,
String title,
String content,
boolean requiredRead,
TargetSelection targetSelection,
String targetPayloadHash
) {
// The target payload has an independent length-delimited digest, so a retry cannot
// reuse a request ID while silently changing mandatory-read or target dimensions.
// Historical M0 rows have no digest and remain replayable only as M0 payloads.
boolean legacyM0TargetPayload = existing.targetPayloadHash() == null
&& !requiredRead && !targetSelection.hasTargets();
if (!title.equals(existing.title()) || !content.equals(existing.content())
|| (!legacyM0TargetPayload && !targetPayloadHash.equals(existing.targetPayloadHash()))) {
throw new ServiceException("该请求ID已用于不同的公司消息", 409);
}
return new PublishResponse(existing.id(), 1);
return new PublishResponse(existing.id(), 1,
existing.matchedRecipientCount(), existing.unmatchedTargetCount());
}
private TargetAudit persistTargetSnapshot(String tenantId, long messageId, TargetSelection selection) {
if (!selection.hasTargets()) {
return TargetAudit.empty();
}
persistTargetRules(tenantId, messageId, "PROJECT", selection.projectCodes());
persistTargetRules(tenantId, messageId, "POSITION", selection.positionNames());
persistTargetRules(tenantId, messageId, "LEVEL", selection.positionLevels());
Set<Long> matchedUsers = new LinkedHashSet<>();
long unmatchedTargets = 0;
for (TargetCandidate candidate : targetCandidates(tenantId, selection)) {
boolean matched = candidate.userId() != null && candidate.userId() > 0;
if (matched) {
matchedUsers.add(candidate.userId());
} else {
unmatchedTargets++;
}
String recipientKey = candidate.subjectRef() + "#" + (matched ? candidate.userId() : "UNMATCHED");
jdbcTemplate.update("""
insert into aihr_broadcast_target_recipient
(tenant_id, message_id, recipient_key, subject_ref, user_id, target_status, target_reason, create_time, update_time)
values (?, ?, ?, ?, ?, ?, ?, now(), now())
on duplicate key update target_status = values(target_status), target_reason = values(target_reason),
update_time = now()
""", tenantId, messageId, recipientKey, candidate.subjectRef(), candidate.userId(),
matched ? "MATCHED" : "UNMATCHED", selection.reason());
}
return new TargetAudit(matchedUsers.size(), unmatchedTargets);
}
private void persistTargetRules(String tenantId, long messageId, String targetType, List<String> targetValues) {
for (String targetValue : targetValues) {
jdbcTemplate.update("""
insert into aihr_broadcast_target_rule
(tenant_id, message_id, target_type, target_value, create_time)
values (?, ?, ?, ?, now())
on duplicate key update target_value = values(target_value)
""", tenantId, messageId, targetType, targetValue);
}
}
private List<TargetCandidate> targetCandidates(String tenantId, TargetSelection selection) {
StringBuilder sql = new StringBuilder("""
select distinct o.ext_party_id as subject_ref, u.user_id
from aihr_org_snapshot o
left join sys_user u
on binary o.tenant_id = binary u.tenant_id
and (binary o.person_phone = binary u.phonenumber or binary o.ext_party_id = binary u.user_name)
and u.user_type = 'app_user'
and u.status = '0'
and u.del_flag = '0'
where binary o.tenant_id = binary ?
and o.employment_status = 'active'
and o.ext_party_id is not null and o.ext_party_id <> ''
""");
List<Object> parameters = new ArrayList<>();
parameters.add(tenantId);
appendTargetFilter(sql, parameters, "o.project_code", selection.projectCodes());
appendTargetFilter(sql, parameters, "o.position_name", selection.positionNames());
appendTargetFilter(sql, parameters, "o.position_level", selection.positionLevels());
sql.append(" order by o.ext_party_id, u.user_id");
return jdbcTemplate.query(sql.toString(), (rs, rowNum) -> new TargetCandidate(
rs.getString("subject_ref"), rs.getObject("user_id", Long.class)
), parameters.toArray());
}
private static void appendTargetFilter(
StringBuilder sql,
List<Object> parameters,
String column,
List<String> values
) {
if (values.isEmpty()) {
return;
}
sql.append(" and ").append(column).append(" in (")
.append(String.join(",", Collections.nCopies(values.size(), "?"))).append(')');
parameters.addAll(values);
}
private static TargetSelection targetSelection(BroadcastTargetRequest rawTargets) {
BroadcastTargetRequest targets = rawTargets == null ? BroadcastTargetRequest.empty() : rawTargets;
List<String> projectCodes = targetValues(targets.projectCodes(), 50, 50, "项目编码");
List<String> positionNames = targetValues(targets.positionNames(), 50, 100, "岗位");
List<String> positionLevels = targetValues(targets.positionLevels(), 20, 30, "职级");
List<String> labels = new ArrayList<>();
if (!projectCodes.isEmpty()) labels.add("项目");
if (!positionNames.isEmpty()) labels.add("岗位");
if (!positionLevels.isEmpty()) labels.add("职级");
return new TargetSelection(projectCodes, positionNames, positionLevels, String.join("、", labels));
}
private static List<String> targetValues(List<String> rawValues, int maximumCount, int maximumLength, String label) {
List<String> values = rawValues == null ? List.of() : rawValues;
if (values.size() > maximumCount) {
throw new ServiceException(label + "不能超过 " + maximumCount + " 个", HttpStatus.BAD_REQUEST);
}
Set<String> normalized = new LinkedHashSet<>();
for (String rawValue : values) {
String value = rawValue == null ? "" : rawValue.trim();
if (value.isEmpty()) {
throw new ServiceException(label + "不能为空", HttpStatus.BAD_REQUEST);
}
if (value.length() > maximumLength) {
throw new ServiceException(label + "不能超过 " + maximumLength + " 个字符", HttpStatus.BAD_REQUEST);
}
normalized.add(value);
}
List<String> sorted = new ArrayList<>(normalized);
sorted.sort(Comparator.naturalOrder());
return List.copyOf(sorted);
}
private static String targetPayloadHash(boolean requiredRead, TargetSelection selection) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
updateDigestField(digest, requiredRead ? "1" : "0");
updateDigestField(digest, "PROJECT");
selection.projectCodes().forEach(value -> updateDigestField(digest, value));
updateDigestField(digest, "POSITION");
selection.positionNames().forEach(value -> updateDigestField(digest, value));
updateDigestField(digest, "LEVEL");
selection.positionLevels().forEach(value -> updateDigestField(digest, value));
return HexFormat.of().formatHex(digest.digest());
} catch (NoSuchAlgorithmException impossible) {
throw new IllegalStateException(impossible);
}
}
private void requireEmployeeAudience(AihrKnowledgePrincipal principal) {
@@ -330,8 +623,8 @@ public class AihrBroadcastService {
select count(*)
from sys_user u
join aihr_org_snapshot o
on o.tenant_id = u.tenant_id
and (o.person_phone = u.phonenumber or o.ext_party_id = u.user_name)
on binary o.tenant_id = binary u.tenant_id
and (binary o.person_phone = binary u.phonenumber or binary o.ext_party_id = binary u.user_name)
where u.tenant_id = ?
and u.user_id = ?
and u.user_type = ?
@@ -439,7 +732,43 @@ public class AihrBroadcastService {
return new ServiceException("仅在职员工可查看公司消息", HttpStatus.FORBIDDEN);
}
private record PublishReplay(long id, String title, String content) {
private enum BroadcastMessageScope {
ALL,
RELATED,
REQUIRED_UNREAD
}
private record EmployeeMessageQuery(String fromWhere, List<Object> parameters) {
}
private record TargetSelection(
List<String> projectCodes,
List<String> positionNames,
List<String> positionLevels,
String reason
) {
private boolean hasTargets() {
return !projectCodes.isEmpty() || !positionNames.isEmpty() || !positionLevels.isEmpty();
}
}
private record TargetCandidate(String subjectRef, Long userId) {
}
private record TargetAudit(long matchedRecipientCount, long unmatchedTargetCount) {
private static TargetAudit empty() {
return new TargetAudit(0, 0);
}
}
private record PublishReplay(
long id,
String title,
String content,
String targetPayloadHash,
long matchedRecipientCount,
long unmatchedTargetCount
) {
}
/**
@@ -17,6 +17,8 @@ import java.lang.reflect.Proxy;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.security.MessageDigest;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -72,12 +74,39 @@ class AihrBroadcastM1ServiceTest {
assertTrue(jdbcTemplate.messageListSql.contains("aihr_broadcast_target_recipient"));
assertTrue(jdbcTemplate.messageListSql.contains("tr.user_id is not null"));
service.messages(1, 20, "ALL");
BroadcastListResponse allMessages = service.messages(1, 20, "ALL");
assertFalse(jdbcTemplate.messageListSql.contains("tr.user_id is not null"));
assertEquals(1, allMessages.rows().size());
assertTrue(jdbcTemplate.messageListSql.contains("m.status = 'PUBLISHED'"));
}
@Test
void idempotentTargetedRetryReturnsTheOriginalRecipientAudit() {
BroadcastTargetRequest targets = new BroadcastTargetRequest(
List.of("PRJ-FCW"), List.of("物业管家"), List.of()
);
TargetingJdbcTemplate jdbcTemplate = new TargetingJdbcTemplate();
jdbcTemplate.publishReplays = List.of(Map.of(
"id", 42L,
"title", "夏季服务标准",
"content", "请项目客服人员在今日完成学习。",
"target_payload_hash", targetPayloadHash(true, targets),
"matched_recipient_count", 4L,
"unmatched_target_count", 2L
));
AihrBroadcastService service = new AihrBroadcastService(jdbcTemplate, RESOLVER);
PublishResponse response = service.publish(new PublishRequest(
"m1-targeted-retry", "夏季服务标准", "请项目客服人员在今日完成学习。", "tenant-a", true, targets
));
assertEquals(42L, response.id());
assertEquals(4L, response.matchedRecipientCount());
assertEquals(2L, response.unmatchedTargetCount());
assertTrue(jdbcTemplate.publishReplaySql.contains("matched_recipient_count"));
assertTrue(jdbcTemplate.publishReplaySql.contains("unmatched_target_count"));
}
private static final AihrKnowledgePrincipalResolver RESOLVER = new AihrKnowledgePrincipalResolver(null) {
@Override
public AihrKnowledgePrincipal current() {
@@ -86,11 +115,38 @@ class AihrBroadcastM1ServiceTest {
}
};
private static String targetPayloadHash(boolean requiredRead, BroadcastTargetRequest targets) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
updateDigestField(digest, requiredRead ? "1" : "0");
updateDigestField(digest, "PROJECT");
targets.projectCodes().stream().sorted().forEach(value -> updateDigestField(digest, value));
updateDigestField(digest, "POSITION");
targets.positionNames().stream().sorted().forEach(value -> updateDigestField(digest, value));
updateDigestField(digest, "LEVEL");
targets.positionLevels().stream().sorted().forEach(value -> updateDigestField(digest, value));
return java.util.HexFormat.of().formatHex(digest.digest());
} catch (Exception exception) {
throw new AssertionError(exception);
}
}
private static void updateDigestField(MessageDigest digest, String value) {
byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
digest.update((byte) (bytes.length >>> 24));
digest.update((byte) (bytes.length >>> 16));
digest.update((byte) (bytes.length >>> 8));
digest.update((byte) bytes.length);
digest.update(bytes);
}
private static final class TargetingJdbcTemplate extends JdbcTemplate {
private List<Map<String, Object>> targetCandidates = List.of();
private List<Map<String, Object>> publishReplays = List.of();
private String targetCandidateSql;
private Object[] targetCandidateArgs;
private String messageListSql;
private String publishReplaySql;
private final List<String> updateSqls = new ArrayList<>();
@Override
@@ -101,7 +157,10 @@ class AihrBroadcastM1ServiceTest {
@Override
public <T> List<T> query(String sql, RowMapper<T> rowMapper, Object... args) {
if (sql.contains("publish_request_key")) return List.of();
if (sql.contains("publish_request_key")) {
publishReplaySql = sql;
return publishReplays.stream().map(values -> map(rowMapper, values)).toList();
}
if (sql.contains("from aihr_org_snapshot o")) {
targetCandidateSql = sql;
targetCandidateArgs = args;
@@ -63,6 +63,9 @@ class AihrBroadcastServiceTest {
assertEquals(HttpStatus.FORBIDDEN, error.getCode());
assertTrue(jdbcTemplate.queryForObjectSql.contains("from sys_user u"));
assertTrue(jdbcTemplate.queryForObjectSql.contains("binary o.tenant_id = binary u.tenant_id"));
assertTrue(jdbcTemplate.queryForObjectSql.contains("binary o.person_phone = binary u.phonenumber"));
assertTrue(jdbcTemplate.queryForObjectSql.contains("binary o.ext_party_id = binary u.user_name"));
assertArrayEquals(new Object[]{"tenant-a", 7L, "app_user"}, jdbcTemplate.queryForObjectArgs);
}
@@ -76,7 +79,7 @@ class AihrBroadcastServiceTest {
assertEquals(HttpStatus.NOT_FOUND, error.getCode());
assertTrue(jdbcTemplate.querySql.contains("m.tenant_id = ? and m.id = ? and m.status = 'PUBLISHED'"));
assertArrayEquals(new Object[]{7L, "tenant-a", 42L}, jdbcTemplate.queryArgs);
assertArrayEquals(new Object[]{7L, "tenant-a", 7L, "tenant-a", 42L}, jdbcTemplate.queryArgs);
}
@Test
@@ -299,10 +302,12 @@ class AihrBroadcastServiceTest {
"withdraw", Long.class, AihrBroadcastDto.WithdrawRequest.class);
Method adminMessages = AihrBroadcastController.class.getMethod(
"adminMessages", Integer.class, Integer.class, String.class);
Method targetOptions = AihrBroadcastController.class.getMethod("targetOptions");
assertRoleGate(publish.getAnnotation(SaCheckRole.class));
assertRoleGate(withdraw.getAnnotation(SaCheckRole.class));
assertRoleGate(adminMessages.getAnnotation(SaCheckRole.class));
assertRoleGate(targetOptions.getAnnotation(SaCheckRole.class));
assertTrue(withdraw.getParameters()[1].isAnnotationPresent(Valid.class));
}
@@ -310,7 +315,10 @@ class AihrBroadcastServiceTest {
void publishRequestValidatesTransportInputBeforeItReachesTheService() throws Exception {
Method publish = AihrBroadcastController.class.getMethod(
"publish", AihrBroadcastDto.PublishRequest.class);
Constructor<?> constructor = AihrBroadcastDto.PublishRequest.class.getDeclaredConstructors()[0];
Constructor<?> constructor = java.util.Arrays.stream(AihrBroadcastDto.PublishRequest.class.getDeclaredConstructors())
.filter(candidate -> candidate.getParameterCount() == 6)
.findFirst()
.orElseThrow();
assertTrue(publish.getParameters()[0].isAnnotationPresent(Valid.class));
assertTrue(constructor.getParameters()[0].isAnnotationPresent(NotBlank.class));