[raft/rid] Add rid Raftstore subscriptions

Author: MariemBaccari

pkg/rid/application/subscription.go

@@ -8,15 +8,11 @@ import (
 	dssmodels "github.com/interuss/dss/pkg/models"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
 	"github.com/interuss/dss/pkg/rid/repos"
+	ridraftstore "github.com/interuss/dss/pkg/rid/store/raftstore"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
 
-const (
-	// Defined in requirement DSS0030.
-	maxSubscriptionsPerArea = 10
-)
-
 // SubscriptionApp provides the interface to the application logic for Subscription entities
 // AppInterface provides the interface to the application logic for ISA entities
 // Note that there is no need for the applciation layer to have the same API as
@@ -60,7 +56,7 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription)
 		return nil, stacktrace.Propagate(err, "Unable to adjust time range")
 	}
 	var sub *ridmodels.Subscription
-	_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
+	raftResult, err := a.store.Transact(ctx, ridraftstore.InsertSubscriptionTransaction, s, func(ctx context.Context, repo repos.Repository) error {
 
 		// ensure it doesn't exist yet
 		old, err := repo.GetSubscription(ctx, s.ID)
@@ -78,7 +74,7 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription)
 			return stacktrace.Propagate(err,
 				"Failed to fetch subscription count, rejecting request")
 		}
-		if count >= maxSubscriptionsPerArea {
+		if count >= ridmodels.MaxSubscriptionsPerArea {
 			return stacktrace.Propagate(
 				stacktrace.NewErrorWithCode(dsserr.Exhausted, "Too many existing subscriptions in this area already"),
 				"%s had %d subscriptions in the area", s.Owner, count)
@@ -91,14 +87,23 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription)
 
 		return nil
 	})
+
+	if raftResult != nil {
+		var ok bool
+		sub, ok = raftResult.(*ridmodels.Subscription)
+		if !ok {
+			return nil, stacktrace.NewError("invalid result type: %T", raftResult)
+		}
+	}
+
 	return sub, err
 }
 
 // InsertSubscription implements the App InsertSubscription method
 func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) {
 	var sub *ridmodels.Subscription
 
-	_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
+	raftResult, err := a.store.Transact(ctx, ridraftstore.UpdateSubscriptionTransaction, s, func(ctx context.Context, repo repos.Repository) error {
 		old, err := repo.GetSubscription(ctx, s.ID)
 		switch {
 		case err != nil:
@@ -128,7 +133,7 @@ func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription)
 			return stacktrace.Propagate(err,
 				"Failed to fetch subscription count, rejecting request")
 		}
-		if count >= maxSubscriptionsPerArea {
+		if count >= ridmodels.MaxSubscriptionsPerArea {
 			return stacktrace.Propagate(
 				stacktrace.NewErrorWithCode(dsserr.Exhausted, "Too many existing subscriptions in this area already"),
 				"%s had %d subscriptions in the area", s.Owner, count)
@@ -139,13 +144,26 @@ func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription)
 		}
 		return nil
 	})
