From 3a0c8f593d8796fc40c9fca9c9b3344db5d9448d Mon Sep 17 00:00:00 2001 From: thotenn Date: Sat, 11 Jul 2026 00:44:37 -0300 Subject: [PATCH] feat(media): backfill existing images for deferred compression (APP-102) Seed pending media_process_task rows for a user's already-stored images so the APP-101 pipeline compresses them. Adds a resumable MediaBackfillTask (mirrors full_text_rebuild) that paginates candidate files and enqueues each image idempotently, a self-service /workflow/mediaBackfill endpoint, and cursor-based candidate listers. A file-level guard (HasHandledForFile) skips already-processed files so re-runs do not re-compress. Bumps the assets submodule for the frontend UI. Co-Authored-By: Claude Opus 4.8 (1M context) --- application/dependency/dependency.go | 1 + assets | 2 +- inventory/file.go | 33 +++ inventory/file_backfill_test.go | 107 ++++++++++ inventory/mediaprocess.go | 16 ++ inventory/mediaprocess_test.go | 41 ++++ pkg/filemanager/workflows/media_backfill.go | 219 ++++++++++++++++++++ pkg/queue/task.go | 1 + routers/controllers/file.go | 15 ++ routers/router.go | 6 + service/explorer/workflows.go | 27 +++ 11 files changed, 467 insertions(+), 1 deletion(-) create mode 100644 inventory/file_backfill_test.go create mode 100644 pkg/filemanager/workflows/media_backfill.go diff --git a/application/dependency/dependency.go b/application/dependency/dependency.go index 9036d2f4..163dc356 100644 --- a/application/dependency/dependency.go +++ b/application/dependency/dependency.go @@ -657,6 +657,7 @@ func (d *dependency) MediaMetaQueue(ctx context.Context) queue.Queue { queue.FullTextRebuildTaskType, queue.FullTextCopyTaskType, queue.FullTextChangeOwnerTaskType, + queue.MediaBackfillTaskType, ), ) return d.mediaMetaQueue diff --git a/assets b/assets index 74d41bbc..5f597e0f 160000 --- a/assets +++ b/assets @@ -1 +1 @@ -Subproject commit 74d41bbcee74020f3f071340f8cea73152f705c9 +Subproject commit 5f597e0fd4342a1727ff699e0235ce804bfd560b diff --git a/inventory/file.go b/inventory/file.go index 3b344bd9..9fbe1853 100644 --- a/inventory/file.go +++ b/inventory/file.go @@ -214,6 +214,12 @@ type FileClient interface { // ListIndexableFiles lists files suitable for FTS indexing, returning up to limit files // with ID strictly greater than afterID. Use afterID=0 to start from the beginning. ListIndexableFiles(ctx context.Context, afterID, limit int) ([]*ent.File, error) + // CountBackfillCandidateFiles counts files eligible for deferred media backfill (file type, + // has primary entity, size >= minSize), optionally scoped by owner and storage policy. + CountBackfillCandidateFiles(ctx context.Context, userID int, storagePolicyIDs []int, minSize int64) (int, error) + // ListBackfillCandidateFiles lists media backfill candidate files with ID strictly greater + // than afterID, up to limit. Use afterID=0 to start from the beginning. + ListBackfillCandidateFiles(ctx context.Context, afterID, limit, userID int, storagePolicyIDs []int, minSize int64) ([]*ent.File, error) // CountByTimeRange counts files created in a given time range CountByTimeRange(ctx context.Context, start, end *time.Time) (int, error) // CountEntityByTimeRange counts entities created in a given time range @@ -351,6 +357,33 @@ func (f *fileClient) indexableFilesQuery() *ent.FileQuery { ).Order(file.ByID()) } +func (f *fileClient) backfillCandidateQuery(userID int, storagePolicyIDs []int, minSize int64) *ent.FileQuery { + q := f.client.File.Query().Where( + file.Type(int(types.FileTypeFile)), + file.PrimaryEntityGT(0), + file.SizeGTE(minSize), + ) + if userID > 0 { + q = q.Where(file.OwnerID(userID)) + } + if len(storagePolicyIDs) > 0 { + q = q.Where(file.StoragePolicyFilesIn(storagePolicyIDs...)) + } + return q.Order(file.ByID()) +} + +func (f *fileClient) CountBackfillCandidateFiles(ctx context.Context, userID int, storagePolicyIDs []int, minSize int64) (int, error) { + return f.backfillCandidateQuery(userID, storagePolicyIDs, minSize).Count(ctx) +} + +func (f *fileClient) ListBackfillCandidateFiles(ctx context.Context, afterID, limit, userID int, storagePolicyIDs []int, minSize int64) ([]*ent.File, error) { + q := f.backfillCandidateQuery(userID, storagePolicyIDs, minSize) + if afterID > 0 { + q = q.Where(file.IDGT(afterID)) + } + return q.Limit(limit).All(ctx) +} + func (f *fileClient) CountEntityByTimeRange(ctx context.Context, start, end *time.Time) (int, error) { if start == nil || end == nil { return f.client.Entity.Query().Count(ctx) diff --git a/inventory/file_backfill_test.go b/inventory/file_backfill_test.go new file mode 100644 index 00000000..c23c2fe3 --- /dev/null +++ b/inventory/file_backfill_test.go @@ -0,0 +1,107 @@ +package inventory + +import ( + "context" + "testing" + + "github.com/cloudreve/Cloudreve/v4/ent" + "github.com/cloudreve/Cloudreve/v4/inventory/types" + "github.com/cloudreve/Cloudreve/v4/pkg/boolset" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func makeBackfillFile(t *testing.T, client *ent.Client, fileType types.FileType, name string, ownerID int, size int64, primaryEntity, policyID int) *ent.File { + t.Helper() + stm := client.File.Create(). + SetType(int(fileType)). + SetName(name). + SetOwnerID(ownerID). + SetSize(size) + if primaryEntity > 0 { + stm.SetPrimaryEntity(primaryEntity) + } + if policyID > 0 { + stm.SetStoragePolicyFiles(policyID) + } + f, err := stm.Save(context.Background()) + require.NoError(t, err) + return f +} + +func backfillFileIDs(files []*ent.File) []int { + out := make([]int, len(files)) + for i, f := range files { + out[i] = f.ID + } + return out +} + +// TestBackfillCandidateFiles covers APP-102: the candidate query returns only +// file-type rows with a primary entity and size >= minSize, honors the owner and +// storage-policy scopes, and paginates by ascending id via the afterID cursor. +func TestBackfillCandidateFiles(t *testing.T) { + ctx := context.Background() + client := newTestClient(t) + c := NewFileClient(client, "sqlite", nil) + + // Minimal graph so File FKs (owner_id, storage_policy_files) resolve. + grp, err := client.Group.Create(). + SetName("g"). + SetPermissions(&boolset.BooleanSet{}). + SetSettings(&types.GroupSetting{}). + Save(ctx) + require.NoError(t, err) + u1, err := client.User.Create().SetEmail("u1@example.com").SetNick("u1").SetGroupUsers(grp.ID).Save(ctx) + require.NoError(t, err) + u2, err := client.User.Create().SetEmail("u2@example.com").SetNick("u2").SetGroupUsers(grp.ID).Save(ctx) + require.NoError(t, err) + p1 := makePolicy(t, client, "p1") + p2 := makePolicy(t, client, "p2") + + const minSize = int64(200000) + + // Eligible: files owned by u1, big enough, with a primary entity. + f1 := makeBackfillFile(t, client, types.FileTypeFile, "a.jpg", u1.ID, 300000, 101, p1.ID) + f2 := makeBackfillFile(t, client, types.FileTypeFile, "b.png", u1.ID, 500000, 102, p1.ID) + // Excluded: below min size. + makeBackfillFile(t, client, types.FileTypeFile, "small.jpg", u1.ID, 1000, 103, p1.ID) + // Excluded: no primary entity. + makeBackfillFile(t, client, types.FileTypeFile, "noentity.jpg", u1.ID, 400000, 0, p1.ID) + // Excluded: folder, not a file. + makeBackfillFile(t, client, types.FileTypeFolder, "folder", u1.ID, 400000, 104, p1.ID) + // Belongs to u2 (excluded when scoping to u1). + makeBackfillFile(t, client, types.FileTypeFile, "other.jpg", u2.ID, 400000, 105, p1.ID) + // Owned by u1 but on a different storage policy. + f7 := makeBackfillFile(t, client, types.FileTypeFile, "policy2.jpg", u1.ID, 400000, 106, p2.ID) + + // Count scoped to u1, all policies: f1, f2, f7. + total, err := c.CountBackfillCandidateFiles(ctx, u1.ID, nil, minSize) + require.NoError(t, err) + assert.Equal(t, 3, total) + + // List scoped to u1, ordered by ascending id. + files, err := c.ListBackfillCandidateFiles(ctx, 0, 100, u1.ID, nil, minSize) + require.NoError(t, err) + require.Len(t, files, 3) + assert.Equal(t, []int{f1.ID, f2.ID, f7.ID}, backfillFileIDs(files)) + + // Cursor pagination: a batch of 2, then the remainder after the last id. + page1, err := c.ListBackfillCandidateFiles(ctx, 0, 2, u1.ID, nil, minSize) + require.NoError(t, err) + require.Len(t, page1, 2) + page2, err := c.ListBackfillCandidateFiles(ctx, page1[len(page1)-1].ID, 2, u1.ID, nil, minSize) + require.NoError(t, err) + require.Len(t, page2, 1) + assert.Equal(t, f7.ID, page2[0].ID) + + // Storage-policy scope: policy p1 only → f1, f2 (f7 is on p2). + pol1, err := c.CountBackfillCandidateFiles(ctx, u1.ID, []int{p1.ID}, minSize) + require.NoError(t, err) + assert.Equal(t, 2, pol1) + + // No owner scope (userID=0): includes u2's file too → f1, f2, u2's, f7. + all, err := c.CountBackfillCandidateFiles(ctx, 0, nil, minSize) + require.NoError(t, err) + assert.Equal(t, 4, all) +} diff --git a/inventory/mediaprocess.go b/inventory/mediaprocess.go index e20aaae6..1ee07000 100644 --- a/inventory/mediaprocess.go +++ b/inventory/mediaprocess.go @@ -21,6 +21,10 @@ type MediaProcessClient interface { // HasActive reports whether an active (pending/processing) row exists for the // given entity id. HasActive(ctx context.Context, entityID int) (bool, error) + // HasHandledForFile reports whether a terminal (done/skipped) row already + // exists for the given file id. Used by the backfill sweep to avoid + // re-enqueuing files it has already processed once. + HasHandledForFile(ctx context.Context, fileID int) (bool, error) // ListPending returns up to limit pending rows of the given media type, // oldest first. ListPending(ctx context.Context, mediaType mediaprocesstask.MediaType, limit int) ([]*ent.MediaProcessTask, error) @@ -75,6 +79,18 @@ func (c *mediaProcessClient) HasActive(ctx context.Context, entityID int) (bool, Exist(ctx) } +func (c *mediaProcessClient) HasHandledForFile(ctx context.Context, fileID int) (bool, error) { + if fileID == 0 { + return false, nil + } + return c.client.MediaProcessTask.Query(). + Where( + mediaprocesstask.FileID(fileID), + mediaprocesstask.StatusIn(mediaprocesstask.StatusDone, mediaprocesstask.StatusSkipped), + ). + Exist(ctx) +} + func (c *mediaProcessClient) Enqueue(ctx context.Context, args *MediaProcessEnqueueArgs) (*ent.MediaProcessTask, error) { mediaType := args.MediaType if mediaType == "" { diff --git a/inventory/mediaprocess_test.go b/inventory/mediaprocess_test.go index 2d628383..d8c4b8cd 100644 --- a/inventory/mediaprocess_test.go +++ b/inventory/mediaprocess_test.go @@ -71,3 +71,44 @@ func TestMediaProcessEnqueueIdempotent(t *testing.T) { require.NoError(t, err) assert.NotEqual(t, row1.ID, row3.ID, "a new pending row is created after the previous one is done") } + +// 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. +func TestMediaProcessHasHandledForFile(t *testing.T) { + ctx := context.Background() + client := newTestClient(t) + c := NewMediaProcessClient(client, "sqlite") + + handled, err := c.HasHandledForFile(ctx, 500) + require.NoError(t, err) + assert.False(t, handled, "no rows yet") + + // A pending row is not terminal → not handled. + row, err := c.Enqueue(ctx, &MediaProcessEnqueueArgs{EntityID: 900, FileID: 500, OwnerID: 1, MediaType: mediaprocesstask.MediaTypeImage}) + require.NoError(t, err) + handled, err = c.HasHandledForFile(ctx, 500) + require.NoError(t, err) + assert.False(t, handled) + + // Done counts as handled. + _, err = c.SetStatus(ctx, row.ID, &MediaProcessStatusArgs{Status: mediaprocesstask.StatusDone, ResultSize: 10}) + require.NoError(t, err) + handled, err = c.HasHandledForFile(ctx, 500) + require.NoError(t, err) + assert.True(t, handled) + + // Skipped also counts as handled. + row2, err := c.Enqueue(ctx, &MediaProcessEnqueueArgs{EntityID: 901, FileID: 501, OwnerID: 1, MediaType: mediaprocesstask.MediaTypeImage}) + require.NoError(t, err) + _, err = c.SetStatus(ctx, row2.ID, &MediaProcessStatusArgs{Status: mediaprocesstask.StatusSkipped}) + require.NoError(t, err) + handled, err = c.HasHandledForFile(ctx, 501) + require.NoError(t, err) + assert.True(t, handled) + + // fileID 0 is never handled. + handled, err = c.HasHandledForFile(ctx, 0) + require.NoError(t, err) + assert.False(t, handled) +} diff --git a/pkg/filemanager/workflows/media_backfill.go b/pkg/filemanager/workflows/media_backfill.go new file mode 100644 index 00000000..ef416fef --- /dev/null +++ b/pkg/filemanager/workflows/media_backfill.go @@ -0,0 +1,219 @@ +package workflows + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync/atomic" + + "github.com/cloudreve/Cloudreve/v4/application/dependency" + "github.com/cloudreve/Cloudreve/v4/ent" + "github.com/cloudreve/Cloudreve/v4/ent/mediaprocesstask" + "github.com/cloudreve/Cloudreve/v4/ent/task" + "github.com/cloudreve/Cloudreve/v4/inventory" + "github.com/cloudreve/Cloudreve/v4/inventory/types" + "github.com/cloudreve/Cloudreve/v4/pkg/hashid" + "github.com/cloudreve/Cloudreve/v4/pkg/logging" + "github.com/cloudreve/Cloudreve/v4/pkg/queue" +) + +type ( + MediaBackfillTask struct { + *queue.DBTask + + l logging.Logger + state *MediaBackfillTaskState + progress queue.Progresses + } + MediaBackfillTaskPhase string + MediaBackfillTaskState struct { + Phase MediaBackfillTaskPhase `json:"phase"` + UserID int `json:"user_id"` + FilteredStoragePolicy []int `json:"filtered_storage_policy"` + Total int `json:"total"` + Scanned int `json:"scanned"` + Seeded int `json:"seeded"` + LastFileID int `json:"last_file_id"` + } +) + +const ( + MediaBackfillPhaseCount MediaBackfillTaskPhase = "count" + MediaBackfillPhaseSeed MediaBackfillTaskPhase = "seed" + + MediaBackfillBatchSize = 1000 + + ProgressTypeMediaBackfill = "media_backfill" + SummaryKeySeeded = "seeded" +) + +func init() { + queue.RegisterResumableTaskFactory(queue.MediaBackfillTaskType, NewMediaBackfillTaskFromModel) +} + +func NewMediaBackfillTask(ctx context.Context, u *ent.User, filteredStoragePolicy []int) (queue.Task, error) { + state := &MediaBackfillTaskState{ + Phase: MediaBackfillPhaseCount, + UserID: u.ID, + FilteredStoragePolicy: filteredStoragePolicy, + } + stateBytes, err := json.Marshal(state) + if err != nil { + return nil, fmt.Errorf("failed to marshal state: %w", err) + } + + return &MediaBackfillTask{ + DBTask: &queue.DBTask{ + Task: &ent.Task{ + Type: queue.MediaBackfillTaskType, + CorrelationID: logging.CorrelationID(ctx), + PrivateState: string(stateBytes), + PublicState: &types.TaskPublicState{}, + }, + DirectOwner: u, + }, + }, nil +} + +func NewMediaBackfillTaskFromModel(t *ent.Task) queue.Task { + return &MediaBackfillTask{ + DBTask: &queue.DBTask{ + Task: t, + }, + } +} + +func (m *MediaBackfillTask) Do(ctx context.Context) (task.Status, error) { + dep := dependency.FromContext(ctx) + m.l = dep.Logger() + + m.Lock() + if m.progress == nil { + m.progress = make(queue.Progresses) + } + m.progress[ProgressTypeMediaBackfill] = &queue.Progress{} + m.Unlock() + + state := &MediaBackfillTaskState{} + if err := json.Unmarshal([]byte(m.State()), state); err != nil { + return task.StatusError, fmt.Errorf("failed to unmarshal state: %s (%w)", err, queue.CriticalErr) + } + m.state = state + + var ( + next = task.StatusCompleted + err error + ) + switch m.state.Phase { + case MediaBackfillPhaseCount, "": + next, err = m.count(ctx, dep) + case MediaBackfillPhaseSeed: + next, err = m.seed(ctx, dep) + default: + next, err = task.StatusError, fmt.Errorf("unknown phase %q: %w", m.state.Phase, queue.CriticalErr) + } + + newStateStr, marshalErr := json.Marshal(m.state) + if marshalErr != nil { + return task.StatusError, fmt.Errorf("failed to marshal state: %w", marshalErr) + } + + m.Lock() + m.Task.PrivateState = string(newStateStr) + m.Unlock() + return next, err +} + +// count computes the number of candidate files to scan, then moves to seeding. +func (m *MediaBackfillTask) count(ctx context.Context, dep dependency.Dep) (task.Status, error) { + minSize := dep.SettingProvider().MediaProcess(ctx).MinSize + total, err := dep.FileClient().CountBackfillCandidateFiles(ctx, m.state.UserID, m.state.FilteredStoragePolicy, minSize) + if err != nil { + return task.StatusError, fmt.Errorf("failed to count backfill candidate files: %w", err) + } + + m.state.Total = total + m.state.Phase = MediaBackfillPhaseSeed + m.state.LastFileID = 0 + m.state.Scanned = 0 + m.state.Seeded = 0 + + m.l.Info("Media backfill: %d candidate file(s) to scan for user %d.", total, m.state.UserID) + m.ResumeAfter(0) + return task.StatusSuspending, nil +} + +// seed scans a batch of candidate files and enqueues each image as a pending +// media_process_task row, then suspends for the next batch. +func (m *MediaBackfillTask) seed(ctx context.Context, dep dependency.Dep) (task.Status, error) { + atomic.StoreInt64(&m.progress[ProgressTypeMediaBackfill].Total, int64(m.state.Total)) + atomic.StoreInt64(&m.progress[ProgressTypeMediaBackfill].Current, int64(m.state.Scanned)) + + minSize := dep.SettingProvider().MediaProcess(ctx).MinSize + files, err := dep.FileClient().ListBackfillCandidateFiles(ctx, m.state.LastFileID, MediaBackfillBatchSize, m.state.UserID, m.state.FilteredStoragePolicy, minSize) + if err != nil { + return task.StatusError, fmt.Errorf("failed to list backfill candidate files after ID %d: %w", m.state.LastFileID, err) + } + + if len(files) == 0 { + m.l.Info("Media backfill complete for user %d. %d image(s) enqueued.", m.state.UserID, m.state.Seeded) + return task.StatusCompleted, nil + } + + mpClient := dep.MediaProcessClient() + detector := dep.MimeDetector(ctx) + for _, f := range files { + mimeType := detector.TypeByName(f.Name) + if !strings.HasPrefix(strings.ToLower(mimeType), "image/") { + continue + } + // Skip files already handled by a previous backfill/upload pass, so a + // re-run does not re-compress the compressed output. + if handled, err := mpClient.HasHandledForFile(ctx, f.ID); err != nil { + m.l.Warning("Media backfill: handled-check failed for file %d: %s", f.ID, err) + continue + } else if handled { + continue + } + if _, err := mpClient.Enqueue(ctx, &inventory.MediaProcessEnqueueArgs{ + EntityID: f.PrimaryEntity, + FileID: f.ID, + OwnerID: f.OwnerID, + MediaType: mediaprocesstask.MediaTypeImage, + }); err != nil { + m.l.Warning("Media backfill: failed to enqueue file %d: %s", f.ID, err) + continue + } + m.state.Seeded++ + } + + m.state.Scanned += len(files) + m.state.LastFileID = files[len(files)-1].ID + atomic.StoreInt64(&m.progress[ProgressTypeMediaBackfill].Current, int64(m.state.Scanned)) + + m.ResumeAfter(0) + return task.StatusSuspending, nil +} + +func (m *MediaBackfillTask) Progress(ctx context.Context) queue.Progresses { + m.Lock() + defer m.Unlock() + return m.progress +} + +func (m *MediaBackfillTask) Summarize(hasher hashid.Encoder) *queue.Summary { + if m.state == nil { + if err := json.Unmarshal([]byte(m.State()), &m.state); err != nil { + return nil + } + } + + return &queue.Summary{ + Phase: string(m.state.Phase), + Props: map[string]any{ + SummaryKeyTotal: m.state.Total, + SummaryKeySeeded: m.state.Seeded, + }, + } +} diff --git a/pkg/queue/task.go b/pkg/queue/task.go index db38536c..f662525b 100644 --- a/pkg/queue/task.go +++ b/pkg/queue/task.go @@ -105,6 +105,7 @@ const ( RemoteDownloadTaskType = "remote_download" ImportTaskType = "import" MediaCompressTaskType = "media_compress" + MediaBackfillTaskType = "media_backfill" FullTextIndexTaskType = "full_text_index" FullTextCopyTaskType = "full_text_copy" diff --git a/routers/controllers/file.go b/routers/controllers/file.go index a956c6b2..403676b8 100644 --- a/routers/controllers/file.go +++ b/routers/controllers/file.go @@ -83,6 +83,21 @@ func RebuildFTSIndex(c *gin.Context) { }) } +// MediaBackfill enqueues a task that seeds pending compression rows for the user's existing images +func MediaBackfill(c *gin.Context) { + service := ParametersFromContext[*explorer.MediaBackfillWorkflowService](c, explorer.CreateMediaBackfillParamCtx{}) + resp, err := service.CreateMediaBackfillTask(c) + if err != nil { + c.JSON(200, serializer.Err(c, err)) + c.Abort() + return + } + + c.JSON(200, serializer.Response{ + Data: resp, + }) +} + // ExtractArchive creates extract archive task func ExtractArchive(c *gin.Context) { service := ParametersFromContext[*explorer.ArchiveWorkflowService](c, explorer.CreateArchiveParamCtx{}) diff --git a/routers/router.go b/routers/router.go index 1b9b23bf..2b6227be 100644 --- a/routers/router.go +++ b/routers/router.go @@ -766,6 +766,12 @@ func initMasterRouter(dep dependency.Dep) *gin.Engine { controllers.FromJSON[explorer.RebuildFTSIndexWorkflowService](explorer.CreateRebuildFTSIndexParamCtx{}), controllers.RebuildFTSIndex, ) + // Create task to backfill existing images for compression + wf.POST("mediaBackfill", + middleware.RequiredScopes(types.ScopeWorkflowWrite), + controllers.FromJSON[explorer.MediaBackfillWorkflowService](explorer.CreateMediaBackfillParamCtx{}), + controllers.MediaBackfill, + ) // 取得文件外链 source := file.Group("source") diff --git a/service/explorer/workflows.go b/service/explorer/workflows.go index 8bef8c95..f40441a4 100644 --- a/service/explorer/workflows.go +++ b/service/explorer/workflows.go @@ -481,3 +481,30 @@ func (service *RebuildFTSIndexWorkflowService) CreateRebuildFTSIndexTask(c *gin. return BuildTaskResponse(t, nil, hasher), nil } + +type ( + MediaBackfillWorkflowService struct { + FilteredStoragePolicy []int `json:"filtered_storage_policy"` + } + CreateMediaBackfillParamCtx struct{} +) + +// CreateMediaBackfillTask seeds pending compression rows for the caller's existing +// images (APP-102). Self-service: it backfills only the caller's own files, so no +// admin permission is required. +func (service *MediaBackfillWorkflowService) CreateMediaBackfillTask(c *gin.Context) (*TaskResponse, error) { + dep := dependency.FromContext(c) + user := inventory.UserFromContext(c) + hasher := dep.HashIDEncoder() + + t, err := workflows.NewMediaBackfillTask(c, user, service.FilteredStoragePolicy) + if err != nil { + return nil, serializer.NewError(serializer.CodeCreateTaskError, "Failed to create task", err) + } + + if err := dep.MediaMetaQueue(c).QueueTask(c, t); err != nil { + return nil, serializer.NewError(serializer.CodeCreateTaskError, "Failed to queue task", err) + } + + return BuildTaskResponse(t, nil, hasher), nil +}