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) <noreply@anthropic.com>pull/3493/head
thotenn
3 weeks ago
parent
51c7654b03
commit
3a0c8f593d
@ -1 +1 @@
|
||||
Subproject commit 74d41bbcee74020f3f071340f8cea73152f705c9
|
||||
Subproject commit 5f597e0fd4342a1727ff699e0235ce804bfd560b
|
||||
@ -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)
|
||||
}
|
||||
@ -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,
|
||||
},
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue