[FEATURE] 저장 링크 주간 재검사 스케줄러 추가

src/main/java/com/linclean/domain/analysis/entity/Analysis.java

@@ -106,6 +106,16 @@ public void updateToSucceeded(
         this.elapsedMs = elapsedMs;
     }
 
+    public void markForRecheck(UUID newRequestId, Instant checkedAt) {
+        this.status = AnalysisStatus.QUEUED;
+        this.requestId = newRequestId;
+        this.lastCheckedAt = checkedAt;
+        // 직전이 FAILED였던 건도 재검사 대상이므로, 과거 오류 흔적을 지우고 다시 시작한다.
+        this.errorCode = null;
+        this.errorStage = null;
+        this.errorMessage = null;
+    }
+
     public void updateToFailed(
             String errorCode, Integer errorStage, String errorMessage,
             String engineVersion, Instant analyzedAt, Integer elapsedMs) {

src/main/java/com/linclean/domain/analysis/repository/AnalysisReasonRepository.java

@@ -8,4 +8,6 @@
 
 public interface AnalysisReasonRepository extends JpaRepository<AnalysisReason, Long> {
     List<AnalysisReason> findAllByAnalysis_AnalysisId(UUID analysisId);
+
+    void deleteAllByAnalysis_AnalysisId(UUID analysisId);
 }

src/main/java/com/linclean/domain/analysis/repository/AnalysisRepository.java

@@ -1,14 +1,34 @@
 package com.linclean.domain.analysis.repository;
 
 import com.linclean.domain.analysis.entity.Analysis;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Slice;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
+import java.time.Instant;
 import java.util.List;
 import java.util.UUID;
 
 public interface AnalysisRepository extends JpaRepository<Analysis, UUID> {
 
     @Query("SELECT a.verdict, COUNT(a) FROM Analysis a WHERE a.verdict IS NOT NULL GROUP BY a.verdict")
     List<Object[]> countGroupByVerdict();
+
+    // SUCCEEDED 뿐 아니라 FAILED(지난 재검사 실패)도 포함해, 일시 장애로 실패한 링크가
+    // 다음 주기에 재시도되도록 한다. QUEUED(초기 분석/재검사 진행 중)는 제외한다.
+    @Query("""
+            SELECT a FROM Analysis a
+            WHERE a.status IN (
+                    com.linclean.domain.analysis.entity.AnalysisStatus.SUCCEEDED,
+                    com.linclean.domain.analysis.entity.AnalysisStatus.FAILED)
+              AND COALESCE(a.lastCheckedAt, a.analyzedAt) < :threshold
+              AND a.analysisId > :cursor
+              AND EXISTS (SELECT 1 FROM SavedLink s WHERE s.analysis = a)
+            ORDER BY a.analysisId
+            """)
+    Slice<Analysis> findStaleSavedAnalyses(@Param("threshold") Instant threshold,
+                                           @Param("cursor") UUID cursor,
+                                           Pageable pageable);
 }

src/main/java/com/linclean/domain/analysis/service/AnalysisCallbackService.java

@@ -79,6 +79,9 @@ private void applySucceeded(Analysis analysis, AnalysisResultCallback callback)
                         .message(r.message())
                         .build())
                 .toList();
+        // 재검사 성공 콜백은 SUCCEEDED 전환을 반복하므로, 기존 reason을 지운 뒤 새로 저장한다.
+        // (같은 @Transactional 안이라 커밋 전까지 다른 조회는 과거 reason을 그대로 본다.)
+        analysisReasonRepository.deleteAllByAnalysis_AnalysisId(analysis.getAnalysisId());
         analysisReasonRepository.saveAll(reasons);
     }
 

src/main/java/com/linclean/domain/link/service/SavedLinkRecheckDispatcher.java

