Spanglish/photo-api/internal/photosync/photosync.go

// Package photosync copies the photo library between the local-disk and the
// S3 backend so the active one (STORAGE_BACKEND) can be switched without
// losing photos. Both backends must be configured in photo-api/.env for a
// sync to run — the direction picks which is the source.
//
// The photos_photos rows are the inventory: every row contributes its
// original key and, once the worker has processed it, its thumb and preview
// key. Objects on disk or in the bucket with no row (worker scratch dirs,
// leftovers of deleted galleries) are deliberately not copied.
//
// Sync only ever writes to the destination. The source is left untouched, so
// a sync is repeatable, safe to interrupt, and leaves the old backend as a
// fallback until it is cleaned up by hand.
package photosync

import (
	"context"
	"errors"
	"fmt"
	"log"
	"sync"
	"sync/atomic"

	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/config"
	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/storage"
	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/store"
)

// Direction is which way objects move.
type Direction string

const (
	ToS3    Direction = "to-s3"
	ToLocal Direction = "to-local"
)

// ParseDirection accepts the short and the spelled-out form.
func ParseDirection(s string) (Direction, error) {
	switch s {
	case "to-s3", "local-to-s3":
		return ToS3, nil
	case "to-local", "s3-to-local":
		return ToLocal, nil
	default:
		return "", fmt.Errorf("unknown direction %q (expected to-s3 or to-local)", s)
	}
}

type Options struct {
	Direction Direction
	// GalleryID limits the sync to one gallery; empty means the whole library.
	GalleryID string
	// Concurrency is how many objects are copied at a time.
	Concurrency int
	// DryRun reports what would be copied without writing anything.
	DryRun bool
	// Overwrite re-copies objects that already exist on the destination with
	// the same size (default: those are skipped, which makes reruns cheap).
	Overwrite bool
}

type Result struct {
	Total   int   // objects in the inventory
	Copied  int   // copied (or, with DryRun, would be copied)
	Skipped int   // already on the destination
	Missing int   // absent from the source — nothing to copy
	Failed  int   // copy attempted and errored
	Bytes   int64 // bytes copied
}

// Run copies the library in the requested direction. It returns a Result even
// on error, and an error if any object failed.
func Run(ctx context.Context, cfg config.Config, db *store.DB, opts Options) (Result, error) {
	if !cfg.S3Configured() {
		return Result{}, errors.New("sync needs both backends configured: set S3_ENDPOINT, S3_BUCKET, " +
			"S3_ACCESS_KEY_ID and S3_SECRET_ACCESS_KEY (next to STORAGE_PATH) in photo-api/.env")
	}
	if opts.Concurrency < 1 {
		opts.Concurrency = 4
	}

	local, err := storage.NewLocal(cfg.StoragePath)
	if err != nil {
		return Result{}, fmt.Errorf("local storage: %w", err)
	}
	s3, err := storage.NewS3(cfg)
	if err != nil {
		return Result{}, fmt.Errorf("s3 storage: %w", err)
	}
	src, dst := local, s3
	srcName, dstName := "local disk "+cfg.StoragePath, "s3 bucket "+cfg.S3Bucket
	if opts.Direction == ToLocal {
		src, dst = s3, local
		srcName, dstName = dstName, srcName
	}

	objects, err := db.AllPhotoObjects(ctx, opts.GalleryID)
	if err != nil {
		return Result{}, fmt.Errorf("list photo objects: %w", err)
	}

	prefix := ""
	if opts.DryRun {
		prefix = "[dry-run] "
	}
	log.Printf("sync: %s%s → %s: %d objects, concurrency %d",
		prefix, srcName, dstName, len(objects), opts.Concurrency)

	var (
		mu   sync.Mutex
		res  = Result{Total: len(objects)}
		done int64
		jobs = make(chan store.PhotoObject)
		wg   sync.WaitGroup
	)
	for i := 0; i < opts.Concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				if ctx.Err() != nil {
					return
				}
				act, size, err := copyObject(ctx, src, dst, o, opts)
				n := atomic.AddInt64(&done, 1)

				mu.Lock()
				switch {
				case err != nil:
					res.Failed++
					log.Printf("sync: [%d/%d] FAILED %s (photo %s %s): %v",
						n, len(objects), o.Key, o.PhotoID, o.Variant, err)
				case act == actionCopied:
					res.Copied++
					res.Bytes += size
					log.Printf("sync: [%d/%d] %scopied %s (%s)", n, len(objects), prefix, o.Key, humanBytes(size))
				case act == actionMissing:
					res.Missing++
					log.Printf("sync: [%d/%d] MISSING on source, skipped: %s (photo %s %s)",
						n, len(objects), o.Key, o.PhotoID, o.Variant)
				default:
					res.Skipped++
				}
				mu.Unlock()
			}
		}()
	}
	for _, o := range objects {
		select {
		case jobs <- o:
		case <-ctx.Done():
		}
		if ctx.Err() != nil {
			break
		}
	}
	close(jobs)
	wg.Wait()

	log.Printf("sync: %sdone — %d copied (%s), %d already present, %d missing on source, %d failed",
		prefix, res.Copied, humanBytes(res.Bytes), res.Skipped, res.Missing, res.Failed)

	if err := ctx.Err(); err != nil {
		return res, fmt.Errorf("interrupted after %d/%d objects: %w", res.Copied+res.Skipped, res.Total, err)
	}
	if res.Failed > 0 {
		return res, fmt.Errorf("%d of %d objects failed to copy (rerun to retry; already-copied objects are skipped)",
			res.Failed, res.Total)
	}
	return res, nil
}

type action int

const (
	actionCopied action = iota
	actionSkipped
	actionMissing
)

func copyObject(ctx context.Context, src, dst storage.Storage, o store.PhotoObject, opts Options) (action, int64, error) {
	srcSize, err := src.Stat(ctx, o.Key)
	if errors.Is(err, storage.ErrNotExist) {
		return actionMissing, 0, nil
	}
	if err != nil {
		return actionSkipped, 0, fmt.Errorf("stat source: %w", err)
	}
	if !opts.Overwrite {
		if dstSize, err := dst.Stat(ctx, o.Key); err == nil && dstSize == srcSize {
			return actionSkipped, 0, nil
		} else if err != nil && !errors.Is(err, storage.ErrNotExist) {
			return actionSkipped, 0, fmt.Errorf("stat destination: %w", err)
		}
	}
	if opts.DryRun {
		return actionCopied, srcSize, nil
	}

	r, size, err := src.Open(ctx, o.Key)
	if err != nil {
		return actionSkipped, 0, fmt.Errorf("read source: %w", err)
	}
	defer r.Close()
	if size <= 0 {
		size = srcSize // local Open reports the stat size; be defensive anyway
	}
	contentType := o.ContentType
	if contentType == "" {
		contentType = "application/octet-stream"
	}
	if err := dst.Put(ctx, o.Key, r, size, contentType); err != nil {
		return actionSkipped, 0, fmt.Errorf("write destination: %w", err)
	}
	return actionCopied, size, nil
}

func humanBytes(n int64) string {
	const unit = 1024
	if n < unit {
		return fmt.Sprintf("%d B", n)
	}
	v, exp := float64(n), 0
	for v >= unit && exp < 4 {
		v /= unit
		exp++
	}
	return fmt.Sprintf("%.1f %ciB", v, "KMGT"[exp-1])
}