fix(proto): Track observations created time

internal/server/postgres/dataserverimpl.go

@@ -566,12 +566,20 @@ func (s *DataPlatformDataServiceServerImpl) GetObservationsAsTimeseries(
 
 	start, end := timeWindowToPgWindow(req.TimeWindow)
 
+	var pivotTime pgtype.Timestamp
+	if req.PivotTimestampUtc != nil {
+		pivotTime = pgtype.Timestamp{Time: req.PivotTimestampUtc.AsTime(), Valid: true}
+	} else {
+		pivotTime = pgtype.Timestamp{Time: time.Now().UTC(), Valid: true}
+	}
+
 	goprms := db.GetObservationsBetweenParams{
 		GeometryUuid: locationUuid,
 		SourceTypeID: int16(req.EnergySource),
 		ObserverUuid: observerResp.ObserverUuid,
 		StartTimeUtc: start,
 		EndTimeUtc:   end,
+		PivotTimeUtc: pivotTime,
 	}
 
 	dbObs, err := querier.GetObservationsBetween(ctx, goprms)
@@ -631,6 +639,16 @@ func (s *DataPlatformDataServiceServerImpl) CreateObservations(
 	}
 
 	// Insert the observations
+	var createdTime pgtype.Timestamp
+	if req.CreatedTimestampUtc != nil {
+		createdTime = pgtype.Timestamp{Time: req.CreatedTimestampUtc.AsTime(), Valid: true}
+	} else {
+		createdTime = pgtype.Timestamp{
+			Time:  time.Now().UTC().Truncate(time.Second),
+			Valid: true,
+		}
+	}
+
 	coprms := make([]db.CreateObservationsBatchParams, len(req.Values))
 	for i, v := range req.Values {
 		coprms[i] = db.CreateObservationsBatchParams{
@@ -640,8 +658,9 @@ func (s *DataPlatformDataServiceServerImpl) CreateObservations(
 				Time:  v.TimestampUtc.AsTime(),
 				Valid: true,
 			},
-			SourceTypeID: dbSource.SourceTypeID,
-			ValueWatts:   int64(v.ValueWatts),
+			SourceTypeID:        dbSource.SourceTypeID,
+			ValueWatts:          int64(v.ValueWatts),
+			CreatedTimestampUtc: createdTime,
 		}
 	}
 

internal/server/postgres/dataserverimpl_test.go

@@ -1043,6 +1043,8 @@ func TestGetObservationsAtTimestamp(t *testing.T) {
 	})
 	require.NoError(t, err)
 
