Implemented RewriteLog functionality and defined handler for REWRITE AOF command.

src/modules/admin/commands.go

@@ -71,7 +71,10 @@ func NewModule() Plugin {
 					return []string{}, nil
 				},
 				HandlerFunc: func(ctx context.Context, cmd []string, server utils.Server, conn *net.Conn) ([]byte, error) {
-					return nil, errors.New("BGREWRITEAOF command not implemented")
+					if err := server.RewriteAOF(); err != nil {
+						return nil, err
+					}
+					return []byte(utils.OK_RESPONSE), nil
 				},
 			},
 		},

src/server/aof/aof.go

@@ -1,6 +1,7 @@
 package aof
 
 import (
+	"encoding/json"
 	"github.com/echovault/echovault/src/utils"
 	"log"
 	"os"
@@ -12,7 +13,8 @@ import (
 // Logging in clusters is handled in the raft layer.
 
 type Opts struct {
-	Config utils.Config
+	Config   utils.Config
+	GetState func() map[string]interface{}
 }
 
 type Engine struct {
@@ -60,10 +62,41 @@ func (engine *Engine) LogCommand(command []byte) error {
 }
 
 func (engine *Engine) RewriteLog() error {
+	engine.mut.Lock()
+	defer engine.mut.Unlock()
+
 	// Get current state.
+	state := engine.options.GetState()
+	o, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+
 	// Replace snapshot contents file with current state.
-	// Close snapshot file.
+	sf, err := os.Create(path.Join(engine.options.Config.DataDir, "aof", "snapshot.bin"))
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err = sf.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+	if _, err = sf.Write(o); err != nil {
+		return err
+	}
+
 	// Replace aof file with empty file.
+	aof, err := os.Create(path.Join(engine.options.Config.DataDir, "aof", "log.aof"))
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err = aof.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+
 	return nil
 }
 

src/server/server.go

@@ -242,7 +242,8 @@ func (server *Server) Start(ctx context.Context) {
 	} else {
 		// Initialize standalone AOF engine
 		server.AOFEngine = aof.NewAOFEngine(aof.Opts{
-			Config: conf,
+			Config:   conf,
+			GetState: server.GetState,
 		})
 		// Initialize and start standalone snapshot engine
 		server.SnapshotEngine = snapshot.NewSnapshotEngine(snapshot.Opts{
@@ -295,6 +296,11 @@ func (server *Server) GetLatestSnapshot() int64 {
 	return server.LatestSnapshotMilliseconds.Load()
 }
 
+func (server *Server) RewriteAOF() error {
+	// TODO: Make this concurrent
+	return server.AOFEngine.RewriteLog()
+}
+
 func (server *Server) ShutDown(ctx context.Context) {
 	if server.IsInCluster() {
 		server.raft.RaftShutdown(ctx)

src/utils/types.go

@@ -23,6 +23,7 @@ type Server interface {
 	FinishSnapshot()
 	SetLatestSnapshot(msec int64)
 	GetLatestSnapshot() int64
+	RewriteAOF() error
 }
 
 type ContextServerID string