From c152aa91dee1d1d913aa21c2d5087ed65d470838 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:13:18 +0100 Subject: [PATCH 1/4] feat(proto): Add CreateLocationEnergySource RPC This adds a new RPC, CreateLocationEnergySource. Currently, there is no way to add a new energy source (e.g. Wind) to an existing location (which is an energy source + geometry combination). This rectifies that lack, and also modifies create location such that it prevents creating an existing one. --- internal/server/dummy/dataserverimpl.go | 12 +++ internal/server/postgres/dataserverimpl.go | 100 +++++++++++++++--- .../server/postgres/dataserverimpl_test.go | 83 +++++++++++++++ proto/ocf/dp/dp-data.messages.proto | 26 +++++ proto/ocf/dp/dp-data.service.proto | 4 + 5 files changed, 209 insertions(+), 16 deletions(-) diff --git a/internal/server/dummy/dataserverimpl.go b/internal/server/dummy/dataserverimpl.go index e2bf5d7..93b33d9 100644 --- a/internal/server/dummy/dataserverimpl.go +++ b/internal/server/dummy/dataserverimpl.go @@ -299,6 +299,18 @@ func (d *DataPlatformDataServiceServerImpl) CreateLocation( }, nil } +// CreateLocationEnergySource implements dp.DataPlatformDataServiceServer. +func (d *DataPlatformDataServiceServerImpl) CreateLocationEnergySource( + ctx context.Context, + req *pb.CreateLocationEnergySourceRequest, +) (*pb.CreateLocationEnergySourceResponse, error) { + return &pb.CreateLocationEnergySourceResponse{ + LocationUuid: req.LocationUuid, + EnergySource: req.EnergySource, + EffectiveCapacityWatts: req.EffectiveCapacityWatts, + }, nil +} + // UpdateLocation implements dp.DataPlatformDataServiceServer. func (d *DataPlatformDataServiceServerImpl) UpdateLocation( ctx context.Context, diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index 1aeb7f6..d35b9fc 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -24,6 +24,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/timestamppb" pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" @@ -963,6 +964,37 @@ func (s *DataPlatformDataServiceServerImpl) GetLocationAsTimeseries( }, nil } +// createLocationSource inserts a source entry for a geometry. +// Importantly, this also refreshes the sources materialised view. +func createLocationSource( + ctx context.Context, + querier *db.Queries, + geometryUuid uuid.UUID, + sourceTypeID int16, + capacityWatts uint64, + metadata *structpb.Struct, + validFrom time.Time, +) (db.CreateSourceEntryRow, error) { + csprms := db.CreateSourceEntryParams{ + GeometryUuid: geometryUuid, + SourceTypeID: sourceTypeID, + CapacityWatts: int64(capacityWatts), + Metadata: metadata, + ValidFromUtc: pgtype.Timestamp{Time: validFrom, Valid: true}, + } + + dbSource, err := querier.CreateSourceEntry(ctx, csprms) + if err != nil { + return db.CreateSourceEntryRow{}, fmt.Errorf("invalid location source: %w", err) + } + + if err := querier.RefreshSourcesMaterializedView(ctx); err != nil { + return db.CreateSourceEntryRow{}, fmt.Errorf("failed to update sources materialised view: %w", err) + } + + return dbSource, nil +} + func (s *DataPlatformDataServiceServerImpl) CreateLocation( ctx context.Context, req *pb.CreateLocationRequest, @@ -1007,17 +1039,12 @@ func (s *DataPlatformDataServiceServerImpl) CreateLocation( req.ValidFromUtc = timestamppb.New(time.Now().UTC().Truncate(time.Minute)) } - csprms := db.CreateSourceEntryParams{ - GeometryUuid: dbLocation.GeometryUuid, - SourceTypeID: int16(req.EnergySource), - CapacityWatts: int64(req.EffectiveCapacityWatts), - Metadata: req.Metadata, - ValidFromUtc: pgtype.Timestamp{Time: req.ValidFromUtc.AsTime(), Valid: true}, - } - - dbSource, err := querier.CreateSourceEntry(ctx, csprms) + dbSource, err := createLocationSource( + ctx, querier, dbLocation.GeometryUuid, int16(req.EnergySource.Number()), + req.EffectiveCapacityWatts, req.Metadata, req.ValidFromUtc.AsTime(), + ) if err != nil { - return nil, fmt.Errorf("invalid location: %w", err) + return nil, err } l.Debug(). @@ -1027,16 +1054,57 @@ func (s *DataPlatformDataServiceServerImpl) CreateLocation( Str("dp.source.valid_from_utc", dbSource.ValidFromUtc.Time.String()). Msg("created source entry for location") - err = querier.RefreshSourcesMaterializedView(ctx) + return &pb.CreateLocationResponse{ + LocationUuid: dbLocation.GeometryUuid.String(), + LocationName: dbLocation.GeometryName, + EffectiveCapacityWatts: uint64(dbSource.CapacityWatts), + }, nil +} + +func (s *DataPlatformDataServiceServerImpl) CreateLocationEnergySource( + ctx context.Context, + req *pb.CreateLocationEnergySourceRequest, +) (*pb.CreateLocationEnergySourceResponse, error) { + l := zerolog.Ctx(ctx) + querier := db.New(ix.GetTxFromContext(ctx)) + + locationUuid := uuid.MustParse(req.LocationUuid) + + if req.ValidFromUtc == nil { + req.ValidFromUtc = timestamppb.New(time.Now().UTC().Truncate(time.Minute)) + } + + // Reject if this energy source already exists for the location at this time. + gsprms := db.GetSourceAtTimestampParams{ + GeometryUuid: locationUuid, + SourceTypeID: int16(req.EnergySource.Number()), + AtTimestampUtc: pgtype.Timestamp{Time: req.ValidFromUtc.AsTime(), Valid: true}, + } + if _, err := querier.GetSourceAtTimestamp(ctx, gsprms); err == nil { + return nil, status.Errorf( + codes.AlreadyExists, + "energy source '%s' already exists for location '%s'", + req.EnergySource, req.LocationUuid, + ) + } + + dbSource, err := createLocationSource( + ctx, querier, locationUuid, int16(req.EnergySource.Number()), + req.EffectiveCapacityWatts, req.Metadata, req.ValidFromUtc.AsTime(), + ) if err != nil { - return nil, fmt.Errorf("failed to update sources materialised view: %w", err) + return nil, err } - l.Debug().Msg("refreshed sources materialised view") + l.Debug(). + Str("dp.source.geometry_uuid", locationUuid.String()). + Int16("dp.source.type_id", int16(req.EnergySource.Number())). + Int64("dp.source.capacity", int64(dbSource.CapacityWatts)). + Msg("created new energy source for location") - return &pb.CreateLocationResponse{ - LocationUuid: dbLocation.GeometryUuid.String(), - LocationName: dbLocation.GeometryName, + return &pb.CreateLocationEnergySourceResponse{ + LocationUuid: locationUuid.String(), + EnergySource: req.EnergySource, EffectiveCapacityWatts: uint64(dbSource.CapacityWatts), }, nil } diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index a69c26e..44a4f1e 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -331,6 +331,89 @@ func TestCreateLocation(t *testing.T) { }) } +func TestCreateLocationEnergySource(t *testing.T) { + metadata := createTestMetadata(t, map[string]any{"source": "test_energy_source"}) + + // Create an initial location with solar + createResp := createTestLocation( + t, + "scotland_region", + "POINT(-0.127 51.507)", + 1230, + time.Now().UTC().Truncate(time.Minute), + metadata, + ) + + testcases := []struct { + name string + req *pb.CreateLocationEnergySourceRequest + shouldErr bool + errCodeCheck codes.Code + }{ + { + name: "Should add wind energy source to existing location", + req: &pb.CreateLocationEnergySourceRequest{ + LocationUuid: createResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_WIND, + EffectiveCapacityWatts: 5000, + Metadata: metadata, + }, + shouldErr: false, + }, + { + name: "Should fail to add already existing energy source", + req: &pb.CreateLocationEnergySourceRequest{ + LocationUuid: createResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, // Already exists + EffectiveCapacityWatts: 2000, + Metadata: metadata, + }, + shouldErr: true, + errCodeCheck: codes.AlreadyExists, + }, + { + name: "Should fail for non-existent location", + req: &pb.CreateLocationEnergySourceRequest{ + LocationUuid: uuid.New().String(), + EnergySource: pb.EnergySource_ENERGY_SOURCE_WIND, + EffectiveCapacityWatts: 5000, + Metadata: metadata, + }, + shouldErr: true, + }, + } + + for _, tc := range testcases { + t.Run(tc.name, func(t *testing.T) { + resp, err := dc.CreateLocationEnergySource(t.Context(), tc.req) + + if tc.shouldErr { + require.Error(t, err, "Expected an error") + if tc.errCodeCheck != codes.OK { + require.Equal(t, tc.errCodeCheck, status.Code(err)) + } + } else { + require.NoError(t, err, "Expected to be able to create the location energy source") + require.Equal(t, tc.req.LocationUuid, resp.LocationUuid) + require.Equal(t, tc.req.EnergySource, resp.EnergySource) + require.Equal(t, tc.req.EffectiveCapacityWatts, resp.EffectiveCapacityWatts) + + // Verify it can be fetched + resp2, err := dc.GetLocation( + t.Context(), + &pb.GetLocationRequest{ + LocationUuid: tc.req.LocationUuid, + EnergySource: tc.req.EnergySource, + IncludeGeometry: false, + }, + ) + require.NoError(t, err, "Expected to be able to fetch the newly created energy source") + require.Equal(t, tc.req.EffectiveCapacityWatts, resp2.EffectiveCapacityWatts) + } + }) + } +} + func TestUpdateLocation(t *testing.T) { metadata := createTestMetadata(t, map[string]any{"source": "test"}) diff --git a/proto/ocf/dp/dp-data.messages.proto b/proto/ocf/dp/dp-data.messages.proto index 6311958..68dc845 100644 --- a/proto/ocf/dp/dp-data.messages.proto +++ b/proto/ocf/dp/dp-data.messages.proto @@ -469,6 +469,32 @@ message CreateLocationResponse { } +message CreateLocationEnergySourceRequest { + string location_uuid = 1 [ + (buf.validate.field).required = true, + (buf.validate.field).string.uuid = true + ]; + EnergySource energy_source = 2 [ + (buf.validate.field).required = true + ]; + // The effective capacity of this source in watts. + uint64 effective_capacity_watts = 3 [ + (buf.validate.field).required = true + ]; + optional google.protobuf.Struct metadata = 4; + // The UTC time from which this source is considered valid. Leave empty to use current time. + optional google.protobuf.Timestamp valid_from_utc = 5 [ + (buf.validate.field).timestamp = { gt: { seconds: 112000000}, lt_now: true } + ]; +} + +message CreateLocationEnergySourceResponse { + string location_uuid = 1; + EnergySource energy_source = 2; + uint64 effective_capacity_watts = 3; +} + + message GetLocationRequest { string location_uuid = 1 [ (buf.validate.field).required = true, diff --git a/proto/ocf/dp/dp-data.service.proto b/proto/ocf/dp/dp-data.service.proto index e32e653..7cde416 100644 --- a/proto/ocf/dp/dp-data.service.proto +++ b/proto/ocf/dp/dp-data.service.proto @@ -34,6 +34,10 @@ service DataPlatformDataService { rpc GetLocationAsTimeseries(GetLocationAsTimeseriesRequest) returns (GetLocationAsTimeseriesResponse) {} /* CreateLocation registers a new location in which to log or forecast generation. */ rpc CreateLocation(CreateLocationRequest) returns (CreateLocationResponse) {} + /* CreateLocationEnergySource attaches a new energy source to an existing location. + * Errors if the location already has a source of this type. + */ + rpc CreateLocationEnergySource(CreateLocationEnergySourceRequest) returns (CreateLocationEnergySourceResponse) {} /* UpdateLocation modifies various attributes associated with a given location. */ rpc UpdateLocation(UpdateLocationRequest) returns (UpdateLocationResponse) {} /* UpdateLocationOwner changes the ownership of a location. */ From 0baaf20aa68b3947bb41720de214b3f17610034a Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:16:12 +0100 Subject: [PATCH 2/4] chore(repo): Linting --- internal/server/postgres/dataserverimpl.go | 5 ++++- internal/server/postgres/dataserverimpl_test.go | 7 ++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index d35b9fc..c9d7b99 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -989,7 +989,10 @@ func createLocationSource( } if err := querier.RefreshSourcesMaterializedView(ctx); err != nil { - return db.CreateSourceEntryRow{}, fmt.Errorf("failed to update sources materialised view: %w", err) + return db.CreateSourceEntryRow{}, fmt.Errorf( + "failed to update sources materialised view: %w", + err, + ) } return dbSource, nil diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index 44a4f1e..2d0784f 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -389,6 +389,7 @@ func TestCreateLocationEnergySource(t *testing.T) { if tc.shouldErr { require.Error(t, err, "Expected an error") + if tc.errCodeCheck != codes.OK { require.Equal(t, tc.errCodeCheck, status.Code(err)) } @@ -407,7 +408,11 @@ func TestCreateLocationEnergySource(t *testing.T) { IncludeGeometry: false, }, ) - require.NoError(t, err, "Expected to be able to fetch the newly created energy source") + require.NoError( + t, + err, + "Expected to be able to fetch the newly created energy source", + ) require.Equal(t, tc.req.EffectiveCapacityWatts, resp2.EffectiveCapacityWatts) } }) From 20524203bf907d324c4f1ba531fa71ea866a3974 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:35:20 +0100 Subject: [PATCH 3/4] fix: Fail CreateLocationEnergySource if it lready exists --- internal/server/postgres/dataserverimpl.go | 17 ++++++++++------- internal/server/postgres/dataserverimpl_test.go | 14 +++++++++++++- .../server/postgres/sql/queries/locations.sql | 9 +++++++++ 3 files changed, 32 insertions(+), 8 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index c9d7b99..30ac8d0 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -1077,16 +1077,19 @@ func (s *DataPlatformDataServiceServerImpl) CreateLocationEnergySource( req.ValidFromUtc = timestamppb.New(time.Now().UTC().Truncate(time.Minute)) } - // Reject if this energy source already exists for the location at this time. - gsprms := db.GetSourceAtTimestampParams{ - GeometryUuid: locationUuid, - SourceTypeID: int16(req.EnergySource.Number()), - AtTimestampUtc: pgtype.Timestamp{Time: req.ValidFromUtc.AsTime(), Valid: true}, + // Reject if this energy source already exists for the location. + cseprms := db.CheckSourceExistsParams{ + GeometryUuid: locationUuid, + SourceTypeID: int16(req.EnergySource.Number()), + } + exists, err := querier.CheckSourceExists(ctx, cseprms) + if err != nil { + return nil, fmt.Errorf("failed to check source existence: %w", err) } - if _, err := querier.GetSourceAtTimestamp(ctx, gsprms); err == nil { + if exists { return nil, status.Errorf( codes.AlreadyExists, - "energy source '%s' already exists for location '%s'", + "energy source '%s' has already been created for location '%s'. New inserts to an existing energy source must go through the UpdateLocation RPC.", req.EnergySource, req.LocationUuid, ) } diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index 2d0784f..969fa0b 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -334,7 +334,7 @@ func TestCreateLocation(t *testing.T) { func TestCreateLocationEnergySource(t *testing.T) { metadata := createTestMetadata(t, map[string]any{"source": "test_energy_source"}) - // Create an initial location with solar + // Create a location with solar valid from today createResp := createTestLocation( t, "scotland_region", @@ -371,6 +371,18 @@ func TestCreateLocationEnergySource(t *testing.T) { shouldErr: true, errCodeCheck: codes.AlreadyExists, }, + { + name: "Should fail if the energy source already exists but we query a past valid_from", + req: &pb.CreateLocationEnergySourceRequest{ + LocationUuid: createResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, // Created today + EffectiveCapacityWatts: 2000, + ValidFromUtc: timestamppb.New(time.Now().UTC().Add(-24 * time.Hour).Truncate(time.Minute)), + Metadata: metadata, + }, + shouldErr: true, + errCodeCheck: codes.AlreadyExists, + }, { name: "Should fail for non-existent location", req: &pb.CreateLocationEnergySourceRequest{ diff --git a/internal/server/postgres/sql/queries/locations.sql b/internal/server/postgres/sql/queries/locations.sql index 59ef8c9..17b36d3 100644 --- a/internal/server/postgres/sql/queries/locations.sql +++ b/internal/server/postgres/sql/queries/locations.sql @@ -111,6 +111,15 @@ WHERE AND s.source_type_id = $2 AND s.sys_period @> sqlc.arg(at_timestamp_utc)::TIMESTAMP; +-- name: CheckSourceExists :one +/* CheckSourceExists returns true if the given geometry and source type has ever had an entry. */ +SELECT EXISTS ( + SELECT 1 + FROM loc.sources_history + WHERE geometry_uuid = $1 + AND source_type_id = $2 +); + -- name: CreateSourceEntry :one /* CreateSourceEntry creates a new source entry for a given geometry and source type. * It fetches the state prior to the input valid time, and only inserts the new row if it differs From dfb8f115e5fa90c045aceb439593c91078ae0c75 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:40:56 +0100 Subject: [PATCH 4/4] chore: linting --- internal/server/postgres/dataserverimpl.go | 5 ++++- internal/server/postgres/dataserverimpl_test.go | 6 ++++-- internal/server/postgres/sql/queries/locations.sql | 2 +- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index 30ac8d0..3fed77a 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -1082,15 +1082,18 @@ func (s *DataPlatformDataServiceServerImpl) CreateLocationEnergySource( GeometryUuid: locationUuid, SourceTypeID: int16(req.EnergySource.Number()), } + exists, err := querier.CheckSourceExists(ctx, cseprms) if err != nil { return nil, fmt.Errorf("failed to check source existence: %w", err) } + if exists { return nil, status.Errorf( codes.AlreadyExists, "energy source '%s' has already been created for location '%s'. New inserts to an existing energy source must go through the UpdateLocation RPC.", - req.EnergySource, req.LocationUuid, + req.EnergySource, + req.LocationUuid, ) } diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index 969fa0b..f76b743 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -377,8 +377,10 @@ func TestCreateLocationEnergySource(t *testing.T) { LocationUuid: createResp.LocationUuid, EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, // Created today EffectiveCapacityWatts: 2000, - ValidFromUtc: timestamppb.New(time.Now().UTC().Add(-24 * time.Hour).Truncate(time.Minute)), - Metadata: metadata, + ValidFromUtc: timestamppb.New( + time.Now().UTC().Add(-24 * time.Hour).Truncate(time.Minute), + ), + Metadata: metadata, }, shouldErr: true, errCodeCheck: codes.AlreadyExists, diff --git a/internal/server/postgres/sql/queries/locations.sql b/internal/server/postgres/sql/queries/locations.sql index 17b36d3..1f0c890 100644 --- a/internal/server/postgres/sql/queries/locations.sql +++ b/internal/server/postgres/sql/queries/locations.sql @@ -113,7 +113,7 @@ WHERE -- name: CheckSourceExists :one /* CheckSourceExists returns true if the given geometry and source type has ever had an entry. */ -SELECT EXISTS ( +SELECT EXISTS( SELECT 1 FROM loc.sources_history WHERE geometry_uuid = $1