Add reliable detection job dispatch

backend-java/src/main/java/com/fengting/aigcforensics/config/JobOutboxProperties.java

@@ -0,0 +1,64 @@
+package com.fengting.aigcforensics.config;
+
+import java.time.Duration;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "app.jobs.outbox")
+public class JobOutboxProperties {
+
+    private boolean dispatcherEnabled = true;
+    private long pollDelayMs = 500;
+    private Duration staleAfter = Duration.ofMinutes(1);
+    private int maxAttempts = 5;
+    private Duration baseRetryDelay = Duration.ofSeconds(1);
+    private Duration maxRetryDelay = Duration.ofMinutes(1);
+
+    public boolean isDispatcherEnabled() {
+        return dispatcherEnabled;
+    }
+
+    public void setDispatcherEnabled(boolean dispatcherEnabled) {
+        this.dispatcherEnabled = dispatcherEnabled;
+    }
+
+    public long getPollDelayMs() {
+        return pollDelayMs;
+    }
+
+    public void setPollDelayMs(long pollDelayMs) {
+        this.pollDelayMs = pollDelayMs;
+    }
+
+    public Duration getStaleAfter() {
+        return staleAfter;
+    }
+
+    public void setStaleAfter(Duration staleAfter) {
+        this.staleAfter = staleAfter;
+    }
+
+    public int getMaxAttempts() {
+        return maxAttempts;
+    }
+
+    public void setMaxAttempts(int maxAttempts) {
+        this.maxAttempts = maxAttempts;
+    }
+
+    public Duration getBaseRetryDelay() {
+        return baseRetryDelay;
+    }
+
+    public void setBaseRetryDelay(Duration baseRetryDelay) {
+        this.baseRetryDelay = baseRetryDelay;
+    }
+
+    public Duration getMaxRetryDelay() {
+        return maxRetryDelay;
+    }
+
+    public void setMaxRetryDelay(Duration maxRetryDelay) {
+        this.maxRetryDelay = maxRetryDelay;
+    }
+}

backend-java/src/main/java/com/fengting/aigcforensics/controller/ApiExceptionHandler.java

@@ -9,6 +9,7 @@
 import org.springframework.web.multipart.support.MissingServletRequestPartException;
 
 import com.fengting.aigcforensics.dto.error.ErrorResponse;
+import com.fengting.aigcforensics.domain.InvalidJobOutboxStateException;
 import com.fengting.aigcforensics.service.ResourceNotFoundException;
 
 @RestControllerAdvice
@@ -20,6 +21,12 @@ public ErrorResponse handleBadRequest(IllegalArgumentException exception) {
         return new ErrorResponse(exception.getMessage(), Instant.now());
     }
 
