@@ -36,11 +36,19 @@ type Service struct {
3636 chConn clickhouse.Conn
3737 // parquetBucket is the object storage bucket for Iceberg Parquet files.
3838 parquetBucket string
39+ // blobBucket is the object storage bucket for externalized event payloads
40+ // (referenced by data_index_key).
41+ blobBucket string
3942}
4043
4144// ObjectInfo is the information about the object in S3.
4245type ObjectInfo struct {
46+ // Key is the index_key — a pointer into parquet (key#row) or a legacy JSON object.
4347 Key string
48+ // DataIndexKey, when non-empty, is the key in the blob bucket holding the
49+ // event's externalized payload. Set by the producer when the payload was
50+ // split out of the inline event.
51+ DataIndexKey string
4452}
4553
4654// ObjectGetter is an interface for getting an object from S3.
@@ -54,37 +62,34 @@ type Presigner interface {
5462 PresignGetObject (ctx context.Context , params * s3.GetObjectInput , optFns ... func (* s3.PresignOptions )) (* v4.PresignedHTTPRequest , error )
5563}
5664
57- // BlobKeyPrefix is the S3 key prefix used for large binary blob objects.
58- // Keys with this prefix are served via presigned URL instead of inline in the response.
59- const BlobKeyPrefix = "cloudevent/blobs/"
60-
6165// presignTTL is the lifetime of generated presigned S3 URLs.
6266const presignTTL = 15 * time .Minute
6367
6468// New creates a new instance of Service.
65- func New (chConn clickhouse.Conn , objGetter ObjectGetter , presigner Presigner , parquetBucket string ) * Service {
69+ func New (chConn clickhouse.Conn , objGetter ObjectGetter , presigner Presigner , parquetBucket , blobBucket string ) * Service {
6670 return & Service {
6771 objGetter : objGetter ,
6872 presigner : presigner ,
6973 chConn : chConn ,
7074 parquetBucket : parquetBucket ,
75+ blobBucket : blobBucket ,
7176 }
7277}
7378
74- // PresignBlobURL returns a short-lived presigned GET URL for the given S3 key in the parquet bucket.
79+ // PresignBlobURL returns a short-lived presigned GET URL for the given S3 key in the blob bucket.
7580func (s * Service ) PresignBlobURL (ctx context.Context , key string ) (string , error ) {
7681 if s .presigner == nil {
7782 return "" , fmt .Errorf ("presigner not configured" )
7883 }
79- if s .parquetBucket == "" {
80- return "" , fmt .Errorf ("parquet bucket not configured" )
84+ if s .blobBucket == "" {
85+ return "" , fmt .Errorf ("blob bucket not configured" )
8186 }
8287 req , err := s .presigner .PresignGetObject (ctx , & s3.GetObjectInput {
83- Bucket : aws .String (s .parquetBucket ),
88+ Bucket : aws .String (s .blobBucket ),
8489 Key : aws .String (key ),
8590 }, s3 .WithPresignExpires (presignTTL ))
8691 if err != nil {
87- return "" , fmt .Errorf ("presign %s/%s: %w" , s .parquetBucket , key , err )
92+ return "" , fmt .Errorf ("presign %s/%s: %w" , s .blobBucket , key , err )
8893 }
8994 return req .URL , nil
9095}
@@ -143,6 +148,7 @@ func (s *Service) ListIndexesAdvanced(ctx context.Context, limit int, advancedOp
143148 chindexer .DataVersionColumn ,
144149 chindexer .ExtrasColumn ,
145150 chindexer .IndexKeyColumn ,
151+ chindexer .DataIndexKeyColumn ,
146152 ),
147153 qm .From (chindexer .TableName ),
148154 qm .OrderBy (chindexer .TimestampColumn + order ),
@@ -164,7 +170,7 @@ func (s *Service) ListIndexesAdvanced(ctx context.Context, limit int, advancedOp
164170 var extras string
165171 for rows .Next () {
166172 var event cloudevent.CloudEvent [ObjectInfo ]
167- err = rows .Scan (& event .Subject , & event .Time , & event .Type , & event .ID , & event .Source , & event .Producer , & event .DataContentType , & event .DataVersion , & extras , & event .Data .Key )
173+ err = rows .Scan (& event .Subject , & event .Time , & event .Type , & event .ID , & event .Source , & event .Producer , & event .DataContentType , & event .DataVersion , & extras , & event .Data .Key , & event . Data . DataIndexKey )
168174 if err != nil {
169175 _ = rows .Close ()
170176 return nil , fmt .Errorf ("failed to scan cloud event: %w" , err )
@@ -349,12 +355,12 @@ func (s *Service) ListCloudEventsFromIndexes(ctx context.Context, indexes []clou
349355 }
350356 defer func () { _ = pr .Close () }()
351357 for _ , item := range items {
352- event , err := pr .SeekToRow (item .rowOffset )
358+ stored , err := pr .SeekToRow (item .rowOffset )
353359 if err != nil {
354360 return fmt .Errorf ("seek to row %d in %s: %w" , item .rowOffset , item .objectKey , err )
355361 }
356- event .Tags = grpc .TagsOrEmpty (event .Tags )
357- events [item .idx ] = event
362+ stored .Tags = grpc .TagsOrEmpty (stored .Tags )
363+ events [item .idx ] = stored . RawEvent
358364 }
359365 return nil
360366 })
@@ -387,7 +393,15 @@ func (s *Service) ListCloudEventsFromIndexes(ctx context.Context, indexes []clou
387393}
388394
389395// GetCloudEventFromIndex fetches and returns the cloud event for the given index.
396+ // Events with a non-empty DataIndexKey have their payload externalized to the
397+ // blob bucket; this method returns header-only since the data is meant to be
398+ // served via presigned URL by the GraphQL layer (see PresignBlobURL).
390399func (s * Service ) GetCloudEventFromIndex (ctx context.Context , index * cloudevent.CloudEvent [ObjectInfo ], bucketName string ) (cloudevent.RawEvent , error ) {
400+ if index .Data .DataIndexKey != "" {
401+ hdr := index .CloudEventHeader
402+ hdr .Tags = grpc .TagsOrEmpty (hdr .Tags )
403+ return cloudevent.RawEvent {CloudEventHeader : hdr }, nil
404+ }
391405 if parquet .IsParquetRef (index .Data .Key ) {
392406 return s .getCloudEventFromParquet (ctx , index .Data .Key )
393407 }
@@ -454,12 +468,12 @@ func (s *Service) getCloudEventFromParquet(ctx context.Context, key string) (clo
454468 return cloudevent.RawEvent {}, fmt .Errorf ("create s3 reader for %s: %w" , objectKey , err )
455469 }
456470
457- event , err := parquet .SeekToRow (reader , reader .Size (), rowOffset )
471+ stored , err := parquet .SeekToRow (reader , reader .Size (), rowOffset )
458472 if err != nil {
459473 return cloudevent.RawEvent {}, fmt .Errorf ("seek to row %d in %s: %w" , rowOffset , objectKey , err )
460474 }
461- event .Tags = grpc .TagsOrEmpty (event .Tags )
462- return event , nil
475+ stored .Tags = grpc .TagsOrEmpty (stored .Tags )
476+ return stored . RawEvent , nil
463477}
464478
465479// parseParquetRef parses a parquet index key into bucket, object key, and row offset.
0 commit comments