package com.fs.live.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.fs.common.constant.LiveKeysConstant;
import com.fs.common.core.domain.R;
import com.fs.common.core.redis.RedisCache;
import com.fs.common.utils.StringUtils;
import com.fs.hisStore.domain.FsUserScrm;
import com.fs.hisStore.service.IFsUserScrmService;
import com.fs.live.domain.LiveOrder;
import com.fs.live.domain.LiveOrderTipConfig;
import com.fs.live.mapper.LiveGoodsMapper;
import com.fs.live.service.ILiveOrderTipConfigService;
import com.fs.live.service.ILiveOrderTipService;
import com.fs.live.service.LiveAppWebSocketNotifyService;
import com.fs.live.vo.LiveGoodsVo;
import com.fs.live.vo.LiveOrderTipVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

/**
 * 直播下单动效提示业务实现。
 *
 * <p><b>高并发优化要点：</b></p>
 * <ol>
 *   <li>限流：使用 Redis ZSet 做滑动窗口 + Lua 脚本原子操作，避免集群竞态。</li>
 *   <li>真实订单去重：以 orderId 作为 Redis SETNX key（过期时间=2 倍窗口），防止回调重复推送。</li>
 *   <li>配置缓存：全局配置采用 5 分钟 TTL 缓存，避免热点读 DB。</li>
 *   <li>商品候选池：直播间商品列表缓存 30 秒，避免假订单每次都查 DB。</li>
 *   <li>推送异步化：不阻塞支付事务，通过 @Async 或调用方自行异步。</li>
 *   <li>消息载体复用现有 WebSocket 广播通道，单直播间 1 万人靠 fs-live-app 的消息队列/广播线程池承载。</li>
 * </ol>
 */
@Slf4j
@Service
public class LiveOrderTipServiceImpl implements ILiveOrderTipService {

    @Autowired
    private ILiveOrderTipConfigService tipConfigService;

    @Autowired
    private LiveAppWebSocketNotifyService wsNotifyService;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private LiveGoodsMapper liveGoodsMapper;

    @Autowired
    private IFsUserScrmService fsUserService;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /** 内置默认昵称库（前端/后台未配置时兜底使用） */
    private static final List<String> DEFAULT_NICKNAMES = Collections.unmodifiableList(Arrays.asList(
            "养生小达人", "健康路上", "乐活青年", "品味生活", "星月同辉",
            "简单快乐", "暖阳微风", "一枕清风", "云淡风轻", "微笑向暖",
            "岁月静好", "小确幸", "明月入怀", "山水有相逢", "梅子黄时雨",
            "温柔星光", "初夏时光", "蒲公英的约定", "晚风轻拂", "清风明月"));

    /**
     * 原子滑动窗口限流脚本：
     * KEYS[1] = 窗口键 (ZSet)
     * ARGV[1] = 当前时间戳（毫秒）
     * ARGV[2] = 窗口起点（毫秒）= now - windowMs
     * ARGV[3] = 最大允许数量
     * ARGV[4] = 唯一 member（避免 score 相同覆盖）
     * ARGV[5] = 键 TTL 秒
     * 返回：1=允许 0=拒绝
     */
    private static final String RATE_LIMIT_LUA = ""
            + "redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, ARGV[2])\n"
            + "local cnt = redis.call('ZCARD', KEYS[1])\n"
            + "if tonumber(cnt) >= tonumber(ARGV[3]) then return 0 end\n"
            + "redis.call('ZADD', KEYS[1], ARGV[1], ARGV[4])\n"
            + "redis.call('EXPIRE', KEYS[1], ARGV[5])\n"
            + "return 1";

