iu
d38ede0201
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.
502 lines
14 KiB
Go
502 lines
14 KiB
Go
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.
|
|
type Worker struct {
|
|
svc *Service
|
|
interval time.Duration
|
|
batch int
|
|
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.
|
|
func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration, concurrency int) *Worker {
|
|
if interval <= 0 {
|
|
interval = 5 * time.Second
|
|
}
|
|
if batch <= 0 {
|
|
batch = 10
|
|
}
|
|
if pendingTTL <= 0 {
|
|
pendingTTL = 10 * time.Minute
|
|
}
|
|
if concurrency <= 0 {
|
|
concurrency = 4
|
|
}
|
|
return &Worker{
|
|
svc: svc,
|
|
interval: interval,
|
|
batch: batch,
|
|
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)
|
|
defer slog.Info("background worker stopped")
|
|
ticker := time.NewTicker(w.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
w.tick(ctx)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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, KindImage, w.batch, w.lease)
|
|
if err != nil {
|
|
slog.Error("claim variant jobs", "error", err)
|
|
return
|
|
}
|
|
w.forEachConcurrent(ids, func(id string) {
|
|
w.generateVariants(ctx, id)
|
|
})
|
|
}
|
|
|
|
// 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.
|
|
func (w *Worker) forEachConcurrent(ids []string, fn func(string)) {
|
|
if len(ids) == 0 {
|
|
return
|
|
}
|
|
if w.concurrency <= 1 {
|
|
for _, id := range ids {
|
|
fn(id)
|
|
}
|
|
return
|
|
}
|
|
sem := make(chan struct{}, w.concurrency)
|
|
var wg sync.WaitGroup
|
|
for _, id := range ids {
|
|
wg.Add(1)
|
|
sem <- struct{}{}
|
|
go func(id string) {
|
|
defer wg.Done()
|
|
defer func() { <-sem }()
|
|
fn(id)
|
|
}(id)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func (w *Worker) generateVariants(ctx context.Context, photoID string) {
|
|
orig, err := w.svc.store.OriginalObject(ctx, photoID)
|
|
if err != nil {
|
|
// The photo or its original binding is gone; nothing to do.
|
|
_ = w.svc.store.CompleteVariantJob(ctx, photoID)
|
|
return
|
|
}
|
|
if orig.State != ObjectReady {
|
|
w.fail(ctx, photoID, "original object is not ready")
|
|
return
|
|
}
|
|
data, err := w.svc.storage.Get(ctx, orig.StorageKey)
|
|
if err != nil {
|
|
w.fail(ctx, photoID, "fetch original: "+err.Error())
|
|
return
|
|
}
|
|
thumb, medium, preview, err := buildVariants(data)
|
|
if err != nil {
|
|
w.fail(ctx, photoID, "build variants: "+err.Error())
|
|
return
|
|
}
|
|
siblings, err := w.svc.store.PhotosByOriginalObject(ctx, orig.ID)
|
|
if err != nil || len(siblings) == 0 {
|
|
siblings = []string{photoID}
|
|
}
|
|
if err := w.createVariant(ctx, siblings, RoleThumb, thumb); err != nil {
|
|
w.fail(ctx, photoID, "store thumb: "+err.Error())
|
|
return
|
|
}
|
|
if err := w.createVariant(ctx, siblings, RoleMedium, medium); err != nil {
|
|
w.fail(ctx, photoID, "store medium: "+err.Error())
|
|
return
|
|
}
|
|
if err := w.createVariant(ctx, siblings, RolePreview, preview); err != nil {
|
|
w.fail(ctx, photoID, "store preview: "+err.Error())
|
|
return
|
|
}
|
|
_ = w.svc.store.CompleteVariantJob(ctx, photoID)
|
|
for _, pid := range siblings {
|
|
w.notify(ctx, pid)
|
|
}
|
|
}
|
|
|
|
// 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, ext)
|
|
obj := &Object{
|
|
ID: objID,
|
|
StorageKey: key,
|
|
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, body, size, mimeType); err != nil {
|
|
_ = w.svc.store.DeleteObjectRow(ctx, objID)
|
|
return err
|
|
}
|
|
if err := w.svc.store.ActivateObject(ctx, objID); err != nil {
|
|
return err
|
|
}
|
|
bound := false
|
|
for _, pid := range photoIDs {
|
|
if err := w.svc.store.BindObject(ctx, pid, objID, role); err != nil {
|
|
if errors.Is(err, ErrObjectNotReady) || errors.Is(err, ErrObjectNotFound) {
|
|
continue
|
|
}
|
|
return err
|
|
}
|
|
bound = true
|
|
}
|
|
if !bound {
|
|
// No photo could take the variant; let the collector reclaim it.
|
|
_ = w.svc.store.MarkObjectDeleting(ctx, objID)
|
|
}
|
|
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 {
|
|
return
|
|
}
|
|
if err := w.svc.attachURLs(ctx, p); err != nil {
|
|
return
|
|
}
|
|
// Variant readiness updates an existing photo; it is not a new photo in the
|
|
// album, so clients update it in place instead of inserting at the top.
|
|
w.svc.hub.Publish(p.AlbumID, "photo.updated", p)
|
|
}
|
|
|
|
func (w *Worker) fail(ctx context.Context, photoID, msg string) {
|
|
slog.Warn("variant job failed", "photo", photoID, "error", msg)
|
|
_ = w.svc.store.FailVariantJob(ctx, photoID, msg, maxVariantAttempts)
|
|
}
|
|
|
|
func (w *Worker) collectGarbage(ctx context.Context) {
|
|
objs, err := w.svc.store.ClaimGarbage(ctx, w.batch)
|
|
if err != nil {
|
|
slog.Error("claim garbage", "error", err)
|
|
return
|
|
}
|
|
for _, o := range objs {
|
|
w.deleteObject(ctx, o)
|
|
}
|
|
}
|
|
|
|
func (w *Worker) reapStalePending(ctx context.Context) {
|
|
objs, err := w.svc.store.ClaimStalePending(ctx, w.pendingTTL, w.batch)
|
|
if err != nil {
|
|
slog.Error("claim stale pending", "error", err)
|
|
return
|
|
}
|
|
for _, o := range objs {
|
|
w.deleteObject(ctx, o)
|
|
}
|
|
}
|
|
|
|
// deleteObject removes an object's blob then its row. If a binding appeared
|
|
// meanwhile the object is returned to ready; if the storage delete fails the
|
|
// row stays deleting and is retried on the next pass.
|
|
func (w *Worker) deleteObject(ctx context.Context, o Object) {
|
|
if n, err := w.svc.store.ObjectRefCount(ctx, o.ID); err == nil && n > 0 {
|
|
_ = w.svc.store.SetObjectState(ctx, o.ID, ObjectReady)
|
|
return
|
|
}
|
|
if err := w.svc.storage.Delete(ctx, o.StorageKey); err != nil {
|
|
slog.Warn("delete object blob", "object", o.ID, "error", err)
|
|
return
|
|
}
|
|
if err := w.svc.store.DeleteObjectRow(ctx, o.ID); err != nil && !errors.Is(err, ErrObjectNotFound) {
|
|
slog.Warn("delete object row", "object", o.ID, "error", err)
|
|
}
|
|
}
|