baiyue-firstleg/firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletionCommandMapper.java

package com.baiyue.firstleg.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Options;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;

import java.time.LocalDateTime;
import java.util.List;

@Mapper
public interface LoadingCompletionCommandMapper extends BaseMapper<LoadingCompletionCommandEntity> {
    @Insert("""
        INSERT IGNORE INTO fl_loading_completion_command
          (id,pick_task_id,operation_id,request_hash,command_status,expected_task_version,remark,
           operator_id,operator_name,attempt_count,next_retry_at)
        VALUES
          (#{id},#{pickTaskId},#{operationId},#{requestHash},'PROCESSING',#{expectedTaskVersion},#{remark},
           #{operatorId},#{operatorName},0,CURRENT_TIMESTAMP(3))
        """)
    int insertCommand(LoadingCompletionCommandEntity command);

    @Select("SELECT * FROM fl_loading_completion_command WHERE operation_id=#{operationId}")
    LoadingCompletionCommandEntity findByOperation(@Param("operationId") String operationId);

    @Select("""
        SELECT * FROM fl_loading_completion_command
         WHERE pick_task_id=#{taskId} AND command_status IN ('PROCESSING','RETRYABLE','FAILED')
         ORDER BY created_at DESC LIMIT 1
        """)
    LoadingCompletionCommandEntity findActiveByTask(@Param("taskId") long taskId);

    @Select("SELECT * FROM fl_loading_completion_command WHERE pick_task_id=#{taskId} ORDER BY created_at DESC LIMIT 1")
    LoadingCompletionCommandEntity findLatestByTask(@Param("taskId") long taskId);

    @Select("""
        SELECT * FROM fl_loading_completion_command
         WHERE command_status IN ('PROCESSING','RETRYABLE') AND next_retry_at<=CURRENT_TIMESTAMP(3)
           AND (lease_until IS NULL OR lease_until<CURRENT_TIMESTAMP(3))
         ORDER BY next_retry_at,id LIMIT #{limit}
        """)
    List<LoadingCompletionCommandEntity> findDue(@Param("limit") int limit);

    @Update("""
        UPDATE fl_loading_completion_command
           SET command_status='PROCESSING',lease_owner=#{worker},lease_until=#{leaseUntil},
               attempt_count=attempt_count+1,updated_at=CURRENT_TIMESTAMP(3)
         WHERE id=#{id} AND command_status IN ('PROCESSING','RETRYABLE')
           AND next_retry_at<=CURRENT_TIMESTAMP(3)
           AND (lease_until IS NULL OR lease_until<CURRENT_TIMESTAMP(3))
        """)
    int claim(@Param("id") long id, @Param("worker") String worker,
              @Param("leaseUntil") LocalDateTime leaseUntil);

    /** 只有尚未过期的当前所有者可以续租，过期执行者必须重新领取。 */
    @Update("""
        UPDATE fl_loading_completion_command SET lease_until=#{leaseUntil},updated_at=CURRENT_TIMESTAMP(3)
         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    int renew(@Param("id") long id, @Param("worker") String worker,
              @Param("leaseUntil") LocalDateTime leaseUntil);

    /** 完成落库必须仍持有有效租约，拒绝过期或已易主的旧执行者。 */
    @Update("""
        UPDATE fl_loading_completion_command
           SET command_status='COMPLETED',completed_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
               lease_owner=NULL,lease_until=NULL,failure_stage=NULL,failure_code=NULL,
               failure_request_id=NULL,failure_trace_id=NULL
         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    int complete(@Param("id") long id, @Param("worker") String worker);

    /** 失败回执也受当前有效租约保护，不能覆盖新执行者的状态。 */
    @Update("""
        UPDATE fl_loading_completion_command
           SET command_status=#{status},next_retry_at=#{nextRetryAt},updated_at=CURRENT_TIMESTAMP(3),
               lease_owner=NULL,lease_until=NULL,failure_stage=#{stage},failure_code=#{code},
               failure_request_id=#{requestId},failure_trace_id=#{traceId}
         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
           AND lease_until>CURRENT_TIMESTAMP(3)
        """)
    int fail(@Param("id") long id, @Param("worker") String worker, @Param("status") String status,
             @Param("nextRetryAt") LocalDateTime nextRetryAt, @Param("stage") String stage,
             @Param("code") String code, @Param("requestId") String requestId, @Param("traceId") String traceId);

    @Update("""
        UPDATE fl_loading_completion_command
           SET command_status='RETRYABLE',next_retry_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
               lease_owner=NULL,lease_until=NULL
         WHERE id=#{id} AND command_status='FAILED'
        """)
    int resume(@Param("id") long id);

    @Select("""
        SELECT checkpoint_stage FROM fl_loading_completion_checkpoint
         WHERE command_id=#{commandId} AND tenant_id=#{tenantId} AND inbound_id=#{inboundId}
        """)
    String checkpoint(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
                      @Param("inboundId") long inboundId);

    /** 检查点事务锁住当前有效租约，与并发重新领取互斥。 */
    @Options(useCache = false, flushCache = Options.FlushCachePolicy.TRUE)
    @Select("""
        SELECT id FROM fl_loading_completion_command
         WHERE id=#{commandId} AND command_status='PROCESSING' AND lease_owner=#{worker}
           AND lease_until>CURRENT_TIMESTAMP(3)
         FOR UPDATE
        """)
    Long lockCheckpointLease(@Param("commandId") long commandId, @Param("worker") String worker);

    /** 在已锁定的命令行上重验有效租约；检查点只能前进，不能被迟到写入降级。 */
    @Insert("""
        INSERT INTO fl_loading_completion_checkpoint
          (command_id,tenant_id,inbound_id,owner_sequence,checkpoint_stage)
        SELECT id,#{tenantId},#{inboundId},#{sequence},#{stage}
          FROM fl_loading_completion_command
         WHERE id=#{commandId} AND command_status='PROCESSING' AND lease_owner=#{worker}
           AND lease_until>CURRENT_TIMESTAMP(3)
        ON DUPLICATE KEY UPDATE checkpoint_stage=CASE
          WHEN checkpoint_stage='INBOUND_COMPLETED' OR #{stage}='INBOUND_COMPLETED'
            THEN 'INBOUND_COMPLETED'
          WHEN checkpoint_stage='SETTLED' OR #{stage}='SETTLED' THEN 'SETTLED'
          ELSE 'READY' END,updated_at=CURRENT_TIMESTAMP(3)
        """)
    int checkpointCompleted(@Param("commandId") long commandId, @Param("worker") String worker,
                            @Param("tenantId") long tenantId,
                            @Param("inboundId") long inboundId, @Param("sequence") int sequence,
                            @Param("stage") String stage);
}