package com.baiyue.eventbridge.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.baiyue.eventbridge.domain.BridgeInbox; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; import java.util.List; /** * 入站事件持久化映射。 */ @Mapper public interface BridgeInboxMapper extends BaseMapper { @Select(""" SELECT event_id FROM bridge_inbox WHERE ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3))) OR (status = 'PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3))) ORDER BY received_at LIMIT #{limit} """) List dispatchCandidates(@Param("limit") int limit); @Update(""" UPDATE bridge_inbox SET status='PROCESSING', lease_owner=#{owner}, lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND) WHERE event_id=#{eventId} AND ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3))) OR (status='PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3))) """) int claim(@Param("eventId") String eventId, @Param("owner") String owner, @Param("leaseSeconds") long leaseSeconds); @Update(""" UPDATE bridge_inbox SET status='PROCESSED', processed_at=CURRENT_TIMESTAMP(3), lease_owner=NULL, lease_until=NULL, next_retry_at=NULL, last_error=NULL WHERE event_id=#{eventId} AND status='PROCESSING' AND lease_owner=#{owner} """) int markProcessed(@Param("eventId") String eventId, @Param("owner") String owner); @Update(""" UPDATE bridge_inbox SET retry_count=retry_count+1, status=CASE WHEN retry_count+1 >= #{maxAttempts} THEN 'DEAD' ELSE 'RETRY' END, next_retry_at=CASE WHEN retry_count+1 >= #{maxAttempts} THEN NULL ELSE DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{backoffSeconds} SECOND) END, lease_owner=NULL, lease_until=NULL, last_error=#{error} WHERE event_id=#{eventId} AND status='PROCESSING' AND lease_owner=#{owner} """) int markFailed(@Param("eventId") String eventId, @Param("owner") String owner, @Param("maxAttempts") int maxAttempts, @Param("backoffSeconds") long backoffSeconds, @Param("error") String error); @Select("SELECT COALESCE(MAX(attempt_no), 0) + 1 FROM bridge_delivery_attempt WHERE event_id=#{eventId}") int nextAttemptNo(@Param("eventId") String eventId); @Update(""" UPDATE bridge_inbox SET status='RETRY', retry_count=0, next_retry_at=CURRENT_TIMESTAMP(3), lease_owner=NULL, lease_until=NULL, last_error=#{reason} WHERE event_id=#{eventId} AND status='DEAD' """) int replayDead(@Param("eventId") String eventId, @Param("reason") String reason); }