fix(education): integrate EDU-011 bounded capability

docs/education/migration/issues/EDU-011-import-export-assets.md

@@ -7,7 +7,7 @@
 
 ## Delivered tenant-admin outcome
 
-Education owns tenant-scoped import asset metadata and durable import jobs while reusing only public Infra APIs for private-file operations and other platform primitives. The delivered job lifecycle has exactly five states: `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, and `FAILED`.
+Education owns tenant-scoped import asset metadata and durable import jobs while reusing only public Infra APIs for private-file operations and other platform primitives. The delivered job lifecycle has exactly five states: `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, and `FAILED`.
 
 The bounded capability provides:
 
@@ -34,7 +34,7 @@ The bounded capability provides:
 ## Acceptance record
 
 - [x] Education-owned import assets and durable jobs are represented by V4130.
-- [x] Jobs use the five-state lifecycle `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, `FAILED`.
+- [x] Jobs use the five-state lifecycle `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, `FAILED`.
 - [x] Claim, lease, heartbeat, expired-lease recovery, retry bounds, and duplicate safety are defined.
 - [x] Scanning fails closed, with default `UNAVAILABLE`.
 - [x] CSV/XLSX can return metadata-only preview when the parser is unavailable.

yudao-module-education/CONTEXT.md

@@ -109,11 +109,11 @@ Tenant-owned metadata that identifies one private Infra-managed source object fo
 _Avoid_: Uploaded file owned by Education, public asset URL
 
 **Content Import Job**:
-A tenant-owned durable Education job that previews and, only when safe and executable, imports content from one Content Import Asset. Its five states are `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, and `FAILED`.
+A tenant-owned durable Education job that previews and, only when safe and executable, imports content from one Content Import Asset. Its five states are `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, and `FAILED`.
 _Avoid_: Generic Infra job, export job
 
 **Import Job Lease**:
-A fenced, expiring claim on a `PROCESSING` Content Import Job, identified by worker and lease token and kept alive by heartbeat. Expiry permits bounded recovery; it is not proof that the prior worker stopped.
+A fenced, expiring claim on a pending Content Import Job, identified by worker and lease token and kept alive by heartbeat. The aggregate remains `PREVIEW_PENDING` or `EXECUTE_PENDING` while claimed; expiry permits bounded recovery and is not proof that the prior worker stopped.
 _Avoid_: Redis lock as job ownership, permanent worker assignment
 
 **Executable Import Preview**:

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java

@@ -40,7 +40,8 @@ public class EducationCapabilityController {
                 .capabilities(List.of("shell", "catalog", "questions", "practice-preview",
                         "answer-save", "session-submit", "practice-report", "vocabulary-review",
                         "exam-reminder", "learning-award", "student-feedback", "learning-summary",
-                        "tenant-leaderboard"))
+                        "tenant-leaderboard", "content-import-assets", "content-import-jobs",
+                        "content-export-policy"))
                 .themes(describeThemes())
                 .catalogReadEnabled(educationProperties.isCatalogReadEnabled())
                 .practiceWriteEnabled(educationProperties.isPracticeWriteEnabled())
