feat(media-supply): add media resource supply marketplace

Introduce an end-to-end media-supply feature: tenant-side resource sync
service/worker backed by a Meijiequan supplier client, ops-side management
APIs, and admin/ops web views for resources, orders, favorites and
submission. Adds a shared digitocr helper, MediaSupply config blocks for
tenant and ops, shared types, and migrations for supplier media resources,
price overrides, customer visibility and order refunds.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/internal/ops/app/media_supply.go

@@ -0,0 +1,631 @@
+package app
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"math"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	sharedconfig "github.com/geo-platform/tenant-api/internal/shared/config"
+	"github.com/geo-platform/tenant-api/internal/shared/response"
+)
+
+const (
+	opsMediaSupplySupplierMeijiequan = "meijiequan"
+)
+
+type MediaSupplyService struct {
+	pool    *pgxpool.Pool
+	audits  *AuditService
+	cfgFunc func() sharedconfig.MediaSupplyConfig
+}
+
+func NewMediaSupplyService(pool *pgxpool.Pool, audits *AuditService, cfgFunc func() sharedconfig.MediaSupplyConfig) *MediaSupplyService {
+	return &MediaSupplyService{pool: pool, audits: audits, cfgFunc: cfgFunc}
+}
+
+type MediaSupplyResource struct {
+	ID                 int64            `json:"id"`
+	SupplierResourceID string           `json:"supplier_resource_id"`
+	ModelID            int              `json:"model_id"`
+	Name               string           `json:"name"`
+	Status             string           `json:"status"`
+	CostPriceCents     int64            `json:"cost_price_cents"`
+	CostPrices         map[string]int64 `json:"cost_prices"`
+	SellPriceCents     int64            `json:"sell_price_cents"`
+	ResourceURL        *string          `json:"resource_url,omitempty"`
+	BaiduWeight        *int             `json:"baidu_weight,omitempty"`
+	ResourceRemark     *string          `json:"resource_remark,omitempty"`
+	CustomerVisible    bool             `json:"customer_visible"`
+	ChannelType        *string          `json:"channel_type,omitempty"`
+	Region             *string          `json:"region,omitempty"`
+	InclusionEffect    *string          `json:"inclusion_effect,omitempty"`
+	LinkType           *string          `json:"link_type,omitempty"`
+	PublishRate        *string          `json:"publish_rate,omitempty"`
+	DeliverySpeed      *string          `json:"delivery_speed,omitempty"`
+	LastSyncedAt       time.Time        `json:"last_synced_at"`
+}
+
+type ListMediaSupplyResourcesInput struct {
+	ModelID         int
+	Keyword         string
+	RemarkKeyword   string
+	ChannelType     string
+	Region          string
+	InclusionEffect string
+	LinkType        string
+	MinPriceCents   int64
+	MaxPriceCents   int64
+	Visibility      string
+	Page            int
+	Size            int
+}
+
+type ListMediaSupplyResourcesResult struct {
+	Items []MediaSupplyResource `json:"items"`
+	Total int64                 `json:"total"`
+	Page  int                   `json:"page"`
+	Size  int                   `json:"size"`
+}
+
+type SetMediaSupplyPriceInput struct {
+	PriceType      string
+	SellPriceCents int64
+	Enabled        bool
+}
+
+type SetMediaSupplyVisibilityInput struct {
+	CustomerVisible bool
+}
+
+type MediaSupplyWalletStatus struct {
+	TenantID      int64      `json:"tenant_id"`
+	UserID        int64      `json:"user_id"`
+	UserName      *string    `json:"user_name,omitempty"`
+	UserPhone     *string    `json:"user_phone,omitempty"`
+	BalanceCents  int64      `json:"balance_cents"`
+	UpdatedAt     time.Time  `json:"updated_at"`
+	LastLedgerAt  *time.Time `json:"last_ledger_at,omitempty"`
+	LedgerEntries int64      `json:"ledger_entries"`
+	TenantName    *string    `json:"tenant_name,omitempty"`
+}
+
+type ListMediaSupplyWalletsInput struct {
+	Keyword  string
+	TenantID int64
+	Page     int
+	Size     int
+}
+
+type ListMediaSupplyWalletsResult struct {
+	Items []MediaSupplyWalletStatus `json:"items"`
+	Total int64                     `json:"total"`
+	Page  int                       `json:"page"`
+	Size  int                       `json:"size"`
+}
+
+type AdjustMediaSupplyWalletInput struct {
+	TenantID    int64
+	UserID      int64
+	DeltaCents  int64
+	Note        string
+	OperatorID  int64
+	OperatorTag string
+}
+
+func (s *MediaSupplyService) ListResources(ctx context.Context, input ListMediaSupplyResourcesInput) (*ListMediaSupplyResourcesResult, error) {
+	if s == nil || s.pool == nil {
+		return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用")
+	}
+	page, size := normalizeOpsMediaSupplyPagination(input.Page, input.Size)
+	args := []any{opsMediaSupplySupplierMeijiequan}
+	where := []string{"r.supplier = $1", "r.deleted_at IS NULL"}
+	cfg := s.mediaSupplyConfig()
+	sellPriceSQL := opsMediaSupplySellPriceSQL(cfg)
+	if input.ModelID > 0 {
+		args = append(args, input.ModelID)
+		where = append(where, fmt.Sprintf("r.model_id = $%d", len(args)))
+	}
+	if keyword := strings.TrimSpace(input.Keyword); keyword != "" {
+		args = append(args, "%"+keyword+"%")
+		where = append(where, fmt.Sprintf("r.name ILIKE $%d", len(args)))
+	}
+	if keyword := strings.TrimSpace(input.RemarkKeyword); keyword != "" {
+		args = append(args, "%"+keyword+"%")
+		where = append(where, fmt.Sprintf("r.resource_remark ILIKE $%d", len(args)))
+	}
+	if value := strings.TrimSpace(input.ChannelType); value != "" {
+		args = append(args, value)
+		where = append(where, fmt.Sprintf("r.channel_type = $%d", len(args)))
+	}
+	if value := strings.TrimSpace(input.Region); value != "" {
+		args = append(args, value)
+		where = append(where, fmt.Sprintf("r.region = $%d", len(args)))
+	}
+	if value := strings.TrimSpace(input.InclusionEffect); value != "" {
+		args = append(args, value)
+		where = append(where, fmt.Sprintf("r.inclusion_effect = $%d", len(args)))
+	}
+	if value := strings.TrimSpace(input.LinkType); value != "" {
+		args = append(args, value)
+		where = append(where, fmt.Sprintf("r.link_type = $%d", len(args)))
+	}
+	if input.MinPriceCents > 0 {
+		args = append(args, input.MinPriceCents)
+		where = append(where, fmt.Sprintf("(%s) >= $%d", sellPriceSQL, len(args)))
+	}
+	if input.MaxPriceCents > 0 {
+		args = append(args, input.MaxPriceCents)
+		where = append(where, fmt.Sprintf("(%s) <= $%d", sellPriceSQL, len(args)))
+	}
+	switch strings.TrimSpace(input.Visibility) {
+	case "visible":
+		where = append(where, "r.customer_visible = TRUE")
+	case "hidden":
+		where = append(where, "r.customer_visible = FALSE")
+	}
+	whereSQL := strings.Join(where, " AND ")
+	var total int64
+	if err := s.pool.QueryRow(ctx, `
+		SELECT COUNT(*)
+		FROM supplier_media_resources r
+		LEFT JOIN supplier_media_price_overrides o
+		  ON o.resource_id = r.id AND o.price_type = 'price'
+		WHERE `+whereSQL, args...).Scan(&total); err != nil {
+		return nil, response.ErrInternal(50060, "media_supply_resource_count_failed", "媒体资源统计失败")
+	}
+	args = append(args, size, (page-1)*size)
+	rows, err := s.pool.Query(ctx, `
+		SELECT r.id, r.supplier_resource_id, r.model_id, r.name, r.status,
+		       r.cost_price_cents, r.cost_prices_json,
+		       `+sellPriceSQL+` AS sell_price_cents,
+		       r.resource_url, r.baidu_weight, r.resource_remark, r.customer_visible,
+		       r.channel_type, r.region, r.inclusion_effect, r.link_type,
+		       r.publish_rate, r.delivery_speed, r.last_synced_at
+		FROM supplier_media_resources r
+		LEFT JOIN supplier_media_price_overrides o
+		  ON o.resource_id = r.id AND o.price_type = 'price'
+		WHERE `+whereSQL+`
+		ORDER BY r.updated_at DESC, r.id DESC
+		LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args))+`
+	`, args...)
+	if err != nil {
+		return nil, response.ErrInternal(50061, "media_supply_resource_query_failed", "媒体资源读取失败")
+	}
+	defer rows.Close()
+	items := make([]MediaSupplyResource, 0, size)
+	for rows.Next() {
+		var item MediaSupplyResource
+		var costPricesRaw []byte
+		if err := rows.Scan(
+			&item.ID,
+			&item.SupplierResourceID,
+			&item.ModelID,
+			&item.Name,
+			&item.Status,
+			&item.CostPriceCents,
+			&costPricesRaw,
+			&item.SellPriceCents,
+			&item.ResourceURL,
+			&item.BaiduWeight,
+			&item.ResourceRemark,
+			&item.CustomerVisible,
+			&item.ChannelType,
+			&item.Region,
+			&item.InclusionEffect,
+			&item.LinkType,
+			&item.PublishRate,
+			&item.DeliverySpeed,
+			&item.LastSyncedAt,
+		); err != nil {
+			return nil, response.ErrInternal(50062, "media_supply_resource_scan_failed", "媒体资源解析失败")
+		}
+		_ = json.Unmarshal(costPricesRaw, &item.CostPrices)
+		if item.CostPrices == nil {
+			item.CostPrices = map[string]int64{}
+		}
+		items = append(items, item)
+	}
+	return &ListMediaSupplyResourcesResult{Items: items, Total: total, Page: page, Size: size}, nil
+}
+
+func (s *MediaSupplyService) SetResourcePrice(ctx context.Context, actor *Actor, resourceID int64, input SetMediaSupplyPriceInput) error {
+	priceType := normalizeOpsMediaSupplyPriceType(input.PriceType)
+	if input.SellPriceCents < 0 {
+		return response.ErrBadRequest(40061, "invalid_media_supply_price", "媒体售价不能为负数")
+	}
+	cost, err := s.resourceCostForPriceType(ctx, resourceID, priceType)
+	if err != nil {
+		return err
+	}
+	sellPriceCents := opsMediaSupplyRoundUpToYuan(input.SellPriceCents)
+	if sellPriceCents < cost {
+		return response.ErrConflict(40960, "media_supply_price_below_cost", "媒体售价不能低于成本价")
+	}
+	if _, err := s.pool.Exec(ctx, `
+		INSERT INTO supplier_media_price_overrides (resource_id, price_type, sell_price_cents, enabled, updated_by)
+		VALUES ($1, $2, $3, $4, NULL)
+		ON CONFLICT (resource_id, price_type) DO UPDATE SET
+		    sell_price_cents = EXCLUDED.sell_price_cents,
+		    enabled = EXCLUDED.enabled,
+		    updated_by = NULL,
+		    updated_at = NOW()
+	`, resourceID, priceType, sellPriceCents, input.Enabled); err != nil {
+		return response.ErrInternal(50064, "media_supply_price_update_failed", "媒体价格更新失败")
+	}
+	if s.audits != nil && actor != nil {
+		_ = s.audits.Append(ctx, actor.audit("media_supply.price_update", "supplier_media_resource", resourceID, map[string]any{
+			"price_type":       priceType,
+			"sell_price_cents": sellPriceCents,
+			"enabled":          input.Enabled,
+		}))
+	}
+	return nil
+}
+
+func (s *MediaSupplyService) SetResourceVisibility(ctx context.Context, actor *Actor, resourceID int64, input SetMediaSupplyVisibilityInput) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE supplier_media_resources
+		SET customer_visible = $3, updated_at = NOW()
+		WHERE id = $1 AND supplier = $2 AND deleted_at IS NULL
+	`, resourceID, opsMediaSupplySupplierMeijiequan, input.CustomerVisible)
+	if err != nil {
+		return response.ErrInternal(50087, "media_supply_visibility_update_failed", "媒体展示状态更新失败")
+	}
+	if tag.RowsAffected() == 0 {
+		return response.ErrNotFound(40460, "media_supply_resource_not_found", "媒体资源不存在或已下架")
+	}
+	if s.audits != nil && actor != nil {
+		action := "media_supply.resource_show"
+		if !input.CustomerVisible {
+			action = "media_supply.resource_hide"
+		}
+		_ = s.audits.Append(ctx, actor.audit(action, "supplier_media_resource", resourceID, map[string]any{
+			"customer_visible": input.CustomerVisible,
+		}))
+	}
+	return nil
+}
+
+func (s *MediaSupplyService) ListWallets(ctx context.Context, input ListMediaSupplyWalletsInput) (*ListMediaSupplyWalletsResult, error) {
+	page, size := normalizeOpsMediaSupplyPagination(input.Page, input.Size)
+	args := make([]any, 0)
+	where := []string{"tm.deleted_at IS NULL"}
+	if input.TenantID > 0 {
+		args = append(args, input.TenantID)
+		where = append(where, fmt.Sprintf("tm.tenant_id = $%d", len(args)))
+	}
+	if keyword := strings.TrimSpace(input.Keyword); keyword != "" {
+		args = append(args, "%"+keyword+"%")
+		where = append(where, fmt.Sprintf("(u.phone ILIKE $%d OR COALESCE(u.email, '') ILIKE $%d OR COALESCE(u.name, '') ILIKE $%d OR COALESCE(t.name, '') ILIKE $%d)", len(args), len(args), len(args), len(args)))
+	}
+	whereSQL := strings.Join(where, " AND ")
+	var total int64
+	if err := s.pool.QueryRow(ctx, `
+		SELECT COUNT(*)
+		FROM tenant_memberships tm
+		JOIN users u ON u.id = tm.user_id AND u.deleted_at IS NULL
+		LEFT JOIN tenants t ON t.id = tm.tenant_id AND t.deleted_at IS NULL
+		WHERE `+whereSQL, args...).Scan(&total); err != nil {
+		return nil, response.ErrInternal(50088, "media_supply_wallet_count_failed", "媒体投稿余额统计失败")
+	}
+	args = append(args, size, (page-1)*size)
+	rows, err := s.pool.Query(ctx, `
+		SELECT tm.tenant_id, t.name, u.id, u.name, u.phone, COALESCE(w.balance_cents, 0),
+		       COALESCE(w.updated_at, u.updated_at), COUNT(l.id), MAX(l.created_at)
+		FROM tenant_memberships tm
+		JOIN users u ON u.id = tm.user_id AND u.deleted_at IS NULL
+		LEFT JOIN tenants t ON t.id = tm.tenant_id AND t.deleted_at IS NULL
+		LEFT JOIN media_supply_user_wallets w ON w.tenant_id = tm.tenant_id AND w.user_id = tm.user_id
+		LEFT JOIN media_supply_wallet_ledgers l ON l.tenant_id = tm.tenant_id AND l.user_id = tm.user_id
+		WHERE `+whereSQL+`
+		GROUP BY tm.tenant_id, t.name, u.id, u.name, u.phone, w.balance_cents, w.updated_at, u.updated_at
+		ORDER BY COALESCE(MAX(l.created_at), COALESCE(w.updated_at, u.updated_at)) DESC, u.id DESC
+		LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args))+`
+	`, args...)
+	if err != nil {
+		return nil, response.ErrInternal(50089, "media_supply_wallet_query_failed", "媒体投稿余额读取失败")
+	}
+	defer rows.Close()
+	items := make([]MediaSupplyWalletStatus, 0, size)
+	for rows.Next() {
+		var item MediaSupplyWalletStatus
+		var tenantName sql.NullString
+		var name sql.NullString
+		var phone sql.NullString
+		var lastLedgerAt sql.NullTime
+		if err := rows.Scan(
+			&item.TenantID,
+			&tenantName,
+			&item.UserID,
+			&name,
+			&phone,
+			&item.BalanceCents,
+			&item.UpdatedAt,
+			&item.LedgerEntries,
+			&lastLedgerAt,
+		); err != nil {
+			return nil, response.ErrInternal(50090, "media_supply_wallet_scan_failed", "媒体投稿余额解析失败")
+		}
+		if name.Valid {
+			item.UserName = &name.String
+		}
+		if tenantName.Valid {
+			item.TenantName = &tenantName.String
+		}
+		if phone.Valid {
+			item.UserPhone = &phone.String
+		}
+		if lastLedgerAt.Valid {
+			item.LastLedgerAt = &lastLedgerAt.Time
+		}
+		items = append(items, item)
+	}
+	return &ListMediaSupplyWalletsResult{Items: items, Total: total, Page: page, Size: size}, nil
+}
+
+func (s *MediaSupplyService) QueueSync(ctx context.Context, actor *Actor, modelID int) (int64, error) {
+	if s == nil || s.pool == nil {
+		return 0, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用")
+	}
+	if modelID <= 0 {
+		modelID = 1
+	}
+	if !isKnownOpsMeijiequanModel(modelID) {
+		return 0, response.ErrBadRequest(40060, "invalid_media_supply_model", "不支持的媒体资源类型")
+	}
+	var id int64
+	if err := s.pool.QueryRow(ctx, `
+		WITH existing AS (
+		    SELECT id
+		    FROM media_supply_sync_jobs
+		    WHERE supplier = $1
+		      AND status IN ($3::varchar, $4::varchar)
+		    ORDER BY created_at ASC
+		    LIMIT 1
+		), inserted AS (
+		    INSERT INTO media_supply_sync_jobs (supplier, model_id, requested_by)
+		    SELECT $1, $2, NULL
+		    WHERE NOT EXISTS (SELECT 1 FROM existing)
+		    RETURNING id
+		)
+		SELECT id FROM inserted
+		UNION ALL
+		SELECT id FROM existing
+		LIMIT 1
+	`, opsMediaSupplySupplierMeijiequan, modelID, "queued", "running").Scan(&id); err != nil {
+		return 0, response.ErrInternal(50063, "media_supply_sync_queue_failed", "媒体资源同步排队失败")
+	}
+	if s.audits != nil && actor != nil {
+		_ = s.audits.Append(ctx, actor.audit("media_supply.sync_queue", "media_supply_sync_job", id, map[string]any{
+			"model_id": modelID,
+		}))
+	}
+	return id, nil
+}
+
+func (s *MediaSupplyService) AdjustWallet(ctx context.Context, actor *Actor, input AdjustMediaSupplyWalletInput) (*MediaSupplyWalletStatus, error) {
+	if input.TenantID <= 0 || input.UserID <= 0 {
+		return nil, response.ErrBadRequest(40067, "invalid_media_supply_wallet_user", "租户或用户参数不正确")
+	}
+	if input.DeltaCents == 0 {
+		return nil, response.ErrBadRequest(40068, "invalid_media_supply_wallet_delta", "调整金额不能为 0")
+	}
+	if err := s.ensureWalletUserInTenant(ctx, input.TenantID, input.UserID); err != nil {
+		return nil, err
+	}
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, response.ErrInternal(50082, "media_supply_wallet_begin_failed", "媒体投稿余额调整失败")
+	}
+	defer tx.Rollback(ctx)
+	balance, err := s.lockWallet(ctx, tx, input.TenantID, input.UserID)
+	if err != nil {
+		return nil, err
+	}
+	balanceAfter := balance + input.DeltaCents
+	if balanceAfter < 0 {
+		return nil, response.ErrConflict(40965, "media_supply_balance_insufficient", "媒体投稿余额不足")
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE media_supply_user_wallets
+		SET balance_cents = $3, updated_at = NOW()
+		WHERE tenant_id = $1 AND user_id = $2
+	`, input.TenantID, input.UserID, balanceAfter); err != nil {
+		return nil, response.ErrInternal(50083, "media_supply_wallet_update_failed", "媒体投稿余额更新失败")
+	}
+	reason := "adjustment"
+	if input.DeltaCents > 0 {
+		reason = "recharge"
+	}
+	note := strings.TrimSpace(input.Note)
+	if tag := strings.TrimSpace(input.OperatorTag); tag != "" {
+		if note == "" {
+			note = tag
+		} else {
+			note = tag + "；" + note
+		}
+	}
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO media_supply_wallet_ledgers (
+		    tenant_id, user_id, delta_cents, balance_after_cents, reason, note, created_by
+		)
+		VALUES ($1, $2, $3, $4, $5, $6, NULL)
+	`, input.TenantID, input.UserID, input.DeltaCents, balanceAfter, reason, nullableOpsMediaSupplyString(note)); err != nil {
+		return nil, response.ErrInternal(50084, "media_supply_wallet_ledger_create_failed", "媒体投稿账单创建失败")
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, response.ErrInternal(50085, "media_supply_wallet_commit_failed", "媒体投稿余额调整失败")
+	}
+	if s.audits != nil && actor != nil {
+		_ = s.audits.Append(ctx, actor.audit("media_supply.wallet_adjust", "media_supply_wallet", input.UserID, map[string]any{
+			"tenant_id":   input.TenantID,
+			"user_id":     input.UserID,
+			"delta_cents": input.DeltaCents,
+			"note":        input.Note,
+		}))
+	}
+	return &MediaSupplyWalletStatus{
+		TenantID:     input.TenantID,
+		UserID:       input.UserID,
+		BalanceCents: balanceAfter,
+		UpdatedAt:    time.Now().UTC(),
+	}, nil
+}
+
+type opsMediaSupplyWalletExecutor interface {
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+}
+
+func (s *MediaSupplyService) ensureWalletRow(ctx context.Context, q opsMediaSupplyWalletExecutor, tenantID, userID int64) error {
+	_, err := q.Exec(ctx, `
+		INSERT INTO media_supply_user_wallets (tenant_id, user_id, balance_cents)
+		VALUES ($1, $2, 0)
+		ON CONFLICT (tenant_id, user_id) DO NOTHING
+	`, tenantID, userID)
+	return err
+}
+
+func (s *MediaSupplyService) ensureWalletUserInTenant(ctx context.Context, tenantID, userID int64) error {
+	var exists bool
+	if err := s.pool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1
+			FROM tenant_memberships
+			WHERE tenant_id = $1 AND user_id = $2 AND deleted_at IS NULL
+		)
+	`, tenantID, userID).Scan(&exists); err != nil {
+		return response.ErrInternal(50086, "media_supply_wallet_user_lookup_failed", "媒体投稿余额用户校验失败")
+	}
+	if !exists {
+		return response.ErrNotFound(40463, "media_supply_wallet_user_not_found", "用户不存在或不属于当前租户")
+	}
+	return nil
+}
+
+func (s *MediaSupplyService) lockWallet(ctx context.Context, tx pgx.Tx, tenantID, userID int64) (int64, error) {
+	if err := s.ensureWalletRow(ctx, tx, tenantID, userID); err != nil {
+		return 0, response.ErrInternal(50077, "media_supply_wallet_prepare_failed", "媒体投稿余额初始化失败")
+	}
+	var balance int64
+	if err := tx.QueryRow(ctx, `
+		SELECT balance_cents
+		FROM media_supply_user_wallets
+		WHERE tenant_id = $1 AND user_id = $2
+		FOR UPDATE
+	`, tenantID, userID).Scan(&balance); err != nil {
+		return 0, response.ErrInternal(50078, "media_supply_wallet_lookup_failed", "媒体投稿余额读取失败")
+	}
+	return balance, nil
+}
+
+func (s *MediaSupplyService) resourceCostForPriceType(ctx context.Context, resourceID int64, priceType string) (int64, error) {
+	var costPrice int64
+	var costPricesRaw []byte
+	if err := s.pool.QueryRow(ctx, `
+		SELECT cost_price_cents, cost_prices_json
+		FROM supplier_media_resources
+		WHERE id = $1 AND supplier = $2 AND deleted_at IS NULL
+	`, resourceID, opsMediaSupplySupplierMeijiequan).Scan(&costPrice, &costPricesRaw); err != nil {
+		if err == pgx.ErrNoRows {
+			return 0, response.ErrNotFound(40460, "media_supply_resource_not_found", "媒体资源不存在或已下架")
+		}
+		return 0, response.ErrInternal(50071, "media_supply_resource_lookup_failed", "媒体资源读取失败")
+	}
+	var costPrices map[string]int64
+	_ = json.Unmarshal(costPricesRaw, &costPrices)
+	if priceTypeCost := costPrices[priceType]; priceTypeCost > 0 {
+		return priceTypeCost, nil
+	}
+	return costPrice, nil
+}
+
+func (s *MediaSupplyService) mediaSupplyConfig() sharedconfig.MediaSupplyConfig {
+	if s != nil && s.cfgFunc != nil {
+		cfg := s.cfgFunc()
+		sharedconfig.NormalizeMediaSupplyConfig(&cfg)
+		return cfg
+	}
+	var cfg sharedconfig.MediaSupplyConfig
+	sharedconfig.NormalizeMediaSupplyConfig(&cfg)
+	return cfg
+}
+
+func normalizeOpsMediaSupplyPagination(page, size int) (int, int) {
+	if page <= 0 {
+		page = 1
+	}
+	if size <= 0 {
+		size = 20
+	}
+	if size > 100 {
+		size = 100
+	}
+	return page, size
+}
+
+func normalizeOpsMediaSupplyPriceType(priceType string) string {
+	priceType = strings.ToLower(strings.TrimSpace(priceType))
+	if priceType == "" {
+		return "price"
+	}
+	return priceType
+}
+
+func nullableOpsMediaSupplyString(value string) *string {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return nil
+	}
+	return &value
+}
+
+func opsMediaSupplySellPrice(costCents, overrideCents int64, overrideEnabled bool, markupPercent float64, minimumMarkupCents int64) int64 {
+	floor := costCents
+	if markupPercent > 0 {
+		floor = int64(math.Ceil(float64(costCents) * (1 + markupPercent/100)))
+	}
+	if minimumMarkupCents > 0 && costCents+minimumMarkupCents > floor {
+		floor = costCents + minimumMarkupCents
+	}
+	if overrideEnabled && overrideCents >= costCents {
+		return opsMediaSupplyRoundUpToYuan(overrideCents)
+	}
+	return opsMediaSupplyRoundUpToYuan(floor)
+}
+
+func opsMediaSupplyRoundUpToYuan(cents int64) int64 {
+	if cents <= 0 {
+		return 0
+	}
+	return ((cents + 99) / 100) * 100
+}
+
+func opsMediaSupplySellPriceSQL(cfg sharedconfig.MediaSupplyConfig) string {
+	floorSQL := "r.cost_price_cents"
+	if cfg.DefaultMarkupPercent > 0 {
+		multiplier := 1 + cfg.DefaultMarkupPercent/100
+		floorSQL = fmt.Sprintf("CEIL(r.cost_price_cents * %.8f)::BIGINT", multiplier)
+	}
+	if cfg.MinimumMarkupCents > 0 {
+		floorSQL = fmt.Sprintf("GREATEST(%s, r.cost_price_cents + %d)", floorSQL, cfg.MinimumMarkupCents)
+	}
+	rawPriceSQL := fmt.Sprintf("CASE WHEN COALESCE(o.enabled, false) AND COALESCE(o.sell_price_cents, 0) >= r.cost_price_cents THEN o.sell_price_cents ELSE (%s) END", floorSQL)
+	return fmt.Sprintf("CEIL((%s) / 100.0)::BIGINT * 100", rawPriceSQL)
+}
+
+func isKnownOpsMeijiequanModel(modelID int) bool {
+	switch modelID {
+	case 1, 2, 10, 3, 4, 7, 5, 6, 9, 11, 13:
+		return true
+	default:
+		return false
+	}
+}

