#include "RedisStreamMQ.h"
#include "../rocketmq/SimpleProducer.h"

#include <hiredis/hiredis.h>
#include <zlib.h>

#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <unistd.h>
#include <dirent.h>
#include <errno.h>
#include <string.h>
#include <stdarg.h>
#include <ctype.h>

#include <iostream>
#include <fstream>
#include <sstream>
#include <vector>
#include <map>
#include <set>
#include <algorithm>
#include <stdexcept>
#include <ctime>
#include <cstdlib>
#include <cstdio>

#include "../log4cplus/log4.h"
#include "../mms/db_interface.h"

/*
 * lnk20260622 修改：运行时切换版 Redis Streams 后端实现。
 *
 * 本文件只导出 RedisStream_xxx 函数，不再导出：
 *   InitializeProducer / ShutdownAndDestroyProducer
 *   rocketmq_producer_send
 *   InitializeConsumer / ShutdownAndDestroyConsumer
 *   rocketmq_consumer_receive
 *
 * 这些统一入口应在 SimpleProducer.cpp 中按配置文件 [MQ] Backend 转发，
 * 否则同时编译 RocketMQ 和 RedisStream 时会发生重复定义链接错误。
 */

extern std::string G_ROCKETMQ_PRODUCER;     // 保留原配置名；Redis 模式下仅用于日志
extern std::string G_ROCKETMQ_IPPORT;       // Redis 模式下建议填 127.0.0.1:6379[/db]
extern std::string G_MQCONSUMER_IPPORT;     // Redis 模式下消费者连接地址
extern std::string G_ROCKETMQ_CONSUMER;     // Redis consumer group
extern std::string FRONT_INST;


// lnk20260626 修改：Redis 模式复用 RocketMq/ConsumerSecretKey 作为 Redis 密码
extern std::string G_MQCONSUMER_SECRETKEY;

extern int          g_front_seg_index;
extern char         subdir[128];

static const int64_t REDIS_BLOCK_MS = 1000; //XREADGROUP 阻塞等待时间
static const int64_t REDIS_IDLE_CLAIM_MS = 60000;   //消息空闲多久被其他消费者抢占
static const int     REDIS_READ_COUNT = 16; //每次 XREADGROUP 的最大读取数量
static const size_t  GZIP_THRESHOLD_BYTES = 1024;   //超过这个大小的消息才压缩
static const char*   SPOOL_DIR = "/FeProject/data/redisstream_spool";   //本地缓存目录

// lnk20260626 修改：XADD 退避与 Redis 内存保护
static const int64_t REDIS_XADD_OOM_BACKOFF_MS      = 30000;  // OOM 后 30 秒内不再 XADD
static const int64_t REDIS_XADD_FAIL_BACKOFF_MS     = 5000;   // 普通连接失败后 5 秒退避
static const int64_t REDIS_MEMORY_CHECK_INTERVAL_MS = 5000;   // 每 5 秒检查一次 Redis 内存
static const int     REDIS_MEMORY_HIGH_WATERMARK    = 85;     // Redis 内存超过 85% 后延迟 XADD
static const int64_t REDIS_SPOOL_FLUSH_INTERVAL_MS  = 2000;   // spool 最多 2 秒补发一次
static const int     REDIS_SPOOL_FLUSH_LIMIT        = 50;     // 每次最多补发 50 条，防止恢复后瞬间冲击 Redis

static int64_t now_ms() //当前时间，单位毫秒
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return (int64_t)tv.tv_sec * 1000 + tv.tv_usec / 1000;
}

/*
 * 注意：如果 RocketMQ 原 SimpleProducer.cpp 已经定义了 G_APP_START_MS / G_START_SKEW_MS，
 * 这里不要再定义全局同名变量，避免多后端同时编译时重复定义。
 */
static int64_t redis_app_start_ms()//RedisStream 后端的程序启动时间
{
    static int64_t v = now_ms();
    return v;
}

static const int64_t REDIS_START_SKEW_MS = 1000;

static std::string to_string_i64(int64_t v)
{
    char buf[64];
    snprintf(buf, sizeof(buf), "%lld", (long long)v);
    return std::string(buf);
}

static std::string shell_safe_name(const std::string& s)
{
    std::string r;
    for (size_t i = 0; i < s.size(); ++i) {
        unsigned char c = (unsigned char)s[i];
        if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
            (c >= '0' && c <= '9') || c == '_' || c == '-' || c == '.') {
            r.push_back((char)c);
        } else {
            r.push_back('_');
        }
    }
    return r.empty() ? "empty" : r;
}

static std::string build_stream_name(const std::string& topic, const std::string& env)//构造stream名称，格式：env_shortTopic
{
    if (!topic.empty() && topic.find('%') != std::string::npos) {
        std::string shortTopic = topic.substr(topic.find('%') + 1);
        if (!env.empty()) return env + "_" + shortTopic;
        return shortTopic;
    }

    if (!env.empty()) {
        const std::string prefix = env + "_";
        if (topic.find(prefix) == 0) return topic;
        return prefix + topic;
    }
    return topic;
}