    @Override
    @Async
    public void publishRealOrderTip(LiveOrder order) {
        if (order == null || order.getLiveId() == null || order.getLiveId() <= 0) {
            return;
        }
        try {
            LiveOrderTipConfig cfg = tipConfigService.getEffectiveConfig();
            if (!isEnabled(cfg)) {
                return;
            }

            // 去重（同一订单只能推一次，TTL = 2 * window，足够覆盖支付回调重试）
            String dedupKey = String.format(LiveKeysConstant.LIVE_ORDER_TIP_REAL_DEDUP,
                    order.getOrderId() == null ? UUID.randomUUID().toString() : order.getOrderId());
            boolean first = redisCache.setIfAbsent(dedupKey, "1",
                    Math.max(60, safeInt(cfg.getWindowSec(), 60) * 2L), TimeUnit.SECONDS);
            if (!first) {
                return;
            }

            // 限流
            if (!tryAcquire(order.getLiveId(), cfg)) {
                log.debug("[订单动效] 限流丢弃真实订单, liveId={}, orderId={}", order.getLiveId(), order.getOrderId());
                return;
            }

            // 取商品名称与昵称
            LiveGoodsVo goods = findGoodsForRealOrder(order);
            if (goods == null || StringUtils.isEmpty(goods.getProductName())) {
                return;
            }
            String nickname = resolveNicknameForRealOrder(order, cfg);

            long now = System.currentTimeMillis();
            long orderTs = resolveOrderTs(order, now);
            long ageSec = Math.max(0, (now - orderTs) / 1000);
            String content = renderContent(nickname, goods.getProductName(), ageSec);

            LiveOrderTipVo vo = LiveOrderTipVo.builder()
                    .liveId(order.getLiveId())
                    .nickname(nickname)
                    .goodsId(goods.getGoodsId())
                    .productName(goods.getProductName())
                    .imgUrl(goods.getImgUrl())
                    .content(content)
                    .orderTime(orderTs)
                    .fake(0)
                    .build();

            broadcast(vo);
        } catch (Exception e) {
            log.error("[订单动效] 推送真实订单异常, orderId={}, liveId={}",
                    order.getOrderId(), order.getLiveId(), e);
        }
    }

    @Override
    public LiveOrderTipVo publishFakeOrderTip(Long liveId) {
        if (liveId == null || liveId <= 0) {
            return null;
        }
        try {
            LiveOrderTipConfig cfg = tipConfigService.getEffectiveConfig();
            if (!isEnabled(cfg) || !isFakeEnabled(cfg)) {
                return null;
            }

            // 假订单自身的最小间隔（二级节流，避免定时任务短周期内重复生成）
            int minInterval = safeInt(cfg.getFakeMinIntervalSec(), 20);
            if (minInterval > 0) {
                String fakeKey = String.format(LiveKeysConstant.LIVE_ORDER_TIP_FAKE_LAST, liveId);
                boolean allowFake = redisCache.setIfAbsent(fakeKey, "1", minInterval, TimeUnit.SECONDS);
                if (!allowFake) {
                    return null;
                }
            }

            // 共享全局窗口限流（真假数据共同计数，保证总频率不超阈值）
            if (!tryAcquire(liveId, cfg)) {
                return null;
            }

            LiveGoodsVo goods = pickGoodsForLive(liveId, cfg);
            if (goods == null || StringUtils.isEmpty(goods.getProductName())) {
                return null;
            }
            String nickname = pickNicknameFromLibrary(cfg);

            long now = System.currentTimeMillis();
            // 假订单时间在 [0, justNowSeconds*2] 秒前随机，更像真实发生
            long justNow = safeInt(cfg.getJustNowSeconds(), 30);
            long ageSec = ThreadLocalRandom.current().nextLong(0, Math.max(1, justNow * 2));
            String content = renderContent(nickname, goods.getProductName(), ageSec);

            LiveOrderTipVo vo = LiveOrderTipVo.builder()
                    .liveId(liveId)
                    .nickname(nickname)
                    .goodsId(goods.getGoodsId())
                    .productName(goods.getProductName())
                    .imgUrl(goods.getImgUrl())
                    .content(content)
                    .orderTime(now - ageSec * 1000)
                    .fake(1)
                    .build();

            broadcast(vo);
            return vo;
        } catch (Exception e) {
            log.error("[订单动效] 生成假订单异常, liveId={}", liveId, e);
            return null;
        }
    }

