signalsLatest/lastSeen via signal_latest projection

charts/telemetry-api/values-prod.yaml

@@ -25,7 +25,6 @@ env:
   CREDIT_TRACKER_ENDPOINT: credit-tracker-prod:8086
   IDENTITY_API_URL: http://identity-api-prod:8080/query
   IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5
-  LATEST_SIGNALS_LOOKBACK_DAYS: 45
   DIMO_REGISTRY_CHAIN_ID: 137
   STORAGE_NODE_DEV_LICENSE: '0x49eAf63eD94FEf3d40692862Eee2C8dB416B1a5f'
 ingress:

charts/telemetry-api/values.yaml

@@ -39,7 +39,6 @@ env:
   CREDIT_TRACKER_ENDPOINT: credit-tracker-dev:8086
   IDENTITY_API_URL: http://identity-api-dev:8080/query
   IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5
-  LATEST_SIGNALS_LOOKBACK_DAYS: 45
   DIMO_REGISTRY_CHAIN_ID: 80002
   STORAGE_NODE_DEV_LICENSE: '0x52fD9Dc294066792785CcD85eFB9A0Bd48DE01E4'
   VIN_DATA_VERSION: vin/v1.0

e2e/clickhouse_container_test.go

@@ -12,21 +12,26 @@ import (
 	"testing"
 	"time"
 
-	"github.com/DIMO-Network/cloudevent"
 	chconfig "github.com/DIMO-Network/clickhouse-infra/pkg/connect/config"
 	"github.com/DIMO-Network/clickhouse-infra/pkg/container"
+	"github.com/DIMO-Network/cloudevent"
 	sigmigrations "github.com/DIMO-Network/model-garage/pkg/migrations"
 	"github.com/DIMO-Network/model-garage/pkg/vss"
 	"github.com/stretchr/testify/require"
 )
 
-const loadBatchSize = 5000
+const (
+	loadBatchSize          = 5000
+	testClickHousePassword = "test-clickhouse-password"
+)
 
 func setupClickhouseContainer(t *testing.T) *container.Container {
 	t.Helper()
 	ctx := context.Background()
 
-	chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{})
+	chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{
+		Password: testClickHousePassword,
+	})
 	if err != nil {
 		t.Fatalf("Failed to create clickhouse container: %v", err)
 	}