static void ensure_dir(const std::string& dir)//创建目录，如果不存在
{
    if (dir.empty()) return;
    if (access(dir.c_str(), F_OK) == 0) return;
    if (mkdir(dir.c_str(), 0755) != 0 && errno != EEXIST) {
        std::cerr << "[REDIS_STREAM][SPOOL] mkdir failed: " << dir
                  << " errno=" << errno << " " << strerror(errno) << std::endl;
    }
}

static std::string base64_encode(const std::string& input)//编码为 base64
{
    static const char* table =
        "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    std::string out;
    int val = 0;
    int valb = -6;
    for (size_t i = 0; i < input.size(); ++i) {
        unsigned char c = (unsigned char)input[i];
        val = (val << 8) + c;
        valb += 8;
        while (valb >= 0) {
            out.push_back(table[(val >> valb) & 0x3F]);
            valb -= 6;
        }
    }
    if (valb > -6) out.push_back(table[((val << 8) >> (valb + 8)) & 0x3F]);
    while (out.size() % 4) out.push_back('=');
    return out;
}

static std::string gzip_compress(const std::string& input)//压缩字符串，返回 gzip 格式
{
    if (input.empty()) return std::string();

    z_stream zs;
    memset(&zs, 0, sizeof(zs));

    static bool zlib_version_logged = false;
    if (!zlib_version_logged) {
        zlib_version_logged = true;
        std::cerr << "[REDIS_STREAM][ZLIB] compile=" << ZLIB_VERSION
                  << " runtime=" << zlibVersion()
                  << " z_stream_size=" << sizeof(z_stream)
                  << std::endl;
    }

    int ret = deflateInit2(&zs, Z_BEST_SPEED, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
    if (ret != Z_OK) {
        char buf[64];
        snprintf(buf, sizeof(buf), "deflateInit2 failed ret=%d", ret);
        throw std::runtime_error(buf);
    }

    zs.next_in = (Bytef*)input.data();
    zs.avail_in = (uInt)input.size();

    char outbuffer[32768];
    std::string out;

    do {
        zs.next_out = reinterpret_cast<Bytef*>(outbuffer);
        zs.avail_out = sizeof(outbuffer);

        ret = deflate(&zs, Z_FINISH);

        if (out.size() < zs.total_out) {
            out.append(outbuffer, zs.total_out - out.size());
        }
    } while (ret == Z_OK);

    deflateEnd(&zs);

    if (ret != Z_STREAM_END) {
        throw std::runtime_error("gzip deflate failed");
    }

    return out;
}

static bool gzip_decompress(const std::string& input, std::string& output)  //解压缩 gzip 格式字符串
{
    output.clear();
    if (input.empty()) return true;

    z_stream zs;
    memset(&zs, 0, sizeof(zs));

    int ret = inflateInit2(&zs, 15 + 32);
    if (ret != Z_OK) return false;

    zs.next_in = (Bytef*)input.data();
    zs.avail_in = (uInt)input.size();

    char outbuffer[32768];

    do {
        zs.next_out = reinterpret_cast<Bytef*>(outbuffer);
        zs.avail_out = sizeof(outbuffer);

        ret = inflate(&zs, 0);

        if (output.size() < zs.total_out) {
            output.append(outbuffer, zs.total_out - output.size());
        }

        if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) {
            inflateEnd(&zs);
            return false;
        }
    } while (ret != Z_STREAM_END);

    inflateEnd(&zs);
    return true;
}

static int base64_value(char c)
{
    if (c >= 'A' && c <= 'Z') return c - 'A';
    if (c >= 'a' && c <= 'z') return c - 'a' + 26;
    if (c >= '0' && c <= '9') return c - '0' + 52;
    if (c == '+') return 62;
    if (c == '/') return 63;
    return -1;
}

static bool base64_decode(const std::string& input, std::string& output)    //解码 base64
{
    output.clear();
    int val = 0;
    int valb = -8;
    for (size_t i = 0; i < input.size(); ++i) {
        char c = input[i];
        if (c == '=') break;
        int d = base64_value(c);
        if (d == -1) {
            if (c == '\r' || c == '\n' || c == ' ' || c == '\t') continue;
            return false;
        }
        val = (val << 6) + d;
        valb += 6;
        if (valb >= 0) {
            output.push_back(char((val >> valb) & 0xFF));
            valb -= 8;
        }
    }
    return true;
}

static bool decode_body(const std::string& enc, //解压缩
                        const std::string& body,
                        std::string& out)
{
    if (enc == "gzip") {
        std::string gz;
        if (!base64_decode(body, gz)) return false;
        return gzip_decompress(gz, out);
    }

    out = body;
    return true;
}

static void encode_body(const std::string& json,
                        std::string& enc,
                        std::string& body)
{
    static bool gzip_disabled = false;

    if (json.size() >= GZIP_THRESHOLD_BYTES && !gzip_disabled) {
        try {
            enc = "gzip";
        body = base64_encode(gzip_compress(json));  //大的压缩后编码
        } catch (const std::exception& e) {
            std::cerr << "[REDIS_STREAM][GZIP_FALLBACK] " << e.what()
                      << ", send plain body" << std::endl;
            if (std::string(e.what()).find("ret=-2") != std::string::npos) {
                gzip_disabled = true;
                std::cerr << "[REDIS_STREAM][GZIP_DISABLED] zlib init stream error, keep sending plain body"
                          << std::endl;
            }
            enc = "plain";
            body = json;
        }
    } else {
        enc = "plain";
        body = json;    //小的保持json
    }
}

