fix media supply citation attribution

server/internal/tenant/app/media_supply_worker.go

@@ -17,9 +17,10 @@ import (
 )
 
 type MediaSupplyWorker struct {
-	service *MediaSupplyService
-	pool    *pgxpool.Pool
-	logger  *zap.Logger
+	service        *MediaSupplyService
+	pool           *pgxpool.Pool
+	monitoringPool *pgxpool.Pool
+	logger         *zap.Logger
 }
 
 type mediaSupplySyncJob struct {
@@ -59,6 +60,13 @@ func NewMediaSupplyWorker(service *MediaSupplyService, pool *pgxpool.Pool, logge
 	return &MediaSupplyWorker{service: service, pool: pool, logger: logger}
 }
 
+func (w *MediaSupplyWorker) WithMonitoringPool(monitoringPool *pgxpool.Pool) *MediaSupplyWorker {
+	if w != nil {
+		w.monitoringPool = monitoringPool
+	}
+	return w
+}
+
 func (w *MediaSupplyWorker) Start(ctx context.Context) {
 	if w == nil || w.service == nil || w.pool == nil {
 		return
@@ -917,7 +925,165 @@ func (w *MediaSupplyWorker) updateOrderBacklinks(ctx context.Context, order medi
 	`, order.ID, "published", remaining, firstMatchedMeijiequanOrderCode(matches), firstMatchedMeijiequanOrderID(matches)); err != nil {
 		return 0, err
 	}
-	return updatedLinks, tx.Commit(ctx)
+	if err := tx.Commit(ctx); err != nil {
+		return 0, err
+	}
+	w.refreshMonitoringAttributionForMediaSupplyBacklinks(ctx, order.ID)
+	return updatedLinks, nil
+}
+
+func (w *MediaSupplyWorker) refreshMonitoringAttributionForMediaSupplyBacklinks(ctx context.Context, orderID int64) {
+	if w == nil || w.monitoringPool == nil || w.pool == nil || orderID <= 0 {
+		return
+	}
+
+	rows, err := w.pool.Query(ctx, `
+		SELECT
+			o.tenant_id,
+			i.id,
+			i.external_article_url
+		FROM media_supply_order_items i
+		JOIN media_supply_orders o ON o.id = i.order_id
+		WHERE o.id = $1
+		  AND o.deleted_at IS NULL
+		  AND NULLIF(TRIM(COALESCE(i.external_article_url, '')), '') IS NOT NULL
+	`, orderID)
+	if err != nil {
+		w.logWarn("media supply monitoring attribution backlink lookup failed", zap.Int64("order_id", orderID), zap.Error(err))
+		return
+	}
+	defer rows.Close()
+
+	type backlinkSource struct {
+		tenantID int64
+		itemID   int64
+		url      string
+	}
+	sources := make([]backlinkSource, 0)
+	for rows.Next() {
+		var source backlinkSource
+		if scanErr := rows.Scan(&source.tenantID, &source.itemID, &source.url); scanErr != nil {
+			w.logWarn("media supply monitoring attribution backlink scan failed", zap.Int64("order_id", orderID), zap.Error(scanErr))
+			return
+		}
+		source.url = strings.TrimSpace(source.url)
+		if source.tenantID <= 0 || source.itemID <= 0 || source.url == "" {
+			continue
+		}
+		sources = append(sources, source)
+	}
+	if err := rows.Err(); err != nil {
+		w.logWarn("media supply monitoring attribution backlink iterate failed", zap.Int64("order_id", orderID), zap.Error(err))
+		return
+	}
+
+	for _, source := range sources {
+		if _, err := w.refreshMonitoringAttributionForMediaSupplyBacklink(ctx, source.tenantID, source.itemID, source.url); err != nil {
+			w.logWarn(
+				"media supply monitoring attribution refresh failed",
+				zap.Int64("tenant_id", source.tenantID),
+				zap.Int64("order_id", orderID),
+				zap.Int64("item_id", source.itemID),
+				zap.Error(err),
+			)
+		}
+	}
+}
+
+func (w *MediaSupplyWorker) refreshMonitoringAttributionForMediaSupplyBacklink(ctx context.Context, tenantID, itemID int64, rawURL string) (int64, error) {
+	matchKeys := monitoringCitationCandidateKeys(rawURL, normalizeCitationURL(rawURL))
+	if len(matchKeys) == 0 {
+		return 0, nil
+	}
+
+	normalizedURL := normalizeCitationURL(rawURL)
+	host, registrableDomain, _, _ := deriveCitationURLParts(rawURL, normalizedURL)
+	candidateURLs := uniqueTrimmedStrings(rawURL, normalizedURL)
+	rows, err := w.monitoringPool.Query(ctx, `
+		SELECT id, cited_url, normalized_url
+		FROM monitoring_citation_facts
+		WHERE tenant_id = $1
+		  AND (
+		    (content_source_type IS NULL AND content_source_id IS NULL)
+		    OR content_source_type = ANY($5::text[])
+		  )
+		  AND (
+		    cited_url = ANY($2::text[])
+		    OR normalized_url = ANY($2::text[])
+		    OR NULLIF(LOWER(TRIM(host)), '') = NULLIF($3, '')
+		    OR NULLIF(LOWER(TRIM(registrable_domain)), '') = NULLIF($4, '')
+		    OR NULLIF(LOWER(TRIM(site_key)), '') = NULLIF($4, '')
+		  )
+	`, tenantID, candidateURLs, strings.ToLower(strings.TrimSpace(host)), strings.ToLower(strings.TrimSpace(registrableDomain)), []string{
+		monitoringContentSourcePublishedArticle,
+		monitoringContentSourceManualMark,
+	})
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+
+	matchKeySet := make(map[string]struct{}, len(matchKeys))
+	for _, key := range matchKeys {
+		matchKeySet[strings.TrimSpace(key)] = struct{}{}
+	}
+	matchedIDs := make([]int64, 0)
+	for rows.Next() {
+		var id int64
+		var citedURL string
+		var factNormalizedURL string
+		if scanErr := rows.Scan(&id, &citedURL, &factNormalizedURL); scanErr != nil {
+			return 0, scanErr
+		}
+		for _, factKey := range monitoringCitationCandidateKeys(citedURL, factNormalizedURL) {
+			if _, ok := matchKeySet[strings.TrimSpace(factKey)]; ok {
+				matchedIDs = append(matchedIDs, id)
+				break
+			}
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	if len(matchedIDs) == 0 {
+		return 0, nil
+	}
+
+	tag, err := w.monitoringPool.Exec(ctx, `
+		UPDATE monitoring_citation_facts
+		SET content_source_type = $3,
+		    content_source_id = $2
+		WHERE tenant_id = $1
+		  AND id = ANY($4::bigint[])
+		  AND (
+		    (content_source_type IS NULL AND content_source_id IS NULL)
+		    OR content_source_type = ANY($5::text[])
+		  )
+	`, tenantID, itemID, monitoringContentSourceMediaSupplyOrderItem, matchedIDs, []string{
+		monitoringContentSourcePublishedArticle,
+		monitoringContentSourceManualMark,
+	})
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
+
+func uniqueTrimmedStrings(values ...string) []string {
+	result := make([]string, 0, len(values))
+	seen := make(map[string]struct{}, len(values))
+	for _, value := range values {
+		trimmed := strings.TrimSpace(value)
+		if trimmed == "" {
+			continue
+		}
+		if _, exists := seen[trimmed]; exists {
+			continue
+		}
+		seen[trimmed] = struct{}{}
+		result = append(result, trimmed)
+	}
+	return result
 }
 
 func (w *MediaSupplyWorker) claimSyncJob(ctx context.Context) (*mediaSupplySyncJob, bool, error) {