// 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()
}

// Cluster 用于和集群中的主机进行交互
type Cluster struct {
    self    string                        // 本机地址，如 127.0.0.1:6107
    peers   cluster.PeerPicker            // 一致性哈希，用于选择节点
    getters map[string]cluster.PeerGetter // 用于和远程节点通信

    idGenerator    *snowflake.Node  // snowflake id生成器，用于生成分布式事务的id
    transactionMap *dict.SimpleDict // 记录所有的分布式事务（本地）
    coordinatorMap *dict.SimpleDict // 记录事务协调者
}

const (
   FsyncAlways   = iota // 每一个命令都会进行刷盘操作
   FsyncEverySec        // 每秒进行一次刷盘操作
   FsyncNo              // 不主动进行刷盘操作，交给操作系统去决定
)

type CmdLine [][]byte

const (
   aofQueueSize = 1 << 16
)

type Persister struct {
   ctx         context.Context
   cancel      context.CancelFunc
   db          database.DBEngine
   tmpDBMaker  func() database.DBEngine
   aofChan     chan *payload
   aofFile     *os.File
   aofFilename string
   aofFsync    int // AOF 刷盘策略
   // aof goroutine will send msg to main goroutine through this channel when aof tasks finished and ready to shut down
   aofFinished chan struct{}
   // pause aof for start/finish aof rewrite progress
   pausingAof sync.Mutex
   // 表示正在aof重写，同时只有一个aof重写
   aofRewriting sync.WaitGroup
   currentDB    int
}

type payload struct {
	cmdLine CmdLine
	dbIndex int
}

type Server struct {
   dbSet        []*atomic.Value // *DB
   AofPersister *aof.Persister  // AOF 持久化
   AofFileSize  int64
   rewriteWait  sync.WaitGroup
   rewriting    atomic.Bool
   closed       chan struct{}
   cluster      *cluster.Cluster
   publish      publish.Publish
}

// Expected check whether given item is equals to expected value
type Expected func(a interface{}) bool

// Consumer traverses list.
// It receives index and value as params, returns true to continue traversal, while returns false to break
type Consumer func(i int, v interface{}) bool

type List interface {
   Add(val interface{})
   Get(index int) (val interface{})
   Set(index int, val interface{})
   Insert(index int, val interface{})
   Remove(index int) (val interface{})
   RemoveLast() (val interface{})
   RemoveAllByVal(expected Expected) int
   RemoveByVal(expected Expected, count int) int
   ReverseRemoveByVal(expected Expected, count int) int
   Len() int
   ForEach(consumer Consumer)
   Contains(expected Expected) bool
   Range(start int, stop int) []interface{}
}

// pageSize must be even
const pageSize = 1024

// QuickList is a linked list of page (which type is []interface{})
// QuickList has better performance than LinkedList of Add, Range and memory usage
type QuickList struct {
   data *list.List // list of []interface{}
   size int
}