static std::string parse_host(const std::string& addr)//解析主机地址，去掉协议、端口和数据库部分
{
    std::string s = addr;
    if (s.find("redis://") == 0) s = s.substr(8);
    size_t slash = s.find('/');
    if (slash != std::string::npos) s = s.substr(0, slash);
    size_t colon = s.rfind(':');
    if (colon == std::string::npos) return s;
    return s.substr(0, colon);
}

static int parse_port(const std::string& addr)//提取端口
{
    std::string s = addr;
    if (s.find("redis://") == 0) s = s.substr(8);
    size_t slash = s.find('/');
    if (slash != std::string::npos) s = s.substr(0, slash);
    size_t colon = s.rfind(':');
    if (colon == std::string::npos) return 6379;
    return atoi(s.substr(colon + 1).c_str());
}

static int parse_db(const std::string& addr)    //提取数据库
{
    size_t slash = addr.rfind('/');
    if (slash == std::string::npos) return -1;
    std::string db = addr.substr(slash + 1);
    if (db.empty()) return -1;
    for (size_t i = 0; i < db.size(); ++i) {
        if (!isdigit((unsigned char)db[i])) return -1;
    }
    return atoi(db.c_str());
}

class RedisConnection { //Redis 连接封装
public:
    RedisConnection() : ctx_(NULL), db_(-1) {}
    ~RedisConnection() { close(); }

    void configure(const std::string& addr)
    {
        addr_ = addr;
        host_ = parse_host(addr);
        port_ = parse_port(addr);
        db_ = parse_db(addr);

        // lnk20260626 修改：Redis 模式复用 RocketMq/ConsumerSecretKey 作为密码
        password_ = G_MQCONSUMER_SECRETKEY; 

        if (host_.empty()) host_ = "127.0.0.1";
    }

    redisContext* get()
    {
        if (ctx_ && ctx_->err == 0) return ctx_;
        reconnect();
        return ctx_;
    }

    redisReply* command(const char* fmt, ...)
    {
        redisContext* c = get();
        if (!c) return NULL;

        va_list ap;
        va_start(ap, fmt);
        void* r = redisvCommand(c, fmt, ap);
        va_end(ap);

        if (!r) {
            close();
            return NULL;
        }
        return (redisReply*)r;
    }

    bool ping()
    {
        redisReply* r = command("PING");
        if (!r) return false;
        bool ok = (r->type == REDIS_REPLY_STATUS && r->str && std::string(r->str) == "PONG");
        freeReplyObject(r);
        return ok;
    }

    void close()
    {
        if (ctx_) {
            redisFree(ctx_);
            ctx_ = NULL;
        }
    }

private:
    void reconnect()
    {
        close();

        struct timeval tv;
        tv.tv_sec = 1;
        tv.tv_usec = 500000;

        ctx_ = redisConnectWithTimeout(host_.c_str(), port_, tv);
        if (!ctx_ || ctx_->err) {
            if (ctx_) {
                std::cerr << "[REDIS_STREAM] connect failed: " << ctx_->errstr << std::endl;
                close();
            } else {
                std::cerr << "[REDIS_STREAM] connect failed: null context" << std::endl;
            }
            return;
        }

        redisSetTimeout(ctx_, tv);

        // lnk20260626 修改：Redis 密码认证
        if (!password_.empty()) {
            redisReply* r = (redisReply*)redisCommand(
                ctx_,
                "AUTH %b",
                password_.data(),
                password_.size());
            
            if (!r || r->type == REDIS_REPLY_ERROR) {
                std::cerr << "[REDIS_STREAM] auth failed" << std::endl;
                if (r) {
                    if (r->str) {
                        std::cerr << "[REDIS_STREAM] auth error: " << r->str << std::endl;
                    }
                    freeReplyObject(r);
                }
                close();
                return;
            }
        
            freeReplyObject(r);
        }

        if (db_ >= 0) {
            redisReply* r = (redisReply*)redisCommand(ctx_, "SELECT %d", db_);
            if (!r || (r->type == REDIS_REPLY_ERROR)) {
                std::cerr << "[REDIS_STREAM] select db failed: " << db_ << std::endl;
                if (r) freeReplyObject(r);
                close();
                return;
            }
            freeReplyObject(r);
        }
    }

private:
    redisContext* ctx_;
    std::string addr_;
    std::string host_;
    std::string password_;   // lnk20260626 修改：Redis AUTH 密码
    int port_;
    int db_;
};

// lnk20260626 修改：区分 XADD 成功、OOM、普通失败
enum RedisXAddResult {
    REDIS_XADD_OK = 0,
    REDIS_XADD_OOM = 1,
    REDIS_XADD_FAIL = 2
};

