63 lines
1.5 KiB
Go
63 lines
1.5 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
filemanager "wucher/internal/domain/file_manager"
|
|
)
|
|
|
|
type FileProcessingJobProcessor interface {
|
|
Process(ctx context.Context, job filemanager.FileProcessingJob, correlationID string) error
|
|
}
|
|
|
|
type FileProcessingPermanentClassifier func(error) bool
|
|
|
|
type FileProcessingJobQueueHandler struct {
|
|
processor FileProcessingJobProcessor
|
|
isPermanent FileProcessingPermanentClassifier
|
|
}
|
|
|
|
func NewFileProcessingJobQueueHandler(
|
|
processor FileProcessingJobProcessor,
|
|
isPermanent FileProcessingPermanentClassifier,
|
|
) *FileProcessingJobQueueHandler {
|
|
return &FileProcessingJobQueueHandler{
|
|
processor: processor,
|
|
isPermanent: isPermanent,
|
|
}
|
|
}
|
|
|
|
func (h *FileProcessingJobQueueHandler) Handle(ctx context.Context, envelope *FileProcessingEnvelope) error {
|
|
if envelope == nil {
|
|
return Permanent(errors.New("file processing envelope is required"))
|
|
}
|
|
if h.processor == nil {
|
|
return Permanent(errors.New("file processing processor is required"))
|
|
}
|
|
|
|
job := envelope.Payload.Canonical()
|
|
if err := job.Validate(); err != nil {
|
|
return Permanent(err)
|
|
}
|
|
|
|
err := h.processor.Process(ctx, filemanager.FileProcessingJob{
|
|
JobID: job.JobID,
|
|
FileUUID: job.FileUUID,
|
|
Bucket: job.Bucket,
|
|
ObjectKey: job.ObjectKey,
|
|
UploadedBy: job.UploadedBy,
|
|
MimeType: job.MimeType,
|
|
SizeBytes: job.SizeBytes,
|
|
TraceID: job.TraceID,
|
|
RequestID: job.RequestID,
|
|
}, envelope.CorrelationID)
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if h.isPermanent != nil && h.isPermanent(err) {
|
|
return Permanent(err)
|
|
}
|
|
return Transient(err)
|
|
}
|