init

backend/internal/downloads/service.go

@@ -0,0 +1,185 @@
+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)
+}