class RedisStreamProducer { //RedisStream 生产者封装
public:
    explicit RedisStreamProducer(const std::string& addr)
        : xadd_pause_until_ms_(0),
          last_flush_ms_(0),
          last_memory_check_ms_(0),
          cached_memory_high_(false),
          last_pressure_log_ms_(0)
    {
        conn_.configure(addr);
        ensure_dir(SPOOL_DIR);
        if (!conn_.ping()) {
            std::cerr << "[REDIS_STREAM][PRODUCER] Redis not ready, spool enabled" << std::endl;
            xadd_pause_until_ms_ = now_ms() + REDIS_XADD_FAIL_BACKOFF_MS;
        }
    }

    void sendMessage(const std::string& json,
                     const std::string& topic,
                     const std::string& tag,
                     const std::string& key)
    {
        const int64_t n = now_ms();

        std::string enc;
        std::string body;
        encode_body(json, enc, body);

        const std::string stream = build_stream_name(topic, tag);
        const std::string ts = to_string_i64(n);

        /*
         * lnk20260626 修改：
         * Redis 处于退避期时，不再尝试 XADD，直接落本地 spool。
         * 这就是“延迟 XADD”。
         */
        if (n < xadd_pause_until_ms_) {
            spool(stream, key, tag, enc, body, ts);
            logPressureOnce("[REDIS_STREAM][DELAY_XADD] redis in backoff, message spooled");
            return;
        }

        /*
         * lnk20260626 修改：
         * 定期检查 Redis 内存，如果超过高水位，提前停止 XADD，避免真正 OOM。
         */
        if (isRedisMemoryHigh()) {
            xadd_pause_until_ms_ = now_ms() + REDIS_XADD_OOM_BACKOFF_MS;
            spool(stream, key, tag, enc, body, ts);
            logPressureOnce("[REDIS_STREAM][MEM_HIGH] redis memory high, delay xadd and spool message");
            return;
        }

        /*
         * lnk20260626 修改：
         * spool 不要每条消息都 flush，避免恢复后瞬间冲击 Redis。
         */
        if (n - last_flush_ms_ >= REDIS_SPOOL_FLUSH_INTERVAL_MS) {
            flushSpool();
            last_flush_ms_ = n;
        }

        RedisXAddResult ret = xadd(stream, key, tag, enc, body, ts);
        if (ret == REDIS_XADD_OK) {
            return;
        }

        /*
         * lnk20260626 修改：
         * XADD OOM 后进入退避，后续消息直接 spool。
         */
        if (ret == REDIS_XADD_OOM) {
            xadd_pause_until_ms_ = now_ms() + REDIS_XADD_OOM_BACKOFF_MS;
            spool(stream, key, tag, enc, body, ts);
            logPressureOnce("[REDIS_STREAM][OOM_BACKOFF] XADD OOM, delay xadd and spool message");
            return;
        }

        /*
         * 普通失败，例如网络断开，也进入短退避。
         */
        xadd_pause_until_ms_ = now_ms() + REDIS_XADD_FAIL_BACKOFF_MS;
        spool(stream, key, tag, enc, body, ts);
        logPressureOnce("[REDIS_STREAM][XADD_FAIL_BACKOFF] XADD failed, message spooled");
    }

    void shutdown() {}

private:

    void logPressureOnce(const char* msg)
    {
        const int64_t n = now_ms();

        /*
         * 压力日志最多 10 秒打一条，避免 OOM 时刷屏。
         */
        if (n - last_pressure_log_ms_ >= 10000) {
            last_pressure_log_ms_ = n;
            std::cerr << msg << std::endl;

            DIY_ERRORLOG_CODE("process", 0, LOG_CODE_MQ,
                "【WARN】前置的%s%d号进程 Redis Stream写入延迟，消息已落本地缓存",
                get_front_msg_from_subdir(), g_front_seg_index);
        }
    }

    static bool parse_info_number(const std::string& info,
                                  const std::string& key,
                                  long long& value)
    {
        value = 0;

        std::string pattern = key + ":";
        size_t pos = info.find(pattern);
        if (pos == std::string::npos) return false;

        pos += pattern.size();
        size_t end = info.find_first_of("\r\n", pos);
        std::string num = info.substr(pos, end == std::string::npos ? std::string::npos : end - pos);

        if (num.empty()) return false;

        value = atoll(num.c_str());
        return value > 0;
    }

    bool isRedisMemoryHigh()
    {
        const int64_t n = now_ms();

        if (n - last_memory_check_ms_ < REDIS_MEMORY_CHECK_INTERVAL_MS) {
            return cached_memory_high_;
        }

        last_memory_check_ms_ = n;
        cached_memory_high_ = false;

        redisReply* r = conn_.command("INFO memory");
        if (!r) {
            return false;
        }

        if (r->type != REDIS_REPLY_STRING || !r->str) {
            freeReplyObject(r);
            return false;
        }

        std::string info(r->str, r->len);
        freeReplyObject(r);

        long long used_memory = 0;
        long long maxmemory = 0;

        if (!parse_info_number(info, "used_memory", used_memory)) {
            return false;
        }

        if (!parse_info_number(info, "maxmemory", maxmemory)) {
            return false;
        }

        /*
         * maxmemory=0 表示 Redis 没有限制内存。
         * 你的环境设置了 2G，所以这里应该能读到 maxmemory。
         */
        if (maxmemory <= 0) {
            return false;
        }

        if (used_memory * 100 >= maxmemory * REDIS_MEMORY_HIGH_WATERMARK) {
            cached_memory_high_ = true;

            std::cerr << "[REDIS_STREAM][MEM_HIGH] used_memory=" << used_memory
                      << " maxmemory=" << maxmemory
                      << " percent=" << (used_memory * 100 / maxmemory)
                      << "% watermark=" << REDIS_MEMORY_HIGH_WATERMARK
                      << "%" << std::endl;
        }

        return cached_memory_high_;
    }

