feat: proxy cloud event queries to dq behind DQ_ENDPOINT flag

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})
 }

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) {

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"`
 }

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?

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
+}