diff --git a/pkg/rid/store/raftstore/store.go b/pkg/rid/store/raftstore/store.go index bc55a3dca..7fd67df32 100644 --- a/pkg/rid/store/raftstore/store.go +++ b/pkg/rid/store/raftstore/store.go @@ -3,37 +3,70 @@ 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) + } }