    RedisXAddResult xadd(const std::string& stream,
                         const std::string& key,
                         const std::string& tag,
                         const std::string& enc,
                         const std::string& body,
                         const std::string& ts)
    {
        redisReply* r = conn_.command(
            "XADD %b * key %b tag %b enc %b body %b ts %b",
            stream.data(), stream.size(),
            key.data(), key.size(),
            tag.data(), tag.size(),
            enc.data(), enc.size(),
            body.data(), body.size(),
            ts.data(), ts.size());

        if (!r) {
            return REDIS_XADD_FAIL;
        }

        if (r->type == REDIS_REPLY_STRING) {
            freeReplyObject(r);
            return REDIS_XADD_OK;
        }

        if (r->type == REDIS_REPLY_ERROR) {
            std::string err = r->str ? r->str : "";
            std::cerr << "[REDIS_STREAM][XADD_ERROR] " << err << std::endl;

            freeReplyObject(r);

            /*
             * lnk20260626 修改：
             * noeviction + maxmemory 满时，Redis 会返回 OOM。
             */
            if (err.find("OOM") != std::string::npos ||
                err.find("maxmemory") != std::string::npos ||
                err.find("command not allowed") != std::string::npos) {
                return REDIS_XADD_OOM;
            }

            return REDIS_XADD_FAIL;
        }

        freeReplyObject(r);
        return REDIS_XADD_FAIL;
    }

    void spool(const std::string& stream,
               const std::string& key,
               const std::string& tag,
               const std::string& enc,
               const std::string& body,
               const std::string& ts)
    {
        ensure_dir(SPOOL_DIR);
        char uniq[128];
        snprintf(uniq, sizeof(uniq), "%lld_%d_%llu",
                 (long long)now_ms(),
                 (int)getpid(),
                 (unsigned long long)rand());

        std::string fn = std::string(SPOOL_DIR) + "/" + uniq + "_" +
                 shell_safe_name(stream) + "_" + shell_safe_name(key) + ".msg.tmp";

        std::string done = fn.substr(0, fn.size() - 4);

        std::ofstream out(fn.c_str(), std::ios::out | std::ios::binary);
        if (!out) {
            std::cerr << "[REDIS_STREAM][SPOOL] open failed: " << fn << std::endl;
            return;
        }

        out << "stream=" << stream << "\n";
        out << "key=" << key << "\n";
        out << "tag=" << tag << "\n";
        out << "enc=" << enc << "\n";
        out << "ts=" << ts << "\n";
        out << "body=" << body << "\n";
        out.close();

        if (rename(fn.c_str(), done.c_str()) != 0) {
            std::cerr << "[REDIS_STREAM][SPOOL] rename failed errno=" << errno << std::endl;
        }
    }

    static bool parse_spool_file(const std::string& path,
                                 std::string& stream,
                                 std::string& key,
                                 std::string& tag,
                                 std::string& enc,
                                 std::string& body,
                                 std::string& ts)
    {
        std::ifstream in(path.c_str(), std::ios::in | std::ios::binary);
        if (!in) return false;

        std::string line;
        while (std::getline(in, line)) {
            if (line.find("stream=") == 0) stream = line.substr(7);
            else if (line.find("key=") == 0) key = line.substr(4);
            else if (line.find("tag=") == 0) tag = line.substr(4);
            else if (line.find("enc=") == 0) enc = line.substr(4);
            else if (line.find("ts=") == 0) ts = line.substr(3);
            else if (line.find("body=") == 0) {
                body = line.substr(5);
                std::string rest;
                while (std::getline(in, rest)) {
                    body += "\n";
                    body += rest;
                }
                break;
            }
        }
        return !stream.empty() && !enc.empty() && !ts.empty();
    }