@@ -161,7 +166,7 @@ func LoadSampleDataInto(t *testing.T, ch *container.Container, signalPath, event
 		events = append(events, vss.Event{
 			CloudEventHeader: cloudevent.CloudEventHeader{
 				Subject:  row[0],
-				Source:    row[1],
+				Source:   row[1],
 				Producer: row[2],
 				ID:       row[3],
 			},

e2e/segments_test.go

@@ -217,7 +217,9 @@ func TestSegmentDetectors(t *testing.T) {
 	ctx := context.Background()
 
 	// Setup ClickHouse container
-	chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{})
+	chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{
+		Password: testClickHousePassword,
+	})
 	require.NoError(t, err, "Failed to create ClickHouse container")
 	t.Cleanup(func() { chContainer.Terminate(ctx) })
 

go.mod

@@ -6,11 +6,11 @@ require (
 	github.com/99designs/gqlgen v0.17.89
 	github.com/ClickHouse/clickhouse-go/v2 v2.40.1
 	github.com/DIMO-Network/attestation-api v0.1.2
-	github.com/DIMO-Network/clickhouse-infra v0.0.7
-	github.com/DIMO-Network/cloudevent v1.0.4
+	github.com/DIMO-Network/clickhouse-infra v0.0.8
+	github.com/DIMO-Network/cloudevent v0.2.7
 	github.com/DIMO-Network/credit-tracker v0.0.6
 	github.com/DIMO-Network/fetch-api v0.0.16
-	github.com/DIMO-Network/model-garage v1.0.9
+	github.com/DIMO-Network/model-garage v1.0.11
 	github.com/DIMO-Network/server-garage v0.1.1
 	github.com/DIMO-Network/shared v1.1.5
 	github.com/DIMO-Network/token-exchange-api v0.4.0

go.sum

@@ -14,16 +14,16 @@ github.com/DATA-DOG/go-sqlmock v1.4.1 h1:ThlnYciV1iM/V0OSF/dtkqWb6xo5qITT1TJBG1M
 github.com/DATA-DOG/go-sqlmock v1.4.1/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/DIMO-Network/attestation-api v0.1.2 h1:mrsProeTVr+aF7rKkoJhCVA7mlGFX6A8AwFKOxO3j8U=
 github.com/DIMO-Network/attestation-api v0.1.2/go.mod h1:sJT/d3Mb+qwYk1+DYKUvM8ZEohDQnM3sAqoreP+ocBI=
-github.com/DIMO-Network/clickhouse-infra v0.0.7 h1:TAsjkFFKu3D5Xg6dwBcRBryjCVSlXsNjVbTwJ4UDlTg=
-github.com/DIMO-Network/clickhouse-infra v0.0.7/go.mod h1:XS80lhSJNWBWGgZ+m4j7++zFj1wAXfmtV2gJfhGlabQ=
-github.com/DIMO-Network/cloudevent v1.0.4 h1:bZ7tRPwCbiRgAlG1gfN7qpOOUI1OEXMMoMMyS9r4omk=
-github.com/DIMO-Network/cloudevent v1.0.4/go.mod h1:I/9NcpMozV5Fw194WimhbkAsJtKVZf5UKYJ9hgc8Cdg=
+github.com/DIMO-Network/clickhouse-infra v0.0.8 h1:54HPXKvNjmn9T0d9ZLYgUm4DnwTavE+Z8admrf39mJI=
+github.com/DIMO-Network/clickhouse-infra v0.0.8/go.mod h1:XS80lhSJNWBWGgZ+m4j7++zFj1wAXfmtV2gJfhGlabQ=
+github.com/DIMO-Network/cloudevent v0.2.7 h1:/cgFhUcWcliZYrmITkB8oIZb+zDhZvYNxWVGS2D3894=
+github.com/DIMO-Network/cloudevent v0.2.7/go.mod h1:I/9NcpMozV5Fw194WimhbkAsJtKVZf5UKYJ9hgc8Cdg=
 github.com/DIMO-Network/credit-tracker v0.0.6 h1:9C9AXah2uCQKCr7Kqzu2H9aXWj7wEOj4z1K2wAmVy0E=
 github.com/DIMO-Network/credit-tracker v0.0.6/go.mod h1:k037ZQMuBcyNj0Czg/jd6GCBC/HihkSsA7c1FRrF2yI=
 github.com/DIMO-Network/fetch-api v0.0.16 h1:jOIM9AHOCTN9Omh5QoRrdVsy0UCPHwB+08/DEI0qnqs=
 github.com/DIMO-Network/fetch-api v0.0.16/go.mod h1:pD9+5wPYjz3ORH9FBsXLlomsZJRPTyOzKy19urT6xew=
-github.com/DIMO-Network/model-garage v1.0.9 h1:MqEC/maSaImduOLEqYbIbLovgKEgzNuTn/R8wpJkARc=
-github.com/DIMO-Network/model-garage v1.0.9/go.mod h1:nvr0Ibiuu7Q7PC/hIfdZF6lPxCmq9eztQmbxUaNyus8=
+github.com/DIMO-Network/model-garage v1.0.11 h1:aLvIyeo58p9pVgz+d3DnU5k5Fxvxh6mq/jE2s3LxXoc=
+github.com/DIMO-Network/model-garage v1.0.11/go.mod h1:oi7EGKQVxFVpXRsu2H+YbizbKcx06aQg2N1Yu4GqOp8=
 github.com/DIMO-Network/server-garage v0.1.1 h1:EYmyy+Fgi2BNW0Bufn04BViDtb8BCWaN7C7BbEuoI5s=
 github.com/DIMO-Network/server-garage v0.1.1/go.mod h1:Z3A1KDUsXey+XhrPhmw/wyCidfrQvmEdWp7nShno7ZM=
 github.com/DIMO-Network/shared v1.1.5 h1:cRI3BbeYOgolMkeBOSqbDVGxymnDfeP/q7xgIXJ5MkY=

internal/config/settings.go

@@ -16,7 +16,6 @@ type Settings struct {
 	TokenExchangeIssuer       string          `yaml:"TOKEN_EXCHANGE_ISSUER_URL"`
 	VehicleNFTAddress         common.Address  `yaml:"VEHICLE_NFT_ADDRESS"`
 	MaxRequestDuration        string          `yaml:"MAX_REQUEST_DURATION"`
-	LatestSignalsLookbackDays int64           `yaml:"LATEST_SIGNALS_LOOKBACK_DAYS"`
 	ChainID                   uint64          `yaml:"DIMO_REGISTRY_CHAIN_ID"`
 	FetchAPIGRPCEndpoint      string          `yaml:"FETCH_API_GRPC_ENDPOINT"`
 	CreditTrackerEndpoint     string          `yaml:"CREDIT_TRACKER_ENDPOINT"`

internal/service/ch/ch.go

@@ -24,8 +24,7 @@ const (
 
 // Service is a ClickHouse service that interacts with the ClickHouse database.
 type Service struct {
-	conn           clickhouse.Conn
-	latestLookback time.Duration
+	conn clickhouse.Conn
 }
 
 // NewService creates a new ClickHouse service.
@@ -60,14 +59,7 @@ func NewService(settings config.Settings) (*Service, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to ping clickhouse: %w", err)
 	}
-	var latestLookback time.Duration
-	if settings.LatestSignalsLookbackDays > 0 {
-		latestLookback = time.Duration(settings.LatestSignalsLookbackDays) * 24 * time.Hour
-	}
-	return &Service{
-		conn:           conn,
-		latestLookback: latestLookback,
-	}, nil
+	return &Service{conn: conn}, nil
 }
 
 func getMaxExecutionTime(maxRequestDuration string) (int, error) {
@@ -81,13 +73,22 @@ func getMaxExecutionTime(maxRequestDuration string) (int, error) {
 	return int(maxExecutionTime.Seconds()), nil
 }
 
-// GetLatestSignals returns the latest signals based on the provided arguments from the ClickHouse database.
+// GetLatestSignals returns the latest signals based on the provided arguments
+// from the ClickHouse database. The queries are answered by the
+// signal_latest_by_subject_source_name projection, so no time bound is
+// applied; results reflect the full history of the subject.
 func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestArgs *model.LatestSignalsArgs) ([]*vss.Signal, error) {
-	lookbackFrom := s.lookbackFrom()
-	stmt, args := getLatestQuery(subject, latestArgs, lookbackFrom)
+	stmt, args := getLatestQuery(subject, latestArgs)
 	if latestArgs.IncludeLastSeen {
-		lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs, lookbackFrom)
-		stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs})
+		lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs)
+		if stmt == "" {
+			stmt, args = lastSeenStmt, lastSeenArgs
+		} else {
+			stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs})
+		}
+	}
+	if stmt == "" {
+		return nil, nil
 	}
 
 	signals, err := s.getSignals(ctx, stmt, args)