server/internal/ops/config/config.go

@@ -28,6 +28,7 @@ type Config struct {
 	RabbitMQ           sharedconfig.RabbitMQConfig      `mapstructure:"rabbitmq"`
 	ObjectStorage      sharedconfig.ObjectStorageConfig `mapstructure:"object_storage"`
 	Log                sharedconfig.LogConfig           `mapstructure:"log"`
+	MediaSupply        sharedconfig.MediaSupplyConfig   `mapstructure:"media_supply"`
 	JWT                JWTConfig                        `mapstructure:"jwt"`
 	Auth               sharedconfig.AuthConfig          `mapstructure:"auth"`
 	DefaultAdmin       DefaultAdminConfig               `mapstructure:"default_admin"`
@@ -244,6 +245,12 @@ func Diff(previous, current *Config) []FieldChange {
 	if previous.Log != current.Log {
 		add("log", false)
 	}
+	if previous.MediaSupply.Worker.PollInterval != current.MediaSupply.Worker.PollInterval ||
+		previous.MediaSupply.Worker.BatchSize != current.MediaSupply.Worker.BatchSize {
+		add("media_supply.worker", false)
+	} else if !reflect.DeepEqual(previous.MediaSupply, current.MediaSupply) {
+		add("media_supply", true)
+	}
 	if previous.IPRegion != current.IPRegion {
 		add("ip_region", false)
 	}
@@ -360,6 +367,7 @@ func normalizeConfig(cfg *Config) {
 	}
 	sharedconfig.NormalizeCacheConfig(&cfg.Cache)
 	normalizeOpsRabbitMQConfig(&cfg.RabbitMQ)
+	sharedconfig.NormalizeMediaSupplyConfig(&cfg.MediaSupply)
 	if strings.TrimSpace(cfg.Log.Level) == "" {
 		cfg.Log.Level = "info"
 	}
@@ -395,6 +403,21 @@ func defaultSettings() map[string]any {
 			"driver": "redis",
 		},
 		"rabbitmq": map[string]any{},
+		"media_supply": map[string]any{
+			"enabled":                true,
+			"default_markup_percent": 50,
+			"minimum_markup_cents":   0,
+			"worker": map[string]any{
+				"enabled":            true,
+				"poll_interval":      5 * time.Second,
+				"batch_size":         1,
+				"order_max_attempts": 3,
+				"sync_max_attempts":  2,
+			},
+			"meijiequan": map[string]any{
+				"base_url": "http://www.meijiequan.com",
+			},
+		},
 		"log": map[string]any{
 			"level":  "info",
 			"format": "json",

server/internal/ops/transport/media_supply_handler.go

@@ -0,0 +1,201 @@
+package transport
+
+import (
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/geo-platform/tenant-api/internal/ops/app"
+	"github.com/geo-platform/tenant-api/internal/shared/response"
+)
+
+type setMediaSupplyPriceRequest struct {
+	PriceType      string `json:"price_type"`
+	SellPriceCents int64  `json:"sell_price_cents"`
+	Enabled        *bool  `json:"enabled,omitempty"`
+}
+
+type setMediaSupplyVisibilityRequest struct {
+	CustomerVisible bool `json:"customer_visible"`
+}
+
+type adjustMediaSupplyWalletRequest struct {
+	TenantID   int64  `json:"tenant_id" binding:"required"`
+	UserID     int64  `json:"user_id" binding:"required"`
+	DeltaCents int64  `json:"delta_cents" binding:"required"`
+	Note       string `json:"note"`
+}
+
+func listMediaSupplyResourcesHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		modelID, err := parseOptionalOpsIntQuery(c, "model_id")
+		if err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_model_id", "媒体资源类型参数必须是数字"))
+			return
+		}
+		minPriceCents, err := parseOptionalOpsInt64Query(c, "min_price_cents")
+		if err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_min_price_cents", "最低价格参数必须是数字"))
+			return
+		}
+		maxPriceCents, err := parseOptionalOpsInt64Query(c, "max_price_cents")
+		if err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_max_price_cents", "最高价格参数必须是数字"))
+			return
+		}
+		page, _ := parseOptionalOpsIntQuery(c, "page")
+		size, _ := parseOptionalOpsIntQuery(c, "page_size")
+		result, err := svc.ListResources(c.Request.Context(), app.ListMediaSupplyResourcesInput{
+			ModelID:         modelID,
+			Keyword:         c.Query("keyword"),
+			RemarkKeyword:   c.Query("remark_keyword"),
+			ChannelType:     c.Query("channel_type"),
+			Region:          c.Query("region"),
+			InclusionEffect: c.Query("inclusion_effect"),
+			LinkType:        c.Query("link_type"),
+			MinPriceCents:   minPriceCents,
+			MaxPriceCents:   maxPriceCents,
+			Visibility:      c.Query("visibility"),
+			Page:            page,
+			Size:            size,
+		})
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, result)
+	}
+}
+
+func setMediaSupplyResourcePriceHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		id, err := parseIDParam(c)
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		var body setMediaSupplyPriceRequest
+		if err := c.ShouldBindJSON(&body); err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_payload", "价格参数不正确"))
+			return
+		}
+		enabled := true
+		if body.Enabled != nil {
+			enabled = *body.Enabled
+		}
+		if err := svc.SetResourcePrice(c.Request.Context(), actorFromGin(c), id, app.SetMediaSupplyPriceInput{
+			PriceType:      body.PriceType,
+			SellPriceCents: body.SellPriceCents,
+			Enabled:        enabled,
+		}); err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, gin.H{"updated": true})
+	}
+}
+
+func setMediaSupplyResourceVisibilityHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		id, err := parseIDParam(c)
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		var body setMediaSupplyVisibilityRequest
+		if err := c.ShouldBindJSON(&body); err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_payload", "展示状态参数不正确"))
+			return
+		}
+		if err := svc.SetResourceVisibility(c.Request.Context(), actorFromGin(c), id, app.SetMediaSupplyVisibilityInput{
+			CustomerVisible: body.CustomerVisible,
+		}); err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, gin.H{"updated": true})
+	}
+}
+
+func queueMediaSupplySyncHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		var body struct {
+			ModelID int `json:"model_id"`
+		}
+		if c.Request.Body != nil {
+			_ = c.ShouldBindJSON(&body)
+		}
+		jobID, err := svc.QueueSync(c.Request.Context(), actorFromGin(c), body.ModelID)
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, gin.H{"job_id": jobID})
+	}
+}
+
+func listMediaSupplyWalletsHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		tenantID, err := parseOptionalOpsInt64Query(c, "tenant_id")
+		if err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_tenant_id", "租户参数必须是数字"))
+			return
+		}
+		page, _ := parseOptionalOpsIntQuery(c, "page")
+		size, _ := parseOptionalOpsIntQuery(c, "page_size")
+		result, err := svc.ListWallets(c.Request.Context(), app.ListMediaSupplyWalletsInput{
+			Keyword:  c.Query("keyword"),
+			TenantID: tenantID,
+			Page:     page,
+			Size:     size,
+		})
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, result)
+	}
+}
+
+func adjustMediaSupplyWalletHandler(svc *app.MediaSupplyService) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		var body adjustMediaSupplyWalletRequest
+		if err := c.ShouldBindJSON(&body); err != nil {
+			response.Error(c, response.ErrBadRequest(40001, "invalid_payload", "余额调整参数不正确"))
+			return
+		}
+		actor := actorFromGin(c)
+		operatorTag := "ops manual adjustment"
+		if actor != nil && actor.DisplayName != "" {
+			operatorTag = "ops " + actor.DisplayName
+		}
+		result, err := svc.AdjustWallet(c.Request.Context(), actor, app.AdjustMediaSupplyWalletInput{
+			TenantID:    body.TenantID,
+			UserID:      body.UserID,
+			DeltaCents:  body.DeltaCents,
+			Note:        body.Note,
+			OperatorTag: operatorTag,
+		})
+		if err != nil {
+			response.Error(c, err)
+			return
+		}
+		response.Success(c, result)
+	}
+}
+
+func parseOptionalOpsIntQuery(c *gin.Context, key string) (int, error) {
+	raw := c.Query(key)
+	if raw == "" {
+		return 0, nil
+	}
+	return strconv.Atoi(raw)
+}
+
+func parseOptionalOpsInt64Query(c *gin.Context, key string) (int64, error) {
+	raw := c.Query(key)
+	if raw == "" {
+		return 0, nil
+	}
+	return strconv.ParseInt(raw, 10, 64)
+}