    void flushSpool()
    {
        const int64_t n = now_ms();

        if (n < xadd_pause_until_ms_) {
            return;
        }

        if (isRedisMemoryHigh()) {
            xadd_pause_until_ms_ = now_ms() + REDIS_XADD_OOM_BACKOFF_MS;
            logPressureOnce("[REDIS_STREAM][SPOOL_DELAY] redis memory high, stop flushing spool");
            return;
        }

        DIR* dir = opendir(SPOOL_DIR);
        if (!dir) return;

        std::vector<std::string> files;
        struct dirent* ent = NULL;
        while ((ent = readdir(dir)) != NULL) {
            std::string name = ent->d_name;
            if (name == "." || name == "..") continue;
            if (name.size() >= 4 && name.substr(name.size() - 4) == ".tmp") continue;
            files.push_back(std::string(SPOOL_DIR) + "/" + name);
        }
        closedir(dir);

        std::sort(files.begin(), files.end());

        int flushed = 0;

        for (size_t i = 0; i < files.size(); ++i) {
            if (flushed >= REDIS_SPOOL_FLUSH_LIMIT) {
                break;
            }

            std::string stream, key, tag, enc, body, ts;
            if (!parse_spool_file(files[i], stream, key, tag, enc, body, ts)) {
                continue;
            }

            RedisXAddResult ret = xadd(stream, key, tag, enc, body, ts);

            if (ret == REDIS_XADD_OK) {
                unlink(files[i].c_str());
                ++flushed;
                std::cout << "[REDIS_STREAM][SPOOL_FLUSHED] " << files[i] << std::endl;
                continue;
            }

            if (ret == REDIS_XADD_OOM) {
                xadd_pause_until_ms_ = now_ms() + REDIS_XADD_OOM_BACKOFF_MS;
                logPressureOnce("[REDIS_STREAM][SPOOL_OOM] XADD OOM while flushing spool");
                break;
            }

            xadd_pause_until_ms_ = now_ms() + REDIS_XADD_FAIL_BACKOFF_MS;
            logPressureOnce("[REDIS_STREAM][SPOOL_FAIL] XADD failed while flushing spool");
            break;
        }
    }

private:
    RedisConnection conn_;

    // lnk20260626 修改：延迟 XADD 控制
    int64_t xadd_pause_until_ms_;
    int64_t last_flush_ms_;

    // lnk20260626 修改：Redis 内存高水位缓存
    int64_t last_memory_check_ms_;
    bool cached_memory_high_;

    // lnk20260626 修改：压力日志限频
    int64_t last_pressure_log_ms_;
};

class RedisStreamConsumer { //RedisStream 消费者封装
public:
    RedisStreamConsumer(const std::string& group, const std::string& addr)
        : group_(group), stop_(false)
    {
        conn_.configure(addr);
        if (group_.empty()) group_ = "front_default";

        int64_t pid = (int64_t)getpid();
        int64_t start_ms = now_ms();

        char buf[128];
        snprintf(buf, sizeof(buf), "_%s_%d_%lld_%lld",
                 subdir, g_front_seg_index,
                 (long long)pid, (long long)start_ms);

        group_ += buf;
        consumer_ = group_;
    }

    void subscribe(const std::string& topic,
                   const std::string& tag,
                   MessageCallBack cb)
    {
        Sub s;
        s.topic = topic;
        s.tag = tag;
        s.stream = build_stream_name(topic, tag);
        s.cb = cb;
        subs_.push_back(s);

        callbacks_[std::make_pair(topic, tag)] = cb;
        callbacks_[std::make_pair(s.stream, tag)] = cb;

        size_t pos = topic.find('%');
        if (pos != std::string::npos) {
            std::string shortTopic = topic.substr(pos + 1);
            callbacks_[std::make_pair(shortTopic, tag)] = cb;
        }

        createGroup(s.stream);

        std::cout << "[REDIS_STREAM][SUB] stream=" << s.stream
                  << " topic=" << topic << " tag=" << tag
                  << " group=" << group_ << " consumer=" << consumer_ << std::endl;
    }

    void start()
    {
        stop_ = false;
        while (!stop_) {
            claimOnce();
            readOnce();
        }
    }

    void shutdown()
    {
        stop_ = true;
        conn_.close();
    }

private:
    struct Sub {
        std::string topic;
        std::string tag;
        std::string stream;
        MessageCallBack cb;
    };

    void createGroup(const std::string& stream)
    {
        redisReply* r = conn_.command("XGROUP CREATE %b %b $ MKSTREAM",
                                      stream.data(), stream.size(),
                                      group_.data(), group_.size());
        if (!r) {
            std::cerr << "[REDIS_STREAM][GROUP] create failed: " << stream << std::endl;
            return;
        }

        if (r->type == REDIS_REPLY_ERROR) {
            std::string err = r->str ? r->str : "";
            if (err.find("BUSYGROUP") == std::string::npos) {
                std::cerr << "[REDIS_STREAM][GROUP_ERROR] " << stream << " " << err << std::endl;
            }
        }
        freeReplyObject(r);
    }

    bool parseFields(redisReply* arr,
                     std::map<std::string, std::string>& fields)
    {
        fields.clear();
        if (!arr || arr->type != REDIS_REPLY_ARRAY) return false;
        for (size_t i = 0; i + 1 < arr->elements; i += 2) {
            redisReply* k = arr->element[i];
            redisReply* v = arr->element[i + 1];
            if (!k || !v || k->type != REDIS_REPLY_STRING || v->type != REDIS_REPLY_STRING) continue;
            fields[std::string(k->str, k->len)] = std::string(v->str, v->len);
        }
        return true;
    }

    static std::string getField(const std::map<std::string, std::string>& m,
                                const std::string& k)
    {
        std::map<std::string, std::string>::const_iterator it = m.find(k);
        if (it == m.end()) return "";
        return it->second;
    }