    @Override
    public String renderContent(String nickname, String productName, long ageSeconds) {
        LiveOrderTipConfig cfg = tipConfigService.getEffectiveConfig();
        int justNow = safeInt(cfg.getJustNowSeconds(), 30);

        String tpl = (ageSeconds <= justNow)
                ? (StringUtils.isEmpty(cfg.getTemplateJustNow())
                    ? "[用户昵称] 刚刚下单了 [商品名称]" : cfg.getTemplateJustNow())
                : (StringUtils.isEmpty(cfg.getTemplateAgo())
                    ? "[时间]前，[用户昵称] 购买了 [商品名称]" : cfg.getTemplateAgo());

        String timeText = humanizeTime(ageSeconds);
        return tpl.replace("[用户昵称]", nickname == null ? "" : nickname)
                .replace("[商品名称]", productName == null ? "" : productName)
                .replace("[时间]", timeText);
    }

    // =============================================================================
    // 内部辅助方法
    // =============================================================================

    /** 使用 Redis Lua 原子滑动窗口限流，超过 rateMax 则拒绝 */
    private boolean tryAcquire(Long liveId, LiveOrderTipConfig cfg) {
        int max = safeInt(cfg.getRateMaxPerMinute(), 3);
        int windowSec = safeInt(cfg.getWindowSec(), 60);
        if (max <= 0 || windowSec <= 0) {
            return false;
        }
        String key = String.format(LiveKeysConstant.LIVE_ORDER_TIP_WINDOW, liveId);
        long now = System.currentTimeMillis();
        long start = now - windowSec * 1000L;
        String member = now + "-" + UUID.randomUUID();

        try {
            DefaultRedisScript<Long> script = new DefaultRedisScript<>(RATE_LIMIT_LUA, Long.class);
            // 使用 StringRedisTemplate，避免 GenericJackson2JsonRedisSerializer 把 ARGV 序列化成带引号字符串，
            // 导致 Redis 报错：ERR min or max is not a float
            Long result = stringRedisTemplate.execute(
                    script,
                    Collections.singletonList(key),
                    String.valueOf(now),
                    String.valueOf(start),
                    String.valueOf(max),
                    member,
                    String.valueOf(windowSec + 10)
            );
            return result != null && result == 1L;
        } catch (Exception e) {
            // 降级：出错时放行，保证功能可用（宁可少量刷屏也不能导致阻塞）
            log.warn("[订单动效] 限流脚本执行异常，本次放行, liveId={}, err={}", liveId, e.getMessage());
            return true;
        }
    }

    /** 当前窗口内的剩余容量，供 scheduler 判断是否需要补充假订单 */
    public int remainingQuota(Long liveId, LiveOrderTipConfig cfg) {
        int max = safeInt(cfg.getRateMaxPerMinute(), 3);
        int windowSec = safeInt(cfg.getWindowSec(), 60);
        String key = String.format(LiveKeysConstant.LIVE_ORDER_TIP_WINDOW, liveId);
        long now = System.currentTimeMillis();
        long start = now - windowSec * 1000L;
        try {
            stringRedisTemplate.opsForZSet().removeRangeByScore(key, 0, start);
            Long cnt = stringRedisTemplate.opsForZSet().zCard(key);
            int used = cnt == null ? 0 : cnt.intValue();
            return Math.max(0, max - used);
        } catch (Exception e) {
            log.warn("[订单动效] 查询剩余配额异常, liveId={}, err={}", liveId, e.getMessage());
            return 1;
        }
    }