+	expectedCreatedTime := time.Date(2025, 2, 26, 12, 5, 0, 0, time.UTC)
+
 	siteUuids := make([]string, 3)
 	for i := range siteUuids {
 		capacity := uint64(1000000 + i*100000)
@@ -1058,9 +1060,10 @@ func TestGetObservationsAtTimestamp(t *testing.T) {
 		siteUuids[i] = siteResp.LocationUuid
 
 		req := &pb.CreateObservationsRequest{
-			LocationUuid: siteResp.LocationUuid,
-			EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
-			ObserverName: observerResp.ObserverName,
+			LocationUuid:        siteResp.LocationUuid,
+			EnergySource:        pb.EnergySource_ENERGY_SOURCE_SOLAR,
+			ObserverName:        observerResp.ObserverName,
+			CreatedTimestampUtc: timestamppb.New(expectedCreatedTime),
 			Values: []*pb.CreateObservationsRequest_Value{
 				{
 					ValueWatts:   uint64(capacity / 10),
@@ -1144,6 +1147,11 @@ func TestGetObservationsAtTimestamp(t *testing.T) {
 
 				for i, obs := range resp.Values {
 					require.Equal(t, tc.expectedFractions[i], obs.ValueFraction)
+
+					require.NotNil(t, obs.CreatedTimestampUtc)
+					created := obs.CreatedTimestampUtc.AsTime()
+					require.True(t, created.Equal(expectedCreatedTime),
+						"created time %s should be equal to %s", created, expectedCreatedTime)
 				}
 			}
 		})
@@ -1385,7 +1393,7 @@ func TestGetForecastAsTimeseries(t *testing.T) {
 		"test_get_forecast_as_timeseries_site",
 		"POINT(-60.25 57.5)",
 		1000000,
-		pivotTime.Add(-time.Hour*49),
+		pivotTime.Add(-time.Hour*24*32),
 		metadata,
 		"GB",
 	)
@@ -1538,6 +1546,42 @@ func TestGetForecastAsTimeseries(t *testing.T) {
 			},
 			shouldErr: false,
 		},
+		{
+			name: "Should allow a time window of exactly 31 days",
+			req: &pb.GetForecastAsTimeseriesRequest{
+				LocationUuid: siteResp.LocationUuid,
+				Forecaster:   fc,
+				EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
+				HorizonMins:  0,
+				TimeWindow: &pb.TimeWindow{
+					StartTimestampUtc: timestamppb.New(
+						pivotTime.Add(time.Hour * 36).Add(-time.Hour * 24 * 31),
+					),
+					EndTimestampUtc: timestamppb.New(pivotTime.Add(time.Hour * 36)),
+				},
+				InitializationTimestampUtc: timestamppb.New(pivotTime.Add(-30 * time.Minute)),
+			},
+			expectedValues: []float32{
+				0.5, 0.501, 0.502, 0.503, 0.504, 0.505, 0.506, 0.507, 0.508, 0.509, 0.510, 0.511,
+			},
+			shouldErr: false,
+		},
+		{
+			name: "Should reject a time window longer than 31 days",
+			req: &pb.GetForecastAsTimeseriesRequest{
+				LocationUuid: siteResp.LocationUuid,
+				Forecaster:   fc,
+				EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
+				HorizonMins:  0,
+				TimeWindow: &pb.TimeWindow{
+					StartTimestampUtc: timestamppb.New(
+						pivotTime.Add(time.Hour * 36).Add(-time.Hour*24*31 - time.Minute),
+					),
+					EndTimestampUtc: timestamppb.New(pivotTime.Add(time.Hour * 36)),
+				},
+			},
+			shouldErr: true,
+		},
 	}
 
 	for _, tc := range testcases {
@@ -1838,28 +1882,52 @@ func TestGetObservationsAsTimeseries(t *testing.T) {
 		}
 	}
 
+	expectedCreatedTime := time.Date(2025, 2, 26, 12, 5, 0, 0, time.UTC)
+
 	_, err = dc.CreateObservations(t.Context(), &pb.CreateObservationsRequest{
-		LocationUuid: siteResp.LocationUuid,
-		EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
-		ObserverName: obsResp.ObserverName,
-		Values:       values,
+		LocationUuid:        siteResp.LocationUuid,
+		EnergySource:        pb.EnergySource_ENERGY_SOURCE_SOLAR,
+		ObserverName:        obsResp.ObserverName,
+		CreatedTimestampUtc: timestamppb.New(expectedCreatedTime),
+		Values:              values,
 	})
 	require.NoError(t, err)
 
 	testcases := []struct {
-		startTime    time.Time
-		endTime      time.Time
-		expectedSize int
+		name            string
+		startTime       time.Time
+		endTime         time.Time
+		expectedSize    int
+		expectedErrCode codes.Code
 	}{
 		{
+			name:         "Should return all values in a day-long window",
 			startTime:    pivotTime.Add(-time.Hour * 24),
 			endTime:      pivotTime,
 			expectedSize: len(values),
 		},
+		{
+			name:         "Should allow windows longer than a week",
+			startTime:    pivotTime.Add(-time.Hour * 24 * 8),
+			endTime:      pivotTime,
+			expectedSize: len(values),
+		},
+		{
+			name:         "Should allow windows of exactly 31 days",
+			startTime:    pivotTime.Add(-time.Hour * 24 * 31),
+			endTime:      pivotTime,
+			expectedSize: len(values),
+		},
+		{
+			name:            "Should reject windows longer than 31 days",
+			startTime:       pivotTime.Add(-time.Hour*24*31 - time.Minute),
+			endTime:         pivotTime,
+			expectedErrCode: codes.InvalidArgument,
+		},
 	}
 
 	for _, tc := range testcases {
-		t.Run(fmt.Sprintf("Size %d", tc.expectedSize), func(t *testing.T) {
+		t.Run(tc.name, func(t *testing.T) {
 			resp, err := dc.GetObservationsAsTimeseries(
 				t.Context(),
 				&pb.GetObservationsAsTimeseriesRequest{
@@ -1872,8 +1940,21 @@ func TestGetObservationsAsTimeseries(t *testing.T) {
 					ObserverName: obsResp.ObserverName,
 				},
 			)
+			if tc.expectedErrCode != codes.OK {
+				require.Error(t, err)
+				require.Equal(t, tc.expectedErrCode, status.Code(err))
+				return
+			}
+
 			require.NoError(t, err)
 			require.Equal(t, tc.expectedSize, len(resp.Values))
+
+			for _, v := range resp.Values {
+				require.NotNil(t, v.CreatedTimestampUtc)
+				created := v.CreatedTimestampUtc.AsTime()
+				require.True(t, created.Equal(expectedCreatedTime),
+					"created time %s should be equal to %s", created, expectedCreatedTime)
+			}
 		})
 	}
 }
@@ -1899,21 +1980,33 @@ func TestGetLatestObservations(t *testing.T) {
 	require.NoError(t, err)
 
 	// Seed some observations for the site.
-	values := []*pb.CreateObservationsRequest_Value{
-		{
-			TimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 2)),
-			ValueWatts:   uint64(0.3 * float64(siteResp.EffectiveCapacityWatts)),
-		},
-		{
-			TimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 1)),
-			ValueWatts:   uint64(0.5 * float64(siteResp.EffectiveCapacityWatts)),
+	// Create the older observation
+	_, err = dc.CreateObservations(t.Context(), &pb.CreateObservationsRequest{
+		LocationUuid: siteResp.LocationUuid,
+		EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
+		ObserverName: obsResp.ObserverName,
+		CreatedTimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 2)),
+		Values: []*pb.CreateObservationsRequest_Value{
+			{
+				TimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 2)),
+				ValueWatts:   uint64(0.3 * float64(siteResp.EffectiveCapacityWatts)),
+			},
 		},