+
+	if raftResult != nil {
+		var ok bool
+		sub, ok = raftResult.(*ridmodels.Subscription)
+		if !ok {
+			return nil, stacktrace.NewError("invalid result type: %T", raftResult)
+		}
+	}
+
 	return sub, err
 }
 
 // DeleteSubscription deletes the Subscription identified by "id" and owned by "owner".
 func (a *app) DeleteSubscription(ctx context.Context, id dssmodels.ID, owner dssmodels.Owner, version *dssmodels.Version) (*ridmodels.Subscription, error) {
 	var ret *ridmodels.Subscription
-	_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
+	raftResult, err := a.store.Transact(ctx, ridraftstore.DeleteSubscriptionTransaction, ridraftstore.DeleteSubscriptionPayload{
+		ID:      id,
+		Owner:   owner,
+		Version: version,
+	}, func(ctx context.Context, repo repos.Repository) error {
 		var err error
 		old, err := repo.GetSubscription(ctx, id)
 		switch {
@@ -169,5 +187,14 @@ func (a *app) DeleteSubscription(ctx context.Context, id dssmodels.ID, owner dss
 		}
 		return nil
 	})
+
+	if raftResult != nil {
+		var ok bool
+		ret, ok = raftResult.(*ridmodels.Subscription)
+		if !ok {
+			return nil, stacktrace.NewError("invalid result type: %T", raftResult)
+		}
+	}
+
 	return ret, err
 }

pkg/rid/models/models.go

@@ -1,8 +1,14 @@
 package models
 
 import (
-	"github.com/interuss/stacktrace"
 	"net/url"
+
+	"github.com/interuss/stacktrace"
+)
+
+const (
+	// Defined in requirement DSS0030.
+	MaxSubscriptionsPerArea = 10
 )
 
 // ValidateURL ensures https

pkg/rid/store/raftstore/store.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"slices"
 
+	"github.com/golang/geo/s2"
 	"github.com/interuss/dss/pkg/memstore"
+	dssmodels "github.com/interuss/dss/pkg/models"
 	"github.com/interuss/dss/pkg/raftstore"
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
@@ -30,13 +32,35 @@ const (
 	DeleteISATransaction raftstore.RequestType = "deleteISATransaction"
 	InsertISATransaction raftstore.RequestType = "insertISATransaction"
 	UpdateISATransaction raftstore.RequestType = "updateISATransaction"
+
+	getSubscription                    raftstore.RequestType = "getSubscription"
+	deleteSubscription                 raftstore.RequestType = "deleteSubscription"
+	insertSubscription                 raftstore.RequestType = "insertSubscription"
+	updateSubscription                 raftstore.RequestType = "updateSubscription"
+	searchSubscriptions                raftstore.RequestType = "searchSubscriptions"
+	searchSubscriptionsByOwner         raftstore.RequestType = "searchSubscriptionsByOwner"
+	updateNotificationIdxsInCells      raftstore.RequestType = "updateNotificationIdxsInCells"
+	maxSubscriptionCountInCellsByOwner raftstore.RequestType = "maxSubscriptionCountInCellsByOwner"
+	listExpiredSubscriptions           raftstore.RequestType = "listExpiredSubscriptions"
+	countSubscriptions                 raftstore.RequestType = "countSubscriptions"
+
+	DeleteSubscriptionTransaction raftstore.RequestType = "deleteSubscriptionTransaction"
+	InsertSubscriptionTransaction raftstore.RequestType = "insertSubscriptionTransaction"
+	UpdateSubscriptionTransaction raftstore.RequestType = "updateSubscriptionTransaction"
 )
 
 var readOnlyRequests = []raftstore.RequestType{
 	getISA,
 	searchISAs,
 	listExpiredISAs,
 	countISAs,
+
+	getSubscription,
+	searchSubscriptions,
+	searchSubscriptionsByOwner,
+	maxSubscriptionCountInCellsByOwner,
+	listExpiredSubscriptions,
+	countSubscriptions,
 }
 
 // repo is a full implementation of rid.repos.Repository for Raft-based storage.
@@ -144,6 +168,83 @@ func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, err
 	case UpdateISATransaction:
 		return r.updateISATransactionApplier(ctx, proposal, mem)
 
+	// Subscriptions
+
+	case getSubscription:
+		var id dssmodels.ID
+		if err := json.Unmarshal(proposal.Value, &id); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", getSubscription)
+		}
+		return mem.GetSubscription(ctx, id)
+
+	case deleteSubscription:
+		var sub ridmodels.Subscription
+		if err := json.Unmarshal(proposal.Value, &sub); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", deleteSubscription)
+		}
+		return mem.DeleteSubscription(ctx, &sub)
+
+	case insertSubscription:
+		var sub ridmodels.Subscription
+		if err := json.Unmarshal(proposal.Value, &sub); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", insertSubscription)
+		}
+		return mem.InsertSubscription(ctx, &sub)
+
+	case updateSubscription:
+		var sub ridmodels.Subscription
+		if err := json.Unmarshal(proposal.Value, &sub); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", updateSubscription)
+		}
+		return mem.UpdateSubscription(ctx, &sub)
+
+	case searchSubscriptions:
+		var cells s2.CellUnion
+		if err := json.Unmarshal(proposal.Value, &cells); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", searchSubscriptions)
+		}
+		return mem.SearchSubscriptions(ctx, cells)
+
+	case searchSubscriptionsByOwner:
+		var p searchSubscriptionsByOwnerPayload
+		if err := json.Unmarshal(proposal.Value, &p); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", searchSubscriptionsByOwner)
+		}
+		return mem.SearchSubscriptionsByOwner(ctx, p.Cells, p.Owner)
+
+	case updateNotificationIdxsInCells:
+		var cells s2.CellUnion
+		if err := json.Unmarshal(proposal.Value, &cells); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", updateNotificationIdxsInCells)
+		}
+		return mem.UpdateNotificationIdxsInCells(ctx, cells)
+
+	case maxSubscriptionCountInCellsByOwner:
+		var p maxSubscriptionCountInCellsByOwnerPayload
+		if err := json.Unmarshal(proposal.Value, &p); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", maxSubscriptionCountInCellsByOwner)
+		}
+		return mem.MaxSubscriptionCountInCellsByOwner(ctx, p.Cells, p.Owner)
+
+	case listExpiredSubscriptions:
+		var p listExpiredSubscriptionsPayload
+		if err := json.Unmarshal(proposal.Value, &p); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", listExpiredSubscriptions)
+		}
+		return mem.ListExpiredSubscriptions(ctx, p.Writer, p.Threshold)
+
+	case countSubscriptions:
+		return mem.CountSubscriptions(ctx)
+
+	case DeleteSubscriptionTransaction:
+		return r.deleteSubscriptionTransactionApplier(ctx, proposal, mem)
+
+	case InsertSubscriptionTransaction:
+		return r.insertSubscriptionTransactionApplier(ctx, proposal, mem)
+
+	case UpdateSubscriptionTransaction:
+		return r.updateSubscriptionTransactionApplier(ctx, proposal, mem)
+
 	default:
 		return nil, stacktrace.NewError("unknown request type: %q", proposal.RequestType)
 	}