diff --git a/internal/app/app.go b/internal/app/app.go index d82c84f..62b6762 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -18,6 +18,7 @@ import ( "github.com/DIMO-Network/telemetry-api/internal/graph" "github.com/DIMO-Network/telemetry-api/internal/limits" "github.com/DIMO-Network/telemetry-api/internal/pricing" + "github.com/DIMO-Network/telemetry-api/internal/proxy" "github.com/DIMO-Network/telemetry-api/internal/queryRecorder" "github.com/DIMO-Network/telemetry-api/internal/repositories" "github.com/DIMO-Network/telemetry-api/internal/repositories/attestation" @@ -70,6 +71,12 @@ func New(settings config.Settings) (*App, error) { VCRepo: vcRepo, AttestationRepo: attRepo, } + if settings.DQEndpoint != "" { + resolver.ProxyClient = proxy.NewClient(settings.DQEndpoint) + resolver.ProxySubjectFunc = func(tokenID int) string { + return proxy.ToSubject(settings.ChainID, settings.VehicleNFTAddress, tokenID) + } + } cfg := graph.Config{Resolvers: resolver} cfg.Directives.RequiresVehicleToken = auth.NewVehicleTokenCheck(settings.VehicleNFTAddress) @@ -99,11 +106,13 @@ func New(settings config.Settings) (*App, error) { serverHandler := PanicRecoveryMiddleware( LoggerMiddleware( - limiter.AddRequestTimeout( - authMiddleware.CheckJWT( - authLoggerMiddleware( - dtcmiddleware.EstimateCostHeaderMiddleware( - auth.AddClaimHandler(server, settings.VehicleNFTAddress), + rawAuthMiddleware( + limiter.AddRequestTimeout( + authMiddleware.CheckJWT( + authLoggerMiddleware( + dtcmiddleware.EstimateCostHeaderMiddleware( + auth.AddClaimHandler(server, settings.VehicleNFTAddress), + ), ), ), ), @@ -125,10 +134,12 @@ func New(settings config.Settings) (*App, error) { // @requiresVehicleToken at the GraphQL resolver level. mcpAuthHandler := PanicRecoveryMiddleware( LoggerMiddleware( - limiter.AddRequestTimeout( - authMiddleware.CheckJWT( - authLoggerMiddleware( - auth.AddClaimHandler(mcpHandler, settings.VehicleNFTAddress), + rawAuthMiddleware( + limiter.AddRequestTimeout( + authMiddleware.CheckJWT( + authLoggerMiddleware( + auth.AddClaimHandler(mcpHandler, settings.VehicleNFTAddress), + ), ), ), ), diff --git a/internal/app/middleware.go b/internal/app/middleware.go index d27083c..d7ffd52 100644 --- a/internal/app/middleware.go +++ b/internal/app/middleware.go @@ -13,6 +13,7 @@ import ( "github.com/DIMO-Network/server-garage/pkg/gql/errorhandler" "github.com/DIMO-Network/telemetry-api/internal/auth" "github.com/DIMO-Network/telemetry-api/internal/limits" + "github.com/DIMO-Network/telemetry-api/internal/proxy" "github.com/rs/zerolog" "github.com/vektah/gqlparser/v2/gqlerror" ) @@ -94,6 +95,17 @@ func authLoggerMiddleware(next http.Handler) http.Handler { }) } +// rawAuthMiddleware captures the raw Authorization header and stores it in context +// so the proxy client can forward it to dq without re-signing. +func rawAuthMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if v := r.Header.Get("Authorization"); v != "" { + r = r.WithContext(context.WithValue(r.Context(), proxy.AuthHeaderKey{}, v)) + } + next.ServeHTTP(w, r) + }) +} + // PanicRecoveryMiddleware recovers from panics and logs them. func PanicRecoveryMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/config/settings.go b/internal/config/settings.go index d4c8f87..e84dfda 100644 --- a/internal/config/settings.go +++ b/internal/config/settings.go @@ -22,4 +22,8 @@ type Settings struct { CreditTrackerEndpoint string `yaml:"CREDIT_TRACKER_ENDPOINT"` StorageNodeDevLicense common.Address `yaml:"STORAGE_NODE_DEV_LICENSE"` VINDataVersion string `yaml:"VIN_DATA_VERSION"` + + // DQEndpoint is the URL of the dq GraphQL endpoint (e.g. http://dq:3000/query). + // When set, signal/event/segment queries are proxied to dq instead of ClickHouse. + DQEndpoint string `yaml:"DQ_ENDPOINT"` } diff --git a/internal/graph/base.resolvers.go b/internal/graph/base.resolvers.go index 422038f..666d976 100644 --- a/internal/graph/base.resolvers.go +++ b/internal/graph/base.resolvers.go @@ -21,6 +21,9 @@ func (r *queryResolver) Signals(ctx context.Context, tokenID int, interval strin if err != nil { return nil, err } + if r.ProxyClient != nil { + return r.ProxyClient.ProxySignals(ctx, r.ProxySubjectFunc(tokenID), interval, from, to, filter, aggArgs) + } return r.BaseRepo.GetSignal(ctx, aggArgs) } @@ -30,11 +33,22 @@ func (r *queryResolver) SignalsLatest(ctx context.Context, tokenID int, filter * if err != nil { return nil, err } + if r.ProxyClient != nil { + coll, err := r.ProxyClient.ProxySignalsLatest(ctx, r.ProxySubjectFunc(tokenID), filter, latestArgs) + if err != nil { + return nil, err + } + repositories.SetApproximateLocationInCollection(coll) + return coll, nil + } return r.BaseRepo.GetSignalLatest(ctx, latestArgs) } // AvailableSignals is the resolver for the AvailableSignals field. func (r *queryResolver) AvailableSignals(ctx context.Context, tokenID int, filter *model.SignalFilter) ([]string, error) { + if r.ProxyClient != nil { + return r.ProxyClient.ProxyAvailableSignals(ctx, r.ProxySubjectFunc(tokenID), filter) + } return r.BaseRepo.GetAvailableSignals(ctx, uint32(tokenID), filter) } @@ -66,6 +80,9 @@ func (r *queryResolver) SignalsSnapshot(ctx context.Context, tokenID int, filter // DataSummary is the resolver for the dataSummary field. func (r *queryResolver) DataSummary(ctx context.Context, tokenID int, filter *model.SignalFilter) (*model.DataSummary, error) { + if r.ProxyClient != nil { + return r.ProxyClient.ProxyDataSummary(ctx, r.ProxySubjectFunc(tokenID), filter) + } return r.BaseRepo.GetDataSummary(ctx, uint32(tokenID), filter) } diff --git a/internal/graph/events.resolvers.go b/internal/graph/events.resolvers.go index 11492df..7753310 100644 --- a/internal/graph/events.resolvers.go +++ b/internal/graph/events.resolvers.go @@ -14,5 +14,8 @@ import ( // Events is the resolver for the events field. func (r *queryResolver) Events(ctx context.Context, tokenID int, from time.Time, to time.Time, filter *model.EventFilter) ([]*model.Event, error) { + if r.ProxyClient != nil { + return r.ProxyClient.ProxyEvents(ctx, r.ProxySubjectFunc(tokenID), from, to, filter) + } return r.BaseRepo.GetEvents(ctx, tokenID, from, to, filter) } diff --git a/internal/graph/resolver.go b/internal/graph/resolver.go index a0d6781..a498335 100644 --- a/internal/graph/resolver.go +++ b/internal/graph/resolver.go @@ -3,6 +3,7 @@ package graph //go:generate go run github.com/DIMO-Network/server-garage/cmd/mcpgen -schema ../../schema/ -prefix telemetry -out mcp_tools_gen.go -package graph import ( + "github.com/DIMO-Network/telemetry-api/internal/proxy" "github.com/DIMO-Network/telemetry-api/internal/repositories" "github.com/DIMO-Network/telemetry-api/internal/repositories/attestation" "github.com/DIMO-Network/telemetry-api/internal/repositories/vc" @@ -16,4 +17,8 @@ type Resolver struct { BaseRepo *repositories.Repository VCRepo *vc.Repository AttestationRepo *attestation.Repository + // ProxyClient, when non-nil, forwards signal/event/segment queries to dq instead of ClickHouse. + ProxyClient *proxy.Client + // ProxySubjectFunc converts a tokenID to a DID subject for the proxy. + ProxySubjectFunc func(tokenID int) string } diff --git a/internal/graph/segments.resolvers.go b/internal/graph/segments.resolvers.go index a15d37f..3bcf55d 100644 --- a/internal/graph/segments.resolvers.go +++ b/internal/graph/segments.resolvers.go @@ -14,10 +14,16 @@ import ( // Segments is the resolver for the segments field. func (r *queryResolver) Segments(ctx context.Context, tokenID int, from time.Time, to time.Time, mechanism model.DetectionMechanism, config *model.SegmentConfig, signalRequests []*model.SegmentSignalRequest, eventRequests []*model.SegmentEventRequest, limit *int, after *time.Time) ([]*model.Segment, error) { + if r.ProxyClient != nil { + return r.ProxyClient.ProxySegments(ctx, r.ProxySubjectFunc(tokenID), from, to, mechanism, config, signalRequests, eventRequests, limit, after) + } return r.BaseRepo.GetSegments(ctx, tokenID, from, to, mechanism, config, signalRequests, eventRequests, limit, after) } // DailyActivity is the resolver for the dailyActivity field. func (r *queryResolver) DailyActivity(ctx context.Context, tokenID int, from time.Time, to time.Time, mechanism model.DetectionMechanism, config *model.SegmentConfig, signalRequests []*model.SegmentSignalRequest, eventRequests []*model.SegmentEventRequest, timezone *string) ([]*model.DailyActivity, error) { + if r.ProxyClient != nil { + return r.ProxyClient.ProxyDailyActivity(ctx, r.ProxySubjectFunc(tokenID), from, to, mechanism, config, signalRequests, eventRequests, timezone) + } return r.BaseRepo.GetDailyActivity(ctx, tokenID, from, to, mechanism, config, signalRequests, eventRequests, timezone) } diff --git a/internal/proxy/client.go b/internal/proxy/client.go new file mode 100644 index 0000000..a5e1e96 --- /dev/null +++ b/internal/proxy/client.go @@ -0,0 +1,70 @@ +package proxy + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" +) + +// AuthHeaderKey is the context key for the raw Authorization header value. +type AuthHeaderKey struct{} + +type gqlRequest struct { + Query string `json:"query"` + Variables map[string]any `json:"variables,omitempty"` +} + +type gqlError struct { + Message string `json:"message"` +} + +type gqlResponse struct { + Data json.RawMessage `json:"data"` + Errors []gqlError `json:"errors"` +} + +// Client is a thin GraphQL HTTP client for forwarding requests to the dq service. +type Client struct { + endpoint string + http *http.Client +} + +// NewClient creates a new proxy client targeting the given GraphQL endpoint URL. +func NewClient(endpoint string) *Client { + return &Client{endpoint: endpoint, http: &http.Client{}} +} + +// Execute posts a GraphQL query to dq, forwarding the Authorization header from ctx. +// It returns the raw JSON "data" object on success. +func (c *Client) Execute(ctx context.Context, query string, variables map[string]any) (json.RawMessage, error) { + body, err := json.Marshal(gqlRequest{Query: query, Variables: variables}) + if err != nil { + return nil, fmt.Errorf("marshaling request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + if auth, _ := ctx.Value(AuthHeaderKey{}).(string); auth != "" { + req.Header.Set("Authorization", auth) + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("executing request: %w", err) + } + defer resp.Body.Close() //nolint:errcheck + + var gqlResp gqlResponse + if err := json.NewDecoder(resp.Body).Decode(&gqlResp); err != nil { + return nil, fmt.Errorf("decoding response: %w", err) + } + if len(gqlResp.Errors) > 0 { + return nil, fmt.Errorf("dq error: %s", gqlResp.Errors[0].Message) + } + return gqlResp.Data, nil +} diff --git a/internal/proxy/client_test.go b/internal/proxy/client_test.go new file mode 100644 index 0000000..b8d5c31 --- /dev/null +++ b/internal/proxy/client_test.go @@ -0,0 +1,155 @@ +package proxy + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/DIMO-Network/telemetry-api/internal/graph/model" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestClientExecute_ForwardsAuthHeader(t *testing.T) { + var gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{}}`) + })) + defer srv.Close() + + ctx := context.WithValue(context.Background(), AuthHeaderKey{}, "Bearer test-token") + client := NewClient(srv.URL) + _, err := client.Execute(ctx, `{__typename}`, nil) + require.NoError(t, err) + assert.Equal(t, "Bearer test-token", gotAuth) +} + +func TestClientExecute_NoAuthHeader(t *testing.T) { + var gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{}}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + _, err := client.Execute(context.Background(), `{__typename}`, nil) + require.NoError(t, err) + assert.Empty(t, gotAuth) +} + +func TestClientExecute_SendsVariables(t *testing.T) { + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotBody)) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{"foo":1}}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + vars := map[string]any{"subject": "did:erc721:137:0xabc:1"} + _, err := client.Execute(context.Background(), `query($subject: String!) { foo }`, vars) + require.NoError(t, err) + + assert.Equal(t, `query($subject: String!) { foo }`, gotBody["query"]) + assert.Equal(t, "did:erc721:137:0xabc:1", gotBody["variables"].(map[string]any)["subject"]) +} + +func TestClientExecute_GraphQLErrors(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"errors":[{"message":"unauthorized"}]}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + _, err := client.Execute(context.Background(), `{__typename}`, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "unauthorized") +} + +func TestClientExecute_ReturnsDataJSON(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{"speed":42.0}}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + data, err := client.Execute(context.Background(), `{speed}`, nil) + require.NoError(t, err) + + var got map[string]float64 + require.NoError(t, json.Unmarshal(data, &got)) + assert.Equal(t, 42.0, got["speed"]) +} + +func TestProxySignals_RoundTrip(t *testing.T) { + ts := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Verify the request contains the right subject and includes timestamp + field selection. + var body map[string]any + require.NoError(t, json.NewDecoder(r.Body).Decode(&body)) + vars := body["variables"].(map[string]any) + assert.Equal(t, "did:erc721:137:0xabc:1", vars["subject"]) + assert.Equal(t, "1h", vars["interval"]) + + query := body["query"].(string) + assert.Contains(t, query, `mySpeed: speed(agg: LAST)`) + + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{"signals":[{"timestamp":"2024-01-01T12:00:00Z","mySpeed":77.5}]}}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{ + {Name: "speed", Agg: model.FloatAggregationLast, Alias: "mySpeed"}, + }, + } + + result, err := client.ProxySignals( + context.Background(), + "did:erc721:137:0xabc:1", "1h", + ts, ts.Add(time.Hour), nil, + aggArgs, + ) + require.NoError(t, err) + require.Len(t, result, 1) + assert.Equal(t, ts, result[0].Timestamp) + assert.Equal(t, 77.5, result[0].ValueNumbers["mySpeed"]) +} + +func TestProxySegments_MechanismMapped(t *testing.T) { + var gotMechanism string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + require.NoError(t, json.NewDecoder(r.Body).Decode(&body)) + gotMechanism = body["variables"].(map[string]any)["mechanism"].(string) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":{"segments":[]}}`) + })) + defer srv.Close() + + client := NewClient(srv.URL) + now := time.Now() + _, err := client.ProxySegments( + context.Background(), + "did:erc721:137:0xabc:1", + now, now.Add(time.Hour), + model.DetectionMechanismIgnitionDetection, + nil, nil, nil, nil, nil, + ) + require.NoError(t, err) + assert.Equal(t, "IGNITION_DETECTION", gotMechanism) +} diff --git a/internal/proxy/transform.go b/internal/proxy/transform.go new file mode 100644 index 0000000..6131e60 --- /dev/null +++ b/internal/proxy/transform.go @@ -0,0 +1,421 @@ +package proxy + +import ( + "context" + "encoding/json" + "fmt" + "math/big" + "strings" + "time" + + "github.com/DIMO-Network/cloudevent" + "github.com/DIMO-Network/model-garage/pkg/vss" + "github.com/DIMO-Network/telemetry-api/internal/graph/model" + "github.com/ethereum/go-ethereum/common" +) + +// ToSubject converts a vehicle token ID into a DID subject string for dq. +func ToSubject(chainID uint64, vehicleAddr common.Address, tokenID int) string { + return cloudevent.ERC721DID{ + ChainID: chainID, + ContractAddress: vehicleAddr, + TokenID: big.NewInt(int64(tokenID)), + }.String() +} + +// dqMechanism maps telemetry-api's camelCase DetectionMechanism values to dq's UPPER_SNAKE_CASE. +var dqMechanism = map[model.DetectionMechanism]string{ + model.DetectionMechanismIgnitionDetection: "IGNITION_DETECTION", + model.DetectionMechanismFrequencyAnalysis: "FREQUENCY_ANALYSIS", + model.DetectionMechanismChangePointDetection: "CHANGE_POINT_DETECTION", + model.DetectionMechanismIdling: "IDLING", + model.DetectionMechanismRefuel: "REFUEL", + model.DetectionMechanismRecharge: "RECHARGE", +} + +// ToDQMechanism returns the dq-format DetectionMechanism string for the given telemetry-api value. +func ToDQMechanism(m model.DetectionMechanism) (string, error) { + v, ok := dqMechanism[m] + if !ok { + return "", fmt.Errorf("unknown detection mechanism: %s", m) + } + return v, nil +} + +// BuildSignalsQuery constructs the dq GraphQL query for the signals resolver. +// Field selections are built from aggArgs to preserve aliases. +// It returns the query string and any per-signal filter variables that must be included +// in the variables map alongside subject/interval/from/to/filter. +func BuildSignalsQuery(aggArgs *model.AggregatedSignalArgs) (query string, filterVars map[string]any) { + filterVars = make(map[string]any) + + var varDecls strings.Builder // extra variable declarations + var fields strings.Builder // field selection set + + fields.WriteString(`timestamp`) + + for i, fa := range aggArgs.FloatArgs { + if fa.Filter != nil { + varName := fmt.Sprintf("ff%d", i) + fmt.Fprintf(&varDecls, `, $%s: SignalFloatFilter`, varName) + fmt.Fprintf(&fields, ` %s: %s(agg: %s, filter: $%s)`, fa.Alias, fa.Name, fa.Agg, varName) + filterVars[varName] = fa.Filter + } else { + fmt.Fprintf(&fields, ` %s: %s(agg: %s)`, fa.Alias, fa.Name, fa.Agg) + } + } + for _, sa := range aggArgs.StringArgs { + fmt.Fprintf(&fields, ` %s: %s(agg: %s)`, sa.Alias, sa.Name, sa.Agg) + } + for i, la := range aggArgs.LocationArgs { + if la.Filter != nil { + varName := fmt.Sprintf("lf%d", i) + fmt.Fprintf(&varDecls, `, $%s: SignalLocationFilter`, varName) + fmt.Fprintf(&fields, ` %s: %s(agg: %s, filter: $%s) { latitude longitude hdop }`, la.Alias, la.Name, la.Agg, varName) + filterVars[varName] = la.Filter + } else { + fmt.Fprintf(&fields, ` %s: %s(agg: %s) { latitude longitude hdop }`, la.Alias, la.Name, la.Agg) + } + } + + query = fmt.Sprintf( + `query Signals($subject: String!, $interval: String!, $from: Time!, $to: Time!, $filter: SignalFilter%s) {signals(subject: $subject, interval: $interval, from: $from, to: $to, filter: $filter) {%s}}`, + varDecls.String(), fields.String(), + ) + return query, filterVars +} + +// UnmarshalSignalsResponse parses the dq response JSON for the signals query into +// []*model.SignalAggregations, preserving the alias→value mapping needed by the +// generated field resolvers. +func UnmarshalSignalsResponse(data json.RawMessage, aggArgs *model.AggregatedSignalArgs) ([]*model.SignalAggregations, error) { + var envelope struct { + Signals []map[string]json.RawMessage `json:"signals"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling signals: %w", err) + } + + result := make([]*model.SignalAggregations, 0, len(envelope.Signals)) + for _, row := range envelope.Signals { + agg := &model.SignalAggregations{ + ValueNumbers: make(map[string]float64), + ValueStrings: make(map[string]string), + ValueLocations: make(map[string]vss.Location), + } + if raw, ok := row["timestamp"]; ok { + if err := json.Unmarshal(raw, &agg.Timestamp); err != nil { + return nil, fmt.Errorf("unmarshaling timestamp: %w", err) + } + } + for _, fa := range aggArgs.FloatArgs { + if raw, ok := row[fa.Alias]; ok && string(raw) != "null" { + var v float64 + if err := json.Unmarshal(raw, &v); err != nil { + return nil, fmt.Errorf("unmarshaling float %s: %w", fa.Alias, err) + } + agg.ValueNumbers[fa.Alias] = v + } + } + for _, sa := range aggArgs.StringArgs { + if raw, ok := row[sa.Alias]; ok && string(raw) != "null" { + var v string + if err := json.Unmarshal(raw, &v); err != nil { + return nil, fmt.Errorf("unmarshaling string %s: %w", sa.Alias, err) + } + agg.ValueStrings[sa.Alias] = v + } + } + for _, la := range aggArgs.LocationArgs { + if raw, ok := row[la.Alias]; ok && string(raw) != "null" { + var loc struct { + Latitude float64 `json:"latitude"` + Longitude float64 `json:"longitude"` + Hdop float64 `json:"hdop"` + } + if err := json.Unmarshal(raw, &loc); err != nil { + return nil, fmt.Errorf("unmarshaling location %s: %w", la.Alias, err) + } + agg.ValueLocations[la.Alias] = vss.Location{ + Latitude: loc.Latitude, + Longitude: loc.Longitude, + HDOP: loc.Hdop, + } + } + } + result = append(result, agg) + } + return result, nil +} + +// BuildSignalsLatestQuery constructs the dq GraphQL query for the signalsLatest resolver. +func BuildSignalsLatestQuery(latestArgs *model.LatestSignalsArgs) string { + var sb strings.Builder + sb.WriteString(`query SignalsLatest($subject: String!, $filter: SignalFilter) {`) + sb.WriteString(`signalsLatest(subject: $subject, filter: $filter) {`) + if latestArgs.IncludeLastSeen { + sb.WriteString(`lastSeen `) + } + for name := range latestArgs.SignalNames { + fmt.Fprintf(&sb, `%s { timestamp value } `, name) + } + for name := range latestArgs.LocationSignalNames { + fmt.Fprintf(&sb, `%s { timestamp value { latitude longitude hdop } } `, name) + } + sb.WriteString(`} }`) + return sb.String() +} + +// UnmarshalSignalsLatestResponse parses the dq response into *model.SignalCollection. +func UnmarshalSignalsLatestResponse(data json.RawMessage) (*model.SignalCollection, error) { + var envelope struct { + SignalsLatest *model.SignalCollection `json:"signalsLatest"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling signalsLatest: %w", err) + } + return envelope.SignalsLatest, nil +} + +// BuildAvailableSignalsQuery constructs the dq query for availableSignals. +func BuildAvailableSignalsQuery() string { + return `query AvailableSignals($subject: String!, $filter: SignalFilter) { + availableSignals(subject: $subject, filter: $filter) + }` +} + +// UnmarshalAvailableSignalsResponse parses the dq response into []string. +func UnmarshalAvailableSignalsResponse(data json.RawMessage) ([]string, error) { + var envelope struct { + AvailableSignals []string `json:"availableSignals"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling availableSignals: %w", err) + } + return envelope.AvailableSignals, nil +} + +// BuildDataSummaryQuery constructs the dq query for dataSummary. +func BuildDataSummaryQuery() string { + return `query DataSummary($subject: String!, $filter: SignalFilter) { + dataSummary(subject: $subject, filter: $filter) { + numberOfSignals availableSignals firstSeen lastSeen + signalDataSummary { name numberOfSignals firstSeen lastSeen } + eventDataSummary { name numberOfEvents firstSeen lastSeen } + } + }` +} + +// UnmarshalDataSummaryResponse parses the dq response into *model.DataSummary. +func UnmarshalDataSummaryResponse(data json.RawMessage) (*model.DataSummary, error) { + var envelope struct { + DataSummary *model.DataSummary `json:"dataSummary"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling dataSummary: %w", err) + } + return envelope.DataSummary, nil +} + +// BuildEventsQuery constructs the dq query for events. +func BuildEventsQuery() string { + return `query Events($subject: String!, $from: Time!, $to: Time!, $filter: EventFilter) { + events(subject: $subject, from: $from, to: $to, filter: $filter) { + timestamp name source durationNs metadata + } + }` +} + +// UnmarshalEventsResponse parses the dq response into []*model.Event. +func UnmarshalEventsResponse(data json.RawMessage) ([]*model.Event, error) { + var envelope struct { + Events []*model.Event `json:"events"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling events: %w", err) + } + return envelope.Events, nil +} + +// BuildSegmentsQuery constructs the dq query for segments. +func BuildSegmentsQuery() string { + return `query Segments($subject: String!, $from: Time!, $to: Time!, $mechanism: DetectionMechanism!, + $config: SegmentConfig, $signalRequests: [SegmentSignalRequest!], $eventRequests: [SegmentEventRequest!], + $limit: Int, $after: Time) { + segments(subject: $subject, from: $from, to: $to, mechanism: $mechanism, + config: $config, signalRequests: $signalRequests, eventRequests: $eventRequests, + limit: $limit, after: $after) { + start { timestamp value { latitude longitude hdop } } + end { timestamp value { latitude longitude hdop } } + duration isOngoing startedBeforeRange + signals { name agg value } + eventCounts { name count } + } + }` +} + +// UnmarshalSegmentsResponse parses the dq response into []*model.Segment. +func UnmarshalSegmentsResponse(data json.RawMessage) ([]*model.Segment, error) { + var envelope struct { + Segments []*model.Segment `json:"segments"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling segments: %w", err) + } + return envelope.Segments, nil +} + +// BuildDailyActivityQuery constructs the dq query for dailyActivity. +func BuildDailyActivityQuery() string { + return `query DailyActivity($subject: String!, $from: Time!, $to: Time!, $mechanism: DetectionMechanism!, + $config: SegmentConfig, $signalRequests: [SegmentSignalRequest!], $eventRequests: [SegmentEventRequest!], + $timezone: String) { + dailyActivity(subject: $subject, from: $from, to: $to, mechanism: $mechanism, + config: $config, signalRequests: $signalRequests, eventRequests: $eventRequests, + timezone: $timezone) { + start { timestamp value { latitude longitude hdop } } + end { timestamp value { latitude longitude hdop } } + segmentCount duration + signals { name agg value } + eventCounts { name count } + } + }` +} + +// UnmarshalDailyActivityResponse parses the dq response into []*model.DailyActivity. +func UnmarshalDailyActivityResponse(data json.RawMessage) ([]*model.DailyActivity, error) { + var envelope struct { + DailyActivity []*model.DailyActivity `json:"dailyActivity"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling dailyActivity: %w", err) + } + return envelope.DailyActivity, nil +} + + +// TimeVar formats a time.Time for use as a GraphQL Time variable. +func TimeVar(t time.Time) string { + return t.UTC().Format(time.RFC3339Nano) +} + +// ProxySignals executes the dq signals query and returns telemetry-api model results. +func (c *Client) ProxySignals(ctx context.Context, subject, interval string, from, to time.Time, filter *model.SignalFilter, aggArgs *model.AggregatedSignalArgs) ([]*model.SignalAggregations, error) { + query, filterVars := BuildSignalsQuery(aggArgs) + vars := map[string]any{ + "subject": subject, + "interval": interval, + "from": TimeVar(from), + "to": TimeVar(to), + "filter": filter, + } + for k, v := range filterVars { + vars[k] = v + } + data, err := c.Execute(ctx, query, vars) + if err != nil { + return nil, err + } + return UnmarshalSignalsResponse(data, aggArgs) +} + +// ProxySignalsLatest executes the dq signalsLatest query and returns the result. +func (c *Client) ProxySignalsLatest(ctx context.Context, subject string, filter *model.SignalFilter, latestArgs *model.LatestSignalsArgs) (*model.SignalCollection, error) { + query := BuildSignalsLatestQuery(latestArgs) + vars := map[string]any{ + "subject": subject, + "filter": filter, + } + data, err := c.Execute(ctx, query, vars) + if err != nil { + return nil, err + } + return UnmarshalSignalsLatestResponse(data) +} + +// ProxyAvailableSignals executes the dq availableSignals query and returns the result. +func (c *Client) ProxyAvailableSignals(ctx context.Context, subject string, filter *model.SignalFilter) ([]string, error) { + data, err := c.Execute(ctx, BuildAvailableSignalsQuery(), map[string]any{ + "subject": subject, + "filter": filter, + }) + if err != nil { + return nil, err + } + return UnmarshalAvailableSignalsResponse(data) +} + +// ProxyDataSummary executes the dq dataSummary query and returns the result. +func (c *Client) ProxyDataSummary(ctx context.Context, subject string, filter *model.SignalFilter) (*model.DataSummary, error) { + data, err := c.Execute(ctx, BuildDataSummaryQuery(), map[string]any{ + "subject": subject, + "filter": filter, + }) + if err != nil { + return nil, err + } + return UnmarshalDataSummaryResponse(data) +} + +// ProxyEvents executes the dq events query and returns the result. +func (c *Client) ProxyEvents(ctx context.Context, subject string, from, to time.Time, filter *model.EventFilter) ([]*model.Event, error) { + data, err := c.Execute(ctx, BuildEventsQuery(), map[string]any{ + "subject": subject, + "from": TimeVar(from), + "to": TimeVar(to), + "filter": filter, + }) + if err != nil { + return nil, err + } + return UnmarshalEventsResponse(data) +} + +// ProxySegments executes the dq segments query and returns the result. +func (c *Client) ProxySegments(ctx context.Context, subject string, from, to time.Time, mechanism model.DetectionMechanism, config *model.SegmentConfig, signalRequests []*model.SegmentSignalRequest, eventRequests []*model.SegmentEventRequest, limit *int, after *time.Time) ([]*model.Segment, error) { + mech, err := ToDQMechanism(mechanism) + if err != nil { + return nil, err + } + vars := map[string]any{ + "subject": subject, + "from": TimeVar(from), + "to": TimeVar(to), + "mechanism": mech, + "config": config, + "signalRequests": signalRequests, + "eventRequests": eventRequests, + "limit": limit, + "after": nil, + } + if after != nil { + vars["after"] = TimeVar(*after) + } + data, err := c.Execute(ctx, BuildSegmentsQuery(), vars) + if err != nil { + return nil, err + } + return UnmarshalSegmentsResponse(data) +} + +// ProxyDailyActivity executes the dq dailyActivity query and returns the result. +func (c *Client) ProxyDailyActivity(ctx context.Context, subject string, from, to time.Time, mechanism model.DetectionMechanism, config *model.SegmentConfig, signalRequests []*model.SegmentSignalRequest, eventRequests []*model.SegmentEventRequest, timezone *string) ([]*model.DailyActivity, error) { + mech, err := ToDQMechanism(mechanism) + if err != nil { + return nil, err + } + data, err := c.Execute(ctx, BuildDailyActivityQuery(), map[string]any{ + "subject": subject, + "from": TimeVar(from), + "to": TimeVar(to), + "mechanism": mech, + "config": config, + "signalRequests": signalRequests, + "eventRequests": eventRequests, + "timezone": timezone, + }) + if err != nil { + return nil, err + } + return UnmarshalDailyActivityResponse(data) +} diff --git a/internal/proxy/transform_test.go b/internal/proxy/transform_test.go new file mode 100644 index 0000000..34c1a76 --- /dev/null +++ b/internal/proxy/transform_test.go @@ -0,0 +1,239 @@ +package proxy + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/DIMO-Network/model-garage/pkg/vss" + "github.com/DIMO-Network/telemetry-api/internal/graph/model" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestToSubject(t *testing.T) { + got := ToSubject(137, common.HexToAddress("0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF"), 42) + assert.Equal(t, "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:42", got) +} + +func TestToDQMechanism(t *testing.T) { + cases := []struct { + in model.DetectionMechanism + want string + }{ + {model.DetectionMechanismIgnitionDetection, "IGNITION_DETECTION"}, + {model.DetectionMechanismFrequencyAnalysis, "FREQUENCY_ANALYSIS"}, + {model.DetectionMechanismChangePointDetection, "CHANGE_POINT_DETECTION"}, + {model.DetectionMechanismIdling, "IDLING"}, + {model.DetectionMechanismRefuel, "REFUEL"}, + {model.DetectionMechanismRecharge, "RECHARGE"}, + } + for _, tc := range cases { + got, err := ToDQMechanism(tc.in) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + } + + _, err := ToDQMechanism("bogus") + require.Error(t, err) +} + +func TestBuildSignalsQuery_NoFilters(t *testing.T) { + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{ + {Name: "speed", Agg: model.FloatAggregationLast, Alias: "speed"}, + }, + StringArgs: []model.StringSignalArgs{ + {Name: "obdFuelTypeName", Agg: model.StringAggregationLast, Alias: "obdFuelTypeName"}, + }, + LocationArgs: []model.LocationSignalArgs{ + {Name: "currentLocationCoordinates", Agg: model.LocationAggregationLast, Alias: "currentLocationCoordinates"}, + }, + } + + query, filterVars := BuildSignalsQuery(aggArgs) + + assert.Empty(t, filterVars) + assert.Contains(t, query, `speed: speed(agg: LAST)`) + assert.Contains(t, query, `obdFuelTypeName: obdFuelTypeName(agg: LAST)`) + assert.Contains(t, query, `currentLocationCoordinates: currentLocationCoordinates(agg: LAST) { latitude longitude hdop }`) + assert.Contains(t, query, `timestamp`) + // No extra variable declarations beyond the base ones. + assert.NotContains(t, query, `SignalFloatFilter`) + assert.NotContains(t, query, `SignalLocationFilter`) +} + +func TestBuildSignalsQuery_AliasPreserved(t *testing.T) { + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{ + {Name: "speed", Agg: model.FloatAggregationLast, Alias: "mySpeed"}, + }, + } + + query, _ := BuildSignalsQuery(aggArgs) + + assert.Contains(t, query, `mySpeed: speed(agg: LAST)`) + assert.NotContains(t, query, `speed: speed`) +} + +func TestBuildSignalsQuery_WithFloatFilter(t *testing.T) { + gt := 50.0 + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{ + {Name: "speed", Agg: model.FloatAggregationAvg, Alias: "speed", Filter: &model.SignalFloatFilter{Gt: >}}, + {Name: "rpm", Agg: model.FloatAggregationAvg, Alias: "rpm"}, + }, + } + + query, filterVars := BuildSignalsQuery(aggArgs) + + require.Len(t, filterVars, 1) + assert.Contains(t, filterVars, "ff0") + assert.Contains(t, query, `$ff0: SignalFloatFilter`) + assert.Contains(t, query, `speed: speed(agg: AVG, filter: $ff0)`) + assert.Contains(t, query, `rpm: rpm(agg: AVG)`) + assert.NotContains(t, query, `$ff1`) // second arg has no filter +} + +func TestBuildSignalsQuery_ApproxCoordinatesAlias(t *testing.T) { + // Approximate location: Name = raw coords field, Alias = approximate field name. + aggArgs := &model.AggregatedSignalArgs{ + LocationArgs: []model.LocationSignalArgs{ + { + Name: vss.FieldCurrentLocationCoordinates, + Agg: model.LocationAggregationLast, + Alias: model.ApproximateCoordinatesField, + }, + }, + } + + query, _ := BuildSignalsQuery(aggArgs) + + // Alias is the approximate field, Name is the raw coordinates field. + assert.Contains(t, query, model.ApproximateCoordinatesField+`: `+vss.FieldCurrentLocationCoordinates+`(agg: LAST) { latitude longitude hdop }`) +} + +func TestUnmarshalSignalsResponse_Basic(t *testing.T) { + ts := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + data := json.RawMessage(`{"signals":[{"timestamp":"2024-01-01T00:00:00Z","speed":55.5,"gear":"D","currentLocationCoordinates":{"latitude":37.0,"longitude":-122.0,"hdop":1.5}}]}`) + + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{{Name: "speed", Alias: "speed"}}, + StringArgs: []model.StringSignalArgs{{Name: "gear", Alias: "gear"}}, + LocationArgs: []model.LocationSignalArgs{ + {Name: "currentLocationCoordinates", Alias: "currentLocationCoordinates"}, + }, + } + + result, err := UnmarshalSignalsResponse(data, aggArgs) + require.NoError(t, err) + require.Len(t, result, 1) + + row := result[0] + assert.Equal(t, ts, row.Timestamp) + assert.Equal(t, 55.5, row.ValueNumbers["speed"]) + assert.Equal(t, "D", row.ValueStrings["gear"]) + loc := row.ValueLocations["currentLocationCoordinates"] + assert.InDelta(t, 37.0, loc.Latitude, 1e-9) + assert.InDelta(t, -122.0, loc.Longitude, 1e-9) + assert.InDelta(t, 1.5, loc.HDOP, 1e-9) +} + +func TestUnmarshalSignalsResponse_AliasMapping(t *testing.T) { + data := json.RawMessage(`{"signals":[{"timestamp":"2024-01-01T00:00:00Z","mySpeed":88.0}]}`) + + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{{Name: "speed", Alias: "mySpeed"}}, + } + + result, err := UnmarshalSignalsResponse(data, aggArgs) + require.NoError(t, err) + require.Len(t, result, 1) + + assert.Equal(t, 88.0, result[0].ValueNumbers["mySpeed"]) + assert.Empty(t, result[0].ValueNumbers["speed"]) // original name not used as key +} + +func TestUnmarshalSignalsResponse_NullFieldSkipped(t *testing.T) { + data := json.RawMessage(`{"signals":[{"timestamp":"2024-01-01T00:00:00Z","speed":null}]}`) + + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{{Name: "speed", Alias: "speed"}}, + } + + result, err := UnmarshalSignalsResponse(data, aggArgs) + require.NoError(t, err) + require.Len(t, result, 1) + assert.Empty(t, result[0].ValueNumbers) // null → not added to map +} + +func TestUnmarshalSignalsResponse_MultipleRows(t *testing.T) { + data := json.RawMessage(`{"signals":[ + {"timestamp":"2024-01-01T00:00:00Z","speed":10.0}, + {"timestamp":"2024-01-01T01:00:00Z","speed":20.0} + ]}`) + + aggArgs := &model.AggregatedSignalArgs{ + FloatArgs: []model.FloatSignalArgs{{Name: "speed", Alias: "speed"}}, + } + + result, err := UnmarshalSignalsResponse(data, aggArgs) + require.NoError(t, err) + require.Len(t, result, 2) + assert.Equal(t, 10.0, result[0].ValueNumbers["speed"]) + assert.Equal(t, 20.0, result[1].ValueNumbers["speed"]) +} + +func TestBuildSignalsLatestQuery_Fields(t *testing.T) { + latestArgs := &model.LatestSignalsArgs{ + IncludeLastSeen: true, + SignalNames: map[string]struct{}{"speed": {}, "obdFuelTypeName": {}}, + LocationSignalNames: map[string]struct{}{ + vss.FieldCurrentLocationCoordinates: {}, + }, + } + + query := BuildSignalsLatestQuery(latestArgs) + + assert.Contains(t, query, `lastSeen`) + assert.Contains(t, query, `speed { timestamp value }`) + assert.Contains(t, query, `obdFuelTypeName { timestamp value }`) + assert.Contains(t, query, vss.FieldCurrentLocationCoordinates+` { timestamp value { latitude longitude hdop } }`) +} + +func TestBuildSignalsLatestQuery_NoLastSeen(t *testing.T) { + latestArgs := &model.LatestSignalsArgs{ + IncludeLastSeen: false, + SignalNames: map[string]struct{}{"speed": {}}, + } + + query := BuildSignalsLatestQuery(latestArgs) + assert.NotContains(t, query, `lastSeen`) +} + +func TestUnmarshalSignalsLatestResponse(t *testing.T) { + ts := "2024-06-01T12:00:00Z" + data := json.RawMessage(`{"signalsLatest":{"speed":{"timestamp":"` + ts + `","value":60.0}}}`) + + coll, err := UnmarshalSignalsLatestResponse(data) + require.NoError(t, err) + require.NotNil(t, coll) + require.NotNil(t, coll.Speed) + assert.Equal(t, 60.0, coll.Speed.Value) +} + +func TestUnmarshalSignalsLatestResponse_Nil(t *testing.T) { + data := json.RawMessage(`{"signalsLatest":null}`) + coll, err := UnmarshalSignalsLatestResponse(data) + require.NoError(t, err) + assert.Nil(t, coll) +} + +func TestBuildSignalsQuery_ContainsRequiredVars(t *testing.T) { + query, _ := BuildSignalsQuery(&model.AggregatedSignalArgs{}) + for _, v := range []string{"$subject", "$interval", "$from", "$to", "$filter"} { + assert.True(t, strings.Contains(query, v), "missing variable %s", v) + } +} diff --git a/internal/repositories/repositories.go b/internal/repositories/repositories.go index b708f7a..a356255 100644 --- a/internal/repositories/repositories.go +++ b/internal/repositories/repositories.go @@ -150,7 +150,7 @@ func (r *Repository) GetSignalLatest(ctx context.Context, latestArgs *model.Late } model.SetCollectionField(coll, signal) } - setApproximateLocationInCollection(coll) + SetApproximateLocationInCollection(coll) return coll, nil } @@ -331,7 +331,9 @@ func GetApproximateLoc(lat, long float64) *h3.LatLng { return &latLong } -func setApproximateLocationInCollection(coll *model.SignalCollection) { +// SetApproximateLocationInCollection derives the approximate location field from the +// raw coordinates already present in coll. Called after fetching signalsLatest data. +func SetApproximateLocationInCollection(coll *model.SignalCollection) { if coll == nil || coll.CurrentLocationCoordinates == nil { return }