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