package tech.easyflow.agent.runtime.document;

import com.mybatisflex.core.query.QueryWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.easyflow.agent.config.AgentDocumentProperties;
import tech.easyflow.agent.entity.AgentDocumentAttachment;
import tech.easyflow.agent.entity.AgentDocumentSnapshot;
import tech.easyflow.agent.mapper.AgentDocumentAttachmentMapper;
import tech.easyflow.agent.mapper.AgentDocumentSnapshotMapper;

import java.time.Instant;
import java.util.Date;
import java.util.List;

/**
 * Agent 文档跨存储部分成功与丢失 MQ 的单实例补偿任务。
 */
@Component
public class AgentDocumentCompensationScheduler {

    private static final Logger LOG = LoggerFactory.getLogger(AgentDocumentCompensationScheduler.class);
    private static final int BATCH_SIZE = 100;

    private final AgentDocumentAttachmentMapper attachmentMapper;
    private final AgentDocumentSnapshotMapper snapshotMapper;
    private final AgentDocumentReaderService readerService;
    private final AgentDocumentService documentService;
    private final AgentDocumentReadTaskProducer taskProducer;
    private final AgentDocumentProperties properties;

    /**
     * 创建补偿任务。
     *
     * @param attachmentMapper 附件 Mapper
     * @param snapshotMapper 快照 Mapper
     * @param readerService 读取服务
     * @param documentService 文档生命周期服务
     * @param taskProducer 任务生产者
     * @param properties 文档配置
     */
    public AgentDocumentCompensationScheduler(AgentDocumentAttachmentMapper attachmentMapper,
                                              AgentDocumentSnapshotMapper snapshotMapper,
                                              AgentDocumentReaderService readerService,
                                              AgentDocumentService documentService,
                                              AgentDocumentReadTaskProducer taskProducer,
                                              AgentDocumentProperties properties) {
        this.attachmentMapper = attachmentMapper;
        this.snapshotMapper = snapshotMapper;
        this.readerService = readerService;
        this.documentService = documentService;
        this.taskProducer = taskProducer;
        this.properties = properties;
    }

    /**
     * 每分钟重新推进状态账本，MQ 只作为及时触发器。
     */
    @Scheduled(fixedDelay = 60_000L)
    public void compensate() {
        if (!properties.isEnabled()) {
            return;
        }
        try {
            repairUploading();
            repairWritingSnapshots();
            recoverStaleReading();
            republishUploaded();
            expireTemporaryAttachments();
            cleanupPending();
        } catch (RuntimeException error) {
            LOG.error("Agent 文档补偿扫描失败", error);
        }
    }

    private void repairUploading() {
        Date cutoff = Date.from(Instant.now().minus(properties.getReadTimeout().multipliedBy(2)));
        List<AgentDocumentAttachment> records = attachmentMapper.selectListByQuery(QueryWrapper.create()
                .eq("status", AgentDocumentStatus.UPLOADING.name())
                .le("modified", cutoff)
                .limit(BATCH_SIZE));
        for (AgentDocumentAttachment attachment : records) {
            if (readerService.repairUploading(attachment)) {
                sendQuietly(attachment.getAttachmentId());
                continue;
            }
            documentService.markDeletePending(attachment);
        }
    }

    private void repairWritingSnapshots() {
        List<AgentDocumentSnapshot> records = snapshotMapper.selectListByQuery(QueryWrapper.create()
                .eq("status", AgentDocumentSnapshotStatus.WRITING.name())
                .limit(BATCH_SIZE));
        for (AgentDocumentSnapshot snapshot : records) {
            readerService.repairWritingSnapshot(snapshot);
        }
    }

    private void recoverStaleReading() {
        Date cutoff = Date.from(Instant.now().minus(properties.getReadTimeout().multipliedBy(2)));
        List<AgentDocumentAttachment> records = attachmentMapper.selectListByQuery(QueryWrapper.create()
                .eq("status", AgentDocumentStatus.READING.name())
                .le("modified", cutoff)
                .limit(BATCH_SIZE));
        for (AgentDocumentAttachment attachment : records) {
            AgentDocumentAttachment update = new AgentDocumentAttachment();
            update.setStatus(AgentDocumentStatus.UPLOADED.name());
            update.setVersion((attachment.getVersion() == null ? 0L : attachment.getVersion()) + 1L);
            update.setModified(new Date());
            int updated = attachmentMapper.updateByQuery(update, QueryWrapper.create()
                    .eq("id", attachment.getId())
                    .eq("status", AgentDocumentStatus.READING.name())
                    .eq("version", attachment.getVersion()));
            if (updated == 1) {
                sendQuietly(attachment.getAttachmentId());
            }
        }
    }

    private void republishUploaded() {
        List<AgentDocumentAttachment> records = attachmentMapper.selectListByQuery(QueryWrapper.create()
                .eq("status", AgentDocumentStatus.UPLOADED.name())
                .limit(BATCH_SIZE));
        for (AgentDocumentAttachment attachment : records) {
            sendQuietly(attachment.getAttachmentId());
        }
    }

    private void expireTemporaryAttachments() {
        List<AgentDocumentAttachment> records = attachmentMapper.selectListByQuery(QueryWrapper.create()
                .in("status", List.of(
                        AgentDocumentStatus.UPLOADING.name(),
                        AgentDocumentStatus.UPLOADED.name(),
                        AgentDocumentStatus.READ_FAILED.name(),
                        AgentDocumentStatus.READY.name()))
                .le("expires_at", new Date())
                .limit(BATCH_SIZE));
        for (AgentDocumentAttachment attachment : records) {
            documentService.markDeletePending(attachment);
        }
    }

    private void cleanupPending() {
        List<AgentDocumentAttachment> records = attachmentMapper.selectListByQuery(QueryWrapper.create()
                .eq("status", AgentDocumentStatus.DELETE_PENDING.name())
                .limit(BATCH_SIZE));
        for (AgentDocumentAttachment attachment : records) {
            try {
                documentService.deletePending(attachment);
            } catch (RuntimeException error) {
                LOG.error("Agent 文档待删除对象清理失败: attachmentId={}",
                        attachment.getAttachmentId(), error);
            }
        }
    }

    private void sendQuietly(String attachmentId) {
        try {
            taskProducer.send(attachmentId);
        } catch (RuntimeException error) {
            LOG.error("Agent 文档补偿消息投递失败: attachmentId={}", attachmentId, error);
        }
    }
}
