type RegistrationDB struct {
    sync.RWMutex
    registrationMap map[iface.IRegistration]iface.ProducerMap
}

type Message struct {
   ID        iface.MessageID `json:"ID"`
   Data      []byte          `json:"Data"`
   Timestamp int64           `json:"Timestamp"`
   Attempts  uint16          `json:"Attempts"`

   // 优先队列中使用到的数据结构
   clientID uint64
   pri      int64
   index    int
}

type Topic struct {
   lmqd iface.ILmqDaemon

   name         string
   isTemporary  bool                      // 标记是否是临时的topic
   isPausing    atomic.Bool               // 标记是否已经暂停
   isExiting    atomic.Bool               // 标记是否已经退出
   channels     map[string]iface.IChannel // 保存所有的channel字典
   channelsLock sync.RWMutex              // 控制对channel字典的互斥访问

   guidFactory *GUIDFactory // message id 生成器

   memoryMsgChan chan iface.IMessage       // 内存chan
   backendQueue  backendqueue.BackendQueue // 当内存chan满了之后，将消息存入到后端队列中（持久化保存）

   deleteCallback func(topic iface.ITopic)
   deleter        sync.Once

   startChan   chan struct{}
   updateChan  chan struct{}
   pauseChan   chan struct{}
   closingChan chan struct{}
   closedChan  chan struct{}

   messageCount atomic.Uint64
   messageBytes atomic.Uint64
}

// Coordinator TCC事务协调者
type Coordinator struct {
    id      string                        // 事务id
    dbIndex int                           // 数据库id
    self    string                        // 自己的地址
    peers   cluster.PeerPicker            // 一致性哈希，用于选择节点
    getters map[string]cluster.PeerGetter // 用于和远程节点通信

    watching map[string]map[string]uint32 // key为peer地址，记录被watch的key的版本号

    cmdLinesNum int
    indexMap    map[string][]int // key为peer，value为cmdline在原来multi队列中的下标，用于重组reply
}

// ExecTx 执行TCC分布式事务
func (coordinator *Coordinator) ExecTx(cmdLines [][][]byte, watching map[string]uint32) redis.Reply {
    // 首先对命令进行分组，以同一个节点上执行的所有命令为一组
    groupByMap := coordinator.groupByCmdLines(cmdLines)

    // 对watch的key进行分组
    coordinator.groupByWatch(watching)

    // 对每一组中发送try命令
    needCancel := false // 记录是否需要回滚
    for peer, cmd := range groupByMap {
       r := coordinator.sendTry(peer, cmd)
       if reply.IsErrorReply(r) {
          needCancel = true
          break
       }
    }

    defer func() {
       // 返回之前发送end消息，通知各个分布式节点结束事务
       for peer := range groupByMap {
          coordinator.sendEnd(peer)
       }
    }()

    // 向各个节点提交/回滚分布式事务
    replies := make(map[string]redis.Reply)
    for peer := range groupByMap {
       if needCancel {
          // 取消事务
          coordinator.sendCancel(peer)
       } else {
          // 提交事务
          r := coordinator.sendCommit(peer)
          if config.Properties.OpenAtomicTx && reply.IsErrorReply(r) {
             // 如果开启了原子事务并且提交发生了错误，则进行所有的节点进行回滚
             needCancel = true
             break
          }
          replies[peer] = r
       }
    }

    // 如果开启了原子事务，并且发生了错误，所有节点进行回滚
    if needCancel && config.Properties.OpenAtomicTx {
       for peer := range groupByMap {
          // 取消事务，进行回滚
          coordinator.sendCancel(peer)
       }

       return reply.MakeErrReply("EXECABORT Transaction rollback because of errors during executing. (atomic tx is open)")
    }

    if needCancel {
       // 没有开启原子事务，但是发送了错误（如watch的key变化了），则返回(nil)
       return reply.MakeNullBulkStringReply()
    }

    // 正常提交，重组reply
    return coordinator.recombineReplies(replies)
}

