query-orchestration/api/docCleanRunner/runner_test.go

// Package doccleanrunner_test contains integration tests for the docCleanRunner.
// Tests use real database and S3 via testcontainers (no mocks).
package doccleanrunner_test

import (
	"bytes"
	"testing"
	"time"

	doccleanrunner "queryorchestration/api/docCleanRunner"
	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/document/batch/outcome"
	documentclean "queryorchestration/internal/document/clean"
	"queryorchestration/internal/server/runner"
	"queryorchestration/internal/serviceconfig/objectstore"
	"queryorchestration/internal/serviceconfig/queue"
	"queryorchestration/internal/test"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

type DocCleanConfig struct {
	runner.BaseConfig[doccleanrunner.Body]
	queue.QueueConfig
	objectstore.ObjectStoreConfig
}

// TestDocCleanRunner_ReportsCleanPassed tests that for a document that has
// already been cleaned (HasDocumentCleanEntry returns true), the runner
// reports clean_passed.
func TestDocCleanRunner_ReportsCleanPassed(t *testing.T) {
	cfg := &DocCleanConfig{}
	test.CreateDB(t, cfg)
	ctx := t.Context()

	clientID := "clean_passed_" + uuid.New().String()[:8]
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Clientid: clientID,
		Name:     "Test Clean Passed",
	})
	require.NoError(t, err)

	// Create batch
	batchID, err := cfg.GetDBQueries().CreateBatchUpload(ctx, &repository.CreateBatchUploadParams{
		ClientID:         clientID,
		OriginalFilename: "test.zip",
		TotalDocuments:   1,
	})
	require.NoError(t, err)

	// Create document
	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "hash_clean_pass_" + uuid.New().String()[:8],
		BatchID:  &batchID,
	})
	require.NoError(t, err)

	// Create a successful clean entry so HasDocumentCleanEntry returns true.
	// This takes the shortcut path in Clean() -> returns {Passed: true}
	bucket := "bucket"
	key := "key"
	hash := "hash"
	cleanID, err := cfg.GetDBQueries().AddDocumentClean(ctx, &repository.AddDocumentCleanParams{
		Documentid: docID,
		Bucket:     &bucket,
		Key:        &key,
		Hash:       &hash,
		Mimetype: repository.NullCleanmimetype{
			Cleanmimetype: repository.CleanmimetypeApplicationPdf,
			Valid:         true,
		},
	})
	require.NoError(t, err)
	err = cfg.GetDBQueries().AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
		Cleanid: cleanID,
		Version: 1,
	})
	require.NoError(t, err)

	// Insert sync_complete outcome (the state before clean)
	err = cfg.GetDBQueries().InsertBatchDocumentOutcome(ctx, &repository.InsertBatchDocumentOutcomeParams{
		BatchID:    batchID,
		Filename:   "invoice.pdf",
		Column3:    repository.BatchOutcomeStatusSyncComplete,
		DocumentID: &docID,
	})
	require.NoError(t, err)

	cleanRunner := doccleanrunner.New(&doccleanrunner.Services{
		Clean:   documentclean.New(cfg),
		Outcome: outcome.New(cfg.GetDBQueries()),
	})

	assert.True(t, cleanRunner.Process(ctx, doccleanrunner.Body{DocumentID: docID}))

	// Verify outcome updated to clean_passed
	rows, err := cfg.GetDBQueries().ListBatchDocumentOutcomes(ctx, batchID)
	require.NoError(t, err)
	require.Len(t, rows, 1)
	assert.Equal(t, repository.BatchOutcomeStatusCleanPassed, rows[0].Outcome)
}

// TestDocCleanRunner_ReportsCleanFailed tests that a document failing clean
// validation gets the clean_failed outcome.
func TestDocCleanRunner_ReportsCleanFailed(t *testing.T) {
	cfg := &DocCleanConfig{}
	test.CreateDB(t, cfg)
	test.CreateAWSResources(t, cfg)
	ctx := t.Context()

	clientID := "clean_failed_" + uuid.New().String()[:8]
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Clientid: clientID,
		Name:     "Test Clean Failed",
	})
	require.NoError(t, err)

	// Create batch
	batchID, err := cfg.GetDBQueries().CreateBatchUpload(ctx, &repository.CreateBatchUploadParams{
		ClientID:         clientID,
		OriginalFilename: "test.zip",
		TotalDocuments:   1,
	})
	require.NoError(t, err)

	// Upload a non-PDF file to S3 (plain text content that will fail mime type check)
	fileContent := []byte("This is NOT a PDF file. It has no PDF signature.")
	location := objectstore.BucketKey{
		Location:  objectstore.Import,
		ClientID:  clientID,
		CreatedAt: time.Now().UTC(),
		EntityID:  uuid.New(),
	}

	_, err = cfg.GetStoreClient().PutObject(ctx, &s3.PutObjectInput{
		Bucket:      aws.String(cfg.GetBucket()),
		Key:         aws.String(location.String()),
		Body:        bytes.NewReader(fileContent),
		ContentType: aws.String("text/plain"),
	})
	require.NoError(t, err)

	// Get the ETag so we can use it as the hash for IfMatch
	headResp, err := cfg.GetStoreClient().HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String(cfg.GetBucket()),
		Key:    aws.String(location.String()),
	})
	require.NoError(t, err)
	etag := *headResp.ETag

	// Create document with hash matching the ETag
	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     etag,
		BatchID:  &batchID,
	})
	require.NoError(t, err)

	// Add document entry pointing to the S3 file
	err = cfg.GetDBQueries().AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
		Documentid: docID,
		Bucket:     cfg.GetBucket(),
		Key:        location.String(),
	})
	require.NoError(t, err)

	// Insert sync_complete outcome
	err = cfg.GetDBQueries().InsertBatchDocumentOutcome(ctx, &repository.InsertBatchDocumentOutcomeParams{
		BatchID:    batchID,
		Filename:   "bad_file.pdf",
		Column3:    repository.BatchOutcomeStatusSyncComplete,
		DocumentID: &docID,
	})
	require.NoError(t, err)

	cleanRunner := doccleanrunner.New(&doccleanrunner.Services{
		Clean:   documentclean.New(cfg),
		Outcome: outcome.New(cfg.GetDBQueries()),
	})

	// The runner should return true (acknowledge), even for validation failures
	result := cleanRunner.Process(ctx, doccleanrunner.Body{DocumentID: docID})
	assert.True(t, result, "validation failures should be acknowledged (not requeued)")

	// Verify outcome updated to clean_failed
	rows, err := cfg.GetDBQueries().ListBatchDocumentOutcomes(ctx, batchID)
	require.NoError(t, err)
	require.Len(t, rows, 1)
	assert.Equal(t, repository.BatchOutcomeStatusCleanFailed, rows[0].Outcome)
}

// TestDocCleanRunner_RequeuesOnInfraError tests that an infrastructure error
// (e.g., document not found in DB) causes the runner to return false (requeue).
func TestDocCleanRunner_RequeuesOnInfraError(t *testing.T) {
	cfg := &DocCleanConfig{}
	test.CreateDB(t, cfg)
	ctx := t.Context()

	cleanRunner := doccleanrunner.New(&doccleanrunner.Services{
		Clean:   documentclean.New(cfg),
		Outcome: outcome.New(cfg.GetDBQueries()),
	})

	// Process a non-existent document -- should fail with DB error
	result := cleanRunner.Process(ctx, doccleanrunner.Body{DocumentID: uuid.New()})
	assert.False(t, result, "infrastructure error should return false for requeue")
}