    rocketmq::MQMessageExt makeMsg(const Sub& sub,
                                   const std::string& id,
                                   const std::map<std::string, std::string>& fields)
    {
        std::string enc = getField(fields, "enc");
        std::string bodyRaw = getField(fields, "body");
        std::string body;

        if (!decode_body(enc, bodyRaw, body)) {
            throw std::runtime_error("decode body failed");
        }

        std::string key = getField(fields, "key");
        std::string tag = getField(fields, "tag");
        std::string ts  = getField(fields, "ts");

        if (tag.empty()) {
            tag = sub.tag;
        }

        int64_t born = ts.empty() ? now_ms() : atoll(ts.c_str());

        rocketmq::MQMessageExt msg;

        msg.setTopic(sub.topic);
        msg.setTags(tag);
        msg.setKeys(key);
        msg.setBody(body);
        msg.setBornTimestamp(born);

        /*
         * 如果你的 RocketMQ MQMessageExt.h 里有 setMsgId，就打开。
         * 如果没有这个函数，就注释掉。
         */
        // msg.setMsgId(id);

        return msg;
    }

    MessageCallBack findCallback(const std::string& topic,
                                 const std::string& stream,
                                 const std::string& tag)
    {
        std::map<std::pair<std::string, std::string>, MessageCallBack>::iterator it;
        it = callbacks_.find(std::make_pair(topic, tag));
        if (it != callbacks_.end()) return it->second;

        it = callbacks_.find(std::make_pair(stream, tag));
        if (it != callbacks_.end()) return it->second;

        return NULL;
    }

    void ack(const std::string& stream, const std::string& id)
    {
        redisReply* r = conn_.command("XACK %b %b %b",
                                      stream.data(), stream.size(),
                                      group_.data(), group_.size(),
                                      id.data(), id.size());
        if (r) freeReplyObject(r);
    }

    void handleOne(const Sub& sub, const std::string& id, redisReply* fieldsReply)
    {
        try {
            std::map<std::string, std::string> fields;
            if (!parseFields(fieldsReply, fields)) {
                std::cerr << "[REDIS_STREAM][BAD_MSG] no fields stream=" << sub.stream << " id=" << id << std::endl;
                ack(sub.stream, id);
                return;
            }

            rocketmq::MQMessageExt msg = makeMsg(sub, id, fields);

            MessageCallBack cb = findCallback(sub.topic, sub.stream, msg.getTags());
            if (!cb) {
                std::cout << "[REDIS_STREAM][NO_CALLBACK] stream=" << sub.stream
                          << " tag=" << msg.getTags() << std::endl;
                ack(sub.stream, id);
                return;
            }

            rocketmq::ConsumeStatus ret = cb(msg);
            if (ret == rocketmq::CONSUME_SUCCESS) {
                ack(sub.stream, id);
            } else {
                std::cout << "[REDIS_STREAM][RETRY_LATER] stream=" << sub.stream
                          << " id=" << id << std::endl;
            }
        } catch (const std::exception& e) {
            std::cerr << "[REDIS_STREAM][HANDLE_EXCEPTION] " << e.what()
                      << " stream=" << sub.stream << " id=" << id << std::endl;
        } catch (...) {
            std::cerr << "[REDIS_STREAM][HANDLE_UNKNOWN_EXCEPTION] stream="
                      << sub.stream << " id=" << id << std::endl;
        }
    }

    void readOnce()
    {
        for (size_t i = 0; i < subs_.size(); ++i) {
            Sub& sub = subs_[i];

            redisReply* r = conn_.command(
                "XREADGROUP GROUP %b %b COUNT %d BLOCK %lld STREAMS %b >",
                group_.data(), group_.size(),
                consumer_.data(), consumer_.size(),
                REDIS_READ_COUNT,
                (long long)REDIS_BLOCK_MS,
                sub.stream.data(), sub.stream.size());

            if (!r) {
                usleep(500000);
                continue;
            }

            if (r->type == REDIS_REPLY_NIL) {
                freeReplyObject(r);
                continue;
            }

            if (r->type == REDIS_REPLY_ERROR) {
                std::string err = r->str ? r->str : "";
                std::cerr << "[REDIS_STREAM][XREADGROUP_ERROR] " << err << std::endl;
                if (err.find("NOGROUP") != std::string::npos) {
                    createGroup(sub.stream);
                }
                freeReplyObject(r);
                continue;
            }

            if (r->type == REDIS_REPLY_ARRAY) {
                for (size_t si = 0; si < r->elements; ++si) {
                    redisReply* streamPair = r->element[si];
                    if (!streamPair || streamPair->type != REDIS_REPLY_ARRAY || streamPair->elements < 2) continue;

                    redisReply* messages = streamPair->element[1];
                    if (!messages || messages->type != REDIS_REPLY_ARRAY) continue;

                    for (size_t mi = 0; mi < messages->elements; ++mi) {
                        redisReply* msgPair = messages->element[mi];
                        if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue;
                        redisReply* idReply = msgPair->element[0];
                        redisReply* fieldsReply = msgPair->element[1];
                        if (!idReply || idReply->type != REDIS_REPLY_STRING) continue;

                        std::string id(idReply->str, idReply->len);
                        handleOne(sub, id, fieldsReply);
                    }
                }
            }

            freeReplyObject(r);
        }
    }

