label_backend/src/test/java/com/label/integration/VideoCallbackIdempotencyTest.java

package com.label.integration;

import com.label.AbstractIntegrationTest;
import com.label.service.RedisService;
import com.label.util.RedisKeyManager;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.http.*;

import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * 视频处理回调幂等与重试集成测试（US8）。
 *
 * 测试场景：
 * 1. 同一 jobId 收到两次 SUCCESS 回调：annotation_task（EXTRACTION）仅创建一次
 * 2. 超出最大重试次数 → job.status = FAILED，source_data.status = PENDING
 */
public class VideoCallbackIdempotencyTest extends AbstractIntegrationTest {

    private static final String ADMIN_TOKEN = "test-admin-token-video";

    @Autowired
    private TestRestTemplate restTemplate;

    @Autowired
    private RedisService redisService;

    private Long companyId;
    private Long adminUserId;
    private Long sourceId;
    private Long jobId;

    @BeforeEach
    void setupTokenAndData() {
        companyId = jdbcTemplate.queryForObject(
                "SELECT id FROM sys_company WHERE company_code = 'DEMO'", Long.class);
        adminUserId = jdbcTemplate.queryForObject(
                "SELECT id FROM sys_user WHERE username = 'admin'", Long.class);

        // 伪造 Redis Token
        redisService.hSetAll(RedisKeyManager.tokenKey(ADMIN_TOKEN),
                Map.of("userId", adminUserId.toString(), "role", "ADMIN",
                        "companyId", companyId.toString(), "username", "admin"),
                3600L);

        // 插入 source_data（PREPROCESSING 状态，模拟视频处理中）
        jdbcTemplate.execute(
                "INSERT INTO source_data (company_id, uploader_id, data_type, file_path, " +
                "file_name, file_size, bucket_name, status) " +
                "VALUES (" + companyId + ", " + adminUserId + ", 'VIDEO', " +
                "'videos/test.mp4', 'test.mp4', 10240, 'label-source-data', 'PREPROCESSING')");
        sourceId = jdbcTemplate.queryForObject(
                "SELECT id FROM source_data ORDER BY id DESC LIMIT 1", Long.class);

        // 插入 PENDING 视频处理任务
        jdbcTemplate.execute(
                "INSERT INTO video_process_job (company_id, source_id, job_type, status, " +
                "params, retry_count, max_retries) " +
                "VALUES (" + companyId + ", " + sourceId + ", 'FRAME_EXTRACT', 'PENDING', " +
                "'{}'::jsonb, 0, 3)");
        jobId = jdbcTemplate.queryForObject(
                "SELECT id FROM video_process_job ORDER BY id DESC LIMIT 1", Long.class);
    }

    @AfterEach
    void cleanupTokens() {
        redisService.delete(RedisKeyManager.tokenKey(ADMIN_TOKEN));
    }

    // ------------------------------------------------------------------ 测试 1: 幂等性 --

    @Test
    @DisplayName("同一 jobId 发送两次 SUCCESS 回调：source_data 仅更新一次，status=PENDING")
    void successCallback_idempotent_sourceUpdatedOnce() {
        // 第一次 SUCCESS 回调
        ResponseEntity<Map> resp1 = sendCallback(jobId, "SUCCESS", "processed/frames.zip", null);
        assertThat(resp1.getStatusCode()).isEqualTo(HttpStatus.OK);

        // 验证第一次回调后状态
        String jobStatus1 = jdbcTemplate.queryForObject(
                "SELECT status FROM video_process_job WHERE id = ?", String.class, jobId);
        assertThat(jobStatus1).isEqualTo("SUCCESS");

        String sourceStatus1 = jdbcTemplate.queryForObject(
                "SELECT status FROM source_data WHERE id = ?", String.class, sourceId);
        assertThat(sourceStatus1).isEqualTo("PENDING");

        // 第二次 SUCCESS 回调（幂等：应直接返回，不重复处理）
        ResponseEntity<Map> resp2 = sendCallback(jobId, "SUCCESS", "processed/frames.zip", null);
        assertThat(resp2.getStatusCode()).isEqualTo(HttpStatus.OK);

        // 状态仍为 SUCCESS + PENDING，未被改变
        String jobStatus2 = jdbcTemplate.queryForObject(
                "SELECT status FROM video_process_job WHERE id = ?", String.class, jobId);
        assertThat(jobStatus2).as("幂等：第二次回调不应改变 job 状态").isEqualTo("SUCCESS");

        String sourceStatus2 = jdbcTemplate.queryForObject(
                "SELECT status FROM source_data WHERE id = ?", String.class, sourceId);
        assertThat(sourceStatus2).as("幂等：第二次回调不应改变 source_data 状态").isEqualTo("PENDING");
    }

