flowchart TB
    Client[客户端 Channel]
    QT[rabbit_queue_type<br/>类型分发器]
    
    subgraph Classic[经典队列 Classic]
        CQ[rabbit_classic_queue]
        VQ[rabbit_variable_queue<br/>变长队列]
        QI[queue_index<br/>索引]
        MS[msg_store<br/>消息存储]
    end
    
    subgraph Quorum[仲裁队列 Quorum]
        QQ[rabbit_quorum_queue]
        RF[rabbit_fifo<br/>Ra 状态机]
        RA[Ra 共识引擎]
        WAL[Write Ahead Log]
    end
    
    subgraph Stream[流队列 Stream]
        SQ[rabbit_stream_queue]
        OSI[Osiris 日志系统]
        SEG[Segment 分段存储]
    end
    
    Client --> QT
    QT --> CQ
    QT --> QQ
    QT --> SQ
    
    CQ --> VQ
    VQ --> QI
    VQ --> MS
    
    QQ --> RF
    RF --> RA
    RA --> WAL
    
    SQ --> OSI
    OSI --> SEG
    
    style Classic fill:#e3f2fd
    style Quorum fill:#fff3e0
    style Stream fill:#f3e5f5

%% amqqueue 记录（定义于 amqqueue_v2.hrl）
-record(amqqueue, {
    name        :: rabbit_amqqueue:name(),  % {resource, VHost, queue, Name}
    durable     :: boolean(),               % 是否持久化到磁盘
    auto_delete :: boolean(),               % 无消费者时自动删除
    exclusive_owner :: none | pid(),        % 独占连接（仅该连接可访问）
    arguments   :: rabbit_framing:amqp_table(), % 扩展参数（如 x-max-length）
    pid         :: pid() | ra_server_id(),  % 队列进程 PID（经典）或 Ra 集群 ID（仲裁/流）
    type        :: atom(),                  % rabbit_classic_queue | rabbit_quorum_queue | rabbit_stream_queue
    vhost       :: binary(),                % 虚拟主机
    state       :: atom(),                  % live | stopped
    policy      :: undefined | binary(),    % 应用的策略名称
    operator_policy :: undefined | binary() % 运维策略
}).

-type declare_result() ::
    {new, amqqueue:amqqueue()} |          % 新创建
    {existing, amqqueue:amqqueue()} |      % 已存在且配置一致
    {owner_died, amqqueue:amqqueue()} |    % 独占队列的所有者进程已死亡
    {protocol_error, Type :: atom(), Reason :: string(), Args :: term()}.

declare(QueueName, Durable, AutoDelete, Args, Owner, ActingUser, Node) ->
    %% 1）参数规范化：合并策略参数、检查最大长度等约束
    NormalizedArgs = augment_declare_args(QueueName, Durable, AutoDelete, Args, ActingUser),
    
    %% 2）确定队列类型（默认 classic，可通过 x-queue-type 指定）
    Type = get_queue_type(NormalizedArgs),
    
    %% 3）检查类型兼容性（如 Quorum 队列必须是 durable=true, auto_delete=false）
    case rabbit_queue_type:is_compatible(Type, Durable, AutoDelete, Owner =/= none) of
        false -> {error, incompatible_queue_arguments};
        true ->
            %% 4）构造 amqqueue 记录
            Q0 = amqqueue:new(QueueName, Owner, Durable, AutoDelete, NormalizedArgs),
            Q = amqqueue:set_type(Q0, Type),
            
            %% 5）委托给队列类型模块声明（事务：检查冲突、写入 Mnesia）
            rabbit_queue_type:declare(Q, Node)
    end.

sequenceDiagram
    autonumber
    participant Client
    participant Channel as rabbit_channel
    participant AM as rabbit_amqqueue
    participant QT as rabbit_queue_type
    participant CQ as rabbit_classic_queue<br/>(或 Quorum/Stream)
    participant DB as rabbit_db_queue
    
    Client->>Channel: queue.declare(durable, args)
    Channel->>AM: declare(QueueName, Durable, Args)
    AM->>QT: get_queue_type(Args)
    QT-->>AM: rabbit_classic_queue
    AM->>QT: is_compatible(Type, Durable, AutoDelete)
    QT-->>AM: true
    AM->>CQ: declare(Q, Node)
    CQ->>DB: insert_or_update(Q)
    DB-->>CQ: {ok, NewQ}
    CQ-->>AM: {new, NewQ}
    AM-->>Channel: {new, NewQ}
    Channel->>Channel: monitor_queue(Q)
    Channel-->>Client: queue.declare_ok

