package com.sino.mci.send.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.sino.mci.channel.adapter.registry.ChannelAdapterRegistry;
import com.sino.mci.channel.common.dto.SendMessageRequest;
import com.sino.mci.channel.common.dto.SendMessageResult;
import com.sino.mci.channel.common.model.ChannelType;
import com.sino.mci.channel.common.spi.ChannelAdapter;
import com.sino.mci.channel.entity.ChannelAccount;
import com.sino.mci.channel.entity.ChannelAccountConversation;
import com.sino.mci.channel.entity.Conversation;
import com.sino.mci.channel.repository.ChannelAccountConversationMapper;
import com.sino.mci.channel.repository.ChannelAccountMapper;
import com.sino.mci.channel.repository.ConversationMapper;
import com.sino.mci.channel.service.AccountSelector;
import com.sino.mci.channel.entity.ConversationTagBinding;
import com.sino.mci.crm.entity.ChannelIdentity;
import com.sino.mci.crm.entity.Person;
import com.sino.mci.crm.entity.PersonTagBinding;
import com.sino.mci.crm.entity.Tag;
import com.sino.mci.channel.repository.ConversationTagBindingMapper;
import com.sino.mci.crm.repository.ChannelIdentityMapper;
import com.sino.mci.crm.repository.PersonMapper;
import com.sino.mci.crm.repository.PersonTagBindingMapper;
import com.sino.mci.crm.repository.TagMapper;
import com.sino.mci.exception.BusinessException;
import com.sino.mci.policy.engine.PolicyCheckResult;
import com.sino.mci.policy.engine.PolicyEngine;
import com.sino.mci.policy.engine.PolicyGroupConfig;
import com.sino.mci.policy.entity.PolicyGroup;
import com.sino.mci.policy.repository.PolicyGroupMapper;
import com.sino.mci.policy.engine.RetryConfig;
import com.sino.mci.send.dto.ChannelStatistics;
import com.sino.mci.send.dto.SendRecordResponse;
import com.sino.mci.send.dto.SendRequest;
import com.sino.mci.send.dto.SendStatisticsResponse;
import com.sino.mci.send.dto.SendTaskStatusResponse;
import com.sino.mci.send.dto.SendResponse;
import com.sino.mci.send.dto.SendTestRequest;
import com.sino.mci.send.entity.SendRecord;
import com.sino.mci.send.mq.SendMessageProducer;
import com.sino.mci.send.mq.SendRecordMessage;
import com.sino.mci.send.repository.SendRecordMapper;
import com.sino.mci.shared.util.JsonUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class SendService {

    private static final int STATUS_ENABLED = 1;
    private static final int SEND_STATUS_PENDING = 0;
    private static final int SEND_STATUS_SENDING = 1;
    private static final int SEND_STATUS_SUCCESS = 2;
    private static final int SEND_STATUS_FAILED = 3;

    private final SendRecordMapper sendRecordMapper;
    private final ConversationMapper conversationMapper;
    private final ChannelAccountMapper accountMapper;
    private final ChannelAccountConversationMapper channelAccountConversationMapper;
    private final ChannelIdentityMapper channelIdentityMapper;
    private final PersonMapper personMapper;
    private final TagMapper tagMapper;
    private final PersonTagBindingMapper personTagBindingMapper;
    private final ConversationTagBindingMapper conversationTagBindingMapper;
    private final ChannelAdapterRegistry adapterRegistry;
    private final PolicyGroupMapper policyGroupMapper;
    private final PolicyEngine policyEngine;
    private final AccountSelector accountSelector;
    private final SendMessageProducer sendMessageProducer;

    @Transactional
    public SendResponse send(String tenantCode, SendRequest request) {
        String taskId = generateTaskId();
        List<SendTarget> targets = resolveTargets(tenantCode, request);
        RetryConfig retryConfig = loadRetryConfig(tenantCode, request);
        List<SendRecord> records = createRecords(taskId, tenantCode, request, targets, retryConfig);

        for (SendRecord record : records) {
            SendRecordMessage message = SendRecordMessage.builder()
                    .recordId(record.getId())
                    .taskId(record.getTaskId())
                    .tenantCode(record.getTenantCode())
                    .attempt(0)
                    .retryPlan(retryConfig)
                    .build();
            sendMessageProducer.publishSendMessage(message);
        }

        return new SendResponse(taskId, records.size(), "accepted");
    }

    @Transactional
    public SendResponse testSend(String tenantCode, SendTestRequest request) {
        String taskId = generateTaskId();
        List<SendTarget> targets = resolveTestTargets(tenantCode, request);
        List<SendRecord> records = createTestRecords(taskId, tenantCode, request, targets);
        return new SendResponse(taskId, records.size(), "simulated");
    }

    public List<SendRecordResponse> listRecords(String tenantCode, String taskId) {
        List<SendRecord> records = sendRecordMapper.selectList(
                new LambdaQueryWrapper<SendRecord>()
                        .eq(SendRecord::getTenantCode, tenantCode)
                        .eq(SendRecord::getTaskId, taskId)
                        .orderByAsc(SendRecord::getId));
        return records.stream().map(this::toResponse).collect(Collectors.toList());
    }

    public List<SendRecordResponse> listRecentRecords(String tenantCode) {
        List<SendRecord> records = sendRecordMapper.selectList(
                new LambdaQueryWrapper<SendRecord>()
                        .eq(SendRecord::getTenantCode, tenantCode)
                        .orderByDesc(SendRecord::getId));
        return records.stream().limit(50).map(this::toResponse).collect(Collectors.toList());
    }

    public com.baomidou.mybatisplus.core.metadata.IPage<SendRecordResponse> queryRecords(String tenantCode, String taskId, LocalDateTime start,
                                                  LocalDateTime end, String channelType, Integer status,
                                                  long page, long size) {
        LambdaQueryWrapper<SendRecord> wrapper = new LambdaQueryWrapper<SendRecord>()
                .eq(SendRecord::getTenantCode, tenantCode)
                .orderByDesc(SendRecord::getId);
        if (StringUtils.hasText(taskId)) {
            wrapper.eq(SendRecord::getTaskId, taskId);
        }
        if (start != null) {
            wrapper.ge(SendRecord::getSendTime, start);
        }
        if (end != null) {
            wrapper.le(SendRecord::getSendTime, end);
        }
        if (StringUtils.hasText(channelType)) {
            wrapper.eq(SendRecord::getChannelType, channelType);
        }
        if (status != null) {
            wrapper.eq(SendRecord::getSendStatus, status);
        }
        Page<SendRecord> pageParam = new Page<>(page, size);
        sendRecordMapper.selectPage(pageParam, wrapper);
        return pageParam.convert(this::toResponse);
    }

    public SendTaskStatusResponse getTaskStatus(String tenantCode, String taskId) {
        List<SendRecord> records = sendRecordMapper.selectList(
                new LambdaQueryWrapper<SendRecord>()
                        .eq(SendRecord::getTenantCode, tenantCode)
                        .eq(SendRecord::getTaskId, taskId));
        SendTaskStatusResponse response = new SendTaskStatusResponse();
        response.setTaskId(taskId);
        response.setTotal(records.size());
        for (SendRecord record : records) {
            switch (record.getSendStatus()) {
                case SEND_STATUS_PENDING:
                    response.setPendingCount(response.getPendingCount() + 1);
                    break;
                case SEND_STATUS_SENDING:
                    response.setSendingCount(response.getSendingCount() + 1);
                    break;
                case SEND_STATUS_SUCCESS:
                    response.setSuccessCount(response.getSuccessCount() + 1);
                    break;
                case SEND_STATUS_FAILED:
                    response.setFailedCount(response.getFailedCount() + 1);
                    break;
                default:
                    break;
            }
        }
        return response;
    }

    @Transactional
    public int cancelTask(String tenantCode, String taskId) {
        return sendRecordMapper.update(null,
                new LambdaUpdateWrapper<SendRecord>()
                        .eq(SendRecord::getTenantCode, tenantCode)
                        .eq(SendRecord::getTaskId, taskId)
                        .eq(SendRecord::getSendStatus, SEND_STATUS_PENDING)
                        .set(SendRecord::getSendStatus, SEND_STATUS_FAILED)
                        .set(SendRecord::getFinishTime, LocalDateTime.now())
                        .set(SendRecord::getChannelResult,
                                JsonUtils.toJson(Collections.singletonMap("canceled", true))));
    }

    public SendStatisticsResponse getStatistics(String tenantCode, LocalDateTime start, LocalDateTime end) {
        LambdaQueryWrapper<SendRecord> wrapper = new LambdaQueryWrapper<SendRecord>()
                .eq(SendRecord::getTenantCode, tenantCode);
        if (start != null) {
            wrapper.ge(SendRecord::getSendTime, start);
        }
        if (end != null) {
            wrapper.le(SendRecord::getSendTime, end);
        }
        List<SendRecord> records = sendRecordMapper.selectList(wrapper);

        SendStatisticsResponse response = new SendStatisticsResponse();
        Map<String, ChannelStatistics> channelMap = new HashMap<>();
        for (SendRecord record : records) {
            response.setTotal(response.getTotal() + 1);
            switch (record.getSendStatus()) {
                case SEND_STATUS_PENDING:
                    response.setPendingCount(response.getPendingCount() + 1);
                    break;
                case SEND_STATUS_SENDING:
                    response.setSendingCount(response.getSendingCount() + 1);
                    break;
                case SEND_STATUS_SUCCESS:
                    response.setSuccessCount(response.getSuccessCount() + 1);
                    break;
                case SEND_STATUS_FAILED:
                    response.setFailedCount(response.getFailedCount() + 1);
                    break;
                default:
                    break;
            }
            ChannelStatistics channelStats = channelMap.computeIfAbsent(record.getChannelType(),
                    k -> {
                        ChannelStatistics cs = new ChannelStatistics();
                        cs.setChannelType(k);
                        return cs;
                    });
            channelStats.setTotal(channelStats.getTotal() + 1);
            switch (record.getSendStatus()) {
                case SEND_STATUS_PENDING:
                    channelStats.setPendingCount(channelStats.getPendingCount() + 1);
                    break;
                case SEND_STATUS_SENDING:
                    channelStats.setSendingCount(channelStats.getSendingCount() + 1);
                    break;
                case SEND_STATUS_SUCCESS:
                    channelStats.setSuccessCount(channelStats.getSuccessCount() + 1);
                    break;
                case SEND_STATUS_FAILED:
                    channelStats.setFailedCount(channelStats.getFailedCount() + 1);
                    break;
                default:
                    break;
            }
        }
        response.setChannelStats(new ArrayList<>(channelMap.values()));
        return response;
    }

    private String generateTaskId() {
        return "SND" + UUID.randomUUID().toString().replace("-", "").toUpperCase();
    }

    List<SendTarget> resolveTargets(String tenantCode, SendRequest request) {
        return resolveByType(tenantCode, request.getTargetType(), request.getTargetValue(), request.getChannelStrategy());
    }

    List<SendTarget> resolveTestTargets(String tenantCode, SendTestRequest request) {
        return resolveByType(tenantCode, request.getTargetType(), request.getTargetValue(), request.getChannelStrategy());
    }

    @SuppressWarnings("unchecked")
    List<SendTarget> resolveByType(String tenantCode, String targetType, Map<String, Object> targetValue,
                                   Map<String, Object> channelStrategy) {
        String channelTypeStr = extractChannelType(channelStrategy);
        ChannelType channelType = parseChannelType(channelTypeStr);

        switch (targetType) {
            case "conversation":
                Number convId = (Number) targetValue.get("conversation_id");
                if (convId == null) {
                    throw new BusinessException("conversation 目标缺少 conversation_id");
                }
                return resolveConversationTarget(tenantCode, convId.longValue(), channelType, channelStrategy);
            case "person":
                Number personId = (Number) targetValue.get("person_id");
                if (personId == null) {
                    throw new BusinessException("person 目标缺少 person_id");
                }
                return resolvePersonTarget(tenantCode, personId.longValue(), channelType);
            case "person_list":
                List<Number> personIds = (List<Number>) targetValue.get("person_ids");
                if (CollectionUtils.isEmpty(personIds)) {
                    throw new BusinessException("person_list 目标缺少 person_ids");
                }
                List<SendTarget> targets = new ArrayList<>();
                for (Number pid : personIds) {
                    targets.addAll(resolvePersonTarget(tenantCode, pid.longValue(), channelType));
                }
                return targets;
            case "tag":
                Number tagId = (Number) targetValue.get("tag_id");
                if (tagId == null) {
                    throw new BusinessException("tag 目标缺少 tag_id");
                }
                String scope = (String) targetValue.getOrDefault("scope", "all");
                return resolveTagTarget(tenantCode, tagId.longValue(), channelType, scope);
            case "rule":
                return resolveRuleTarget(tenantCode, targetValue, channelType, channelStrategy);
            default:
                throw new BusinessException("不支持的目标类型: " + targetType);
        }
    }

    private List<SendTarget> resolveConversationTarget(String tenantCode, Long conversationId,
                                                       ChannelType channelType, Map<String, Object> channelStrategy) {
        Conversation conversation = conversationMapper.selectById(conversationId);
        if (conversation == null || !tenantCode.equals(conversation.getTenantCode())) {
            throw new BusinessException("会话不存在: " + conversationId);
        }
        Long accountId = selectSendAccount(tenantCode, conversationId, channelType, channelStrategy);
        if (accountId == null) {
            accountId = conversation.getOwnerAccountId();
        }
        if (accountId == null) {
            throw new BusinessException("会话未配置发送账号: " + conversationId);
        }
        ChannelAccount account = accountMapper.selectById(accountId);
        if (!AccountSelector.isHealthy(account, tenantCode, channelType)) {
            throw new BusinessException("会话没有可用的发送账号: " + conversationId);
        }
        SendTarget target = new SendTarget();
        target.setConversationId(conversationId);
        target.setPersonId(null);
        target.setAccountId(accountId);
        target.setChannelType(channelType);
        target.setChannelConversationId(conversation.getChannelConversationId());
        target.setReceiverId(conversation.getChannelConversationId());
        return Collections.singletonList(target);
    }

    private List<SendTarget> resolvePersonTarget(String tenantCode, Long personId, ChannelType channelType) {
        ChannelIdentity identity = channelIdentityMapper.selectOne(
                new LambdaQueryWrapper<ChannelIdentity>()
                        .eq(ChannelIdentity::getTenantCode, tenantCode)
                        .eq(ChannelIdentity::getPersonId, personId)
                        .eq(ChannelIdentity::getChannelType, channelType.name().toLowerCase())
                        .eq(ChannelIdentity::getStatus, STATUS_ENABLED));
        if (identity == null) {
            throw new BusinessException("人员未配置渠道身份: person_id=" + personId + ", channel=" + channelType.name().toLowerCase());
        }
        Long accountId = selectDefaultAccount(tenantCode, channelType);
        if (accountId == null) {
            throw new BusinessException("租户未配置可用发送账号: channel=" + channelType.name().toLowerCase());
        }
        ChannelAccount account = accountMapper.selectById(accountId);
        if (!AccountSelector.isHealthy(account, tenantCode, channelType)) {
            throw new BusinessException("租户没有可用的发送账号: channel=" + channelType.name().toLowerCase());
        }
        SendTarget target = new SendTarget();
        target.setConversationId(null);
        target.setPersonId(personId);
        target.setAccountId(accountId);
        target.setChannelType(channelType);
        target.setChannelConversationId(null);
        target.setReceiverId(identity.getChannelUserId());
        return Collections.singletonList(target);
    }

    private List<SendTarget> resolveTagTarget(String tenantCode, Long tagId, ChannelType channelType, String scope) {
        List<SendTarget> targets = new ArrayList<>();
        if ("all".equals(scope) || "conversation".equals(scope)) {
            List<ConversationTagBinding> convBindings = conversationTagBindingMapper.selectList(
                    new LambdaQueryWrapper<ConversationTagBinding>()
                            .eq(ConversationTagBinding::getTenantCode, tenantCode)
                            .eq(ConversationTagBinding::getTagId, tagId));
            for (ConversationTagBinding binding : convBindings) {
                targets.addAll(resolveConversationTarget(tenantCode, binding.getConversationId(), channelType,
                        Collections.singletonMap("strategy", "primary")));
            }
        }
        if ("all".equals(scope) || "person".equals(scope)) {
            List<PersonTagBinding> personBindings = personTagBindingMapper.selectList(
                    new LambdaQueryWrapper<PersonTagBinding>()
                            .eq(PersonTagBinding::getTenantCode, tenantCode)
                            .eq(PersonTagBinding::getTagId, tagId));
            for (PersonTagBinding binding : personBindings) {
                targets.addAll(resolvePersonTarget(tenantCode, binding.getPersonId(), channelType));
            }
        }
        return targets;
    }

    private List<SendTarget> resolveRuleTarget(String tenantCode, Map<String, Object> targetValue,
                                               ChannelType channelType, Map<String, Object> channelStrategy) {
        RuleTargetValue ruleTarget = parseRuleTargetValue(targetValue);
        List<RuleMatchResult> results = new ArrayList<>();
        for (RuleCondition condition : ruleTarget.getRules()) {
            results.add(evaluateRuleCondition(tenantCode, condition));
        }
        Set<Long> personIds = combinePersonIds(ruleTarget.getLogic(), results);
        Set<Long> conversationIds = combineConversationIds(ruleTarget.getLogic(), results);

        List<SendTarget> targets = new ArrayList<>();
        for (Long personId : personIds) {
            targets.addAll(resolvePersonTarget(tenantCode, personId, channelType));
        }
        for (Long conversationId : conversationIds) {
            targets.addAll(resolveConversationTarget(tenantCode, conversationId, channelType, channelStrategy));
        }
        return targets;
    }

    private RuleTargetValue parseRuleTargetValue(Map<String, Object> targetValue) {
        if (targetValue == null) {
            throw new BusinessException("rule 目标缺少 rules");
        }
        Object rulesObj = targetValue.get("rules");
        if (!(rulesObj instanceof List) || ((List<?>) rulesObj).isEmpty()) {
            throw new BusinessException("rule 目标缺少 rules");
        }
        List<RuleCondition> rules = new ArrayList<>();
        for (Object item : (List<?>) rulesObj) {
            if (!(item instanceof Map)) {
                throw new BusinessException("rule 条件格式错误");
            }
            Map<?, ?> map = (Map<?, ?>) item;
            String field = toRuleString(map.get("field"));
            String operator = toRuleString(map.get("operator"));
            if (!StringUtils.hasText(field) || !StringUtils.hasText(operator)) {
                throw new BusinessException("rule 条件缺少 field 或 operator");
            }
            rules.add(new RuleCondition(field, operator, map.get("value")));
        }
        String logic = toRuleString(targetValue.getOrDefault("logic", "and"));
        if (!"and".equals(logic) && !"or".equals(logic)) {
            throw new BusinessException("rule 逻辑仅支持 and 或 or");
        }
        return new RuleTargetValue(rules, logic);
    }

    private RuleMatchResult evaluateRuleCondition(String tenantCode, RuleCondition condition) {
        switch (condition.getField()) {
            case "tag":
                return evaluateTagCondition(tenantCode, condition);
            case "contract_id":
                return new RuleMatchResult(Collections.emptySet(),
                        evaluateFieldCondition(tenantCode, condition, conversationMapper,
                                Conversation::getTenantCode, Conversation::getId, Conversation::getContractId, this::toLong));
            case "institution_id":
                return new RuleMatchResult(
                        evaluateFieldCondition(tenantCode, condition, personMapper,
                                Person::getTenantCode, Person::getId, Person::getInstitutionId, this::toLong),
                        evaluateFieldCondition(tenantCode, condition, conversationMapper,
                                Conversation::getTenantCode, Conversation::getId, Conversation::getInstitutionId, this::toLong));
            case "supplier_id":
                return new RuleMatchResult(
                        evaluateFieldCondition(tenantCode, condition, personMapper,
                                Person::getTenantCode, Person::getId, Person::getSupplierId, this::toLong),
                        evaluateFieldCondition(tenantCode, condition, conversationMapper,
                                Conversation::getTenantCode, Conversation::getId, Conversation::getSupplierId, this::toLong));
            case "person_type":
                return new RuleMatchResult(
                        evaluateFieldCondition(tenantCode, condition, personMapper,
                                Person::getTenantCode, Person::getId, Person::getPersonType, this::toInteger),
                        Collections.emptySet());
            case "channel_type":
                return new RuleMatchResult(Collections.emptySet(),
                        evaluateFieldCondition(tenantCode, condition, conversationMapper,
                                Conversation::getTenantCode, Conversation::getId, Conversation::getChannelType, this::toLowerCaseString));
            default:
                throw new BusinessException("rule 不支持的字段: " + condition.getField());
        }
    }

    private <T, R> Set<Long> evaluateFieldCondition(String tenantCode, RuleCondition condition, BaseMapper<T> mapper,
                                                    SFunction<T, String> tenantCodeGetter, SFunction<T, Long> idGetter,
                                                    SFunction<T, R> fieldGetter, Function<Object, R> converter) {
        List<R> values = normalizeValues(condition.getValue()).stream()
                .map(converter)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        LambdaQueryWrapper<T> wrapper = new LambdaQueryWrapper<T>()
                .eq(tenantCodeGetter, tenantCode);
        if (values.isEmpty()) {
            if ("not_in".equals(condition.getOperator())) {
                return mapper.selectList(wrapper).stream().map(idGetter).collect(Collectors.toSet());
            }
            return Collections.emptySet();
        }
        applyOperator(wrapper, fieldGetter, condition.getOperator(), values);
        return mapper.selectList(wrapper).stream().map(idGetter).collect(Collectors.toSet());
    }

    private <T, R> void applyOperator(LambdaQueryWrapper<T> wrapper, SFunction<T, R> fieldGetter,
                                      String operator, List<R> values) {
        switch (operator) {
            case "eq":
                wrapper.eq(fieldGetter, values.get(0));
                break;
            case "ne":
                wrapper.ne(fieldGetter, values.get(0));
                break;
            case "in":
                wrapper.in(fieldGetter, values);
                break;
            case "not_in":
                wrapper.notIn(fieldGetter, values);
                break;
            default:
                throw new BusinessException("rule 不支持的算子: " + operator);
        }
    }

    private RuleMatchResult evaluateTagCondition(String tenantCode, RuleCondition condition) {
        List<Object> values = normalizeValues(condition.getValue());
        Set<Long> tagIds = new HashSet<>();
        for (Object value : values) {
            if (value instanceof Number) {
                tagIds.add(((Number) value).longValue());
            } else if (value instanceof String) {
                List<Tag> tags = tagMapper.selectList(
                        new LambdaQueryWrapper<Tag>()
                                .eq(Tag::getTenantCode, tenantCode)
                                .eq(Tag::getTagPath, value));
                for (Tag tag : tags) {
                    tagIds.add(tag.getId());
                }
            }
        }
        if (tagIds.isEmpty()) {
            if ("not_in".equals(condition.getOperator())) {
                return new RuleMatchResult(
                        selectAllPersonIds(tenantCode),
                        selectAllConversationIds(tenantCode));
            }
            return new RuleMatchResult(Collections.emptySet(), Collections.emptySet());
        }
        Set<Long> personIds = personTagBindingMapper.selectList(
                new LambdaQueryWrapper<PersonTagBinding>()
                        .eq(PersonTagBinding::getTenantCode, tenantCode)
                        .in(PersonTagBinding::getTagId, tagIds))
                .stream()
                .map(PersonTagBinding::getPersonId)
                .collect(Collectors.toSet());
        Set<Long> conversationIds = conversationTagBindingMapper.selectList(
                new LambdaQueryWrapper<ConversationTagBinding>()
                        .eq(ConversationTagBinding::getTenantCode, tenantCode)
                        .in(ConversationTagBinding::getTagId, tagIds))
                .stream()
                .map(ConversationTagBinding::getConversationId)
                .collect(Collectors.toSet());
        if ("ne".equals(condition.getOperator()) || "not_in".equals(condition.getOperator())) {
            Set<Long> allPersonIds = selectAllPersonIds(tenantCode);
            Set<Long> allConversationIds = selectAllConversationIds(tenantCode);
            allPersonIds.removeAll(personIds);
            allConversationIds.removeAll(conversationIds);
            return new RuleMatchResult(allPersonIds, allConversationIds);
        }
        return new RuleMatchResult(personIds, conversationIds);
    }

    private Set<Long> selectAllPersonIds(String tenantCode) {
        return personMapper.selectList(
                new LambdaQueryWrapper<Person>().eq(Person::getTenantCode, tenantCode))
                .stream()
                .map(Person::getId)
                .collect(Collectors.toSet());
    }

    private Set<Long> selectAllConversationIds(String tenantCode) {
        return conversationMapper.selectList(
                new LambdaQueryWrapper<Conversation>().eq(Conversation::getTenantCode, tenantCode))
                .stream()
                .map(Conversation::getId)
                .collect(Collectors.toSet());
    }

    private Set<Long> combinePersonIds(String logic, List<RuleMatchResult> results) {
        return "or".equals(logic) ? unionPersonIds(results) : intersectPersonIds(results);
    }

    private Set<Long> unionPersonIds(List<RuleMatchResult> results) {
        Set<Long> ids = new HashSet<>();
        for (RuleMatchResult result : results) {
            ids.addAll(result.getPersonIds());
        }
        return ids;
    }

    private Set<Long> intersectPersonIds(List<RuleMatchResult> results) {
        Set<Long> ids = null;
        for (RuleMatchResult result : results) {
            if (!result.getPersonIds().isEmpty()) {
                if (ids == null) {
                    ids = new HashSet<>(result.getPersonIds());
                } else {
                    ids.retainAll(result.getPersonIds());
                }
            }
        }
        return ids == null ? Collections.emptySet() : ids;
    }

    private Set<Long> combineConversationIds(String logic, List<RuleMatchResult> results) {
        return "or".equals(logic) ? unionConversationIds(results) : intersectConversationIds(results);
    }

    private Set<Long> unionConversationIds(List<RuleMatchResult> results) {
        Set<Long> ids = new HashSet<>();
        for (RuleMatchResult result : results) {
            ids.addAll(result.getConversationIds());
        }
        return ids;
    }

    private Set<Long> intersectConversationIds(List<RuleMatchResult> results) {
        Set<Long> ids = null;
        for (RuleMatchResult result : results) {
            if (!result.getConversationIds().isEmpty()) {
                if (ids == null) {
                    ids = new HashSet<>(result.getConversationIds());
                } else {
                    ids.retainAll(result.getConversationIds());
                }
            }
        }
        return ids == null ? Collections.emptySet() : ids;
    }

    private List<Object> normalizeValues(Object value) {
        if (value == null) {
            return Collections.emptyList();
        }
        if (value instanceof List) {
            return new ArrayList<>((List<?>) value);
        }
        return Collections.singletonList(value);
    }

    private Long toLong(Object value) {
        if (value instanceof Number) {
            return ((Number) value).longValue();
        }
        if (value instanceof String) {
            try {
                return Long.parseLong((String) value);
            } catch (NumberFormatException e) {
                return null;
            }
        }
        return null;
    }

    private Integer toInteger(Object value) {
        if (value instanceof Number) {
            return ((Number) value).intValue();
        }
        if (value instanceof String) {
            try {
                return Integer.parseInt((String) value);
            } catch (NumberFormatException e) {
                return null;
            }
        }
        return null;
    }

    private String toLowerCaseString(Object value) {
        String str = toRuleString(value);
        return str != null ? str.toLowerCase() : null;
    }

    private String toRuleString(Object value) {
        return value != null ? String.valueOf(value) : null;
    }

    private Long selectSendAccount(String tenantCode, Long conversationId, ChannelType channelType,
                                   Map<String, Object> channelStrategy) {
        List<Long> accounts = accountSelector.selectSendAccounts(tenantCode, conversationId, channelType,
                extractStrategy(channelStrategy));
        return CollectionUtils.isEmpty(accounts) ? null : accounts.get(0);
    }

    private Long selectDefaultAccount(String tenantCode, ChannelType channelType) {
        List<ChannelAccount> accounts = accountMapper.selectList(
                new LambdaQueryWrapper<ChannelAccount>()
                        .eq(ChannelAccount::getTenantCode, tenantCode)
                        .eq(ChannelAccount::getChannelType, channelType.name().toLowerCase())
                        .eq(ChannelAccount::getStatus, STATUS_ENABLED));
        return CollectionUtils.isEmpty(accounts) ? null : accounts.get(0).getId();
    }

    private String extractChannelType(Map<String, Object> channelStrategy) {
        if (channelStrategy == null) {
            return "wecom";
        }
        Object value = channelStrategy.get("channel_type");
        if (value == null) {
            value = channelStrategy.get("channelType");
        }
        return value != null ? String.valueOf(value) : "wecom";
    }

    private String extractStrategy(Map<String, Object> channelStrategy) {
        if (channelStrategy == null) {
            return "primary";
        }
        Object value = channelStrategy.get("strategy");
        return value != null ? String.valueOf(value) : "primary";
    }

    private ChannelType parseChannelType(String channelTypeStr) {
        if (!StringUtils.hasText(channelTypeStr)) {
            return ChannelType.WECOM;
        }
        try {
            return ChannelType.valueOf(channelTypeStr.toUpperCase());
        } catch (IllegalArgumentException e) {
            throw new BusinessException("不支持的渠道类型: " + channelTypeStr);
        }
    }

    private RetryConfig loadRetryConfig(String tenantCode, SendRequest request) {
        String channelTypeStr = extractChannelType(request.getChannelStrategy());
        PolicyGroup policyGroup = loadPolicyGroup(tenantCode, channelTypeStr);
        if (policyGroup == null) {
            return null;
        }
        PolicyGroupConfig config = PolicyGroupConfig.parseJson(policyGroup.getPolicies());
        return config.getRetryConfig().orElse(null);
    }

    private List<SendRecord> createRecords(String taskId, String tenantCode, SendRequest request,
                                           List<SendTarget> targets, RetryConfig retryConfig) {
        List<SendRecord> records = new ArrayList<>();
        int maxRetry = retryConfig != null ? retryConfig.getMaxRetry() : 0;
        for (SendTarget target : targets) {
            SendRecord record = new SendRecord();
            record.setTaskId(taskId);
            record.setTenantCode(tenantCode);
            record.setTargetType(request.getTargetType());
            record.setTargetValue(JsonUtils.toJson(request.getTargetValue()));
            record.setConversationId(target.getConversationId());
            record.setPersonId(target.getPersonId());
            record.setAccountId(target.getAccountId());
            record.setChannelType(target.getChannelType().name().toLowerCase());
            record.setChannelStrategy(JsonUtils.toJson(request.getChannelStrategy()));
            record.setContentType(request.getContentType());
            record.setContentBody(JsonUtils.toJson(request.getContent()));
            record.setCallbackUrl(request.getCallbackUrl());
            record.setSendStatus(SEND_STATUS_PENDING);
            record.setRetryCount(0);
            record.setMaxRetry(maxRetry);
            record.setSendTime(LocalDateTime.now());
            sendRecordMapper.insert(record);
            records.add(record);
        }
        return records;
    }

    private List<SendRecord> createTestRecords(String taskId, String tenantCode, SendTestRequest request,
                                               List<SendTarget> targets) {
        List<SendRecord> records = new ArrayList<>();
        for (SendTarget target : targets) {
            SendRecord record = new SendRecord();
            record.setTaskId(taskId);
            record.setTenantCode(tenantCode);
            record.setTargetType(request.getTargetType());
            record.setTargetValue(JsonUtils.toJson(request.getTargetValue()));
            record.setConversationId(target.getConversationId());
            record.setPersonId(target.getPersonId());
            record.setAccountId(target.getAccountId());
            record.setChannelType(target.getChannelType().name().toLowerCase());
            record.setChannelStrategy(JsonUtils.toJson(request.getChannelStrategy()));
            record.setContentType(request.getContentType());
            record.setContentBody(JsonUtils.toJson(request.getContent()));
            record.setCallbackUrl(request.getCallbackUrl());
            record.setSendStatus(SEND_STATUS_SUCCESS);
            record.setChannelResult(JsonUtils.toJson(Collections.singletonMap("simulated", true)));
            record.setSendTime(LocalDateTime.now());
            record.setFinishTime(LocalDateTime.now());
            sendRecordMapper.insert(record);
            records.add(record);
        }
        return records;
    }

    @Transactional
    public void dispatchRecord(SendRecord record) {

        ChannelType channelType = parseChannelType(record.getChannelType());
        ChannelAdapter adapter = adapterRegistry.getAdapter(channelType);
        if (adapter == null) {
            record.setSendStatus(SEND_STATUS_FAILED);
            record.setChannelResult(JsonUtils.toJson(Collections.singletonMap("error", "渠道适配器未实现: " + channelType)));
            record.setFinishTime(LocalDateTime.now());
            sendRecordMapper.updateById(record);
            return;
        }

        PolicyCheckResult policyResult = checkPolicy(record);
        if (policyResult != null && !policyResult.isAllowed()) {
            record.setSendStatus(SEND_STATUS_FAILED);
            Map<String, Object> resultMap = new HashMap<>();
            resultMap.put("policy_rejected", true);
            resultMap.put("reason", policyResult.getReasonCode());
            resultMap.put("message", policyResult.getReasonMessage());
            record.setChannelResult(JsonUtils.toJson(resultMap));
            record.setFinishTime(LocalDateTime.now());
            sendRecordMapper.updateById(record);
            return;
        }

        List<Long> candidateAccountIds = selectSendAccountCandidates(record, channelType);
        if (candidateAccountIds.isEmpty()) {
            record.setSendStatus(SEND_STATUS_FAILED);
            record.setChannelResult(JsonUtils.toJson(Collections.singletonMap("error", "no_healthy_account")));
            record.setFinishTime(LocalDateTime.now());
            sendRecordMapper.updateById(record);
            return;
        }

        List<Map<String, Object>> attempts = new ArrayList<>();
        boolean success = false;
        SendMessageResult lastResult = null;

        for (Long accountId : candidateAccountIds) {
            record.setAccountId(accountId);
            record.setSendStatus(SEND_STATUS_SENDING);
            sendRecordMapper.updateById(record);

            ChannelAccount account = accountMapper.selectById(accountId);
            SendMessageResult result = doSendAttempt(record, adapter, account);
            lastResult = result;

            attempts.add(buildAttemptResult(accountId, result));

            if (result != null && result.isSuccess()) {
                success = true;
                break;
            }

            markAccountUnhealthy(account);
        }

        record.setFinishTime(LocalDateTime.now());
        if (success) {
            record.setSendStatus(SEND_STATUS_SUCCESS);
            record.setChannelResult(JsonUtils.toJson(buildSuccessResult(lastResult, record.getAccountId(), attempts)));
        } else {
            record.setSendStatus(SEND_STATUS_FAILED);
            record.setChannelResult(JsonUtils.toJson(buildAllAttemptsFailureResult(attempts, policyResult)));
        }
        sendRecordMapper.updateById(record);
    }

    private List<Long> selectSendAccountCandidates(SendRecord record, ChannelType channelType) {
        if (record.getConversationId() == null) {
            return record.getAccountId() != null ? Collections.singletonList(record.getAccountId()) : Collections.emptyList();
        }
        String preferredStrategy = extractStrategy(parseChannelStrategy(record.getChannelStrategy()));
        return accountSelector.selectSendAccounts(record.getTenantCode(), record.getConversationId(), channelType, preferredStrategy);
    }

    private Map<String, Object> parseChannelStrategy(String strategyJson) {
        if (!StringUtils.hasText(strategyJson)) {
            return null;
        }
        Map<?, ?> map = JsonUtils.fromJson(strategyJson, Map.class);
        if (map == null) {
            return null;
        }
        Map<String, Object> result = new HashMap<>();
        for (Map.Entry<?, ?> entry : map.entrySet()) {
            result.put(String.valueOf(entry.getKey()), entry.getValue());
        }
        return result;
    }

    private SendMessageResult doSendAttempt(SendRecord record, ChannelAdapter adapter, ChannelAccount account) {
        SendMessageRequest messageRequest = new SendMessageRequest();
        messageRequest.setConversationId(record.getConversationId() != null ? String.valueOf(record.getConversationId()) : null);
        messageRequest.setReceiverId(record.getPersonId() != null ? String.valueOf(record.getPersonId()) : null);
        messageRequest.setContentType(record.getContentType());
        messageRequest.setContent(JsonUtils.fromJson(record.getContentBody(), Map.class));

        Map<String, Object> extra = new HashMap<>();
        if (account != null && StringUtils.hasText(account.getAccountId())) {
            extra.put("account_id", account.getAccountId());
        }
        extra.put("record_id", record.getId());
        extra.put("tenant_code", record.getTenantCode());
        messageRequest.setExtra(extra);

        try {
            return adapter.sendMessage(messageRequest);
        } catch (Exception e) {
            log.warn("发送消息失败: task_id={}, record_id={}, account_id={}", record.getTaskId(), record.getId(),
                    account != null ? account.getId() : null, e);
            SendMessageResult result = new SendMessageResult();
            result.setSuccess(false);
            result.setMessage(e.getMessage());
            return result;
        }
    }

    private void markAccountUnhealthy(ChannelAccount account) {
        if (account == null || account.getAccountType() == null || account.getAccountType() != 3) {
            return;
        }
        account.setRiskLevel(2);
        account.setOnlineStatus(0);
        accountMapper.updateById(account);
    }

    private Map<String, Object> buildAttemptResult(Long accountId, SendMessageResult result) {
        Map<String, Object> map = new HashMap<>();
        map.put("account_id", accountId);
        if (result != null) {
            map.put("success", result.isSuccess());
            map.put("message", result.getMessage());
        } else {
            map.put("success", false);
            map.put("message", "unknown error");
        }
        return map;
    }

    private Map<String, Object> buildSuccessResult(SendMessageResult result, Long accountId, List<Map<String, Object>> attempts) {
        Map<String, Object> map = new HashMap<>();
        map.put("success", true);
        map.put("account_id", accountId);
        if (result != null) {
            map.put("channel_message_id", result.getChannelMessageId());
            map.put("message", result.getMessage());
        }
        map.put("attempts", attempts);
        return map;
    }

    private Map<String, Object> buildAllAttemptsFailureResult(List<Map<String, Object>> attempts, PolicyCheckResult policyResult) {
        Map<String, Object> map = new HashMap<>();
        map.put("error", "all_accounts_failed");
        map.put("attempts", attempts);
        appendRetryPlan(map, policyResult);
        return map;
    }

    private PolicyCheckResult checkPolicy(SendRecord record) {
        PolicyGroup policyGroup = loadPolicyGroup(record.getTenantCode(), record.getChannelType());
        if (policyGroup == null) {
            return null;
        }
        PolicyGroupConfig config = PolicyGroupConfig.parseJson(policyGroup.getPolicies());
        return policyEngine.check(record, config);
    }

    private PolicyGroup loadPolicyGroup(String tenantCode, String channelType) {
        LambdaQueryWrapper<PolicyGroup> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(PolicyGroup::getTenantCode, tenantCode)
                .eq(PolicyGroup::getStatus, STATUS_ENABLED);
        if (StringUtils.hasText(channelType)) {
            wrapper.eq(PolicyGroup::getChannelType, channelType);
        } else {
            wrapper.and(w -> w.isNull(PolicyGroup::getChannelType).or().eq(PolicyGroup::getChannelType, ""));
        }
        wrapper.orderByDesc(PolicyGroup::getId).last("LIMIT 1");
        PolicyGroup group = policyGroupMapper.selectOne(wrapper);
        if (group != null) {
            return group;
        }

        group = policyGroupMapper.selectOne(
                new LambdaQueryWrapper<PolicyGroup>()
                        .eq(PolicyGroup::getTenantCode, tenantCode)
                        .eq(PolicyGroup::getStatus, STATUS_ENABLED)
                        .and(w -> w.isNull(PolicyGroup::getChannelType).or().eq(PolicyGroup::getChannelType, ""))
                        .orderByDesc(PolicyGroup::getId)
                        .last("LIMIT 1"));
        if (group != null) {
            return group;
        }

        return policyGroupMapper.selectOne(
                new LambdaQueryWrapper<PolicyGroup>()
                        .eq(PolicyGroup::getGroupCode, "default")
                        .eq(PolicyGroup::getStatus, STATUS_ENABLED)
                        .orderByDesc(PolicyGroup::getId)
                        .last("LIMIT 1"));
    }

    private void appendRetryPlan(Map<String, Object> map, PolicyCheckResult policyResult) {
        if (policyResult != null && policyResult.getRetryConfig() != null) {
            map.put("retry_plan", policyResult.getRetryConfig());
            log.info("发送失败，按策略计划重试: {}", policyResult.getRetryConfig().getIntervals());
        }
    }

    private SendRecordResponse toResponse(SendRecord record) {
        SendRecordResponse response = new SendRecordResponse();
        response.setId(record.getId());
        response.setTaskId(record.getTaskId());
        response.setTargetType(record.getTargetType());
        response.setTargetValue(record.getTargetValue());
        response.setConversationId(record.getConversationId());
        response.setPersonId(record.getPersonId());
        response.setAccountId(record.getAccountId());
        response.setChannelType(record.getChannelType());
        response.setContentType(record.getContentType());
        response.setSendStatus(record.getSendStatus());
        response.setChannelResult(record.getChannelResult());
        response.setContentBody(record.getContentBody());
        response.setCallbackUrl(record.getCallbackUrl());
        response.setSendTime(record.getSendTime());
        response.setFinishTime(record.getFinishTime());
        response.setCreateTime(record.getCreateTime());
        return response;
    }

    private static class RuleCondition {

        private final String field;
        private final String operator;
        private final Object value;

        RuleCondition(String field, String operator, Object value) {
            this.field = field;
            this.operator = operator;
            this.value = value;
        }

        public String getField() {
            return field;
        }

        public String getOperator() {
            return operator;
        }

        public Object getValue() {
            return value;
        }
    }

    private static class RuleTargetValue {

        private final List<RuleCondition> rules;
        private final String logic;

        RuleTargetValue(List<RuleCondition> rules, String logic) {
            this.rules = rules;
            this.logic = logic;
        }

        public List<RuleCondition> getRules() {
            return rules;
        }

        public String getLogic() {
            return logic;
        }
    }

    private static class RuleMatchResult {

        private final Set<Long> personIds;
        private final Set<Long> conversationIds;

        RuleMatchResult(Set<Long> personIds, Set<Long> conversationIds) {
            this.personIds = personIds;
            this.conversationIds = conversationIds;
        }

        public Set<Long> getPersonIds() {
            return personIds;
        }

        public Set<Long> getConversationIds() {
            return conversationIds;
        }
    }

}
