feat(media-supply): resource cache sync and order flow refinements

Add a resource cache sync entrypoint (RunMediaResourceSyncOnce) driven by the new scheduler job, and rework the backlink/sync worker around it. Tune the meijiequan client and slow the default backlink sync interval from 10m to 30m to reduce upstream pressure. Trim unused public/transport surface (handler + router + swagger) and simplify admin-web order management and ops-web media-supply views, dropping stale shared-types fields.
2026-06-02 14:50:36 +08:00
parent 842782b3dd
commit 723c3ffb86
14 changed files with 529 additions and 208 deletions
@@ -22,6 +22,7 @@ import (

 type MediaSupplyService struct {
 	pool   *pgxpool.Pool
+	redis  *goredis.Client
 	logger *zap.Logger
 	cfg    config.Provider
 	client *MeijiequanClient
@@ -35,6 +36,7 @@ func NewMediaSupplyService(pool *pgxpool.Pool, redis *goredis.Client, logger *za
 	client := NewMeijiequanClient(redis, current.MediaSupply.Meijiequan)
 	return &MediaSupplyService{
 		pool:   pool,
+		redis:  redis,
 		logger: logger,
 		cfg:    cfg,
 		client: client,
@@ -403,19 +405,6 @@ func (s *MediaSupplyService) SessionStatus(ctx context.Context) (*MeijiequanSess
 	return status, nil
 }

-func (s *MediaSupplyService) SyncBacklinks(ctx context.Context) (*SyncMediaSupplyBacklinksResponse, error) {
-	if !isTenantAdmin(auth.MustActor(ctx)) {
-		return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限")
-	}
-	_ = s.mediaSupplyConfig()
-	worker := NewMediaSupplyWorker(s, s.pool, s.logger)
-	result, err := worker.syncPublishedBacklinksNow(ctx)
-	if err != nil {
-		return nil, mediaSupplyClientError(err)
-	}
-	return result, nil
-}
-
 func (s *MediaSupplyService) SetResourcePrice(ctx context.Context, resourceID int64, req SetSupplierMediaPriceRequest) error {
 	if actor, ok := auth.ActorFromCtx(ctx); ok && !isTenantAdmin(actor) {
 		return response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限")
@@ -982,18 +971,34 @@ func (s *MediaSupplyService) GetOrder(ctx context.Context, orderID int64) (*Medi
 	return &item, nil
 }

+type mediaResourceSyncStats struct {
+	Synced               int
+	RemovedPriceOverride int
+}
+
 func (s *MediaSupplyService) UpsertSyncedResources(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (int, error) {
+	stats, err := s.upsertSyncedResourcesWithStats(ctx, resources)
+	return stats.Synced, err
+}
+
+func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (mediaResourceSyncStats, error) {
 	if len(resources) == 0 {
-		return 0, nil
+		return mediaResourceSyncStats{}, nil
 	}
-	count := 0
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return mediaResourceSyncStats{}, err
+	}
+	defer tx.Rollback(ctx)
+	stats := mediaResourceSyncStats{}
 	for _, resource := range resources {
 		raw := compactJSON(resource.Raw)
 		costPrices, _ := json.Marshal(resource.CostPrices)
 		if len(costPrices) == 0 {
 			costPrices = []byte(`{}`)
 		}
-		if _, err := s.pool.Exec(ctx, `
+		var resourceID int64
+		if err := tx.QueryRow(ctx, `
 			INSERT INTO supplier_media_resources (
 			    supplier, supplier_resource_id, model_id, name, status, cost_price_cents,
 			    cost_prices_json, sale_price_label, resource_url, baidu_weight, resource_remark,
@@ -1022,18 +1027,111 @@ func (s *MediaSupplyService) UpsertSyncedResources(ctx context.Context, resource
 			    raw_json = EXCLUDED.raw_json,
 			    updated_at = NOW(),
 			    deleted_at = NULL
+			RETURNING id
 		`, resource.Supplier, resource.SupplierResourceID, resource.ModelID, resource.Name, resource.Status,
 			resource.CostPriceCents, costPrices, nullableTrimmedString(resource.SalePriceLabel),
 			nullableTrimmedString(resource.ResourceURL), resource.BaiduWeight, nullableTrimmedString(resource.ResourceRemark),
 			nullableTrimmedString(resource.ChannelType), nullableTrimmedString(resource.Region),
 			nullableTrimmedString(resource.InclusionEffect), nullableTrimmedString(resource.LinkType),
 			nullableTrimmedString(resource.PublishRate), nullableTrimmedString(resource.DeliverySpeed),
-			resource.SupplierUpdatedAt, raw); err != nil {
-			return count, err
+			resource.SupplierUpdatedAt, raw).Scan(&resourceID); err != nil {
+			return stats, err
 		}
-		count++
+		removed, err := s.deletePriceOverridesBelowCost(ctx, tx, resourceID)
+		if err != nil {
+			return stats, err
+		}
+		stats.RemovedPriceOverride += removed
+		stats.Synced++
 	}
-	return count, nil
+	if err := tx.Commit(ctx); err != nil {
+		return stats, err
+	}
+	return stats, nil
+}
+
+func (s *MediaSupplyService) deletePriceOverridesBelowCost(ctx context.Context, q interface {
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+}, resourceID int64) (int, error) {
+	if s == nil || q == nil || resourceID <= 0 {
+		return 0, nil
+	}
+	tag, err := q.Exec(ctx, `
+		DELETE FROM supplier_media_price_overrides o
+		USING supplier_media_resources r
+		WHERE o.resource_id = r.id
+		  AND r.id = $1
+		  AND r.supplier = $2
+		  AND r.deleted_at IS NULL
+		  AND o.enabled = TRUE
+		  AND o.sell_price_cents < COALESCE(
+		        CASE
+		          WHEN o.price_type = 'price' THEN COALESCE(NULLIF(r.cost_prices_json->>'price', '')::bigint, NULLIF(r.cost_prices_json->>'sale_price', '')::bigint)
+		          ELSE NULLIF(r.cost_prices_json->>o.price_type, '')::bigint
+		        END,
+		        r.cost_price_cents
+		      )
+	`, resourceID, mediaSupplySupplierMeijiequan)
+	return int(tag.RowsAffected()), err
+}
+
+func (s *MediaSupplyService) RunMediaResourceSyncOnce(ctx context.Context, modelID int) (map[string]any, error) {
+	if s == nil || s.pool == nil || s.client == nil {
+		return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用")
+	}
+	cfg := s.mediaSupplyConfig()
+	if !cfg.Enabled {
+		return map[string]any{"synced": 0, "model_id": modelID, "disabled": true}, nil
+	}
+	if modelID <= 0 {
+		modelID = 1
+	}
+	if status, statusErr := s.client.SessionStatus(ctx); statusErr != nil || status == nil || !status.Available {
+		if loginStatus, loginErr := s.client.LoginWithConfiguredAccount(ctx); loginErr != nil {
+			return nil, mediaSupplyClientError(loginErr)
+		} else if loginStatus == nil || !loginStatus.Available {
+			return nil, mediaSupplyClientError(errMeijiequanSessionMissing)
+		}
+	}
+	stats, err := s.syncMediaResourceModel(ctx, modelID, cfg.Meijiequan.SyncPageSize, cfg.Meijiequan.MaxSyncPages, cfg.Meijiequan.SyncPageDelay)
+	if err != nil {
+		return nil, mediaSupplyClientError(err)
+	}
+	return map[string]any{
+		"model_id":                modelID,
+		"synced":                  stats.Synced,
+		"removed_price_overrides": stats.RemovedPriceOverride,
+	}, nil
+}
+
+func (s *MediaSupplyService) syncMediaResourceModel(ctx context.Context, modelID, pageSize, maxPages int, pageDelay time.Duration) (mediaResourceSyncStats, error) {
+	var total mediaResourceSyncStats
+	for page := 1; page <= maxPages; page++ {
+		if page > 1 && pageDelay > 0 {
+			select {
+			case <-ctx.Done():
+				return total, ctx.Err()
+			case <-time.After(pageDelay):
+			}
+		}
+		mediaPage, err := s.client.ListMedia(ctx, modelID, page, pageSize)
+		if err != nil {
+			return total, err
+		}
+		stats, err := s.upsertSyncedResourcesWithStats(ctx, mediaPage.Items)
+		if err != nil {
+			return total, err
+		}
+		total.Synced += stats.Synced
+		total.RemovedPriceOverride += stats.RemovedPriceOverride
+		if len(mediaPage.Items) == 0 {
+			break
+		}
+		if mediaPage.AllPage > 0 && page >= mediaPage.AllPage {
+			break
+		}
+	}
+	return total, nil
 }

 func (s *MediaSupplyService) resourceCostForPriceType(ctx context.Context, resourceID int64, priceType string) (int64, error) {
@@ -1253,6 +1351,14 @@ func costForResourcePriceType(resource SupplierMediaResource, priceType string)
 	return resource.CostPriceCents
 }

+func mediaSupplyOverrideBelowCost(resource SupplierMediaResource, priceType string, overrideCents int64, overrideEnabled bool) bool {
+	if !overrideEnabled {
+		return false
+	}
+	cost := costForResourcePriceType(resource, normalizeMediaSupplyPriceType(priceType))
+	return cost > 0 && overrideCents < cost
+}
+
 func mediaSupplyClientError(err error) error {
 	switch {
 	case errors.Is(err, errMeijiequanCredentialsMissing):
@@ -1335,5 +1441,14 @@ func scanMediaSupplyOrder(row mediaSupplyOrderScanner) (MediaSupplyOrderDetail,
 		message := mediaSupplyPublicGenericErrorMessage(*item.ErrorMessage)
 		item.ErrorMessage = &message
 	}
+	item.ExternalOrderID = nil
+	if item.ExternalOrderCode != nil {
+		code := strings.ToUpper(strings.TrimSpace(*item.ExternalOrderCode))
+		if !isLikelyMeijiequanOrderCode(code) {
+			item.ExternalOrderCode = nil
+		} else {
+			item.ExternalOrderCode = &code
+		}
+	}
 	return item, nil
 }