@@ -53,10 +54,10 @@ public class EducationCapabilityController {
         List<String> catalogDependencies = educationProperties.getCatalogMode() == CatalogProviderMode.SCALAR_READ
                 ? List.of("scalar-catalog") : List.of();
         return List.of(
-                theme("EDU-011", "content-import-export-assets", "BLOCKED",
+                theme("EDU-011", "content-import-export-assets", "PARTIAL",
                         List.of("education", "infra"),
-                        List.of("asset-admission-contract", "scanner-ownership", "durable-job-lease"),
-                        List.of(), "FEATURE_FLAG", "CONTRACT_ONLY"),
+                        List.of("production-scanner-adapter", "production-import-parser", "export-artifact-generation"),
+                        List.of(), "FEATURE_FLAG", "POSTGRESQL_INTEGRATION_TESTED"),
                 theme("EDU-012", "classes-relationships", "BLOCKED",
                         List.of("education", "member", "system"),
                         List.of("relationship-role-model", "class-data-scope", "invitation-contract"),

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/EducationImportAssetDO.java

@@ -2,23 +2,29 @@ package cn.iocoder.yudao.module.education.dal.dataobject;
 
 import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
 import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
-@TableName("education_import_asset")
-@KeySequence("education_import_asset_seq")
+@TableName("education_content_import_asset")
+@KeySequence("education_content_import_asset_seq")
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class EducationImportAssetDO extends TenantBaseDO {
     @TableId
     private Long id;
     private Long ownerUserId;
+    @TableField("file_name")
     private String originalName;
+    @TableField("mime_type")
     private String contentType;
+    @TableField("file_size_bytes")
     private Long size;
+    @TableField("checksum_sha256")
     private String sha256;
+    @TableField("object_key")
     private String fileReference;
     private String scanStatus;
 }

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportAssetDO.java

@@ -14,9 +14,11 @@ import lombok.EqualsAndHashCode;
 public class ContentImportAssetDO extends TenantBaseDO {
     @TableId
     private Long id;
+    private Long ownerUserId;
     private String objectKey;
     private String fileName;
     private String mimeType;
     private Long fileSizeBytes;
     private String checksumSha256;
+    private String scanStatus;
 }

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportJobDO.java

@@ -15,12 +15,17 @@ import java.time.LocalDateTime;
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class ContentImportJobDO extends TenantBaseDO {
-    @TableId
-    private Long id;
+    @TableId private Long id;
     private Long assetId;
-    private String previewKey;
+    private Long actorId;
+    @TableField("preview_key") private String previewKey;
+    private String previewRequestHash;
+    @TableField("execute_key") private String executeKey;
+    private String executeRequestHash;
     private String importType;
     private String status;
+    private String scanStatus;
+    private String parserStatus;
     private Integer attemptCount;
     private Integer maxAttempts;
     private LocalDateTime nextAttemptAt;
@@ -30,9 +35,11 @@ public class ContentImportJobDO extends TenantBaseDO {
     private LocalDateTime lastHeartbeatAt;
     private LocalDateTime startedAt;
     private LocalDateTime finishedAt;
-    private String errorMessage;
-    @TableField("preview_payload")
+    private String failureCode;
+    private String failureMessage;
     private String previewPayload;
-    @TableField("result_summary")
+    private String parsedPayload;
     private String resultSummary;
+    private Integer previewQuestionCount;
+    private Integer importedQuestionCount;
 }

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java

@@ -11,10 +11,12 @@ public interface ContentImportJobMapper extends BaseMapperX<ContentImportJobDO>
 
     @Insert("""
             INSERT INTO education_content_import_job
-              (tenant_id, asset_id, preview_key, import_type, status, attempt_count, max_attempts,
+              (tenant_id, asset_id, actor_id, preview_key, preview_request_hash, import_type, status,
+               scan_status, parser_status, attempt_count, max_attempts, next_attempt_at,
                preview_payload, creator, create_time, updater, update_time, deleted)
             VALUES
-              (#{tenantId}, #{assetId}, #{previewKey}, #{importType}, #{status}, 0, #{maxAttempts},
+              (#{tenantId}, #{assetId}, #{actorId}, #{previewKey}, #{previewRequestHash}, #{importType},
+               'PREVIEW_PENDING', 'PENDING', 'PENDING', 0, #{maxAttempts}, CURRENT_TIMESTAMP,
                CAST(#{previewPayload} AS JSONB), '', CURRENT_TIMESTAMP, '', CURRENT_TIMESTAMP, FALSE)
             ON CONFLICT (tenant_id, preview_key) DO NOTHING
             """)
@@ -23,71 +25,67 @@ public interface ContentImportJobMapper extends BaseMapperX<ContentImportJobDO>
 
     @Update("""
             UPDATE education_content_import_job
-               SET status = 'PENDING', next_attempt_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP
-             WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PREVIEW' AND deleted = FALSE
+               SET status='EXECUTE_PENDING', execute_key=#{executeKey}, execute_request_hash=#{executeRequestHash},
+                   next_attempt_at=CURRENT_TIMESTAMP, update_time=CURRENT_TIMESTAMP
+             WHERE tenant_id=#{tenantId} AND id=#{id} AND status='PREVIEW_READY'
+               AND scan_status='CLEAN' AND parser_status='PARSED' AND deleted=false
             """)
-    int executePreview(@Param("tenantId") Long tenantId, @Param("id") Long id);
+    int requestExecute(@Param("tenantId") Long tenantId, @Param("id") Long id,
+                       @Param("executeKey") String executeKey, @Param("executeRequestHash") String executeRequestHash);
 
     @Select("""
-            WITH candidates AS (
+            WITH candidate AS (
                 SELECT id FROM education_content_import_job
-                 WHERE deleted = FALSE AND attempt_count < max_attempts
-                   AND ((status = 'PENDING' AND (next_attempt_at IS NULL OR next_attempt_at <= CURRENT_TIMESTAMP))
-                     OR (status = 'PROCESSING' AND lease_expires_at <= CURRENT_TIMESTAMP))
-                 ORDER BY CASE WHEN status = 'PROCESSING' THEN 0 ELSE 1 END,
-                          COALESCE(lease_expires_at, next_attempt_at, create_time), id
+                 WHERE deleted=false AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING')
+                   AND attempt_count < max_attempts
+                   AND (next_attempt_at IS NULL OR next_attempt_at <= CURRENT_TIMESTAMP)
+                   AND (lease_expires_at IS NULL OR lease_expires_at <= CURRENT_TIMESTAMP)
+                 ORDER BY COALESCE(next_attempt_at, create_time), id
                  LIMIT #{limit} FOR UPDATE SKIP LOCKED
             )
             UPDATE education_content_import_job job
-               SET status = 'PROCESSING', locked_by = #{workerId}, lease_token = #{leaseToken},
-                   lease_expires_at = CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL),
-                   last_heartbeat_at = CURRENT_TIMESTAMP, attempt_count = job.attempt_count + 1,
-                   next_attempt_at = NULL, started_at = COALESCE(job.started_at, CURRENT_TIMESTAMP),
-                   finished_at = NULL, error_message = NULL, update_time = CURRENT_TIMESTAMP
-              FROM candidates WHERE job.id = candidates.id
+               SET locked_by=#{workerId}, lease_token=#{leaseToken},
+                   lease_expires_at=CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL),
+                   last_heartbeat_at=CURRENT_TIMESTAMP, attempt_count=job.attempt_count + 1,
+                   next_attempt_at=NULL, started_at=COALESCE(job.started_at,CURRENT_TIMESTAMP), update_time=CURRENT_TIMESTAMP
+              FROM candidate WHERE job.id=candidate.id
             RETURNING job.*
             """)
-    List<ContentImportJobDO> claimReady(@Param("workerId") String workerId,
-                                        @Param("leaseToken") String leaseToken,
-                                        @Param("leaseSeconds") long leaseSeconds,
-                                        @Param("limit") int limit);
+    List<ContentImportJobDO> claimReady(@Param("workerId") String workerId, @Param("leaseToken") String leaseToken,
+                                        @Param("leaseSeconds") long leaseSeconds, @Param("limit") int limit);
 
     @Update("""
             UPDATE education_content_import_job
-               SET lease_expires_at = CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL),
-                   last_heartbeat_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP
-             WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PROCESSING'
-               AND lease_token = #{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP
+               SET lease_expires_at=CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL),
+                   last_heartbeat_at=CURRENT_TIMESTAMP, update_time=CURRENT_TIMESTAMP
+             WHERE tenant_id=#{tenantId} AND id=#{id} AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING')
+               AND lease_token=#{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP
             """)
     int heartbeat(@Param("tenantId") Long tenantId, @Param("id") Long id,
                   @Param("leaseToken") String leaseToken, @Param("leaseSeconds") long leaseSeconds);
 
-    @Update("""
-            WITH expired AS (
-                SELECT id FROM education_content_import_job
-                 WHERE status = 'PROCESSING' AND lease_expires_at <= CURRENT_TIMESTAMP
-                   AND attempt_count < max_attempts
-                 ORDER BY lease_expires_at, id
-                 LIMIT #{limit} FOR UPDATE SKIP LOCKED
-            )
-            UPDATE education_content_import_job job
-               SET status = 'PENDING', locked_by = NULL, lease_token = NULL, lease_expires_at = NULL,
-                   last_heartbeat_at = NULL, next_attempt_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP
-              FROM expired WHERE job.id = expired.id
-            """)
-    int recoverExpired(@Param("limit") int limit);
-
     @Update("""
             UPDATE education_content_import_job
-               SET status = 'COMPLETED', result_summary = CAST(#{resultSummary} AS JSONB),
-                   locked_by = NULL, lease_token = NULL, lease_expires_at = NULL, last_heartbeat_at = NULL,
-                   finished_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP
-             WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PROCESSING'
-               AND lease_token = #{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP
+               SET status=#{nextStatus}, scan_status=COALESCE(#{scanStatus},scan_status),
+                   parser_status=COALESCE(#{parserStatus},parser_status),
+                   preview_payload=COALESCE(CAST(#{previewPayload} AS JSONB),preview_payload),
+                   parsed_payload=CAST(#{parsedPayload} AS JSONB), preview_question_count=#{previewQuestionCount},
+                   result_summary=COALESCE(CAST(#{resultSummary} AS JSONB),result_summary),
+                   imported_question_count=#{importedQuestionCount}, failure_code=#{failureCode},
+                   failure_message=#{failureMessage}, locked_by=NULL, lease_token=NULL, lease_expires_at=NULL,
+                   last_heartbeat_at=NULL, finished_at=CASE WHEN #{nextStatus} IN ('COMPLETED','FAILED') THEN CURRENT_TIMESTAMP ELSE finished_at END,
+                   update_time=CURRENT_TIMESTAMP
+             WHERE tenant_id=#{tenantId} AND id=#{id} AND status=#{expectedStatus}
+               AND lease_token=#{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP
             """)
-    int complete(@Param("tenantId") Long tenantId, @Param("id") Long id,
-                 @Param("leaseToken") String leaseToken, @Param("resultSummary") String resultSummary);
+    int finishClaim(@Param("tenantId") Long tenantId, @Param("id") Long id,
+                    @Param("expectedStatus") String expectedStatus, @Param("nextStatus") String nextStatus,
+                    @Param("leaseToken") String leaseToken, @Param("scanStatus") String scanStatus,
+                    @Param("parserStatus") String parserStatus, @Param("previewPayload") String previewPayload,
+                    @Param("parsedPayload") String parsedPayload, @Param("previewQuestionCount") Integer previewQuestionCount,
+                    @Param("resultSummary") String resultSummary, @Param("importedQuestionCount") Integer importedQuestionCount,
+                    @Param("failureCode") String failureCode, @Param("failureMessage") String failureMessage);
 
-    @Update("UPDATE education_content_import_job SET lease_expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = #{id}")
+    @Update("UPDATE education_content_import_job SET lease_expires_at=CURRENT_TIMESTAMP-INTERVAL '1 second' WHERE id=#{id}")
     int expireLeaseForTest(@Param("id") Long id);
 }

yudao-module-education/src/main/resources/db/migration/education/V4130__add_content_import_jobs.sql

@@ -1,35 +1,48 @@
--- EDU-011 persistence slice: tenant-owned import assets and durable preview/execute jobs.
+-- EDU-011: tenant-bound import assets and durable preview/execute jobs.
 
 CREATE TABLE education_content_import_asset (
     id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
     tenant_id BIGINT NOT NULL,
-    object_key VARCHAR(512) NOT NULL,
+    owner_user_id BIGINT NOT NULL,
+    object_key VARCHAR(1024) NOT NULL,
     file_name VARCHAR(255) NOT NULL,
     mime_type VARCHAR(128) NOT NULL,
     file_size_bytes BIGINT NOT NULL,
     checksum_sha256 VARCHAR(64) NOT NULL,
-    creator VARCHAR(64) DEFAULT '',
-    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    updater VARCHAR(64) DEFAULT '',
-    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    scan_status VARCHAR(16) NOT NULL DEFAULT 'UNAVAILABLE',
+    creator VARCHAR(64) DEFAULT '', create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) DEFAULT '', update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     deleted BOOLEAN NOT NULL DEFAULT false,
-    CONSTRAINT ck_education_content_import_asset_size CHECK (file_size_bytes >= 0),
+    CONSTRAINT ck_education_content_import_asset_size CHECK (file_size_bytes > 0 AND file_size_bytes <= 10485760),
+    CONSTRAINT ck_education_content_import_asset_scan CHECK (scan_status IN ('PENDING','CLEAN','INFECTED','ERROR','UNAVAILABLE')),
     CONSTRAINT uk_education_content_import_asset_tenant_id UNIQUE (tenant_id, id),
     CONSTRAINT uk_education_content_import_asset_object UNIQUE (tenant_id, object_key)
 );
-
-COMMENT ON TABLE education_content_import_asset IS '教育内容导入源文件元数据；文件存储与扫描由 Infra 持有';
-COMMENT ON COLUMN education_content_import_asset.object_key IS 'Infra 私有对象键，不存储公开 URL';
+COMMENT ON TABLE education_content_import_asset IS 'Education-owned tenant binding for an Infra file reference admitted for content import';
+COMMENT ON COLUMN education_content_import_asset.object_key IS 'Opaque reference returned by the public Infra File API; never client supplied';
 
 CREATE TABLE education_content_import_job (
     id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
     tenant_id BIGINT NOT NULL,
     asset_id BIGINT NOT NULL,
+    actor_id BIGINT NOT NULL,
+    import_type VARCHAR(32) NOT NULL DEFAULT 'QUESTION_DRAFT',
+    status VARCHAR(24) NOT NULL DEFAULT 'PREVIEW_PENDING',
     preview_key VARCHAR(128) NOT NULL,
-    import_type VARCHAR(32) NOT NULL,
-    status VARCHAR(16) NOT NULL DEFAULT 'PREVIEW',
+    preview_request_hash VARCHAR(64) NOT NULL,
+    execute_key VARCHAR(128),
+    execute_request_hash VARCHAR(64),
+    scan_status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
+    parser_status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
+    preview_payload JSONB NOT NULL DEFAULT '{}'::JSONB,
+    parsed_payload JSONB,
+    result_summary JSONB NOT NULL DEFAULT '{}'::JSONB,
+    preview_question_count INTEGER,
+    imported_question_count INTEGER,
+    failure_code VARCHAR(64),
+    failure_message VARCHAR(500),
     attempt_count INTEGER NOT NULL DEFAULT 0,
-    max_attempts INTEGER NOT NULL DEFAULT 3,
+    max_attempts INTEGER NOT NULL DEFAULT 5,
     next_attempt_at TIMESTAMP,
     locked_by VARCHAR(128),
     lease_token VARCHAR(64),
@@ -37,38 +50,32 @@ CREATE TABLE education_content_import_job (
     last_heartbeat_at TIMESTAMP,
     started_at TIMESTAMP,
     finished_at TIMESTAMP,
-    error_message VARCHAR(2000),
-    preview_payload JSONB NOT NULL DEFAULT '{}'::JSONB,
-    result_summary JSONB NOT NULL DEFAULT '{}'::JSONB,
-    creator VARCHAR(64) DEFAULT '',
-    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    updater VARCHAR(64) DEFAULT '',
-    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    creator VARCHAR(64) DEFAULT '', create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) DEFAULT '', update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     deleted BOOLEAN NOT NULL DEFAULT false,
-    CONSTRAINT fk_education_content_import_job_asset_tenant
-        FOREIGN KEY (tenant_id, asset_id)
+    CONSTRAINT fk_education_content_import_job_asset_tenant FOREIGN KEY (tenant_id, asset_id)
         REFERENCES education_content_import_asset (tenant_id, id),
     CONSTRAINT uk_education_content_import_job_preview UNIQUE (tenant_id, preview_key),
-    CONSTRAINT ck_education_content_import_job_status
-        CHECK (status IN ('PREVIEW', 'PENDING', 'PROCESSING', 'COMPLETED', 'FAILED')),
-    CONSTRAINT ck_education_content_import_job_attempts
-        CHECK (attempt_count >= 0 AND max_attempts > 0 AND attempt_count <= max_attempts),
+    CONSTRAINT uk_education_content_import_job_execute UNIQUE (tenant_id, execute_key),
+    CONSTRAINT ck_education_content_import_job_status CHECK (status IN
+        ('PREVIEW_PENDING','PREVIEW_READY','EXECUTE_PENDING','COMPLETED','FAILED')),
+    CONSTRAINT ck_education_content_import_job_attempts CHECK
+        (attempt_count >= 0 AND max_attempts > 0 AND attempt_count <= max_attempts),
+    CONSTRAINT ck_education_content_import_job_counts CHECK
+        (COALESCE(preview_question_count, 0) >= 0 AND COALESCE(imported_question_count, 0) >= 0),
     CONSTRAINT ck_education_content_import_job_lease CHECK (
-        (status = 'PROCESSING' AND locked_by IS NOT NULL AND lease_token IS NOT NULL
-             AND lease_expires_at IS NOT NULL AND last_heartbeat_at IS NOT NULL)
-        OR
-        (status <> 'PROCESSING' AND locked_by IS NULL AND lease_token IS NULL
-             AND lease_expires_at IS NULL AND last_heartbeat_at IS NULL)
-    )
+        (lease_token IS NULL AND locked_by IS NULL AND lease_expires_at IS NULL AND last_heartbeat_at IS NULL)
+        OR (lease_token IS NOT NULL AND locked_by IS NOT NULL AND lease_expires_at IS NOT NULL AND last_heartbeat_at IS NOT NULL
+            AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING'))),
+    CONSTRAINT ck_education_content_import_job_execute CHECK
+        (status <> 'EXECUTE_PENDING' OR (execute_key IS NOT NULL AND execute_request_hash IS NOT NULL)),
+    CONSTRAINT ck_education_content_import_job_failed CHECK
+        (status <> 'FAILED' OR failure_code IS NOT NULL)
 );
+COMMENT ON TABLE education_content_import_job IS 'Education import aggregate with preview/execute lifecycle and token-fenced worker lease';
+COMMENT ON COLUMN education_content_import_job.lease_token IS 'Fencing token required by heartbeat and terminal worker writes';
 
-COMMENT ON TABLE education_content_import_job IS '教育内容导入预览与执行作业';
-COMMENT ON COLUMN education_content_import_job.preview_key IS '租户内重复预览幂等键';
-COMMENT ON COLUMN education_content_import_job.lease_token IS '处理租约 fencing token';
-
-CREATE INDEX idx_education_content_import_job_ready
-    ON education_content_import_job (next_attempt_at, create_time, id)
-    WHERE status = 'PENDING' AND deleted = false;
-CREATE INDEX idx_education_content_import_job_expired
-    ON education_content_import_job (lease_expires_at, create_time, id)
-    WHERE status = 'PROCESSING' AND deleted = false;
+CREATE INDEX idx_education_content_import_job_claim ON education_content_import_job
+    (next_attempt_at, lease_expires_at, create_time, id)
+    WHERE status IN ('PREVIEW_PENDING','EXECUTE_PENDING') AND deleted=false;
+CREATE INDEX idx_education_content_import_job_tenant ON education_content_import_job (tenant_id, create_time DESC, id DESC);

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapperPostgreSqlIntegrationTest.java

@@ -5,9 +5,7 @@ import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportA
 import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportJobDO;
 import cn.iocoder.yudao.module.education.test.PostgreSqlDbIntegrationTest;
 import jakarta.annotation.Resource;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.*;
 import org.springframework.dao.DataIntegrityViolationException;
 
 import java.time.LocalDateTime;
@@ -16,92 +14,58 @@ import java.util.List;
 import static org.junit.jupiter.api.Assertions.*;
 
 class ContentImportJobMapperPostgreSqlIntegrationTest extends PostgreSqlDbIntegrationTest {
+    @Resource(name="contentImportAssetMapper") ContentImportAssetMapper assetMapper;
+    @Resource(name="contentImportJobMapper") ContentImportJobMapper jobMapper;
 
-    @Resource(name = "contentImportAssetMapper") private ContentImportAssetMapper assetMapper;
-    @Resource(name = "contentImportJobMapper") private ContentImportJobMapper jobMapper;
+    @BeforeEach void setUp(){ TenantContextHolder.setTenantId(10L); }
+    @AfterEach void clear(){ TenantContextHolder.clear(); }
 
-    @BeforeEach
-    void setUp() {
-        TenantContextHolder.setTenantId(10L);
-    }
+    @Test void tenantBindingDuplicateCommandsAndLeaseFencingAreAtomic() {
+        ContentImportAssetDO asset=asset(10L,"ref-10"); assetMapper.insert(asset);
+        ContentImportJobDO preview=preview(10L,asset.getId(),"preview-1");
+        assertEquals(1,jobMapper.insertPreviewIgnore(preview));
+        assertEquals(0,jobMapper.insertPreviewIgnore(preview(10L,asset.getId(),"preview-1")));
+        ContentImportAssetDO other=asset(20L,"ref-20"); assetMapper.insert(other);
+        assertThrows(DataIntegrityViolationException.class,()->jobMapper.insertPreviewIgnore(preview(10L,other.getId(),"cross")));
 
-    @AfterEach
-    void tearDown() {
-        TenantContextHolder.clear();
-    }
-
-    @Test
-    void shouldEnforceTenantBoundJobLifecycleAndLeaseFencing() {
-        ContentImportAssetDO asset = asset(10L, "imports/10/questions.xlsx");
-        assetMapper.insert(asset);
-
-        ContentImportJobDO preview = preview(10L, asset.getId(), "preview-1");
-        assertEquals(1, jobMapper.insertPreviewIgnore(preview));
-        assertEquals(0, jobMapper.insertPreviewIgnore(preview(10L, asset.getId(), "preview-1")));
-
-        ContentImportAssetDO otherTenantAsset = asset(20L, "imports/20/questions.xlsx");
-        assetMapper.insert(otherTenantAsset);
-        assertThrows(DataIntegrityViolationException.class,
-                () -> jobMapper.insertPreviewIgnore(preview(10L, otherTenantAsset.getId(), "cross-tenant")));
-
-        assertEquals(1, jobMapper.executePreview(10L, preview.getId()));
-        assertEquals(0, jobMapper.executePreview(10L, preview.getId()));
-
-        List<ContentImportJobDO> claimed = jobMapper.claimReady("worker-a", "token-a", 60, 1);
-        assertEquals(1, claimed.size());
-        ContentImportJobDO job = claimed.getFirst();
-        assertEquals("PROCESSING", job.getStatus());
-        assertEquals(1, job.getAttemptCount());
-        assertEquals("token-a", job.getLeaseToken());
-        assertTrue(jobMapper.claimReady("worker-b", "token-b", 60, 1).isEmpty());
-
-        LocalDateTime firstExpiry = job.getLeaseExpiresAt();
-        assertEquals(1, jobMapper.heartbeat(10L, job.getId(), "token-a", 120));
+        List<ContentImportJobDO> claimed=jobMapper.claimReady("worker-a","token-a",60,1);
+        assertEquals(1,claimed.size());
+        ContentImportJobDO job=claimed.getFirst();
+        assertEquals("PREVIEW_PENDING",job.getStatus());
+        assertEquals(1,job.getAttemptCount());
+        assertTrue(jobMapper.claimReady("worker-b","token-b",60,1).isEmpty());
+        LocalDateTime firstExpiry=job.getLeaseExpiresAt();
+        assertEquals(1,jobMapper.heartbeat(10L,job.getId(),"token-a",120));
         assertTrue(jobMapper.selectById(job.getId()).getLeaseExpiresAt().isAfter(firstExpiry));
-        assertEquals(0, jobMapper.heartbeat(10L, job.getId(), "stale-token", 120));
-        assertEquals(0, jobMapper.complete(10L, job.getId(), "stale-token", "{}"));
+        assertEquals(0,jobMapper.heartbeat(10L,job.getId(),"stale",120));
+        assertEquals(0,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","stale",
+                "CLEAN","UNAVAILABLE","{}",null,null,null,null,null,null));
 
         jobMapper.expireLeaseForTest(job.getId());
-        assertEquals(1, jobMapper.recoverExpired(10));
-        assertEquals("PENDING", jobMapper.selectById(job.getId()).getStatus());
-
-        ContentImportJobDO reclaimed = jobMapper.claimReady("worker-b", "token-b", 60, 1).getFirst();
-        assertEquals(2, reclaimed.getAttemptCount());
-        assertEquals(0, jobMapper.complete(10L, job.getId(), "token-a", "{}"));
-        assertEquals(1, jobMapper.complete(10L, job.getId(), "token-b", "{\"inserted\":1}"));
-        assertEquals("COMPLETED", jobMapper.selectById(job.getId()).getStatus());
+        ContentImportJobDO recovered=jobMapper.claimReady("worker-b","token-b",60,1).getFirst();
+        assertEquals(2,recovered.getAttemptCount());
+        assertEquals(0,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","token-a",
+                "CLEAN","UNAVAILABLE","{}",null,null,null,null,null,null));
+        assertEquals(1,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","token-b",
+                "CLEAN","UNAVAILABLE","{\"mode\":\"METADATA_ONLY\",\"executable\":false}",null,null,null,null,null,null));
+        assertEquals("PREVIEW_READY",jobMapper.selectById(job.getId()).getStatus());
+        assertEquals(0,jobMapper.requestExecute(10L,job.getId(),"execute-1","b".repeat(64)));
     }
 
-    @Test
-    void shouldAllowOnlyFiveJobStates() {
-        ContentImportAssetDO asset = asset(10L, "imports/10/states.xlsx");
-        assetMapper.insert(asset);
-        ContentImportJobDO job = preview(10L, asset.getId(), "states");
-        job.setStatus("REJECTED");
-
-        assertThrows(DataIntegrityViolationException.class, () -> jobMapper.insert(job));
+    @Test void onlyRequiredFiveStatesAreAccepted(){
+        ContentImportAssetDO asset=asset(10L,"states"); assetMapper.insert(asset);
+        ContentImportJobDO job=preview(10L,asset.getId(),"states"); job.setStatus("PROCESSING");
+        assertThrows(DataIntegrityViolationException.class,()->jobMapper.insert(job));
     }
 
-    private ContentImportAssetDO asset(Long tenantId, String objectKey) {
-        ContentImportAssetDO asset = new ContentImportAssetDO();
-        asset.setTenantId(tenantId);
-        asset.setObjectKey(objectKey);
-        asset.setFileName("questions.xlsx");
-        asset.setMimeType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
-        asset.setFileSizeBytes(128L);
-        asset.setChecksumSha256("a".repeat(64));
-        return asset;
+    private ContentImportAssetDO asset(Long tenant,String key){
+        ContentImportAssetDO a=new ContentImportAssetDO(); a.setTenantId(tenant); a.setOwnerUserId(7L); a.setObjectKey(key);
+        a.setFileName("questions.xlsx"); a.setMimeType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
+        a.setFileSizeBytes(128L); a.setChecksumSha256("a".repeat(64)); a.setScanStatus("UNAVAILABLE"); return a;
     }
-
-    private ContentImportJobDO preview(Long tenantId, Long assetId, String previewKey) {
-        ContentImportJobDO job = new ContentImportJobDO();
-        job.setTenantId(tenantId);
-        job.setAssetId(assetId);
-        job.setPreviewKey(previewKey);
-        job.setImportType("QUESTIONS");
-        job.setStatus("PREVIEW");
-        job.setMaxAttempts(3);
-        job.setPreviewPayload("{}");
-        return job;
+    private ContentImportJobDO preview(Long tenant,Long asset,String key){
+        ContentImportJobDO j=new ContentImportJobDO(); j.setTenantId(tenant); j.setAssetId(asset); j.setActorId(7L);
+        j.setPreviewKey(key); j.setPreviewRequestHash("a".repeat(64)); j.setImportType("QUESTION_DRAFT");
+        j.setStatus("PREVIEW_PENDING"); j.setMaxAttempts(5); j.setPreviewPayload("{}"); return j;
     }
 }

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/asset/EducationAssetAdmissionServiceImplTest.java

@@ -28,7 +28,7 @@ class EducationAssetAdmissionServiceImplTest {
     @BeforeEach
     void setUp() {
         TenantContextHolder.setTenantId(42L);
-        when(mapper.insert(any())).thenAnswer(invocation -> {
+        when(mapper.insert(any(EducationImportAssetDO.class))).thenAnswer(invocation -> {
             EducationImportAssetDO asset = invocation.getArgument(0);
             asset.setId(99L);
             return 1;