diff --git a/api/storeEventRunner/runner.go b/api/storeEventRunner/runner.go index d5132cc7..a1ef7d51 100644 --- a/api/storeEventRunner/runner.go +++ b/api/storeEventRunner/runner.go @@ -3,6 +3,7 @@ package storeeventrunner import ( "context" "log/slog" + "strings" documentstore "queryorchestration/internal/document/store" ) @@ -62,6 +63,15 @@ const ( func (s Runner) Process(ctx context.Context, body S3EventNotification) bool { for _, record := range body.Records { + // Skip batch processing keys - these are handled separately + if strings.HasPrefix(record.S3.Object.Key, "batches/") { + slog.Debug("discarding batch message", + "key", record.S3.Object.Key, + "bucket", record.S3.Bucket.Name, + "event", record.EventName) + continue + } + err := s.svc.Store.Process(ctx, documentstore.Params{ Bucket: record.S3.Bucket.Name, Key: record.S3.Object.Key, diff --git a/api/storeEventRunner/runner_test.go b/api/storeEventRunner/runner_test.go index b07e473c..9d4b3b74 100644 --- a/api/storeEventRunner/runner_test.go +++ b/api/storeEventRunner/runner_test.go @@ -72,3 +72,113 @@ func TestStoreEventRunner(t *testing.T) { test.AssertMessageBody(t, cfg, cfg.GetDocInitURL(), regexp.MustCompile(`^\{"bucket":"bucketName","key":"hi\/import\/.+\/0\/.+","hash":"example_hash"\}$`)) } + +func TestStoreEventRunner_BatchFiltering(t *testing.T) { + cfg := &DocInitConfig{} + test.CreateDB(t, cfg) + acfg := test.CreateAWSContainer(t, cfg) + + test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint) + cfg.DocInitURL = test.CreateQueue(t, cfg, test.DocInitRunnerName) + + runner := New(&Services{ + documentstore.New(cfg), + }) + + bucketName := "bucketName" + + // Test that batch keys are filtered out + batchDoc := S3EventNotification{ + Records: []S3EventRecord{ + { + EventName: EventS3ObjectCreatedPut, + S3: S3EventRecordDetails{ + Bucket: S3Bucket{ + Name: bucketName, + }, + Object: S3Object{ + Key: "batches/test_client/2025/10/06/20/batch-id/file.pdf", + ETag: "example_hash", + }, + }, + }, + }, + } + + // Should return true (success) but not process the batch message + assert.True(t, runner.Process(t.Context(), batchDoc)) + + // Verify no message was sent to the queue (batch was filtered) + // Try to receive a message with a short wait time - should get nothing + result, err := cfg.ReceiveFromQueue(t.Context(), &queue.ReceiveParams{ + QueueURL: cfg.GetDocInitURL(), + VisibilityTimeout: 1, + WaitTimeSeconds: 1, + }) + require.NoError(t, err) + assert.Empty(t, result.Messages, "expected no messages in queue for batch key") +} + +func TestStoreEventRunner_MixedMessages(t *testing.T) { + cfg := &DocInitConfig{} + test.CreateDB(t, cfg) + acfg := test.CreateAWSContainer(t, cfg) + + test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint) + cfg.DocInitURL = test.CreateQueue(t, cfg, test.DocInitRunnerName) + + runner := New(&Services{ + documentstore.New(cfg), + }) + + clientId := "hi" + bucketName := "bucketName" + location := objectstore.BucketKey{ + Location: objectstore.Import, + ClientID: clientId, + CreatedAt: time.Now().UTC(), + EntityID: uuid.New(), + } + + err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{ + Clientid: clientId, + Name: "name", + }) + require.NoError(t, err) + + // Mix of batch and normal messages + mixedDoc := S3EventNotification{ + Records: []S3EventRecord{ + { + EventName: EventS3ObjectCreatedPut, + S3: S3EventRecordDetails{ + Bucket: S3Bucket{ + Name: bucketName, + }, + Object: S3Object{ + Key: "batches/test_client/2025/10/06/20/batch-id/file.pdf", + ETag: "batch_hash", + }, + }, + }, + { + EventName: EventS3ObjectCreatedPut, + S3: S3EventRecordDetails{ + Bucket: S3Bucket{ + Name: bucketName, + }, + Object: S3Object{ + Key: location.String(), + ETag: "example_hash", + }, + }, + }, + }, + } + + // Should return true and only process the non-batch message + assert.True(t, runner.Process(t.Context(), mixedDoc)) + + // Verify only one message was sent (the non-batch one) + test.AssertMessageBody(t, cfg, cfg.GetDocInitURL(), regexp.MustCompile(`^\{"bucket":"bucketName","key":"hi\/import\/.+\/0\/.+","hash":"example_hash"\}$`)) +} diff --git a/internal/document/store/process.go b/internal/document/store/process.go index ba6b6f56..1786820e 100644 --- a/internal/document/store/process.go +++ b/internal/document/store/process.go @@ -3,6 +3,7 @@ package documentstore import ( "context" "log/slog" + "net/url" docinitrunner "queryorchestration/api/docInitRunner" "queryorchestration/internal/serviceconfig/objectstore" @@ -29,9 +30,16 @@ func (s *Service) Process(ctx context.Context, params Params) error { return nil } - key, err := objectstore.ParseBucketKey(params.Key) + // URL decode the key (AWS S3 sends URL-encoded keys, LocalStack doesn't) + decodedKey, err := url.QueryUnescape(params.Key) if err != nil { - slog.Error("unable to parse key", "key", params.Key) + slog.Error("unable to decode key", "key", params.Key, "error", err) + return nil + } + + key, err := objectstore.ParseBucketKey(decodedKey) + if err != nil { + slog.Error("unable to parse key", "key", decodedKey, "original_key", params.Key, "error", err) return nil }