Merged in feature/decode-sqs-messages (pull request #188)

decode and filter sqs messages

* decode and filter

* code coverage

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,

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"\}$`))
+}

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
 	}