msb-public
/
cloudreve
mirror of https://github.com/cloudreve/Cloudreve
1
0
Code Issues Projects Releases Wiki Activity

feat(media): deferred video transcoding with ffmpeg (APP-103)

Browse Source 
Extend the media post-processing subsystem (APP-101) to video, sharing
the media_process_task table via the media_type discriminator but running
on its own dedicated queue so a long transcode never blocks image work.

- Upload hook (enqueueMediaProcessIfEligible) now branches by mime: image/*
  and video/* each gate on their own master switch + per-user opt-in +
  min_size, then Enqueue with the matching MediaType.
- Cron runs two independent lanes (enqueueMediaProcessLane): images ->
  MediaProcessQueue, videos -> MediaVideoQueue.
- New dedicated MediaVideoQueue (dependency.go), MediaCompressVideoTaskType,
  started in application.go, live-reloaded via mediaVideoQueuePostProcessor,
  metric in GetQueueMetrics.
- MediaCompressTask.compress branches by media_type; new compressVideo builds
  the ffmpeg command (codec/CRF/preset, videoScaleFilter cap without upscale,
  audio, -threads, optional nice on non-Windows); normalizeVideoContainer
  defaults to "keep" so the file is not renamed.
- Settings block media_compress_video_* + MediaProcessVideo(ctx) getter;
  engine is always ffmpeg for video. cpu_num virtual key in GetSetting feeds
  the frontend threads selector. AutoCompressVideos user opt-in + Codec added
  to the JSON props (no schema/migration/BackendVersion bump).
- Tests: TestMediaProcessEnqueueVideo (discriminator isolation),
  TestNormalizeVideoContainer + TestVideoScaleFilter (ffmpeg arg building).

Bumps the assets submodule pointer to the APP-103 frontend.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
pull/3493/head
thotenn 2 weeks ago
parent 3a0c8f593d
commit 87d9210a32
15 changed files with 637 additions and 202 deletions
Show Stats Download Patch File Download Diff File

1
application/application.go
Unescape View File

@ -109,6 +109,7 @@ func (s *server) Start() error {
s.dep.IoIntenseQueue(context.Background()).Start()
s.dep.RemoteDownloadQueue(context.Background()).Start()
s.dep.MediaProcessQueue(context.Background()).Start()
s.dep.MediaVideoQueue(context.Background()).Start()
// Start cron jobs
c, err := crontab.NewCron(context.Background(), s.dep)

37
application/dependency/dependency.go
Unescape View File

@ -122,6 +122,8 @@ type Dep interface {
EntityRecycleQueue(ctx context.Context) queue.Queue
// MediaProcessQueue Get a singleton queue.Queue instance for media post-processing (APP-101).
MediaProcessQueue(ctx context.Context) queue.Queue
// MediaVideoQueue Get a singleton queue.Queue instance for deferred video transcoding (APP-103).
MediaVideoQueue(ctx context.Context) queue.Queue
// MediaProcessClient Creates a new inventory.MediaProcessClient instance for the media_process_task store.
MediaProcessClient() inventory.MediaProcessClient
// MimeDetector Get a singleton fs.MimeDetector instance for MIME type detection.
@ -186,6 +188,7 @@ type dependency struct {
mediaMetaQueue queue.Queue
entityRecycleQueue queue.Queue
mediaProcessQueue queue.Queue
mediaVideoQueue queue.Queue
slaveQueue queue.Queue
remoteDownloadQueue queue.Queue
ioIntenseQueueTask queue.Task
@ -787,6 +790,40 @@ func (d *dependency) MediaProcessQueue(ctx context.Context) queue.Queue {
return d.mediaProcessQueue
}
func (d *dependency) MediaVideoQueue(ctx context.Context) queue.Queue {
d.mu.Lock()
defer d.mu.Unlock()
_, reload := ctx.Value(ReloadCtx{}).(bool)
if d.mediaVideoQueue != nil && !reload {
return d.mediaVideoQueue
}
if d.mediaVideoQueue != nil {
d.mediaVideoQueue.Shutdown()
}
settings := d.SettingProvider()
queueSetting := settings.Queue(context.Background(), setting.QueueTypeMediaProcessVideo)
// Worker count comes from the dedicated media_compress_video_worker_num key
// (default 1) so a single long transcode stays CPU-bounded — the generic queue
// worker_num getter has an upstream key bug that always yields its fallback.
workerNum := settings.MediaProcessVideo(context.Background()).WorkerNum
d.mediaVideoQueue = queue.New(d.Logger(), d.TaskClient(), d.TaskRegistry(), d,
queue.WithBackoffFactor(queueSetting.BackoffFactor),
queue.WithMaxRetry(queueSetting.MaxRetry),
queue.WithBackoffMaxDuration(queueSetting.BackoffMaxDuration),
queue.WithRetryDelay(queueSetting.RetryDelay),
queue.WithWorkerCount(workerNum),
queue.WithName("MediaVideoQueue"),
queue.WithMaxTaskExecution(queueSetting.MaxExecution),
queue.WithResumeTaskType(queue.MediaCompressVideoTaskType),
queue.WithTaskPullInterval(10*time.Second),
)
return d.mediaVideoQueue
}
func (d *dependency) MediaProcessClient() inventory.MediaProcessClient {
if d.mediaProcessClient != nil {
return d.mediaProcessClient

2
assets

@ -1 +1 @@
Subproject commit 5f597e0fd4342a1727ff699e0235ce804bfd560b
Subproject commit 8788a60ec1edc1fc9062b2e385e778fb372b1fd3

48
inventory/mediaprocess_test.go
Unescape View File

@ -72,6 +72,54 @@ func TestMediaProcessEnqueueIdempotent(t *testing.T) {
assert.NotEqual(t, row1.ID, row3.ID, "a new pending row is created after the previous one is done")
}
// TestMediaProcessEnqueueVideo covers APP-103: video rows are enqueued and listed
// independently of image rows via the media_type discriminator, sharing the same
// idempotency (active-row) and terminal-state semantics.
func TestMediaProcessEnqueueVideo(t *testing.T) {
ctx := context.Background()
client := newTestClient(t)
c := NewMediaProcessClient(client, "sqlite")
// An image and a video row coexist on the same table.
img, err := c.Enqueue(ctx, &MediaProcessEnqueueArgs{
EntityID: 10, FileID: 1, OwnerID: 1, MediaType: mediaprocesstask.MediaTypeImage,
})
require.NoError(t, err)
vid, err := c.Enqueue(ctx, &MediaProcessEnqueueArgs{
EntityID: 20, FileID: 2, OwnerID: 1, MediaType: mediaprocesstask.MediaTypeVideo,
})
require.NoError(t, err)
assert.Equal(t, mediaprocesstask.MediaTypeVideo, vid.MediaType)
// ListPending is scoped by the discriminator: each lane sees only its own rows.
imgs, err := c.ListPending(ctx, mediaprocesstask.MediaTypeImage, 50)
require.NoError(t, err)
assert.Len(t, imgs, 1)
assert.Equal(t, img.ID, imgs[0].ID)
vids, err := c.ListPending(ctx, mediaprocesstask.MediaTypeVideo, 50)
require.NoError(t, err)
require.Len(t, vids, 1)
assert.Equal(t, vid.ID, vids[0].ID)
// Idempotency guard is per entity regardless of media type.
dup, err := c.Enqueue(ctx, &MediaProcessEnqueueArgs{
EntityID: 20, FileID: 2, OwnerID: 1, MediaType: mediaprocesstask.MediaTypeVideo,
})
require.NoError(t, err)
assert.Equal(t, vid.ID, dup.ID, "duplicate video enqueue must reuse the active row")
// Completing the video drops it from its lane without touching the image lane.
_, err = c.SetStatus(ctx, vid.ID, &MediaProcessStatusArgs{Status: mediaprocesstask.StatusDone, ResultSize: 42})
require.NoError(t, err)
vids, err = c.ListPending(ctx, mediaprocesstask.MediaTypeVideo, 50)
require.NoError(t, err)
assert.Len(t, vids, 0)
imgs, err = c.ListPending(ctx, mediaprocesstask.MediaTypeImage, 50)
require.NoError(t, err)
assert.Len(t, imgs, 1, "the image lane is unaffected by video completion")
}
// TestMediaProcessHasHandledForFile covers APP-102: the backfill sweep skips
// files that already have a terminal (done/skipped) row, so a re-run does not
// re-compress already-processed files.

177
inventory/setting.go
View File

File diff suppressed because one or more lines are too long

7
inventory/types/types.go
Unescape View File

@ -20,13 +20,20 @@ type (
// AutoCompressImages opts this user's uploaded images into the deferred
// media post-processing (compression) pipeline (APP-101). Default off.
AutoCompressImages bool `json:"auto_compress_images,omitempty"`
// AutoCompressVideos opts this user's uploaded videos into the deferred
// video transcoding pipeline (APP-103). Default off.
AutoCompressVideos bool `json:"auto_compress_videos,omitempty"`
}
// MediaProcessTaskProps is the JSON payload of a media_process_task row
// (APP-101). Signature is the idempotency fingerprint (engine+quality+format
// +original size) so a blob is not recompressed with the same parameters.
// For video (APP-103) Engine is "ffmpeg", Codec holds the video codec, Quality
// holds the CRF and Format the target container. New fields are JSON-only
// (props is a JSON column) — no schema migration.
MediaProcessTaskProps struct {
Engine string `json:"engine,omitempty"`
Codec string `json:"codec,omitempty"`
Quality int `json:"quality,omitempty"`
Format string `json:"format,omitempty"`
OriginalSize int64 `json:"original_size,omitempty"`

304
pkg/filemanager/manager/mediacompress.go
Unescape View File

@ -9,6 +9,7 @@ import (
"os"
"os/exec"
"path/filepath"
"runtime"
"strconv"
"strings"
@ -62,54 +63,73 @@ type (
func init() {
queue.RegisterResumableTaskFactory(queue.MediaCompressTaskType, NewMediaCompressTaskFromModel)
queue.RegisterResumableTaskFactory(queue.MediaCompressVideoTaskType, NewMediaCompressTaskFromModel)
// Cron: enqueue a bounded batch of pending images onto the dedicated
// MediaProcessQueue. Mirrors pkg/filemanager/manager/recycle.go.
// Cron: enqueue bounded batches of pending media onto their dedicated queues.
// Image and video are independent lanes so either can run with the other off,
// and a long video transcode never blocks image compression (APP-103). Mirrors
// pkg/filemanager/manager/recycle.go.
crontab.Register(setting.CronTypeMediaProcess, func(ctx context.Context) {
dep := dependency.FromContext(ctx)
l := dep.Logger()
sp := dep.SettingProvider()
mp := dep.SettingProvider().MediaProcess(ctx)
if !mp.ImageEnabled {
return
if mp := sp.MediaProcess(ctx); mp.ImageEnabled {
enqueueMediaProcessLane(ctx, dep, mediaprocesstask.MediaTypeImage, mp.BatchSize,
queue.MediaCompressTaskType, dep.MediaProcessQueue(ctx))
}
if v := sp.MediaProcessVideo(ctx); v.Enabled {
enqueueMediaProcessLane(ctx, dep, mediaprocesstask.MediaTypeVideo, v.BatchSize,
queue.MediaCompressVideoTaskType, dep.MediaVideoQueue(ctx))
}
})
}
rows, err := dep.MediaProcessClient().ListPending(ctx, mediaprocesstask.MediaTypeImage, mp.BatchSize)
// enqueueMediaProcessLane lists a bounded batch of pending rows of the given media
// type and enqueues one MediaCompressTask per row onto the provided queue, tagged
// with taskType so it lands on the right (image vs video) lane.
func enqueueMediaProcessLane(ctx context.Context, dep dependency.Dep, mediaType mediaprocesstask.MediaType, batchSize int, taskType string, q queue.Queue) {
l := dep.Logger()
rows, err := dep.MediaProcessClient().ListPending(ctx, mediaType, batchSize)
if err != nil {
l.Error("Failed to list pending media process tasks (%s): %s", mediaType, err)
return
}
if len(rows) == 0 {
return
}
uc := dep.UserClient()
enqueued := 0
for _, row := range rows {
// The task must carry an owner (DBTask.Owner() is dereferenced when the
// queue persists the task); attribute it to the blob's owner.
owner, err := uc.GetByID(ctx, row.OwnerID)
if err != nil {
l.Error("Failed to list pending media process tasks: %s", err)
return
l.Error("Failed to load owner %d for media compress row %d: %s", row.OwnerID, row.ID, err)
continue
}
if len(rows) == 0 {
return
t, err := newMediaCompressTask(ctx, row.ID, owner, taskType)
if err != nil {
l.Error("Failed to create media compress task for row %d: %s", row.ID, err)
continue
}
q := dep.MediaProcessQueue(ctx)
uc := dep.UserClient()
enqueued := 0
for _, row := range rows {
// The task must carry an owner (DBTask.Owner() is dereferenced when the
// queue persists the task); attribute it to the blob's owner.
owner, err := uc.GetByID(ctx, row.OwnerID)
if err != nil {
l.Error("Failed to load owner %d for media compress row %d: %s", row.OwnerID, row.ID, err)
continue
}
t, err := NewMediaCompressTask(ctx, row.ID, owner)
if err != nil {
l.Error("Failed to create media compress task for row %d: %s", row.ID, err)
continue
}
if err := q.QueueTask(ctx, t); err != nil {
l.Error("Failed to queue media compress task for row %d: %s", row.ID, err)
continue
}
enqueued++
if err := q.QueueTask(ctx, t); err != nil {
l.Error("Failed to queue media compress task for row %d: %s", row.ID, err)
continue
}
l.Info("Enqueued %d media compress task(s) from cron.", enqueued)
})
enqueued++
}
l.Info("Enqueued %d media compress task(s) (%s) from cron.", enqueued, mediaType)
}
func NewMediaCompressTask(ctx context.Context, rowID int, owner *ent.User) (queue.Task, error) {
return newMediaCompressTask(ctx, rowID, owner, queue.MediaCompressTaskType)
}
// newMediaCompressTask builds a MediaCompressTask tagged with the given queue task
// type, which selects the lane it runs on: MediaCompressTaskType (image) →
// MediaProcessQueue, MediaCompressVideoTaskType (video) → MediaVideoQueue.
func newMediaCompressTask(ctx context.Context, rowID int, owner *ent.User, taskType string) (queue.Task, error) {
state := &MediaCompressTaskState{RowID: rowID}
stateBytes, err := json.Marshal(state)
if err != nil {
@ -122,7 +142,7 @@ func NewMediaCompressTask(ctx context.Context, rowID int, owner *ent.User) (queu
// persisting the task (pkg/queue/task.go).
DirectOwner: owner,
Task: &ent.Task{
Type: queue.MediaCompressTaskType,
Type: taskType,
CorrelationID: logging.CorrelationID(ctx),
PrivateState: string(stateBytes),
PublicState: &types.TaskPublicState{},
@ -162,8 +182,14 @@ func (m *MediaCompressTask) Do(ctx context.Context) (task.Status, error) {
return task.StatusCompleted, nil
}
settings := dep.SettingProvider().MediaProcess(ctx)
if !settings.ImageEnabled {
// Master-switch gate per media type: image and video have independent switches
// so one lane can be off while the other runs (APP-103).
sp := dep.SettingProvider()
enabled := sp.MediaProcess(ctx).ImageEnabled
if row.MediaType == mediaprocesstask.MediaTypeVideo {
enabled = sp.MediaProcessVideo(ctx).Enabled
}
if !enabled {
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{Status: mediaprocesstask.StatusSkipped})
return task.StatusCompleted, nil
}
@ -177,7 +203,7 @@ func (m *MediaCompressTask) Do(ctx context.Context) (task.Status, error) {
return task.StatusError, fmt.Errorf("failed to mark processing: %w", err)
}
if err := m.compress(ctx, dep, settings, row); err != nil {
if err := m.compress(ctx, dep, row); err != nil {
// Persist the error; the queue backoff will retry up to MaxRetry, after
// which the row stays "processing" — a follow-up sweep can requeue it.
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{
@ -190,10 +216,14 @@ func (m *MediaCompressTask) Do(ctx context.Context) (task.Status, error) {
return task.StatusCompleted, nil
}
// compress runs the full pipeline for one row: read → compress → write-back as a
// new version → mark done. Returns nil on success or a documented skip.
func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, settings *setting.MediaProcessSetting, row *ent.MediaProcessTask) error {
// compress runs the full pipeline for one row: read → compress/transcode →
// write-back as a new version → mark done. The read, size/idempotency guards,
// write-back and quota accounting are shared between image and video; only the
// engine step branches by media_type. Returns nil on success or a documented skip.
func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, row *ent.MediaProcessTask) error {
mpClient := dep.MediaProcessClient()
sp := dep.SettingProvider()
isVideo := row.MediaType == mediaprocesstask.MediaTypeVideo
owner, err := dep.UserClient().GetByID(ctx, row.OwnerID)
if err != nil {
@ -212,8 +242,12 @@ func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, se
}
defer es.Close()
minSize := sp.MediaProcess(ctx).MinSize
if isVideo {
minSize = sp.MediaProcessVideo(ctx).MinSize
}
originalSize := es.Entity().Size()
if originalSize < settings.MinSize {
if originalSize < minSize {
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{Status: mediaprocesstask.StatusSkipped})
return nil
}
@ -227,15 +261,37 @@ func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, se
}
sourceExt := strings.ToLower(strings.TrimPrefix(filepath.Ext(file.Name()), "."))
// Idempotency signature: engine + quality + format + original size.
signature := fmt.Sprintf("%s|q%d|%s|%d", settings.Engine, settings.Quality, settings.Format, originalSize)
// Idempotency signature + the props recorded on completion. Both branch by
// media type: image encodes engine+quality+format, video encodes the ffmpeg
// transcode parameters. A matching signature means the same parameters already
// ran against the same original bytes → skip.
var (
signature string
props *types.MediaProcessTaskProps
)
if isVideo {
v := sp.MediaProcessVideo(ctx)
signature = fmt.Sprintf("ffmpeg|%s|crf%d|%s|%s|%s|%s|%d",
v.Codec, v.CRF, v.Preset, v.Container, v.MaxResolution, v.AudioCodec, originalSize)
props = &types.MediaProcessTaskProps{
Engine: "ffmpeg", Codec: v.Codec, Quality: v.CRF, Format: v.Container,
OriginalSize: originalSize, Signature: signature,
}
} else {
mp := sp.MediaProcess(ctx)
signature = fmt.Sprintf("%s|q%d|%s|%d", mp.Engine, mp.Quality, mp.Format, originalSize)
props = &types.MediaProcessTaskProps{
Engine: mp.Engine, Quality: mp.Quality, Format: mp.Format,
OriginalSize: originalSize, Signature: signature,
}
}
if row.Props != nil && row.Props.Signature == signature {
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{Status: mediaprocesstask.StatusSkipped})
return nil
}
// Materialize a local input path (download if the blob is remote/encrypted).
tempDir := filepath.Join(util.DataPath(dep.SettingProvider().TempPath(ctx)), mediaCompressTempFolder)
tempDir := filepath.Join(util.DataPath(sp.TempPath(ctx)), mediaCompressTempFolder)
if err := util.CreatNestedFolder(tempDir); err != nil {
return fmt.Errorf("failed to create temp folder: %w", err)
}
@ -245,11 +301,19 @@ func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, se
}
defer cleanupInput()
// Compress.
outputPath, targetExt, err := compressImage(ctx, settings, dep.SettingProvider(), inputPath, sourceExt, tempDir)
// Compress/transcode.
var (
outputPath string
targetExt string
)
if isVideo {
outputPath, targetExt, err = compressVideo(ctx, sp.MediaProcessVideo(ctx), sp, inputPath, sourceExt, tempDir)
} else {
outputPath, targetExt, err = compressImage(ctx, sp.MediaProcess(ctx), sp, inputPath, sourceExt, tempDir)
}
if err != nil {
if err == errUnsupportedFormat {
m.l.Info("Unsupported image format %q for row %d, skipping.", sourceExt, row.ID)
m.l.Info("Unsupported %s format %q for row %d, skipping.", row.MediaType, sourceExt, row.ID)
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{Status: mediaprocesstask.StatusSkipped})
return nil
}
@ -264,16 +328,14 @@ func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, se
resultSize := outInfo.Size()
// No gain (or larger): keep the original, mark done with the signature so it
// is not retried with the same parameters.
// is not retried with the same parameters. This guard is more often hit for
// video (an already well-encoded clip may not shrink on re-encode).
if resultSize == 0 || resultSize >= originalSize {
m.l.Info("Compression yielded no gain for row %d (%d -> %d), keeping original.", row.ID, originalSize, resultSize)
_, _ = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{
Status: mediaprocesstask.StatusDone,
ResultSize: originalSize,
Props: &types.MediaProcessTaskProps{
Engine: settings.Engine, Quality: settings.Quality, Format: settings.Format,
OriginalSize: originalSize, Signature: signature,
},
Props: props,
})
return nil
}
@ -305,16 +367,13 @@ func (m *MediaCompressTask) compress(ctx context.Context, dep dependency.Dep, se
_, err = mpClient.SetStatus(ctx, row.ID, &inventory.MediaProcessStatusArgs{
Status: mediaprocesstask.StatusDone,
ResultSize: resultSize,
Props: &types.MediaProcessTaskProps{
Engine: settings.Engine, Quality: settings.Quality, Format: settings.Format,
OriginalSize: originalSize, Signature: signature,
},
Props: props,
})
if err != nil {
return fmt.Errorf("failed to mark done: %w", err)
}
m.l.Info("Compressed row %d: %s %d -> %d bytes (target .%s).", row.ID, sourceExt, originalSize, resultSize, targetExt)
m.l.Info("Compressed row %d (%s): %s %d -> %d bytes (target .%s).", row.ID, row.MediaType, sourceExt, originalSize, resultSize, targetExt)
return nil
}
@ -444,6 +503,101 @@ func ffmpegCompressArgs(inputPath, outputPath, targetExt string, quality int) ([
}
}
// compressVideo transcodes a local video with ffmpeg (the only video engine; vips
// does not process video) and returns the output path + its extension. The target
// container defaults to "keep" (same extension as the source, so the file is not
// renamed). CPU is bounded by -threads and, when enabled, a low process priority
// via nice (best effort, non-Windows only).
func compressVideo(ctx context.Context, v *setting.MediaProcessVideoSetting, settings setting.Provider, inputPath, sourceExt, tempDir string) (string, string, error) {
targetExt := normalizeVideoContainer(v.Container, sourceExt)
if targetExt == "" {
return "", "", errUnsupportedFormat
}
outputPath := filepath.Join(tempDir, fmt.Sprintf("out_%s.%s", uuid.Must(uuid.NewV4()).String(), targetExt))
codec := strings.TrimSpace(v.Codec)
if codec == "" {
codec = "libx264"
}
args := []string{"-y", "-i", inputPath, "-c:v", codec, "-crf", strconv.Itoa(v.CRF)}
if preset := strings.TrimSpace(v.Preset); preset != "" {
args = append(args, "-preset", preset)
}
if scale := videoScaleFilter(v.MaxResolution); scale != "" {
args = append(args, "-vf", scale)
}
if ac := strings.TrimSpace(v.AudioCodec); ac != "" {
args = append(args, "-c:a", ac)
if ab := strings.TrimSpace(v.AudioBitrate); ab != "" {
args = append(args, "-b:a", ab)
}
}
if v.Threads > 0 {
args = append(args, "-threads", strconv.Itoa(v.Threads))
}
if extra := strings.TrimSpace(v.ExtraArgs); extra != "" {
args = append(args, strings.Fields(extra)...)
}
args = append(args, outputPath)
bin := settings.FFMpegPath(ctx)
// Yield CPU to real traffic by running the encode at low priority. `nice` is a
// POSIX tool; skip it on Windows where it is unavailable.
if v.Nice && runtime.GOOS != "windows" {
args = append([]string{"-n", "10", bin}, args...)
bin = "nice"
}
cmd := exec.CommandContext(ctx, bin, args...)
var stdErr bytes.Buffer
cmd.Stderr = &stdErr
if err := cmd.Run(); err != nil {
os.Remove(outputPath)
return "", "", fmt.Errorf("failed to invoke %s: %w, output: %s", bin, err, stdErr.String())
}
return outputPath, targetExt, nil
}
// normalizeVideoContainer resolves the output container/extension. "keep" (default)
// preserves the source extension — but only for recognised video containers, so a
// non-video file yields "" and is skipped. An explicit container overrides it (the
// file is not renamed, matching the image caveat: the bytes change container but
// the name/extension stays — hence "keep" is the recommended default).
func normalizeVideoContainer(container, sourceExt string) string {
c := strings.ToLower(strings.TrimSpace(container))
if c == "" || c == "keep" {
c = strings.ToLower(strings.TrimSpace(sourceExt))
}
switch c {
case "mp4", "m4v", "mov", "mkv", "webm", "avi", "flv", "wmv", "mpeg", "mpg", "ts", "m2ts", "3gp", "ogv":
return c
default:
return ""
}
}
// videoScaleFilter builds an ffmpeg -vf scale filter that caps the output at the
// given WxH without upscaling and keeping the aspect ratio, forcing even
// dimensions (required by yuv420p). An empty or unparseable spec disables scaling.
func videoScaleFilter(maxResolution string) string {
spec := strings.TrimSpace(maxResolution)
if spec == "" {
return ""
}
parts := strings.SplitN(strings.ToLower(spec), "x", 2)
if len(parts) != 2 {
return ""
}
w, errW := strconv.Atoi(strings.TrimSpace(parts[0]))
h, errH := strconv.Atoi(strings.TrimSpace(parts[1]))
if errW != nil || errH != nil || w <= 0 || h <= 0 {
return ""
}
return fmt.Sprintf("scale='min(%d,iw)':'min(%d,ih)':force_original_aspect_ratio=decrease:force_divisible_by=2", w, h)
}
// entitySourceReader is the subset of entitysource.EntitySource the compression
// pipeline consumes (kept local to avoid widening the import surface).
type entitySourceReader interface {
@ -466,11 +620,6 @@ func (m *manager) enqueueMediaProcessIfEligible(ctx context.Context, session *fs
return
}
mp := m.settings.MediaProcess(ctx)
if !mp.ImageEnabled || !m.user.Settings.AutoCompressImages {
return
}
mimeType := ""
if session != nil && session.Props != nil {
mimeType = session.Props.MimeType
@ -478,12 +627,33 @@ func (m *manager) enqueueMediaProcessIfEligible(ctx context.Context, session *fs
if mimeType == "" {
mimeType = m.dep.MimeDetector(ctx).TypeByName(file.Name())
}
if !strings.HasPrefix(strings.ToLower(mimeType), "image/") {
mimeType = strings.ToLower(mimeType)
// Route by mime to the matching media type; each has its own master switch,
// per-user opt-in and min-size gate (APP-101 image, APP-103 video).
var (
mediaType mediaprocesstask.MediaType
minSize int64
)
switch {
case strings.HasPrefix(mimeType, "image/"):
mp := m.settings.MediaProcess(ctx)
if !mp.ImageEnabled || !m.user.Settings.AutoCompressImages {
return
}
mediaType, minSize = mediaprocesstask.MediaTypeImage, mp.MinSize
case strings.HasPrefix(mimeType, "video/"):
v := m.settings.MediaProcessVideo(ctx)
if !v.Enabled || !m.user.Settings.AutoCompressVideos {
return
}
mediaType, minSize = mediaprocesstask.MediaTypeVideo, v.MinSize
default:
return
}
entity := file.PrimaryEntity()
if entity == nil || entity.Size() < mp.MinSize {
if entity == nil || entity.Size() < minSize {
return
}
@ -499,7 +669,7 @@ func (m *manager) enqueueMediaProcessIfEligible(ctx context.Context, session *fs
EntityID: entity.ID(),
FileID: file.ID(),
OwnerID: m.user.ID,
MediaType: mediaprocesstask.MediaTypeImage,
MediaType: mediaType,
}); err != nil {
m.l.Warning("media process: failed to enqueue pending row for entity %d: %s", entity.ID(), err)
}

49
pkg/filemanager/manager/mediacompress_video_test.go
Unescape View File

@ -0,0 +1,49 @@
package manager
import (
"strings"
"testing"
"github.com/stretchr/testify/assert"
)
// TestNormalizeVideoContainer covers APP-103: "keep" preserves the source
// extension for recognised video containers, an explicit container overrides it,
// and non-video / unknown inputs yield "" (→ skipped).
func TestNormalizeVideoContainer(t *testing.T) {
cases := []struct {
container string
sourceExt string
want string
}{
{"keep", "mp4", "mp4"},
{"keep", "MOV", "mov"}, // case-insensitive
{"", "mkv", "mkv"}, // empty behaves like keep
{"keep", "txt", ""}, // non-video source → skip
{"keep", "", ""}, // no extension → skip
{"mp4", "avi", "mp4"}, // explicit container overrides source
{"webm", "mp4", "webm"}, // explicit container overrides source
{"exe", "mp4", ""}, // unknown explicit container → skip
}
for _, c := range cases {
assert.Equal(t, c.want, normalizeVideoContainer(c.container, c.sourceExt),
"normalizeVideoContainer(%q, %q)", c.container, c.sourceExt)
}
}
// TestVideoScaleFilter covers APP-103: a valid WxH yields a scale filter that caps
// without upscaling (min(...)) and forces even dimensions; empty/invalid disables
// scaling (empty string → no -vf).
func TestVideoScaleFilter(t *testing.T) {
assert.Equal(t, "", videoScaleFilter(""), "empty spec disables scaling")
assert.Equal(t, "", videoScaleFilter("1080"), "missing dimension is invalid")
assert.Equal(t, "", videoScaleFilter("axb"), "non-numeric is invalid")
assert.Equal(t, "", videoScaleFilter("0x0"), "non-positive is invalid")
f := videoScaleFilter("1920x1080")
assert.Contains(t, f, "min(1920,iw)")
assert.Contains(t, f, "min(1080,ih)")
assert.Contains(t, f, "force_original_aspect_ratio=decrease")
assert.Contains(t, f, "force_divisible_by=2")
assert.True(t, strings.HasPrefix(f, "scale="))
}

1
pkg/queue/task.go
Unescape View File

@ -105,6 +105,7 @@ const (
RemoteDownloadTaskType = "remote_download"
ImportTaskType = "import"
MediaCompressTaskType = "media_compress"
MediaCompressVideoTaskType = "media_compress_video"
MediaBackfillTaskType = "media_backfill"
FullTextIndexTaskType = "full_text_index"

29
pkg/setting/provider.go
Unescape View File

@ -158,6 +158,8 @@ type (
Cron(ctx context.Context, t CronType) string
// MediaProcess returns the media post-processing (image compression) settings.
MediaProcess(ctx context.Context) *MediaProcessSetting
// MediaProcessVideo returns the deferred video transcoding settings (APP-103).
MediaProcessVideo(ctx context.Context) *MediaProcessVideoSetting
// Theme returns the theme settings.
Theme(ctx context.Context) *Theme
// Logo returns the logo settings.
@ -459,6 +461,33 @@ func (s *settingProvider) MediaProcess(ctx context.Context) *MediaProcessSetting
}
}
func (s *settingProvider) MediaProcessVideo(ctx context.Context) *MediaProcessVideoSetting {
workerNum := s.getInt(ctx, "media_compress_video_worker_num", 1)
if workerNum <= 0 {
workerNum = 1
}
threads := s.getInt(ctx, "media_compress_video_threads", 1)
if threads <= 0 {
threads = 1
}
return &MediaProcessVideoSetting{
Enabled: s.getBoolean(ctx, "media_compress_video_enabled", false),
Codec: s.getString(ctx, "media_compress_video_codec", "libx264"),
CRF: s.getInt(ctx, "media_compress_video_crf", 28),
Preset: s.getString(ctx, "media_compress_video_preset", "medium"),
Container: s.getString(ctx, "media_compress_video_container", "keep"),
MaxResolution: s.getString(ctx, "media_compress_video_max_resolution", "1920x1080"),
AudioCodec: s.getString(ctx, "media_compress_video_audio_codec", "aac"),
AudioBitrate: s.getString(ctx, "media_compress_video_audio_bitrate", "128k"),
ExtraArgs: s.getString(ctx, "media_compress_video_args", ""),
WorkerNum: workerNum,
BatchSize: s.getInt(ctx, "media_compress_video_batch_size", 10),
Threads: threads,
Nice: s.getBoolean(ctx, "media_compress_video_nice", true),
MinSize: s.getInt64(ctx, "media_compress_video_min_size", 10485760),
}
}
func (s *settingProvider) BuiltinThumbGeneratorEnabled(ctx context.Context) bool {
return s.getBoolean(ctx, "thumb_builtin_enabled", true)
}

25
pkg/setting/types.go
Unescape View File

@ -103,6 +103,27 @@ type (
ResultMode string // "version" | "replace" | "auto"
MinSize int64 // skip blobs smaller than this (bytes)
}
// MediaProcessVideoSetting carries the deferred video transcoding parameters
// (APP-103), all editable from the admin panel. The engine is always ffmpeg
// (vips does not process video). Video runs on its own dedicated queue so a
// long transcode never blocks image compression.
MediaProcessVideoSetting struct {
Enabled bool // master switch for video transcoding
Codec string // video codec, e.g. "libx264" | "libx265"
CRF int // constant rate factor (lower = better quality/bigger)
Preset string // ffmpeg preset (speed/size trade-off), e.g. "medium"
Container string // "keep" | "mp4" | "webm" — output container/extension
MaxResolution string // downscale cap, e.g. "1920x1080" (empty = no cap)
AudioCodec string // audio codec, e.g. "aac"
AudioBitrate string // audio bitrate, e.g. "128k"
ExtraArgs string // extra ffmpeg flags
WorkerNum int // transcoding concurrency (dedicated key; default 1)
BatchSize int // max rows the cron enqueues per run
Threads int // ffmpeg -threads per encode (approximate CPU cap)
Nice bool // run ffmpeg at low priority (best effort, non-Windows)
MinSize int64 // skip blobs smaller than this (bytes)
}
)
type ThumbEncode struct {
@ -118,6 +139,10 @@ var (
QueueTypeSlave = QueueType("slave")
QueueTypeRemoteDownload = QueueType("remote_download")
QueueTypeMediaProcess = QueueType("media_process")
// QueueTypeMediaProcessVideo is the dedicated queue for video transcoding
// (APP-103), kept separate from media_process so a long transcode does not
// block image compression.
QueueTypeMediaProcessVideo = QueueType("media_process_video")
)
type CronType string

142
service/admin/site.go
Unescape View File

@ -7,6 +7,8 @@ import (
"fmt"
"net/url"
"reflect"
"runtime"
"strconv"
"strings"
"time"
@ -190,9 +192,17 @@ type (
GetSettingParamCtx struct{}
)
// cpuNumSettingKey is a virtual (non-DB) key: when requested, GetSetting returns
// the CPU count visible to the process (runtime.NumCPU()) so the admin UI can cap
// the "threads per compression" selector to the available cores (APP-103).
const cpuNumSettingKey = "cpu_num"
func (s *GetSettingService) GetSetting(c *gin.Context) (map[string]string, error) {
dep := dependency.FromContext(c)
res, err := dep.SettingClient().Gets(c, lo.Filter(s.Keys, func(item string, index int) bool {
if strings.ToLower(item) == cpuNumSettingKey {
return false
}
_, ok := inventory.RedactedSettings[strings.ToLower(item)]
return !ok
}))
@ -200,6 +210,10 @@ func (s *GetSettingService) GetSetting(c *gin.Context) (map[string]string, error
return nil, serializer.NewError(serializer.CodeDBError, "Failed to get settings", err)
}
if lo.ContainsBy(s.Keys, func(item string) bool { return strings.ToLower(item) == cpuNumSettingKey }) {
res[cpuNumSettingKey] = strconv.Itoa(runtime.NumCPU())
}
return res, nil
}
@ -219,64 +233,70 @@ var (
"secret_key": secretKeyPreProcessor,
}
postprocessors = map[string]SettingPostProcessor{
"mime_mapping": mimeMappingPostProcessor,
"media_meta_exif": mediaMetaPostProcessor,
"media_meta_music": mediaMetaPostProcessor,
"media_meta_ffprobe": mediaMetaPostProcessor,
"smtpUser": emailPostProcessor,
"smtpPass": emailPostProcessor,
"smtpHost": emailPostProcessor,
"smtpPort": emailPostProcessor,
"smtpEncryption": emailPostProcessor,
"smtpFrom": emailPostProcessor,
"replyTo": emailPostProcessor,
"fromName": emailPostProcessor,
"fromAdress": emailPostProcessor,
"queue_media_meta_worker_num": mediaMetaQueuePostProcessor,
"queue_media_meta_max_execution": mediaMetaQueuePostProcessor,
"queue_media_meta_backoff_factor": mediaMetaQueuePostProcessor,
"queue_media_meta_backoff_max_duration": mediaMetaQueuePostProcessor,
"queue_media_meta_max_retry": mediaMetaQueuePostProcessor,
"queue_media_meta_retry_delay": mediaMetaQueuePostProcessor,
"queue_thumb_worker_num": thumbQueuePostProcessor,
"queue_thumb_max_execution": thumbQueuePostProcessor,
"queue_thumb_backoff_factor": thumbQueuePostProcessor,
"queue_thumb_backoff_max_duration": thumbQueuePostProcessor,
"queue_thumb_max_retry": thumbQueuePostProcessor,
"queue_thumb_retry_delay": thumbQueuePostProcessor,
"queue_recycle_worker_num": entityRecycleQueuePostProcessor,
"queue_recycle_max_execution": entityRecycleQueuePostProcessor,
"queue_recycle_backoff_factor": entityRecycleQueuePostProcessor,
"queue_recycle_backoff_max_duration": entityRecycleQueuePostProcessor,
"queue_recycle_max_retry": entityRecycleQueuePostProcessor,
"queue_recycle_retry_delay": entityRecycleQueuePostProcessor,
"queue_io_intense_worker_num": ioIntenseQueuePostProcessor,
"queue_io_intense_max_execution": ioIntenseQueuePostProcessor,
"queue_io_intense_backoff_factor": ioIntenseQueuePostProcessor,
"queue_io_intense_backoff_max_duration": ioIntenseQueuePostProcessor,
"queue_io_intense_max_retry": ioIntenseQueuePostProcessor,
"queue_io_intense_retry_delay": ioIntenseQueuePostProcessor,
"queue_remote_download_worker_num": remoteDownloadQueuePostProcessor,
"queue_remote_download_max_execution": remoteDownloadQueuePostProcessor,
"queue_remote_download_backoff_factor": remoteDownloadQueuePostProcessor,
"queue_remote_download_backoff_max_duration": remoteDownloadQueuePostProcessor,
"queue_remote_download_max_retry": remoteDownloadQueuePostProcessor,
"queue_remote_download_retry_delay": remoteDownloadQueuePostProcessor,
"media_compress_worker_num": mediaProcessQueuePostProcessor,
"queue_media_process_max_execution": mediaProcessQueuePostProcessor,
"queue_media_process_backoff_factor": mediaProcessQueuePostProcessor,
"queue_media_process_backoff_max_duration": mediaProcessQueuePostProcessor,
"queue_media_process_max_retry": mediaProcessQueuePostProcessor,
"queue_media_process_retry_delay": mediaProcessQueuePostProcessor,
"secret_key": secretKeyPostProcessor,
"fts_meilisearch_embed_config": meilisearchPostProcessor,
"fts_meilisearch_endpoint": meilisearchPostProcessor,
"fts_meilisearch_api_key": meilisearchPostProcessor,
"fts_meilisearch_embed_enabled": meilisearchPostProcessor,
"fts_meilisearch_page_size": meilisearchPostProcessor,
"fts_tika_endpoint": tikaPostProcessor,
"fts_tika_exts": tikaPostProcessor,
"fts_tika_max_file_size": tikaPostProcessor,
"mime_mapping": mimeMappingPostProcessor,
"media_meta_exif": mediaMetaPostProcessor,
"media_meta_music": mediaMetaPostProcessor,
"media_meta_ffprobe": mediaMetaPostProcessor,
"smtpUser": emailPostProcessor,
"smtpPass": emailPostProcessor,
"smtpHost": emailPostProcessor,
"smtpPort": emailPostProcessor,
"smtpEncryption": emailPostProcessor,
"smtpFrom": emailPostProcessor,
"replyTo": emailPostProcessor,
"fromName": emailPostProcessor,
"fromAdress": emailPostProcessor,
"queue_media_meta_worker_num": mediaMetaQueuePostProcessor,
"queue_media_meta_max_execution": mediaMetaQueuePostProcessor,
"queue_media_meta_backoff_factor": mediaMetaQueuePostProcessor,
"queue_media_meta_backoff_max_duration": mediaMetaQueuePostProcessor,
"queue_media_meta_max_retry": mediaMetaQueuePostProcessor,
"queue_media_meta_retry_delay": mediaMetaQueuePostProcessor,
"queue_thumb_worker_num": thumbQueuePostProcessor,
"queue_thumb_max_execution": thumbQueuePostProcessor,
"queue_thumb_backoff_factor": thumbQueuePostProcessor,
"queue_thumb_backoff_max_duration": thumbQueuePostProcessor,
"queue_thumb_max_retry": thumbQueuePostProcessor,
"queue_thumb_retry_delay": thumbQueuePostProcessor,
"queue_recycle_worker_num": entityRecycleQueuePostProcessor,
"queue_recycle_max_execution": entityRecycleQueuePostProcessor,
"queue_recycle_backoff_factor": entityRecycleQueuePostProcessor,
"queue_recycle_backoff_max_duration": entityRecycleQueuePostProcessor,
"queue_recycle_max_retry": entityRecycleQueuePostProcessor,
"queue_recycle_retry_delay": entityRecycleQueuePostProcessor,
"queue_io_intense_worker_num": ioIntenseQueuePostProcessor,
"queue_io_intense_max_execution": ioIntenseQueuePostProcessor,
"queue_io_intense_backoff_factor": ioIntenseQueuePostProcessor,
"queue_io_intense_backoff_max_duration": ioIntenseQueuePostProcessor,
"queue_io_intense_max_retry": ioIntenseQueuePostProcessor,
"queue_io_intense_retry_delay": ioIntenseQueuePostProcessor,
"queue_remote_download_worker_num": remoteDownloadQueuePostProcessor,
"queue_remote_download_max_execution": remoteDownloadQueuePostProcessor,
"queue_remote_download_backoff_factor": remoteDownloadQueuePostProcessor,
"queue_remote_download_backoff_max_duration": remoteDownloadQueuePostProcessor,
"queue_remote_download_max_retry": remoteDownloadQueuePostProcessor,
"queue_remote_download_retry_delay": remoteDownloadQueuePostProcessor,
"media_compress_worker_num": mediaProcessQueuePostProcessor,
"queue_media_process_max_execution": mediaProcessQueuePostProcessor,
"queue_media_process_backoff_factor": mediaProcessQueuePostProcessor,
"queue_media_process_backoff_max_duration": mediaProcessQueuePostProcessor,
"queue_media_process_max_retry": mediaProcessQueuePostProcessor,
"queue_media_process_retry_delay": mediaProcessQueuePostProcessor,
"media_compress_video_worker_num": mediaVideoQueuePostProcessor,
"queue_media_process_video_max_execution": mediaVideoQueuePostProcessor,
"queue_media_process_video_backoff_factor": mediaVideoQueuePostProcessor,
"queue_media_process_video_backoff_max_duration": mediaVideoQueuePostProcessor,
"queue_media_process_video_max_retry": mediaVideoQueuePostProcessor,
"queue_media_process_video_retry_delay": mediaVideoQueuePostProcessor,
"secret_key": secretKeyPostProcessor,
"fts_meilisearch_embed_config": meilisearchPostProcessor,
"fts_meilisearch_endpoint": meilisearchPostProcessor,
"fts_meilisearch_api_key": meilisearchPostProcessor,
"fts_meilisearch_embed_enabled": meilisearchPostProcessor,
"fts_meilisearch_page_size": meilisearchPostProcessor,
"fts_tika_endpoint": tikaPostProcessor,
"fts_tika_exts": tikaPostProcessor,
"fts_tika_max_file_size": tikaPostProcessor,
}
)
@ -425,6 +445,12 @@ func mediaProcessQueuePostProcessor(ctx context.Context, settings map[string]str
return nil
}
func mediaVideoQueuePostProcessor(ctx context.Context, settings map[string]string) error {
dep := dependency.FromContext(ctx)
dep.MediaVideoQueue(context.WithValue(ctx, dependency.ReloadCtx{}, true)).Start()
return nil
}
func secretKeyPostProcessor(ctx context.Context, settings map[string]string) error {
dep := dependency.FromContext(ctx)
dep.KV().Delete(manager.EntityUrlCacheKeyPrefix)

9
service/admin/task.go
Unescape View File

@ -28,6 +28,7 @@ func GetQueueMetrics(c *gin.Context) ([]QueueMetric, error) {
remoteDownload := dep.RemoteDownloadQueue(c)
thumb := dep.ThumbQueue(c)
mediaProcess := dep.MediaProcessQueue(c)
mediaVideo := dep.MediaVideoQueue(c)
res = append(res, QueueMetric{
Name: setting.QueueTypeMediaMeta,
@ -77,6 +78,14 @@ func GetQueueMetrics(c *gin.Context) ([]QueueMetric, error) {
SubmittedTasks: mediaProcess.SubmittedTasks(),
SuspendingTasks: mediaProcess.SuspendingTasks(),
})
res = append(res, QueueMetric{
Name: setting.QueueTypeMediaProcessVideo,
BusyWorkers: mediaVideo.BusyWorkers(),
SuccessTasks: mediaVideo.SuccessTasks(),
FailureTasks: mediaVideo.FailureTasks(),
SubmittedTasks: mediaVideo.SubmittedTasks(),
SuspendingTasks: mediaVideo.SuspendingTasks(),
})
return res, nil
}

2
service/user/response.go
Unescape View File

@ -33,6 +33,7 @@ type UserSettings struct {
DisableViewSync bool `json:"disable_view_sync"`
ShareLinksInProfile string `json:"share_links_in_profile"`
AutoCompressImages bool `json:"auto_compress_images"`
AutoCompressVideos bool `json:"auto_compress_videos"`
OAuthGrants []OauthGrant `json:"oauth_grants,omitempty"`
}
@ -49,6 +50,7 @@ func BuildUserSettings(u *ent.User, passkeys []*ent.Passkey, parser *uaparser.Pa
DisableViewSync: u.Settings.DisableViewSync,
ShareLinksInProfile: string(u.Settings.ShareLinksInProfile),
AutoCompressImages: u.Settings.AutoCompressImages,
AutoCompressVideos: u.Settings.AutoCompressVideos,
OAuthGrants: lo.Map(grants, func(item *ent.OAuthGrant, index int) OauthGrant {
return BuildOauthGrant(item)
}),

6
service/user/setting.go
Unescape View File

@ -231,6 +231,7 @@ type (
DisableViewSync *bool `json:"disable_view_sync" binding:"omitempty"`
ShareLinksInProfile *string `json:"share_links_in_profile" binding:"omitempty"`
AutoCompressImages *bool `json:"auto_compress_images" binding:"omitempty"`
AutoCompressVideos *bool `json:"auto_compress_videos" binding:"omitempty"`
}
PatchUserSettingParamsCtx struct{}
)
@ -282,6 +283,11 @@ func (s *PatchUserSetting) Patch(c *gin.Context) error {
saveSetting = true
}
if s.AutoCompressVideos != nil {
u.Settings.AutoCompressVideos = *s.AutoCompressVideos
saveSetting = true
}
if s.ShareLinksInProfile != nil {
u.Settings.ShareLinksInProfile = types.ShareLinksInProfileLevel(*s.ShareLinksInProfile)
saveSetting = true

Write Preview
Loading…
Cancel
Save