@@ -0,0 +1,48 @@
+package com.linclean.domain.link.service;
+
+import com.linclean.domain.analysis.entity.Analysis;
+import com.linclean.domain.analysis.repository.AnalysisRepository;
+import com.linclean.domain.analysis.service.AnalysisAsyncRunner;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.time.Instant;
+import java.util.UUID;
+
+/**
+ * 저장 링크 재검사 1건을 처리한다.
+ * 기존 Analysis 를 QUEUED 로 리셋하고 새 requestId 를 발급한 뒤(콜백 수용 조건),
+ * 커밋 후 기존 위임 로직({@link AnalysisAsyncRunner})을 재사용해 FastAPI 에 재검사를 요청한다.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class SavedLinkRecheckDispatcher {
+
+    private final AnalysisRepository analysisRepository;
+    private final AnalysisAsyncRunner asyncRunner;
+
+    @Transactional
+    public void dispatch(UUID analysisId) {
+        Analysis analysis = analysisRepository.findById(analysisId).orElse(null);
+        if (analysis == null) {
+            log.warn("재검사 대상 분석 없음 - analysisId={}", analysisId);
+            return;
+        }
+
+        UUID requestId = UUID.randomUUID();
+        String url = analysis.getOriginalUrl();
+        analysis.markForRecheck(requestId, Instant.now());
+
+        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+            @Override
+            public void afterCommit() {
+                asyncRunner.run(analysisId, url, requestId);
+            }
+        });
+    }
+}

src/main/java/com/linclean/domain/link/service/SavedLinkRecheckScheduler.java

@@ -0,0 +1,28 @@
+package com.linclean.domain.link.service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 저장 링크 주간 재검사 스케줄러.
+ * 기본 매주 월요일 04:00 실행 (cron 은 프로퍼티로 조정 가능).
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class SavedLinkRecheckScheduler {
+
+    private final SavedLinkRecheckService recheckService;
+
+    @Scheduled(cron = "${link.recheck.cron:0 0 4 * * MON}")
+    public void recheckStaleSavedLinks() {
+        log.info("저장 링크 재검사 스케줄러 시작");
+        try {
+            recheckService.runRecheck();
+        } catch (Exception e) {
+            log.error("저장 링크 재검사 스케줄러 실패", e);
+        }
+    }
+}

src/main/java/com/linclean/domain/link/service/SavedLinkRecheckService.java

@@ -0,0 +1,73 @@
+package com.linclean.domain.link.service;
+
+import com.linclean.domain.analysis.entity.Analysis;
+import com.linclean.domain.analysis.repository.AnalysisRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Slice;
+import org.springframework.stereotype.Service;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * 저장 링크 주간 재검사 오케스트레이션.
+ * 마지막 검사 후 N일 지난 저장 링크(참조 Analysis)를 keyset 페이지로 순회하며
+ * 건별로 {@link SavedLinkRecheckDispatcher} 에 재검사를 위임한다.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class SavedLinkRecheckService {
+
+    private static final int BATCH_SIZE = 100;
+    /** UUID 오름차순 순회 시작 커서 (모든 값보다 작음). */
+    private static final UUID START_CURSOR = new UUID(0L, 0L);
+
+    private final AnalysisRepository analysisRepository;
+    private final SavedLinkRecheckDispatcher dispatcher;
+
+    @Value("${link.recheck.stale-days:7}")
+    private int staleDays;
+
+    public void runRecheck() {
+        Instant threshold = Instant.now().minus(staleDays, ChronoUnit.DAYS);
+        UUID cursor = START_CURSOR;
+        int dispatched = 0;
+        int failed = 0;
+
+        while (true) {
+            Slice<Analysis> slice = analysisRepository.findStaleSavedAnalyses(
+                    threshold, cursor, PageRequest.of(0, BATCH_SIZE));
+            List<Analysis> batch = slice.getContent();
+            if (batch.isEmpty()) {
+                break;
+            }
+
+            // TODO(follow-up): 현재 위임은 AnalysisAsyncRunner 의 analysisTaskExecutor(실시간 분석과 공유)에
+            //  백프레셔를 위임한다. stale 규모가 커지면 재검사 전용 bounded executor 분리 + 레이트 리밋 필요.
+            for (Analysis analysis : batch) {
+                UUID analysisId = analysis.getAnalysisId();
+                cursor = analysisId;
+                try {
+                    dispatcher.dispatch(analysisId);
+                    dispatched++;
+                } catch (Exception e) {
+                    failed++;
+                    log.error("재검사 위임 실패 - analysisId={}", analysisId, e);
+                }
+            }
+
+            if (!slice.hasNext()) {
+                break;
+            }
+        }
+
+        log.info("저장 링크 재검사 완료 - staleDays={}, dispatched={}, failed={}",
+                staleDays, dispatched, failed);
+    }
+}