server/internal/ops/transport/router.go

@@ -30,6 +30,7 @@ type Deps struct {
 	ObjectStorage *app.ObjectStorageService
 	Compliance    *app.ComplianceService
 	Scheduler     *app.SchedulerService
+	MediaSupply   *app.MediaSupplyService
 }
 
 func (d Deps) ServerAllowedOrigins() []string {
@@ -123,6 +124,13 @@ func RegisterRoutes(d Deps) {
 		authed.POST("/jobs/:source/:id/retry", retryJobHandler(d.Jobs))
 		authed.POST("/jobs/:source/:id/cancel", cancelJobHandler(d.Jobs))
 
+		authed.GET("/media-supply/resources", listMediaSupplyResourcesHandler(d.MediaSupply))
+		authed.PUT("/media-supply/resources/:id/price", setMediaSupplyResourcePriceHandler(d.MediaSupply))
+		authed.PUT("/media-supply/resources/:id/visibility", setMediaSupplyResourceVisibilityHandler(d.MediaSupply))
+		authed.POST("/media-supply/sync-jobs", queueMediaSupplySyncHandler(d.MediaSupply))
+		authed.GET("/media-supply/wallets", listMediaSupplyWalletsHandler(d.MediaSupply))
+		authed.POST("/media-supply/wallets/adjustments", adjustMediaSupplyWalletHandler(d.MediaSupply))
+
 		authed.GET("/scheduler/jobs", listSchedulerJobsHandler(d.Scheduler))
 		authed.GET("/scheduler/jobs/:key", getSchedulerJobHandler(d.Scheduler))
 		authed.PATCH("/scheduler/jobs/:key", updateSchedulerJobHandler(d.Scheduler))