Fix(core): datastreams create with tenantID

services/core/measurements/application.go

@@ -105,8 +105,13 @@ func (s *Service) storePipelineMeasurement(ctx context.Context, msg pipeline.Mes
 		m.ObservedProperty = m.SensorExternalID + "_" + m.ObservedProperty
 	}
 
+	tenantID, err := auth.GetTenant(ctx)
+	if err != nil {
+		return err
+	}
+
 	// Find or create datastream
-	ds, err := FindOrCreateDatastream(sensor.ID, m.ObservedProperty, m.UnitOfMeasurement, s.store)
+	ds, err := FindOrCreateDatastream(tenantID, sensor.ID, m.ObservedProperty, m.UnitOfMeasurement, s.store)
 	if err != nil {
 		return err
 	}

services/core/measurements/application_test.go

@@ -130,7 +130,7 @@ func TestShouldErrorIfNoDeviceOrNoSensor(t *testing.T) {
 			msg.AccessToken = authtest.CreateToken()
 			require.NoError(t, err)
 			store := &StoreMock{
-				FindDatastreamFunc: func(sensorID int64, obs string) (*measurements.Datastream, error) {
+				FindDatastreamFunc: func(tenantID, sensorID int64, obs string) (*measurements.Datastream, error) {
 					assert.Equal(t, tC.expectedObservationProperty, obs, "expected observation property to match in query")
 					return &measurements.Datastream{}, nil
 				},
@@ -189,7 +189,7 @@ func TestShouldCopyOverDefaultFields(t *testing.T) {
 		UnitOfMeasurement: msg.Measurements[0].UnitOfMeasurement,
 	}
 	store := &StoreMock{
-		FindDatastreamFunc: func(sensorID int64, obs string) (*measurements.Datastream, error) {
+		FindDatastreamFunc: func(tenantID, sensorID int64, obs string) (*measurements.Datastream, error) {
 			return &ds, nil
 		},
 		InsertFunc: func(measurement measurements.Measurement) error {
@@ -326,7 +326,7 @@ func TestShouldChooseMeasurementLocationOverDeviceLocation(t *testing.T) {
 				UnitOfMeasurement: msg.Measurements[0].UnitOfMeasurement,
 			}
 			store := &StoreMock{
-				FindDatastreamFunc: func(sensorID int64, obs string) (*measurements.Datastream, error) {
+				FindDatastreamFunc: func(tenantID, sensorID int64, obs string) (*measurements.Datastream, error) {
 					return &ds, nil
 				},
 				InsertFunc: func(measurement measurements.Measurement) error {
@@ -407,7 +407,7 @@ func TestShouldSetExpirationDate(t *testing.T) {
 			UnitOfMeasurement: msg.Measurements[0].UnitOfMeasurement,
 		}
 		store := &StoreMock{
-			FindDatastreamFunc: func(sensorID int64, obs string) (*measurements.Datastream, error) {
+			FindDatastreamFunc: func(tenantID, sensorID int64, obs string) (*measurements.Datastream, error) {
 				return &ds, nil
 			},
 			InsertFunc: func(measurement measurements.Measurement) error {

services/core/measurements/datastreams.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"sensorbucket.nl/sensorbucket/internal/web"
 )
 
@@ -16,7 +17,7 @@ var (
 )
 
 type DatastreamFinderCreater interface {
-	FindDatastream(sensorID int64, observedProperty string) (*Datastream, error)
+	FindDatastream(tenantID, sensorID int64, observedProperty string) (*Datastream, error)
 	CreateDatastream(*Datastream) error
 }
 
@@ -27,9 +28,10 @@ type Datastream struct {
 	ObservedProperty  string    `json:"observed_property" db:"observed_property"`
 	UnitOfMeasurement string    `json:"unit_of_measurement" db:"unit_of_measurement"`
 	CreatedAt         time.Time `json:"created_at" db:"created_at"`
+	TenantID          int64     `json:"-"`
 }
 
-func newDatastream(sensorID int64, obs, uom string) (*Datastream, error) {
+func newDatastream(tenantID, sensorID int64, obs, uom string) (*Datastream, error) {
 	// TODO: Check UoM conforms to UCUM
 	if uom == "" || false {
 		return nil, ErrUoMInvalid
@@ -39,6 +41,7 @@ func newDatastream(sensorID int64, obs, uom string) (*Datastream, error) {
 	}
 	return &Datastream{
 		ID:                uuid.New(),
+		TenantID:          tenantID,
 		Description:       "",
 		SensorID:          sensorID,
 		ObservedProperty:  obs,
@@ -47,10 +50,10 @@ func newDatastream(sensorID int64, obs, uom string) (*Datastream, error) {
 	}, nil
 }
 
-func FindOrCreateDatastream(sensorID int64, obs, uom string, store DatastreamFinderCreater) (*Datastream, error) {
-	ds, err := store.FindDatastream(sensorID, obs)
+func FindOrCreateDatastream(tenantID, sensorID int64, obs, uom string, store DatastreamFinderCreater) (*Datastream, error) {
+	ds, err := store.FindDatastream(tenantID, sensorID, obs)
 	if errors.Is(err, ErrDatastreamNotFound) {
-		ds, err := newDatastream(sensorID, obs, uom)
+		ds, err := newDatastream(tenantID, sensorID, obs, uom)
 		if err != nil {
 			return nil, err
 		}

services/core/measurements/datastreams_test.go

@@ -6,12 +6,14 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"sensorbucket.nl/sensorbucket/pkg/authtest"
 	"sensorbucket.nl/sensorbucket/services/core/measurements"
 )
 
 func TestFindOrCreateDatastreamWorks(t *testing.T) {
 	store := &DatastreamFinderCreaterMock{
-		FindDatastreamFunc: func(sensorID int64, observationProperty string) (*measurements.Datastream, error) {
+		FindDatastreamFunc: func(tenantID, sensorID int64, observationProperty string) (*measurements.Datastream, error) {
 			return nil, measurements.ErrDatastreamNotFound
 		},
 		CreateDatastreamFunc: func(datastream *measurements.Datastream) error {
@@ -22,7 +24,7 @@ func TestFindOrCreateDatastreamWorks(t *testing.T) {
 	obs := "test_obs"
 	uom := "1/cm3"
 
-	ds, err := measurements.FindOrCreateDatastream(sensorID, obs, uom, store)
+	ds, err := measurements.FindOrCreateDatastream(authtest.DefaultTenantID, sensorID, obs, uom, store)
 	require.NoError(t, err)
 	assert.NotNil(t, ds, "FindOrCreateDatastream must return datastream if no error")
 
@@ -37,5 +39,4 @@ func TestFindOrCreateDatastreamWorks(t *testing.T) {
 	assert.Equal(t, sensorID, cds.SensorID)
 	assert.Equal(t, obs, cds.ObservedProperty)
 	assert.Equal(t, uom, cds.UnitOfMeasurement)
-
 }

services/core/measurements/infra/psql_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/jmoiron/sqlx"
 	"github.com/samber/lo"
@@ -16,6 +17,7 @@ import (
 	"github.com/testcontainers/testcontainers-go/wait"
 
 	"sensorbucket.nl/sensorbucket/internal/pagination"
+	"sensorbucket.nl/sensorbucket/pkg/authtest"
 	"sensorbucket.nl/sensorbucket/services/core/measurements"
 	measurementsinfra "sensorbucket.nl/sensorbucket/services/core/measurements/infra"
 	"sensorbucket.nl/sensorbucket/services/core/migrations"
@@ -106,3 +108,29 @@ func TestShouldQueryCorrectly(t *testing.T) {
 		})
 	}
 }
+
+func TestDatastreamCreated(t *testing.T) {
+	db := createPostgresServer(t)
+	store := measurementsinfra.NewPSQL(db)
+
+	ds := &measurements.Datastream{
+		ID:                uuid.New(),
+		UnitOfMeasurement: "#",
+		Description:       "",
+		SensorID:          1,
+		ObservedProperty:  "none",
+		CreatedAt:         time.Now(),
+		TenantID:          authtest.DefaultTenantID,
+	}
+	err := store.CreateDatastream(ds)
+	require.NoError(t, err)
+	ds2, err := store.FindDatastream(ds.TenantID, ds.SensorID, ds.ObservedProperty)
+	require.NoError(t, err)
+	assert.Equal(t, ds.ID, ds2.ID)
+	assert.Equal(t, ds.UnitOfMeasurement, ds2.UnitOfMeasurement)
+	assert.Equal(t, ds.Description, ds2.Description)
+	assert.Equal(t, ds.SensorID, ds2.SensorID)
+	assert.Equal(t, ds.ObservedProperty, ds2.ObservedProperty)
+	assert.WithinDuration(t, ds.CreatedAt, ds2.CreatedAt, time.Second)
+	assert.Equal(t, ds.TenantID, ds2.TenantID)
+}

services/core/measurements/infra/store_psql.go

@@ -231,21 +231,21 @@ func (s *MeasurementStorePSQL) Query(query measurements.Filter, r pagination.Req
 	return &page, nil
 }
 
-func (s *MeasurementStorePSQL) FindDatastream(sensorID int64, obs string) (*measurements.Datastream, error) {
+func (s *MeasurementStorePSQL) FindDatastream(tenantID, sensorID int64, obs string) (*measurements.Datastream, error) {
 	var ds measurements.Datastream
-	query := `
-		SELECT
-			"id", "description", "sensor_id", "observed_property", "unit_of_measurement",
-			"created_at"
-		FROM 
-			"datastreams"
-		WHERE
-			"sensor_id"=$1 AND "observed_property"=$2
-	`
-	if err := s.db.Get(&ds, query, sensorID, obs); err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
-			return nil, measurements.ErrDatastreamNotFound
-		}
+	err := pq.Select("id", "description", "sensor_id", "observed_property", "unit_of_measurement",
+		"created_at", "tenant_id").From("datastreams").Where(sq.Eq{
+		"sensor_id":         sensorID,
+		"observed_property": obs,
+		"tenant_id":         tenantID,
+	}).RunWith(s.db).Scan(
+		&ds.ID, &ds.Description, &ds.SensorID, &ds.ObservedProperty, &ds.UnitOfMeasurement, &ds.CreatedAt,
+		&ds.TenantID,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, measurements.ErrDatastreamNotFound
+	}
+	if err != nil {
 		return nil, fmt.Errorf("database error querying datastream: %w", err)
 	}
 	return &ds, nil
@@ -262,11 +262,11 @@ func (s *MeasurementStorePSQL) CreateDatastream(ds *measurements.Datastream) err
 	INSERT INTO
 		"datastreams" (
 			id, "description", "sensor_id", "observed_property", "unit_of_measurement",
-			"created_at"
+			"created_at", "tenant_id"
 		)
 	VALUES 
-		($1, $2, $3, $4, $5, $6)
-	`, uuidB, ds.Description, ds.SensorID, ds.ObservedProperty, ds.UnitOfMeasurement, ds.CreatedAt)
+		($1, $2, $3, $4, $5, $6, $7)
+	`, uuidB, ds.Description, ds.SensorID, ds.ObservedProperty, ds.UnitOfMeasurement, ds.CreatedAt, ds.TenantID)
 	if err != nil {
 		return fmt.Errorf("database error inserting datastream: %w", err)
 	}