-	}
+	})
+	require.NoError(t, err)
+
+	// Create the newer observation
 	_, err = dc.CreateObservations(t.Context(), &pb.CreateObservationsRequest{
 		LocationUuid: siteResp.LocationUuid,
 		EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR,
 		ObserverName: obsResp.ObserverName,
-		Values:       values,
+		CreatedTimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 1)),
+		Values: []*pb.CreateObservationsRequest_Value{
+			{
+				TimestampUtc: timestamppb.New(pivotTime.Add(-time.Hour * 1)),
+				ValueWatts:   uint64(0.5 * float64(siteResp.EffectiveCapacityWatts)),
+			},
+		},
 	})
 	require.NoError(t, err)
 
@@ -2127,6 +2220,16 @@ func TestCreateObservations(t *testing.T) {
 				Values:       validObservations,
 			},
 		},
+		{
+			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,
+				CreatedTimestampUtc: timestamppb.New(time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)),
+				Values:       validObservations,
+			},
+		},
 		{
 			name: "Shouldn't create invalid observations",
 			req: &pb.CreateObservationsRequest{

internal/server/postgres/mappers.go

@@ -277,6 +277,7 @@ func mapObservationAsTimeseries(
 		ValueFraction:          sipToFraction(obs.ValueSip),
 		TimestampUtc:           timestamppb.New(obs.ObservationTimestampUtc.Time),
 		EffectiveCapacityWatts: uint64(obs.CapacityWatts),
+		CreatedTimestampUtc:    timestamppb.New(obs.CreatedTimestampUtc.Time),
 	}
 }
 
@@ -335,6 +336,7 @@ func mapObservationAtTimestamp(
 			Latitude:  obs.Latitude,
 			Longitude: obs.Longitude,
 		},
+		CreatedTimestampUtc: timestamppb.New(obs.CreatedTimestampUtc.Time),
 	}
 }
 

internal/server/postgres/mappers_test.go

@@ -409,3 +409,41 @@ func Test_mapForecastAsTimeseriesFromLocationValue(t *testing.T) {
 		})
 	}
 }
+
+func Test_mapObservationAsTimeseries(t *testing.T) {
+	obsTime := time.Date(2025, 2, 26, 12, 0, 0, 0, time.UTC)
+	createdTime := time.Date(2025, 2, 26, 12, 5, 0, 0, time.UTC)
+
+	res := mapObservationAsTimeseries(db.GetObservationsBetweenRow{
+		ObservationTimestampUtc: pgtype.Timestamp{Time: obsTime, Valid: true},
+		CreatedTimestampUtc:     pgtype.Timestamp{Time: createdTime, Valid: true},
+		ValueSip:                15000,
+		CapacityWatts:           2000,
+	})
+
+	require.Equal(t, obsTime, res.TimestampUtc.AsTime())
+	require.Equal(t, createdTime, res.CreatedTimestampUtc.AsTime())
+	require.Equal(t, float32(0.5), res.ValueFraction)
+	require.Equal(t, uint64(2000), res.EffectiveCapacityWatts)
+}
+
+func Test_mapObservationAtTimestamp(t *testing.T) {
+	id := uuid.New()
+	createdTime := time.Date(2025, 2, 26, 12, 5, 0, 0, time.UTC)
+
+	res := mapObservationAtTimestamp(db.ListObservationsAtTimeForLocationsRow{
+		GeometryUuid:        id,
+		CreatedTimestampUtc: pgtype.Timestamp{Time: createdTime, Valid: true},
+		ValueSip:            30000,
+		CapacityWatts:       1000,
+		Latitude:            51.5,
+		Longitude:           -0.1,
+	})
+
+	require.Equal(t, id.String(), res.LocationUuid)
+	require.Equal(t, createdTime, res.CreatedTimestampUtc.AsTime())
+	require.Equal(t, float32(1), res.ValueFraction)
+	require.Equal(t, uint64(1000), res.EffectiveCapacityWatts)
+	require.Equal(t, float32(51.5), res.Latlng.Latitude)
+	require.Equal(t, float32(-0.1), res.Latlng.Longitude)
+}

internal/server/postgres/sql/migrations/00015_consolidated_schema.sql → internal/server/postgres/sql/migrations/00016_consolidated_schema.sql

@@ -138,9 +138,9 @@ CREATE TABLE loc.geometries (
     REFERENCES loc.entities(entity_id)
     ON UPDATE CASCADE
     ON DELETE SET NULL,
-    country_code CHAR(3),
+    country_code CHAR(2),
     CONSTRAINT country_code_format_check CHECK (
-	country_code IS NULL OR country_code ~ '^[A-Z]{3}$'
+	country_code IS NULL OR country_code ~ '^[A-Z]{2}$'
     ),
     PRIMARY KEY (geometry_uuid),
     UNIQUE (geometry_name, geom_hash)