feat(t240): cache freight item images
This commit is contained in:
QiuSW
@@ -0,0 +1,270 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"cmroubao/backend-api/internal/domain"
|
||||
)
|
||||
|
||||
const (
|
||||
maxFreightImageJobsPerRun = 1000
|
||||
freightImageWorkers = 4
|
||||
freightImageCleanup = 5 * time.Second
|
||||
)
|
||||
|
||||
type FreightImageService struct {
|
||||
repository FreightImageRepository
|
||||
source FreightImageSource
|
||||
store ReferenceImageStore
|
||||
clock Clock
|
||||
}
|
||||
|
||||
type FreightItemImageContent struct {
|
||||
Image domain.FreightItemImage
|
||||
Content io.ReadCloser
|
||||
}
|
||||
|
||||
func NewFreightImageService(
|
||||
repository FreightImageRepository,
|
||||
source FreightImageSource,
|
||||
store ReferenceImageStore,
|
||||
clock Clock,
|
||||
) (*FreightImageService, error) {
|
||||
if repository == nil || source == nil || store == nil || clock == nil {
|
||||
return nil, errors.New("freight image service dependencies are required")
|
||||
}
|
||||
return &FreightImageService{
|
||||
repository: repository,
|
||||
source: source,
|
||||
store: store,
|
||||
clock: clock,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (service *FreightImageService) CacheRun(
|
||||
ctx context.Context,
|
||||
creatorSubject, runID string,
|
||||
) error {
|
||||
jobs, err := service.repository.ListFreightImageJobs(
|
||||
ctx,
|
||||
creatorSubject,
|
||||
runID,
|
||||
maxFreightImageJobsPerRun,
|
||||
)
|
||||
if err != nil {
|
||||
return wrapRepositoryError(err)
|
||||
}
|
||||
if len(jobs) == 0 {
|
||||
return nil
|
||||
}
|
||||
workerCount := freightImageWorkers
|
||||
if len(jobs) < workerCount {
|
||||
workerCount = len(jobs)
|
||||
}
|
||||
queue := make(chan domain.FreightItemImageJob)
|
||||
var workers sync.WaitGroup
|
||||
workers.Add(workerCount)
|
||||
for range workerCount {
|
||||
go func() {
|
||||
defer workers.Done()
|
||||
for job := range queue {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
service.cacheJob(ctx, job)
|
||||
}
|
||||
}()
|
||||
}
|
||||
for _, job := range jobs {
|
||||
select {
|
||||
case queue <- job:
|
||||
case <-ctx.Done():
|
||||
close(queue)
|
||||
workers.Wait()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
close(queue)
|
||||
workers.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (service *FreightImageService) cacheJob(
|
||||
ctx context.Context,
|
||||
job domain.FreightItemImageJob,
|
||||
) {
|
||||
sourceImage, err := service.source.FetchProductImage(
|
||||
ctx,
|
||||
job.ProductThumbRef,
|
||||
)
|
||||
if err != nil {
|
||||
if ctx.Err() == nil {
|
||||
service.saveFailure(ctx, job, freightImageSourceFailure(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
defer sourceImage.Content.Close()
|
||||
normalized, err := service.store.Put(
|
||||
ctx,
|
||||
job.FreightOrderItemID,
|
||||
sourceImage.MediaType,
|
||||
sourceImage.Content,
|
||||
)
|
||||
if err != nil {
|
||||
if ctx.Err() == nil {
|
||||
service.saveFailure(ctx, job, freightImageStoreFailure(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
candidate := domain.FreightItemImage{
|
||||
CreatorSubject: job.CreatorSubject,
|
||||
FreightOrderItemID: job.FreightOrderItemID,
|
||||
ProductThumbRef: job.ProductThumbRef,
|
||||
Status: domain.FreightItemImageReady,
|
||||
MediaType: normalized.MediaType,
|
||||
SizeBytes: normalized.SizeBytes,
|
||||
SHA256: normalized.SHA256,
|
||||
StorageKey: normalized.StorageKey,
|
||||
UpdatedAt: service.clock.Now().UTC(),
|
||||
}
|
||||
replaced, err := service.repository.SaveFreightItemImage(ctx, candidate)
|
||||
if err != nil {
|
||||
service.deleteStoredImage(normalized.StorageKey)
|
||||
return
|
||||
}
|
||||
if replaced != nil {
|
||||
service.deleteStoredImage(*replaced)
|
||||
}
|
||||
}
|
||||
|
||||
func (service *FreightImageService) saveFailure(
|
||||
ctx context.Context,
|
||||
job domain.FreightItemImageJob,
|
||||
failure freightImageFailure,
|
||||
) {
|
||||
errorCode := failure.code
|
||||
replaced, err := service.repository.SaveFreightItemImage(
|
||||
ctx,
|
||||
domain.FreightItemImage{
|
||||
CreatorSubject: job.CreatorSubject,
|
||||
FreightOrderItemID: job.FreightOrderItemID,
|
||||
ProductThumbRef: job.ProductThumbRef,
|
||||
Status: failure.status,
|
||||
ErrorCode: &errorCode,
|
||||
UpdatedAt: service.clock.Now().UTC(),
|
||||
},
|
||||
)
|
||||
if err == nil && replaced != nil {
|
||||
service.deleteStoredImage(*replaced)
|
||||
}
|
||||
}
|
||||
|
||||
func (service *FreightImageService) deleteStoredImage(storageKey string) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), freightImageCleanup)
|
||||
defer cancel()
|
||||
_ = service.store.Delete(ctx, storageKey)
|
||||
}
|
||||
|
||||
func (service *FreightImageService) OpenItemImage(
|
||||
ctx context.Context,
|
||||
creatorSubject, itemID string,
|
||||
) (FreightItemImageContent, error) {
|
||||
if creatorSubject == "" || !isUUID(itemID) {
|
||||
return FreightItemImageContent{}, freightImageNotFoundError(nil)
|
||||
}
|
||||
image, err := service.repository.GetReadyFreightItemImage(
|
||||
ctx,
|
||||
creatorSubject,
|
||||
itemID,
|
||||
)
|
||||
if err != nil {
|
||||
wrapped := wrapRepositoryError(err)
|
||||
var typed *Error
|
||||
if errors.As(wrapped, &typed) &&
|
||||
typed.Kind == ErrorKindNotFound {
|
||||
return FreightItemImageContent{},
|
||||
freightImageNotFoundError(err)
|
||||
}
|
||||
return FreightItemImageContent{}, wrapped
|
||||
}
|
||||
content, err := service.store.Open(ctx, image.StorageKey)
|
||||
if err != nil {
|
||||
var storeError *ImageStoreError
|
||||
if errors.As(err, &storeError) &&
|
||||
storeError.Kind == ImageStoreErrorNotFound {
|
||||
return FreightItemImageContent{},
|
||||
freightImageNotFoundError(err)
|
||||
}
|
||||
return FreightItemImageContent{}, mapImageStoreError(err)
|
||||
}
|
||||
return FreightItemImageContent{Image: image, Content: content}, nil
|
||||
}
|
||||
|
||||
type freightImageFailure struct {
|
||||
status domain.FreightItemImageStatus
|
||||
code string
|
||||
}
|
||||
|
||||
func freightImageSourceFailure(err error) freightImageFailure {
|
||||
switch {
|
||||
case errors.Is(err, domain.ErrFreightImageNotFound):
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageMissing,
|
||||
code: "ERP_IMAGE_NOT_FOUND",
|
||||
}
|
||||
case errors.Is(err, domain.ErrFreightImageInvalid):
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "ERP_IMAGE_INVALID",
|
||||
}
|
||||
case errors.Is(err, domain.ErrFreightSourceSessionNeeded):
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "ERP_SESSION_REQUIRED",
|
||||
}
|
||||
case errors.Is(err, domain.ErrFreightSourceNotConfigured):
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "ERP_NOT_CONFIGURED",
|
||||
}
|
||||
default:
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "ERP_IMAGE_UNAVAILABLE",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func freightImageStoreFailure(err error) freightImageFailure {
|
||||
var storeError *ImageStoreError
|
||||
if errors.As(err, &storeError) {
|
||||
switch storeError.Kind {
|
||||
case ImageStoreErrorTooLarge,
|
||||
ImageStoreErrorUnsupported,
|
||||
ImageStoreErrorInvalid:
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "ERP_IMAGE_INVALID",
|
||||
}
|
||||
}
|
||||
}
|
||||
return freightImageFailure{
|
||||
status: domain.FreightItemImageFailed,
|
||||
code: "IMAGE_STORE_UNAVAILABLE",
|
||||
}
|
||||
}
|
||||
|
||||
func freightImageNotFoundError(cause error) error {
|
||||
return newError(
|
||||
ErrorKindNotFound,
|
||||
"FREIGHT_IMAGE_NOT_FOUND",
|
||||
"freight item image not found",
|
||||
cause,
|
||||
)
|
||||
}
|
||||
|
||||
var _ FreightImageCache = (*FreightImageService)(nil)
|
||||
Reference in New Issue
Block a user