src/main/resources/application.yml

@@ -64,6 +64,11 @@ member:
     retention-days: ${MEMBER_WITHDRAWAL_RETENTION_DAYS:30}
     hard-delete-cron: ${MEMBER_WITHDRAWAL_HARD_DELETE_CRON:0 0 3 * * *}
 
+link:
+  recheck:
+    cron: ${LINK_RECHECK_CRON:0 0 4 * * MON}
+    stale-days: ${LINK_RECHECK_STALE_DAYS:7}
+
 springdoc:
   swagger-ui:
     path: /swagger-ui.html

src/test/java/com/linclean/domain/analysis/repository/AnalysisRepositoryTest.java

@@ -0,0 +1,158 @@
+package com.linclean.domain.analysis.repository;
+
+import com.linclean.domain.analysis.entity.Analysis;
+import jakarta.persistence.EntityManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Slice;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.springframework.transaction.annotation.Transactional;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@Testcontainers
+@ActiveProfiles("test")
+class AnalysisRepositoryTest {
+
+    private static final UUID START_CURSOR = new UUID(0L, 0L);
+    private static final PageRequest PAGE = PageRequest.of(0, 100);
+
+    @Container
+    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:17")
+            .withDatabaseName("linclean_test")
+            .withUsername("test")
+            .withPassword("test");
+
+    @Container
+    static GenericContainer<?> redis = new GenericContainer<>("redis:7-alpine")
+            .withExposedPorts(6379);
+
+    @DynamicPropertySource
+    static void overrideProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.datasource.url", postgres::getJdbcUrl);
+        registry.add("spring.datasource.username", postgres::getUsername);
+        registry.add("spring.datasource.password", postgres::getPassword);
+        registry.add("spring.data.redis.host", redis::getHost);
+        registry.add("spring.data.redis.port", () -> redis.getMappedPort(6379));
+    }
+
+    @Autowired
+    AnalysisRepository analysisRepository;
+
+    @Autowired
+    EntityManager em;
+
+    private long memberId;
+
+    @BeforeEach
+    void insertMember() {
+        em.createNativeQuery("""
+                INSERT INTO member (public_id, clerk_id, created_at, updated_at, deleted_at)
+                VALUES (gen_random_uuid(), 'recheck-repo-test', now(), now(), NULL)
+                """).executeUpdate();
+        memberId = ((Number) em.createNativeQuery(
+                "SELECT id FROM member WHERE clerk_id = 'recheck-repo-test'"
+        ).getSingleResult()).longValue();
+    }
+
+    private void insertAnalysis(UUID id, String status, String analyzedAtExpr, String lastCheckedExpr) {
+        em.createNativeQuery("""
+                INSERT INTO analysis
+                    (analysis_id, version, member_id, original_url, status, verdict, score, summary,
+                     analyzed_at, last_checked_at, created_at, updated_at)
+                VALUES ('%s'::uuid, 0, %d, 'https://%s.com', '%s', 'safe', 90, 'ok', %s, %s, now(), now())
+                """.formatted(id, memberId, id, status, analyzedAtExpr, lastCheckedExpr))
+                .executeUpdate();
+    }
+
+    private void insertSavedLink(UUID analysisId, String title) {
+        em.createNativeQuery("""
+                INSERT INTO saved_link (member_id, analysis_id, title, is_bookmarked, created_at, updated_at)
+                VALUES (%d, '%s'::uuid, '%s', false, now(), now())
+                """.formatted(memberId, analysisId, title))
+                .executeUpdate();
+    }
+
+    @Nested
+    class findStaleSavedAnalyses {
+
+        @Test
+        @Transactional
+        void 저장링크가_참조하고_마지막_검사후_N일_지난_SUCCEEDED_분석만_조회한다() {
+            UUID staleByAnalyzedAt = UUID.randomUUID();  // last_checked_at 없음 → analyzed_at 폴백으로 stale
+            UUID staleByLastChecked = UUID.randomUUID(); // last_checked_at 10일 전 → stale
+            UUID staleFailed = UUID.randomUUID();        // FAILED(지난 재검사 실패) → 재시도 대상으로 포함
+            UUID fresh = UUID.randomUUID();              // last_checked_at 1일 전 → 최신
+            UUID notSaved = UUID.randomUUID();           // 저장 링크 없음 → 제외
+            UUID queued = UUID.randomUUID();             // QUEUED → 제외
+
+            insertAnalysis(staleByAnalyzedAt, "succeeded", "now() - INTERVAL '10 days'", "NULL");
+            insertAnalysis(staleByLastChecked, "succeeded", "now() - INTERVAL '10 days'", "now() - INTERVAL '10 days'");
+            insertAnalysis(staleFailed, "failed", "now() - INTERVAL '10 days'", "NULL");
+            insertAnalysis(fresh, "succeeded", "now() - INTERVAL '10 days'", "now() - INTERVAL '1 day'");
+            insertAnalysis(notSaved, "succeeded", "now() - INTERVAL '10 days'", "NULL");
+            insertAnalysis(queued, "queued", "now() - INTERVAL '10 days'", "NULL");
+
+            insertSavedLink(staleByAnalyzedAt, "link-1");
+            insertSavedLink(staleByLastChecked, "link-2");
+            insertSavedLink(staleFailed, "link-3");
+            insertSavedLink(fresh, "link-4");
+            insertSavedLink(queued, "link-5");
+            em.flush();
+
+            Instant threshold = Instant.now().minus(7, ChronoUnit.DAYS);
+            Slice<Analysis> result = analysisRepository.findStaleSavedAnalyses(threshold, START_CURSOR, PAGE);
+
+            assertThat(result.getContent())
+                    .extracting(Analysis::getAnalysisId)
+                    .containsExactlyInAnyOrder(staleByAnalyzedAt, staleByLastChecked, staleFailed);
+        }
+
+        @Test
+        @Transactional
+        void 커서보다_큰_analysisId만_조회하여_keyset_페이지네이션을_지원한다() {
+            UUID a = UUID.randomUUID();
+            UUID b = UUID.randomUUID();
+            insertAnalysis(a, "succeeded", "now() - INTERVAL '10 days'", "NULL");
+            insertAnalysis(b, "succeeded", "now() - INTERVAL '10 days'", "NULL");
+            insertSavedLink(a, "cursor-link-a");
+            insertSavedLink(b, "cursor-link-b");
+            em.flush();
+
+            Instant threshold = Instant.now().minus(7, ChronoUnit.DAYS);
+
+            // DB(Postgres) 정렬 기준으로 결과 순서를 그대로 사용해 커서 의미를 검증한다.
+            List<Analysis> all = analysisRepository
+                    .findStaleSavedAnalyses(threshold, START_CURSOR, PAGE).getContent();
+            assertThat(all).hasSize(2);
+
+            UUID first = all.get(0).getAnalysisId();
+            UUID second = all.get(1).getAnalysisId();
+
+            // 첫 항목 커서 이후 → 두 번째만
+            assertThat(analysisRepository.findStaleSavedAnalyses(threshold, first, PAGE).getContent())
+                    .extracting(Analysis::getAnalysisId)
+                    .containsExactly(second);
+
+            // 마지막 항목 커서 이후 → 없음
+            assertThat(analysisRepository.findStaleSavedAnalyses(threshold, second, PAGE).getContent())
+                    .isEmpty();
+        }
+    }
+}