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) }