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 fce9bcc2..31e6550c 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 @@ -128,6 +128,10 @@ public class ScanTaskConsumer extends AbstractStreamConsumer candidates = new LinkedHashMap<>(); + repository.findPendingDue(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); + repository.findExpiredLeases(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); + for (ScanTaskOutbox outbox : candidates.values()) { + if (!outbox.claim(now, lease)) continue; + repository.saveAndFlush(outbox); + try { + producer.publishScanTask(outbox.toScanTask()); + outbox.markSent(Instant.now(clock)); + repository.save(outbox); + } catch (Exception e) { + Duration delay = retryDelay(outbox.getRetryCount() + 1); + outbox.markRetry(Instant.now(clock), delay, e.toString()); + repository.save(outbox); + log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", + outbox.getTaskId(), outbox.getRetryCount(), delay, e); + } + } + } + + @Scheduled(cron = "0 20 2 * * ?") + @Transactional + public void cleanupSent() { + int deleted = repository.deleteSentBefore(Instant.now(clock).minus(Duration.ofDays(7))); + if (deleted > 0) log.info("Cleaned up {} sent scan outbox records", deleted); + } + + private Duration retryDelay(int retryCount) { + long seconds = Math.min(maxBackoff.toSeconds(), 1L << Math.min(retryCount, 16)); + return Duration.ofSeconds(Math.max(seconds, 1)); + } +} \ No newline at end of file diff --git a/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql b/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql new file mode 100644 index 00000000..2b56fb36 --- /dev/null +++ b/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql @@ -0,0 +1,28 @@ +CREATE TABLE scan_task_outbox ( + id BIGSERIAL PRIMARY KEY, + task_id VARCHAR(100) NOT NULL, + version_id BIGINT NOT NULL, + skill_path VARCHAR(1000), + bundle_key VARCHAR(1000), + publisher_id VARCHAR(255), + status VARCHAR(20) NOT NULL, + retry_count INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL, + lease_until TIMESTAMPTZ, + last_error VARCHAR(2000), + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + entity_version BIGINT NOT NULL DEFAULT 0, + CONSTRAINT uk_scan_task_outbox_task_id UNIQUE (task_id), + CONSTRAINT ck_scan_task_outbox_status CHECK (status IN ('PENDING', 'SENDING', 'SENT', 'FAILED')) +); + +CREATE INDEX idx_scan_task_outbox_pending + ON scan_task_outbox (status, next_attempt_at, created_at); +CREATE INDEX idx_scan_task_outbox_lease + ON scan_task_outbox (status, lease_until); +CREATE INDEX idx_scan_task_outbox_version + ON scan_task_outbox (version_id); + +ALTER TABLE security_audit ADD COLUMN task_id VARCHAR(100); +CREATE INDEX idx_security_audit_task_id ON security_audit (task_id); \ No newline at end of file diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java new file mode 100644 index 00000000..eae1e0ec --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java @@ -0,0 +1,49 @@ +package com.iflytek.skillhub.task; + +import com.iflytek.skillhub.domain.security.ScanTask; +import com.iflytek.skillhub.domain.security.ScanTaskOutbox; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskProducer; +import com.iflytek.skillhub.domain.security.ScannerType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; + +@ExtendWith(MockitoExtension.class) +class ScanTaskOutboxDispatcherTest { + @Mock ScanTaskOutboxRepository repository; + @Mock ScanTaskProducer producer; + + @Test + void failedRedisPublishLeavesTaskPendingForRetry() { + Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC); + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-1", 1L, "/tmp/1", null, "user", 1L, + java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()))); + given(repository.findPendingDue(any(), any(Integer.class))).willReturn(List.of(outbox)); + given(repository.findExpiredLeases(any(), any(Integer.class))).willReturn(List.of()); + doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); + ScanTaskOutboxDispatcher dispatcher = new ScanTaskOutboxDispatcher( + repository, producer, clock, 50, Duration.ofMinutes(2), Duration.ofMinutes(5)); + + dispatcher.dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getRetryCount()).isEqualTo(1); + verify(producer).publishScanTask(any()); + verify(repository).saveAndFlush(outbox); + } +} \ No newline at end of file diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java new file mode 100644 index 00000000..8f93fce1 --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java @@ -0,0 +1,110 @@ +package com.iflytek.skillhub.domain.security; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.PrePersist; +import jakarta.persistence.Table; +import jakarta.persistence.Version; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; + +@Entity +@Table(name = "scan_task_outbox") +public class ScanTaskOutbox { + @Id @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + @Column(name = "task_id", nullable = false, unique = true, length = 100) + private String taskId; + @Column(name = "version_id", nullable = false) + private Long versionId; + @Column(name = "skill_path", length = 1000) + private String skillPath; + @Column(name = "bundle_key", length = 1000) + private String bundleKey; + @Column(name = "publisher_id", length = 255) + private String publisherId; + @Enumerated(EnumType.STRING) @Column(nullable = false, length = 20) + private ScanTaskOutboxStatus status; + @Column(name = "retry_count", nullable = false) + private int retryCount; + @Column(name = "next_attempt_at", nullable = false) + private Instant nextAttemptAt; + @Column(name = "lease_until") + private Instant leaseUntil; + @Column(name = "last_error", length = 2000) + private String lastError; + @Column(name = "created_at", nullable = false, updatable = false) + private Instant createdAt; + @Column(name = "updated_at", nullable = false) + private Instant updatedAt; + @Version @Column(nullable = false) + private long entityVersion; + + protected ScanTaskOutbox() { } + + public ScanTaskOutbox(ScanTask task) { + this.taskId = task.taskId(); + this.versionId = task.versionId(); + this.skillPath = task.skillPath(); + this.bundleKey = task.bundleKey(); + this.publisherId = task.publisherId(); + this.status = ScanTaskOutboxStatus.PENDING; + this.nextAttemptAt = Instant.now(Clock.systemUTC()); + } + + @PrePersist + protected void onCreate() { + Instant now = Instant.now(Clock.systemUTC()); + createdAt = now; + updatedAt = now; + if (nextAttemptAt == null) nextAttemptAt = now; + } + + public ScanTask toScanTask() { + return new ScanTask(taskId, versionId, skillPath, bundleKey, publisherId, + createdAt == null ? System.currentTimeMillis() : createdAt.toEpochMilli(), + Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + } + + public boolean claim(Instant now, Duration lease) { + if (status != ScanTaskOutboxStatus.PENDING + && !(status == ScanTaskOutboxStatus.SENDING && leaseUntil != null && leaseUntil.isBefore(now))) return false; + status = ScanTaskOutboxStatus.SENDING; + leaseUntil = now.plus(lease); + updatedAt = now; + return true; + } + + public void markSent(Instant now) { + status = ScanTaskOutboxStatus.SENT; + leaseUntil = null; + lastError = null; + updatedAt = now; + } + + public void markRetry(Instant now, Duration delay, String error) { + retryCount++; + status = ScanTaskOutboxStatus.PENDING; + nextAttemptAt = now.plus(delay); + leaseUntil = null; + lastError = error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + updatedAt = now; + } + + public Long getId() { return id; } + public String getTaskId() { return taskId; } + public Long getVersionId() { return versionId; } + public ScanTaskOutboxStatus getStatus() { return status; } + public int getRetryCount() { return retryCount; } + public Instant getNextAttemptAt() { return nextAttemptAt; } + public Instant getLeaseUntil() { return leaseUntil; } + public Instant getCreatedAt() { return createdAt; } +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java new file mode 100644 index 00000000..43c5c24e --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java @@ -0,0 +1,13 @@ +package com.iflytek.skillhub.domain.security; + +import java.time.Instant; +import java.util.List; + +public interface ScanTaskOutboxRepository { + ScanTaskOutbox save(ScanTaskOutbox outbox); + ScanTaskOutbox saveAndFlush(ScanTaskOutbox outbox); + List findPendingDue(Instant now, int limit); + List findExpiredLeases(Instant now, int limit); + int deleteSentBefore(Instant cutoff); + int deleteByVersionId(Long versionId); +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java new file mode 100644 index 00000000..dc0fd83a --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java @@ -0,0 +1,8 @@ +package com.iflytek.skillhub.domain.security; + +public enum ScanTaskOutboxStatus { + PENDING, + SENDING, + SENT, + FAILED +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java index 055287f0..0b7844de 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java @@ -26,6 +26,9 @@ public class SecurityAudit { @Column(name = "skill_version_id", nullable = false) private Long skillVersionId; + @Column(name = "task_id", length = 100) + private String taskId; + @Column(name = "scan_id", length = 100) private String scanId; @@ -66,8 +69,13 @@ public class SecurityAudit { } public SecurityAudit(Long skillVersionId, ScannerType scannerType) { + this(skillVersionId, scannerType, null); + } + + public SecurityAudit(Long skillVersionId, ScannerType scannerType, String taskId) { this.skillVersionId = skillVersionId; this.scannerType = scannerType; + this.taskId = taskId; this.verdict = SecurityVerdict.SUSPICIOUS; this.isSafe = false; this.findingsCount = 0; @@ -91,6 +99,10 @@ public class SecurityAudit { return scanId; } + public String getTaskId() { + return taskId; + } + public ScannerType getScannerType() { return scannerType; } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java index 2ae16856..9bde6ea6 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java @@ -12,6 +12,8 @@ public interface SecurityAuditRepository { Optional findByScanId(String scanId); + boolean existsByTaskIdAndScannedAtIsNotNull(String taskId); + boolean existsBySkillVersionId(Long skillVersionId); /** diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java index 2d455064..43aa8962 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java @@ -9,6 +9,7 @@ import com.iflytek.skillhub.domain.skill.SkillVersionStatus; import com.iflytek.skillhub.domain.skill.validation.PackageEntry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -32,23 +33,36 @@ public class SecurityScanService { private final SecurityAuditRepository auditRepository; private final SkillVersionRepository skillVersionRepository; + private final ScanTaskOutboxRepository scanTaskOutboxRepository; private final ScanTaskProducer scanTaskProducer; private final ObjectMapper objectMapper; private final String scanMode; private final boolean enabled; + @Autowired public SecurityScanService(SecurityAuditRepository auditRepository, SkillVersionRepository skillVersionRepository, ScanTaskProducer scanTaskProducer, ObjectMapper objectMapper, @Value("${skillhub.security.scanner.mode:local}") String scanMode, - @Value("${skillhub.security.scanner.enabled:false}") boolean enabled) { + @Value("${skillhub.security.scanner.enabled:false}") boolean enabled, + ScanTaskOutboxRepository scanTaskOutboxRepository) { this.auditRepository = auditRepository; this.skillVersionRepository = skillVersionRepository; this.scanTaskProducer = scanTaskProducer; this.objectMapper = objectMapper; this.scanMode = scanMode; this.enabled = enabled; + this.scanTaskOutboxRepository = scanTaskOutboxRepository; + } + + public SecurityScanService(SecurityAuditRepository auditRepository, + SkillVersionRepository skillVersionRepository, + ScanTaskProducer scanTaskProducer, + ObjectMapper objectMapper, + String scanMode, + boolean enabled) { + this(auditRepository, skillVersionRepository, scanTaskProducer, objectMapper, scanMode, enabled, null); } public boolean isEnabled() { @@ -74,7 +88,6 @@ public class SecurityScanService { packagePath = saveTempDirectory(versionId, entries).toString(); } // Always create a new audit record — supports multiple rounds per version - auditRepository.save(new SecurityAudit(versionId, ScannerType.SKILL_SCANNER)); final ScanTask scanTask = new ScanTask( UUID.randomUUID().toString(), versionId, @@ -84,9 +97,12 @@ public class SecurityScanService { System.currentTimeMillis(), Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()) ); - // The stream consumer must not observe this task before skill_version / - // security_audit rows are committed and visible. - TransactionCommitCallbacks.afterCommitOrNow(() -> scanTaskProducer.publishScanTask(scanTask)); + auditRepository.save(new SecurityAudit(versionId, ScannerType.SKILL_SCANNER, scanTask.taskId())); + if (scanTaskOutboxRepository != null) { + scanTaskOutboxRepository.save(new ScanTaskOutbox(scanTask)); + } else { + TransactionCommitCallbacks.afterCommitOrNow(() -> scanTaskProducer.publishScanTask(scanTask)); + } // Only transition to SCANNING if the version is not already published (auto-publish flow) if (version.getStatus() != SkillVersionStatus.PUBLISHED) { version.setStatus(SkillVersionStatus.SCANNING); @@ -94,6 +110,10 @@ public class SecurityScanService { } } + public boolean isTaskAlreadyProcessed(String taskId) { + return taskId != null && auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId); + } + @Transactional public void processScanResult(Long versionId, ScannerType scannerType, SecurityScanResponse response) { SecurityAudit audit = auditRepository.findLatestActiveByVersionIdAndScannerType(versionId, scannerType) @@ -203,5 +223,8 @@ public class SecurityScanService { @Transactional public void hardDeleteByVersionId(Long versionId) { auditRepository.deleteBySkillVersionId(versionId); + if (scanTaskOutboxRepository != null) { + scanTaskOutboxRepository.deleteByVersionId(versionId); + } } } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java new file mode 100644 index 00000000..8abf6ca7 --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java @@ -0,0 +1,2 @@ +/** Security scanning domain model and durable task dispatch ports. */ +package com.iflytek.skillhub.domain.security; diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java new file mode 100644 index 00000000..968fd1e8 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java @@ -0,0 +1,40 @@ +package com.iflytek.skillhub.domain.security; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class ScanTaskOutboxTest { + @Test + void claimAndMarkSentProducesStableTaskPayload() { + ScanTask task = new ScanTask("task-1", 7L, "/tmp/7", null, "u1", 123L, + Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + ScanTaskOutbox outbox = new ScanTaskOutbox(task); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + + assertThat(outbox.claim(now, Duration.ofMinutes(2))).isTrue(); + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENDING); + outbox.markSent(now.plusSeconds(1)); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); + assertThat(outbox.toScanTask().taskId()).isEqualTo("task-1"); + assertThat(outbox.toScanTask().versionId()).isEqualTo(7L); + } + + @Test + void failedPublishReturnsToPendingWithBackoffAndTruncatesError() { + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-2", 8L, null, "packages/1/8/bundle.zip", null, 1L, Map.of())); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + outbox.claim(now, Duration.ofMinutes(2)); + outbox.markRetry(now, Duration.ofSeconds(5), "x".repeat(5000)); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getRetryCount()).isEqualTo(1); + assertThat(outbox.getNextAttemptAt()).isEqualTo(now.plusSeconds(5)); + } +} \ No newline at end of file diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java new file mode 100644 index 00000000..ce0b247e --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java @@ -0,0 +1,47 @@ +package com.iflytek.skillhub.domain.security; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.iflytek.skillhub.domain.skill.SkillVersion; +import com.iflytek.skillhub.domain.skill.SkillVersionRepository; +import com.iflytek.skillhub.domain.skill.validation.PackageEntry; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.lang.reflect.Field; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +@ExtendWith(MockitoExtension.class) +class SecurityScanOutboxTest { + @Mock SecurityAuditRepository auditRepository; + @Mock SkillVersionRepository versionRepository; + @Mock ScanTaskProducer producer; + @Mock ScanTaskOutboxRepository outboxRepository; + + @Test + void triggerPersistsAuditStateAndOutboxWithoutPublishingInsideTransaction() throws Exception { + SkillVersion version = new SkillVersion(9L, "1.0.0", "publisher"); + Field id = SkillVersion.class.getDeclaredField("id"); + id.setAccessible(true); + id.set(version, 42L); + given(versionRepository.findById(42L)).willReturn(Optional.of(version)); + SecurityScanService service = new SecurityScanService(auditRepository, versionRepository, producer, + new ObjectMapper(), "upload", true, outboxRepository); + + service.triggerScan(42L, List.of(new PackageEntry("SKILL.md", new byte[0], 0, "text/markdown")), "publisher"); + + ArgumentCaptor outbox = ArgumentCaptor.forClass(ScanTaskOutbox.class); + verify(outboxRepository).save(outbox.capture()); + verify(producer, never()).publishScanTask(org.mockito.ArgumentMatchers.any()); + assertThat(outbox.getValue().getVersionId()).isEqualTo(42L); + assertThat(outbox.getValue().getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); + } +} \ No newline at end of file diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java new file mode 100644 index 00000000..04f7eb4e --- /dev/null +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -0,0 +1,43 @@ +package com.iflytek.skillhub.infra.jpa; + +import com.iflytek.skillhub.domain.security.ScanTaskOutbox; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; + +import java.time.Instant; +import java.util.List; + +@Repository +public interface ScanTaskOutboxJpaRepository extends JpaRepository, ScanTaskOutboxRepository { + @Override + default List findPendingDue(Instant now, int limit) { + return findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( + ScanTaskOutboxStatus.PENDING, now, PageRequest.of(0, limit)); + } + + @Override + default List findExpiredLeases(Instant now, int limit) { + return findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( + ScanTaskOutboxStatus.SENDING, now, PageRequest.of(0, limit)); + } + + @Override + @Modifying + @Query("DELETE FROM ScanTaskOutbox o WHERE o.status = com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.SENT AND o.updatedAt < :cutoff") + int deleteSentBefore(@Param("cutoff") Instant cutoff); + + @Override + void deleteByVersionId(Long versionId); + + List findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( + ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); + + List findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( + ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); +} \ No newline at end of file