cmroubao/backend-api/internal/usecase/freight_image_service.go

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)