[raft/rid] Raftstore skeleton

Author: MariemBaccari

pkg/rid/store/raftstore/store.go

@@ -3,37 +3,69 @@ package raftstore
 import (
 	"context"
 
-	dsserr "github.com/interuss/dss/pkg/errors"
+	"github.com/interuss/dss/pkg/memstore"
 	"github.com/interuss/dss/pkg/raftstore"
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	"github.com/interuss/dss/pkg/rid/actions"
 	"github.com/interuss/dss/pkg/rid/repos"
+	ridmemstore "github.com/interuss/dss/pkg/rid/store/memstore"
 	ridraftparams "github.com/interuss/dss/pkg/rid/store/raftstore/params"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
 
 // repo is a full implementation of rid.repos.Repository for Raft-based storage.
-type repo struct{}
+type repo struct {
+	consensus *consensus.Consensus
+	memStore  *memstore.Store[repos.Repository]
+	memRepo   repos.Repository
+}
 
 func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) {
 	params, err := ridraftparams.GetConnectParameters()
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to get rid raft parameters")
 	}
-	return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, &repo{}, actions.Registry)
+	memStore, err := ridmemstore.Init(ctx, logger)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to initialize rid memstore")
+	}
+
+	r := &repo{memStore: memStore, memRepo: memStore.GetRepo()}
+	store, err := raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, r, actions.Registry)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to initialize rid raftstore")
+	}
+
+	r.consensus = store.Consensus
+
+	return store, nil
 }
 
 func (r *repo) GetRepo() repos.Repository { return r }
 
 func (r *repo) GetSnapshot() ([]byte, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+	return r.memStore.GetSnapshot()
 }
 
-func (r *repo) RestoreFromSnapshot([]byte) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+func (r *repo) RestoreFromSnapshot(data []byte) error {
+	return r.memStore.RestoreFromSnapshot(data)
 }
 
-func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, error) {
+	switch proposal.RequestType {
+
+	default:
+		handler, ok := actions.Registry[string(proposal.RequestType)]
+		if !ok {
+			return nil, stacktrace.NewError("unrecognized request type: %s", proposal.RequestType)
+		}
+
+		request, err := handler.Decode(proposal.Value)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "failed to decode %s payload", proposal.RequestType)
+		}
+
+		return handler.Execute(ctx, r.memRepo, request)
+	}
 }