    // ------------------------------------------------------------------ 测试 2: 超出重试上限 → FAILED --

    @Test
    @DisplayName("超出最大重试次数后 → job.status=FAILED，source_data.status=PENDING")
    void failedCallback_exceedsMaxRetries_jobBecomesFailedAndSourceReverts() {
        // 将 retry_count 设为 max_retries-1（再失败一次就超限）
        jdbcTemplate.execute(
                "UPDATE video_process_job SET retry_count = 2, max_retries = 3, " +
                "status = 'RETRYING' WHERE id = " + jobId);

        // 发送最后一次 FAILED 回调（retry_count 变为 3 = max_retries → 超限）
        ResponseEntity<Map> resp = sendCallback(jobId, "FAILED", null, "ffmpeg 处理超时");
        assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);

        // 验证 job → FAILED
        Map<String, Object> jobRow = jdbcTemplate.queryForMap(
                "SELECT status, retry_count, error_message FROM video_process_job WHERE id = ?", jobId);
        assertThat(jobRow.get("status")).as("超出重试上限后 job 应为 FAILED").isEqualTo("FAILED");
        assertThat(((Number) jobRow.get("retry_count")).intValue()).isEqualTo(3);
        assertThat(jobRow.get("error_message")).isEqualTo("ffmpeg 处理超时");

        // 验证 source_data → PENDING（管理员可重新处理）
        String sourceStatus = jdbcTemplate.queryForObject(
                "SELECT status FROM source_data WHERE id = ?", String.class, sourceId);
        assertThat(sourceStatus).as("超出重试上限后 source_data 应回退为 PENDING").isEqualTo("PENDING");
    }

    // ------------------------------------------------------------------ 测试 3: 管理员重置 --

    @Test
    @DisplayName("管理员重置 FAILED 任务 → job.status=PENDING，retryCount=0")
    void resetFailedJob_succeeds() {
        // 先将任务置为 FAILED 状态
        jdbcTemplate.execute(
                "UPDATE video_process_job SET status = 'FAILED', retry_count = 3 WHERE id = " + jobId);

        // 重置
        HttpHeaders headers = new HttpHeaders();
        headers.set("Authorization", "Bearer " + ADMIN_TOKEN);
        ResponseEntity<Map> resp = restTemplate.exchange(
                baseUrl("/api/video/jobs/" + jobId + "/reset"),
                HttpMethod.POST,
                new HttpEntity<>(headers),
                Map.class);
        assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);

        // 验证
        Map<String, Object> jobRow = jdbcTemplate.queryForMap(
                "SELECT status, retry_count FROM video_process_job WHERE id = ?", jobId);
        assertThat(jobRow.get("status")).isEqualTo("PENDING");
        assertThat(((Number) jobRow.get("retry_count")).intValue()).isEqualTo(0);
    }

    // ------------------------------------------------------------------ 工具方法 --

    private ResponseEntity<Map> sendCallback(Long jobId, String status,
                                              String outputPath, String errorMessage) {
        Map<String, Object> body;
        if ("SUCCESS".equals(status)) {
            body = Map.of("jobId", jobId, "status", status, "outputPath", outputPath);
        } else {
            body = Map.of("jobId", jobId, "status", status, "errorMessage",
                    errorMessage != null ? errorMessage : "");
        }

        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        return restTemplate.exchange(
                baseUrl("/api/video/callback"),
                HttpMethod.POST,
                new HttpEntity<>(body, headers),
                Map.class);
    }
}