feat(video): add video upload with remux/transcode playback

Add video as a media kind in photos: streaming upload with per-kind limits, ffprobe metadata, poster frame reusing the image variant pipeline, and a playback object (remux when web-compatible, else h264/yuv420p/aac transcode). Video jobs run on a separate pool with a long lease. Frontend gains a video viewer, grid badge/duration and video upload.
2026-10-11 20:37:53 +03:00
parent d059d042f0
commit d38ede0201
52 changed files with 2717 additions and 82 deletions
@@ -3,16 +3,29 @@ package photos
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
+	"io"
 	"log/slog"
+	"os"
 	"sync"
 	"time"

 	"galo/internal/auth"
+	"galo/internal/video"
 )

 const maxVariantAttempts = 5

+// videoJobLease is the processing lease for a video job. It is far longer than
+// the image lease because a transcode can run for minutes, and must outlast the
+// worst-case processing so a job is never claimed twice concurrently.
+const videoJobLease = 30 * time.Minute
+
+// videoJobTimeout bounds a single ffmpeg/ffprobe invocation.
+const videoJobTimeout = 20 * time.Minute
+
 // Worker is the background reconciler. It drives the variant generation queue
 // and the object garbage collector entirely from the state recorded in the
 // metadata layer, so it resumes correctly after a restart.
@@ -23,6 +36,11 @@ type Worker struct {
 	pendingTTL  time.Duration
 	lease       time.Duration
 	concurrency int
+
+	tools      *video.Toolchain
+	videoConc  int
+	videoLease time.Duration
+	tempDir    string
 }

 // NewWorker builds a worker. Parameters are clamped to sane defaults.
@@ -46,9 +64,23 @@ func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.
 		pendingTTL:  pendingTTL,
 		lease:       interval * 4,
 		concurrency: concurrency,
+		videoConc:   1,
+		videoLease:  videoJobLease,
 	}
 }

+// SetVideo enables video processing with the given toolchain and a separate,
+// small transcode concurrency. A nil or unavailable toolchain leaves video jobs
+// untouched.
+func (w *Worker) SetVideo(tools *video.Toolchain, concurrency int) {
+	w.tools = tools
+	if concurrency > 0 {
+		w.videoConc = concurrency
+	}
+	w.videoLease = videoJobLease
+	w.tempDir = tools.TempDir()
+}
+
 // Run loops until the context is cancelled.
 func (w *Worker) Run(ctx context.Context) {
 	slog.Info("background worker started", "interval", w.interval)
@@ -65,14 +97,21 @@ func (w *Worker) Run(ctx context.Context) {
 	}
 }

+// ProcessOnce runs a single round of all queues. It is used by tests and manual
+// reconciliation and does not block.
+func (w *Worker) ProcessOnce(ctx context.Context) {
+	w.tick(ctx)
+}
+
 func (w *Worker) tick(ctx context.Context) {
 	w.processVariantJobs(ctx)
+	w.processVideoJobs(ctx)
 	w.collectGarbage(ctx)
 	w.reapStalePending(ctx)
 }

 func (w *Worker) processVariantJobs(ctx context.Context) {
-	ids, err := w.svc.store.ClaimVariantJobs(ctx, w.batch, w.lease)
+	ids, err := w.svc.store.ClaimVariantJobs(ctx, KindImage, w.batch, w.lease)
 	if err != nil {
 		slog.Error("claim variant jobs", "error", err)
 		return
@@ -82,6 +121,26 @@ func (w *Worker) processVariantJobs(ctx context.Context) {
 	})
 }

+// processVideoJobs claims and processes video jobs on a separate, small pool
+// with a long lease so transcoding does not starve image variant generation.
+func (w *Worker) processVideoJobs(ctx context.Context) {
+	if w.tools == nil || !w.tools.Available() {
+		return
+	}
+	ids, err := w.svc.store.ClaimVariantJobs(ctx, KindVideo, w.batch, w.videoLease)
+	if err != nil {
+		slog.Error("claim video jobs", "error", err)
+		return
+	}
+	if len(ids) == 0 {
+		return
+	}
+	pool := &Worker{concurrency: w.videoConc}
+	pool.forEachConcurrent(ids, func(id string) {
+		w.generateVideo(ctx, id)
+	})
+}
+
 // forEachConcurrent runs fn over ids with at most w.concurrency functions in
 // flight. Generation is CPU-bound (decode + resize + encode), so parallelism
 // scales with cores; a concurrency of 1 stays strictly sequential.