+    @ExceptionHandler(InvalidJobOutboxStateException.class)
+    @ResponseStatus(HttpStatus.CONFLICT)
+    public ErrorResponse handleConflict(InvalidJobOutboxStateException exception) {
+        return new ErrorResponse(exception.getMessage(), Instant.now());
+    }
+
     @ExceptionHandler(MissingServletRequestPartException.class)
     @ResponseStatus(HttpStatus.BAD_REQUEST)
     public ErrorResponse handleMissingPart(MissingServletRequestPartException exception) {

backend-java/src/main/java/com/fengting/aigcforensics/controller/JobOutboxOperationsController.java

@@ -0,0 +1,65 @@
+package com.fengting.aigcforensics.controller;
+
+import java.util.List;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.fengting.aigcforensics.domain.JobOutboxEvent;
+import com.fengting.aigcforensics.domain.JobOutboxStatus;
+import com.fengting.aigcforensics.dto.operations.JobOutboxEventResponse;
+import com.fengting.aigcforensics.service.JobOutboxService;
+
+@RestController
+@RequestMapping("/api/operations/job-outbox")
+public class JobOutboxOperationsController {
+
+    private static final int DEFAULT_LIMIT = 50;
+    private static final int MAX_LIMIT = 100;
+
+    private final JobOutboxService outboxService;
+
+    public JobOutboxOperationsController(JobOutboxService outboxService) {
+        this.outboxService = outboxService;
+    }
+
+    @GetMapping
+    public List<JobOutboxEventResponse> list(
+            @RequestParam(defaultValue = "FAILED") JobOutboxStatus status,
+            @RequestParam(defaultValue = "50") int limit) {
+        validateLimit(limit);
+        return outboxService.list(status, limit).stream()
+                .map(this::toResponse)
+                .toList();
+    }
+
+    @PostMapping("/{eventId}/replay")
+    public JobOutboxEventResponse replay(@PathVariable String eventId) {
+        return toResponse(outboxService.replay(eventId));
+    }
+
+    private void validateLimit(int limit) {
+        if (limit < 1 || limit > MAX_LIMIT) {
+            throw new IllegalArgumentException("limit must be between 1 and 100");
+        }
+    }
+
+    private JobOutboxEventResponse toResponse(JobOutboxEvent event) {
+        return new JobOutboxEventResponse(
+                event.getEventId(),
+                event.getEventType(),
+                event.getAggregateType(),
+                event.getAggregateId(),
+                event.getStatus(),
+                event.getAttemptCount(),
+                event.getAvailableAt(),
+                event.getLastError(),
+                event.getCreatedAt(),
+                event.getUpdatedAt(),
+                event.getPublishedAt());
+    }
+}

backend-java/src/main/java/com/fengting/aigcforensics/domain/InvalidJobOutboxStateException.java

@@ -0,0 +1,8 @@
+package com.fengting.aigcforensics.domain;
+
+public class InvalidJobOutboxStateException extends IllegalStateException {
+
+    public InvalidJobOutboxStateException(String message) {
+        super(message);
+    }
+}

backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEvent.java

@@ -0,0 +1,234 @@
+package com.fengting.aigcforensics.domain;
+
+import java.time.Instant;
+
+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.Table;
+import jakarta.persistence.Version;
+
+@Entity
+@Table(name = "job_outbox")
+public class JobOutboxEvent {
+
+    private static final int MAX_ERROR_LENGTH = 2048;
+    private static final String DETECTION_AGGREGATE_TYPE = "DETECTION_TASK";
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(name = "event_id", nullable = false, unique = true, length = 64)
+    private String eventId;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "event_type", nullable = false, length = 64)
+    private JobOutboxEventType eventType;
+
+    @Column(name = "aggregate_type", nullable = false, length = 64)
+    private String aggregateType;
+
+    @Column(name = "aggregate_id", nullable = false, length = 64)
+    private String aggregateId;
+
+    @Column(name = "payload_json", nullable = false, columnDefinition = "text")
+    private String payloadJson;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false, length = 32)
+    private JobOutboxStatus status;
+
+    @Column(name = "attempt_count", nullable = false)
+    private int attemptCount;
+
+    @Column(name = "available_at")
+    private Instant availableAt;
+
+    @Column(name = "last_error", length = MAX_ERROR_LENGTH)
+    private String lastError;
+
+    @Column(name = "created_at", nullable = false)
+    private Instant createdAt;
+
+    @Column(name = "updated_at", nullable = false)
+    private Instant updatedAt;
+
+    @Column(name = "published_at")
+    private Instant publishedAt;
+
+    @Version
+    @Column(name = "lock_version", nullable = false)
+    private long lockVersion;
+
+    protected JobOutboxEvent() {
+    }
+
+    private JobOutboxEvent(
+            String eventId,
+            JobOutboxEventType eventType,
+            String aggregateType,
+            String aggregateId,
+            String payloadJson,
+            Instant now) {
+        this.eventId = requireText(eventId, "eventId");
+        this.eventType = eventType;
+        this.aggregateType = requireText(aggregateType, "aggregateType");
+        this.aggregateId = requireText(aggregateId, "aggregateId");
+        this.payloadJson = requireText(payloadJson, "payloadJson");
+        this.status = JobOutboxStatus.PENDING;
+        this.availableAt = now;
+        this.createdAt = now;
+        this.updatedAt = now;
+    }
+
+    public static JobOutboxEvent detectionRequested(
+            String eventId,
+            String taskId,
+            String payloadJson,
+            Instant now) {
+        return new JobOutboxEvent(
+                eventId,
+                JobOutboxEventType.DETECTION_REQUESTED,
+                DETECTION_AGGREGATE_TYPE,
+                taskId,
+                payloadJson,
+                now);
+    }
+
+    public void claim(Instant claimedAt) {
+        requireStatus(JobOutboxStatus.PENDING, "claim");
+        if (availableAt == null || availableAt.isAfter(claimedAt)) {
+            throw new InvalidJobOutboxStateException(
+                    "Outbox event is not available for publishing: " + eventId);
+        }
+        status = JobOutboxStatus.PUBLISHING;
+        attemptCount++;
+        updatedAt = claimedAt;
+    }
+
+    public void markPublished(Instant completedAt) {
+        requireStatus(JobOutboxStatus.PUBLISHING, "mark published");
+        status = JobOutboxStatus.PUBLISHED;
+        publishedAt = completedAt;
+        availableAt = null;
+        lastError = null;
+        updatedAt = completedAt;
+    }
+
+    public void markPublishFailed(String error, Instant retryAt, int maxAttempts, Instant failedAt) {
+        requireStatus(JobOutboxStatus.PUBLISHING, "mark publish failed");
+        if (maxAttempts < 1) {
+            throw new IllegalArgumentException("maxAttempts must be at least 1");
+        }
+        lastError = truncateError(error);
+        updatedAt = failedAt;
+        if (attemptCount >= maxAttempts) {
+            status = JobOutboxStatus.FAILED;
+            availableAt = null;
+            return;
+        }
+        status = JobOutboxStatus.PENDING;
+        availableAt = retryAt;
+    }
+
+    public void recoverStaleClaim(Instant recoveredAt) {
+        requireStatus(JobOutboxStatus.PUBLISHING, "recover stale claim");
+        status = JobOutboxStatus.PENDING;
+        availableAt = recoveredAt;
+        lastError = "Recovered stale publishing claim";
+        updatedAt = recoveredAt;
+    }
+
+    public void replay(Instant replayedAt) {
+        if (status != JobOutboxStatus.PUBLISHED && status != JobOutboxStatus.FAILED) {
+            throw new InvalidJobOutboxStateException(
+                    "Only terminal outbox events can be replayed: " + eventId);
+        }
+        status = JobOutboxStatus.PENDING;
+        attemptCount = 0;
+        availableAt = replayedAt;
+        lastError = null;
+        publishedAt = null;
+        updatedAt = replayedAt;
+    }
+
+    private void requireStatus(JobOutboxStatus expected, String action) {
+        if (status != expected) {
+            throw new InvalidJobOutboxStateException(
+                    "Cannot " + action + " outbox event " + eventId + " while status is " + status);
+        }
+    }
+
+    private static String requireText(String value, String fieldName) {
+        if (value == null || value.isBlank()) {
+            throw new IllegalArgumentException(fieldName + " must not be blank");
+        }
+        return value;
+    }
+
+    private String truncateError(String error) {
+        String message = error == null || error.isBlank() ? "Unknown publish failure" : error;
+        return message.length() <= MAX_ERROR_LENGTH ? message : message.substring(0, MAX_ERROR_LENGTH);
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public String getEventId() {
+        return eventId;
+    }
+
+    public JobOutboxEventType getEventType() {
+        return eventType;
+    }
+
+    public String getAggregateType() {
+        return aggregateType;
+    }
+
+    public String getAggregateId() {
+        return aggregateId;
+    }
+
+    public String getPayloadJson() {
+        return payloadJson;
+    }
+
+    public JobOutboxStatus getStatus() {
+        return status;
+    }
+
+    public int getAttemptCount() {
+        return attemptCount;
+    }
+
+    public Instant getAvailableAt() {
+        return availableAt;
+    }
+
+    public String getLastError() {
+        return lastError;
+    }
+
+    public Instant getCreatedAt() {
+        return createdAt;
+    }
+
+    public Instant getUpdatedAt() {
+        return updatedAt;
+    }
+
+    public Instant getPublishedAt() {
+        return publishedAt;
+    }
+
+    public long getLockVersion() {
+        return lockVersion;
+    }
+}