diff --git a/server/skillhub-app/pom.xml b/server/skillhub-app/pom.xml index 4b57b318..aeeee7a3 100644 --- a/server/skillhub-app/pom.xml +++ b/server/skillhub-app/pom.xml @@ -77,6 +77,11 @@ org.springframework.session spring-session-data-redis + + org.redisson + redisson + 3.51.0 + org.springframework.boot spring-boot-devtools diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java index 7add1244..daf4e2d0 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java @@ -6,14 +6,14 @@ import com.iflytek.skillhub.domain.security.SecurityScanService; import com.iflytek.skillhub.domain.security.SecurityScanner; import com.iflytek.skillhub.domain.skill.SkillRepository; import com.iflytek.skillhub.domain.skill.SkillVersionRepository; -import com.iflytek.skillhub.stream.RedisScanTaskProducer; +import com.iflytek.skillhub.stream.RedissonScanTaskProducer; import com.iflytek.skillhub.stream.ScanTaskConsumer; +import org.redisson.api.RedissonClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.core.StringRedisTemplate; +import java.time.Duration; @Configuration @ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true") @@ -25,13 +25,25 @@ public class RedisStreamConfig { @Value("${skillhub.security.stream.group:skillhub-scanners}") private String groupName; + @Value("${skillhub.security.stream.reclaim-enabled:true}") + private boolean reclaimEnabled; + + @Value("${skillhub.security.stream.reclaim-min-idle:PT2M}") + private Duration reclaimMinIdle; + + @Value("${skillhub.security.stream.reclaim-batch-size:20}") + private int reclaimBatchSize; + + @Value("${skillhub.security.stream.reclaim-interval:PT30S}") + private Duration reclaimInterval; + @Bean - public RedisScanTaskProducer redisScanTaskProducer(StringRedisTemplate redisTemplate) { - return new RedisScanTaskProducer(redisTemplate, streamKey); + public RedissonScanTaskProducer redisScanTaskProducer(RedissonClient redissonClient) { + return new RedissonScanTaskProducer(redissonClient, streamKey); } @Bean - public ScanTaskConsumer scanTaskConsumer(RedisConnectionFactory connectionFactory, + public ScanTaskConsumer scanTaskConsumer(RedissonClient redissonClient, SecurityScanner securityScanner, SecurityScanService securityScanService, SkillVersionRepository skillVersionRepository, @@ -39,7 +51,7 @@ public class RedisStreamConfig { ReviewTaskRepository reviewTaskRepository, ScanTaskProducer scanTaskProducer) { return new ScanTaskConsumer( - connectionFactory, + redissonClient, streamKey, groupName, securityScanner, @@ -47,7 +59,11 @@ public class RedisStreamConfig { skillVersionRepository, skillRepository, reviewTaskRepository, - scanTaskProducer + scanTaskProducer, + reclaimEnabled, + reclaimMinIdle, + reclaimBatchSize, + reclaimInterval ); } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedissonConfig.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedissonConfig.java new file mode 100644 index 00000000..d9897c42 --- /dev/null +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedissonConfig.java @@ -0,0 +1,56 @@ +package com.iflytek.skillhub.config; + +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.redisson.config.SingleServerConfig; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.data.redis.RedisProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.util.StringUtils; + +@Configuration +@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true") +public class RedissonConfig { + + @Bean(destroyMethod = "shutdown") + public RedissonClient redissonClient(RedisProperties redisProperties) { + return Redisson.create(createConfig(redisProperties)); + } + + static Config createConfig(RedisProperties redisProperties) { + Config config = new Config(); + SingleServerConfig singleServerConfig = config.useSingleServer() + .setAddress(resolveAddress(redisProperties)) + .setDatabase(redisProperties.getDatabase()); + + if (StringUtils.hasText(redisProperties.getUsername())) { + singleServerConfig.setUsername(redisProperties.getUsername()); + } + if (StringUtils.hasText(redisProperties.getPassword())) { + singleServerConfig.setPassword(redisProperties.getPassword()); + } + if (StringUtils.hasText(redisProperties.getClientName())) { + singleServerConfig.setClientName(redisProperties.getClientName()); + } + if (redisProperties.getTimeout() != null) { + singleServerConfig.setTimeout(Math.toIntExact(redisProperties.getTimeout().toMillis())); + } + if (redisProperties.getConnectTimeout() != null) { + singleServerConfig.setConnectTimeout(Math.toIntExact(redisProperties.getConnectTimeout().toMillis())); + } + + return config; + } + + private static String resolveAddress(RedisProperties redisProperties) { + if (StringUtils.hasText(redisProperties.getUrl())) { + return redisProperties.getUrl(); + } + String scheme = redisProperties.getSsl() != null && redisProperties.getSsl().isEnabled() + ? "rediss" + : "redis"; + return scheme + "://" + redisProperties.getHost() + ":" + redisProperties.getPort(); + } +} diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java index 812aa84c..229af851 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java @@ -2,113 +2,214 @@ package com.iflytek.skillhub.stream; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; +import org.redisson.api.AutoClaimResult; +import org.redisson.api.RStream; +import org.redisson.api.RedissonClient; +import org.redisson.api.StreamMessageId; +import org.redisson.api.stream.StreamCreateGroupArgs; +import org.redisson.api.stream.StreamReadGroupArgs; +import org.redisson.client.codec.StringCodec; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.data.redis.stream.StreamListener; -import org.springframework.data.redis.stream.StreamMessageListenerContainer; -import org.springframework.data.redis.stream.Subscription; import java.time.Duration; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; -public abstract class AbstractStreamConsumer implements StreamListener> { +public abstract class AbstractStreamConsumer { protected final Logger log = LoggerFactory.getLogger(getClass()); private static final String FIELD_RETRY_COUNT = "retryCount"; private static final int MAX_RETRY_COUNT = 3; + private static final int READ_BATCH_SIZE = 10; + private static final Duration POLL_TIMEOUT = Duration.ofSeconds(2); - private final RedisConnectionFactory connectionFactory; + private final RedissonClient redissonClient; private final String streamKey; private final String groupName; private final String consumerName; - private StringRedisTemplate redisTemplate; + private final boolean reclaimEnabled; + private final Duration reclaimMinIdle; + private final int reclaimBatchSize; + private final Duration reclaimInterval; + private final AtomicBoolean running = new AtomicBoolean(false); - private StreamMessageListenerContainer> container; + private RStream stream; + private ExecutorService consumerExecutor; + private ScheduledExecutorService reclaimExecutor; - protected AbstractStreamConsumer(RedisConnectionFactory connectionFactory, + protected AbstractStreamConsumer(RedissonClient redissonClient, String streamKey, String groupName) { - this.connectionFactory = connectionFactory; + this(redissonClient, streamKey, groupName, true, Duration.ofMinutes(2), 20, Duration.ofSeconds(30)); + } + + protected AbstractStreamConsumer(RedissonClient redissonClient, + String streamKey, + String groupName, + boolean reclaimEnabled, + Duration reclaimMinIdle, + int reclaimBatchSize, + Duration reclaimInterval) { + this.redissonClient = redissonClient; this.streamKey = streamKey; this.groupName = groupName; + this.reclaimEnabled = reclaimEnabled; + this.reclaimMinIdle = reclaimMinIdle; + this.reclaimBatchSize = reclaimBatchSize; + this.reclaimInterval = reclaimInterval; this.consumerName = consumerPrefix() + "-" + UUID.randomUUID().toString().substring(0, 8); } @PostConstruct public void init() { - if (connectionFactory == null) { + if (redissonClient == null) { return; } - this.redisTemplate = createRedisTemplate(); + this.stream = createStream(); initializeStreamAndGroup(); startConsumer(); + startPendingReclaimer(); } @PreDestroy public void shutdown() { - if (container != null) { - container.stop(); + running.set(false); + if (consumerExecutor != null) { + consumerExecutor.shutdownNow(); + } + if (reclaimExecutor != null) { + reclaimExecutor.shutdownNow(); } } private void initializeStreamAndGroup() { try { - StringRedisTemplate template = redisTemplate(); - if (Boolean.FALSE.equals(template.hasKey(streamKey))) { - template.opsForStream().add(streamKey, Map.of("_init", "true")); - } - try { - template.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupName); - } catch (Exception e) { - if (e.getMessage() == null || !e.getMessage().contains("BUSYGROUP")) { - log.warn("Failed to create consumer group: stream={}, group={}", streamKey, groupName, e); - } - } + stream().createGroup(StreamCreateGroupArgs.name(groupName).makeStream()); } catch (Exception e) { - throw new RuntimeException("Failed to initialize Redis Stream consumer", e); + if (!isConsumerGroupAlreadyExists(e)) { + log.warn("Failed to create consumer group: stream={}, group={}", streamKey, groupName, e); + } } } + static boolean isConsumerGroupAlreadyExists(Throwable error) { + Throwable current = error; + while (current != null) { + if (current.getMessage() != null && current.getMessage().contains("BUSYGROUP")) { + return true; + } + current = current.getCause(); + } + return false; + } + private void startConsumer() { - StreamMessageListenerContainer.StreamMessageListenerContainerOptions> options = - StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder() - .pollTimeout(Duration.ofSeconds(2)) - .build(); - - container = StreamMessageListenerContainer.create(connectionFactory, options); - Subscription ignored = container.receive( - Consumer.from(groupName, consumerName), - StreamOffset.create(streamKey, ReadOffset.lastConsumed()), - this - ); - container.start(); + running.set(true); + consumerExecutor = Executors.newSingleThreadExecutor(threadFactory(consumerPrefix() + "-consumer")); + consumerExecutor.submit(this::consumeLoop); } - @Override - public void onMessage(MapRecord message) { - T payload = parsePayload(message.getId().getValue(), message.getValue()); + private void startPendingReclaimer() { + if (!reclaimEnabled) { + return; + } + reclaimExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory(consumerPrefix() + "-reclaimer")); + long intervalMillis = Math.max(1L, reclaimInterval.toMillis()); + reclaimExecutor.scheduleWithFixedDelay(this::safeReclaimPendingMessages, + intervalMillis, intervalMillis, TimeUnit.MILLISECONDS); + } + + private void consumeLoop() { + while (running.get()) { + try { + consumeAvailableMessages(); + } catch (Exception e) { + if (Thread.currentThread().isInterrupted()) { + Thread.currentThread().interrupt(); + break; + } + log.error("Failed while consuming stream messages: stream={}, group={}, consumer={}", + streamKey, groupName, consumerName, e); + } + } + } + + private void safeReclaimPendingMessages() { + try { + reclaimPendingMessages(); + } catch (Exception e) { + log.error("Failed while reclaiming pending stream messages: stream={}, group={}, consumer={}", + streamKey, groupName, consumerName, e); + } + } + + void consumeAvailableMessages() { + Map> messages = stream().readGroup( + groupName, + consumerName, + StreamReadGroupArgs.neverDelivered() + .count(READ_BATCH_SIZE) + .timeout(POLL_TIMEOUT) + ); + processMessages(messages); + } + + void reclaimPendingMessages() { + if (!reclaimEnabled) { + return; + } + StreamMessageId startId = StreamMessageId.MIN; + while (true) { + AutoClaimResult result = stream().autoClaim( + groupName, + consumerName, + reclaimMinIdle.toMillis(), + TimeUnit.MILLISECONDS, + startId, + reclaimBatchSize + ); + if (result == null || result.getMessages() == null || result.getMessages().isEmpty()) { + return; + } + processMessages(result.getMessages()); + if (result.getMessages().size() < reclaimBatchSize || result.getNextId() == null) { + return; + } + startId = result.getNextId(); + } + } + + private void processMessages(Map> messages) { + if (messages == null || messages.isEmpty()) { + return; + } + messages.forEach(this::handleMessage); + } + + void handleMessage(StreamMessageId messageId, Map data) { + T payload = parsePayload(messageId.toString(), data); if (payload == null) { - acknowledge(message); + acknowledge(messageId); return; } - int retryCount = parseRetryCount(message.getValue()); + int retryCount = parseRetryCount(data); try { markProcessing(payload); processBusiness(payload); markCompleted(payload); - acknowledge(message); + acknowledge(messageId); } catch (Exception e) { handleFailure(payload, retryCount, e); - acknowledge(message); + acknowledge(messageId); } } @@ -137,19 +238,31 @@ public abstract class AbstractStreamConsumer implements StreamListener 500 ? error.substring(0, 500) : error; } - protected StringRedisTemplate createRedisTemplate() { - return new StringRedisTemplate(connectionFactory); + protected RStream createStream() { + return redissonClient.getStream(streamKey, StringCodec.INSTANCE); } - protected void acknowledge(MapRecord message) { - redisTemplate().opsForStream().acknowledge(streamKey, groupName, message.getId()); + protected void acknowledge(StreamMessageId messageId) { + stream().ack(groupName, messageId); } - private StringRedisTemplate redisTemplate() { - if (redisTemplate == null) { - redisTemplate = createRedisTemplate(); + protected final RStream stream() { + if (stream == null) { + stream = createStream(); } - return redisTemplate; + return stream; + } + + protected final String consumerName() { + return consumerName; + } + + private ThreadFactory threadFactory(String namePrefix) { + return runnable -> { + Thread thread = new Thread(runnable, namePrefix + "-" + consumerName); + thread.setDaemon(true); + return thread; + }; } protected abstract String taskDisplayName(); diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedisScanTaskProducer.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedissonScanTaskProducer.java similarity index 57% rename from server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedisScanTaskProducer.java rename to server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedissonScanTaskProducer.java index 46d18373..9502582d 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedisScanTaskProducer.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedissonScanTaskProducer.java @@ -2,25 +2,26 @@ package com.iflytek.skillhub.stream; import com.iflytek.skillhub.domain.security.ScanTask; import com.iflytek.skillhub.domain.security.ScanTaskProducer; +import org.redisson.api.RStream; +import org.redisson.api.RedissonClient; +import org.redisson.api.StreamMessageId; +import org.redisson.api.stream.StreamAddArgs; +import org.redisson.client.codec.StringCodec; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.data.redis.connection.stream.RecordId; -import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.data.redis.connection.stream.StringRecord; -import org.springframework.data.redis.core.StringRedisTemplate; import java.util.HashMap; import java.util.Map; -public class RedisScanTaskProducer implements ScanTaskProducer { +public class RedissonScanTaskProducer implements ScanTaskProducer { - private static final Logger log = LoggerFactory.getLogger(RedisScanTaskProducer.class); + private static final Logger log = LoggerFactory.getLogger(RedissonScanTaskProducer.class); - private final StringRedisTemplate redisTemplate; + private final RedissonClient redissonClient; private final String streamKey; - public RedisScanTaskProducer(StringRedisTemplate redisTemplate, String streamKey) { - this.redisTemplate = redisTemplate; + public RedissonScanTaskProducer(RedissonClient redissonClient, String streamKey) { + this.redissonClient = redissonClient; this.streamKey = streamKey; } @@ -36,9 +37,9 @@ public class RedisScanTaskProducer implements ScanTaskProducer { fields.putAll(task.metadata()); } - StringRecord record = StreamRecords.string(fields).withStreamKey(streamKey); - RecordId recordId = redisTemplate.opsForStream().add(record); + RStream stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE); + StreamMessageId messageId = stream.add(StreamAddArgs.entries(fields)); log.info("Published scan task: taskId={}, versionId={}, recordId={}", - task.taskId(), task.versionId(), recordId); + task.taskId(), task.versionId(), messageId); } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java index 05142b9a..2921042e 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java @@ -12,12 +12,13 @@ import com.iflytek.skillhub.domain.security.SecurityScanner; import com.iflytek.skillhub.domain.skill.SkillRepository; import com.iflytek.skillhub.domain.skill.SkillVersionRepository; import com.iflytek.skillhub.domain.skill.SkillVersionStatus; -import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.redisson.api.RedissonClient; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.time.Duration; import java.util.Comparator; import java.util.Map; @@ -31,7 +32,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer streamOperations = mock(StreamOperations.class); - StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); - org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations); - TestConsumer consumer = new TestConsumer(redisTemplate); - MapRecord message = StreamRecords.newRecord() - .in("scan-stream") - .withId(RecordId.of("1-0")) - .ofMap(Map.of("payload", "ok")); + void handleMessage_acknowledgesAfterSuccessfulProcessing() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + TestConsumer consumer = new TestConsumer(stream); + StreamMessageId messageId = new StreamMessageId(1, 0); - consumer.onMessage(message); + consumer.handleMessage(messageId, Map.of("payload", "ok")); - verify(streamOperations).acknowledge("scan-stream", "scan-group", message.getId()); + verify(stream).ack("scan-group", messageId); } @Test - void onMessage_acknowledgesAfterRetryableFailure() { - StreamOperations streamOperations = mock(StreamOperations.class); - StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); - org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations); - TestConsumer consumer = new TestConsumer(redisTemplate); + void handleMessage_acknowledgesAfterRetryableFailure() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + TestConsumer consumer = new TestConsumer(stream); consumer.fail = true; - MapRecord message = StreamRecords.newRecord() - .in("scan-stream") - .withId(RecordId.of("2-0")) - .ofMap(Map.of("payload", "boom")); + StreamMessageId messageId = new StreamMessageId(2, 0); - consumer.onMessage(message); + consumer.handleMessage(messageId, Map.of("payload", "boom")); - verify(streamOperations).acknowledge("scan-stream", "scan-group", message.getId()); - verify(streamOperations, times(1)).acknowledge("scan-stream", "scan-group", message.getId()); + verify(stream).ack("scan-group", messageId); + verify(stream, times(1)).ack("scan-group", messageId); } @Test - void onMessage_reusesRedisTemplateForAcknowledgement() { - StreamOperations streamOperations = mock(StreamOperations.class); - StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class); - org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations); - CountingConsumer consumer = new CountingConsumer(redisTemplate); - MapRecord first = StreamRecords.newRecord() - .in("scan-stream") - .withId(RecordId.of("3-0")) - .ofMap(Map.of("payload", "one")); - MapRecord second = StreamRecords.newRecord() - .in("scan-stream") - .withId(RecordId.of("4-0")) - .ofMap(Map.of("payload", "two")); + void consumeAvailableMessages_processesNeverDeliveredMessages() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + TestConsumer consumer = new TestConsumer(stream); + StreamMessageId first = new StreamMessageId(3, 0); + StreamMessageId second = new StreamMessageId(4, 0); + Map> messages = new LinkedHashMap<>(); + messages.put(first, Map.of("payload", "one")); + messages.put(second, Map.of("payload", "two")); + when(stream.readGroup(eq("scan-group"), anyString(), org.mockito.ArgumentMatchers.any())) + .thenReturn(messages); - consumer.onMessage(first); - consumer.onMessage(second); + consumer.consumeAvailableMessages(); - org.junit.jupiter.api.Assertions.assertEquals(1, consumer.templateCreationCount.get()); + verify(stream).ack("scan-group", first); + verify(stream).ack("scan-group", second); + } + + @Test + void reclaimPendingMessages_autoClaimsAndProcessesMessages() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + TestConsumer consumer = new TestConsumer(stream); + StreamMessageId reclaimedId = new StreamMessageId(5, 0); + when(stream.autoClaim(eq("scan-group"), anyString(), anyLong(), eq(java.util.concurrent.TimeUnit.MILLISECONDS), + eq(StreamMessageId.MIN), anyInt())) + .thenReturn(new AutoClaimResult<>( + StreamMessageId.MAX, + Map.of(reclaimedId, Map.of("payload", "reclaimed")), + java.util.List.of() + )); + + consumer.reclaimPendingMessages(); + + verify(stream).ack("scan-group", reclaimedId); + } + + @Test + void handleMessage_reusesStreamInstanceForAcknowledgement() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + CountingConsumer consumer = new CountingConsumer(stream); + + consumer.handleMessage(new StreamMessageId(6, 0), Map.of("payload", "one")); + consumer.handleMessage(new StreamMessageId(7, 0), Map.of("payload", "two")); + + assertThat(consumer.streamCreationCount.get()).isEqualTo(1); + } + + @Test + void detectsBusyGroupWhenWrappedInRedisSystemException() { + RedisSystemException wrapped = new RedisSystemException( + "Error in execution", + new RedisBusyException("BUSYGROUP Consumer Group name already exists") + ); + + assertThat(AbstractStreamConsumer.isConsumerGroupAlreadyExists(wrapped)).isTrue(); } private static class TestConsumer extends AbstractStreamConsumer { - private final StringRedisTemplate redisTemplate; + private final RStream stream; private boolean fail; - private TestConsumer(StringRedisTemplate redisTemplate) { - super(mock(RedisConnectionFactory.class), "scan-stream", "scan-group"); - this.redisTemplate = redisTemplate; + private TestConsumer(RStream stream) { + super(mock(RedissonClient.class), "scan-stream", "scan-group", true, Duration.ofMinutes(2), 20, Duration.ofSeconds(30)); + this.stream = stream; } @Override - protected StringRedisTemplate createRedisTemplate() { - return redisTemplate; + protected RStream createStream() { + return stream; } @Override @@ -130,16 +173,16 @@ class AbstractStreamConsumerTest { } private static final class CountingConsumer extends TestConsumer { - private final AtomicInteger templateCreationCount = new AtomicInteger(); + private final AtomicInteger streamCreationCount = new AtomicInteger(); - private CountingConsumer(StringRedisTemplate redisTemplate) { - super(redisTemplate); + private CountingConsumer(RStream stream) { + super(stream); } @Override - protected StringRedisTemplate createRedisTemplate() { - templateCreationCount.incrementAndGet(); - return super.createRedisTemplate(); + protected RStream createStream() { + streamCreationCount.incrementAndGet(); + return super.createStream(); } } } diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/RedissonScanTaskProducerTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/RedissonScanTaskProducerTest.java new file mode 100644 index 00000000..18bfa24d --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/RedissonScanTaskProducerTest.java @@ -0,0 +1,48 @@ +package com.iflytek.skillhub.stream; + +import com.iflytek.skillhub.domain.security.ScanTask; +import org.junit.jupiter.api.Test; +import org.redisson.api.RStream; +import org.redisson.api.RedissonClient; +import org.redisson.api.StreamMessageId; +import org.redisson.client.codec.StringCodec; +import org.redisson.api.stream.StreamAddArgs; +import org.mockito.ArgumentCaptor; + +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class RedissonScanTaskProducerTest { + + @Test + void publishScanTask_writesExpectedFieldsToConfiguredStream() { + @SuppressWarnings("unchecked") + RStream stream = mock(RStream.class); + @SuppressWarnings("unchecked") + RStream typedStream = (RStream) (RStream) stream; + RedissonClient redissonClient = mock(RedissonClient.class); + doReturn(typedStream).when(redissonClient).getStream("skillhub:scan:requests", StringCodec.INSTANCE); + when(stream.add(any())).thenReturn(new StreamMessageId(1, 0)); + RedissonScanTaskProducer producer = new RedissonScanTaskProducer(redissonClient, "skillhub:scan:requests"); + + producer.publishScanTask(new ScanTask( + "task-1", + 42L, + "/tmp/skill", + "publisher-1", + 1711260000000L, + Map.of("scannerType", "skill-scanner") + )); + + verify(redissonClient).getStream("skillhub:scan:requests", StringCodec.INSTANCE); + ArgumentCaptor> argsCaptor = ArgumentCaptor.forClass(StreamAddArgs.class); + verify(stream).add(argsCaptor.capture()); + assertThat(argsCaptor.getValue()).isNotNull(); + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerPathSafetyTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerPathSafetyTest.java index 4f8eb535..d3016115 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerPathSafetyTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerPathSafetyTest.java @@ -12,14 +12,14 @@ import java.lang.reflect.Method; import java.nio.file.Files; import java.nio.file.Path; import org.junit.jupiter.api.Test; -import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.redisson.api.RedissonClient; class ScanTaskConsumerPathSafetyTest { @Test void cleanupTempPath_ignoresPathsOutsideScanTempDirectory() throws Exception { ScanTaskConsumer consumer = new ScanTaskConsumer( - org.mockito.Mockito.mock(RedisConnectionFactory.class), + org.mockito.Mockito.mock(RedissonClient.class), "scan-stream", "scan-group", org.mockito.Mockito.mock(SecurityScanner.class), diff --git a/web/src/features/security-audit/display-state.ts b/web/src/features/security-audit/display-state.ts new file mode 100644 index 00000000..611c86e4 --- /dev/null +++ b/web/src/features/security-audit/display-state.ts @@ -0,0 +1,7 @@ +import type { SecurityAuditDisplayState, SecurityAuditRecord } from './types' + +export function getSecurityAuditDisplayState( + audit: Pick +): SecurityAuditDisplayState { + return audit.scannedAt ? audit.verdict : 'SCANNING' +} diff --git a/web/src/features/security-audit/security-audit-section.test.tsx b/web/src/features/security-audit/security-audit-section.test.tsx index 5c51de34..fdb77398 100644 --- a/web/src/features/security-audit/security-audit-section.test.tsx +++ b/web/src/features/security-audit/security-audit-section.test.tsx @@ -98,6 +98,16 @@ describe('SecurityAuditSection', () => { expect(html).toContain('securityAudit.verdict.SUSPICIOUS') }) + it('renders scanning status when the audit has not completed yet', () => { + mockAudits = [createAudit({ verdict: 'SUSPICIOUS', scannedAt: null })] + mockIsLoading = false + + const html = renderToStaticMarkup() + + expect(html).toContain('securityAudit.statusScanning') + expect(html).not.toContain('securityAudit.verdict.SUSPICIOUS') + }) + it('renders the findings count', () => { mockAudits = [createAudit({ findingsCount: 3 })] mockIsLoading = false diff --git a/web/src/features/security-audit/security-audit-section.tsx b/web/src/features/security-audit/security-audit-section.tsx index 3abc999f..2a3600dc 100644 --- a/web/src/features/security-audit/security-audit-section.tsx +++ b/web/src/features/security-audit/security-audit-section.tsx @@ -4,6 +4,7 @@ import { ChevronDown, ChevronUp, Shield } from 'lucide-react' import { Card } from '@/shared/ui/card' import { Button } from '@/shared/ui/button' import { useSecurityAudits } from './use-security-audit' +import { getSecurityAuditDisplayState } from './display-state' import { VerdictBadge } from './verdict-badge' import { FindingItem } from './finding-item' import type { FindingSeverity, SecurityAuditRecord } from './types' @@ -72,7 +73,7 @@ function ScannerCard({ audit }: { audit: SecurityAuditRecord }) {
{audit.scannerType} - +
diff --git a/web/src/features/security-audit/security-audit-summary.test.tsx b/web/src/features/security-audit/security-audit-summary.test.tsx index c668275c..efdd9344 100644 --- a/web/src/features/security-audit/security-audit-summary.test.tsx +++ b/web/src/features/security-audit/security-audit-summary.test.tsx @@ -97,6 +97,15 @@ describe('SecurityAuditSummary', () => { expect(html).toContain('securityAudit.verdict.BLOCKED') }) + it('renders scanning status when the audit has not completed yet', () => { + mockAudits = [createAudit({ verdict: 'SUSPICIOUS', scannedAt: null })] + + const html = renderToStaticMarkup() + + expect(html).toContain('securityAudit.statusScanning') + expect(html).not.toContain('securityAudit.verdict.SUSPICIOUS') + }) + it('renders the total findings count across all audits', () => { mockAudits = [ createAudit({ id: 1, findingsCount: 3 }), diff --git a/web/src/features/security-audit/security-audit-summary.tsx b/web/src/features/security-audit/security-audit-summary.tsx index cd35f57d..0fe4344b 100644 --- a/web/src/features/security-audit/security-audit-summary.tsx +++ b/web/src/features/security-audit/security-audit-summary.tsx @@ -5,6 +5,7 @@ import { Card } from '@/shared/ui/card' import { Button } from '@/shared/ui/button' import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogDescription } from '@/shared/ui/dialog' import { useSecurityAudits } from './use-security-audit' +import { getSecurityAuditDisplayState } from './display-state' import { VerdictBadge } from './verdict-badge' import { SecurityAuditSection } from './security-audit-section' @@ -40,7 +41,7 @@ export function SecurityAuditSummary({ skillId, versionId }: SecurityAuditSummar className="flex items-center justify-between rounded-xl border border-border/60 bg-secondary/20 p-3" > {audit.scannerType} - +
))}
diff --git a/web/src/features/security-audit/types.ts b/web/src/features/security-audit/types.ts index f0ca1cbb..6990956d 100644 --- a/web/src/features/security-audit/types.ts +++ b/web/src/features/security-audit/types.ts @@ -1,4 +1,5 @@ export type SecurityVerdict = 'SAFE' | 'SUSPICIOUS' | 'DANGEROUS' | 'BLOCKED' +export type SecurityAuditDisplayState = SecurityVerdict | 'SCANNING' export type FindingSeverity = 'CRITICAL' | 'HIGH' | 'MEDIUM' | 'LOW' | 'INFO' export interface SecurityFinding { diff --git a/web/src/features/security-audit/verdict-badge.test.tsx b/web/src/features/security-audit/verdict-badge.test.tsx index 6a722359..6c4ca8e9 100644 --- a/web/src/features/security-audit/verdict-badge.test.tsx +++ b/web/src/features/security-audit/verdict-badge.test.tsx @@ -47,6 +47,12 @@ describe('VerdictBadge', () => { expect(html).toContain('text-red-700') }) + it('renders the scanning label for pending scan display state', () => { + const html = renderToStaticMarkup() + + expect(html).toContain('securityAudit.statusScanning') + }) + it('renders as a span with rounded-full pill styling', () => { const html = renderToStaticMarkup() diff --git a/web/src/features/security-audit/verdict-badge.tsx b/web/src/features/security-audit/verdict-badge.tsx index e9d3adf7..78d4f7db 100644 --- a/web/src/features/security-audit/verdict-badge.tsx +++ b/web/src/features/security-audit/verdict-badge.tsx @@ -1,25 +1,36 @@ import { useTranslation } from 'react-i18next' -import type { SecurityVerdict } from './types' +import type { SecurityAuditDisplayState, SecurityVerdict } from './types' interface VerdictBadgeProps { - verdict: SecurityVerdict + verdict?: SecurityVerdict + displayState?: SecurityAuditDisplayState } -export function VerdictBadge({ verdict }: VerdictBadgeProps) { +export function VerdictBadge({ verdict, displayState }: VerdictBadgeProps) { const { t } = useTranslation() + const state = displayState ?? verdict + + if (!state) { + return null + } const styles = { + SCANNING: 'bg-sky-500/10 text-sky-700 dark:text-sky-400', SAFE: 'bg-emerald-500/10 text-emerald-700 dark:text-emerald-400', SUSPICIOUS: 'bg-amber-500/10 text-amber-700 dark:text-amber-400', DANGEROUS: 'bg-orange-500/10 text-orange-700 dark:text-orange-400', BLOCKED: 'bg-red-500/10 text-red-700 dark:text-red-400', } + const label = state === 'SCANNING' + ? t('securityAudit.statusScanning') + : t(`securityAudit.verdict.${state}`) + return ( - {t(`securityAudit.verdict.${verdict}`)} + {label} ) } diff --git a/web/src/i18n/locales/en.json b/web/src/i18n/locales/en.json index 832b9f91..e4e00378 100644 --- a/web/src/i18n/locales/en.json +++ b/web/src/i18n/locales/en.json @@ -1268,13 +1268,14 @@ "findingsCount": "{{count}} findings", "totalFindings": "{{count}} findings total", "scanDuration": "{{seconds}}s", + "statusScanning": "Scanning", "remediation": "Remediation", "viewDetails": "View Details", "verdict": { "SAFE": "Safe", "SUSPICIOUS": "Suspicious", "DANGEROUS": "Dangerous", - "BLOCKED": "Blocked" + "BLOCKED": "High Risk" }, "severity": { "CRITICAL": "Critical", diff --git a/web/src/i18n/locales/zh.json b/web/src/i18n/locales/zh.json index 725ce11f..2ec0194b 100644 --- a/web/src/i18n/locales/zh.json +++ b/web/src/i18n/locales/zh.json @@ -1268,13 +1268,14 @@ "findingsCount": "{{count}} 项发现", "totalFindings": "共 {{count}} 项发现", "scanDuration": "{{seconds}}s", + "statusScanning": "扫描中", "remediation": "修复建议", "viewDetails": "查看详情", "verdict": { "SAFE": "安全", "SUSPICIOUS": "可疑", "DANGEROUS": "危险", - "BLOCKED": "已拦截" + "BLOCKED": "高风险" }, "severity": { "CRITICAL": "严重", diff --git a/web/src/i18n/security-audit-locale.test.ts b/web/src/i18n/security-audit-locale.test.ts new file mode 100644 index 00000000..3cb5e5f6 --- /dev/null +++ b/web/src/i18n/security-audit-locale.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from 'vitest' +import en from './locales/en.json' +import zh from './locales/zh.json' + +describe('security audit locales', () => { + it('defines the scanning label in both locales', () => { + expect(zh.securityAudit.statusScanning).toBe('扫描中') + expect(en.securityAudit.statusScanning).toBe('Scanning') + }) + + it('uses the updated blocked wording in both locales', () => { + expect(zh.securityAudit.verdict.BLOCKED).toBe('高风险') + expect(en.securityAudit.verdict.BLOCKED).toBe('High Risk') + }) +})