package app import ( "context" "encoding/json" "errors" "fmt" "regexp" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/config" ) type MediaSupplyWorker struct { service *MediaSupplyService pool *pgxpool.Pool logger *zap.Logger } type mediaSupplySyncJob struct { ID int64 ModelID *int AttemptCount int } type mediaSupplySubmitOrder struct { ID int64 TenantID int64 UserID int64 ModelID int Title string ContentSnapshot string Remark *string OrderBrand *string ExternalOrderID *string ExternalOrderCode *string RequestPayload json.RawMessage AttemptCount int } var ( errMediaSupplySelectedResourcesMissing = errors.New("failed to refresh supplier price") errMediaSupplyLockedSellBelowCost = errors.New("supplier cost increased above locked sell price") ) var mediaSupplyOrderCodePattern = regexp.MustCompile(`(?i)\bWM\d{3,}\b`) const ( mediaSupplyValidExternalOrderCodeSQL = `TRIM(COALESCE(external_order_code, '')) ~* '^WM[0-9]{3,}$'` mediaSupplyBacklinkSyncCooldownKey = "media_supply:backlink_sync:cooldown" ) func NewMediaSupplyWorker(service *MediaSupplyService, pool *pgxpool.Pool, logger *zap.Logger) *MediaSupplyWorker { return &MediaSupplyWorker{service: service, pool: pool, logger: logger} } func (w *MediaSupplyWorker) Start(ctx context.Context) { if w == nil || w.service == nil || w.pool == nil { return } cfg := w.service.mediaSupplyConfig() if !cfg.Enabled || !cfg.Worker.Enabled { return } go w.loop(ctx) } func (w *MediaSupplyWorker) loop(ctx context.Context) { cfg := w.service.mediaSupplyConfig() ticker := time.NewTicker(cfg.Worker.PollInterval) defer ticker.Stop() w.logInfo("media supply worker started") for { w.tick(ctx) select { case <-ctx.Done(): w.logInfo("media supply worker stopped") return case <-ticker.C: } } } func (w *MediaSupplyWorker) tick(ctx context.Context) { cfg := w.service.mediaSupplyConfig() for i := 0; i < cfg.Worker.BatchSize; i++ { worked := false if err := w.processOneSyncJob(ctx); err != nil { w.logWarn("media supply sync job failed", zap.Error(err)) } else { worked = true } if err := w.processOneOrder(ctx); err != nil { w.logWarn("media supply order submit failed", zap.Error(err)) } else { worked = true } if !worked { break } } if err := w.processBacklinkSync(ctx); err != nil { w.logWarn("media supply backlink sync failed", zap.Error(err)) } } func (w *MediaSupplyWorker) processOneSyncJob(ctx context.Context) error { job, ok, err := w.claimSyncJob(ctx) if err != nil || !ok { return err } modelID := 1 if job.ModelID != nil && *job.ModelID > 0 { modelID = *job.ModelID } stats, syncErr := w.service.RunMediaResourceSyncOnce(ctx, modelID) if syncErr != nil { return w.failSyncJob(ctx, job, syncErr) } raw, _ := json.Marshal(stats) if _, err := w.pool.Exec(ctx, ` UPDATE media_supply_sync_jobs SET status = $2, finished_at = NOW(), stats_json = $3, updated_at = NOW() WHERE id = $1 `, job.ID, mediaSupplySyncStatusSuccess, raw); err != nil { return err } return nil } func (w *MediaSupplyWorker) syncModel(ctx context.Context, modelID, pageSize, maxPages int, pageDelay time.Duration) (int, error) { stats, err := w.service.syncMediaResourceModel(ctx, modelID, pageSize, maxPages, pageDelay) return stats.Synced, err } func (w *MediaSupplyWorker) processOneOrder(ctx context.Context) error { order, ok, err := w.claimSubmitOrder(ctx) if err != nil || !ok { return err } cfg := w.service.mediaSupplyConfig() userID := strings.TrimSpace(cfg.Meijiequan.UserID) if status, statusErr := w.service.client.SessionStatus(ctx); statusErr != nil || status == nil || !status.Available { if loginStatus, loginErr := w.service.client.LoginWithConfiguredAccount(ctx); loginErr != nil { return w.failOrder(ctx, order, loginErr) } else if userID == "" && loginStatus != nil { userID = strings.TrimSpace(loginStatus.UserID) } } else if userID == "" { userID = strings.TrimSpace(status.UserID) } if userID == "" { return w.failOrder(ctx, order, errMediaSupplyAccountUserIDMissing) } items, err := w.loadSubmitOrderItems(ctx, order.ID) if err != nil { return w.failOrder(ctx, order, err) } if err := w.refreshOrderCostsBeforeSubmit(ctx, order, items); err != nil { return w.failOrder(ctx, order, err) } content, err := w.service.prepareMeijiequanArticleContent(ctx, order.TenantID, order.ContentSnapshot) if err != nil { return w.failOrder(ctx, order, err) } req := MeijiequanCreateOrderRequest{ UID: userID, ModelID: order.ModelID, Title: order.Title, Content: content, Remark: mediaSupplyStringPtrValue(order.Remark), OrderBrand: mediaSupplyStringPtrValue(order.OrderBrand), ResourceIDArr: make([]string, 0, len(items)), Extra: mediaSupplySubmitOrderExtra(order.RequestPayload), } for _, item := range items { req.ResourceIDArr = append(req.ResourceIDArr, formatMeijiequanResourceID(item)) } resp, err := w.service.client.CreateOrder(ctx, req) if err != nil { return w.failOrder(ctx, order, err) } raw := json.RawMessage(`{}`) if resp != nil && len(resp.Raw) > 0 { raw = resp.Raw } externalID, externalCode := extractMeijiequanOrderIDs(resp) w.logInfo("media supply create order response", zap.Int64("order_id", order.ID), zap.Int("upstream_status", mediaSupplyCreateOrderStatus(resp)), zap.String("upstream_info", mediaSupplyCreateOrderInfo(resp)), zap.String("raw_response", string(raw)), zap.Stringp("external_order_id", externalID), zap.Stringp("external_order_code", externalCode), ) if _, err := w.pool.Exec(ctx, ` UPDATE media_supply_orders SET status = $2, supplier_status = $3, external_order_id = COALESCE($4, external_order_id), external_order_code = COALESCE($5, external_order_code), response_payload_json = $6, submitted_at = COALESCE(submitted_at, NOW()), updated_at = NOW(), error_message = NULL WHERE id = $1 `, order.ID, mediaSupplyOrderStatusSubmitted, "submitted", externalID, externalCode, raw); err != nil { return err } _, _ = w.pool.Exec(ctx, ` UPDATE media_supply_order_items SET status = $2, updated_at = NOW() WHERE order_id = $1 `, order.ID, mediaSupplyOrderStatusSubmitted) return nil } func (w *MediaSupplyWorker) refreshOrderCostsBeforeSubmit(ctx context.Context, order *mediaSupplySubmitOrder, items []MediaSupplyOrderItem) error { if order == nil || len(items) == 0 { return nil } if err := w.refreshSelectedResourcesFromUpstream(ctx, order.ModelID, items); err != nil { return err } // Resource sync jobs keep the catalog fresh. The submit path rechecks the // persisted supplier cost under a row lock so a manual price override cannot // slip below the latest known cost. tx, err := w.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) for _, item := range items { resource, _, _, loadErr := w.service.loadResourceForOrder(ctx, tx, item.ResourceID, order.ModelID, item.PriceType) if loadErr != nil { return loadErr } currentCost := costForResourcePriceType(resource, item.PriceType) if currentCost > item.LockedSellPriceCents { return fmt.Errorf("%w for resource %s", errMediaSupplyLockedSellBelowCost, item.SupplierResourceID) } if currentCost > item.LockedCostPriceCents { if _, err := tx.Exec(ctx, ` UPDATE media_supply_order_items SET locked_cost_price_cents = $3, updated_at = NOW() WHERE order_id = $1 AND id = $2 `, order.ID, item.ID, currentCost); err != nil { return err } } } _, err = tx.Exec(ctx, ` UPDATE media_supply_orders o SET cost_total_cents = totals.cost_total, updated_at = NOW() FROM ( SELECT order_id, SUM(locked_cost_price_cents) AS cost_total FROM media_supply_order_items WHERE order_id = $1 GROUP BY order_id ) totals WHERE o.id = totals.order_id `, order.ID) if err != nil { return err } return tx.Commit(ctx) } func (w *MediaSupplyWorker) refreshSelectedResourcesFromUpstream(ctx context.Context, modelID int, items []MediaSupplyOrderItem) error { cfg := w.service.mediaSupplyConfig().Meijiequan pending := make(map[string]struct{}, len(items)) for _, item := range items { pending[strings.TrimSpace(item.SupplierResourceID)] = struct{}{} } delete(pending, "") for page := 1; page <= cfg.MaxSyncPages && len(pending) > 0; page++ { if page > 1 && cfg.SyncPageDelay > 0 { select { case <-ctx.Done(): return ctx.Err() case <-time.After(cfg.SyncPageDelay): } } mediaPage, err := w.service.client.ListMedia(ctx, modelID, page, cfg.SyncPageSize) if err != nil { return err } if _, err := w.service.UpsertSyncedResources(ctx, mediaPage.Items); err != nil { return err } for _, resource := range mediaPage.Items { delete(pending, resource.SupplierResourceID) } if len(mediaPage.Items) == 0 { break } if mediaPage.AllPage > 0 && page >= mediaPage.AllPage { break } } if len(pending) > 0 { return w.validatePendingResourcesFromCache(ctx, modelID, items, pending) } return nil } func (w *MediaSupplyWorker) validatePendingResourcesFromCache(ctx context.Context, modelID int, items []MediaSupplyOrderItem, pending map[string]struct{}) error { if len(pending) == 0 { return nil } tx, err := w.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) missing := 0 for _, item := range items { resourceID := strings.TrimSpace(item.SupplierResourceID) if _, ok := pending[resourceID]; !ok { continue } resource, _, _, loadErr := w.service.loadResourceForOrder(ctx, tx, item.ResourceID, modelID, item.PriceType) if loadErr != nil { missing++ continue } currentCost := costForResourcePriceType(resource, item.PriceType) if currentCost <= 0 { missing++ continue } if currentCost > item.LockedSellPriceCents { return fmt.Errorf("%w for resource %s", errMediaSupplyLockedSellBelowCost, item.SupplierResourceID) } } if missing > 0 { return fmt.Errorf("%w for %d selected resources", errMediaSupplySelectedResourcesMissing, missing) } return tx.Commit(ctx) } func (w *MediaSupplyWorker) processBacklinkSync(ctx context.Context) error { cfg := w.service.mediaSupplyConfig() if cfg.Worker.BacklinkSyncInterval <= 0 { return nil } cleared, err := w.clearInvalidBacklinks(ctx) if err != nil { return err } lockKey := "media_supply_backlink_sync:meijiequan" tx, err := w.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) var locked bool if err := tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1))`, lockKey).Scan(&locked); err != nil { return err } if !locked { return nil } var due bool if err := tx.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM media_supply_orders WHERE supplier = $1 AND status = $2 AND deleted_at IS NULL AND ( completed_at IS NULL OR EXISTS ( SELECT 1 FROM media_supply_order_items item WHERE item.order_id = media_supply_orders.id AND NULLIF(TRIM(COALESCE(item.external_article_url, '')), '') IS NULL ) ) ) `, mediaSupplySupplierMeijiequan, mediaSupplyOrderStatusSubmitted).Scan(&due); err != nil { return err } if err := tx.Commit(ctx); err != nil { return err } if !due && cleared == 0 { return nil } if allowed, err := w.claimBacklinkSyncCooldown(ctx, cfg.Worker.BacklinkSyncInterval); err != nil || !allowed { return err } _, err = w.syncPublishedBacklinksNowUnthrottled(ctx, cfg) return err } func (w *MediaSupplyWorker) claimBacklinkSyncCooldown(ctx context.Context, interval time.Duration) (bool, error) { if interval <= 0 || w == nil || w.service == nil || w.service.redis == nil { return true, nil } return w.service.redis.SetNX(ctx, mediaSupplyBacklinkSyncCooldownKey, time.Now().UTC().Format(time.RFC3339), interval).Result() } func (w *MediaSupplyWorker) clearInvalidBacklinks(ctx context.Context) (int64, error) { tag, err := w.pool.Exec(ctx, ` UPDATE media_supply_order_items SET external_article_url = NULL, updated_at = NOW() WHERE external_article_url IS NOT NULL AND ( LOWER(external_article_url) LIKE '%show.meitidaren.top%' OR LOWER(external_article_url) LIKE '%meijiequan.com%' OR LOWER(external_article_url) LIKE '%/advsupply/%' OR LOWER(external_article_url) LIKE '%/supply/%' OR LOWER(external_article_url) LIKE '%/api/%' ) `) return tag.RowsAffected(), err } func (w *MediaSupplyWorker) syncPublishedBacklinks(ctx context.Context) error { cfg := w.service.mediaSupplyConfig() if allowed, err := w.claimBacklinkSyncCooldown(ctx, cfg.Worker.BacklinkSyncInterval); err != nil { return err } else if !allowed { return nil } _, err := w.syncPublishedBacklinksNowUnthrottled(ctx, cfg) return err } func (w *MediaSupplyWorker) syncPublishedBacklinksNowUnthrottled(ctx context.Context, cfg config.MediaSupplyConfig) (*SyncMediaSupplyBacklinksResponse, error) { result := &SyncMediaSupplyBacklinksResponse{} if _, err := w.clearInvalidBacklinks(ctx); err != nil { return result, err } userID := strings.TrimSpace(cfg.Meijiequan.UserID) if status, statusErr := w.service.client.SessionStatus(ctx); statusErr != nil || status == nil || !status.Available { loginStatus, loginErr := w.service.client.LoginWithConfiguredAccount(ctx) if loginErr != nil { return result, loginErr } if userID == "" && loginStatus != nil { userID = strings.TrimSpace(loginStatus.UserID) } } else if userID == "" { userID = strings.TrimSpace(status.UserID) } if userID == "" { return result, errMediaSupplyAccountUserIDMissing } allArticles, err := w.loadArticleList(ctx, userID, "all", cfg.Meijiequan.PublishedPageSize, cfg.Meijiequan.PublishedMaxPages, cfg.Meijiequan.SyncPageDelay) if err != nil { return result, err } pendingArticles := filterMeijiequanPendingArticles(allArticles) problemArticles := filterMeijiequanProblemArticles(allArticles) publishedArticles := filterMeijiequanPublishedArticles(allArticles) result.PendingFetched = len(pendingArticles) result.ProblemFetched = len(problemArticles) result.PublishedFetched = len(publishedArticles) if len(allArticles) > 0 { orders, err := w.loadOrderCodeSyncCandidates(ctx, cfg.Worker.BacklinkBatchSize) if err != nil { return result, err } result.OrdersChecked += len(orders) for _, order := range orders { updated, err := w.updateOrderCode(ctx, order, allArticles) if err != nil { return result, err } if updated { result.OrdersUpdated++ result.OrderCodesUpdated++ } } } if len(problemArticles) > 0 { orders, err := w.loadRefundSyncCandidates(ctx, cfg.Worker.BacklinkBatchSize) if err != nil { return result, err } result.OrdersChecked += len(orders) for _, order := range orders { refunded, err := w.refundProblemOrder(ctx, order, problemArticles) if err != nil { return result, err } if refunded { result.OrdersUpdated++ result.RefundedOrders++ } } } if len(publishedArticles) == 0 { return result, nil } orders, err := w.loadBacklinkSyncCandidates(ctx, cfg.Worker.BacklinkBatchSize) if err != nil { return result, err } result.OrdersChecked += len(orders) if len(orders) == 0 { return result, nil } for _, order := range orders { updatedLinks, err := w.updateOrderBacklinks(ctx, order, publishedArticles) if err != nil { return result, err } if updatedLinks > 0 { result.OrdersUpdated++ result.LinksUpdated += updatedLinks } } return result, nil } func (w *MediaSupplyWorker) loadPublishedArticles(ctx context.Context, userID string, pageSize, maxPages int, pageDelay time.Duration) ([]MeijiequanPublishedArticle, error) { return w.loadArticleList(ctx, userID, "published", pageSize, maxPages, pageDelay) } func (w *MediaSupplyWorker) loadAllArticles(ctx context.Context, userID string, pageSize, maxPages int, pageDelay time.Duration) ([]MeijiequanPublishedArticle, error) { return w.loadArticleList(ctx, userID, "all", pageSize, maxPages, pageDelay) } func (w *MediaSupplyWorker) loadProblemArticles(ctx context.Context, userID string, pageSize, maxPages int, pageDelay time.Duration) ([]MeijiequanPublishedArticle, error) { return w.loadArticleList(ctx, userID, "problem", pageSize, maxPages, pageDelay) } func (w *MediaSupplyWorker) loadArticleList(ctx context.Context, userID, listType string, pageSize, maxPages int, pageDelay time.Duration) ([]MeijiequanPublishedArticle, error) { if pageSize <= 0 { pageSize = 20 } if maxPages <= 0 { maxPages = 1 } out := make([]MeijiequanPublishedArticle, 0, pageSize) seen := make(map[string]struct{}, pageSize) for page := 1; page <= maxPages; page++ { if page > 1 && pageDelay > 0 { select { case <-ctx.Done(): return out, ctx.Err() case <-time.After(pageDelay): } } var result MeijiequanPublishedArticlePage var err error switch listType { case "all": result, err = w.service.client.ListAllArticles(ctx, userID, page, pageSize) case "pending": result, err = w.service.client.ListPendingArticles(ctx, userID, page, pageSize) case "problem": result, err = w.service.client.ListProblemArticles(ctx, userID, page, pageSize) default: result, err = w.service.client.ListPublishedArticles(ctx, userID, page, pageSize) } if err != nil { return out, err } for _, item := range result.Items { item.ExternalArticleURL = strings.TrimSpace(item.ExternalArticleURL) key := articleListDedupeKey(item) if key == "" { continue } if _, exists := seen[key]; exists { continue } seen[key] = struct{}{} out = append(out, item) } if len(result.Items) == 0 { break } if result.AllPage > 0 && page >= result.AllPage { break } } return out, nil } func articleListDedupeKey(item MeijiequanPublishedArticle) string { if item.ExternalArticleURL != "" && isLikelyPublishedArticleURL(item.ExternalArticleURL) { return "url:" + item.ExternalArticleURL } if code := strings.TrimSpace(item.OrderCode); code != "" { return "code:" + strings.ToUpper(code) } if id := strings.TrimSpace(item.OrderID); id != "" { return "id:" + id } title := normalizeMeijiequanMatchText(item.Title) resource := normalizeMeijiequanMatchText(item.ResourceName) if title != "" || resource != "" { return "text:" + title + ":" + resource } return "" } func (w *MediaSupplyWorker) loadOrderCodeSyncCandidates(ctx context.Context, limit int) ([]mediaSupplySubmitOrder, error) { if limit <= 0 { limit = 20 } rows, err := w.pool.Query(ctx, ` SELECT id, tenant_id, user_id, model_id, title, content_snapshot, remark, order_brand, external_order_id, external_order_code, attempt_count, request_payload_json FROM media_supply_orders WHERE supplier = $1 AND status = $2 AND deleted_at IS NULL AND NOT (`+mediaSupplyValidExternalOrderCodeSQL+`) ORDER BY submitted_at DESC NULLS LAST, id DESC LIMIT $3 `, mediaSupplySupplierMeijiequan, mediaSupplyOrderStatusSubmitted, limit) if err != nil { return nil, err } defer rows.Close() orders := make([]mediaSupplySubmitOrder, 0, limit) for rows.Next() { var order mediaSupplySubmitOrder var requestPayload []byte if err := rows.Scan( &order.ID, &order.TenantID, &order.UserID, &order.ModelID, &order.Title, &order.ContentSnapshot, &order.Remark, &order.OrderBrand, &order.ExternalOrderID, &order.ExternalOrderCode, &order.AttemptCount, &requestPayload, ); err != nil { return nil, err } order.RequestPayload = append(order.RequestPayload[:0], requestPayload...) orders = append(orders, order) } return orders, rows.Err() } func (w *MediaSupplyWorker) updateOrderCode(ctx context.Context, order mediaSupplySubmitOrder, pending []MeijiequanPublishedArticle) (bool, error) { if order.ExternalOrderCode != nil && isLikelyMeijiequanOrderCode(*order.ExternalOrderCode) { return false, nil } items, err := w.service.listOrderItems(ctx, order.ID) if err != nil { return false, err } article := matchMeijiequanOrderCodeArticle(order, items, pending) if article == nil || strings.TrimSpace(article.OrderCode) == "" { return false, nil } raw := article.Raw if len(raw) == 0 { raw, _ = json.Marshal(article) } if len(raw) == 0 { raw = json.RawMessage(`{}`) } tag, err := w.pool.Exec(ctx, ` UPDATE media_supply_orders SET external_order_code = COALESCE(NULLIF($2::text, ''), external_order_code), external_order_id = COALESCE(NULLIF($3::text, ''), external_order_id), response_payload_json = CASE WHEN response_payload_json IS NULL OR response_payload_json = '{}'::jsonb THEN $4 ELSE response_payload_json END, updated_at = NOW() WHERE id = $1 AND NOT (`+mediaSupplyValidExternalOrderCodeSQL+`) `, order.ID, article.OrderCode, article.OrderID, compactJSON(raw)) if err != nil { return false, err } return tag.RowsAffected() > 0, nil } func (w *MediaSupplyWorker) loadBacklinkSyncCandidates(ctx context.Context, limit int) ([]mediaSupplySubmitOrder, error) { if limit <= 0 { limit = 20 } rows, err := w.pool.Query(ctx, ` SELECT id, tenant_id, user_id, model_id, title, content_snapshot, remark, order_brand, external_order_id, external_order_code, attempt_count, request_payload_json FROM media_supply_orders WHERE supplier = $1 AND status = $2 AND deleted_at IS NULL AND ( completed_at IS NULL OR EXISTS ( SELECT 1 FROM media_supply_order_items item WHERE item.order_id = media_supply_orders.id AND NULLIF(TRIM(COALESCE(item.external_article_url, '')), '') IS NULL ) ) ORDER BY submitted_at ASC NULLS LAST, id ASC LIMIT $3 `, mediaSupplySupplierMeijiequan, mediaSupplyOrderStatusSubmitted, limit) if err != nil { return nil, err } defer rows.Close() orders := make([]mediaSupplySubmitOrder, 0, limit) for rows.Next() { var order mediaSupplySubmitOrder var requestPayload []byte if err := rows.Scan( &order.ID, &order.TenantID, &order.UserID, &order.ModelID, &order.Title, &order.ContentSnapshot, &order.Remark, &order.OrderBrand, &order.ExternalOrderID, &order.ExternalOrderCode, &order.AttemptCount, &requestPayload, ); err != nil { return nil, err } order.RequestPayload = append(order.RequestPayload[:0], requestPayload...) orders = append(orders, order) } return orders, rows.Err() } func (w *MediaSupplyWorker) loadRefundSyncCandidates(ctx context.Context, limit int) ([]mediaSupplySubmitOrder, error) { if limit <= 0 { limit = 20 } rows, err := w.pool.Query(ctx, ` SELECT id, tenant_id, user_id, model_id, title, content_snapshot, remark, order_brand, external_order_id, external_order_code, attempt_count, request_payload_json FROM media_supply_orders WHERE supplier = $1 AND status = $2 AND wallet_debited_at IS NOT NULL AND wallet_refunded_at IS NULL AND deleted_at IS NULL ORDER BY submitted_at ASC NULLS LAST, id ASC LIMIT $3 `, mediaSupplySupplierMeijiequan, mediaSupplyOrderStatusSubmitted, limit) if err != nil { return nil, err } defer rows.Close() orders := make([]mediaSupplySubmitOrder, 0, limit) for rows.Next() { var order mediaSupplySubmitOrder var requestPayload []byte if err := rows.Scan( &order.ID, &order.TenantID, &order.UserID, &order.ModelID, &order.Title, &order.ContentSnapshot, &order.Remark, &order.OrderBrand, &order.ExternalOrderID, &order.ExternalOrderCode, &order.AttemptCount, &requestPayload, ); err != nil { return nil, err } order.RequestPayload = append(order.RequestPayload[:0], requestPayload...) orders = append(orders, order) } return orders, rows.Err() } func (w *MediaSupplyWorker) refundProblemOrder(ctx context.Context, order mediaSupplySubmitOrder, problemArticles []MeijiequanPublishedArticle) (bool, error) { items, err := w.service.listOrderItems(ctx, order.ID) if err != nil { return false, err } article := matchMeijiequanProblemArticle(order, items, problemArticles) if article == nil { return false, nil } raw := article.Raw if len(raw) == 0 { raw, _ = json.Marshal(article) } if len(raw) == 0 { raw = json.RawMessage(`{}`) } reason := mediaSupplyProblemArticleReason(*article) tx, err := w.pool.Begin(ctx) if err != nil { return false, err } defer tx.Rollback(ctx) tag, err := tx.Exec(ctx, ` UPDATE media_supply_orders SET status = $2, supplier_status = $3, external_order_code = COALESCE(NULLIF($4::text, ''), external_order_code), external_order_id = COALESCE(NULLIF($5::text, ''), external_order_id), response_payload_json = CASE WHEN response_payload_json IS NULL OR response_payload_json = '{}'::jsonb THEN $6 ELSE response_payload_json END, error_message = $7, next_attempt_at = NOW(), updated_at = NOW() WHERE id = $1 AND status = $8 AND wallet_refunded_at IS NULL `, order.ID, mediaSupplyOrderStatusFailed, "rejected", article.OrderCode, article.OrderID, compactJSON(raw), reason, mediaSupplyOrderStatusSubmitted) if err != nil { return false, err } if tag.RowsAffected() == 0 { return false, nil } if _, err := tx.Exec(ctx, ` UPDATE media_supply_order_items SET status = $2, raw_json = $3, updated_at = NOW() WHERE order_id = $1 `, order.ID, mediaSupplyOrderStatusFailed, compactJSON(raw)); err != nil { return false, err } if err := w.service.refundMediaSupplyWalletForOrder(ctx, tx, order.ID, reason); err != nil { return false, err } return true, tx.Commit(ctx) } func (w *MediaSupplyWorker) updateOrderBacklinks(ctx context.Context, order mediaSupplySubmitOrder, published []MeijiequanPublishedArticle) (int, error) { items, err := w.service.listOrderItems(ctx, order.ID) if err != nil { return 0, err } if len(items) == 0 { return 0, nil } matches := matchMeijiequanPublishedArticles(order, items, published) if len(matches) == 0 { _, err := w.pool.Exec(ctx, ` UPDATE media_supply_orders SET updated_at = NOW() WHERE id = $1 `, order.ID) return 0, err } tx, err := w.pool.Begin(ctx) if err != nil { return 0, err } defer tx.Rollback(ctx) updatedLinks := 0 for itemID, article := range matches { article.ExternalArticleURL = strings.TrimSpace(article.ExternalArticleURL) if article.ExternalArticleURL == "" || !isLikelyPublishedArticleURL(article.ExternalArticleURL) { continue } raw := article.Raw if len(raw) == 0 { raw, _ = json.Marshal(article) } if len(raw) == 0 { raw = json.RawMessage(`{}`) } if _, err := tx.Exec(ctx, ` UPDATE media_supply_order_items SET external_article_url = $3, status = $4, raw_json = $5, updated_at = NOW() WHERE order_id = $1 AND id = $2 `, order.ID, itemID, article.ExternalArticleURL, mediaSupplyOrderStatusSubmitted, compactJSON(raw)); err != nil { return 0, err } updatedLinks++ } if updatedLinks == 0 { return 0, nil } var remaining int if err := tx.QueryRow(ctx, ` SELECT COUNT(*) FROM media_supply_order_items WHERE order_id = $1 AND NULLIF(TRIM(COALESCE(external_article_url, '')), '') IS NULL `, order.ID).Scan(&remaining); err != nil { return 0, err } if _, err := tx.Exec(ctx, ` UPDATE media_supply_orders SET supplier_status = $2, completed_at = CASE WHEN $3::int = 0 THEN COALESCE(completed_at, NOW()) ELSE completed_at END, external_order_code = COALESCE(NULLIF($4::text, ''), external_order_code), external_order_id = COALESCE(NULLIF($5::text, ''), external_order_id), updated_at = NOW() WHERE id = $1 `, order.ID, "published", remaining, firstMatchedMeijiequanOrderCode(matches), firstMatchedMeijiequanOrderID(matches)); err != nil { return 0, err } return updatedLinks, tx.Commit(ctx) } func (w *MediaSupplyWorker) claimSyncJob(ctx context.Context) (*mediaSupplySyncJob, bool, error) { tx, err := w.pool.Begin(ctx) if err != nil { return nil, false, err } defer tx.Rollback(ctx) var locked bool if err := tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1))`, "media_supply_sync:meijiequan").Scan(&locked); err != nil { return nil, false, err } if !locked { return nil, false, nil } var job mediaSupplySyncJob var modelID *int if err := tx.QueryRow(ctx, ` SELECT id, model_id, attempt_count FROM media_supply_sync_jobs WHERE supplier = $1 AND status = $2 AND NOT EXISTS ( SELECT 1 FROM media_supply_sync_jobs running WHERE running.supplier = $1 AND running.status = $3 ) ORDER BY created_at ASC, id ASC FOR UPDATE SKIP LOCKED LIMIT 1 `, mediaSupplySupplierMeijiequan, mediaSupplySyncStatusQueued, mediaSupplySyncStatusRunning).Scan(&job.ID, &modelID, &job.AttemptCount); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, false, nil } return nil, false, err } job.ModelID = modelID if _, err := tx.Exec(ctx, ` UPDATE media_supply_sync_jobs SET status = $2, started_at = NOW(), attempt_count = attempt_count + 1, updated_at = NOW() WHERE id = $1 `, job.ID, mediaSupplySyncStatusRunning); err != nil { return nil, false, err } if err := tx.Commit(ctx); err != nil { return nil, false, err } return &job, true, nil } func (w *MediaSupplyWorker) claimSubmitOrder(ctx context.Context) (*mediaSupplySubmitOrder, bool, error) { tx, err := w.pool.Begin(ctx) if err != nil { return nil, false, err } defer tx.Rollback(ctx) var order mediaSupplySubmitOrder var requestPayload []byte if err := tx.QueryRow(ctx, ` SELECT id, tenant_id, user_id, model_id, title, content_snapshot, remark, order_brand, attempt_count, request_payload_json FROM media_supply_orders WHERE supplier = $1 AND status = $2 AND next_attempt_at <= NOW() AND deleted_at IS NULL ORDER BY queued_at ASC, id ASC FOR UPDATE SKIP LOCKED LIMIT 1 `, mediaSupplySupplierMeijiequan, mediaSupplyOrderStatusQueued).Scan( &order.ID, &order.TenantID, &order.UserID, &order.ModelID, &order.Title, &order.ContentSnapshot, &order.Remark, &order.OrderBrand, &order.AttemptCount, &requestPayload, ); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, false, nil } return nil, false, err } order.RequestPayload = append(order.RequestPayload[:0], requestPayload...) if _, err := tx.Exec(ctx, ` UPDATE media_supply_orders SET status = $2, attempt_count = attempt_count + 1, updated_at = NOW() WHERE id = $1 `, order.ID, mediaSupplyOrderStatusSubmitting); err != nil { return nil, false, err } if err := tx.Commit(ctx); err != nil { return nil, false, err } return &order, true, nil } func (w *MediaSupplyWorker) loadSubmitOrderItems(ctx context.Context, orderID int64) ([]MediaSupplyOrderItem, error) { return w.service.listOrderItems(ctx, orderID) } func (w *MediaSupplyWorker) failSyncJob(ctx context.Context, job *mediaSupplySyncJob, cause error) error { if job == nil { return cause } cfg := w.service.mediaSupplyConfig() status := mediaSupplySyncStatusFailed if job.AttemptCount+1 < cfg.Worker.SyncMaxAttempts { status = mediaSupplySyncStatusQueued } _, err := w.pool.Exec(ctx, ` UPDATE media_supply_sync_jobs SET status = $2::varchar, error_message = $3, finished_at = CASE WHEN $2::varchar = $4::varchar THEN NOW() ELSE finished_at END, updated_at = NOW() WHERE id = $1 `, job.ID, status, truncateText(mediaSupplyPublicErrorMessage(cause)), mediaSupplySyncStatusFailed) if err != nil { return err } return cause } func (w *MediaSupplyWorker) failOrder(ctx context.Context, order *mediaSupplySubmitOrder, cause error) error { if order == nil { return cause } cfg := w.service.mediaSupplyConfig() status := mediaSupplyOrderStatusFailed nextAttemptAt := time.Now().UTC() if order.AttemptCount+1 < cfg.Worker.OrderMaxAttempts && retryableMediaSupplyError(cause) { status = mediaSupplyOrderStatusQueued nextAttemptAt = nextAttemptAt.Add(time.Duration(order.AttemptCount+1) * time.Minute) } publicMessage := truncateText(mediaSupplyPublicErrorMessage(cause)) tx, err := w.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) _, err = tx.Exec(ctx, ` UPDATE media_supply_orders SET status = $2, error_message = $3, next_attempt_at = $4, updated_at = NOW() WHERE id = $1 `, order.ID, status, publicMessage, nextAttemptAt) if err != nil { return err } if _, err := tx.Exec(ctx, ` UPDATE media_supply_order_items SET status = $2, updated_at = NOW() WHERE order_id = $1 AND status IN ($3::varchar, $4::varchar) `, order.ID, status, mediaSupplyOrderStatusQueued, mediaSupplyOrderStatusSubmitting); err != nil { return err } if status == mediaSupplyOrderStatusFailed { if err := w.service.refundMediaSupplyWalletForOrder(ctx, tx, order.ID, publicMessage); err != nil { return err } } if err := tx.Commit(ctx); err != nil { return err } return cause } func retryableMediaSupplyError(err error) bool { if err == nil { return false } if errors.Is(err, errMeijiequanSessionMissing) || errors.Is(err, errMeijiequanSessionExpired) || errors.Is(err, errMeijiequanAuthRequired) { return false } if errors.Is(err, errMeijiequanOrderRejected) || errors.Is(err, errMediaSupplySelectedResourcesMissing) || errors.Is(err, errMediaSupplyLockedSellBelowCost) { return false } return true } func formatMeijiequanResourceID(item MediaSupplyOrderItem) string { priceType := normalizeMediaSupplyPriceType(item.PriceType) switch priceType { case "price": return fmt.Sprintf("%s_price_0", item.SupplierResourceID) default: return fmt.Sprintf("%s_%s_0", item.SupplierResourceID, priceType) } } func mediaSupplySubmitOrderExtra(raw json.RawMessage) map[string]any { if len(raw) == 0 { return map[string]any{} } var payload map[string]any decoder := json.NewDecoder(strings.NewReader(string(raw))) decoder.UseNumber() if err := decoder.Decode(&payload); err != nil { return map[string]any{} } extra := make(map[string]any) if nested, ok := payload["extra"].(map[string]any); ok { for key, value := range nested { extra[key] = value } } for _, key := range []string{ "source", "laiyuan", "description", "miaoshu", "promotion", "tuiguang", "propaganda", "xuanchuan", "news_link", "xinwenlianjie", "business_license", "yingyezhizhao", "enterprise_logo", "qiyelogo", "coupon", "youhuiquan", "reference_link", "cankaolianjie", "daily_price", "richangjia", "live_price", "zhibojia", "stock", "kucunliang", "cover", "fengmian", "original_link", "yuanwenlianjie", "brand_id", "extra_field_1", "extra_field_2", "order_brand", "batch_resource_ids", "point_time", "point_price_range", "limit_time_end", "resource_spare_arr", "resource_spare_ids", "prop", } { if value, ok := payload[key]; ok { extra[key] = value } } if extra == nil { return map[string]any{} } return extra } func matchMeijiequanPublishedArticles(order mediaSupplySubmitOrder, items []MediaSupplyOrderItem, published []MeijiequanPublishedArticle) map[int64]MeijiequanPublishedArticle { matches := make(map[int64]MeijiequanPublishedArticle) if len(items) == 0 || len(published) == 0 { return matches } orderTitle := normalizeMeijiequanMatchText(order.Title) orderCode := mediaSupplyOrderCodeValue(order) unmatched := make(map[int64]MediaSupplyOrderItem, len(items)) for _, item := range items { if item.ExternalArticleURL != nil && isLikelyPublishedArticleURL(strings.TrimSpace(*item.ExternalArticleURL)) { continue } unmatched[item.ID] = item } for _, article := range published { if len(unmatched) == 0 { break } article.ExternalArticleURL = strings.TrimSpace(article.ExternalArticleURL) if article.ExternalArticleURL == "" || !isLikelyPublishedArticleURL(article.ExternalArticleURL) { continue } if orderCode != "" && strings.EqualFold(strings.TrimSpace(article.OrderCode), orderCode) && len(unmatched) == 1 { for itemID := range unmatched { matches[itemID] = article delete(unmatched, itemID) break } continue } itemID := matchMeijiequanPublishedArticleItem(orderTitle, orderCode, unmatched, article) if itemID <= 0 { continue } matches[itemID] = article delete(unmatched, itemID) } return matches } func matchMeijiequanOrderCodeArticle(order mediaSupplySubmitOrder, items []MediaSupplyOrderItem, articles []MeijiequanPublishedArticle) *MeijiequanPublishedArticle { if len(articles) == 0 { return nil } if orderCode := mediaSupplyOrderCodeValue(order); orderCode != "" { for idx := range articles { if strings.EqualFold(strings.TrimSpace(articles[idx].OrderCode), orderCode) { return &articles[idx] } } return nil } orderTitle := normalizeMeijiequanMatchText(order.Title) for idx := range articles { article := &articles[idx] if strings.TrimSpace(article.OrderCode) == "" { continue } if articleMatchesOrderFallback(orderTitle, *article, items) { return article } } return nil } func matchMeijiequanProblemArticle(order mediaSupplySubmitOrder, items []MediaSupplyOrderItem, articles []MeijiequanPublishedArticle) *MeijiequanPublishedArticle { if len(articles) == 0 { return nil } if orderCode := mediaSupplyOrderCodeValue(order); orderCode != "" { for idx := range articles { article := &articles[idx] if !isMeijiequanProblemArticle(*article) { continue } if strings.EqualFold(strings.TrimSpace(article.OrderCode), orderCode) { return article } } return nil } orderTitle := normalizeMeijiequanMatchText(order.Title) for idx := range articles { article := &articles[idx] if !isMeijiequanProblemArticle(*article) { continue } if articleMatchesOrderFallback(orderTitle, *article, items) { return article } } return nil } func mediaSupplyOrderCodeValue(order mediaSupplySubmitOrder) string { if order.ExternalOrderCode != nil { if value := strings.TrimSpace(*order.ExternalOrderCode); isLikelyMeijiequanOrderCode(value) { return value } } if order.ExternalOrderID != nil { if value := strings.TrimSpace(*order.ExternalOrderID); isLikelyMeijiequanOrderCode(value) { return value } } return "" } func filterMeijiequanPendingArticles(articles []MeijiequanPublishedArticle) []MeijiequanPublishedArticle { out := make([]MeijiequanPublishedArticle, 0, len(articles)) for _, article := range articles { if !isMeijiequanPublishedArticle(article) && !isMeijiequanProblemArticle(article) { out = append(out, article) } } return out } func filterMeijiequanPublishedArticles(articles []MeijiequanPublishedArticle) []MeijiequanPublishedArticle { out := make([]MeijiequanPublishedArticle, 0, len(articles)) for _, article := range articles { if isMeijiequanPublishedArticle(article) { out = append(out, article) } } return out } func filterMeijiequanProblemArticles(articles []MeijiequanPublishedArticle) []MeijiequanPublishedArticle { out := make([]MeijiequanPublishedArticle, 0, len(articles)) for _, article := range articles { if isMeijiequanProblemArticle(article) { out = append(out, article) } } return out } func isMeijiequanPublishedArticle(article MeijiequanPublishedArticle) bool { status := strings.TrimSpace(article.Status) return strings.Contains(status, "已发布") || strings.Contains(status, "已发表") || strings.Contains(status, "发布成功") || (article.ExternalArticleURL != "" && isLikelyPublishedArticleURL(article.ExternalArticleURL)) } func isMeijiequanProblemArticle(article MeijiequanPublishedArticle) bool { status := strings.TrimSpace(article.Status) return strings.Contains(status, "退稿") || strings.Contains(status, "失败") || strings.Contains(status, "驳回") || strings.Contains(status, "拒") || strings.TrimSpace(article.RejectReason) != "" } func mediaSupplyProblemArticleReason(article MeijiequanPublishedArticle) string { reason := strings.TrimSpace(article.RejectReason) if reason == "" { reason = strings.TrimSpace(article.Status) } if reason == "" { reason = "投稿未通过" } return "退稿原因:" + reason } func articleMatchesAnyOrderItem(article MeijiequanPublishedArticle, items []MediaSupplyOrderItem) bool { if len(items) == 0 { return true } resourceID := normalizeMeijiequanPublishedResourceID(article.ResourceID) resourceName := normalizeMeijiequanMatchText(article.ResourceName) for _, item := range items { if resourceID != "" && strings.EqualFold(strings.TrimSpace(item.SupplierResourceID), resourceID) { return true } itemName := normalizeMeijiequanMatchText(item.ResourceNameSnapshot) if resourceName != "" && itemName != "" && (strings.Contains(itemName, resourceName) || strings.Contains(resourceName, itemName)) { return true } } return false } func articleMatchesOrderFallback(orderTitle string, article MeijiequanPublishedArticle, items []MediaSupplyOrderItem) bool { if !articleTitleMatchesOrder(orderTitle, article) { return false } if len(items) == 0 { return false } for _, item := range items { if articleMatchesOrderItemFallback(article, item) { return true } } return false } func articleTitleMatchesOrder(orderTitle string, article MeijiequanPublishedArticle) bool { articleTitle := normalizeMeijiequanMatchText(article.Title) return orderTitle != "" && articleTitle != "" && (strings.Contains(articleTitle, orderTitle) || strings.Contains(orderTitle, articleTitle)) } func articleMatchesOrderItemFallback(article MeijiequanPublishedArticle, item MediaSupplyOrderItem) bool { return articleMatchesOrderItemResource(article, item) && articleMatchesOrderItemPrice(article, item) } func articleMatchesOrderItemResource(article MeijiequanPublishedArticle, item MediaSupplyOrderItem) bool { resourceID := normalizeMeijiequanPublishedResourceID(article.ResourceID) if resourceID != "" && strings.EqualFold(strings.TrimSpace(item.SupplierResourceID), resourceID) { return true } resourceName := normalizeMeijiequanMatchText(article.ResourceName) itemName := normalizeMeijiequanMatchText(item.ResourceNameSnapshot) return resourceName != "" && itemName != "" && (strings.Contains(itemName, resourceName) || strings.Contains(resourceName, itemName)) } func articleMatchesOrderItemPrice(article MeijiequanPublishedArticle, item MediaSupplyOrderItem) bool { if article.PriceCents <= 0 { return false } return article.PriceCents == item.LockedCostPriceCents } func matchMeijiequanPublishedArticleItem(orderTitle, orderCode string, items map[int64]MediaSupplyOrderItem, article MeijiequanPublishedArticle) int64 { if orderCode != "" { if !strings.EqualFold(strings.TrimSpace(article.OrderCode), orderCode) { return 0 } for itemID := range items { return itemID } } if articleTitleMatchesOrder(orderTitle, article) { if len(items) == 1 { for itemID, item := range items { if articleMatchesOrderItemFallback(article, item) { return itemID } } } for itemID, item := range items { if articleMatchesOrderItemFallback(article, item) { return itemID } } } return 0 } func normalizeMeijiequanMatchText(value string) string { value = strings.ToLower(strings.TrimSpace(value)) value = strings.ReplaceAll(value, " ", "") value = strings.ReplaceAll(value, "\t", "") value = strings.ReplaceAll(value, "\n", "") value = strings.ReplaceAll(value, "\r", "") return value } func firstMatchedMeijiequanOrderCode(matches map[int64]MeijiequanPublishedArticle) string { for _, article := range matches { if text := strings.TrimSpace(article.OrderCode); text != "" { return text } } return "" } func firstMatchedMeijiequanOrderID(matches map[int64]MeijiequanPublishedArticle) string { for _, article := range matches { if text := strings.TrimSpace(article.OrderID); text != "" { return text } } return "" } func pgInterval(value time.Duration) string { if value <= 0 { return "0 seconds" } return fmt.Sprintf("%f seconds", value.Seconds()) } func mediaSupplyCreateOrderStatus(resp *MeijiequanCreateOrderResponse) int { if resp == nil { return 0 } return resp.Status } func mediaSupplyCreateOrderInfo(resp *MeijiequanCreateOrderResponse) string { if resp == nil { return "" } return resp.Info } func extractMeijiequanOrderIDs(resp *MeijiequanCreateOrderResponse) (*string, *string) { if resp == nil { return nil, nil } id, code := extractMeijiequanOrderIDValue(resp.Data) if id == "" && code == "" { id, code = extractMeijiequanOrderIDValue(resp.Raw) } if code == "" { code = firstMeijiequanOrderCode(resp.Info) } return optionalTextPtr(id), optionalTextPtr(code) } func extractMeijiequanOrderIDValue(raw json.RawMessage) (string, string) { if len(raw) == 0 { return "", "" } var value any decoder := json.NewDecoder(strings.NewReader(string(raw))) decoder.UseNumber() if err := decoder.Decode(&value); err != nil { return "", firstMeijiequanOrderCode(string(raw)) } id, code := extractMeijiequanOrderIDFromValue(value) if code == "" { code = firstMeijiequanOrderCode(string(raw)) } return id, code } func extractMeijiequanOrderIDFromValue(value any) (string, string) { switch typed := value.(type) { case map[string]any: id := strings.TrimSpace(meijiequanAnyString(firstMapValue(typed, "id", "order_id", "article_id", "article_order_id", ))) code := strings.TrimSpace(meijiequanAnyString(firstMapValue(typed, "order_code", "code", "sn", "order_sn", "order_no", "orderno", "danhao", "dingdanhao", "no", ))) if code != "" && !isLikelyMeijiequanOrderCode(code) { if extracted := firstMeijiequanOrderCode(code); extracted != "" { code = extracted } else { code = "" } } for _, key := range []string{"data", "result", "order", "article", "info", "message", "msg"} { nestedID, nestedCode := extractMeijiequanOrderIDFromValue(typed[key]) if id == "" { id = nestedID } if code == "" { code = nestedCode } if id != "" && code != "" { return id, code } } if code == "" { for _, value := range typed { if text := firstMeijiequanOrderCode(meijiequanAnyString(value)); text != "" { code = text break } } } return id, code case []any: for _, item := range typed { if id, code := extractMeijiequanOrderIDFromValue(item); id != "" || code != "" { return id, code } } case string: return "", firstMeijiequanOrderCode(typed) } return "", "" } func firstMeijiequanOrderCode(value string) string { return strings.ToUpper(strings.TrimSpace(mediaSupplyOrderCodePattern.FindString(value))) } func optionalTextPtr(value string) *string { value = strings.TrimSpace(value) if value == "" { return nil } return &value } func mediaSupplyStringPtrValue(value *string) string { if value == nil { return "" } return strings.TrimSpace(*value) } func truncateError(err error) string { if err == nil { return "" } return truncateText(err.Error()) } func truncateText(value string) string { if len(value) > 2000 { return value[:2000] } return value } func (w *MediaSupplyWorker) logInfo(message string, fields ...zap.Field) { if w != nil && w.logger != nil { w.logger.Info(message, fields...) } } func (w *MediaSupplyWorker) logWarn(message string, fields ...zap.Field) { if w != nil && w.logger != nil { w.logger.Warn(message, fields...) } }