-type deliver_options() :: #{
    correlation := term(),     % 关联 ID，用于 Publisher Confirms
    mandatory := boolean(),    % 若无队列可路由则返回 basic.return
    delivery_mode := 1 | 2     % 1=transient, 2=persistent
}.

-type deliver_result() ::
    ok |                        % 所有队列投递成功
    {delivered, [amqqueue:amqqueue()]} |  % 部分队列投递（其余崩溃/停止）
    {unroutable, mc:state()}.   % 没有队列接收（mandatory=true 时触发 basic.return）

deliver(Qs, Msg, Options) ->
    %% 经典队列：同步投递到队列进程
    Correlation = maps:get(correlation, Options),
    lists:foldl(
        fun(Q, Acc) ->
            QPid = amqqueue:get_pid(Q),
            case is_process_alive(QPid) of
                true ->
                    %% 发送消息到队列进程的 mailbox
                    QPid ! {deliver, Msg, Correlation},
                    [Q | Acc];
                false ->
                    %% 队列崩溃，跳过
                    Acc
            end
        end, [], Qs).

deliver(Qs, Msg, Options) ->
    %% 仲裁队列：通过 Ra 客户端异步提交日志
    Correlation = maps:get(correlation, Options),
    lists:foldl(
        fun(Q, Acc) ->
            RaName = amqqueue:get_pid(Q),  % Ra 集群标识 {Name, Node}
            Cmd = rabbit_fifo:make_enqueue(undefined, Msg),
            case ra:pipeline_command(RaName, Cmd, Correlation) of
                ok -> [Q | Acc];
                {error, _} -> Acc
            end
        end, [], Qs).

sequenceDiagram
    autonumber
    participant Channel as rabbit_channel
    participant QT as rabbit_queue_type
    participant CQ as Classic Queue Process
    participant VQ as rabbit_variable_queue
    participant Index as queue_index
    
    Channel->>QT: deliver([Q1, Q2], Msg, Options)
    QT->>CQ: QPid ! {deliver, Msg, Correlation}
    CQ->>VQ: publish(Msg, MsgProps, Correlation)
    VQ->>Index: publish(MsgId, SeqId, MsgProps)
    Index-->>VQ: ok
    VQ->>VQ: 将消息加入 q3（内存队列）或 delta（磁盘队列）
    VQ-->>CQ: {ok, State1}
    CQ->>CQ: 检查是否有待投递消费者
    CQ->>CQ: 立即投递或排队等待 consume
    CQ-->>QT: ok
    QT-->>Channel: {delivered, [Q1, Q2]}

sequenceDiagram
    autonumber
    participant Channel
    participant QQ as rabbit_quorum_queue
    participant Ra as Ra Consensus
    participant Fifo as rabbit_fifo<br/>(State Machine)
    participant Follower1
    participant Follower2
    
    Channel->>QQ: deliver([Q], Msg, Opts)
    QQ->>Ra: pipeline_command(enqueue, Msg, Corr)
    Ra->>Fifo: apply(enqueue, State)
    Fifo->>Fifo: 将消息加入 messages 队列<br/>分配 RaftIndex
    Fifo-->>Ra: {State1, Effects}
    Ra->>Follower1: AppendEntries RPC（异步复制）
    Ra->>Follower2: AppendEntries RPC
    Follower1-->>Ra: ack
    Follower2-->>Ra: ack
    Note over Ra: 达到 Quorum（2/3）后提交
    Ra-->>QQ: {ok, Correlation}
    QQ-->>Channel: {delivered, [Q]}

-record(basic_consume, {
    queue :: binary(),          % 队列名
    consumer_tag :: binary(),   % 消费者标签（客户端提供或自动生成）
    no_ack :: boolean(),        % true=自动确认，false=手动确认
    exclusive :: boolean(),     % true=独占消费（单消费者模式）
    arguments :: rabbit_framing:amqp_table()  % 扩展参数（如 x-priority）
}).

consume(Q, #basic_consume{consumer_tag = CTag, no_ack = NoAck, exclusive = Exclusive}, QState) ->
    QPid = amqqueue:get_pid(Q),
    %% 1）向队列进程发送 consume 请求
    case gen_server2:call(QPid, {basic_consume, NoAck, self(), CTag, Exclusive, Args}, infinity) of
        ok ->
            %% 2）将消费者记录到 Channel 状态
            Consumer = #consumer{tag = CTag, ack_required = not NoAck},
            NewState = QState#{consumers => maps:put(CTag, Consumer, maps:get(consumers, QState))},
            {ok, NewState};
        {error, exclusive_consume_unavailable} ->
            {error, <<"queue is exclusively consumed">>}
    end.

handle_call({basic_consume, NoAck, ChPid, CTag, Exclusive, Args}, _From, State) ->
    %% 1）检查是否独占冲突
    case Exclusive andalso has_consumers(State) of
        true -> {reply, {error, exclusive_consume_unavailable}, State};
        false ->
            %% 2）创建消费者记录
            Consumer = #consumer{
                channel_pid = ChPid,
                tag = CTag,
                ack_required = not NoAck,
                prefetch_count = get_prefetch_count(Args)
            },
            %% 3）添加到消费者列表
            State1 = add_consumer(CTag, Consumer, State),
            %% 4）立即尝试投递消息
            State2 = deliver_msgs_to_consumers(State1),
            {reply, ok, State2}
    end.

sequenceDiagram
    autonumber
    participant Client
    participant Channel as rabbit_channel
    participant QProc as Queue Process
    participant VQ as variable_queue
    
    Client->>Channel: basic.consume(queue, tag, no_ack)
    Channel->>QProc: gen_server2:call({basic_consume, ...})
    QProc->>QProc: 检查独占冲突
    QProc->>QProc: add_consumer(Tag, Consumer)
    QProc->>VQ: fetch(AckRequired)
    VQ-->>QProc: {Msg, State1} | empty
    alt 有消息可投递
        QProc->>Channel: deliver_to_consumer(Msg)
        Channel-->>Client: basic.deliver(Msg)
    end
    QProc-->>Channel: ok
    Channel-->>Client: basic.consume_ok(tag)

publish(Msg, MsgProps, IsDelivered, ChPid, Flow, State) ->
    %% 1）检查消息是否重复（基于 Msg ID 去重）
    case is_duplicate(Msg, State) of
        {true, State1} -> {ok, State1};  % 幂等：已发布过，跳过
        {false, State1} ->
            %% 2）分配序列号（SeqId）
            SeqId = State1#vqstate.next_seq_id,
            State2 = State1#vqstate{next_seq_id = SeqId + 1},
            
            %% 3）持久化消息（如果 durable=true 且 delivery_mode=2）
            State3 = case needs_confirm(MsgProps, State2) of
                true ->
                    %% 记录未确认消息，用于 Publisher Confirms
                    track_confirm(Msg, SeqId, ChPid, State2);
                false -> State2
            end,
            
            %% 4）写入索引（将 SeqId -> MsgId 映射持久化）
            State4 = publish_to_index(Msg, SeqId, MsgProps, State3),
            
            %% 5）根据内存压力决定消息位置
            State5 = case should_keep_in_ram(State4) of
                true ->
                    %% 消息保留在 q3（内存队列）
                    enqueue_message(q3, Msg, SeqId, State4);
                false ->
                    %% 消息直接写入 delta（磁盘队列）
                    enqueue_message(delta, Msg, SeqId, State4)
            end,
            
            {ok, State5}
    end.

should_keep_in_ram(#vqstate{ram_msg_count = RamCount, target_ram_count = Target}) ->
    %% 目标内存消息数由消费速率动态调整（最少1条，最多2048条）
    RamCount < Target.

classDiagram
    class amqqueue {
        +name: resource()
        +durable: boolean()
        +auto_delete: boolean()
        +exclusive_owner: none | pid()
        +arguments: amqp_table()
        +pid: pid()
        +type: atom()
        +vhost: binary()
        +state: live | stopped
        +policy: binary()
        +operator_policy: binary()
        +decorators: [atom()]
        +slave_pids: [pid()]
        +sync_slave_pids: [pid()]
        +gm_pids: [pid()]
    }
    
    note for amqqueue "持久化到 Mnesia 的队列元数据\n经典队列：pid 是本地进程\n仲裁队列：pid 是 {RaName, Node} 元组"

-record(vqstate, {
    %% 消息队列（简化版：q1/q2/q4 已废弃）
    delta,              % 磁盘队列：{DeltaCount, StartSeqId, EndSeqId, PendingAcks}
    q3,                 % 内存队列：lqueue（双端队列）
    
    %% 序列号管理
    next_seq_id,        % 下一个消息的 SeqId
    next_deliver_seq_id,% 首次投递的 SeqId 分界线（用于判断 redelivered）
    
    %% 未确认消息
    ram_pending_ack,    % 内存中的未确认消息：#{SeqId => Msg}
    disk_pending_ack,   % 磁盘上的未确认消息：#{SeqId => {MsgId, IsDelivered}}
    
    %% 索引与存储
    index_state,        % rabbit_classic_queue_index_v2 的状态
    store_state,        % rabbit_classic_queue_store_v2 的状态
    msg_store_clients,  % {PersistentClient, TransientClient} 到共享 msg_store 的连接
    
    %% 统计信息
    len,                % 队列长度（不含未确认）
    bytes,              % 队列字节数
    persistent_count,   % 持久化消息数
    ram_msg_count,      % 内存中的消息数
    target_ram_count,   % 目标内存消息数（根据消费速率动态调整：1~2048）
    
    %% 流控
    rates,              % 入队/出队速率统计
    durable             % 是否持久化队列
}).

classDiagram
    class vqstate {
        +delta: delta_queue()
        +q3: lqueue()
        +next_seq_id: integer()
        +ram_pending_ack: map()
        +disk_pending_ack: map()
        +index_state: index_state()
        +len: integer()
        +target_ram_count: 1..2048
    }
    
    class msg_status {
        +seq_id: integer()
        +msg_id: binary()
        +msg: mc:state()
        +is_persistent: boolean()
        +is_delivered: boolean()
        +msg_props: map()
    }
    
    class delta_queue {
        +count: integer()
        +start_seq_id: integer()
        +end_seq_id: integer()
    }
    
    vqstate --> msg_status : q3 包含
    vqstate --> delta_queue
    vqstate --> index_state

-record(rabbit_fifo, {
    cfg :: #cfg{},                      % 配置：名称、资源名、死信交换器
    messages = rabbit_fifo_q:new(),     % 待投递消息队列（加权优先级队列）
    messages_total = 0,                 % 总消息数
    returns = lqueue:new(),             % 被 nack/reject 返回的消息
    enqueue_count = 0,                  % 入队计数器（用于触发快照）
    
    %% 生产者管理
    enqueuers = #{},                    % #{Pid => #enqueuer{}}，跟踪活跃发布者
    
    %% Raft 索引管理
    ra_indexes = rabbit_fifo_index:empty(), % 已投递消息的 Raft Index 集合
    
    %% 消费者管理
    consumers = #{},                    % #{ConsumerKey => #consumer{}}
    service_queue = priority_queue:new(), % 待服务的消费者队列
    
    %% 死信队列
    dlx = rabbit_fifo_dlx:init(),       % 死信交换器状态
    
    %% 统计
    msg_bytes_enqueue = 0,              % 入队字节数
    msg_bytes_checkout = 0,             % 已投递字节数
    
    %% 单活消费者（SAC）
    waiting_consumers = []              % SAC 模式下的等待消费者列表
}).

-record(consumer, {
    meta = #{},                  % 元数据：CTag, ChPid, Prefetch
    credit = 0,                  % 剩余信用额度（QoS prefetch）
    checked_out = #{},           % 未确认消息：#{RaftIndex => {SeqId, Msg}}
    next_msg_id = 0,             % 下一个消息序列号
    status = up,                 % up | suspected_down | cancelled
    lifetime = once              % once（断开即取消） | auto（自动恢复）
}).

classDiagram
    class rabbit_fifo {
        +cfg: config()
        +messages: rabbit_fifo_q()
        +returns: lqueue()
        +enqueuers: map()
        +consumers: map()
        +ra_indexes: fifo_index()
        +dlx: dlx_state()
    }
    
    class consumer {
        +credit: integer()
        +checked_out: map()
        +next_msg_id: integer()
        +status: up | down
    }
    
    class enqueuer {
        +msg_seqs: [RaftIndex]
        +pending_confirms: [SeqId]
    }
    
    rabbit_fifo --> consumer
    rabbit_fifo --> enqueuer

-record(stream_client, {
    stream_id :: string(),          % 流名称（内部标识）
    name :: rabbit_amqqueue:name(), % 队列资源名
    leader :: pid(),                % Osiris Leader PID
    local_pid :: undefined | pid(), % 本地副本 PID（如果有）
    next_seq = 1,                   % 下一个发布序列号
    correlation = #{}               % #{AppenderSeq => {Correlation, Msg}}（Publisher Confirms）
}).

-record(stream, {
    mode :: simple | automatic,     % 消费模式：simple（手动 ack）| automatic（自动）
    credit :: integer(),            % 剩余信用额度
    ack :: boolean(),               % 是否需要确认
    start_offset = 0,               % 起始偏移量
    listening_offset = 0,           % 当前监听偏移量
    log :: osiris_log:state(),      % Osiris 日志句柄
    chunk_iterator :: chunk_iter(), % 块迭代器（用于批量读取）
    buffer_msgs_rev = []            % 预读消息缓冲（逆序）
}).

sequenceDiagram
    autonumber
    participant Ch as Channel
    participant Ex as Exchange
    participant QProc as Queue Process
    participant VQ as variable_queue
    participant Index as queue_index
    participant Store as msg_store
    
    Ch->>Ex: route(Msg, RoutingKey)
    Ex-->>Ch: [Q1, Q2]
    Ch->>QProc: QPid ! {deliver, Msg, Corr}
    QProc->>VQ: publish(Msg, Props)
    VQ->>VQ: 分配 SeqId = 1001
    VQ->>Index: publish(SeqId, MsgId, Props)
    Index-->>VQ: ok
    alt 消息 >= 4KB
        VQ->>Store: write(MsgId, Msg)
        Store-->>VQ: ok
    end
    VQ->>VQ: should_keep_in_ram() ?
    alt ram_msg_count < target
        VQ->>VQ: q3 = lqueue:in(#msg_status{}, q3)
    else
        VQ->>VQ: delta = increment(delta)
    end
    VQ-->>QProc: {ok, State1}
    QProc->>QProc: 检查待投递消费者
    QProc-->>Ch: （异步 Publisher Confirm）

sequenceDiagram
    autonumber
    participant Ch as Channel
    participant Leader as Ra Leader
    participant Fifo as rabbit_fifo
    participant F1 as Follower1
    participant F2 as Follower2
    
    Ch->>Leader: pipeline_command(enqueue, Msg)
    Leader->>Leader: 写入本地 WAL<br/>{Index=100, Term=5, Cmd}
    Leader->>Fifo: apply(enqueue, State)
    Fifo->>Fifo: messages = fifo_q:in(Msg)
    Fifo-->>Leader: {State1, Effects}
    
    par 异步复制
        Leader->>F1: AppendEntries(Index=100, Cmd)
        Leader->>F2: AppendEntries(Index=100, Cmd)
    end
    
    F1->>F1: 写入 WAL
    F1-->>Leader: ack(Index=100)
    F2->>F2: 写入 WAL
    F2-->>Leader: ack(Index=100)
    
    Note over Leader: 达到 Quorum（2/3）
    Leader->>Leader: commit_index = 100
    Leader->>Ch: {confirm, Seq, RaftIndex}
    Ch-->>Ch: 标记消息已确认

sequenceDiagram
    autonumber
    participant Client
    participant Ch as Channel
    participant QProc as Queue Process
    participant VQ as variable_queue
    
    Client->>Ch: basic.consume(queue, tag)
    Ch->>QProc: gen_server2:call({basic_consume})
    QProc->>QProc: add_consumer(Tag, Consumer)
    
    loop 持续投递
        QProc->>VQ: fetch(AckRequired=true)
        VQ->>VQ: lqueue:out(q3)
        alt q3 有消息
            VQ-->>QProc: {{Msg, IsDelivered=false, AckTag}}
        else q3 为空
            VQ->>VQ: fetch_from_delta()
            VQ-->>QProc: {{Msg, IsDelivered=true, AckTag}}
        end
        QProc->>Ch: ChPid ! {deliver, Tag, Msg}
        Ch->>Client: basic.deliver(Tag, Msg)
        Ch->>Ch: 记录 unacked_message_q[DeliveryTag]
        
        alt 客户端发送 ack
            Client->>Ch: basic.ack(DeliveryTag)
            Ch->>QProc: gen_server2:cast({ack, [AckTag]})
            QProc->>VQ: ack([AckTag])
            VQ->>VQ: 从 ram_pending_ack 删除
            VQ-->>QProc: ok
        end
    end