func (coordinator *Coordinator) sendTry(peer string, cmdLines [][][]byte) redis.Reply {
    // 获取getter
    getter := coordinator.getters[peer]

    var r redis.Reply
    // 发送try开始命令
    r = getter.RemoteExec(coordinator.dbIndex, utils.StringsToCmdLine("try", coordinator.id, "start"))
    if reply.IsErrorReply(r) {
       // 如果发生错误，则中断try，直接返回
       return r
    }

    // 发送watch时的version，比较version是否发生变化
    watching := coordinator.watching[peer]
    for key, version := range watching {
       versionStr := strconv.FormatInt(int64(version), 10)
       r = getter.RemoteExec(coordinator.dbIndex, utils.StringsToCmdLine("try", coordinator.id, "watched", key, versionStr))
       if reply.IsErrorReply(r) {
          // 如果发生变化，则中断try，直接返回
          return r
       }
    }

    // 依次发送需要执行的命令，每一条命令=try tx_id cmdline，如 try 123456 set k1 v1
    for _, cmdLine := range cmdLines {
       tryCmd := make([][]byte, 0, 2+len(cmdLine))
       tryCmd = append(tryCmd, utils.StringsToCmdLine("try", coordinator.id)...)
       tryCmd = append(tryCmd, cmdLine...)
       r = getter.RemoteExec(coordinator.dbIndex, tryCmd)
       if reply.IsErrorReply(r) {
          // 如果发生错误，则中断try，直接返回
          return r
       }
    }

    // 发送try结束命令
    r = getter.RemoteExec(coordinator.dbIndex, utils.StringsToCmdLine("try", coordinator.id, "end"))
    if reply.IsErrorReply(r) {
       // 如果发生错误，则中断try，直接返回
       return r
    }

    return reply.MakeOkReply()
}

// Transaction TCC本地事务
type Transaction struct {
    id    string // 事务id
    phase int    // 阶段，分别是try、commit、cancel三个阶段

    writeKeys   []string
    readKeys    []string
    watchedKeys map[string]uint32 // 记录watch时的版本

    cmdLines       [][][]byte   // 本次事务中包含的命令
    undoLogs       [][][][]byte // 本次事务的undo log
    addVersionKeys []string     // 需要增加版本的key

    db *engine.DB

    mu sync.Mutex
}

// Try TCC事务的结束try阶段，锁定key
func (tx *Transaction) Try() redis.Reply {
    if tx.phase != phaseTry {
       return reply.MakeErrReply("ERR TRY ERROR")
    }

    // 锁定需要读写的key
    tx.db.RWLocks(tx.writeKeys, tx.readKeys)

    // 在时间轮中添加任务, 自动回滚超时未提交的事务
    taskKey := tx.id
    timewheel.Delay(maxLockTime, taskKey, func() {
       if tx.phase == phaseTry {
          logger.Info("abort transaction: " + tx.id)
          _ = tx.Cancel()
       }
    })

    // 读取版本是否变化
    for key, oldVersion := range tx.watchedKeys {
       newVersion := tx.db.GetVersion(key)
       if oldVersion != newVersion {
          // 获取key的版本，如果版本变化了，则返回错误
          return reply.MakeErrReply("ERR VERSION CHANGED")
       }
    }

    return reply.MakeOkReply()
}

// Commit TCC事务的commit阶段
func (tx *Transaction) Commit() redis.Reply {
    tx.mu.Lock() // Commit互斥进行操作，防止commit时调用cancel，发生冲突
    defer tx.mu.Unlock()
    defer func() {
       // 设置阶段
       tx.phase = phaseCommit
    }()

    // 依次执行命令
    var results [][]byte // 存储命令执行的结果
    for _, cmdLine := range tx.cmdLines {
       cmdName := string(cmdLine[0])
       key := string(cmdLine[1])
       if !engine.IsReadOnlyCommand(cmdName) {
          // 写命令，则记录需要add版本的key
          tx.addVersionKeys = append(tx.addVersionKeys, key)
          if config.Properties.OpenAtomicTx {
             // 开启了原子性事务，则记录undo log
             tx.undoLogs = append(tx.undoLogs, tx.db.GetUndoLog(key))
          }
       }

       // 执行命令
       r := tx.db.ExecWithLock(cmdLine)

       if config.Properties.OpenAtomicTx && reply.IsErrorReply(r) {
          // 开启了原子性事务，并且如果发生错误，直接返回错误
          if !engine.IsReadOnlyCommand(cmdName) {
             // 删除错误命令的undo log，不需要增加版本号
             tx.addVersionKeys = tx.addVersionKeys[:len(tx.addVersionKeys)-1]
             tx.undoLogs = tx.undoLogs[:len(tx.undoLogs)-1]
          }

          return r
       }

       results = append(results, []byte(r.DataString()))
    }

    return reply.MakeMultiBulkStringReply(results)
}

// Cancel TCC事务的cancel阶段
func (tx *Transaction) Cancel() redis.Reply {
    tx.mu.Lock() // Cancel互斥进行操作，防止commit时调用cancel，发生冲突
    defer tx.mu.Unlock()
    defer func() {
       // 设置阶段
       tx.phase = phaseCancel
    }()

    // 从后向前依次执行undo日志，只有开启原子性事务时，len(tx.undoLogs)才可能大于0
    for i := len(tx.undoLogs) - 1; i >= 0; i-- {
       undoLog := tx.undoLogs[i]
       if len(undoLog) == 0 {
          continue
       }

       for _, cmdLine := range undoLog {
          tx.db.ExecWithLock(cmdLine)
       }

    }

    // 清空
    tx.undoLogs = nil

    return reply.MakeOkReply()
}