// 存儲引擎，負責數(shù)據(jù)的CRUD
type Engine struct {
	// *DB
	dbSet []*atomic.Value
	// 時間輪(延遲任務(wù))
	delay *timewheel.Delay
	// Append Only File
	aof *aof.AOF
}

func NewEngine() *Engine {
	//.....省略....
	// 啟用AOF日志
	if conf.GlobalConfig.AppendOnly {
		// 創(chuàng)建*AOF對象
		aof, err := aof.NewAOF(conf.GlobalConfig.AppendFilename, engine, true, conf.GlobalConfig.AppendFsync)
		if err != nil {
			panic(err)
		}
		engine.aof = aof
		// 設(shè)定每個db，使用aof寫入日志
		engine.aofBindEveryDB()
	}
	return engine
}

func (e *Engine) aofBindEveryDB() {
	for _, dbSet := range e.dbSet {
		db := dbSet.Load().(*DB)
		db.writeAof = func(redisCommand [][]byte) {
			if conf.GlobalConfig.AppendOnly {
                // 調(diào)用e.aof對象方法，保存命令
				e.aof.SaveRedisCommand(db.index, aof.Command(redisCommand))
			}
		}
	}
}

func cmdSet(db *DB, args [][]byte) protocol.Reply {
    //.....省略....
	if result > 0 { // 1 表示存儲成功
		//TODO： 過期時間處理
		if ttl != nolimitedTTL { // 設(shè)定key過期
			expireTime := time.Now().Add(time.Duration(ttl) * time.Millisecond)
			db.ExpireAt(key, expireTime)
			//寫入日志
			db.writeAof(aof.SetCmd([][]byte{args[0], args[1]}...))
			db.writeAof(aof.PExpireAtCmd(string(args[0]), expireTime))
		} else { // 設(shè)定key不過期
			db.Persist(key)
			//寫入日志
			db.writeAof(aof.SetCmd(args...))
		}
		return protocol.NewOkReply()
	}
	return protocol.NewNullBulkReply()
}

func (aof *AOF) SaveRedisCommand(dbIndex int, command Command) {
	// 關(guān)閉
	if aof.atomicClose.Load() {
		return
	}
	// 寫入文件 & 刷盤
	if aof.aofFsync == FsyncAlways {
		record := aofRecord{
			dbIndex: dbIndex,
			command: command,
		}
		aof.writeAofRecord(record)
		return
	}
	// 寫入緩沖
	aof.aofChan <- aofRecord{
		dbIndex: dbIndex,
		command: command,
	}
}

func (aof *AOF) watchChan() {
	for record := range aof.aofChan {
		aof.writeAofRecord(record)
	}
	aof.aofFinished <- struct{}{}
}

func (aof *AOF) writeAofRecord(record aofRecord) {
	aof.mu.Lock()
	defer aof.mu.Unlock()
	// 因為aof對象是所有數(shù)據(jù)庫對象【復用】寫入文件方法，每個數(shù)據(jù)庫的索引不同
	// 所以，每個命令的執(zhí)行，有個前提就是操作的不同的數(shù)據(jù)庫
	if record.dbIndex != aof.lastDBIndex {
		// 構(gòu)建select index 命令 & 寫入文件
		selectCommand := [][]byte{[]byte("select"), []byte(strconv.Itoa(record.dbIndex))}
		data := protocol.NewMultiBulkReply(selectCommand).ToBytes()
		_, err := aof.aofFile.Write(data)
		if err != nil {
			logger.Warn(err)
			return
		}
		aof.lastDBIndex = record.dbIndex
	}
	// redis命令
	data := protocol.NewMultiBulkReply(record.command).ToBytes()
	_, err := aof.aofFile.Write(data)
	if err != nil {
		logger.Warn(err)
	}
	logger.Debugf("write aof command:%q", data)
	// 每次寫入刷盤
	if aof.aofFsync == FsyncAlways {
		aof.aofFile.Sync()
	}
}

// 構(gòu)建AOF對象
func NewAOF(aofFileName string, engine abstract.Engine, load bool, fsync string) (*AOF, error) {
	//...省略...
	// 啟動加載aof文件
	if load {
		aof.LoadAof(0)
	}
    //...省略...
}

func (aof *AOF) LoadAof(maxBytes int) {
	// 目的：當加載aof文件的時候，因為需要復用engine對象，內(nèi)部重放命令的時候會自動寫aof日志，加載aof 禁用 SaveRedisCommand的寫入
	aof.atomicClose.Store(true)
	deferfunc() {
		aof.atomicClose.Store(false)
	}()
	// 只讀打開文件
	file, err := os.Open(aof.aofFileName)
	if err != nil {
		logger.Error(err.Error())
		return
	}
	defer file.Close()
	file.Seek(0, io.SeekStart)
	var reader io.Reader
	if maxBytes > 0 { // 限定讀取的字節(jié)大小
		reader = io.LimitReader(file, int64(maxBytes))
	} else { // 不限定，直接讀取到文件結(jié)尾（為止）
		reader = file
	}
	// 文件中保存的格式和網(wǎng)絡(luò)傳輸?shù)母袷揭恢?
	ch := parser.ParseStream(reader)
	virtualConn := connection.NewVirtualConn()
	for payload := range ch {
		if payload.Err != nil {
			// 文件已經(jīng)讀取到“完成“
			if payload.Err == io.EOF {
				break
			}
			// 讀取到非法的格式
			logger.Errorf("LoadAof parser error %+v:", payload.Err)
			continue
		}
		if payload.Reply == nil {
			logger.Error("empty payload data")
			continue
		}
		// 從文件中讀取到命令
		reply, ok := payload.Reply.(*protocol.MultiBulkReply)
		if !ok {
			logger.Error("require multi bulk protocol")
			continue
		}
		// 利用數(shù)據(jù)庫引擎，將命令數(shù)據(jù)保存到內(nèi)存中（命令重放）
		ret := aof.engine.Exec(virtualConn, reply.RedisCommand)
		// 判斷是否執(zhí)行失敗
		if protocol.IsErrReply(ret) {
			logger.Error("exec err ", string(ret.ToBytes()))
		}
		// 判斷命令是否是"select"
		if strings.ToLower(string(reply.RedisCommand[0])) == "select" {
			dbIndex, err := strconv.Atoi(string(reply.RedisCommand[1]))
			if err == nil {
				aof.lastDBIndex = dbIndex // 記錄下數(shù)據(jù)恢復過程中，選中的數(shù)據(jù)庫索引
			}
		}
	}
}

func (aof *AOF) Rewrite(engine abstract.Engine) {
	//1.對現(xiàn)有的aof文件做一次快照
	snapShot, err := aof.startRewrite()
	if err != nil {
		logger.Errorf("StartRewrite err: %+v", err)
		return
	}
	//2. 將現(xiàn)在的aof文件數(shù)據(jù)，加在到新（內(nèi)存）對象中,并重寫入新aof文件中
	err = aof.doRewrite(snapShot, engine)
	if err != nil {
		logger.Errorf("doRewrite err: %+v", err)
		return
	}
	//3. 將重寫過程中的增量命令寫入到新文件中
	err = aof.finishRewrite(snapShot)
	if err != nil {
		logger.Errorf("finishRewrite err: %+v", err)
	}
}