feat: 实现平台抽成、提现审批、异步任务集成及安全审计功能

backend/app/jobs/args/media.go

@@ -0,0 +1,19 @@
+package args
+
+import "github.com/riverqueue/river"
+
+type MediaProcessArgs struct {
+	AssetID int64 `json:"asset_id"`
+}
+
+func (MediaProcessArgs) Kind() string {
+	return "media_process"
+}
+
+func (MediaProcessArgs) InsertOpts() river.InsertOpts {
+	return river.InsertOpts{}
+}
+
+func (MediaProcessArgs) UniqueID() string {
+	return "media_process"
+}

backend/app/jobs/args/notification.go

@@ -0,0 +1,22 @@
+package args
+
+import "github.com/riverqueue/river"
+
+type NotificationArgs struct {
+	UserID  int64  `json:"user_id"`
+	Type    string `json:"type"`
+	Title   string `json:"title"`
+	Content string `json:"content"`
+}
+
+func (NotificationArgs) Kind() string {
+	return "notification"
+}
+
+func (NotificationArgs) InsertOpts() river.InsertOpts {
+	return river.InsertOpts{}
+}
+
+func (NotificationArgs) UniqueID() string {
+	return "notification"
+}

backend/app/jobs/media_process_job.go

@@ -0,0 +1,39 @@
+package jobs
+
+import (
+	"context"
+	"time"
+
+	"github.com/riverqueue/river"
+	"quyun/v2/app/jobs/args"
+	"quyun/v2/database/models"
+	"quyun/v2/pkg/consts"
+)
+
+// @provider(job)
+type MediaProcessWorker struct {
+	river.WorkerDefaults[args.MediaProcessArgs]
+}
+
+func (j *MediaProcessWorker) Work(ctx context.Context, job *river.Job[args.MediaProcessArgs]) error {
+	arg := job.Args
+	// 1. Fetch Asset
+	asset, err := models.MediaAssetQuery.WithContext(ctx).Where(models.MediaAssetQuery.ID.Eq(arg.AssetID)).First()
+	if err != nil {
+		return err
+	}
+
+	// 2. Mock Processing
+	// Update status to processing
+	_, err = models.MediaAssetQuery.WithContext(ctx).Where(models.MediaAssetQuery.ID.Eq(asset.ID)).UpdateSimple(models.MediaAssetQuery.Status.Value(consts.MediaAssetStatusProcessing))
+	if err != nil {
+		return err
+	}
+
+	// 3. Update status to ready
+	_, err = models.MediaAssetQuery.WithContext(ctx).Where(models.MediaAssetQuery.ID.Eq(asset.ID)).Updates(&models.MediaAsset{
+		Status:    consts.MediaAssetStatusReady,
+		UpdatedAt: time.Now(),
+	})
+	return err
+}

backend/app/jobs/notification_job.go

@@ -0,0 +1,27 @@
+package jobs
+
+import (
+	"context"
+
+	"quyun/v2/app/jobs/args"
+	"quyun/v2/database/models"
+
+	"github.com/riverqueue/river"
+)
+
+// @provider(job)
+type NotificationWorker struct {
+	river.WorkerDefaults[args.NotificationArgs]
+}
+
+func (j *NotificationWorker) Work(ctx context.Context, job *river.Job[args.NotificationArgs]) error {
+	arg := job.Args
+	n := &models.Notification{
+		UserID:  arg.UserID,
+		Type:    arg.Type,
+		Title:   arg.Title,
+		Content: arg.Content,
+		IsRead:  false,
+	}
+	return models.NotificationQuery.WithContext(ctx).Create(n)
+}

backend/app/jobs/provider.gen.go

@@ -1,9 +1,39 @@
 package jobs
 
 import (
+	"quyun/v2/providers/job"
+
+	"github.com/riverqueue/river"
+	"go.ipao.vip/atom"
+	"go.ipao.vip/atom/container"
+	"go.ipao.vip/atom/contracts"
 	"go.ipao.vip/atom/opt"
 )
 
 func Provide(opts ...opt.Option) error {
+	if err := container.Container.Provide(func(
+		__job *job.Job,
+	) (contracts.Initial, error) {
+		obj := &MediaProcessWorker{}
+		if err := river.AddWorkerSafely(__job.Workers, obj); err != nil {
+			return nil, err
+		}
+
+		return obj, nil
+	}, atom.GroupInitial); err != nil {
+		return err
+	}
+	if err := container.Container.Provide(func(
+		__job *job.Job,
+	) (contracts.Initial, error) {
+		obj := &NotificationWorker{}
+		if err := river.AddWorkerSafely(__job.Workers, obj); err != nil {
+			return nil, err
+		}
+
+		return obj, nil
+	}, atom.GroupInitial); err != nil {
+		return err
+	}
 	return nil
 }