[raft/rid] Extract InsertSubscription and route Create through the store

cmds/core-service/main.go

@@ -112,6 +112,15 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string
 		return nil, stacktrace.Propagate(err, "Unable to interact with store")
 	}
 
+	ctx = timestamp.NewContext(ctx, time.Now())
+
+	seed, err := random.NewSeed()
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "Unable to generate seed")
+	}
+
+	ctx = random.NewContext(ctx, seed)
+
 	err = repo.SaveOwnMetadata(ctx, locality, publicEndpoint)
 
 	if err != nil {

pkg/random/random.go

@@ -17,7 +17,7 @@ import (
 
 type key struct{}
 
-func newSeed() (int64, error) {
+func NewSeed() (int64, error) {
 	var buf [8]byte
 	_, err := rand.Read(buf[:])
 	if err != nil {
@@ -66,7 +66,7 @@ func Generator(seed int64, label string) (*mrand.Rand, error) {
 // deterministically via Generator.
 func Middleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		seed, err := newSeed()
+		seed, err := NewSeed()
 		if err != nil {
 			http.Error(w, "failed to generate request seed", http.StatusInternalServerError)
 			return

pkg/rid/application/isa_test.go

@@ -11,6 +11,7 @@ import (
 	dsserr "github.com/interuss/dss/pkg/errors"
 	dssmodels "github.com/interuss/dss/pkg/models"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
+	"github.com/interuss/dss/pkg/timestamp"
 	"github.com/interuss/stacktrace"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -95,7 +96,7 @@ func (store *isaStore) CountISAs(ctx context.Context) (int64, error) {
 }
 
 func TestISAUpdateIdxCells(t *testing.T) {
-	ctx := context.Background()
+	ctx := timestamp.NewContext(t.Context(), fakeClock.Now())
 	app, cleanup := setUpISAApp(ctx, t)
 
 	defer cleanup()
@@ -119,21 +120,28 @@ func TestISAUpdateIdxCells(t *testing.T) {
 	// with the soon to be new version of the isa. both should increase their
 	// notification index.
 
-	_, err = app.InsertSubscription(ctx, &ridmodels.Subscription{
-		ID:        dssmodels.ID(uuid.New().String()),
-		Owner:     "owner",
-		StartTime: &startTime,
-		EndTime:   &endTime,
-		Cells:     s2.CellUnion{17106221850767130624, 17106221919486607360},
+	repo, err := app.store.Interact(ctx)
+	require.NoError(t, err)
+
+	_, err = repo.InsertSubscription(ctx, &ridmodels.Subscription{
+		ID:    dssmodels.ID(uuid.New().String()),
+		Owner: "owner",
+		CellsVolume4D: &dssmodels.CellsVolume4D{
+			StartTime: &startTime,
+			EndTime:   &endTime,
+			Cells:     s2.CellUnion{17106221850767130624, 17106221919486607360},
+		},
 	})
 	require.NoError(t, err)
 
-	_, err = app.InsertSubscription(ctx, &ridmodels.Subscription{
-		ID:        dssmodels.ID(uuid.New().String()),
-		Owner:     "owner",
-		StartTime: &startTime,
-		EndTime:   &endTime,
-		Cells:     s2.CellUnion{17106221953846345728},
+	_, err = repo.InsertSubscription(ctx, &ridmodels.Subscription{
+		ID:    dssmodels.ID(uuid.New().String()),
+		Owner: "owner",
+		CellsVolume4D: &dssmodels.CellsVolume4D{
+			StartTime: &startTime,
+			EndTime:   &endTime,
+			Cells:     s2.CellUnion{17106221953846345728},
+		},
 	})
 	require.NoError(t, err)
 
@@ -326,15 +334,18 @@ func TestUpdateISA(t *testing.T) {
 
 func TestAppDeleteISAs(t *testing.T) {
 	var (
-		ctx          = context.Background()
+		ctx          = timestamp.NewContext(t.Context(), fakeClock.Now())
 		app, cleanup = setUpISAApp(ctx, t)
 	)
 	defer cleanup()
 
 	insertedSubscriptions := []*ridmodels.Subscription{}
 	for _, r := range subscriptionsPool {
 		sunscriptionCopy := *r.input
-		s1, err := app.InsertSubscription(ctx, &sunscriptionCopy)
+		repo, err := app.store.Interact(ctx)
+		require.NoError(t, err)
+
+		s1, err := repo.InsertSubscription(ctx, &sunscriptionCopy)
 		require.NoError(t, err)
 		require.NotNil(t, s1)
 		require.Equal(t, 42, s1.NotificationIndex)

pkg/rid/application/subscription.go

@@ -25,9 +25,6 @@ const (
 type SubscriptionApp interface {
 	GetSubscription(ctx context.Context, id dssmodels.ID) (*ridmodels.Subscription, error)
 
-	// InsertSubscription inserts or updates an Subscription.
-	InsertSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error)
-
 	// UpdateSubscription
 	UpdateSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error)
 
@@ -51,46 +48,6 @@ func (a *app) SearchSubscriptionsByOwner(ctx context.Context, cells s2.CellUnion
 	return repo.SearchSubscriptionsByOwner(ctx, cells, owner)
 }
 
-func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) {
-	// Validate and perhaps correct StartTime and EndTime.
-	if err := s.AdjustTimeRange(a.clock.Now(), nil); err != nil {
-		return nil, stacktrace.Propagate(err, "Unable to adjust time range")
-	}
-	var sub *ridmodels.Subscription
-	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
-
-		// ensure it doesn't exist yet
-		old, err := repo.GetSubscription(ctx, s.ID)
-		if err != nil {
-			return stacktrace.Propagate(err, "Error getting Subscription from repo")
-		}
-		if old != nil {
-			return stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "Subscription %s already exists", s.ID)
-		}
-
-		// Check the user hasn't created too many subscriptions in this area.
-		count, err := repo.MaxSubscriptionCountInCellsByOwner(ctx, s.Cells, s.Owner)
-		if err != nil {
-			a.logger.Error("Error fetching max subscription count", zap.Error(err))
-			return stacktrace.Propagate(err,
-				"Failed to fetch subscription count, rejecting request")
-		}
-		if count >= 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)
-		}
-
-		sub, err = repo.InsertSubscription(ctx, s)
-		if err != nil {
-			return stacktrace.Propagate(err, "Error inserting Subscription into repo")
-		}
-
-		return nil
-	}))
-	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