@@ -152,22 +211,28 @@ func (w *Worker) generateVariants(ctx context.Context, photoID string) {
 	}
 }

-// createVariant stores one derived object and binds it to every photo sharing
-// the original, so copies reuse the same physical variant.
+// createVariant stores one derived image object and binds it to every photo
+// sharing the original, so copies reuse the same physical variant.
 func (w *Worker) createVariant(ctx context.Context, photoIDs []string, role string, data []byte) error {
+	return w.createObject(ctx, photoIDs, role, ".webp", "image/webp", int64(len(data)), checksum(data), bytes.NewReader(data))
+}
+
+// createObject stores one derived object and binds it to every photo sharing
+// the original.
+func (w *Worker) createObject(ctx context.Context, photoIDs []string, role, ext, mimeType string, size int64, sum string, body io.Reader) error {
 	objID := auth.NewID()
-	key := objectKey(objID, role, ".webp")
+	key := objectKey(objID, role, ext)
 	obj := &Object{
 		ID:         objID,
 		StorageKey: key,
-		MimeType:   "image/webp",
-		Size:       int64(len(data)),
-		Checksum:   checksum(data),
+		MimeType:   mimeType,
+		Size:       size,
+		Checksum:   sum,
 	}
 	if err := w.svc.store.CreateObject(ctx, obj); err != nil {
 		return err
 	}
-	if err := w.svc.storage.Put(ctx, key, bytes.NewReader(data), int64(len(data)), "image/webp"); err != nil {
+	if err := w.svc.storage.Put(ctx, key, body, size, mimeType); err != nil {
 		_ = w.svc.store.DeleteObjectRow(ctx, objID)
 		return err
 	}
@@ -191,6 +256,193 @@ func (w *Worker) createVariant(ctx context.Context, photoIDs []string, role stri
 	return nil
 }

+// generateVideo prepares a video job: probe, poster frame (reusing the image
+// variant pipeline) and a web-friendly playback variant, then records the
+// probed facts and notifies subscribers.
+func (w *Worker) generateVideo(ctx context.Context, photoID string) {
+	orig, err := w.svc.store.OriginalObject(ctx, photoID)
+	if err != nil {
+		_ = w.svc.store.CompleteVariantJob(ctx, photoID)
+		return
+	}
+	if orig.State != ObjectReady {
+		w.fail(ctx, photoID, "original object is not ready")
+		return
+	}
+	inPath, cleanup, err := w.spoolOriginal(ctx, orig.StorageKey)
+	if err != nil {
+		w.fail(ctx, photoID, "spool original: "+err.Error())
+		return
+	}
+	defer cleanup()
+
+	info, err := video.Probe(ctx, w.tools.FFprobePath(), inPath)
+	if err != nil {
+		w.fail(ctx, photoID, "probe: "+err.Error())
+		return
+	}
+
+	siblings, err := w.svc.store.PhotosByOriginalObject(ctx, orig.ID)
+	if err != nil || len(siblings) == 0 {
+		siblings = []string{photoID}
+	}
+
+	if err := w.storePoster(ctx, inPath, siblings); err != nil {
+		w.fail(ctx, photoID, "poster: "+err.Error())
+		return
+	}
+	if err := w.storePlayback(ctx, inPath, info, siblings); err != nil {
+		w.fail(ctx, photoID, "playback: "+err.Error())
+		return
+	}
+	if err := w.svc.store.SetMediaMeta(ctx, photoID, orig.ID, info.DurationMS, info.Width, info.Height); err != nil {
+		w.fail(ctx, photoID, "save media meta: "+err.Error())
+		return
+	}
+
+	_ = w.svc.store.CompleteVariantJob(ctx, photoID)
+	for _, pid := range siblings {
+		w.notify(ctx, pid)
+	}
+}
+
+// spoolOriginal streams the original blob into a temporary file so ffmpeg reads
+// a seekable input. The returned cleanup removes the file.
+func (w *Worker) spoolOriginal(ctx context.Context, key string) (string, func(), error) {
+	src, err := w.svc.storage.Open(ctx, key)
+	if err != nil {
+		return "", nil, err
+	}
+	defer src.Close()
+	f, err := os.CreateTemp(w.tempDir, "video-in-*")
+	if err != nil {
+		return "", nil, err
+	}
+	name := f.Name()
+	cleanup := func() { _ = os.Remove(name) }
+	if _, err := io.Copy(f, src); err != nil {
+		_ = f.Close()
+		cleanup()
+		return "", nil, err
+	}
+	if err := f.Close(); err != nil {
+		cleanup()
+		return "", nil, err
+	}
+	return name, cleanup, nil
+}
+
+// storePoster extracts a frame, retrying at the start for videos whose first
+// second is unavailable, then generates the image variants from it.
+func (w *Worker) storePoster(ctx context.Context, inPath string, siblings []string) error {
+	pf, err := os.CreateTemp(w.tempDir, "video-poster-*.jpg")
+	if err != nil {
+		return err
+	}
+	posterPath := pf.Name()
+	_ = pf.Close()
+	defer func() { _ = os.Remove(posterPath) }()
+
+	// Seek to 1s for a representative frame, falling back to the first frame
+	// when that yields no image (for example a clip shorter than one second).
+	if err := w.runPoster(ctx, inPath, posterPath, 1); err != nil {
+		if err := w.runPoster(ctx, inPath, posterPath, 0); err != nil {
+			return err
+		}
+	}
+	poster, err := os.ReadFile(posterPath)
+	if err != nil {
+		return err
+	}
+	if len(poster) == 0 {
+		return errors.New("poster frame is empty")
+	}
+	thumb, medium, preview, err := buildVariants(poster)
+	if err != nil {
+		return err
+	}
+	if err := w.createVariant(ctx, siblings, RoleThumb, thumb); err != nil {
+		return err
+	}
+	if err := w.createVariant(ctx, siblings, RoleMedium, medium); err != nil {
+		return err
+	}
+	return w.createVariant(ctx, siblings, RolePreview, preview)
+}
+
+// runPoster extracts a poster frame at the given second and reports an error
+// when ffmpeg fails or produces no image.
+func (w *Worker) runPoster(ctx context.Context, inPath, outPath string, atSecond int) error {
+	if err := w.runFFmpeg(ctx, video.PosterArgs(inPath, outPath, atSecond)); err != nil {
+		return err
+	}
+	info, err := os.Stat(outPath)
+	if err != nil {
+		return err
+	}
+	if info.Size() == 0 {
+		return errors.New("poster frame is empty")
+	}
+	return nil
+}
+
+// storePlayback produces the web-friendly mp4 and binds it as the playback
+// role. Web-compatible sources are remuxed; everything else is transcoded.
+func (w *Worker) storePlayback(ctx context.Context, inPath string, info *video.Info, siblings []string) error {
+	of, err := os.CreateTemp(w.tempDir, "video-out-*.mp4")
+	if err != nil {
+		return err
+	}
+	outPath := of.Name()
+	_ = of.Close()
+	defer func() { _ = os.Remove(outPath) }()
+
+	remuxable := info.CanRemux(video.MaxLongSide)
+	if remuxable {
+		if err := w.runFFmpeg(ctx, video.RemuxArgs(inPath, outPath)); err != nil {
+			// Some remux-eligible files still fail; fall back to a transcode.
+			if err2 := w.runFFmpeg(ctx, video.TranscodeArgs(inPath, outPath, video.MaxLongSide)); err2 != nil {
+				return err
+			}
+		}
+	} else if err := w.runFFmpeg(ctx, video.TranscodeArgs(inPath, outPath, video.MaxLongSide)); err != nil {
+		return err
+	}
+
+	f, err := os.Open(outPath)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	stat, err := f.Stat()
+	if err != nil {
+		return err
+	}
+	sum, err := hashReader(f)
+	if err != nil {
+		return err
+	}
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+	return w.createObject(ctx, siblings, RolePlayback, ".mp4", "video/mp4", stat.Size(), sum, f)
+}
+
+// runFFmpeg runs ffmpeg with a bounded timeout.
+func (w *Worker) runFFmpeg(ctx context.Context, args []string) error {
+	tctx, cancel := context.WithTimeout(ctx, videoJobTimeout)
+	defer cancel()
+	return video.Run(tctx, w.tools.FFmpegPath(), args...)
+}
+
+func hashReader(r io.Reader) (string, error) {
+	h := sha256.New()
+	if _, err := io.Copy(h, r); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(h.Sum(nil)), nil
+}
+
 func (w *Worker) notify(ctx context.Context, photoID string) {
 	p, err := w.svc.store.GetByID(ctx, photoID, "")
 	if err != nil {