    /**
     * 广播到 fs-live-app（通过 HTTP 调 WebSocket 广播接口）。
     *
     * <p>消息结构与现有 WebSocket 广播保持一致：<code>R.ok().put("data", SendMsgVo)</code>，
     * 前端按 <code>cmd=orderTip</code> 识别订单动效提示。</p>
     */
    private void broadcast(LiveOrderTipVo vo) {
        JSONObject sendMsg = new JSONObject();
        sendMsg.put("liveId", vo.getLiveId());
        sendMsg.put("cmd", "orderTip");
        sendMsg.put("msg", vo.getContent());
        sendMsg.put("data", JSONObject.toJSONString(vo));
        sendMsg.put("on", true);
        String fullJson = JSONObject.toJSONString(R.ok().put("data", sendMsg));
        // 走订单动效专用入口：内部进入 fs-live-app 的优先级消息队列，具备背压保护，
        // 高峰（1万人+）下不阻塞调用方，避免 WebSocket 广播线程反压到业务主流程。
        wsNotifyService.broadcastOrderTipToLive(vo.getLiveId(), fullJson);
    }

    /** 真实订单优先从订单的 itemJson 里取商品；没有则查 live_goods 展示中的商品 */
    private LiveGoodsVo findGoodsForRealOrder(LiveOrder order) {
        // 1) 按 productId 定位（准确）
        if (order.getProductId() != null && order.getLiveId() != null) {
            try {
                com.fs.live.domain.LiveGoods lg = liveGoodsMapper.selectLiveGoodsByProductId(
                        order.getLiveId(), order.getProductId());
                if (lg != null) {
                    LiveGoodsVo vo = liveGoodsMapper.selectLiveGoodsVoByGoodsId(lg.getGoodsId());
                    if (vo != null) return vo;
                }
            } catch (Exception ignore) {
                // 兜底继续走下方
            }
        }
        // 2) 解析 itemJson 取 productName
        if (StringUtils.isNotEmpty(order.getItemJson())) {
            try {
                JSONObject obj = JSONObject.parseObject(order.getItemJson());
                String productName = obj.getString("productName");
                if (StringUtils.isNotEmpty(productName)) {
                    LiveGoodsVo vo = new LiveGoodsVo();
                    vo.setGoodsId(null);
                    vo.setProductName(productName);
                    vo.setImgUrl(obj.getString("image"));
                    return vo;
                }
            } catch (Exception ignore) {}
        }
        // 3) 随机直播间展示商品
        return pickGoodsForLive(order.getLiveId(), tipConfigService.getEffectiveConfig());
    }

    /** 为假订单/兜底选择一个可用商品（随机） */
    private LiveGoodsVo pickGoodsForLive(Long liveId, LiveOrderTipConfig cfg) {
        String scope = cfg == null || StringUtils.isEmpty(cfg.getGoodsScope()) ? "live" : cfg.getGoodsScope();
        // custom 指定商品池
        if ("custom".equalsIgnoreCase(scope) && StringUtils.isNotEmpty(cfg.getCustomGoodsIds())) {
            List<Long> ids = parseLongList(cfg.getCustomGoodsIds());
            if (!ids.isEmpty()) {
                Long goodsId = ids.get(ThreadLocalRandom.current().nextInt(ids.size()));
                LiveGoodsVo vo = liveGoodsMapper.selectLiveGoodsVoByGoodsId(goodsId);
                if (vo != null) return vo;
            }
        }
        // live/all：从当前直播间在售商品里挑
        List<LiveGoodsVo> pool = getLiveGoodsPool(liveId);
        if (pool == null || pool.isEmpty()) {
            return null;
        }
        return pool.get(ThreadLocalRandom.current().nextInt(pool.size()));
    }

    /** 直播间可用商品缓存池（30 秒 TTL） */
    @SuppressWarnings("unchecked")
    private List<LiveGoodsVo> getLiveGoodsPool(Long liveId) {
        String cacheKey = "live:order:tip:goods:pool:" + liveId;
        Object cached = redisCache.getCacheObject(cacheKey);
        if (cached instanceof List) {
            return (List<LiveGoodsVo>) cached;
        }
        com.fs.live.domain.LiveGoods query = new com.fs.live.domain.LiveGoods();
        query.setLiveId(liveId);
        List<LiveGoodsVo> list = liveGoodsMapper.selectProductListByLiveId(query);
        if (list == null) {
            list = Collections.emptyList();
        } else {
            // 只保留有效的：有商品名
            list = list.stream()
                    .filter(g -> g != null && StringUtils.isNotEmpty(g.getProductName()))
                    .collect(Collectors.toList());
        }
        redisCache.setCacheObject(cacheKey, list, 30, TimeUnit.SECONDS);
        return list;
    }

