Jacob Mathison
a080ca59d8
Implement v2 upload batch cleanup * Implement v2 upload batch cleanup * Merge remote-tracking branch 'origin/main' into jmathison/v2-upload-batch * Address upload batch review feedback * Raise batch worker coverage * Fix batch cleanup review issues Approved-by: Jay Brown
79 lines
2.1 KiB
Go
79 lines
2.1 KiB
Go
package docsyncrunner
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
|
|
"queryorchestration/internal/database/repository"
|
|
"queryorchestration/internal/document/batch/foldercleanup"
|
|
"queryorchestration/internal/document/batch/outcome"
|
|
documentsync "queryorchestration/internal/document/sync"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
const Name = "docSyncRunner"
|
|
|
|
// Services holds the dependencies for the docSyncRunner.
|
|
//
|
|
// Fields:
|
|
// - Document: service for syncing documents
|
|
// - Outcome: reporter for writing batch pipeline outcomes
|
|
type Services struct {
|
|
Document *documentsync.Service
|
|
Outcome *outcome.Reporter
|
|
Cleanup *foldercleanup.Service
|
|
}
|
|
|
|
// Runner processes document sync messages from the SQS queue.
|
|
type Runner struct {
|
|
svc *Services
|
|
}
|
|
|
|
// New creates a new docSyncRunner.
|
|
//
|
|
// 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 docSyncRunner.
|
|
type Body struct {
|
|
DocumentID uuid.UUID `json:"id" validate:"required,uuid"`
|
|
}
|
|
|
|
// Process handles a single document sync message. Returns true to acknowledge
|
|
// the message (success), or false to requeue (infrastructure error).
|
|
func (s Runner) Process(ctx context.Context, body Body) bool {
|
|
result, err := s.svc.Document.Sync(ctx, body.DocumentID)
|
|
if err != nil {
|
|
slog.Error("unable to process", "error", err)
|
|
return false
|
|
}
|
|
|
|
outcomeName := repository.BatchOutcomeStatusSyncComplete
|
|
if result.Skipped {
|
|
outcomeName = repository.BatchOutcomeStatusSyncSkipped
|
|
}
|
|
batchIDs, updateErr := s.svc.Outcome.Update(ctx, body.DocumentID, outcomeName, nil)
|
|
if updateErr != nil {
|
|
slog.Error("failed to update batch outcome",
|
|
"document_id", body.DocumentID, "outcome", outcomeName, "error", updateErr)
|
|
}
|
|
if outcomeName == repository.BatchOutcomeStatusSyncSkipped && 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
|
|
}
|