Add presigned S3 URL support for large blob payloads (#66)

* Add S3 presigned URL support for blob CloudEvents

Objects stored under the cloudevent/blobs/ key prefix are now served
via a short-lived presigned S3 GET URL (dataUrl field) instead of being
downloaded and embedded inline in the GraphQL response. This avoids
ballooning the JSON payload for large binary objects (e.g. scans).

- Add Presigner interface + PresignBlobURL method to eventrepo.Service
- Export BlobKeyPrefix constant ("cloudevent/blobs/")
- Add dataUrl: String field to GraphQL CloudEvent type
- Detect blob prefix in LatestCloudEvent / CloudEvents resolvers;
  skip GetObject and presign against the primary bucket instead
- Wire s3.NewPresignClient into both GraphQL and gRPC app paths
- Update eventrepo.New signature; pass nil in tests that don't need presigning
- Add MockPresigner to generated mock file

https://claude.ai/code/session_015ReeLGeCywJfYkkrrng5wU

* Improve dataUrl field description in GraphQL schema

Describe the field in terms of behavior (large files) rather than the
internal key prefix convention, which is an implementation detail that
may change.

https://claude.ai/code/session_015ReeLGeCywJfYkkrrng5wU

* Add tests for presigned URL generation and blob resolver fields

- pkg/eventrepo/presign_test.go: unit tests for PresignBlobURL covering
  correct bucket/key routing, 15-minute TTL, presigner error propagation,
  and nil-presigner guard
- internal/graph/blob_resolver_test.go: unit tests for the DataUrl resolver
  (nil wrapper, empty DataURL, populated DataURL) and a composite test that
  a blob wrapper returns nil for data/dataBase64 and a URL for dataUrl

https://claude.ai/code/session_015ReeLGeCywJfYkkrrng5wU

* Get rid of these imagined handlers

* These tests don't do much

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: elffjs

internal/app/app.go

@@ -14,6 +14,7 @@ import (
 	"github.com/DIMO-Network/fetch-api/internal/limits"
 	"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"
 	"github.com/DIMO-Network/server-garage/pkg/gql/errorhandler"
 	gqlmetrics "github.com/DIMO-Network/server-garage/pkg/gql/metrics"
 	"github.com/DIMO-Network/shared/pkg/middleware/metrics"
@@ -42,7 +43,7 @@ func New(settings config.Settings) (*App, error) {
 	}
 	s3Client := s3ClientFromSettings(&settings)
 	buckets := []string{settings.CloudEventBucket, settings.EphemeralBucket, settings.ParquetBucket}
-	eventService := eventrepo.New(chConn, s3Client, settings.ParquetBucket)
+	eventService := eventrepo.New(chConn, s3Client, s3.NewPresignClient(s3Client), settings.ParquetBucket)
 
 	var identityClient identity.Client
 	if settings.IdentityAPIURL != "" {
@@ -118,7 +119,7 @@ func CreateGRPCServer(logger *zerolog.Logger, settings *config.Settings) (*grpc.
 	}
 
 	s3Client := s3ClientFromSettings(settings)
-	eventService := eventrepo.New(chConn, s3Client, settings.ParquetBucket)
+	eventService := eventrepo.New(chConn, s3Client, s3.NewPresignClient(s3Client), settings.ParquetBucket)
 
 	rpcServer := rpc.NewServer([]string{settings.CloudEventBucket, settings.EphemeralBucket, settings.ParquetBucket}, eventService)
 

internal/config/settings.go

@@ -7,13 +7,13 @@ import (
 
 // Settings contains the application config.
 type Settings struct {
-	Port                      int    `yaml:"PORT"`
-	MonPort                   int    `yaml:"MON_PORT"`
-	GRPCPort                  int    `yaml:"GRPC_PORT"`
-	EnablePprof               bool   `yaml:"ENABLE_PPROF"`
-	MaxRequestDuration        string `yaml:"MAX_REQUEST_DURATION"`
-	TokenExchangeJWTKeySetURL string `yaml:"TOKEN_EXCHANGE_JWK_KEY_SET_URL"`
-	TokenExchangeIssuer       string `yaml:"TOKEN_EXCHANGE_ISSUER_URL"`
+	Port                      int             `yaml:"PORT"`
+	MonPort                   int             `yaml:"MON_PORT"`
+	GRPCPort                  int             `yaml:"GRPC_PORT"`
+	EnablePprof               bool            `yaml:"ENABLE_PPROF"`
+	MaxRequestDuration        string          `yaml:"MAX_REQUEST_DURATION"`
+	TokenExchangeJWTKeySetURL string          `yaml:"TOKEN_EXCHANGE_JWK_KEY_SET_URL"`
+	TokenExchangeIssuer       string          `yaml:"TOKEN_EXCHANGE_ISSUER_URL"`
 	CloudEventBucket          string          `yaml:"CLOUDEVENT_BUCKET"`
 	EphemeralBucket           string          `yaml:"EPHEMERAL_BUCKET"`
 	ParquetBucket             string          `yaml:"PARQUET_BUCKET"`

internal/graph/base.resolvers.go

@@ -7,9 +7,10 @@ import (
 )
 
 // CloudEventWrapper holds a pointer to a RawEvent so resolvers can expose
-// header, data, and dataBase64 without copying the underlying event.
+// header, data, dataBase64, and dataUrl without copying the underlying event.
 type CloudEventWrapper struct {
-	Raw *cloudevent.RawEvent
+	Raw     *cloudevent.RawEvent
+	DataURL string // non-empty when the payload is a blob served via presigned URL
 }
 
 // RawJSON is the raw bytes of a JSON value. It implements graphql.Marshaler by

internal/graph/cloud_event.go

@@ -93,7 +93,7 @@ func TestGetLatestIndexKey(t *testing.T) {
 		},
 	}
 
-	indexService := eventrepo.New(conn, nil, "")
+	indexService := eventrepo.New(conn, nil, nil, "")
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -172,7 +172,7 @@ func TestGetDataFromIndex(t *testing.T) {
 		ContentLength: ref(int64(len(content))),
 	}, nil).AnyTimes()
 
-	indexService := eventrepo.New(conn, mockS3Client, "")
+	indexService := eventrepo.New(conn, mockS3Client, nil, "")
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -204,7 +204,7 @@ func TestStoreObject(t *testing.T) {
 	mockS3Client := NewMockObjectGetter(ctrl)
 	mockS3Client.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3.PutObjectOutput{}, nil).AnyTimes()
 
-	indexService := eventrepo.New(conn, mockS3Client, "")
+	indexService := eventrepo.New(conn, mockS3Client, nil, "")
 
 	content := []byte(`{"vin": "1HGCM82633A123456"}`)
 	did := cloudevent.ERC721DID{
@@ -333,7 +333,7 @@ func TestGetData(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			mockS3Client := NewMockObjectGetter(ctrl)
 
-			indexService := eventrepo.New(conn, mockS3Client, "")
+			indexService := eventrepo.New(conn, mockS3Client, nil, "")
 			// Allow GetObject calls in any order since fetches are concurrent.
 			if len(tt.expectedIndexKeys) > 0 {
 				mockS3Client.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
@@ -424,7 +424,7 @@ func TestGetEventWithAllHeaderFields(t *testing.T) {
 	eventDataEnvelope := []byte(`{"data":` + string(eventData) + `}`)
 
 	// Create service
-	indexService := eventrepo.New(conn, mockS3Client, "")
+	indexService := eventrepo.New(conn, mockS3Client, nil, "")
 
 	// Test retrieving the event
 	t.Run("retrieve event with full headers", func(t *testing.T) {
@@ -576,7 +576,7 @@ func TestGetCloudEventFromIndex_ParquetRef(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	mockS3 := mockS3ParquetReader(t, ctrl, parquetBytes)
 
-	indexService := eventrepo.New(conn, mockS3, "test-parquet-bucket")
+	indexService := eventrepo.New(conn, mockS3, nil, "test-parquet-bucket")
 
 	// Build the index object as GetCloudEventFromIndex expects
 	index := cloudevent.CloudEvent[eventrepo.ObjectInfo]{
@@ -655,7 +655,7 @@ func TestListCloudEventsFromIndexes_ParquetCaching(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	mockS3 := mockS3ParquetReader(t, ctrl, parquetBytes)
 
-	indexService := eventrepo.New(conn, mockS3, "test-parquet-bucket")
+	indexService := eventrepo.New(conn, mockS3, nil, "test-parquet-bucket")
 
 	indexes := []cloudevent.CloudEvent[eventrepo.ObjectInfo]{
 		{CloudEventHeader: hdr0, Data: eventrepo.ObjectInfo{Key: indexKeys[0]}},
@@ -747,7 +747,7 @@ func TestListIndexesAdvanced(t *testing.T) {
 	keyTypeStatusSource1Producer3 := insertTestData(t, ctx, conn, eventIdx3)
 	keyTypeStatusSource3Producer4 := insertTestData(t, ctx, conn, eventIdx4)
 
-	indexService := eventrepo.New(conn, nil, "")
+	indexService := eventrepo.New(conn, nil, nil, "")
 
 	tests := []struct {
 		name              string
@@ -1032,7 +1032,7 @@ func TestGetCloudEventTypeSummaries(t *testing.T) {
 	insertTestData(t, ctx, conn, status3)
 	insertTestData(t, ctx, conn, fp1)
 
-	indexService := eventrepo.New(conn, nil, "")
+	indexService := eventrepo.New(conn, nil, nil, "")
 
 	t.Run("no filter returns all types", func(t *testing.T) {
 		opts := &grpc.SearchOptions{

internal/graph/generated.go

@@ -15,6 +15,8 @@ import (
 	chindexer "github.com/DIMO-Network/cloudevent/clickhouse"
 	"github.com/DIMO-Network/cloudevent/parquet"
 	"github.com/DIMO-Network/fetch-api/pkg/grpc"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/volatiletech/sqlboiler/v4/drivers"
 	"github.com/volatiletech/sqlboiler/v4/queries"
@@ -30,6 +32,7 @@ const tagsColumn = "JSONExtract(extras, 'tags', 'Array(String)')"
 // Service manages and retrieves data messages from indexed objects in S3.
 type Service struct {
 	objGetter ObjectGetter
+	presigner Presigner
 	chConn    clickhouse.Conn
 	// parquetBucket is the object storage bucket for Iceberg Parquet files.
 	parquetBucket string
@@ -46,15 +49,43 @@ type ObjectGetter interface {
 	PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
 }
 
+// Presigner generates presigned S3 GET URLs.
+type Presigner interface {
+	PresignGetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.PresignOptions)) (*v4.PresignedHTTPRequest, error)
+}
+
+// BlobKeyPrefix is the S3 key prefix used for large binary blob objects.
+// Keys with this prefix are served via presigned URL instead of inline in the response.
+const BlobKeyPrefix = "cloudevent/blobs/"
+
+// presignTTL is the lifetime of generated presigned S3 URLs.
+const presignTTL = 15 * time.Minute
+
 // New creates a new instance of Service.
-func New(chConn clickhouse.Conn, objGetter ObjectGetter, parquetBucket string) *Service {
+func New(chConn clickhouse.Conn, objGetter ObjectGetter, presigner Presigner, parquetBucket string) *Service {
 	return &Service{
 		objGetter:     objGetter,
+		presigner:     presigner,
 		chConn:        chConn,
 		parquetBucket: parquetBucket,
 	}
 }
 
+// PresignBlobURL returns a short-lived presigned GET URL for the given S3 key and bucket.
+func (s *Service) PresignBlobURL(ctx context.Context, key, bucket string) (string, error) {
+	if s.presigner == nil {
+		return "", fmt.Errorf("presigner not configured")
+	}
+	req, err := s.presigner.PresignGetObject(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+	}, s3.WithPresignExpires(presignTTL))
+	if err != nil {
+		return "", fmt.Errorf("presign %s/%s: %w", bucket, key, err)
+	}
+	return req.URL, nil
+}
+
 // GetLatestIndex returns the latest cloud event index that matches the given options.
 func (s *Service) GetLatestIndex(ctx context.Context, opts *grpc.SearchOptions) (cloudevent.CloudEvent[ObjectInfo], error) {
 	advancedOpts := convertSearchOptionsToAdvanced(opts)