wijtransferen/backend/internal/downloads/service.go

package downloads

import (
	"archive/zip"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/example/sndit/backend/internal/media"
	"github.com/example/sndit/backend/internal/storage"
	"github.com/google/uuid"
)

type Service struct {
	repository *Repository
	media      *media.Repository
	storage    storage.Storage
	jobs       chan uuid.UUID
	stop       chan struct{}
	waitGroup  sync.WaitGroup
}

var placeholderClient = &http.Client{Timeout: time.Minute}

func NewService(repository *Repository, mediaRepository *media.Repository, objectStorage storage.Storage, workers int) *Service {
	if workers < 1 {
		workers = 1
	}
	service := &Service{
		repository: repository,
		media:      mediaRepository,
		storage:    objectStorage,
		jobs:       make(chan uuid.UUID, 32),
		stop:       make(chan struct{}),
	}
	for index := 0; index < workers; index++ {
		service.waitGroup.Add(1)
		go service.worker()
	}
	return service
}

func (s *Service) Create(ctx context.Context, galleryID uuid.UUID, visitorID string) (Job, error) {
	job, err := s.repository.Create(ctx, galleryID, visitorID)
	if err != nil {
		return Job{}, err
	}
	select {
	case s.jobs <- job.ID:
	case <-s.stop:
		return Job{}, fmt.Errorf("download service is stopping")
	}
	return job, nil
}

func (s *Service) Get(ctx context.Context, jobID, galleryID uuid.UUID, visitorID string) (Job, error) {
	return s.repository.GetForVisitor(ctx, jobID, galleryID, visitorID)
}

func (s *Service) Close() {
	close(s.stop)
	s.waitGroup.Wait()
}

func (s *Service) worker() {
	defer s.waitGroup.Done()
	for {
		select {
		case jobID := <-s.jobs:
			s.process(jobID)
		case <-s.stop:
			return
		}
	}
}

func (s *Service) process(jobID uuid.UUID) {
	ctx := context.Background()
	if err := s.repository.MarkProcessing(ctx, jobID); err != nil {
		return
	}

	var job Job
	// The worker needs the gallery ID and visitor only for status storage. The
	// job lookup below is intentionally not visitor-scoped because the ID is
	// generated internally and never exposed before creation succeeds.
	job, err := s.repository.Get(ctx, jobID)
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	items, err := s.media.ListByGallery(ctx, job.GalleryID)
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}

	temporary, err := os.CreateTemp("", "gallery-download-*.zip")
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	temporaryPath := temporary.Name()
	defer os.Remove(temporaryPath)

	archive := zip.NewWriter(temporary)
	for _, item := range items {
		if item.ProcessingStatus != media.StatusReady {
			continue
		}
		object, err := s.openItem(ctx, item)
		if err != nil {
			_ = archive.Close()
			_ = temporary.Close()
			_ = s.repository.MarkFailed(ctx, jobID, err.Error())
			return
		}
		entry, err := archive.Create(filepath.Base(item.OriginalFilename))
		if err == nil {
			_, err = io.Copy(entry, object)
		}
		_ = object.Close()
		if err != nil {
			_ = archive.Close()
			_ = temporary.Close()
			_ = s.repository.MarkFailed(ctx, jobID, err.Error())
			return
		}
	}
	if err := archive.Close(); err != nil {
		_ = temporary.Close()
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	if err := temporary.Close(); err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	fileInfo, err := os.Stat(temporaryPath)
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	key := fmt.Sprintf("galleries/%s/downloads/%s.zip", job.GalleryID, job.ID)
	file, err := os.Open(temporaryPath)
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	err = s.storage.Put(ctx, key, file, fileInfo.Size(), "application/zip")
	_ = file.Close()
	if err != nil {
		_ = s.repository.MarkFailed(ctx, jobID, err.Error())
		return
	}
	_ = s.repository.MarkReady(ctx, jobID, key)
}

func (s *Service) openItem(ctx context.Context, item media.Record) (io.ReadCloser, error) {
	if item.ExternalURL != "" {
		request, err := http.NewRequestWithContext(ctx, http.MethodGet, item.ExternalURL, nil)
		if err != nil {
			return nil, err
		}
		response, err := placeholderClient.Do(request)
		if err != nil {
			return nil, err
		}
		if response.StatusCode >= http.StatusBadRequest {
			_ = response.Body.Close()
			return nil, fmt.Errorf("download placeholder returned %s", response.Status)
		}
		return response.Body, nil
	}
	if strings.TrimSpace(item.StorageKey) == "" {
		return nil, fmt.Errorf("media has no storage object")
	}
	return s.storage.Get(ctx, item.StorageKey)
}