    void claimOnce()
    {
        for (size_t i = 0; i < subs_.size(); ++i) {
            Sub& sub = subs_[i];

            redisReply* r = conn_.command(
                "XAUTOCLAIM %b %b %b %lld 0-0 COUNT %d",
                sub.stream.data(), sub.stream.size(),
                group_.data(), group_.size(),
                consumer_.data(), consumer_.size(),
                (long long)REDIS_IDLE_CLAIM_MS,
                REDIS_READ_COUNT);

            if (!r) continue;

            if (r->type == REDIS_REPLY_ERROR) {
                std::string err = r->str ? r->str : "";
                if (err.find("NOGROUP") != std::string::npos) createGroup(sub.stream);
                freeReplyObject(r);
                continue;
            }

            if (r->type == REDIS_REPLY_ARRAY && r->elements >= 2) {
                redisReply* messages = r->element[1];
                if (messages && messages->type == REDIS_REPLY_ARRAY) {
                    for (size_t mi = 0; mi < messages->elements; ++mi) {
                        redisReply* msgPair = messages->element[mi];
                        if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue;
                        redisReply* idReply = msgPair->element[0];
                        redisReply* fieldsReply = msgPair->element[1];
                        if (!idReply || idReply->type != REDIS_REPLY_STRING) continue;

                        std::string id(idReply->str, idReply->len);
                        std::cout << "[REDIS_STREAM][CLAIM] stream=" << sub.stream
                                  << " id=" << id << std::endl;
                        handleOne(sub, id, fieldsReply);
                    }
                }
            }

            freeReplyObject(r);
        }
    }

private:
    RedisConnection conn_;
    std::string group_;
    std::string consumer_;
    bool stop_;
    std::vector<Sub> subs_;
    std::map<std::pair<std::string, std::string>, MessageCallBack> callbacks_;
};

static RedisStreamProducer* g_redis_producer = NULL;    //全局生产者实例
static RedisStreamConsumer* g_redis_consumer = NULL;    //全局消费者实例

bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg)  //判断消息是否在程序启动后产生
{
    const int64_t born_ts = static_cast<int64_t>(msg.getBornTimestamp());
    return born_ts >= (redis_app_start_ms() - REDIS_START_SKEW_MS);
}

void RedisStream_InitializeProducer()   //初始化生产者
{
    if (g_redis_producer == NULL) {
        std::string addr = G_ROCKETMQ_IPPORT.empty() ? G_MQCONSUMER_IPPORT : G_ROCKETMQ_IPPORT;
        g_redis_producer = new RedisStreamProducer(addr);
        std::cout << "[REDIS_STREAM] producer initialized addr=" << addr << std::endl;
    }
}

void RedisStream_ShutdownAndDestroyProducer()//关闭生产者
{
    if (g_redis_producer != NULL) {
        g_redis_producer->shutdown();
        delete g_redis_producer;
        g_redis_producer = NULL;
    }
}

void RedisStream_producer_send(const std::string& body, //发送消息
                               const std::string& topic,
                               const std::string& tags,
                               const std::string& keys)
{
    if (g_redis_producer == NULL) RedisStream_InitializeProducer();

    try {
        g_redis_producer->sendMessage(body, topic, tags, keys);
    } catch (const std::exception& e) {
        std::cerr << "[REDIS_STREAM][SEND_FAIL] " << e.what() << std::endl;
        DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
            "【ERROR】前置的%s%d号进程 redis stream发送失败,已尝试本地落盘",
            get_front_msg_from_subdir(), g_front_seg_index);
    } catch (...) {
        std::cerr << "[REDIS_STREAM][SEND_FAIL] unknown exception" << std::endl;
        DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
            "【ERROR】前置的%s%d号进程 redis stream发送未知异常",
            get_front_msg_from_subdir(), g_front_seg_index);
    }
}

void RedisStream_InitializeConsumer(const std::string& consumerName,    //初始化消费者
                                    const std::string& nameServer,
                                    const std::vector<Subscription>& subscriptions)
{
    if (g_redis_consumer == NULL) {
        g_redis_consumer = new RedisStreamConsumer(consumerName, nameServer);
        for (size_t i = 0; i < subscriptions.size(); ++i) {
            g_redis_consumer->subscribe(subscriptions[i].topic,
                                        subscriptions[i].tag,
                                        subscriptions[i].callback);
        }
    }
}

void RedisStream_ShutdownAndDestroyConsumer()//关闭消费者
{
    if (g_redis_consumer != NULL) {
        g_redis_consumer->shutdown();
        delete g_redis_consumer;
        g_redis_consumer = NULL;
    }
}

void RedisStream_consumer_receive(const std::string& consumerName, //启动消费者接收消息
                                  const std::string& nameServer,
                                  const std::vector<Subscription>& subscriptions)
{
    RedisStream_InitializeConsumer(consumerName, nameServer, subscriptions);
    if (g_redis_consumer) {
        g_redis_consumer->start();
    }
}