From 79be46c8a9e55c818cff3bafa0450c194d577b43 Mon Sep 17 00:00:00 2001 From: Dylan Moreland Date: Wed, 22 Apr 2026 13:33:45 -0400 Subject: [PATCH] feat: proxy cloud event queries to dq behind DQ_ENDPOINT flag When DQ_ENDPOINT is set, all five GraphQL query resolvers forward to the unified dq service instead of hitting ClickHouse/S3 directly. Auth is validated locally first; the raw JWT is forwarded to dq. latestIndex/indexes are mapped to dq's latestCloudEvent/cloudEvents (header-only selection) and return indexKey: "" since dq no longer exposes that field. Co-Authored-By: Claude Sonnet 4.6 --- internal/app/app.go | 20 ++- internal/app/middleware.go | 13 ++ internal/config/settings.go | 3 + internal/graph/base.resolvers.go | 49 +++++++ internal/graph/resolver.go | 3 + internal/proxy/client.go | 70 ++++++++++ internal/proxy/transform.go | 212 +++++++++++++++++++++++++++++++ 7 files changed, 364 insertions(+), 6 deletions(-) create mode 100644 internal/proxy/client.go create mode 100644 internal/proxy/transform.go diff --git a/internal/app/app.go b/internal/app/app.go index 19d66d0..5cbcf44 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -13,6 +13,7 @@ import ( "github.com/DIMO-Network/fetch-api/internal/graph" "github.com/DIMO-Network/fetch-api/internal/identity" "github.com/DIMO-Network/fetch-api/internal/limits" + "github.com/DIMO-Network/fetch-api/internal/proxy" "github.com/DIMO-Network/fetch-api/pkg/eventrepo" fetchgrpc "github.com/DIMO-Network/fetch-api/pkg/grpc" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -53,7 +54,11 @@ func New(settings config.Settings) (*App, error) { identityClient = identity.New(settings.IdentityAPIURL) } - es := newExecutableSchema(eventService, buckets, identityClient) + var proxyClient *proxy.Client + if settings.DQEndpoint != "" { + proxyClient = proxy.NewClient(settings.DQEndpoint) + } + es := newExecutableSchema(eventService, buckets, identityClient, proxyClient) gqlSrv := newGraphQLHandler(es) jwtMiddleware, err := auth.NewJWTMiddleware(settings.TokenExchangeIssuer, settings.TokenExchangeJWTKeySetURL) @@ -73,10 +78,12 @@ func New(settings config.Settings) (*App, error) { authChain := func(inner http.Handler) http.Handler { return PanicRecoveryMiddleware( LoggerMiddleware( - limiter.AddRequestTimeout( - jwtMiddleware.CheckJWT( - authLoggerMiddleware( - auth.AddClaimHandler(inner), + rawAuthMiddleware( + limiter.AddRequestTimeout( + jwtMiddleware.CheckJWT( + authLoggerMiddleware( + auth.AddClaimHandler(inner), + ), ), ), ), @@ -109,11 +116,12 @@ func (a *App) Cleanup() { } // newExecutableSchema builds the gqlgen ExecutableSchema shared by the GraphQL and MCP handlers. -func newExecutableSchema(eventService *eventrepo.Service, buckets []string, identityClient identity.Client) graphql.ExecutableSchema { +func newExecutableSchema(eventService *eventrepo.Service, buckets []string, identityClient identity.Client, proxyClient *proxy.Client) graphql.ExecutableSchema { resolver := &graph.Resolver{ EventService: eventService, Buckets: buckets, IdentityClient: identityClient, + ProxyClient: proxyClient, } return graph.NewExecutableSchema(graph.Config{Resolvers: resolver}) } diff --git a/internal/app/middleware.go b/internal/app/middleware.go index 184ae12..ee487e0 100644 --- a/internal/app/middleware.go +++ b/internal/app/middleware.go @@ -1,12 +1,14 @@ package app import ( + "context" "fmt" "net/http" "os" "runtime/debug" "github.com/DIMO-Network/fetch-api/internal/auth" + "github.com/DIMO-Network/fetch-api/internal/proxy" "github.com/rs/zerolog" ) @@ -45,6 +47,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 ada6e0d..159480f 100644 --- a/internal/config/settings.go +++ b/internal/config/settings.go @@ -22,4 +22,7 @@ type Settings struct { S3AWSSecretAccessKey string `yaml:"S3_AWS_SECRET_ACCESS_KEY"` Clickhouse config.Settings `yaml:",inline"` IdentityAPIURL string `yaml:"IDENTITY_API_URL"` + // DQEndpoint is the URL of the dq GraphQL endpoint (e.g. http://dq:3000/query). + // When set, all 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 02877eb..7348339 100644 --- a/internal/graph/base.resolvers.go +++ b/internal/graph/base.resolvers.go @@ -48,6 +48,13 @@ func (r *queryResolver) LatestIndex(ctx context.Context, did string, filter *mod if err != nil { return nil, err } + if r.ProxyClient != nil { + result, err := r.ProxyClient.ProxyLatestIndex(ctx, opts.Subject.In[0], filter) + if err != nil { + return nil, err + } + return &model.CloudEventIndex{Header: &result.Header, IndexKey: ""}, nil + } idx, err := r.EventService.GetLatestIndexAdvanced(ctx, opts) if err != nil { return nil, err @@ -61,6 +68,17 @@ func (r *queryResolver) Indexes(ctx context.Context, did string, limit *int, fil if err != nil { return nil, err } + if r.ProxyClient != nil { + results, err := r.ProxyClient.ProxyIndexes(ctx, opts.Subject.In[0], limit, filter) + if err != nil { + return nil, err + } + out := make([]*model.CloudEventIndex, len(results)) + for i := range results { + out[i] = &model.CloudEventIndex{Header: &results[i].Header, IndexKey: ""} + } + return out, nil + } list, err := r.EventService.ListIndexesAdvanced(ctx, resolveLimit(limit), opts) if err != nil { if errors.Is(err, sql.ErrNoRows) { @@ -81,6 +99,18 @@ func (r *queryResolver) LatestCloudEvent(ctx context.Context, did string, filter if err != nil { return nil, err } + if r.ProxyClient != nil { + result, err := r.ProxyClient.ProxyLatestCloudEvent(ctx, opts.Subject.In[0], filter) + if err != nil { + return nil, err + } + raw := &cloudevent.RawEvent{ + CloudEventHeader: result.Header, + DataBase64: result.DataBase64, + } + raw.Data = result.Data + return &CloudEventWrapper{Raw: raw, DataURL: result.DataURL}, nil + } idx, err := r.EventService.GetLatestIndexAdvanced(ctx, opts) if err != nil { return nil, err @@ -106,6 +136,22 @@ func (r *queryResolver) CloudEvents(ctx context.Context, did string, limit *int, if err != nil { return nil, err } + if r.ProxyClient != nil { + results, err := r.ProxyClient.ProxyCloudEvents(ctx, opts.Subject.In[0], limit, filter) + if err != nil { + return nil, err + } + out := make([]*CloudEventWrapper, len(results)) + for i, result := range results { + raw := &cloudevent.RawEvent{ + CloudEventHeader: result.Header, + DataBase64: result.DataBase64, + } + raw.Data = result.Data + out[i] = &CloudEventWrapper{Raw: raw, DataURL: result.DataURL} + } + return out, nil + } list, err := r.EventService.ListIndexesAdvanced(ctx, resolveLimit(limit), opts) if err != nil { if errors.Is(err, sql.ErrNoRows) { @@ -152,6 +198,9 @@ func (r *queryResolver) AvailableCloudEventTypes(ctx context.Context, did string if err != nil { return nil, err } + if r.ProxyClient != nil { + return r.ProxyClient.ProxyAvailableCloudEventTypes(ctx, opts.Subject.In[0], filter) + } summaries, err := r.EventService.GetCloudEventTypeSummariesAdvanced(ctx, opts) if err != nil { return nil, err diff --git a/internal/graph/resolver.go b/internal/graph/resolver.go index 0a50835..21461e3 100644 --- a/internal/graph/resolver.go +++ b/internal/graph/resolver.go @@ -10,6 +10,7 @@ import ( "github.com/DIMO-Network/cloudevent" "github.com/DIMO-Network/fetch-api/internal/graph/model" "github.com/DIMO-Network/fetch-api/internal/identity" + "github.com/DIMO-Network/fetch-api/internal/proxy" "github.com/DIMO-Network/fetch-api/pkg/eventrepo" "github.com/DIMO-Network/fetch-api/pkg/grpc" "github.com/DIMO-Network/token-exchange-api/pkg/tokenclaims" @@ -26,6 +27,8 @@ type Resolver struct { EventService *eventrepo.Service Buckets []string IdentityClient identity.Client + // ProxyClient, when non-nil, forwards all queries to dq instead of ClickHouse. + ProxyClient *proxy.Client } // TODO(elffjs): Shouldn't these be Errors? 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/transform.go b/internal/proxy/transform.go new file mode 100644 index 0000000..2d3b756 --- /dev/null +++ b/internal/proxy/transform.go @@ -0,0 +1,212 @@ +package proxy + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/DIMO-Network/cloudevent" + "github.com/DIMO-Network/fetch-api/internal/graph/model" +) + +// cloudEventHeaderFields is the GraphQL selection set for CloudEventHeader, shared across queries. +const cloudEventHeaderFields = `header { + specversion type source subject id time + datacontenttype dataschema dataversion producer signature raweventid tags +}` + +// ProxiedCloudEvent holds a cloud event returned from dq, in fetch-api-native types. +// This avoids a circular import between the proxy and graph packages. +type ProxiedCloudEvent struct { + Header cloudevent.CloudEventHeader + Data []byte // raw JSON payload; nil when absent + DataBase64 string + DataURL string +} + +// ProxiedCloudIndex holds a cloud event header returned from dq for index-style queries. +type ProxiedCloudIndex struct { + Header cloudevent.CloudEventHeader +} + +// dqCloudEventJSON is an intermediate struct for unmarshaling dq cloud event JSON. +// dq uses camelCase field names (dataBase64, dataUrl) that differ from the cloudevent +// library's wire format (data_base64), so we cannot unmarshal directly into RawEvent. +type dqCloudEventJSON struct { + Header cloudevent.CloudEventHeader `json:"header"` + Data json.RawMessage `json:"data"` + DataBase64 string `json:"dataBase64"` + DataUrl string `json:"dataUrl"` +} + +func dqJSONToProxied(ce dqCloudEventJSON) ProxiedCloudEvent { + if ce.Header.Tags == nil { + ce.Header.Tags = []string{} + } + var data []byte + if len(ce.Data) > 0 && string(ce.Data) != "null" { + data = ce.Data + } + return ProxiedCloudEvent{ + Header: ce.Header, + Data: data, + DataBase64: ce.DataBase64, + DataURL: ce.DataUrl, + } +} + +func dqJSONToProxiedIndex(ce dqCloudEventJSON) ProxiedCloudIndex { + if ce.Header.Tags == nil { + ce.Header.Tags = []string{} + } + return ProxiedCloudIndex{Header: ce.Header} +} + +// BuildLatestCloudEventQuery constructs the dq query for latestCloudEvent, requesting all payload fields. +func BuildLatestCloudEventQuery() string { + return `query LatestCloudEvent($subject: String!, $filter: CloudEventFilter) { + latestCloudEvent(subject: $subject, filter: $filter) { + ` + cloudEventHeaderFields + ` + data dataBase64 dataUrl + } + }` +} + +// BuildCloudEventsQuery constructs the dq query for cloudEvents, requesting all payload fields. +func BuildCloudEventsQuery() string { + return `query CloudEvents($subject: String!, $limit: Int, $filter: CloudEventFilter) { + cloudEvents(subject: $subject, limit: $limit, filter: $filter) { + ` + cloudEventHeaderFields + ` + data dataBase64 dataUrl + } + }` +} + +// BuildLatestIndexQuery maps latestIndex to dq's latestCloudEvent with header fields only, +// avoiding an unnecessary S3 fetch since the index queries return metadata only. +func BuildLatestIndexQuery() string { + return `query LatestIndex($subject: String!, $filter: CloudEventFilter) { + latestCloudEvent(subject: $subject, filter: $filter) { + ` + cloudEventHeaderFields + ` + } + }` +} + +// BuildIndexesQuery maps indexes to dq's cloudEvents with header fields only. +func BuildIndexesQuery() string { + return `query Indexes($subject: String!, $limit: Int, $filter: CloudEventFilter) { + cloudEvents(subject: $subject, limit: $limit, filter: $filter) { + ` + cloudEventHeaderFields + ` + } + }` +} + +// BuildAvailableCloudEventTypesQuery constructs the dq query for availableCloudEventTypes. +func BuildAvailableCloudEventTypesQuery() string { + return `query AvailableCloudEventTypes($subject: String!, $filter: CloudEventFilter) { + availableCloudEventTypes(subject: $subject, filter: $filter) { + type count firstSeen lastSeen + } + }` +} + +// ProxyLatestCloudEvent forwards latestCloudEvent to dq and returns the result. +func (c *Client) ProxyLatestCloudEvent(ctx context.Context, subject string, filter *model.CloudEventFilter) (ProxiedCloudEvent, error) { + data, err := c.Execute(ctx, BuildLatestCloudEventQuery(), map[string]any{ + "subject": subject, + "filter": filter, + }) + if err != nil { + return ProxiedCloudEvent{}, err + } + var envelope struct { + LatestCloudEvent dqCloudEventJSON `json:"latestCloudEvent"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return ProxiedCloudEvent{}, fmt.Errorf("unmarshaling latestCloudEvent: %w", err) + } + return dqJSONToProxied(envelope.LatestCloudEvent), nil +} + +// ProxyCloudEvents forwards cloudEvents to dq and returns the results. +func (c *Client) ProxyCloudEvents(ctx context.Context, subject string, limit *int, filter *model.CloudEventFilter) ([]ProxiedCloudEvent, error) { + data, err := c.Execute(ctx, BuildCloudEventsQuery(), map[string]any{ + "subject": subject, + "limit": limit, + "filter": filter, + }) + if err != nil { + return nil, err + } + var envelope struct { + CloudEvents []dqCloudEventJSON `json:"cloudEvents"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling cloudEvents: %w", err) + } + out := make([]ProxiedCloudEvent, len(envelope.CloudEvents)) + for i, ce := range envelope.CloudEvents { + out[i] = dqJSONToProxied(ce) + } + return out, nil +} + +// ProxyLatestIndex forwards latestIndex to dq via latestCloudEvent (header only) and returns the result. +func (c *Client) ProxyLatestIndex(ctx context.Context, subject string, filter *model.CloudEventFilter) (ProxiedCloudIndex, error) { + data, err := c.Execute(ctx, BuildLatestIndexQuery(), map[string]any{ + "subject": subject, + "filter": filter, + }) + if err != nil { + return ProxiedCloudIndex{}, err + } + var envelope struct { + LatestCloudEvent dqCloudEventJSON `json:"latestCloudEvent"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return ProxiedCloudIndex{}, fmt.Errorf("unmarshaling latestCloudEvent (for latestIndex): %w", err) + } + return dqJSONToProxiedIndex(envelope.LatestCloudEvent), nil +} + +// ProxyIndexes forwards indexes to dq via cloudEvents (header only) and returns the results. +func (c *Client) ProxyIndexes(ctx context.Context, subject string, limit *int, filter *model.CloudEventFilter) ([]ProxiedCloudIndex, error) { + data, err := c.Execute(ctx, BuildIndexesQuery(), map[string]any{ + "subject": subject, + "limit": limit, + "filter": filter, + }) + if err != nil { + return nil, err + } + var envelope struct { + CloudEvents []dqCloudEventJSON `json:"cloudEvents"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling cloudEvents (for indexes): %w", err) + } + out := make([]ProxiedCloudIndex, len(envelope.CloudEvents)) + for i, ce := range envelope.CloudEvents { + out[i] = dqJSONToProxiedIndex(ce) + } + return out, nil +} + +// ProxyAvailableCloudEventTypes forwards availableCloudEventTypes to dq and returns the results. +func (c *Client) ProxyAvailableCloudEventTypes(ctx context.Context, subject string, filter *model.CloudEventFilter) ([]*model.CloudEventTypeSummary, error) { + data, err := c.Execute(ctx, BuildAvailableCloudEventTypesQuery(), map[string]any{ + "subject": subject, + "filter": filter, + }) + if err != nil { + return nil, err + } + var envelope struct { + AvailableCloudEventTypes []*model.CloudEventTypeSummary `json:"availableCloudEventTypes"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, fmt.Errorf("unmarshaling availableCloudEventTypes: %w", err) + } + return envelope.AvailableCloudEventTypes, nil +} +