package doccleanrunner import ( "context" "log/slog" "queryorchestration/internal/database/repository" "queryorchestration/internal/document/batch/foldercleanup" "queryorchestration/internal/document/batch/outcome" documentclean "queryorchestration/internal/document/clean" "github.com/google/uuid" ) const Name = "docCleanRunner" // Services holds the dependencies for the docCleanRunner. // // Fields: // - Clean: service for cleaning/validating documents // - Outcome: reporter for writing batch pipeline outcomes type Services struct { Clean *documentclean.Service Outcome *outcome.Reporter Cleanup *foldercleanup.Service } // Runner processes document clean messages from the SQS queue. type Runner struct { svc *Services } // New creates a new docCleanRunner. // // Parameters: // - svc: the runner's service dependencies // // Returns: // - Runner: the runner instance func New(svc *Services) Runner { return Runner{ svc: svc, } } // Body is the SQS message body for docCleanRunner. type Body struct { DocumentID uuid.UUID `json:"id" validate:"required,uuid"` } // Process handles a single document clean message. Returns true to acknowledge // the message (success or validation failure), or false to requeue (infrastructure error). // Validation failures are acknowledged (not requeued) to prevent infinite retry loops // for permanently-invalid documents. func (s Runner) Process(ctx context.Context, body Body) bool { result, err := s.svc.Clean.Clean(ctx, body.DocumentID) if err != nil { slog.Error("unable to process", "error", err) return false // infrastructure error -> requeue } outcomeName := repository.BatchOutcomeStatusCleanPassed var errorDetail *string if !result.Passed { outcomeName = repository.BatchOutcomeStatusCleanFailed errorDetail = result.FailReason } batchIDs, updateErr := s.svc.Outcome.Update(ctx, body.DocumentID, outcomeName, errorDetail) if updateErr != nil { slog.Error("failed to update batch outcome", "document_id", body.DocumentID, "outcome", outcomeName, "error", updateErr) } if s.svc.Cleanup != nil { for _, batchID := range batchIDs { if _, err := s.svc.Cleanup.TryFinalizeBatch(ctx, batchID); err != nil { slog.Error("failed to finalize batch folder cleanup", "batch_id", batchID, "error", err) } } } return true // validation pass or fail -> acknowledge (don't requeue bad PDFs) }