    private String resolveNicknameForRealOrder(LiveOrder order, LiveOrderTipConfig cfg) {
        String nickname = null;
        try {
            if (order.getUserId() != null) {
                FsUserScrm user = fsUserService.selectFsUserById(Long.valueOf(order.getUserId()));
                if (user != null) {
                    nickname = user.getNickname();
                    if (StringUtils.isEmpty(nickname)) {
                        nickname = user.getPhone();
                    }
                }
            }
        } catch (Exception ignore) {}

        if (StringUtils.isEmpty(nickname)) {
            // 用户信息缺失时，从昵称库随机（避免用户隐私信息泄露）
            return pickNicknameFromLibrary(cfg);
        }
        return maskNickname(nickname, cfg);
    }

    private String pickNicknameFromLibrary(LiveOrderTipConfig cfg) {
        List<String> lib = parseStringList(cfg == null ? null : cfg.getNicknameLibrary());
        if (lib.isEmpty()) {
            lib = DEFAULT_NICKNAMES;
        }
        return lib.get(ThreadLocalRandom.current().nextInt(lib.size()));
    }

    /** 脱敏：如 "张三" -> "张*"; "小明同学" -> "小***" ; 包含字母数字则保留首尾 */
    private String maskNickname(String nickname, LiveOrderTipConfig cfg) {
        if (cfg == null || cfg.getNicknameMask() == null || cfg.getNicknameMask() == 0) {
            return nickname;
        }
        if (nickname == null) return "";
        int len = nickname.codePointCount(0, nickname.length());
        if (len <= 1) return nickname + "*";
        int firstEnd = nickname.offsetByCodePoints(0, 1);
        String head = nickname.substring(0, firstEnd);
        int stars = Math.min(len - 1, 3);
        StringBuilder sb = new StringBuilder(head);
        for (int i = 0; i < stars; i++) {
            sb.append('*');
        }
        return sb.toString();
    }

    private String humanizeTime(long ageSec) {
        if (ageSec < 60) {
            return ageSec + "秒";
        }
        long mins = ageSec / 60;
        if (mins < 60) {
            return mins + "分钟";
        }
        long hours = mins / 60;
        return hours + "小时";
    }

    private long resolveOrderTs(LiveOrder order, long fallback) {
        try {
            if (order.getPayTime() != null) {
                return order.getPayTime().atZone(java.time.ZoneId.systemDefault())
                        .toInstant().toEpochMilli();
            }
            if (order.getCreateTime() != null) {
                return order.getCreateTime().getTime();
            }
        } catch (Exception ignore) {}
        return fallback;
    }

    private static boolean isEnabled(LiveOrderTipConfig cfg) {
        return cfg != null && cfg.getEnabled() != null && cfg.getEnabled() == 1;
    }

    private static boolean isFakeEnabled(LiveOrderTipConfig cfg) {
        return cfg != null && cfg.getFakeEnabled() != null && cfg.getFakeEnabled() == 1;
    }

    private static int safeInt(Integer v, int def) {
        return v == null ? def : v;
    }

    private static List<Long> parseLongList(String raw) {
        if (StringUtils.isEmpty(raw)) return Collections.emptyList();
        List<Long> list = new ArrayList<>();
        for (String part : raw.split("[,，\\s]+")) {
            if (part == null || part.trim().isEmpty()) continue;
            try {
                list.add(Long.parseLong(part.trim()));
            } catch (NumberFormatException ignore) {}
        }
        return list;
    }

    private static List<String> parseStringList(String raw) {
        if (StringUtils.isEmpty(raw)) return Collections.emptyList();
        List<String> list = new ArrayList<>();
        for (String part : raw.split("[,，\\n\\r]+")) {
            if (part == null) continue;
            String p = part.trim();
            if (!p.isEmpty()) {
                list.add(p);
            }
        }
        return list;
    }

}
