// 源码路径: include/maxscale/filter.hh

class FilterSession : public MXS_FILTER_SESSION {
public:
    // 处理从客户端发来的请求（请求方向）
    // 返回 1 表示继续传递，0 表示终止
    virtual int routeQuery(GWBUF* buffer) = 0;
    
    // 处理从后端返回的响应（响应方向）
    virtual int clientReply(GWBUF* buffer,
                           const mxs::ReplyRoute& down,
                           const mxs::Reply& reply) = 0;
    
    // 诊断信息输出
    virtual json_t* diagnostics() const;
};

class Filter : public MXS_FILTER {
public:
    // 创建 Filter 实例（进程级别，全局唯一）
    static Filter* create(const char* name, mxs::ConfigParameters* params);
    
    // 为每个客户端会话创建 FilterSession
    virtual FilterSession* newSession(MXS_SESSION* session,
                                      SERVICE* service) = 0;
};

// 伪代码：Filter Chain 构建过程
// 源码参考: server/core/service.cc

void Service::create_filter_chain() {
    // 配置中 filters=QLA|Cache|Tee
    // 组装顺序：Client → QLA → Cache → Tee → Router
    
    FilterSession* chain_tail = router_session;  // 链尾是 Router
    
    // 从后向前构建链表
    for (int i = filter_count - 1; i >= 0; i--) {
        FilterSession* fs = filters[i]->newSession(session, this);
        fs->set_downstream(chain_tail);   // 设置下游
        chain_tail->set_upstream(fs);     // 设置上游（用于 clientReply）
        chain_tail = fs;
    }
    
    session->set_head(chain_tail);  // 链头 = 第一个 Filter
}

// GWBUF 结构简化表示
struct GWBUF {
    uint8_t* start;     // 数据起始指针
    uint8_t* end;       // 数据结束指针
    GWBUF*   next;      // 链表下一个节点（支持分包）
    uint32_t gwbuf_type; // 缓冲区类型标记
};

class RWSplitSession : public RouterSession {
    // 当前 Master 连接
    mxs::RWBackend* m_current_master;
    
    // 所有可用的 Slave 连接
    mxs::PRWBackends m_slave_connections;
    
    // 事务状态
    bool m_in_transaction;
    
    // 上一次写操作的 GTID（用于因果读）
    std::string m_gtid;
    
    // 目标选择策略
    select_criteria_t m_criteria;
};

// 路由决策伪代码（基于源码简化）
route_target_t RWSplitSession::get_target(GWBUF* buffer) {
    
    uint32_t type = qc_get_type_mask(buffer);  // SQL 分类
    
    // 1. 事务内 → Master
    if (m_in_transaction) {
        return TARGET_MASTER;
    }
    
    // 2. 写操作 → Master
    if (type & (QUERY_TYPE_WRITE | QUERY_TYPE_CREATE | QUERY_TYPE_DROP)) {
        return TARGET_MASTER;
    }
    
    // 3. SELECT ... FOR UPDATE → Master
    if (type & QUERY_TYPE_WRITE_LOCK) {
        return TARGET_MASTER;
    }
    
    // 4. 检查 Hint
    // -- maxscale route to server server1
    if (buffer_has_hint(buffer, HINT_ROUTE_TO_MASTER)) {
        return TARGET_MASTER;
    }
    if (buffer_has_hint(buffer, HINT_ROUTE_TO_SLAVE)) {
        return TARGET_SLAVE;
    }
    
    // 5. Prepared Statement
    // PREPARE 走 Master（需要状态绑定）
    // EXECUTE 根据 PREPARE 时的 SQL 类型决定
    if (type & QUERY_TYPE_PREPARE_STMT) {
        return TARGET_MASTER;  // 保守策略
    }
    
    // 6. 多语句（Multi-Statement）→ Master
    // 无法逐条判断，保守路由到 Master
    if (type & QUERY_TYPE_MULTI_STATEMENT) {
        return TARGET_MASTER;
    }
    
    // 7. 普通 SELECT → Slave
    if (type & QUERY_TYPE_READ) {
        return TARGET_SLAVE;
    }
    
    // 8. 无法判断 → Master（保守）
    return TARGET_MASTER;
}

mxs::RWBackend* RWSplitSession::get_slave_backend() {
    mxs::RWBackend* best = nullptr;
    
    switch (m_criteria) {
        case LEAST_CURRENT_OPERATIONS:
            // 选择当前活跃查询数最少的 Slave
            for (auto& backend : m_slave_connections) {
                if (!best || backend->current_ops() < best->current_ops()) {
                    best = backend;
                }
            }
            break;
            
        case LEAST_GLOBAL_CONNECTIONS:
            // 选择全局连接数最少的 Slave
            break;
            
        case ADAPTIVE_ROUTING:
            // 基于平均响应时间选择最快的 Slave
            break;
    }
    
    return best;
}

