diff --git a/go.mod b/go.mod index a8fa1da50..d0722521b 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/go-jose/go-jose/v4 v4.1.4 github.com/golang-jwt/jwt/v4 v4.5.2 github.com/golang/geo v0.0.0-20230421003525-6adc56603217 + github.com/google/go-cmp v0.7.0 github.com/google/uuid v1.6.0 github.com/interuss/stacktrace v1.0.0 github.com/jackc/pgx/v5 v5.9.2 @@ -27,11 +28,13 @@ require ( go.opentelemetry.io/otel v1.43.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.39.0 go.opentelemetry.io/otel/exporters/prometheus v0.65.0 + go.opentelemetry.io/otel/metric v1.43.0 go.opentelemetry.io/otel/sdk v1.43.0 go.opentelemetry.io/otel/sdk/metric v1.43.0 go.opentelemetry.io/otel/trace v1.43.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.27.0 + golang.org/x/sync v0.20.0 ) require ( @@ -71,13 +74,11 @@ require ( go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect - go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect golang.org/x/crypto v0.49.0 // indirect golang.org/x/net v0.52.0 // indirect golang.org/x/oauth2 v0.34.0 // indirect - golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect golang.org/x/text v0.36.0 // indirect golang.org/x/time v0.14.0 // indirect diff --git a/pkg/aux_/store/memstore/dss.go b/pkg/aux_/store/memstore/dss.go index 38fa4d5bf..dc86c8e90 100644 --- a/pkg/aux_/store/memstore/dss.go +++ b/pkg/aux_/store/memstore/dss.go @@ -2,24 +2,96 @@ package memstore import ( "context" + "database/sql" auxmodels "github.com/interuss/dss/pkg/aux_/models" dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" ) -func (r *repo) SaveOwnMetadata(_ context.Context, locality string, publicEndpoint string) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SaveOwnMetadata not implemented for memstore") +func (r *repo) SaveOwnMetadata(ctx context.Context, locality string, publicEndpoint string) error { + if locality == "" { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Locality not set") + } + if publicEndpoint == "" { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Public endpoint not set") + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return err + } + + r.state.Participants[locality] = &participant{ + PublicEndpoint: publicEndpoint, + UpdatedAt: now, + } + return nil } func (r *repo) GetDSSMetadata(_ context.Context) ([]*auxmodels.DSSMetadata, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSMetadata not implemented for memstore") + metadata := make([]*auxmodels.DSSMetadata, 0, len(r.state.Participants)) + for locality, p := range r.state.Participants { + updatedAt := p.UpdatedAt + m := &auxmodels.DSSMetadata{ + Locality: locality, + PublicEndpoint: p.PublicEndpoint, + UpdatedAt: &updatedAt, + } + + // Find the latest heartbeat across all sources for this locality. + var latest auxmodels.Heartbeat + found := false + for key, hb := range r.state.Heartbeats { + if key.Locality != locality { + continue + } + if !found || hb.Timestamp.After(*latest.Timestamp) { + latest = hb + found = true + } + } + + if found { + m.LatestTimestamp.Source = sql.NullString{String: latest.Source, Valid: true} + m.LatestTimestamp.Timestamp = latest.Timestamp + m.LatestTimestamp.NextHeartbeatExpectedBefore = latest.NextHeartbeatExpectedBefore + m.LatestTimestamp.Reporter = sql.NullString{String: latest.Reporter, Valid: true} + } + + metadata = append(metadata, m) + } + return metadata, nil } -func (r *repo) RecordHeartbeat(_ context.Context, heartbeat auxmodels.Heartbeat) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "RecordHeartbeat not implemented for memstore") +func (r *repo) RecordHeartbeat(ctx context.Context, heartbeat auxmodels.Heartbeat) error { + if heartbeat.Locality == "" { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Locality not set") + } + if heartbeat.Source == "" { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Source not set") + } + + if heartbeat.Timestamp == nil { + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return err + } + heartbeat.Timestamp = &now + } + + if heartbeat.NextHeartbeatExpectedBefore != nil && heartbeat.NextHeartbeatExpectedBefore.Before(*heartbeat.Timestamp) { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Cannot expect the timestamp of the next heartbeat before the timestamp of the new heartbeat") + } + + r.state.Heartbeats[heartbeatKey{Locality: heartbeat.Locality, Source: heartbeat.Source}] = heartbeat + return nil } func (r *repo) GetDSSAirspaceRepresentationID(_ context.Context) (string, error) { - return "", stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSAirspaceRepresentationID not implemented for memstore") + return "", stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSAirspaceRepresentationID not implementable for memstore") } diff --git a/pkg/aux_/store/memstore/dss_test.go b/pkg/aux_/store/memstore/dss_test.go new file mode 100644 index 000000000..332c2b409 --- /dev/null +++ b/pkg/aux_/store/memstore/dss_test.go @@ -0,0 +1,136 @@ +package memstore + +import ( + "context" + "testing" + "time" + + auxmodels "github.com/interuss/dss/pkg/aux_/models" + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/timestamp" + "github.com/interuss/stacktrace" + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/require" +) + +var fakeClock = clockwork.NewFakeClock() + +func TestSaveOwnMetadataValidation(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.SaveOwnMetadata(ctx, "", "https://example.com"))) + require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.SaveOwnMetadata(ctx, "dss-1", ""))) +} + +func TestSaveOwnMetadataRoundTrip(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + + require.Len(t, md, 1) + require.Equal(t, "dss-1", md[0].Locality) + require.Equal(t, "https://example.com", md[0].PublicEndpoint) + require.NotNil(t, md[0].UpdatedAt) + + // No heartbeat recorded yet. + require.False(t, md[0].LatestTimestamp.Source.Valid) + require.Nil(t, md[0].LatestTimestamp.Timestamp) +} + +func TestSaveOwnMetadataUpsert(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://old.example.com")) + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://new.example.com")) + + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + + require.Len(t, md, 1) + require.Equal(t, "https://new.example.com", md[0].PublicEndpoint) +} + +func TestRecordHeartbeatValidation(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Source: "source1"}))) + require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1"}))) + + ts := time.Now() + before := ts.Add(-time.Minute) + err := r.RecordHeartbeat(ctx, auxmodels.Heartbeat{ + Locality: "dss-1", + Source: "source1", + Timestamp: &ts, + NextHeartbeatExpectedBefore: &before, + }) + + require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(err)) +} + +func TestRecordHeartbeatDefaultsTimestamp(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1"})) + + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + + require.Len(t, md, 1) + require.True(t, md[0].LatestTimestamp.Source.Valid) + require.NotNil(t, md[0].LatestTimestamp.Timestamp) +} + +func TestGetDSSMetadataPicksLatestHeartbeat(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + + older := time.Now().Add(-time.Hour) + newer := time.Now() + require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &older, Reporter: "uss1"})) + require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source2", Timestamp: &newer, Reporter: "uss2"})) + + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + + require.Len(t, md, 1) + require.True(t, md[0].LatestTimestamp.Timestamp.Equal(newer)) + require.Equal(t, "source2", md[0].LatestTimestamp.Source.String) + require.Equal(t, "uss2", md[0].LatestTimestamp.Reporter.String) +} + +func TestGetDSSMetadataUpdatesHeartbeatPerSource(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + + first := time.Now().Add(-time.Hour) + second := time.Now() + require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &first})) + require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &second})) + + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + + require.Len(t, md, 1) + require.True(t, md[0].LatestTimestamp.Timestamp.Equal(second)) +} diff --git a/pkg/aux_/store/memstore/snapshot.go b/pkg/aux_/store/memstore/snapshot.go new file mode 100644 index 000000000..15deb274b --- /dev/null +++ b/pkg/aux_/store/memstore/snapshot.go @@ -0,0 +1,35 @@ +package memstore + +import ( + "bytes" + "encoding/gob" + + "github.com/interuss/stacktrace" +) + +const snapshotVersion = 1 + +type snapshotEnvelope struct { + Version int + State state +} + +func (r *repo) GetSnapshot() ([]byte, error) { + var buf bytes.Buffer + if err := gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion, State: r.state}); err != nil { + return nil, stacktrace.Propagate(err, "Failed to encode memstore snapshot") + } + return buf.Bytes(), nil +} + +func (r *repo) RestoreFromSnapshot(data []byte) error { + var env snapshotEnvelope + if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&env); err != nil { + return stacktrace.Propagate(err, "Failed to decode memstore snapshot") + } + if env.Version != snapshotVersion { + return stacktrace.NewError("Unsupported memstore snapshot version %d, expected %d", env.Version, snapshotVersion) + } + r.state = env.State + return nil +} diff --git a/pkg/aux_/store/memstore/snapshot_test.go b/pkg/aux_/store/memstore/snapshot_test.go new file mode 100644 index 000000000..cd2f03a9f --- /dev/null +++ b/pkg/aux_/store/memstore/snapshot_test.go @@ -0,0 +1,67 @@ +package memstore + +import ( + "bytes" + "context" + "encoding/gob" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + auxmodels "github.com/interuss/dss/pkg/aux_/models" + "github.com/interuss/dss/pkg/models" + "github.com/interuss/dss/pkg/timestamp" + "github.com/stretchr/testify/require" +) + +func TestSnapshotRoundTrip(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + src := newRepo() + require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + ts := time.Now().UTC() + require.NoError(t, src.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source-1", Timestamp: &ts, Reporter: "uss-1"})) + + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := newRepo() + require.NoError(t, dst.RestoreFromSnapshot(data)) + + want, err := src.GetDSSMetadata(ctx) + require.NoError(t, err) + got, err := dst.GetDSSMetadata(ctx) + require.NoError(t, err) + if diff := cmp.Diff(want, got, cmpopts.EquateApproxTime(0), cmp.AllowUnexported(models.Version{})); diff != "" { + t.Errorf("Store mismatch (-want +got):\n%s", diff) + } +} + +func TestRestoreFromSnapshotReplacesState(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + src := newRepo() + require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := newRepo() + require.NoError(t, dst.SaveOwnMetadata(ctx, "dss-2", "https://other.example.com")) + require.NoError(t, dst.RestoreFromSnapshot(data)) + + md, err := dst.GetDSSMetadata(ctx) + require.NoError(t, err) + require.Len(t, md, 1) + require.Equal(t, "dss-1", md[0].Locality) +} + +func TestRestoreFromSnapshotInvalidData(t *testing.T) { + require.Error(t, newRepo().RestoreFromSnapshot([]byte("random value that is definitely not valid"))) +} + +func TestRestoreFromSnapshotVersionMismatch(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion + 1})) + require.Error(t, newRepo().RestoreFromSnapshot(buf.Bytes())) +} diff --git a/pkg/aux_/store/memstore/store.go b/pkg/aux_/store/memstore/store.go index c2b875f9b..bdb8d35d3 100644 --- a/pkg/aux_/store/memstore/store.go +++ b/pkg/aux_/store/memstore/store.go @@ -2,17 +2,82 @@ package memstore import ( "context" + "time" + auxmodels "github.com/interuss/dss/pkg/aux_/models" "github.com/interuss/dss/pkg/aux_/repos" "github.com/interuss/dss/pkg/memstore" + "github.com/interuss/stacktrace" "go.uber.org/zap" ) // repo is a full implementation of aux_.repos.Repository for memory-based storage. -type repo struct{} +type repo struct { + state state +} + +// state is the serializable in-memory state. +type state struct { + // Participants holds pool participants metadata, keyed by locality. + Participants map[string]*participant + // Heartbeats holds the latest heartbeat per (locality, source). + Heartbeats map[heartbeatKey]auxmodels.Heartbeat + // participants holds pool participants metadata, keyed by locality. + participants map[string]*participant + // heartbeats holds the latest heartbeat per (locality, source). + heartbeats map[heartbeatKey]auxmodels.Heartbeat +} + +type participant struct { + PublicEndpoint string + UpdatedAt time.Time +} + +type heartbeatKey struct { + Locality string + Source string +} + +func newRepo() *repo { + return &repo{ + state: state{ + Participants: map[string]*participant{}, + Heartbeats: map[heartbeatKey]auxmodels.Heartbeat{}, + }} +} func Init(ctx context.Context, logger *zap.Logger) (*memstore.Store[repos.Repository], error) { - return memstore.Init(ctx, logger, "aux_", &repo{}) + return memstore.Init(ctx, logger, "aux_", newRepo()) } func (r *repo) GetRepo() repos.Repository { return r } + +// clone returns a copy of s with independent maps and participant records. +func (s state) clone() state { + ps := make(map[string]*participant, len(s.Participants)) + for k, v := range s.Participants { + cp := *v + ps[k] = &cp + } + hb := make(map[heartbeatKey]auxmodels.Heartbeat, len(s.Heartbeats)) + for k, v := range s.Heartbeats { + hb[k] = v + } + return state{Participants: ps, Heartbeats: hb} +} + +// Checkpoint returns a fast, restorable in-memory copy of the current state. +func (r *repo) Checkpoint() any { + return r.state.clone() +} + +// Restore replaces the current state with a checkpoint previously returned by +// Checkpoint. The checkpoint is copied, so it stays reusable. +func (r *repo) Restore(cp any) error { + s, ok := cp.(state) + if !ok { + return stacktrace.NewError("Invalid checkpoint type %T", cp) + } + r.state = s.clone() + return nil +} diff --git a/pkg/aux_/store/memstore/store_test.go b/pkg/aux_/store/memstore/store_test.go new file mode 100644 index 000000000..c0f41b158 --- /dev/null +++ b/pkg/aux_/store/memstore/store_test.go @@ -0,0 +1,55 @@ +package memstore + +import ( + "context" + "testing" + + "github.com/interuss/dss/pkg/timestamp" + "github.com/stretchr/testify/require" +) + +func TestCheckpointRestore(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com")) + + cp := r.Checkpoint() + + // Mutate after the checkpoint. + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-2", "https://other.example.com")) + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + require.Len(t, md, 2) + + // Restore drops dss-2. + require.NoError(t, r.Restore(cp)) + md, err = r.GetDSSMetadata(ctx) + require.NoError(t, err) + require.Len(t, md, 1) + require.Equal(t, "dss-1", md[0].Locality) +} + +func TestCheckpointIsolatesUpsert(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + r := newRepo() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://old.example.com")) + + cp := r.Checkpoint() + + require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://new.example.com")) + + require.NoError(t, r.Restore(cp)) + md, err := r.GetDSSMetadata(ctx) + require.NoError(t, err) + require.Len(t, md, 1) + require.Equal(t, "https://old.example.com", md[0].PublicEndpoint) +} + +func TestRestoreInvalidType(t *testing.T) { + require.Error(t, newRepo().Restore("not a checkpoint")) +} diff --git a/pkg/memstore/store.go b/pkg/memstore/store.go index 9c850f2c3..20ac4bfe5 100644 --- a/pkg/memstore/store.go +++ b/pkg/memstore/store.go @@ -18,6 +18,10 @@ import ( type MemRepo[R any] interface { GetRepo() R + GetSnapshot() ([]byte, error) + RestoreFromSnapshot([]byte) error + Checkpoint() any + Restore(any) error } type Store[R any] struct { @@ -58,6 +62,16 @@ func (s *Store[R]) Interact(_ context.Context) (R, error) { return s.memRepo.GetRepo(), nil } +// Checkpoint returns a fast, restorable in-memory copy of the current state. +func (s *Store[R]) Checkpoint() any { + return s.memRepo.Checkpoint() +} + +// Restore replaces the current state with a checkpoint returned by Checkpoint. +func (s *Store[R]) Restore(cp any) error { + return s.memRepo.Restore(cp) +} + func (s *Store[R]) Close() error { return nil } diff --git a/pkg/rid/store/memstore/identification_service_area.go b/pkg/rid/store/memstore/identification_service_area.go index 3bd315fb7..4dc666fb1 100644 --- a/pkg/rid/store/memstore/identification_service_area.go +++ b/pkg/rid/store/memstore/identification_service_area.go @@ -8,33 +8,159 @@ 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" ) -func (r *repo) GetISA(_ context.Context, id dssmodels.ID, forUpdate bool) (*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetISA not implemented for memstore") +func isaRecordFromModel(isa *ridmodels.IdentificationServiceArea, updatedAt time.Time) *isaRecord { + return &isaRecord{ + ID: isa.ID, + URL: isa.URL, + Owner: isa.Owner, + Cells: cloneCells(isa.Cells), + StartTime: cloneTime(isa.StartTime), + EndTime: cloneTime(isa.EndTime), + AltitudeHi: cloneFloat32(isa.AltitudeHi), + AltitudeLo: cloneFloat32(isa.AltitudeLo), + Writer: isa.Writer, + UpdatedAt: updatedAt, + } } -func (r *repo) DeleteISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteISA not implemented for memstore") +// toModel rebuilds the ISA model +func (rec *isaRecord) toModel() *ridmodels.IdentificationServiceArea { + return &ridmodels.IdentificationServiceArea{ + ID: rec.ID, + URL: rec.URL, + Owner: rec.Owner, + Cells: cloneCells(rec.Cells), + StartTime: cloneTime(rec.StartTime), + EndTime: cloneTime(rec.EndTime), + Version: dssmodels.VersionFromTime(rec.UpdatedAt), + AltitudeHi: cloneFloat32(rec.AltitudeHi), + AltitudeLo: cloneFloat32(rec.AltitudeLo), + Writer: rec.Writer, + } } -func (r *repo) InsertISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "InsertISA not implemented for memstore") +func (r *repo) GetISA(_ context.Context, id dssmodels.ID, _ bool) (*ridmodels.IdentificationServiceArea, error) { + rec, ok := r.state.ISAs[id] + if !ok { + return nil, nil + } + return rec.toModel(), nil } -func (r *repo) UpdateISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpdateISA not implemented for memstore") +func (r *repo) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { + if err := validateWriteData(isa.Cells, isa.StartTime, isa.EndTime); err != nil { + return nil, err + } + if _, ok := r.state.ISAs[isa.ID]; ok { + return nil, stacktrace.NewError("ISA with id %s already exists", isa.ID) + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := isaRecordFromModel(isa, now) + r.state.ISAs[isa.ID] = rec + return rec.toModel(), nil +} + +func (r *repo) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { + if err := validateWriteData(isa.Cells, isa.StartTime, isa.EndTime); err != nil { + return nil, err + } + prev, ok := r.state.ISAs[isa.ID] + if !ok { + return nil, nil + } + if !dssmodels.VersionFromTime(prev.UpdatedAt).Matches(isa.Version) { + return nil, nil + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := isaRecordFromModel(isa, now) + rec.Owner = prev.Owner + r.state.ISAs[isa.ID] = rec + return rec.toModel(), nil +} + +func (r *repo) DeleteISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) { + rec, ok := r.state.ISAs[isa.ID] + if !ok { + return nil, nil + } + if !dssmodels.VersionFromTime(rec.UpdatedAt).Matches(isa.Version) { + return nil, nil + } + out := rec.toModel() + delete(r.state.ISAs, isa.ID) + return out, nil } func (r *repo) SearchISAs(_ context.Context, cells s2.CellUnion, earliest *time.Time, latest *time.Time) ([]*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchISAs not implemented for memstore") + if len(cells) == 0 { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Missing cell IDs for query") + } + if earliest == nil { + return nil, stacktrace.NewError("Earliest start time is missing") + } + + want := cellSet(cells) + var out []*ridmodels.IdentificationServiceArea + for _, rec := range r.state.ISAs { + // ends_at >= earliest + if rec.EndTime == nil || rec.EndTime.Before(*earliest) { + continue + } + // COALESCE(starts_at <= latest, true) + if latest != nil && rec.StartTime != nil && rec.StartTime.After(*latest) { + continue + } + if !overlaps(rec.Cells, want) { + continue + } + out = append(out, rec.toModel()) + + if len(out) > dssmodels.MaxResultLimit { // This miminc sqlstore behaviour, but it's not very good. + break + } + } + return out, nil } func (r *repo) ListExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredISAs not implemented for memstore") + var out []*ridmodels.IdentificationServiceArea + for _, rec := range r.state.ISAs { + // ends_at <= threshold + if rec.EndTime == nil || rec.EndTime.After(threshold) { + continue + } + if writer == "" { + if rec.Writer != "" { + continue + } + } else if rec.Writer != writer { + continue + } + out = append(out, rec.toModel()) + + if len(out) > dssmodels.MaxResultLimit { // This miminc sqlstore behaviour, but it's not very good. + break + } + } + return out, nil } func (r *repo) CountISAs(_ context.Context) (int64, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountISAs not implemented for memstore") + return int64(len(r.state.ISAs)), nil } diff --git a/pkg/rid/store/memstore/identification_service_area_test.go b/pkg/rid/store/memstore/identification_service_area_test.go new file mode 100644 index 000000000..dc27053a7 --- /dev/null +++ b/pkg/rid/store/memstore/identification_service_area_test.go @@ -0,0 +1,324 @@ +package memstore + +import ( + "context" + "testing" + "time" + + "github.com/golang/geo/s2" + "github.com/google/uuid" + dssmodels "github.com/interuss/dss/pkg/models" + ridmodels "github.com/interuss/dss/pkg/rid/models" + "github.com/interuss/dss/pkg/rid/repos" + "github.com/interuss/dss/pkg/timestamp" + "github.com/stretchr/testify/require" +) + +var ( + // Ensure the struct conforms to the interface + _ repos.ISA = &repo{} + overflow = uint64(17106221850767130624) // face 5 L13 overflows + serviceArea = &ridmodels.IdentificationServiceArea{ + ID: dssmodels.ID(uuid.New().String()), + Owner: dssmodels.Owner(uuid.New().String()), + URL: "https://no/place/like/home/for/flights", + StartTime: &startTime, + EndTime: &endTime, + Writer: writer, + Cells: s2.CellUnion{ + s2.CellID(uint64(overflow)), + s2.CellID(17106221850767130624), + }, + } +) + +func TestStoreSearchISAs(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + cells := s2.CellUnion{ + s2.CellID(17106221850767130624), + s2.CellID(17106221885126868992), + s2.CellID(17106221919486607360), + s2.CellID(uint64(overflow)), + } + repo := setUpStore(t) + + isa := *serviceArea + isa.Cells = cells + saOut, err := repo.InsertISA(ctx, &isa) + require.NoError(t, err) + require.NotNil(t, saOut) + require.Equal(t, isa.ID, saOut.ID) + + for _, r := range []struct { + name string + cells s2.CellUnion + timestampMutator func(time.Time, time.Time) (*time.Time, *time.Time) + expectedLen int + }{ + { + name: "search for empty cell", + cells: s2.CellUnion{s2.CellID(17106221953846345728)}, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + return &start, nil + }, + expectedLen: 0, + }, + { + name: "search for only one cell", + cells: s2.CellUnion{s2.CellID(17106221850767130624)}, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + return &start, nil + }, + expectedLen: 1, + }, + { + name: "search for only one cell with high bit set", + cells: s2.CellUnion{s2.CellID(uint64(overflow))}, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + return &start, nil + }, + expectedLen: 1, + }, + { + name: "search with nil ends_at", + cells: cells, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + return &start, nil + }, + expectedLen: 1, + }, + { + name: "search with exact timestamps", + cells: cells, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + return &start, &end + }, + expectedLen: 1, + }, + { + name: "search with non-matching time span", + cells: cells, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + var ( + offset = time.Duration(100 * time.Second) + earliest = end.Add(offset) + latest = end.Add(offset * 2) + ) + + return &earliest, &latest + }, + expectedLen: 0, + }, + { + name: "search with expanded time span", + cells: cells, + timestampMutator: func(start time.Time, end time.Time) (*time.Time, *time.Time) { + var ( + offset = time.Duration(100 * time.Second) + earliest = start.Add(-offset) + latest = end.Add(offset) + ) + + return &earliest, &latest + }, + expectedLen: 1, + }, + } { + t.Run(r.name, func(t *testing.T) { + earliest, latest := r.timestampMutator(*saOut.StartTime, *saOut.EndTime) + + serviceAreas, err := repo.SearchISAs(ctx, r.cells, earliest, latest) + require.NoError(t, err) + require.Len(t, serviceAreas, r.expectedLen) + }) + } +} + +func TestBadVersion(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + saOut1, err := repo.InsertISA(ctx, serviceArea) + require.NoError(t, err) + require.NotNil(t, saOut1) + + // Rewriting service area should fail + saOut2, err := repo.UpdateISA(ctx, serviceArea) + require.NoError(t, err) + require.Nil(t, saOut2) + + // Rewriting, but with the correct version should work. + newEndTime := saOut1.EndTime.Add(time.Minute) + saOut1.EndTime = &newEndTime + saOut3, err := repo.UpdateISA(ctx, saOut1) + require.NoError(t, err) + require.NotNil(t, saOut3) +} + +func TestStoreExpiredISA(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + saOut, err := repo.InsertISA(ctx, serviceArea) + require.NoError(t, err) + require.NotNil(t, saOut) + + // The ISA's endTime is one hour from now. + now := fakeClock.Now() + now = now.Add(59 * time.Minute) + + // We should still be able to find the ISA by searching and by ID. + serviceAreas, err := repo.SearchISAs(ctx, serviceArea.Cells, &now, nil) + require.NoError(t, err) + require.Len(t, serviceAreas, 1) + + ret, err := repo.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + require.NotNil(t, ret) + + // But now the ISA has expired. + now = now.Add(2 * time.Minute) + + serviceAreas, err = repo.SearchISAs(ctx, serviceArea.Cells, &now, nil) + require.NoError(t, err) + require.Len(t, serviceAreas, 0) + + // A get should work even if it is expired. + ret, err = repo.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + require.NotNil(t, ret) +} + +func TestStoreDeleteISAs(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + // Insert the ISA. + copy := *serviceArea + isa, err := repo.InsertISA(ctx, ©) + require.NoError(t, err) + require.NotNil(t, isa) + + // Delete the ISA. + // Ensure a fresh Get, then delete still updates the sub indexes + isa, err = repo.GetISA(ctx, isa.ID, false) + require.NoError(t, err) + + serviceAreaOut, err := repo.DeleteISA(ctx, isa) + require.NoError(t, err) + require.Equal(t, isa, serviceAreaOut) +} + +func TestStoreISAWithNoGeoData(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + endTime := fakeClock.Now().Add(24 * time.Hour) + sub := &ridmodels.IdentificationServiceArea{ + ID: dssmodels.ID(uuid.New().String()), + Owner: dssmodels.Owner("original owner"), + EndTime: &endTime, + } + _, err := repo.InsertISA(ctx, sub) + require.Error(t, err) +} + +func TestListExpiredISAs(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + // Insert ISA with endtime 1 day from now + isa1 := *serviceArea + startTime := fakeClock.Now() + isa1.StartTime = &startTime + endTime := fakeClock.Now().Add(24 * time.Hour) + isa1.EndTime = &endTime + saOut1, err := repo.InsertISA(ctx, &isa1) + require.NoError(t, err) + require.NotNil(t, saOut1) + + // Insert ISA with endtime to 30 minutes ago + isa2 := *serviceArea + startTime = fakeClock.Now().Add(-1 * time.Hour) + isa2.StartTime = &startTime + endTime = fakeClock.Now().Add(-30 * time.Minute) + isa2.EndTime = &endTime + isa2.ID = dssmodels.ID(uuid.New().String()) + saOut2, err := repo.InsertISA(ctx, &isa2) + require.NoError(t, err) + require.NotNil(t, saOut2) + + serviceAreas, err := repo.ListExpiredISAs(ctx, writer, fakeClock.Now().Add(-30*time.Minute)) + require.NoError(t, err) + require.Len(t, serviceAreas, 1) +} + +func TestListExpiredISAsWithEmptyWriter(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + // Insert ISA with endtime 1 day from now + isa1 := *serviceArea + startTime := fakeClock.Now() + isa1.StartTime = &startTime + endTime := fakeClock.Now().Add(24 * time.Hour) + isa1.EndTime = &endTime + isa1.Writer = "" + saOut1, err := repo.InsertISA(ctx, &isa1) + require.NoError(t, err) + require.NotNil(t, saOut1) + + // Insert ISA with endtime to 30 minutes ago + isa2 := *serviceArea + startTime = fakeClock.Now().Add(-1 * time.Hour) + isa2.StartTime = &startTime + endTime = fakeClock.Now().Add(-30 * time.Minute) + isa2.EndTime = &endTime + isa2.ID = dssmodels.ID(uuid.New().String()) + isa2.Writer = "" + saOut2, err := repo.InsertISA(ctx, &isa2) + require.NoError(t, err) + require.NotNil(t, saOut2) + + serviceAreas, err := repo.ListExpiredISAs(ctx, "", fakeClock.Now().Add(-30*time.Minute)) + require.NoError(t, err) + require.Len(t, serviceAreas, 1) +} + +func TestStoreCountISAs(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + // Insert the ISA. + copy := *serviceArea + isa, err := repo.InsertISA(ctx, ©) + require.NoError(t, err) + require.NotNil(t, isa) + + //Cound should be one + count, err := repo.CountISAs(ctx) + require.NoError(t, err) + require.Equal(t, count, int64(1)) + + // Delete the ISA. + // Ensure a fresh Get, then delete still updates the sub indexes + isa, err = repo.GetISA(ctx, isa.ID, false) + require.NoError(t, err) + + serviceAreaOut, err := repo.DeleteISA(ctx, isa) + require.NoError(t, err) + require.Equal(t, isa, serviceAreaOut) + + //Cound should be zero + count, err = repo.CountISAs(ctx) + require.NoError(t, err) + require.Equal(t, count, int64(0)) +} diff --git a/pkg/rid/store/memstore/snapshot.go b/pkg/rid/store/memstore/snapshot.go new file mode 100644 index 000000000..fe2aa2c70 --- /dev/null +++ b/pkg/rid/store/memstore/snapshot.go @@ -0,0 +1,43 @@ +package memstore + +import ( + "bytes" + "encoding/gob" + + dssmodels "github.com/interuss/dss/pkg/models" + "github.com/interuss/stacktrace" +) + +const snapshotVersion = 1 + +type snapshotEnvelope struct { + Version int + State state +} + +func (r *repo) GetSnapshot() ([]byte, error) { + var buf bytes.Buffer + if err := gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion, State: r.state}); err != nil { + return nil, stacktrace.Propagate(err, "Failed to encode memstore snapshot") + } + return buf.Bytes(), nil +} + +func (r *repo) RestoreFromSnapshot(data []byte) error { + var env snapshotEnvelope + if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&env); err != nil { + return stacktrace.Propagate(err, "Failed to decode memstore snapshot") + } + if env.Version != snapshotVersion { + return stacktrace.NewError("Unsupported memstore snapshot version %d, expected %d", env.Version, snapshotVersion) + } + r.state = env.State + // gob decodes an empty map as nil; re-initialize to keep the repo writable. + if r.state.ISAs == nil { + r.state.ISAs = map[dssmodels.ID]*isaRecord{} + } + if r.state.Subscriptions == nil { + r.state.Subscriptions = map[dssmodels.ID]*subscriptionRecord{} + } + return nil +} diff --git a/pkg/rid/store/memstore/snapshot_test.go b/pkg/rid/store/memstore/snapshot_test.go new file mode 100644 index 000000000..a795d5f5f --- /dev/null +++ b/pkg/rid/store/memstore/snapshot_test.go @@ -0,0 +1,85 @@ +package memstore + +import ( + "bytes" + "context" + "encoding/gob" + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/interuss/dss/pkg/models" + "github.com/interuss/dss/pkg/timestamp" + "github.com/stretchr/testify/require" +) + +func TestSnapshotRoundTrip(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + src := setUpStore(t) + _, err := src.InsertISA(ctx, serviceArea) + require.NoError(t, err) + _, err = src.InsertSubscription(ctx, subscriptionsPool[0].input) + require.NoError(t, err) + + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := setUpStore(t) + require.NoError(t, dst.RestoreFromSnapshot(data)) + + wantISA, err := src.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + gotISA, err := dst.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + + if diff := cmp.Diff(wantISA, gotISA, cmpopts.EquateApproxTime(0), cmp.AllowUnexported(models.Version{})); diff != "" { + t.Errorf("IdentificationServiceArea mismatch (-want +got):\n%s", diff) + } + + wantSub, err := src.GetSubscription(ctx, subscriptionsPool[0].input.ID) + require.NoError(t, err) + gotSub, err := dst.GetSubscription(ctx, subscriptionsPool[0].input.ID) + require.NoError(t, err) + + if diff := cmp.Diff(wantSub, gotSub, cmpopts.EquateApproxTime(0), cmp.AllowUnexported(models.Version{})); diff != "" { + t.Errorf("Subscription mismatch (-want +got):\n%s", diff) + } +} + +func TestRestoreFromSnapshotReplacesState(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + src := setUpStore(t) + _, err := src.InsertISA(ctx, serviceArea) + require.NoError(t, err) + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := setUpStore(t) + other := *serviceArea + other.ID = "00000000-0000-4000-8000-000000000002" + _, err = dst.InsertISA(ctx, &other) + require.NoError(t, err) + require.NoError(t, dst.RestoreFromSnapshot(data)) + + count, err := dst.CountISAs(ctx) + require.NoError(t, err) + require.Equal(t, int64(1), count) + got, err := dst.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + require.NotNil(t, got) + gone, err := dst.GetISA(ctx, other.ID, false) + require.NoError(t, err) + require.Nil(t, gone) +} + +func TestRestoreFromSnapshotInvalidData(t *testing.T) { + require.Error(t, setUpStore(t).RestoreFromSnapshot([]byte("random value that is definitely not valid"))) +} + +func TestRestoreFromSnapshotVersionMismatch(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion + 1})) + require.Error(t, setUpStore(t).RestoreFromSnapshot(buf.Bytes())) +} diff --git a/pkg/rid/store/memstore/store.go b/pkg/rid/store/memstore/store.go index b50c2b169..cefd0bb4c 100644 --- a/pkg/rid/store/memstore/store.go +++ b/pkg/rid/store/memstore/store.go @@ -2,17 +2,169 @@ package memstore import ( "context" + "time" + "github.com/golang/geo/s2" + "github.com/interuss/dss/pkg/geo" "github.com/interuss/dss/pkg/memstore" + dssmodels "github.com/interuss/dss/pkg/models" "github.com/interuss/dss/pkg/rid/repos" + "github.com/interuss/stacktrace" "go.uber.org/zap" ) // repo is a full implementation of rid.repos.Repository for memory-based storage. -type repo struct{} +type repo struct { + state state +} + +// state is the serializable in-memory state. +type state struct { + // ISAs holds the stored ISAs keyed by ID. + ISAs map[dssmodels.ID]*isaRecord + // Subscriptions holds the stored subscriptions keyed by ID. + Subscriptions map[dssmodels.ID]*subscriptionRecord +} + +// isaRecord is the gob-serializable representation of an ISA. It intentionally +// stores only primitive fields: the model's Version is never persisted, it is +// derived from UpdatedAt on read. +type isaRecord struct { + ID dssmodels.ID + URL string + Owner dssmodels.Owner + Cells s2.CellUnion + StartTime *time.Time + EndTime *time.Time + AltitudeHi *float32 + AltitudeLo *float32 + Writer string + UpdatedAt time.Time +} + +// subscriptionRecord is the gob-serializable representation of a Subscription. +type subscriptionRecord struct { + ID dssmodels.ID + URL string + NotificationIndex int + Owner dssmodels.Owner + Cells s2.CellUnion + StartTime *time.Time + EndTime *time.Time + AltitudeHi *float32 + AltitudeLo *float32 + Writer string + UpdatedAt time.Time +} + +func newRepo() *repo { + r := &repo{} + r.resetState() + return r +} + +func (r *repo) resetState() { + r.state = state{ + ISAs: map[dssmodels.ID]*isaRecord{}, + Subscriptions: map[dssmodels.ID]*subscriptionRecord{}, + } +} func Init(ctx context.Context, logger *zap.Logger) (*memstore.Store[repos.Repository], error) { - return memstore.Init(ctx, logger, "rid", &repo{}) + return memstore.Init(ctx, logger, "rid", newRepo()) } func (r *repo) GetRepo() repos.Repository { return r } + +// validateWriteData validate constraints on an ISA +func validateWriteData(cells s2.CellUnion, start, end *time.Time) error { + if len(cells) == 0 { + return stacktrace.NewError("At least one cell must be provided") + } + for _, c := range cells { + if err := geo.ValidateCell(c); err != nil { + return stacktrace.Propagate(err, "Error validating cell") + } + } + if start != nil && end != nil && !start.Before(*end) { + return stacktrace.NewError("Start time must be strictly before end time") + } + return nil +} + +// cellSet builds a lookup set from a cell union. +func cellSet(cells s2.CellUnion) map[s2.CellID]struct{} { + set := make(map[s2.CellID]struct{}, len(cells)) + for _, c := range cells { + set[c] = struct{}{} + } + return set +} + +// overlaps reports whether any cell is present in set (equivalent to the SQL +// "cells && $x" array-overlap operator). +func overlaps(cells s2.CellUnion, set map[s2.CellID]struct{}) bool { + for _, c := range cells { + if _, ok := set[c]; ok { + return true + } + } + return false +} + +func cloneCells(cells s2.CellUnion) s2.CellUnion { + if cells == nil { + return nil + } + return append(s2.CellUnion(nil), cells...) +} + +func cloneTime(t *time.Time) *time.Time { + if t == nil { + return nil + } + v := *t + return &v +} + +func cloneFloat32(f *float32) *float32 { + if f == nil { + return nil + } + v := *f + return &v +} + +// clone returns a copy of s with independent maps and records. Cell slices and +// time pointers are shared, as they are never mutated in place. +func (s state) clone() state { + isas := make(map[dssmodels.ID]*isaRecord, len(s.ISAs)) + for id, rec := range s.ISAs { + cp := *rec + isas[id] = &cp + } + subs := make(map[dssmodels.ID]*subscriptionRecord, len(s.Subscriptions)) + for id, rec := range s.Subscriptions { + cp := *rec + subs[id] = &cp + } + return state{ISAs: isas, Subscriptions: subs} +} + +// Checkpoint returns a fast, restorable in-memory copy of the current state. +// Unlike GetSnapshot it does not serialize, so it is cheap but only valid +// in-process. +func (r *repo) Checkpoint() any { + return r.state.clone() +} + +// Restore replaces the current state with a checkpoint previously returned by +// Checkpoint. The checkpoint is copied, so it stays reusable. +func (r *repo) Restore(cp any) error { + s, ok := cp.(state) + if !ok { + return stacktrace.NewError("Invalid checkpoint type %T", cp) + } + r.state = s.clone() + return nil +} diff --git a/pkg/rid/store/memstore/store_test.go b/pkg/rid/store/memstore/store_test.go new file mode 100644 index 000000000..8a798b5d8 --- /dev/null +++ b/pkg/rid/store/memstore/store_test.go @@ -0,0 +1,101 @@ +package memstore + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + dssmodels "github.com/interuss/dss/pkg/models" + ridmodels "github.com/interuss/dss/pkg/rid/models" + "github.com/interuss/dss/pkg/timestamp" + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/require" +) + +var ( + fakeClock = clockwork.NewFakeClock() + startTime = fakeClock.Now().UTC().Add(-time.Minute) + endTime = fakeClock.Now().UTC().Add(time.Hour) + writer = "writer" +) + +// setUpStore returns a fresh in-memory repo whose clock is the (reset) package +// fakeClock, so tests can advance time deterministically. +func setUpStore(t *testing.T) *repo { + t.Helper() + r := newRepo() + return r +} + +func TestDatabaseEnsuresBeginsBeforeExpires(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + var ( + begins = time.Now().UTC() + expires = begins.Add(-5 * time.Minute) + ) + _, err := repo.InsertSubscription(ctx, &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: "me-myself-and-i", + URL: "https://no/place/like/home", + NotificationIndex: 42, + StartTime: &begins, + EndTime: &expires, + }) + require.Error(t, err) +} + +func TestCheckpointRestoreISA(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + _, err := repo.InsertISA(ctx, serviceArea) + require.NoError(t, err) + + cp := repo.Checkpoint() + + // Mutate after the checkpoint. + isa, err := repo.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + _, err = repo.DeleteISA(ctx, isa) + require.NoError(t, err) + gone, err := repo.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + require.Nil(t, gone) + + // Restore brings it back. + require.NoError(t, repo.Restore(cp)) + back, err := repo.GetISA(ctx, serviceArea.ID, false) + require.NoError(t, err) + require.NotNil(t, back) +} + +func TestCheckpointIsolatesNotificationIndex(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + sub, err := repo.InsertSubscription(ctx, subscriptionsPool[0].input) + require.NoError(t, err) + + cp := repo.Checkpoint() + + // In-place notification-index bump must not leak into the checkpoint. + updated, err := repo.UpdateNotificationIdxsInCells(ctx, sub.Cells) + require.NoError(t, err) + require.Len(t, updated, 1) + require.Equal(t, sub.NotificationIndex+1, updated[0].NotificationIndex) + + require.NoError(t, repo.Restore(cp)) + restored, err := repo.GetSubscription(ctx, sub.ID) + require.NoError(t, err) + require.Equal(t, sub.NotificationIndex, restored.NotificationIndex) +} + +func TestRestoreInvalidType(t *testing.T) { + require.Error(t, setUpStore(t).Restore("not a checkpoint")) +} diff --git a/pkg/rid/store/memstore/subscriptions.go b/pkg/rid/store/memstore/subscriptions.go index ac744ab10..9037b0f38 100644 --- a/pkg/rid/store/memstore/subscriptions.go +++ b/pkg/rid/store/memstore/subscriptions.go @@ -8,45 +8,240 @@ 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" ) +func subRecordFromModel(s *ridmodels.Subscription, updatedAt time.Time) *subscriptionRecord { + return &subscriptionRecord{ + ID: s.ID, + URL: s.URL, + NotificationIndex: s.NotificationIndex, + Owner: s.Owner, + Cells: cloneCells(s.Cells), + StartTime: cloneTime(s.StartTime), + EndTime: cloneTime(s.EndTime), + AltitudeHi: cloneFloat32(s.AltitudeHi), + AltitudeLo: cloneFloat32(s.AltitudeLo), + Writer: s.Writer, + UpdatedAt: updatedAt, + } +} + +func (rec *subscriptionRecord) toModel() *ridmodels.Subscription { + return &ridmodels.Subscription{ + ID: rec.ID, + URL: rec.URL, + NotificationIndex: rec.NotificationIndex, + Owner: rec.Owner, + Cells: cloneCells(rec.Cells), + StartTime: cloneTime(rec.StartTime), + EndTime: cloneTime(rec.EndTime), + Version: dssmodels.VersionFromTime(rec.UpdatedAt), + AltitudeHi: cloneFloat32(rec.AltitudeHi), + AltitudeLo: cloneFloat32(rec.AltitudeLo), + Writer: rec.Writer, + } +} + func (r *repo) GetSubscription(_ context.Context, id dssmodels.ID) (*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetSubscription not implemented for memstore") + rec, ok := r.state.Subscriptions[id] + if !ok { + return nil, nil + } + return rec.toModel(), nil } -func (r *repo) DeleteSubscription(_ context.Context, sub *ridmodels.Subscription) (*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteSubscription not implemented for memstore") +func (r *repo) InsertSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) { + if err := validateWriteData(s.Cells, s.StartTime, s.EndTime); err != nil { + return nil, err + } + if _, ok := r.state.Subscriptions[s.ID]; ok { + return nil, stacktrace.NewError("Subscription with id %s already exists", s.ID) + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := subRecordFromModel(s, now) + r.state.Subscriptions[s.ID] = rec + return rec.toModel(), nil } -func (r *repo) InsertSubscription(_ context.Context, sub *ridmodels.Subscription) (*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "InsertSubscription not implemented for memstore") +func (r *repo) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) { + if err := validateWriteData(s.Cells, s.StartTime, s.EndTime); err != nil { + return nil, err + } + prev, ok := r.state.Subscriptions[s.ID] + if !ok { + return nil, nil + } + if !dssmodels.VersionFromTime(prev.UpdatedAt).Matches(s.Version) { + return nil, nil + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := subRecordFromModel(s, now) + rec.Owner = prev.Owner + r.state.Subscriptions[s.ID] = rec + return rec.toModel(), nil } -func (r *repo) UpdateSubscription(_ context.Context, sub *ridmodels.Subscription) (*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpdateSubscription not implemented for memstore") +func (r *repo) DeleteSubscription(_ context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) { + rec, ok := r.state.Subscriptions[s.ID] + if !ok { + return nil, nil + } + if !dssmodels.VersionFromTime(rec.UpdatedAt).Matches(s.Version) { + return nil, nil + } + out := rec.toModel() + delete(r.state.Subscriptions, s.ID) + return out, nil } -func (r *repo) SearchSubscriptions(_ context.Context, cells s2.CellUnion) ([]*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchSubscriptions not implemented for memstore") +func (r *repo) SearchSubscriptions(ctx context.Context, cells s2.CellUnion) ([]*ridmodels.Subscription, error) { + if len(cells) == 0 { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "no location provided") + } + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + want := cellSet(cells) + var out []*ridmodels.Subscription + for _, rec := range r.state.Subscriptions { + if rec.EndTime == nil || rec.EndTime.Before(now) { + continue + } + if !overlaps(rec.Cells, want) { + continue + } + out = append(out, rec.toModel()) + + if len(out) > dssmodels.MaxResultLimit { // This miminc sqlstore behaviour, but it's not very good. + break + } + } + return out, nil } -func (r *repo) SearchSubscriptionsByOwner(_ context.Context, cells s2.CellUnion, owner dssmodels.Owner) ([]*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchSubscriptionsByOwner not implemented for memstore") +func (r *repo) SearchSubscriptionsByOwner(ctx context.Context, cells s2.CellUnion, owner dssmodels.Owner) ([]*ridmodels.Subscription, error) { + if len(cells) == 0 { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "no location provided") + } + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + want := cellSet(cells) + var out []*ridmodels.Subscription + for _, rec := range r.state.Subscriptions { + if rec.Owner != owner { + continue + } + if rec.EndTime == nil || rec.EndTime.Before(now) { + continue + } + if !overlaps(rec.Cells, want) { + continue + } + out = append(out, rec.toModel()) + + if len(out) > dssmodels.MaxResultLimit { // This miminc sqlstore behaviour, but it's not very good. + break + } + } + return out, nil } -func (r *repo) UpdateNotificationIdxsInCells(_ context.Context, cells s2.CellUnion) ([]*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpdateNotificationIdxsInCells not implemented for memstore") +// UpdateNotificationIdxsInCells increments the notification index for each +// subscription in the given cells. +func (r *repo) UpdateNotificationIdxsInCells(ctx context.Context, cells s2.CellUnion) ([]*ridmodels.Subscription, error) { + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + want := cellSet(cells) + var out []*ridmodels.Subscription + for _, rec := range r.state.Subscriptions { + if rec.EndTime == nil || rec.EndTime.Before(now) { + continue + } + if !overlaps(rec.Cells, want) { + continue + } + rec.NotificationIndex++ + out = append(out, rec.toModel()) + } + return out, nil } -func (r *repo) MaxSubscriptionCountInCellsByOwner(_ context.Context, cells s2.CellUnion, owner dssmodels.Owner) (int, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "MaxSubscriptionCountInCellsByOwner not implemented for memstore") +func (r *repo) MaxSubscriptionCountInCellsByOwner(ctx context.Context, cells s2.CellUnion, owner dssmodels.Owner) (int, error) { + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return 0, err + } + + want := cellSet(cells) + counts := make(map[s2.CellID]int, len(cells)) + for _, rec := range r.state.Subscriptions { + if rec.Owner != owner { + continue + } + if rec.EndTime == nil || rec.EndTime.Before(now) { + continue + } + for _, c := range rec.Cells { + if _, ok := want[c]; ok { + counts[c]++ + } + } + } + best := 0 + for _, n := range counts { + if n > best { + best = n + } + } + return best, nil } func (r *repo) ListExpiredSubscriptions(_ context.Context, writer string, threshold time.Time) ([]*ridmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredSubscriptions not implemented for memstore") + var out []*ridmodels.Subscription + for _, rec := range r.state.Subscriptions { + // ends_at <= threshold + if rec.EndTime == nil || rec.EndTime.After(threshold) { + continue + } + if writer == "" { + if rec.Writer != "" { + continue + } + } else if rec.Writer != writer { + continue + } + out = append(out, rec.toModel()) + + // TODO: This miminc sqlstore inconsistency of not limiting results there, comparted to ISAs. Should it be normalized? + } + return out, nil } func (r *repo) CountSubscriptions(_ context.Context) (int64, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountSubscriptions not implemented for memstore") + return int64(len(r.state.Subscriptions)), nil } diff --git a/pkg/rid/store/memstore/subscriptions_test.go b/pkg/rid/store/memstore/subscriptions_test.go new file mode 100644 index 000000000..47797f10d --- /dev/null +++ b/pkg/rid/store/memstore/subscriptions_test.go @@ -0,0 +1,369 @@ +package memstore + +import ( + "context" + "testing" + "time" + + "github.com/golang/geo/s2" + "github.com/google/uuid" + dssmodels "github.com/interuss/dss/pkg/models" + ridmodels "github.com/interuss/dss/pkg/rid/models" + "github.com/interuss/dss/pkg/rid/repos" + "github.com/interuss/dss/pkg/timestamp" + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/require" +) + +var ( + // Ensure the struct conforms to the interface + _ repos.Subscription = &repo{} + subscriptionsPool = []struct { + name string + input *ridmodels.Subscription + }{ + { + name: "a subscription with startTime and endTime", + input: &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: "myself", + URL: "https://no/place/like/home", + StartTime: &startTime, + EndTime: &endTime, + NotificationIndex: 42, + Writer: writer, + Cells: s2.CellUnion{ + s2.CellID(uint64(overflow)), + 12494535935418957824, + }, + }, + }, + { + name: "a subscription without startTime and with endTime", + input: &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: "myself", + URL: "https://no/place/like/home", + EndTime: &endTime, + NotificationIndex: 42, + Cells: s2.CellUnion{ + 12494535935418957824, + }, + }, + }, + { + name: "a subscription without startTime and with endTime", + input: &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: "me", + URL: "https://no/place/like/home", + StartTime: &startTime, + EndTime: &endTime, + NotificationIndex: 42, + Cells: s2.CellUnion{ + 12494535935418957824, + }, + }, + }, + } +) + +func TestStoreGetSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + for _, r := range subscriptionsPool { + t.Run(r.name, func(t *testing.T) { + sub1, err := repo.InsertSubscription(ctx, r.input) + require.NoError(t, err) + require.NotNil(t, sub1) + + sub2, err := repo.GetSubscription(ctx, sub1.ID) + require.NoError(t, err) + require.NotNil(t, sub2) + + require.Equal(t, *sub1, *sub2) + }) + } +} + +func TestStoreInsertSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + for _, r := range subscriptionsPool { + t.Run(r.name, func(t *testing.T) { + sub1, err := repo.InsertSubscription(ctx, r.input) + require.NoError(t, err) + require.NotNil(t, sub1) + + // Test changes without the version differing. + r2 := *sub1 + r2.URL = "new url" + sub2, err := repo.UpdateSubscription(ctx, &r2) + require.NoError(t, err) + require.NotNil(t, sub2) + require.Equal(t, "new url", sub2.URL) + + // Test it doesn't work when Version is nil. + r3 := *sub2 + r3.URL = "new url 2" + r3.Version = nil + sub3, err := repo.UpdateSubscription(ctx, &r3) + require.NoError(t, err) + require.Nil(t, sub3) + + // Bad version doesn't work. + r4 := *sub2 + r4.URL = "new url 3" + r4.Version = dssmodels.VersionFromTime(time.Now()) + sub4, err := repo.UpdateSubscription(ctx, &r4) + require.NoError(t, err) + require.Nil(t, sub4) + + sub5, err := repo.GetSubscription(ctx, sub1.ID) + require.NoError(t, err) + require.NotNil(t, sub5) + + require.Equal(t, *sub2, *sub5) + }) + } +} + +func TestStoreDeleteSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + for _, r := range subscriptionsPool { + t.Run(r.name, func(t *testing.T) { + sub1, err := repo.InsertSubscription(ctx, r.input) + require.NoError(t, err) + require.NotNil(t, sub1) + + // Ensure mismatched versions returns nothing + sub1BadVersion := *sub1 + sub1BadVersion.Version, err = dssmodels.VersionFromString("a3cg3tcuhk00") + require.NoError(t, err) + sub2, err := repo.DeleteSubscription(ctx, &sub1BadVersion) + require.NoError(t, err) + require.Nil(t, sub2) + + sub4, err := repo.DeleteSubscription(ctx, sub1) + require.NoError(t, err) + require.NotNil(t, sub4) + + require.Equal(t, *sub1, *sub4) + }) + } +} + +func TestStoreSearchSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now().UTC()) + repo := setUpStore(t) + + var ( + // pick an L13 value that overflows. + overflow = uint64(17106221850767130624) + + cells = s2.CellUnion{ + s2.CellID(12494535935418957824), + s2.CellID(12494535866699481088), + s2.CellID(12494535901059219456), + s2.CellID(12494535866699481088), + s2.CellID(overflow), + } + owners = []dssmodels.Owner{ + "me", + "my", + "self", + } + ) + + for i, r := range subscriptionsPool { + subscription := *r.input + subscription.Owner = owners[i] + subscription.Cells = cells[:i+1] + sub1, err := repo.InsertSubscription(ctx, &subscription) + require.NoError(t, err) + require.NotNil(t, sub1) + } + // Test normal search + found, err := repo.SearchSubscriptions(ctx, cells) + require.NoError(t, err) + require.Len(t, found, 3) + for _, owner := range owners { + found, err := repo.SearchSubscriptionsByOwner(ctx, cells, owner) + require.NoError(t, err) + require.NotNil(t, found) + // We insert one subscription per owner. Hence, no matter how many cells are touched by the subscription, + // the result should always be 1. + require.Len(t, found, 1) + } +} + +func TestStoreExpiredSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + endTime := fakeClock.Now().Add(24 * time.Hour) + sub := &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: dssmodels.Owner("original owner"), + Cells: s2.CellUnion{s2.CellID(12494535866699481088)}, + EndTime: &endTime, + } + _, err := repo.InsertSubscription(ctx, sub) + require.NoError(t, err) + + // The subscription's endTime is 24 hours from now. + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now().Add(23*time.Hour)) + + // We should still be able to find the subscription by searching and by ID. + subs, err := repo.SearchSubscriptionsByOwner(ctx, sub.Cells, "original owner") + require.NoError(t, err) + require.Len(t, subs, 1) + + ret, err := repo.GetSubscription(ctx, sub.ID) + require.NoError(t, err) + require.NotNil(t, &ret) + + // But now the subscription has expired. + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now().Add(25*time.Hour)) + + subs, err = repo.SearchSubscriptionsByOwner(ctx, sub.Cells, "original owner") + require.NoError(t, err) + require.Len(t, subs, 0) + + ret, err = repo.GetSubscription(ctx, sub.ID) + require.NotNil(t, ret) + require.NoError(t, err) +} + +func TestStoreSubscriptionWithNoGeoData(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + endTime := fakeClock.Now().Add(24 * time.Hour) + sub := &ridmodels.Subscription{ + ID: dssmodels.ID(uuid.New().String()), + Owner: dssmodels.Owner("original owner"), + EndTime: &endTime, + } + _, err := repo.InsertSubscription(ctx, sub) + require.Error(t, err) +} + +func TestMaxSubscriptionCountInCellsByOwner(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + for _, s := range subscriptionsPool { + _, err := repo.InsertSubscription(ctx, s.input) + require.NoError(t, err) + } + + count, err := repo.MaxSubscriptionCountInCellsByOwner(ctx, s2.CellUnion{12494535935418957824}, "myself") + require.NoError(t, err) + require.Equal(t, 2, count) +} + +func TestListExpiredSubscriptions(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + fakeClock := clockwork.NewFakeClockAt(time.Now()) + + // Insert Subscription with endtime 1 day from now + subscripiton1 := *subscriptionsPool[0].input + startTime := fakeClock.Now() + subscripiton1.StartTime = &startTime + endTime := fakeClock.Now().Add(24 * time.Hour) + subscripiton1.EndTime = &endTime + subOut1, err := repo.InsertSubscription(ctx, &subscripiton1) + require.NoError(t, err) + require.NotNil(t, subOut1) + + // Insert Subscription with endtime to 30 minutes ago + subscripiton2 := *subscriptionsPool[0].input + startTime = fakeClock.Now().Add(-1 * time.Hour) + subscripiton2.StartTime = &startTime + endTime = fakeClock.Now().Add(-30 * time.Minute) + subscripiton2.EndTime = &endTime + subscripiton2.ID = dssmodels.ID(uuid.New().String()) + subOut2, err := repo.InsertSubscription(ctx, &subscripiton2) + require.NoError(t, err) + require.NotNil(t, subOut2) + + subscriptions, err := repo.ListExpiredSubscriptions(ctx, writer, fakeClock.Now().Add(-30*time.Minute)) + require.NoError(t, err) + require.Len(t, subscriptions, 1) +} + +func TestListExpiredSubscriptionsWithEmptyWriter(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + // Insert Subscription with endtime 1 day from now + subscripiton1 := *subscriptionsPool[0].input + startTime := fakeClock.Now() + subscripiton1.StartTime = &startTime + endTime := fakeClock.Now().Add(24 * time.Hour) + subscripiton1.EndTime = &endTime + subscripiton1.Writer = "" + subOut1, err := repo.InsertSubscription(ctx, &subscripiton1) + require.NoError(t, err) + require.NotNil(t, subOut1) + + // Insert Subscription with endtime to 30 minutes ago + subscripiton2 := *subscriptionsPool[0].input + startTime = fakeClock.Now().Add(-1 * time.Hour) + subscripiton2.StartTime = &startTime + endTime = fakeClock.Now().Add(-30 * time.Minute) + subscripiton2.EndTime = &endTime + subscripiton2.ID = dssmodels.ID(uuid.New().String()) + subscripiton2.Writer = "" + subOut2, err := repo.InsertSubscription(ctx, &subscripiton2) + require.NoError(t, err) + require.NotNil(t, subOut2) + + subscriptions, err := repo.ListExpiredSubscriptions(ctx, "", fakeClock.Now().Add(-30*time.Minute)) + require.NoError(t, err) + require.Len(t, subscriptions, 1) +} + +func TestStoreCountSubscription(t *testing.T) { + ctx := context.Background() + ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now()) + repo := setUpStore(t) + + for _, r := range subscriptionsPool { + t.Run(r.name, func(t *testing.T) { + sub1, err := repo.InsertSubscription(ctx, r.input) + require.NoError(t, err) + require.NotNil(t, sub1) + + //Cound should be one + count, err := repo.CountSubscriptions(ctx) + require.NoError(t, err) + require.Equal(t, count, int64(1)) + + sub4, err := repo.DeleteSubscription(ctx, sub1) + require.NoError(t, err) + require.NotNil(t, sub4) + + //Cound should be zero + count, err = repo.CountSubscriptions(ctx) + require.NoError(t, err) + require.Equal(t, count, int64(0)) + }) + } +} diff --git a/pkg/scd/store/memstore/availability.go b/pkg/scd/store/memstore/availability.go index 3579672e1..e0d0ba5b5 100644 --- a/pkg/scd/store/memstore/availability.go +++ b/pkg/scd/store/memstore/availability.go @@ -3,16 +3,43 @@ package memstore import ( "context" - dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" scdmodels "github.com/interuss/dss/pkg/scd/models" - "github.com/interuss/stacktrace" + "github.com/interuss/dss/pkg/timestamp" + "github.com/jackc/pgx/v5" ) +func (rec *availabilityRecord) toModel() *scdmodels.UssAvailabilityStatus { + return &scdmodels.UssAvailabilityStatus{ + Uss: rec.Uss, + Availability: rec.Availability, + Version: scdmodels.NewOVNFromTime(rec.UpdatedAt, rec.Uss.String()), + } +} + +// GetUssAvailability implements scd.repos.UssAvailability.GetUssAvailability. func (r *repo) GetUssAvailability(_ context.Context, id dssmodels.Manager) (*scdmodels.UssAvailabilityStatus, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetUssAvailability not implemented for memstore") + rec, ok := r.state.Availabilities[id] + if !ok { + return nil, pgx.ErrNoRows + } + return rec.toModel(), nil } -func (r *repo) UpsertUssAvailability(_ context.Context, ussa *scdmodels.UssAvailabilityStatus) (*scdmodels.UssAvailabilityStatus, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpsertUssAvailability not implemented for memstore") +// UpsertUssAvailability implements scd.repos.UssAvailability.UpsertUssAvailability. +func (r *repo) UpsertUssAvailability(ctx context.Context, s *scdmodels.UssAvailabilityStatus) (*scdmodels.UssAvailabilityStatus, error) { + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := &availabilityRecord{ + Uss: s.Uss, + Availability: s.Availability, + UpdatedAt: now, + } + r.state.Availabilities[s.Uss] = rec + return rec.toModel(), nil } diff --git a/pkg/scd/store/memstore/availability_test.go b/pkg/scd/store/memstore/availability_test.go new file mode 100644 index 000000000..aa4b33e5e --- /dev/null +++ b/pkg/scd/store/memstore/availability_test.go @@ -0,0 +1,32 @@ +package memstore + +import ( + "errors" + "testing" + + scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestUssAvailabilityUpsertGet(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + got, err := r.UpsertUssAvailability(ctx, sampleAvailability()) + require.NoError(t, err) + require.Equal(t, manager, got.Uss) + require.Equal(t, scdmodels.UssAvailabilityStateNormal, got.Availability) + require.NotEmpty(t, got.Version) + + fetched, err := r.GetUssAvailability(ctx, manager) + require.NoError(t, err) + require.Equal(t, got.Version, fetched.Version) + require.Equal(t, scdmodels.UssAvailabilityStateNormal, fetched.Availability) +} + +func TestGetUssAvailabilityMissingReturnsErrNoRows(t *testing.T) { + r := setUpStore(t) + _, err := r.GetUssAvailability(writeCtx(), manager) + require.True(t, errors.Is(err, pgx.ErrNoRows)) +} diff --git a/pkg/scd/store/memstore/constraints.go b/pkg/scd/store/memstore/constraints.go index be3e46eee..415cf60c3 100644 --- a/pkg/scd/store/memstore/constraints.go +++ b/pkg/scd/store/memstore/constraints.go @@ -3,28 +3,107 @@ package memstore import ( "context" - dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" scdmodels "github.com/interuss/dss/pkg/scd/models" + dsssql "github.com/interuss/dss/pkg/sql" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" + "github.com/jackc/pgx/v5" ) +func (rec *constraintRecord) toModel() *scdmodels.Constraint { + return &scdmodels.Constraint{ + ID: rec.ID, + Manager: rec.Manager, + Version: rec.Version, + OVN: scdmodels.NewOVNFromTime(rec.UpdatedAt, rec.ID.String()), + StartTime: cloneTime(rec.StartTime), + EndTime: cloneTime(rec.EndTime), + USSBaseURL: rec.USSBaseURL, + AltitudeLower: cloneFloat32(rec.AltitudeLower), + AltitudeUpper: cloneFloat32(rec.AltitudeUpper), + Cells: cloneCells(rec.Cells), + } +} + +// SearchConstraints implements scd.repos.Constraint.SearchConstraints. func (r *repo) SearchConstraints(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Constraint, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchConstraints not implemented for memstore") + cells, err := v4d.CalculateSpatialCovering() + if err != nil { + return nil, stacktrace.Propagate(err, "Could not calculate spatial covering") + } + if len(cells) == 0 { + return []*scdmodels.Constraint{}, nil + } + + want := cellSet(cells) + var out []*scdmodels.Constraint + for _, rec := range r.state.Constraints { + if !overlaps(rec.Cells, want) { + continue + } + // COALESCE(starts_at <= $3, true) with $3 = v4d.EndTime + if rec.StartTime != nil && v4d.EndTime != nil && rec.StartTime.After(*v4d.EndTime) { + continue + } + // COALESCE(ends_at >= $2, true) with $2 = v4d.StartTime + if rec.EndTime != nil && v4d.StartTime != nil && rec.EndTime.Before(*v4d.StartTime) { + continue + } + out = append(out, rec.toModel()) + if len(out) >= dssmodels.MaxResultLimit { // mirror SQL "LIMIT MaxResultLimit" + break + } + } + return out, nil } +// GetConstraint implements scd.repos.Constraint.GetConstraint. func (r *repo) GetConstraint(_ context.Context, id dssmodels.ID) (*scdmodels.Constraint, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetConstraint not implemented for memstore") + rec, ok := r.state.Constraints[id] + if !ok { + return nil, pgx.ErrNoRows + } + return rec.toModel(), nil } -func (r *repo) UpsertConstraint(_ context.Context, constraint *scdmodels.Constraint) (*scdmodels.Constraint, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpsertConstraint not implemented for memstore") +// UpsertConstraint implements scd.repos.Constraint.UpsertConstraint. +func (r *repo) UpsertConstraint(ctx context.Context, s *scdmodels.Constraint) (*scdmodels.Constraint, error) { + if _, err := dsssql.CellUnionToCellIdsWithValidation(s.Cells); err != nil { + return nil, stacktrace.Propagate(err, "Failed to convert array to jackc/pgtype") + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := &constraintRecord{ + ID: s.ID, + Manager: s.Manager, + Version: s.Version, + StartTime: cloneTime(s.StartTime), + EndTime: cloneTime(s.EndTime), + USSBaseURL: s.USSBaseURL, + AltitudeLower: cloneFloat32(s.AltitudeLower), + AltitudeUpper: cloneFloat32(s.AltitudeUpper), + Cells: cloneCells(s.Cells), + UpdatedAt: now, + } + r.state.Constraints[s.ID] = rec + return rec.toModel(), nil } +// DeleteConstraint implements scd.repos.Constraint.DeleteConstraint. func (r *repo) DeleteConstraint(_ context.Context, id dssmodels.ID) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteConstraint not implemented for memstore") + if _, ok := r.state.Constraints[id]; !ok { + return pgx.ErrNoRows + } + delete(r.state.Constraints, id) + return nil } func (r *repo) CountConstraints(_ context.Context) (int64, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountConstraint not implemented for memstore") + return int64(len(r.state.Constraints)), nil } diff --git a/pkg/scd/store/memstore/constraints_test.go b/pkg/scd/store/memstore/constraints_test.go new file mode 100644 index 000000000..3515820c5 --- /dev/null +++ b/pkg/scd/store/memstore/constraints_test.go @@ -0,0 +1,73 @@ +package memstore + +import ( + "errors" + "testing" + "time" + + "github.com/golang/geo/s2" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestConstraintUpsertGetDelete(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + got, err := r.UpsertConstraint(ctx, sampleConstraint()) + require.NoError(t, err) + require.Equal(t, constraintId, got.ID) + require.Equal(t, manager, got.Manager) + require.NotEmpty(t, got.OVN) + + fetched, err := r.GetConstraint(ctx, constraintId) + require.NoError(t, err) + require.Equal(t, got.OVN, fetched.OVN) + require.Equal(t, cells, fetched.Cells) + + count, err := r.CountConstraints(ctx) + require.NoError(t, err) + require.Equal(t, int64(1), count) + + require.NoError(t, r.DeleteConstraint(ctx, constraintId)) + + _, err = r.GetConstraint(ctx, constraintId) + require.True(t, errors.Is(err, pgx.ErrNoRows)) +} + +func TestConstraintGetMissingReturnsErrNoRows(t *testing.T) { + r := setUpStore(t) + _, err := r.GetConstraint(writeCtx(), constraintId) + require.True(t, errors.Is(err, pgx.ErrNoRows)) +} + +func TestConstraintDeleteMissingReturnsErrNoRows(t *testing.T) { + r := setUpStore(t) + err := r.DeleteConstraint(writeCtx(), constraintId) + require.True(t, errors.Is(err, pgx.ErrNoRows)) +} + +func TestSearchConstraints(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + _, err := r.UpsertConstraint(ctx, sampleConstraint()) + require.NoError(t, err) + + // Overlapping volume with no time bounds matches. + res, err := r.SearchConstraints(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, res, 1) + + // Time window after the constraint's end excludes it. + afterStart := endTime.Add(time.Hour) + afterEnd := afterStart.Add(time.Hour) + res, err = r.SearchConstraints(ctx, volume4D(cells, &afterStart, &afterEnd, nil, nil)) + require.NoError(t, err) + require.Empty(t, res) + + // No covering cells returns an empty (non-nil) slice. + res, err = r.SearchConstraints(ctx, volume4D(s2.CellUnion{}, nil, nil, nil, nil)) + require.NoError(t, err) + require.NotNil(t, res) + require.Empty(t, res) +} diff --git a/pkg/scd/store/memstore/operational_intents.go b/pkg/scd/store/memstore/operational_intents.go index b39f81793..d68b972ec 100644 --- a/pkg/scd/store/memstore/operational_intents.go +++ b/pkg/scd/store/memstore/operational_intents.go @@ -2,38 +2,210 @@ package memstore import ( "context" + "errors" "time" dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" + "github.com/jackc/pgx/v5" ) -func (r *repo) GetOperationalIntent(_ context.Context, id dssmodels.ID) (*scdmodels.OperationalIntent, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetOperationalIntent not implemented for memstore") +// toModel rebuilds the OperationalIntent model without its UssAvailability, +// which is attached separately (see buildOperationalIntents). +func (rec *operationalIntentRecord) toModel() *scdmodels.OperationalIntent { + // If the managing USS has requested a specific OVN it is persisted, otherwise + // a default DSS-generated OVN based on the last update time is used. + var ovn scdmodels.OVN + if rec.USSRequestedOVN != "" { + ovn = scdmodels.OVN(rec.USSRequestedOVN) + } else { + ovn = scdmodels.NewOVNFromTime(rec.UpdatedAt, rec.ID.String()) + } + return &scdmodels.OperationalIntent{ + ID: rec.ID, + Manager: rec.Manager, + Version: rec.Version, + State: rec.State, + OVN: ovn, + PastOVNs: clonePastOVNs(rec.PastOVNs), + StartTime: cloneTime(rec.StartTime), + EndTime: cloneTime(rec.EndTime), + USSBaseURL: rec.USSBaseURL, + SubscriptionID: cloneID(rec.SubscriptionID), + AltitudeLower: cloneFloat32(rec.AltitudeLower), + AltitudeUpper: cloneFloat32(rec.AltitudeUpper), + Cells: cloneCells(rec.Cells), + } } +// buildOperationalIntents converts records to models and attaches the +// UssAvailability of each managing USS +func (r *repo) buildOperationalIntents(ctx context.Context, recs []*operationalIntentRecord) ([]*scdmodels.OperationalIntent, error) { + ussAvailabilities := map[dssmodels.Manager]scdmodels.UssAvailabilityState{} + payload := make([]*scdmodels.OperationalIntent, 0, len(recs)) + for _, rec := range recs { + o := rec.toModel() + ussAvailabilities[o.Manager] = scdmodels.UssAvailabilityStateUnknown + payload = append(payload, o) + } + + for manager := range ussAvailabilities { + ussAvailability, err := r.GetUssAvailability(ctx, manager) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return nil, stacktrace.Propagate(err, "Error getting USS availability of %s", manager) + } + if ussAvailability != nil { + ussAvailabilities[manager] = ussAvailability.Availability + } + } + + for _, op := range payload { + op.UssAvailability = ussAvailabilities[op.Manager] + } + return payload, nil +} + +// GetOperationalIntent implements scd.repos.OperationalIntent.GetOperationalIntent. +func (r *repo) GetOperationalIntent(ctx context.Context, id dssmodels.ID) (*scdmodels.OperationalIntent, error) { + rec, ok := r.state.OperationalIntents[id] + if !ok { + return nil, nil + } + built, err := r.buildOperationalIntents(ctx, []*operationalIntentRecord{rec}) + if err != nil { + return nil, err + } + return built[0], nil +} + +// DeleteOperationalIntent implements scd.repos.OperationalIntent.DeleteOperationalIntent. func (r *repo) DeleteOperationalIntent(_ context.Context, id dssmodels.ID) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteOperationalIntent not implemented for memstore") + if _, ok := r.state.OperationalIntents[id]; !ok { + return stacktrace.NewError("Could not delete Operation that does not exist") + } + delete(r.state.OperationalIntents, id) + return nil } -func (r *repo) UpsertOperationalIntent(_ context.Context, operation *scdmodels.OperationalIntent) (*scdmodels.OperationalIntent, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpsertOperationalIntent not implemented for memstore") +// UpsertOperationalIntent implements scd.repos.OperationalIntent.UpsertOperationalIntent. +func (r *repo) UpsertOperationalIntent(ctx context.Context, operation *scdmodels.OperationalIntent) (*scdmodels.OperationalIntent, error) { + // An empty OVN means the DSS generates it; it is persisted as NULL in the + // sqlstore (represented here by an empty USSRequestedOVN). + var ussRequestedOVN string + if operation.OVN != "" { + ussRequestedOVN = operation.OVN.String() + } + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := &operationalIntentRecord{ + ID: operation.ID, + Manager: operation.Manager, + Version: operation.Version, + State: operation.State, + StartTime: cloneTime(operation.StartTime), + EndTime: cloneTime(operation.EndTime), + USSBaseURL: operation.USSBaseURL, + SubscriptionID: cloneID(operation.SubscriptionID), + AltitudeLower: cloneFloat32(operation.AltitudeLower), + AltitudeUpper: cloneFloat32(operation.AltitudeUpper), + Cells: cloneCells(operation.Cells), + USSRequestedOVN: ussRequestedOVN, + PastOVNs: clonePastOVNs(operation.PastOVNs), + UpdatedAt: now, + } + r.state.OperationalIntents[operation.ID] = rec + + built, err := r.buildOperationalIntents(ctx, []*operationalIntentRecord{rec}) + if err != nil { + return nil, err + } + return built[0], nil } -func (r *repo) SearchOperationalIntents(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.OperationalIntent, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchOperationalIntents not implemented for memstore") +// SearchOperationalIntents implements scd.repos.OperationalIntent.SearchOperationalIntents. +func (r *repo) SearchOperationalIntents(ctx context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.OperationalIntent, error) { + if v4d.SpatialVolume == nil || v4d.SpatialVolume.Footprint == nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Missing geospatial footprint for query") + } + cells, err := v4d.SpatialVolume.Footprint.CalculateCovering() + if err != nil { + return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to calculate footprint covering") + } + if len(cells) == 0 { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Missing cell IDs for query") + } + + want := cellSet(cells) + var matched []*operationalIntentRecord + for _, rec := range r.state.OperationalIntents { + if !overlaps(rec.Cells, want) { + continue + } + // COALESCE(altitude_upper >= $2, true) with $2 = SpatialVolume.AltitudeLo + if rec.AltitudeUpper != nil && v4d.SpatialVolume.AltitudeLo != nil && *rec.AltitudeUpper < *v4d.SpatialVolume.AltitudeLo { + continue + } + // COALESCE(altitude_lower <= $3, true) with $3 = SpatialVolume.AltitudeHi + if rec.AltitudeLower != nil && v4d.SpatialVolume.AltitudeHi != nil && *rec.AltitudeLower > *v4d.SpatialVolume.AltitudeHi { + continue + } + // COALESCE(ends_at >= $4, true) with $4 = v4d.StartTime + if rec.EndTime != nil && v4d.StartTime != nil && rec.EndTime.Before(*v4d.StartTime) { + continue + } + // COALESCE(starts_at <= $5, true) with $5 = v4d.EndTime + if rec.StartTime != nil && v4d.EndTime != nil && rec.StartTime.After(*v4d.EndTime) { + continue + } + matched = append(matched, rec) + if len(matched) >= dssmodels.MaxResultLimit { // mirror SQL "LIMIT MaxResultLimit" + break + } + } + return r.buildOperationalIntents(ctx, matched) } +// GetDependentOperationalIntents implements scd.repos.OperationalIntent.GetDependentOperationalIntents. func (r *repo) GetDependentOperationalIntents(_ context.Context, subscriptionID dssmodels.ID) ([]dssmodels.ID, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDependentOperationalIntents not implemented for memstore") + var dependentOps []dssmodels.ID + for _, rec := range r.state.OperationalIntents { + if rec.SubscriptionID != nil && *rec.SubscriptionID == subscriptionID { + dependentOps = append(dependentOps, rec.ID) + } + } + return dependentOps, nil } -func (r *repo) ListExpiredOperationalIntents(_ context.Context, threshold time.Time) ([]*scdmodels.OperationalIntent, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredOperationalIntents not implemented for memstore") +// ListExpiredOperationalIntents implements scd.repos.OperationalIntent.ListExpiredOperationalIntents. +func (r *repo) ListExpiredOperationalIntents(ctx context.Context, threshold time.Time) ([]*scdmodels.OperationalIntent, error) { + var matched []*operationalIntentRecord + for _, rec := range r.state.OperationalIntents { + // (ends_at IS NOT NULL AND ends_at <= threshold) OR (ends_at IS NULL AND updated_at <= threshold) + var expired bool + if rec.EndTime != nil { + expired = !rec.EndTime.After(threshold) + } else { + expired = !rec.UpdatedAt.After(threshold) + } + if !expired { + continue + } + matched = append(matched, rec) + if len(matched) >= dssmodels.MaxResultLimit { // mirror SQL "LIMIT MaxResultLimit" + break + } + } + return r.buildOperationalIntents(ctx, matched) } func (r *repo) CountOperationalIntents(_ context.Context) (int64, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountOperationalIntents not implemented for memstore") + return int64(len(r.state.OperationalIntents)), nil } diff --git a/pkg/scd/store/memstore/operational_intents_test.go b/pkg/scd/store/memstore/operational_intents_test.go new file mode 100644 index 000000000..4eaabb575 --- /dev/null +++ b/pkg/scd/store/memstore/operational_intents_test.go @@ -0,0 +1,208 @@ +package memstore + +import ( + "testing" + "time" + + dssmodels "github.com/interuss/dss/pkg/models" + scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/stretchr/testify/require" +) + +func TestOperationalIntentUpsertGetDelete(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + got, err := r.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + require.Equal(t, operationalIntentId, got.ID) + require.Equal(t, scdmodels.OperationalIntentStateAccepted, got.State) + require.NotEmpty(t, got.OVN) + // No availability stored yet: defaults to Unknown. + require.Equal(t, scdmodels.UssAvailabilityStateUnknown, got.UssAvailability) + + count, err := r.CountOperationalIntents(ctx) + require.NoError(t, err) + require.Equal(t, int64(1), count) + + require.NoError(t, r.DeleteOperationalIntent(ctx, operationalIntentId)) + gone, err := r.GetOperationalIntent(ctx, operationalIntentId) + require.NoError(t, err) + require.Nil(t, gone) +} + +func TestOperationalIntentGetMissingReturnsNil(t *testing.T) { + r := setUpStore(t) + got, err := r.GetOperationalIntent(writeCtx(), operationalIntentId) + require.NoError(t, err) + require.Nil(t, got) +} + +func TestOperationalIntentDeleteMissingErrors(t *testing.T) { + r := setUpStore(t) + require.Error(t, r.DeleteOperationalIntent(writeCtx(), operationalIntentId)) +} + +func TestOperationalIntentUssAvailabilityAttached(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + _, err := r.UpsertUssAvailability(ctx, sampleAvailability()) + require.NoError(t, err) + _, err = r.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + + got, err := r.GetOperationalIntent(ctx, operationalIntentId) + require.NoError(t, err) + require.Equal(t, scdmodels.UssAvailabilityStateNormal, got.UssAvailability) +} + +func TestSearchOperationalIntents(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + _, err := r.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + + res, err := r.SearchOperationalIntents(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, res, 1) + + // Altitude window entirely above the operational intent excludes it. + var lo float32 = 200 + res, err = r.SearchOperationalIntents(ctx, volume4D(cells, nil, nil, &lo, nil)) + require.NoError(t, err) + require.Empty(t, res) + + // Missing footprint is a bad request. + _, err = r.SearchOperationalIntents(ctx, &dssmodels.Volume4D{}) + require.Error(t, err) +} + +func TestGetDependentOperationalIntents(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + _, err := r.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + + deps, err := r.GetDependentOperationalIntents(ctx, subscriptionId) + require.NoError(t, err) + require.Equal(t, []dssmodels.ID{operationalIntentId}, deps) + + deps, err = r.GetDependentOperationalIntents(ctx, "other") + require.NoError(t, err) + require.Nil(t, deps) +} + +var ( + oi1ID = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca30000") + oi2ID = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca30001") + oi3ID = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca30003") + + start1 = time.Date(2024, time.August, 14, 15, 48, 36, 0, time.UTC) + end1 = start1.Add(time.Hour) + start2 = time.Date(2024, time.September, 15, 15, 48, 36, 0, time.UTC) + end2 = start2.Add(time.Hour) + start3 = time.Date(2024, time.September, 16, 15, 48, 36, 0, time.UTC) + end3 = start3.Add(time.Hour) +) + +var ( + oi1 = &scdmodels.OperationalIntent{ + ID: oi1ID, + Manager: "unittest", + Version: 1, + State: scdmodels.OperationalIntentStateAccepted, + StartTime: &start1, + EndTime: &end1, + USSBaseURL: "https://dummy.uss", + SubscriptionID: &sub1ID, + AltitudeLower: &altLow, + AltitudeUpper: &altHigh, + Cells: cells, + } + oi2 = &scdmodels.OperationalIntent{ + ID: oi2ID, + Manager: "unittest", + Version: 1, + State: scdmodels.OperationalIntentStateAccepted, + StartTime: &start2, + EndTime: &end2, + USSBaseURL: "https://dummy.uss", + SubscriptionID: &sub2ID, + AltitudeLower: &altLow, + AltitudeUpper: &altHigh, + Cells: cells, + } + oi3 = &scdmodels.OperationalIntent{ + ID: oi3ID, + Manager: "unittest", + Version: 1, + State: scdmodels.OperationalIntentStateAccepted, + StartTime: &start3, + EndTime: &end3, + USSBaseURL: "https://dummy.uss", + SubscriptionID: &sub3ID, + AltitudeLower: &altLow, + AltitudeUpper: &altHigh, + Cells: cells, + } +) + +func TestListExpiredOperationalIntents(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + _, err := r.UpsertSubscription(ctx, sub1) + require.NoError(t, err) + _, err = r.UpsertOperationalIntent(ctx, oi1) + require.NoError(t, err) + + _, err = r.UpsertSubscription(ctx, sub2) + require.NoError(t, err) + _, err = r.UpsertOperationalIntent(ctx, oi2) + require.NoError(t, err) + + _, err = r.UpsertSubscription(ctx, sub3) + require.NoError(t, err) + _, err = r.UpsertOperationalIntent(ctx, oi3) + require.NoError(t, err) + + testCases := []struct { + name string + timeRef time.Time + ttl time.Duration + expired []dssmodels.ID + }{{ + name: "none expired, one in close past", + timeRef: time.Date(2024, time.August, 25, 15, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{}, + }, { + name: "one recently expired, one current, one in future", + timeRef: time.Date(2024, time.September, 15, 16, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{oi1ID}, + }, { + name: "two expired, one in future", + timeRef: time.Date(2024, time.September, 16, 16, 0, 0, 0, time.UTC), + ttl: time.Hour * 2, + expired: []dssmodels.ID{oi1ID, oi2ID}, + }, { + name: "all expired", + timeRef: time.Date(2024, time.December, 15, 15, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{oi1ID, oi2ID, oi3ID}, + }} + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + threshold := testCase.timeRef.Add(-testCase.ttl) + expired, err := r.ListExpiredOperationalIntents(ctx, threshold) + require.NoError(t, err) + + expiredIDs := make([]dssmodels.ID, 0, len(expired)) + for _, expiredOi := range expired { + expiredIDs = append(expiredIDs, expiredOi.ID) + } + require.ElementsMatch(t, expiredIDs, testCase.expired) + }) + } +} diff --git a/pkg/scd/store/memstore/snapshot.go b/pkg/scd/store/memstore/snapshot.go new file mode 100644 index 000000000..6688121f8 --- /dev/null +++ b/pkg/scd/store/memstore/snapshot.go @@ -0,0 +1,49 @@ +package memstore + +import ( + "bytes" + "encoding/gob" + + dssmodels "github.com/interuss/dss/pkg/models" + "github.com/interuss/stacktrace" +) + +const snapshotVersion = 1 + +type snapshotEnvelope struct { + Version int + State state +} + +func (r *repo) GetSnapshot() ([]byte, error) { + var buf bytes.Buffer + if err := gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion, State: r.state}); err != nil { + return nil, stacktrace.Propagate(err, "Failed to encode memstore snapshot") + } + return buf.Bytes(), nil +} + +func (r *repo) RestoreFromSnapshot(data []byte) error { + var env snapshotEnvelope + if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&env); err != nil { + return stacktrace.Propagate(err, "Failed to decode memstore snapshot") + } + if env.Version != snapshotVersion { + return stacktrace.NewError("Unsupported memstore snapshot version %d, expected %d", env.Version, snapshotVersion) + } + r.state = env.State + // gob decodes an empty map as nil; re-initialize to keep the repo writable. + if r.state.Constraints == nil { + r.state.Constraints = map[dssmodels.ID]*constraintRecord{} + } + if r.state.Subscriptions == nil { + r.state.Subscriptions = map[dssmodels.ID]*subscriptionRecord{} + } + if r.state.OperationalIntents == nil { + r.state.OperationalIntents = map[dssmodels.ID]*operationalIntentRecord{} + } + if r.state.Availabilities == nil { + r.state.Availabilities = map[dssmodels.Manager]*availabilityRecord{} + } + return nil +} diff --git a/pkg/scd/store/memstore/snapshot_test.go b/pkg/scd/store/memstore/snapshot_test.go new file mode 100644 index 000000000..5f7d24380 --- /dev/null +++ b/pkg/scd/store/memstore/snapshot_test.go @@ -0,0 +1,99 @@ +package memstore + +import ( + "bytes" + "encoding/gob" + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/stretchr/testify/require" +) + +func TestSnapshotRoundTrip(t *testing.T) { + ctx := writeCtx() + src := setUpStore(t) + _, err := src.UpsertConstraint(ctx, sampleConstraint()) + require.NoError(t, err) + _, err = src.UpsertSubscription(ctx, sampleSubscription()) + require.NoError(t, err) + _, err = src.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + _, err = src.UpsertUssAvailability(ctx, sampleAvailability()) + require.NoError(t, err) + + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := setUpStore(t) + require.NoError(t, dst.RestoreFromSnapshot(data)) + + opt := cmpopts.EquateApproxTime(0) + + wantCon, err := src.GetConstraint(ctx, constraintId) + require.NoError(t, err) + gotCon, err := dst.GetConstraint(ctx, constraintId) + require.NoError(t, err) + if diff := cmp.Diff(wantCon, gotCon, opt); diff != "" { + t.Errorf("Constraint mismatch (-want +got):\n%s", diff) + } + + wantSub, err := src.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + gotSub, err := dst.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + if diff := cmp.Diff(wantSub, gotSub, opt); diff != "" { + t.Errorf("Subscription mismatch (-want +got):\n%s", diff) + } + + wantOI, err := src.GetOperationalIntent(ctx, operationalIntentId) + require.NoError(t, err) + gotOI, err := dst.GetOperationalIntent(ctx, operationalIntentId) + require.NoError(t, err) + if diff := cmp.Diff(wantOI, gotOI, opt); diff != "" { + t.Errorf("OperationalIntent mismatch (-want +got):\n%s", diff) + } + + wantAvail, err := src.GetUssAvailability(ctx, manager) + require.NoError(t, err) + gotAvail, err := dst.GetUssAvailability(ctx, manager) + require.NoError(t, err) + if diff := cmp.Diff(wantAvail, gotAvail, opt); diff != "" { + t.Errorf("UssAvailability mismatch (-want +got):\n%s", diff) + } +} + +func TestRestoreFromSnapshotReplacesState(t *testing.T) { + ctx := writeCtx() + src := setUpStore(t) + _, err := src.UpsertConstraint(ctx, sampleConstraint()) + require.NoError(t, err) + data, err := src.GetSnapshot() + require.NoError(t, err) + + dst := setUpStore(t) + other := sampleConstraint() + other.ID = "00000185-e36d-40be-8d38-beca6ca39999" + _, err = dst.UpsertConstraint(ctx, other) + require.NoError(t, err) + require.NoError(t, dst.RestoreFromSnapshot(data)) + + count, err := dst.CountConstraints(ctx) + require.NoError(t, err) + require.Equal(t, int64(1), count) + got, err := dst.GetConstraint(ctx, constraintId) + require.NoError(t, err) + require.NotNil(t, got) + _, err = dst.GetConstraint(ctx, other.ID) + require.Error(t, err) +} + +func TestRestoreFromSnapshotInvalidData(t *testing.T) { + require.Error(t, setUpStore(t).RestoreFromSnapshot([]byte("random value that is definitely not valid"))) +} + +func TestRestoreFromSnapshotVersionMismatch(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion + 1})) + require.Error(t, setUpStore(t).RestoreFromSnapshot(buf.Bytes())) +} diff --git a/pkg/scd/store/memstore/store.go b/pkg/scd/store/memstore/store.go index 45365614a..f2d6e7e0c 100644 --- a/pkg/scd/store/memstore/store.go +++ b/pkg/scd/store/memstore/store.go @@ -2,17 +2,217 @@ package memstore import ( "context" + "time" + "github.com/golang/geo/s2" "github.com/interuss/dss/pkg/memstore" + dssmodels "github.com/interuss/dss/pkg/models" + scdmodels "github.com/interuss/dss/pkg/scd/models" "github.com/interuss/dss/pkg/scd/repos" + "github.com/interuss/stacktrace" "go.uber.org/zap" ) // repo is a full implementation of scd.repos.Repository for memory-based storage. -type repo struct{} +type repo struct { + state state +} + +// state is the serializable in-memory state. +type state struct { + // Constraints holds the stored constraints keyed by ID. + Constraints map[dssmodels.ID]*constraintRecord + // Subscriptions holds the stored subscriptions keyed by ID. + Subscriptions map[dssmodels.ID]*subscriptionRecord + // OperationalIntents holds the stored operational intents keyed by ID. + OperationalIntents map[dssmodels.ID]*operationalIntentRecord + // Availabilities holds the stored USS availabilities keyed by USS Manager. + Availabilities map[dssmodels.Manager]*availabilityRecord +} + +// constraintRecord is the gob-serializable representation of a Constraint. The +// model's OVN is never persisted: it is derived from UpdatedAt on read +type constraintRecord struct { + ID dssmodels.ID + Manager dssmodels.Manager + Version scdmodels.VersionNumber + StartTime *time.Time + EndTime *time.Time + USSBaseURL string + AltitudeLower *float32 + AltitudeUpper *float32 + Cells s2.CellUnion + UpdatedAt time.Time +} + +// subscriptionRecord is the gob-serializable representation of a Subscription. +// The sqlstore stores the version column but always writes 0 and discards it on +// read (the model Version is derived from UpdatedAt), so it is not kept here. +type subscriptionRecord struct { + ID dssmodels.ID + Manager dssmodels.Manager + NotificationIndex int + USSBaseURL string + NotifyForOperationalIntents bool + NotifyForConstraints bool + ImplicitSubscription bool + StartTime *time.Time + EndTime *time.Time + Cells s2.CellUnion + UpdatedAt time.Time +} + +// operationalIntentRecord is the gob-serializable representation of an +// OperationalIntent. USSRequestedOVN is empty when the OVN is DSS-generated. +type operationalIntentRecord struct { + ID dssmodels.ID + Manager dssmodels.Manager + Version scdmodels.VersionNumber + State scdmodels.OperationalIntentState + StartTime *time.Time + EndTime *time.Time + USSBaseURL string + SubscriptionID *dssmodels.ID + AltitudeLower *float32 + AltitudeUpper *float32 + Cells s2.CellUnion + USSRequestedOVN string + PastOVNs []scdmodels.OVN + UpdatedAt time.Time +} + +// availabilityRecord is the gob-serializable representation of a +// UssAvailabilityStatus. The model's Version is derived from UpdatedAt on read. +type availabilityRecord struct { + Uss dssmodels.Manager + Availability scdmodels.UssAvailabilityState + UpdatedAt time.Time +} + +func newRepo() *repo { + r := &repo{} + r.resetState() + return r +} + +func (r *repo) resetState() { + r.state = state{ + Constraints: map[dssmodels.ID]*constraintRecord{}, + Subscriptions: map[dssmodels.ID]*subscriptionRecord{}, + OperationalIntents: map[dssmodels.ID]*operationalIntentRecord{}, + Availabilities: map[dssmodels.Manager]*availabilityRecord{}, + } +} func Init(ctx context.Context, logger *zap.Logger) (*memstore.Store[repos.Repository], error) { - return memstore.Init(ctx, logger, "scd", &repo{}) + return memstore.Init(ctx, logger, "scd", newRepo()) } func (r *repo) GetRepo() repos.Repository { return r } + +// cellSet builds a lookup set from a cell union. +func cellSet(cells s2.CellUnion) map[s2.CellID]struct{} { + set := make(map[s2.CellID]struct{}, len(cells)) + for _, c := range cells { + set[c] = struct{}{} + } + return set +} + +// overlaps reports whether any cell is present in set (equivalent to the SQL +// "cells && $x" array-overlap operator). +func overlaps(cells s2.CellUnion, set map[s2.CellID]struct{}) bool { + for _, c := range cells { + if _, ok := set[c]; ok { + return true + } + } + return false +} + +func cloneCells(cells s2.CellUnion) s2.CellUnion { + if cells == nil { + return nil + } + return append(s2.CellUnion(nil), cells...) +} + +func cloneTime(t *time.Time) *time.Time { + if t == nil { + return nil + } + v := *t + return &v +} + +func cloneFloat32(f *float32) *float32 { + if f == nil { + return nil + } + v := *f + return &v +} + +func cloneID(id *dssmodels.ID) *dssmodels.ID { + if id == nil { + return nil + } + v := *id + return &v +} + +func clonePastOVNs(ovns []scdmodels.OVN) []scdmodels.OVN { + if ovns == nil { + return nil + } + return append([]scdmodels.OVN(nil), ovns...) +} + +// clone returns a copy of s with independent maps and records. Cell slices, +// time pointers and OVN slices are shared, as they are never mutated in place. +func (s state) clone() state { + constraints := make(map[dssmodels.ID]*constraintRecord, len(s.Constraints)) + for id, rec := range s.Constraints { + cp := *rec + constraints[id] = &cp + } + subs := make(map[dssmodels.ID]*subscriptionRecord, len(s.Subscriptions)) + for id, rec := range s.Subscriptions { + cp := *rec + subs[id] = &cp + } + ois := make(map[dssmodels.ID]*operationalIntentRecord, len(s.OperationalIntents)) + for id, rec := range s.OperationalIntents { + cp := *rec + ois[id] = &cp + } + avails := make(map[dssmodels.Manager]*availabilityRecord, len(s.Availabilities)) + for id, rec := range s.Availabilities { + cp := *rec + avails[id] = &cp + } + return state{ + Constraints: constraints, + Subscriptions: subs, + OperationalIntents: ois, + Availabilities: avails, + } +} + +// Checkpoint returns a fast, restorable in-memory copy of the current state. +// Unlike GetSnapshot it does not serialize, so it is cheap but only valid +// in-process. +func (r *repo) Checkpoint() any { + return r.state.clone() +} + +// Restore replaces the current state with a checkpoint previously returned by +// Checkpoint. The checkpoint is copied, so it stays reusable. +func (r *repo) Restore(cp any) error { + s, ok := cp.(state) + if !ok { + return stacktrace.NewError("Invalid checkpoint type %T", cp) + } + r.state = s.clone() + return nil +} diff --git a/pkg/scd/store/memstore/store_test.go b/pkg/scd/store/memstore/store_test.go new file mode 100644 index 000000000..152399077 --- /dev/null +++ b/pkg/scd/store/memstore/store_test.go @@ -0,0 +1,170 @@ +package memstore + +import ( + "context" + "testing" + "time" + + "github.com/golang/geo/s2" + dssmodels "github.com/interuss/dss/pkg/models" + scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/interuss/dss/pkg/timestamp" + "github.com/stretchr/testify/require" +) + +var ( + manager = dssmodels.Manager("unittest") + + constraintId = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca31000") + subscriptionId = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca31001") + operationalIntentId = dssmodels.ID("00000185-e36d-40be-8d38-beca6ca31002") + + cells = s2.CellUnion{ + s2.CellID(int64(8768904281496485888)), + s2.CellID(int64(8768904178417270784)), + } + + startTime = time.Date(2024, time.August, 14, 15, 48, 36, 0, time.UTC) + endTime = startTime.Add(time.Hour) + writeTime = time.Date(2024, time.August, 1, 0, 0, 0, 0, time.UTC) + + altLow, altHigh float32 = 84, 169 +) + +// setUpStore returns a fresh in-memory repo. +func setUpStore(t *testing.T) *repo { + t.Helper() + return newRepo() +} + +// writeCtx returns a context carrying a deterministic write timestamp so that +// updated_at is controlled in tests. +func writeCtx() context.Context { + return timestamp.WithRequestTimestamp(context.Background(), writeTime) +} + +func sampleConstraint() *scdmodels.Constraint { + return &scdmodels.Constraint{ + ID: constraintId, + Manager: manager, + Version: 1, + StartTime: &startTime, + EndTime: &endTime, + USSBaseURL: "https://dummy.uss", + AltitudeLower: &altLow, + AltitudeUpper: &altHigh, + Cells: cells, + } +} + +func sampleSubscription() *scdmodels.Subscription { + return &scdmodels.Subscription{ + ID: subscriptionId, + Manager: manager, + NotificationIndex: 1, + USSBaseURL: "https://dummy.uss", + NotifyForOperationalIntents: true, + NotifyForConstraints: true, + StartTime: &startTime, + EndTime: &endTime, + Cells: cells, + } +} + +func sampleOperationalIntent() *scdmodels.OperationalIntent { + sid := subscriptionId + return &scdmodels.OperationalIntent{ + ID: operationalIntentId, + Manager: manager, + Version: 1, + State: scdmodels.OperationalIntentStateAccepted, + StartTime: &startTime, + EndTime: &endTime, + USSBaseURL: "https://dummy.uss", + SubscriptionID: &sid, + AltitudeLower: &altLow, + AltitudeUpper: &altHigh, + Cells: cells, + } +} + +func sampleAvailability() *scdmodels.UssAvailabilityStatus { + return &scdmodels.UssAvailabilityStatus{ + Uss: manager, + Availability: scdmodels.UssAvailabilityStateNormal, + } +} + +// volume4D builds a Volume4D whose footprint covers the provided cells. +func volume4D(cu s2.CellUnion, start, end *time.Time, altLo, altHi *float32) *dssmodels.Volume4D { + return &dssmodels.Volume4D{ + StartTime: start, + EndTime: end, + SpatialVolume: &dssmodels.Volume3D{ + AltitudeLo: altLo, + AltitudeHi: altHi, + Footprint: dssmodels.GeometryFunc(func() (s2.CellUnion, error) { + return cu, nil + }), + }, + } +} + +func TestCheckpointRestoreRoundTrip(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + _, err := r.UpsertConstraint(ctx, sampleConstraint()) + require.NoError(t, err) + _, err = r.UpsertSubscription(ctx, sampleSubscription()) + require.NoError(t, err) + _, err = r.UpsertOperationalIntent(ctx, sampleOperationalIntent()) + require.NoError(t, err) + _, err = r.UpsertUssAvailability(ctx, sampleAvailability()) + require.NoError(t, err) + + cp := r.Checkpoint() + + // Mutate after the checkpoint. + require.NoError(t, r.DeleteConstraint(ctx, constraintId)) + require.NoError(t, r.DeleteSubscription(ctx, subscriptionId)) + require.NoError(t, r.DeleteOperationalIntent(ctx, operationalIntentId)) + + // Restore brings everything back. + require.NoError(t, r.Restore(cp)) + + con, err := r.GetConstraint(ctx, constraintId) + require.NoError(t, err) + require.NotNil(t, con) + sub, err := r.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + require.NotNil(t, sub) + oi, err := r.GetOperationalIntent(ctx, operationalIntentId) + require.NoError(t, err) + require.NotNil(t, oi) +} + +func TestCheckpointIsolatesNotificationIndex(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + sub, err := r.UpsertSubscription(ctx, sampleSubscription()) + require.NoError(t, err) + + cp := r.Checkpoint() + + // In-place notification-index bump must not leak into the checkpoint. + bumped, err := r.IncrementNotificationIndicesForOperationalIntents(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, bumped, 1) + require.Equal(t, sub.NotificationIndex+1, bumped[0].NotificationIndex) + + require.NoError(t, r.Restore(cp)) + restored, err := r.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + require.Equal(t, sub.NotificationIndex, restored.NotificationIndex) +} + +func TestRestoreInvalidType(t *testing.T) { + require.Error(t, setUpStore(t).Restore("not a checkpoint")) +} diff --git a/pkg/scd/store/memstore/subscriptions.go b/pkg/scd/store/memstore/subscriptions.go index c284e6df8..1d4d067c3 100644 --- a/pkg/scd/store/memstore/subscriptions.go +++ b/pkg/scd/store/memstore/subscriptions.go @@ -5,44 +5,200 @@ import ( "time" "github.com/golang/geo/s2" - dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" ) +func (rec *subscriptionRecord) toModel() *scdmodels.Subscription { + return &scdmodels.Subscription{ + ID: rec.ID, + Version: scdmodels.NewOVNFromTime(rec.UpdatedAt, rec.ID.String()), + NotificationIndex: rec.NotificationIndex, + Manager: rec.Manager, + StartTime: cloneTime(rec.StartTime), + EndTime: cloneTime(rec.EndTime), + USSBaseURL: rec.USSBaseURL, + NotifyForOperationalIntents: rec.NotifyForOperationalIntents, + NotifyForConstraints: rec.NotifyForConstraints, + ImplicitSubscription: rec.ImplicitSubscription, + Cells: cloneCells(rec.Cells), + } +} + +// SearchSubscriptions implements scd.repos.Subscription.SearchSubscriptions. func (r *repo) SearchSubscriptions(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchSubscriptions not implemented for memstore") + cells, err := v4d.CalculateSpatialCovering() + if err != nil { + return nil, stacktrace.Propagate(err, "Could not calculate spatial covering") + } + if len(cells) == 0 { + return nil, nil + } + + want := cellSet(cells) + var out []*scdmodels.Subscription + for _, rec := range r.state.Subscriptions { + if !overlaps(rec.Cells, want) { + continue + } + // COALESCE(starts_at <= $3, true) with $3 = v4d.EndTime + if rec.StartTime != nil && v4d.EndTime != nil && rec.StartTime.After(*v4d.EndTime) { + continue + } + // COALESCE(ends_at >= $2, true) with $2 = v4d.StartTime + if rec.EndTime != nil && v4d.StartTime != nil && rec.EndTime.Before(*v4d.StartTime) { + continue + } + out = append(out, rec.toModel()) + if len(out) >= dssmodels.MaxResultLimit { // mirror SQL "LIMIT MaxResultLimit" + break + } + } + return out, nil } +// GetSubscription implements scd.repos.Subscription.GetSubscription. func (r *repo) GetSubscription(_ context.Context, id dssmodels.ID) (*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetSubscription not implemented for memstore") + rec, ok := r.state.Subscriptions[id] + if !ok { + return nil, nil + } + return rec.toModel(), nil } -func (r *repo) UpsertSubscription(_ context.Context, sub *scdmodels.Subscription) (*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpsertSubscription not implemented for memstore") +// UpsertSubscription implements scd.repos.Subscription.UpsertSubscription. +func (r *repo) UpsertSubscription(ctx context.Context, s *scdmodels.Subscription) (*scdmodels.Subscription, error) { + + now, err := timestamp.RequestTimestampFromContext(ctx) + + if err != nil { + return nil, err + } + + rec := &subscriptionRecord{ + ID: s.ID, + Manager: s.Manager, + NotificationIndex: s.NotificationIndex, + USSBaseURL: s.USSBaseURL, + NotifyForOperationalIntents: s.NotifyForOperationalIntents, + NotifyForConstraints: s.NotifyForConstraints, + ImplicitSubscription: s.ImplicitSubscription, + StartTime: cloneTime(s.StartTime), + EndTime: cloneTime(s.EndTime), + Cells: cloneCells(s.Cells), + UpdatedAt: now, + } + r.state.Subscriptions[s.ID] = rec + return rec.toModel(), nil } +// DeleteSubscription implements scd.repos.Subscription.DeleteSubscription. func (r *repo) DeleteSubscription(_ context.Context, id dssmodels.ID) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteSubscription not implemented for memstore") + if _, ok := r.state.Subscriptions[id]; !ok { + return stacktrace.NewError("Attempted to delete non-existent Subscription") + } + delete(r.state.Subscriptions, id) + return nil } +// IncrementNotificationIndicesForOperationalIntents implements +// scd.repos.Subscription.IncrementNotificationIndicesForOperationalIntents. func (r *repo) IncrementNotificationIndicesForOperationalIntents(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "IncrementNotificationIndicesForOperationalIntents not implemented for memstore") + cells, err := v4d.CalculateSpatialCovering() + if err != nil { + return nil, stacktrace.Propagate(err, "Could not calculate spatial covering") + } + if len(cells) == 0 { + return nil, nil + } + + want := cellSet(cells) + var out []*scdmodels.Subscription + for _, rec := range r.state.Subscriptions { + if !overlaps(rec.Cells, want) { + continue + } + if !rec.NotifyForOperationalIntents { + continue + } + // COALESCE(starts_at <= $3, true) with $3 = v4d.EndTime + if rec.StartTime != nil && v4d.EndTime != nil && rec.StartTime.After(*v4d.EndTime) { + continue + } + // COALESCE(ends_at >= $2, true) with $2 = v4d.StartTime + if rec.EndTime != nil && v4d.StartTime != nil && rec.EndTime.Before(*v4d.StartTime) { + continue + } + rec.NotificationIndex++ + out = append(out, rec.toModel()) + } + return out, nil } +// IncrementNotificationIndicesForConstraints implements +// scd.repos.Subscription.IncrementNotificationIndicesForConstraints. func (r *repo) IncrementNotificationIndicesForConstraints(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "IncrementNotificationIndicesForConstraints not implemented for memstore") + cells, err := v4d.CalculateSpatialCovering() + if err != nil { + return nil, stacktrace.Propagate(err, "Could not calculate spatial covering") + } + if len(cells) == 0 { + return nil, nil + } + + want := cellSet(cells) + var out []*scdmodels.Subscription + for _, rec := range r.state.Subscriptions { + if !overlaps(rec.Cells, want) { + continue + } + if !rec.NotifyForConstraints { + continue + } + // COALESCE(starts_at <= $3, true) with $3 = v4d.EndTime + if rec.StartTime != nil && v4d.EndTime != nil && rec.StartTime.After(*v4d.EndTime) { + continue + } + // COALESCE(ends_at >= $2, true) with $2 = v4d.StartTime + if rec.EndTime != nil && v4d.StartTime != nil && rec.EndTime.Before(*v4d.StartTime) { + continue + } + rec.NotificationIndex++ + out = append(out, rec.toModel()) + } + return out, nil } -func (r *repo) LockSubscriptionsOnCells(_ context.Context, cells s2.CellUnion, subscriptionIds []dssmodels.ID, startTime *time.Time, endTime *time.Time) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "LockSubscriptionsOnCells not implemented for memstore") +// LockSubscriptionsOnCells implements scd.repos.Subscription.LockSubscriptionsOnCells. +func (r *repo) LockSubscriptionsOnCells(_ context.Context, _ s2.CellUnion, _ []dssmodels.ID, _ *time.Time, _ *time.Time) error { + // For the memory store, that a no-op + return nil } +// ListExpiredSubscriptions implements scd.repos.Subscription.ListExpiredSubscriptions. func (r *repo) ListExpiredSubscriptions(_ context.Context, threshold time.Time) ([]*scdmodels.Subscription, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredSubscriptions not implemented for memstore") + var out []*scdmodels.Subscription + for _, rec := range r.state.Subscriptions { + // (ends_at IS NOT NULL AND ends_at <= threshold) OR (ends_at IS NULL AND updated_at <= threshold) + var expired bool + if rec.EndTime != nil { + expired = !rec.EndTime.After(threshold) + } else { + expired = !rec.UpdatedAt.After(threshold) + } + if !expired { + continue + } + out = append(out, rec.toModel()) + if len(out) >= dssmodels.MaxResultLimit { // mirror SQL "LIMIT MaxResultLimit" + break + } + } + return out, nil } func (r *repo) CountSubscriptions(_ context.Context) (int64, error) { - return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountSubscriptions not implemented for memstore") + return int64(len(r.state.Subscriptions)), nil } diff --git a/pkg/scd/store/memstore/subscriptions_test.go b/pkg/scd/store/memstore/subscriptions_test.go new file mode 100644 index 000000000..2176e411b --- /dev/null +++ b/pkg/scd/store/memstore/subscriptions_test.go @@ -0,0 +1,234 @@ +package memstore + +import ( + "testing" + "time" + + "github.com/golang/geo/s2" + dssmodels "github.com/interuss/dss/pkg/models" + scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/stretchr/testify/require" +) + +func TestSubscriptionUpsertGetDelete(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + got, err := r.UpsertSubscription(ctx, sampleSubscription()) + require.NoError(t, err) + require.Equal(t, subscriptionId, got.ID) + require.Equal(t, 1, got.NotificationIndex) + require.NotEmpty(t, got.Version) + + fetched, err := r.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + require.Equal(t, got.Version, fetched.Version) + require.True(t, fetched.NotifyForOperationalIntents) + + count, err := r.CountSubscriptions(ctx) + require.NoError(t, err) + require.Equal(t, int64(1), count) + + require.NoError(t, r.DeleteSubscription(ctx, subscriptionId)) + gone, err := r.GetSubscription(ctx, subscriptionId) + require.NoError(t, err) + require.Nil(t, gone) +} + +func TestSubscriptionGetMissingReturnsNil(t *testing.T) { + r := setUpStore(t) + got, err := r.GetSubscription(writeCtx(), subscriptionId) + require.NoError(t, err) + require.Nil(t, got) +} + +func TestSubscriptionDeleteMissingErrors(t *testing.T) { + r := setUpStore(t) + require.Error(t, r.DeleteSubscription(writeCtx(), subscriptionId)) +} + +func TestSearchSubscriptions(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + _, err := r.UpsertSubscription(ctx, sampleSubscription()) + require.NoError(t, err) + + res, err := r.SearchSubscriptions(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, res, 1) + + // No covering cells returns nil. + res, err = r.SearchSubscriptions(ctx, volume4D(s2.CellUnion{}, nil, nil, nil, nil)) + require.NoError(t, err) + require.Nil(t, res) +} + +func TestIncrementNotificationIndicesForOperationalIntents(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + // notify_for_operations = true. + opSub := sampleSubscription() + _, err := r.UpsertSubscription(ctx, opSub) + require.NoError(t, err) + + // A second subscription that only wants constraint notifications must be skipped. + conSub := sampleSubscription() + conSub.ID = "00000185-e36d-40be-8d38-beca6ca31aaa" + conSub.NotifyForOperationalIntents = false + conSub.NotifyForConstraints = true + _, err = r.UpsertSubscription(ctx, conSub) + require.NoError(t, err) + + got, err := r.IncrementNotificationIndicesForOperationalIntents(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, opSub.ID, got[0].ID) + require.Equal(t, opSub.NotificationIndex+1, got[0].NotificationIndex) + + // The bump is persisted. + fetched, err := r.GetSubscription(ctx, opSub.ID) + require.NoError(t, err) + require.Equal(t, opSub.NotificationIndex+1, fetched.NotificationIndex) + + // The constraint-only subscription was untouched. + other, err := r.GetSubscription(ctx, conSub.ID) + require.NoError(t, err) + require.Equal(t, conSub.NotificationIndex, other.NotificationIndex) + + // No covering cells returns nil. + got, err = r.IncrementNotificationIndicesForOperationalIntents(ctx, volume4D(s2.CellUnion{}, nil, nil, nil, nil)) + require.NoError(t, err) + require.Nil(t, got) +} + +func TestIncrementNotificationIndicesForConstraints(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + // notify_for_constraints = true (sample sets both notify flags). + conSub := sampleSubscription() + _, err := r.UpsertSubscription(ctx, conSub) + require.NoError(t, err) + + // A subscription that does not want constraint notifications must be skipped. + opSub := sampleSubscription() + opSub.ID = "00000185-e36d-40be-8d38-beca6ca31bbb" + opSub.NotifyForConstraints = false + _, err = r.UpsertSubscription(ctx, opSub) + require.NoError(t, err) + + got, err := r.IncrementNotificationIndicesForConstraints(ctx, volume4D(cells, nil, nil, nil, nil)) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, conSub.ID, got[0].ID) + require.Equal(t, conSub.NotificationIndex+1, got[0].NotificationIndex) + + other, err := r.GetSubscription(ctx, opSub.ID) + require.NoError(t, err) + require.Equal(t, opSub.NotificationIndex, other.NotificationIndex) +} + +func TestLockSubscriptionsOnCellsNoop(t *testing.T) { + r := setUpStore(t) + require.NoError(t, r.LockSubscriptionsOnCells(writeCtx(), cells, []dssmodels.ID{subscriptionId}, nil, nil)) +} + +var ( + sub1ID = dssmodels.ID("189ec22f-5e61-418a-940b-36de2d201fd5") + sub2ID = dssmodels.ID("78f98cc5-94f3-4c04-8da9-a8398feba3f3") + sub3ID = dssmodels.ID("9f0d4575-b275-4a4c-a261-e1e04d324565") +) + +var ( + sub1 = &scdmodels.Subscription{ + ID: sub1ID, + NotificationIndex: 1, + Manager: "unittest", + StartTime: &start1, + EndTime: &end1, + USSBaseURL: "https://dummy.uss", + NotifyForOperationalIntents: true, + NotifyForConstraints: false, + ImplicitSubscription: true, + Cells: cells, + } + sub2 = &scdmodels.Subscription{ + ID: sub2ID, + NotificationIndex: 1, + Manager: "unittest", + StartTime: &start2, + EndTime: &end2, + USSBaseURL: "https://dummy.uss", + NotifyForOperationalIntents: true, + NotifyForConstraints: false, + ImplicitSubscription: true, + Cells: cells, + } + sub3 = &scdmodels.Subscription{ + ID: sub3ID, + NotificationIndex: 1, + Manager: "unittest", + StartTime: &start3, + EndTime: &end3, + USSBaseURL: "https://dummy.uss", + NotifyForOperationalIntents: true, + NotifyForConstraints: false, + ImplicitSubscription: true, + Cells: cells, + } +) + +func TestListExpiredSubscriptions(t *testing.T) { + ctx := writeCtx() + r := setUpStore(t) + + _, err := r.UpsertSubscription(ctx, sub1) + require.NoError(t, err) + + _, err = r.UpsertSubscription(ctx, sub2) + require.NoError(t, err) + + _, err = r.UpsertSubscription(ctx, sub3) + require.NoError(t, err) + + testCases := []struct { + name string + timeRef time.Time + ttl time.Duration + expired []dssmodels.ID + }{{ + name: "none expired, one in close past", + timeRef: time.Date(2024, time.August, 25, 15, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{}, + }, { + name: "one recently expired, one current, one in future", + timeRef: time.Date(2024, time.September, 15, 16, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{sub1ID}, + }, { + name: "two expired, one in future", + timeRef: time.Date(2024, time.September, 16, 16, 0, 0, 0, time.UTC), + ttl: time.Hour * 2, + expired: []dssmodels.ID{sub1ID, sub2ID}, + }, { + name: "all expired", + timeRef: time.Date(2024, time.December, 15, 15, 0, 0, 0, time.UTC), + ttl: time.Hour * 24 * 30, + expired: []dssmodels.ID{sub1ID, sub2ID, sub3ID}, + }} + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + threshold := testCase.timeRef.Add(-testCase.ttl) + expired, err := r.ListExpiredSubscriptions(ctx, threshold) + require.NoError(t, err) + + expiredIDs := make([]dssmodels.ID, 0, len(expired)) + for _, expiredSub := range expired { + expiredIDs = append(expiredIDs, expiredSub.ID) + } + require.ElementsMatch(t, expiredIDs, testCase.expired) + }) + } +}