@@ -97,20 +98,10 @@ func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestAr
 	return signals, nil
 }
 
-// lookbackFrom returns the earliest timestamp the "latest" queries will
-// scan, or the zero Time if no lower bound is configured.
-func (s *Service) lookbackFrom() time.Time {
-	if s.latestLookback <= 0 {
-		return time.Time{}
-	}
-	return time.Now().Add(-s.latestLookback)
-}
-
 // GetAllLatestSignals returns the latest value for every signal stored for a subject.
 func (s *Service) GetAllLatestSignals(ctx context.Context, subject string, filter *model.SignalFilter) ([]*vss.Signal, error) {
-	lookbackFrom := s.lookbackFrom()
-	stmt, args := getAllLatestQuery(subject, filter, lookbackFrom)
-	lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &model.SignalArgs{Filter: filter}, lookbackFrom)
+	stmt, args := getAllLatestQuery(subject, filter)
+	lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &model.SignalArgs{Filter: filter})
 	stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs})
 
 	signals, err := s.getSignals(ctx, stmt, args)
@@ -322,7 +313,9 @@ type EventSummary struct {
 	LastSeen  time.Time
 }
 
-// GetEventSummaries returns per-event summaries (name, count, first/last seen) for a subject (vehicle), all time.
+// GetEventSummaries returns per-event summaries (name, count, first/last seen)
+// for a subject (vehicle), over all time. Relies on the (subject, source, name)
+// projection on the event table for acceptable performance.
 func (s *Service) GetEventSummaries(ctx context.Context, subject string) ([]*EventSummary, error) {
 	stmt, args := getEventSummariesQuery(subject)
 	rows, err := s.conn.Query(ctx, stmt, args...)

internal/service/ch/ch_test.go

@@ -9,9 +9,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/DIMO-Network/cloudevent"
 	chconfig "github.com/DIMO-Network/clickhouse-infra/pkg/connect/config"
 	"github.com/DIMO-Network/clickhouse-infra/pkg/container"
+	"github.com/DIMO-Network/cloudevent"
 	"github.com/DIMO-Network/model-garage/pkg/migrations"
 	"github.com/DIMO-Network/model-garage/pkg/vss"
 	"github.com/DIMO-Network/telemetry-api/internal/config"
@@ -20,8 +20,9 @@ import (
 )
 
 const (
-	day        = time.Hour * 24
-	dataPoints = 10
+	day                    = time.Hour * 24
+	dataPoints             = 10
+	testClickHousePassword = "test-clickhouse-password"
 
 	testSubject1   = "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:1"
 	testSubject2   = "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:2"
@@ -42,7 +43,9 @@ func TestCHService(t *testing.T) {
 func (c *CHServiceTestSuite) SetupSuite() {
 	ctx := context.Background()
 	var err error
-	c.container, err = container.CreateClickHouseContainer(ctx, chconfig.Settings{})
+	c.container, err = container.CreateClickHouseContainer(ctx, chconfig.Settings{
+		Password: testClickHousePassword,
+	})
 	c.Require().NoError(err, "Failed to create clickhouse container")
 
 	db, err := c.container.GetClickhouseAsDB()
@@ -71,9 +74,10 @@ func (c *CHServiceTestSuite) TestGetAggSignal() {
 	endTs := c.dataStartTime.Add(time.Second * time.Duration(30*dataPoints))
 	ctx := context.Background()
 	testCases := []struct {
-		name     string
-		aggArgs  model.AggregatedSignalArgs
-		expected []AggSignal
+		name                  string
+		aggArgs               model.AggregatedSignalArgs
+		expected              []AggSignal
+		allowedTopValueString []string
 	}{
 		{
 			name: "no aggs",
@@ -233,6 +237,7 @@ func (c *CHServiceTestSuite) TestGetAggSignal() {
 					ValueString: "value2",
 				},
 			},
+			allowedTopValueString: []string{"value2", "value5", "value8"},
 		},
 		{
 			name: "first float",
@@ -817,6 +822,15 @@ func (c *CHServiceTestSuite) TestGetAggSignal() {
 			})
 
 			for i, sig := range result {
+				if len(tc.allowedTopValueString) > 0 {
+					c.Require().Equal(tc.expected[i].SignalType, sig.SignalType)
+					c.Require().Equal(tc.expected[i].SignalIndex, sig.SignalIndex)
+					c.Require().Equal(tc.expected[i].Timestamp, sig.Timestamp)
+					c.Require().Equal(tc.expected[i].ValueNumber, sig.ValueNumber)
+					c.Require().Equal(tc.expected[i].ValueLocation, sig.ValueLocation)
+					c.Require().Contains(tc.allowedTopValueString, sig.ValueString)
+					continue
+				}
 				c.Require().Equal(tc.expected[i], *sig)
 			}
 		})