Add photo content-hash dedup and local↔S3 library sync.
Uploads skip per-gallery duplicates, checksums can be backfilled, and STORAGE_BACKEND plus sync tooling make switching storage backends safe.
This commit is contained in:
Michilis
@@ -0,0 +1,229 @@
|
||||
// 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])
|
||||
}
|
||||
Reference in New Issue
Block a user