// 因果读处理伪代码
bool RWSplitSession::handle_causal_read(GWBUF* buffer) {
    if (!m_config.causal_reads || m_gtid.empty()) {
        return false;  // 未启用因果读 或 无写操作记录
    }
    
    // 在实际查询前，先发送 GTID WAIT 命令
    std::string wait_query = 
        "SELECT MASTER_GTID_WAIT('" + m_gtid + "', " 
        + std::to_string(m_config.causal_reads_timeout) + ")";
    
    GWBUF* wait_buf = modutil_create_query(wait_query.c_str());
    
    // 发送 WAIT 查询到目标 Slave
    m_target_slave->write(wait_buf);
    
    // 保存原始查询，等 WAIT 返回后再发送
    m_pending_query = buffer;
    m_state = WAITING_FOR_GTID;
    
    return true;
}

// WAIT 返回后的回调处理
void RWSplitSession::handle_gtid_wait_reply(GWBUF* reply) {
    int result = extract_wait_result(reply);
    
    if (result >= 0) {
        // WAIT 成功，Slave 已追上，发送原始查询到 Slave
        m_target_slave->write(m_pending_query);
    } else {
        // WAIT 超时，回退到 Master
        m_current_master->write(m_pending_query);
    }
    
    m_state = ROUTING;
}

// 源码路径: server/modules/filter/qlafilter/

class QlaFilterSession : public FilterSession {
    FILE* m_logfile;  // 日志文件句柄
    
    int routeQuery(GWBUF* buffer) override {
        // 提取 SQL 文本
        char* sql = modutil_get_SQL(buffer);
        
        // 写入日志
        fprintf(m_logfile, "%s: %s\n", timestamp(), sql);
        free(sql);
        
        // 透传给下游（不修改请求）
        return FilterSession::routeQuery(buffer);
    }
    
    int clientReply(GWBUF* buffer, 
                    const mxs::ReplyRoute& down,
                    const mxs::Reply& reply) override {
        // 响应方向：直接透传
        return FilterSession::clientReply(buffer, down, reply);
    }
};

class TeeFilterSession : public FilterSession {
    MXS_SESSION* m_branch_session;  // 分支 Session（影子库连接）
    
    int routeQuery(GWBUF* buffer) override {
        // 关键：克隆 GWBUF
        GWBUF* clone = gwbuf_clone(buffer);
        
        // 将克隆发送到分支 Session
        m_branch_session->route_query(clone);
        
        // 原始请求继续透传到主链路
        return FilterSession::routeQuery(buffer);
    }
    
    int clientReply(GWBUF* buffer,
                    const mxs::ReplyRoute& down,
                    const mxs::Reply& reply) override {
        // 只返回主链路的响应，丢弃分支响应
        return FilterSession::clientReply(buffer, down, reply);
    }
};

class CacheFilterSession : public FilterSession {
    Cache* m_cache;
    
    int routeQuery(GWBUF* buffer) override {
        std::string sql = get_sql(buffer);
        std::string key = hash(sql);
        
        CacheEntry* entry = m_cache->lookup(key);
        
        if (entry && !entry->expired()) {
            // 缓存命中：直接构造响应返回给客户端
            GWBUF* cached_response = entry->to_gwbuf();
            
            // 注意：这里不调用下游的 routeQuery
            // 而是直接调用上游的 clientReply
            FilterSession::clientReply(cached_response, ...);
            return 0;  // 终止链路
        }
        
        // 缓存未命中：记录 key，透传到下游
        m_pending_key = key;
        return FilterSession::routeQuery(buffer);
    }
    
    int clientReply(GWBUF* buffer,
                    const mxs::ReplyRoute& down,
                    const mxs::Reply& reply) override {
        if (!m_pending_key.empty()) {
            // 将响应写入缓存
            m_cache->store(m_pending_key, buffer);
            m_pending_key.clear();
        }
        
        return FilterSession::clientReply(buffer, down, reply);
    }
};

// myfilter.cc
#include <maxscale/ccdefs.hh>
#include <maxscale/filter.hh>

class MyFilter : public Filter {
public:
    static MyFilter* create(const char* name, mxs::ConfigParameters* params) {
        return new MyFilter(name, params);
    }
    
    MyFilterSession* newSession(MXS_SESSION* session, SERVICE* service) override {
        return new MyFilterSession(session, service, this);
    }
};

class MyFilterSession : public FilterSession {
public:
    int routeQuery(GWBUF* buffer) override {
        // 你的请求处理逻辑
        return FilterSession::routeQuery(buffer);
    }
    
    int clientReply(GWBUF* buffer,
                    const mxs::ReplyRoute& down,
                    const mxs::Reply& reply) override {
        // 你的响应处理逻辑
        return FilterSession::clientReply(buffer, down, reply);
    }
};

// 模块注册宏
MXS_MODULE* MXS_CREATE_MODULE() {
    static MXS_MODULE info = {
        .api_version = MXS_MODULE_API_VERSION,
        .module_init = NULL,
        .module_finish = NULL,
        .object = &MyFilter::s_object,
    };
    return &info;
}