From 02a3a9e3c1c1cf321487ab3d5f93958de25164fa Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:06:36 +0100 Subject: [PATCH] fix(proto): Add metadata updating to the Batch route The locations that need to have their capacities updated in a batch, also often require bulk metadata modification. This adds that functionality to the BatchUpdateLocationCapacities route. --- internal/server/postgres/dataserverimpl.go | 36 ++- .../server/postgres/dataserverimpl_test.go | 220 +++++++++++++----- .../server/postgres/sql/queries/locations.sql | 49 +++- proto/ocf/dp/dp-data.messages.proto | 29 ++- proto/ocf/dp/dp-data.service.proto | 5 +- 5 files changed, 256 insertions(+), 83 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index 32b6dcf..770d7c5 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -1282,10 +1282,11 @@ func (s *DataPlatformDataServiceServerImpl) UpdateLocationOwner( }, nil } -// BatchUpdateLocationCapacity updates the effective capacity of many locations of a single -// energy source in one transaction, refreshing the sources materialised view only once at the -// end. This is intended for bulk maintenance jobs that would otherwise overload the database by -// calling UpdateLocation once per location. +// BatchUpdateLocationCapacity updates the effective capacity and/or metadata of many locations of +// a single energy source in one transaction, refreshing the sources materialised view only once +// at the end. Fields left unset on an update keep their existing values. This is intended for +// bulk maintenance jobs that would otherwise overload the database by calling UpdateLocation once +// per location. func (s *DataPlatformDataServiceServerImpl) BatchUpdateLocationCapacity( ctx context.Context, req *pb.BatchUpdateLocationCapacityRequest, @@ -1298,19 +1299,38 @@ func (s *DataPlatformDataServiceServerImpl) BatchUpdateLocationCapacity( validFrom = req.ValidFromUtc.AsTime().UTC() } - geometryUuids := make([]uuid.UUID, len(req.Updates)) - capacityWatts := make([]int64, len(req.Updates)) + // Fields left unset on an update keep their existing values. The placeholder values for unset + // fields are ignored by the query. + n := len(req.Updates) + geometryUuids := make([]uuid.UUID, n) + setCapacity := make([]bool, n) + capacityWatts := make([]int64, n) + setMetadata := make([]bool, n) + metadata := make([]*structpb.Struct, n) for i, u := range req.Updates { geometryUuids[i] = uuid.MustParse(u.LocationUuid) - capacityWatts[i] = int64(u.NewEffectiveCapacityWatts) + + if u.NewEffectiveCapacityWatts != nil { + setCapacity[i] = true + capacityWatts[i] = int64(*u.NewEffectiveCapacityWatts) + } + + metadata[i] = &structpb.Struct{} + if u.NewMetadata != nil { + setMetadata[i] = true + metadata[i] = u.NewMetadata + } } bcprms := db.BatchCreateSourceEntriesParams{ SourceTypeID: int16(req.EnergySource.Number()), ValidFromUtc: pgtype.Timestamp{Time: validFrom, Valid: true}, GeometryUuids: geometryUuids, + SetCapacity: setCapacity, CapacityWatts: capacityWatts, + SetMetadata: setMetadata, + Metadata: metadata, } dbUpdated, err := querier.BatchCreateSourceEntries(ctx, bcprms) @@ -1328,7 +1348,7 @@ func (s *DataPlatformDataServiceServerImpl) BatchUpdateLocationCapacity( Int16("dp.source.type_id", bcprms.SourceTypeID). Int("dp.locations.requested", len(req.Updates)). Int("dp.locations.updated", len(dbUpdated)). - Msg("batch updated location capacities") + Msg("batch updated location capacities and metadata") return &pb.BatchUpdateLocationCapacityResponse{ UpdatedCount: uint32(len(dbUpdated)), diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index 6cfcc99..c42ee5b 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -643,78 +643,182 @@ func TestUpdateLocationOwner(t *testing.T) { } func TestBatchUpdateLocationCapacity(t *testing.T) { - metadata := createTestMetadata(t, map[string]any{"source": "test"}) + initialMetadata := createTestMetadata(t, map[string]any{"source": "test"}) + newMetadata := createTestMetadata(t, map[string]any{"source": "updated", "extra": 1}) pivotTime := time.Date(2019, 5, 6, 6, 0, 0, 0, time.UTC) - locA := createTestLocation( - t, "test_batch_capacity_site_a", "POINT(-0.1 51.5)", 1000e6, pivotTime, metadata, "GB", - ) - locB := createTestLocation( - t, "test_batch_capacity_site_b", "POINT(-0.2 51.6)", 2000e6, pivotTime, metadata, "GB", - ) + type expectedState struct { + capacityWatts uint64 + metadata map[string]any + } + // Each case operates on its own pair of locations, A and B, both created with 1000MW/2000MW + // capacity and the initial metadata. The updates are built from the created location UUIDs. testcases := []struct { name string - req *pb.BatchUpdateLocationCapacityRequest + updates func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update expectedUpdatedCount uint32 expectedUnchangedCount uint32 - expectedCapacityWatts map[string]uint64 - shouldErr bool + expectedA expectedState + expectedB expectedState + expectedErrCode codes.Code }{ { - name: "Should update only the location whose capacity changed", - req: &pb.BatchUpdateLocationCapacityRequest{ - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - ValidFromUtc: timestamppb.New(pivotTime.Add(time.Hour)), - Updates: []*pb.BatchUpdateLocationCapacityRequest_Update{ - {LocationUuid: locA.LocationUuid, NewEffectiveCapacityWatts: 1500e6}, - {LocationUuid: locB.LocationUuid, NewEffectiveCapacityWatts: 2000e6}, - }, + name: "Should update only the location whose capacity changed, keeping metadata", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewEffectiveCapacityWatts: ptr(uint64(1500e6))}, + {LocationUuid: b, NewEffectiveCapacityWatts: ptr(uint64(2000e6))}, + } }, expectedUpdatedCount: 1, expectedUnchangedCount: 1, - expectedCapacityWatts: map[string]uint64{ - locA.LocationUuid: 1500e6, - locB.LocationUuid: 2000e6, + expectedA: expectedState{1500e6, initialMetadata.AsMap()}, + expectedB: expectedState{2000e6, initialMetadata.AsMap()}, + }, + { + name: "Should update metadata only, keeping capacity", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewMetadata: newMetadata}, + } + }, + expectedUpdatedCount: 1, + expectedUnchangedCount: 0, + expectedA: expectedState{1000e6, newMetadata.AsMap()}, + expectedB: expectedState{2000e6, initialMetadata.AsMap()}, + }, + { + name: "Should update capacity and metadata together", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + { + LocationUuid: a, + NewEffectiveCapacityWatts: ptr(uint64(1100e6)), + NewMetadata: newMetadata, + }, + } + }, + expectedUpdatedCount: 1, + expectedUnchangedCount: 0, + expectedA: expectedState{1100e6, newMetadata.AsMap()}, + expectedB: expectedState{2000e6, initialMetadata.AsMap()}, + }, + { + name: "Should apply a mix of capacity-only and metadata-only updates to the right rows", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewMetadata: newMetadata}, + {LocationUuid: b, NewEffectiveCapacityWatts: ptr(uint64(2500e6))}, + } }, + expectedUpdatedCount: 2, + expectedUnchangedCount: 0, + expectedA: expectedState{1000e6, newMetadata.AsMap()}, + expectedB: expectedState{2500e6, initialMetadata.AsMap()}, + }, + { + name: "Should clear metadata when given an empty object", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewMetadata: &structpb.Struct{}}, + } + }, + expectedUpdatedCount: 1, + expectedUnchangedCount: 0, + expectedA: expectedState{1000e6, map[string]any{}}, + expectedB: expectedState{2000e6, initialMetadata.AsMap()}, + }, + { + name: "Should treat identical capacity and metadata as unchanged", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + { + LocationUuid: a, + NewEffectiveCapacityWatts: ptr(uint64(1000e6)), + NewMetadata: initialMetadata, + }, + {LocationUuid: b, NewMetadata: initialMetadata}, + } + }, + expectedUpdatedCount: 0, + expectedUnchangedCount: 2, + expectedA: expectedState{1000e6, initialMetadata.AsMap()}, + expectedB: expectedState{2000e6, initialMetadata.AsMap()}, }, { name: "Should fail the whole batch for an unknown location uuid", - req: &pb.BatchUpdateLocationCapacityRequest{ - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - ValidFromUtc: timestamppb.New(pivotTime.Add(2 * time.Hour)), - Updates: []*pb.BatchUpdateLocationCapacityRequest_Update{ - {LocationUuid: locA.LocationUuid, NewEffectiveCapacityWatts: 1600e6}, - {LocationUuid: uuid.NewString(), NewEffectiveCapacityWatts: 1e6}, - }, + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewEffectiveCapacityWatts: ptr(uint64(1600e6))}, + {LocationUuid: uuid.NewString(), NewEffectiveCapacityWatts: ptr(uint64(1e6))}, + } }, - shouldErr: true, + expectedErrCode: codes.FailedPrecondition, + }, + { + name: "Should reject an update with neither capacity nor metadata", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a}, + } + }, + expectedErrCode: codes.InvalidArgument, + }, + { + name: "Should reject duplicate location uuids in a batch", + updates: func(a, b string) []*pb.BatchUpdateLocationCapacityRequest_Update { + return []*pb.BatchUpdateLocationCapacityRequest_Update{ + {LocationUuid: a, NewEffectiveCapacityWatts: ptr(uint64(1100e6))}, + {LocationUuid: a, NewMetadata: newMetadata}, + } + }, + expectedErrCode: codes.InvalidArgument, }, } - for _, tc := range testcases { + for i, tc := range testcases { t.Run(tc.name, func(t *testing.T) { - resp, err := dc.BatchUpdateLocationCapacity(t.Context(), tc.req) + locA := createTestLocation( + t, fmt.Sprintf("test_batch_capacity_site_a_%d", i), "POINT(-0.1 51.5)", + 1000e6, pivotTime, initialMetadata, "GB", + ) + locB := createTestLocation( + t, fmt.Sprintf("test_batch_capacity_site_b_%d", i), "POINT(-0.2 51.6)", + 2000e6, pivotTime, initialMetadata, "GB", + ) - if tc.shouldErr { + req := &pb.BatchUpdateLocationCapacityRequest{ + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + ValidFromUtc: timestamppb.New(pivotTime.Add(time.Hour)), + Updates: tc.updates(locA.LocationUuid, locB.LocationUuid), + } + + resp, err := dc.BatchUpdateLocationCapacity(t.Context(), req) + + if tc.expectedErrCode != codes.OK { require.Error(t, err) - } else { - require.NoError(t, err) - require.Equal(t, tc.expectedUpdatedCount, resp.UpdatedCount) - require.Equal(t, tc.expectedUnchangedCount, resp.UnchangedCount) + require.Equal(t, tc.expectedErrCode, status.Code(err), err.Error()) + return + } - for locationUuid, expectedCapacity := range tc.expectedCapacityWatts { - getResp, err := dc.GetLocation(t.Context(), &pb.GetLocationRequest{ - LocationUuid: locationUuid, - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - PivotTimestampUtc: timestamppb.New( - tc.req.ValidFromUtc.AsTime().Add(time.Minute), - ), - }) - require.NoError(t, err) - require.Equal(t, expectedCapacity, getResp.EffectiveCapacityWatts) - require.Equal(t, metadata.AsMap(), getResp.Metadata.AsMap()) - } + require.NoError(t, err) + require.Equal(t, tc.expectedUpdatedCount, resp.UpdatedCount) + require.Equal(t, tc.expectedUnchangedCount, resp.UnchangedCount) + + expected := map[string]expectedState{ + locA.LocationUuid: tc.expectedA, + locB.LocationUuid: tc.expectedB, + } + for locationUuid, want := range expected { + getResp, err := dc.GetLocation(t.Context(), &pb.GetLocationRequest{ + LocationUuid: locationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + PivotTimestampUtc: timestamppb.New(req.ValidFromUtc.AsTime().Add(time.Minute)), + }) + require.NoError(t, err) + require.Equal(t, want.capacityWatts, getResp.EffectiveCapacityWatts) + require.Equal(t, want.metadata, getResp.Metadata.AsMap()) } }) } @@ -1982,9 +2086,9 @@ func TestGetLatestObservations(t *testing.T) { // Seed some observations for the site. // Create the older observation _, err = dc.CreateObservations(t.Context(), &pb.CreateObservationsRequest{ - LocationUuid: siteResp.LocationUuid, - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - ObserverName: obsResp.ObserverName, + LocationUuid: siteResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + ObserverName: obsResp.ObserverName, CreatedTimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 2)), Values: []*pb.CreateObservationsRequest_Value{ { @@ -1997,9 +2101,9 @@ func TestGetLatestObservations(t *testing.T) { // Create the newer observation _, err = dc.CreateObservations(t.Context(), &pb.CreateObservationsRequest{ - LocationUuid: siteResp.LocationUuid, - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - ObserverName: obsResp.ObserverName, + LocationUuid: siteResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + ObserverName: obsResp.ObserverName, CreatedTimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 1)), Values: []*pb.CreateObservationsRequest_Value{ { @@ -2223,11 +2327,11 @@ func TestCreateObservations(t *testing.T) { { name: "Should create valid observations with explicitly provided created timestamp", req: &pb.CreateObservationsRequest{ - LocationUuid: siteResp.LocationUuid, - EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, - ObserverName: obsResp.ObserverName, + LocationUuid: siteResp.LocationUuid, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + ObserverName: obsResp.ObserverName, CreatedTimestampUtc: timestamppb.New(time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)), - Values: validObservations, + Values: validObservations, }, }, { diff --git a/internal/server/postgres/sql/queries/locations.sql b/internal/server/postgres/sql/queries/locations.sql index 8134934..36c8447 100644 --- a/internal/server/postgres/sql/queries/locations.sql +++ b/internal/server/postgres/sql/queries/locations.sql @@ -172,42 +172,71 @@ RETURNING geometry_uuid, source_type_id, capacity_watts, valid_from_utc, metadat -- name: BatchCreateSourceEntries :many /* BatchCreateSourceEntries creates new source history entries for many geometries of a single - * source type at once. Like CreateSourceEntry, it only inserts rows whose capacity differs from - * the geometry's previous state, but does so set-based instead of one row at a time. + * source type at once. Like CreateSourceEntry, it only inserts rows whose capacity or metadata + * differ from the geometry's previous state, but does so set-based instead of one row at a time. + * Where set_capacity (or set_metadata) is false for a geometry, the corresponding capacity_watts + * (or metadata) value is ignored and the previous state's value is kept. An empty metadata object + * clears the metadata. * Callers are responsible for refreshing loc.sources_mv afterwards - this is intentionally not * done here so that many geometries can be updated with a single refresh. */ WITH input AS ( SELECT u.geometry_uuid, - c.capacity_watts + sc.set_capacity, + c.capacity_watts, + sm.set_metadata, + m.metadata FROM UNNEST(sqlc.arg(geometry_uuids)::UUID[]) WITH ORDINALITY AS u (geometry_uuid, ord) + INNER JOIN UNNEST(sqlc.arg(set_capacity)::BOOLEAN[]) WITH ORDINALITY AS sc (set_capacity, ord) + ON u.ord = sc.ord INNER JOIN UNNEST(sqlc.arg(capacity_watts)::BIGINT[]) WITH ORDINALITY AS c (capacity_watts, ord) ON u.ord = c.ord + INNER JOIN UNNEST(sqlc.arg(set_metadata)::BOOLEAN[]) WITH ORDINALITY AS sm (set_metadata, ord) + ON u.ord = sm.ord + INNER JOIN UNNEST(sqlc.arg(metadata)::JSONB[]) WITH ORDINALITY AS m (metadata, ord) + ON u.ord = m.ord ), prev_state AS ( SELECT DISTINCT ON (sh.geometry_uuid) sh.geometry_uuid, sh.capacity_watts, + sh.capacity_limit_sip, sh.metadata FROM loc.sources_history AS sh INNER JOIN input USING (geometry_uuid) WHERE sh.source_type_id = sqlc.arg(source_type_id)::SMALLINT AND sh.valid_from_utc <= sqlc.arg(valid_from_utc)::TIMESTAMP ORDER BY sh.geometry_uuid ASC, sh.valid_from_utc DESC +), +new_state AS ( + SELECT + i.geometry_uuid, + p.capacity_watts AS prev_capacity_watts, + p.metadata AS prev_metadata, + p.capacity_limit_sip, + CASE WHEN i.set_capacity THEN i.capacity_watts ELSE p.capacity_watts END AS capacity_watts, + CASE + WHEN NOT i.set_metadata THEN p.metadata + WHEN i.metadata = '{}'::JSONB THEN NULL + ELSE i.metadata + END AS metadata + FROM input AS i + LEFT OUTER JOIN prev_state AS p USING (geometry_uuid) ) INSERT INTO loc.sources_history ( - geometry_uuid, source_type_id, capacity_watts, valid_from_utc, metadata + geometry_uuid, source_type_id, capacity_watts, capacity_limit_sip, valid_from_utc, metadata ) SELECT - i.geometry_uuid, + n.geometry_uuid, sqlc.arg(source_type_id)::SMALLINT, - i.capacity_watts, + n.capacity_watts, + n.capacity_limit_sip, sqlc.arg(valid_from_utc)::TIMESTAMP, - p.metadata -FROM input AS i - LEFT OUTER JOIN prev_state AS p USING (geometry_uuid) -WHERE p.capacity_watts IS DISTINCT FROM i.capacity_watts + n.metadata +FROM new_state AS n +WHERE n.prev_capacity_watts IS DISTINCT FROM n.capacity_watts + OR n.prev_metadata IS DISTINCT FROM n.metadata RETURNING geometry_uuid, capacity_watts; -- name: RefreshSourcesMaterializedView :exec diff --git a/proto/ocf/dp/dp-data.messages.proto b/proto/ocf/dp/dp-data.messages.proto index 951c97d..e142f18 100644 --- a/proto/ocf/dp/dp-data.messages.proto +++ b/proto/ocf/dp/dp-data.messages.proto @@ -638,7 +638,21 @@ message BatchUpdateLocationCapacityRequest { (buf.validate.field).required = true, (buf.validate.field).string.uuid = true ]; - uint64 new_effective_capacity_watts = 2; + // The new effective capacity. Leave unset to keep the existing capacity. + optional uint64 new_effective_capacity_watts = 2; + // The new metadata object to set for the location. Leave unset to keep the existing metadata. + // Note that this will replace any existing metadata, so be sure to include existing fields where + // needed. An empty object clears the metadata. + optional google.protobuf.Struct new_metadata = 3; + + // Ensure one of the updatable fields is set. + option (buf.validate.message).cel = { + id: "has_update_field" + message: "at least one updatable field must be set" + expression: + "has(this.new_effective_capacity_watts) " + "|| has(this.new_metadata)" + }; } EnergySource energy_source = 1 [ @@ -646,7 +660,12 @@ message BatchUpdateLocationCapacityRequest { ]; repeated BatchUpdateLocationCapacityRequest.Update updates = 2 [ (buf.validate.field).repeated.min_items = 1, - (buf.validate.field).repeated.max_items = 5000 + (buf.validate.field).repeated.max_items = 5000, + (buf.validate.field).cel = { + id: "unique_location_uuids" + message: "each location_uuid may only appear once per batch" + expression: "this.map(u, u.location_uuid).unique()" + } ]; // The UTC time from which these capacities are considered valid. Leave empty to use current time. optional google.protobuf.Timestamp valid_from_utc = 3 [ @@ -655,10 +674,10 @@ message BatchUpdateLocationCapacityRequest { } message BatchUpdateLocationCapacityResponse { - // The number of locations whose capacity changed and was recorded. + // The number of locations whose capacity or metadata changed and was recorded. uint32 updated_count = 1; - // The number of locations whose requested capacity matched their existing capacity, and so - // were left unchanged. + // The number of locations whose requested capacity and metadata matched their existing state, + // and so were left unchanged. uint32 unchanged_count = 2; } diff --git a/proto/ocf/dp/dp-data.service.proto b/proto/ocf/dp/dp-data.service.proto index 67eb9d3..bba4de6 100644 --- a/proto/ocf/dp/dp-data.service.proto +++ b/proto/ocf/dp/dp-data.service.proto @@ -42,8 +42,9 @@ service DataPlatformDataService { rpc UpdateLocation(UpdateLocationRequest) returns (UpdateLocationResponse) {} /* UpdateLocationOwner changes the ownership of a location. */ rpc UpdateLocationOwner(UpdateLocationOwnerRequest) returns (UpdateLocationOwnerResponse) {} - /* BatchUpdateLocationCapacity updates the effective capacity of many locations of a single - * energy source in one call. Intended for bulk maintenance jobs. + /* BatchUpdateLocationCapacity updates the effective capacity and/or metadata of many locations + * of a single energy source in one call. Fields left unset on an update keep their existing + * values. Intended for bulk maintenance jobs. */ rpc BatchUpdateLocationCapacity(BatchUpdateLocationCapacityRequest) returns (BatchUpdateLocationCapacityResponse) {}