fix: implement resource reconciliation and update handling in media supply favorites

server/internal/tenant/app/media_supply_service.go

@@ -983,17 +983,21 @@ func (s *MediaSupplyService) GetOrder(ctx context.Context, orderID int64) (*Medi
 type mediaResourceSyncStats struct {
 	Synced               int
 	RemovedPriceOverride int
+	RemovedStale         int
 }
 
 func (s *MediaSupplyService) UpsertSyncedResources(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (int, error) {
-	stats, err := s.upsertSyncedResourcesWithStats(ctx, resources)
+	stats, err := s.upsertSyncedResourcesWithStats(ctx, resources, time.Now().UTC())
 	return stats.Synced, err
 }
 
-func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (mediaResourceSyncStats, error) {
+func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context, resources []UpsertSupplierMediaResourceInput, syncedAt time.Time) (mediaResourceSyncStats, error) {
 	if len(resources) == 0 {
 		return mediaResourceSyncStats{}, nil
 	}
+	if syncedAt.IsZero() {
+		syncedAt = time.Now().UTC()
+	}
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {
 		return mediaResourceSyncStats{}, err
@@ -1014,7 +1018,7 @@ func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context,
 			    channel_type, region, inclusion_effect, link_type, publish_rate, delivery_speed,
 			    supplier_updated_at, last_synced_at, raw_json
 			)
-			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, NOW(), $19)
+			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
 			ON CONFLICT (supplier, supplier_resource_id) DO UPDATE SET
 			    model_id = EXCLUDED.model_id,
 			    name = EXCLUDED.name,
@@ -1032,7 +1036,7 @@ func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context,
 			    publish_rate = EXCLUDED.publish_rate,
 			    delivery_speed = EXCLUDED.delivery_speed,
 			    supplier_updated_at = EXCLUDED.supplier_updated_at,
-			    last_synced_at = NOW(),
+			    last_synced_at = EXCLUDED.last_synced_at,
 			    raw_json = EXCLUDED.raw_json,
 			    updated_at = NOW(),
 			    deleted_at = NULL
@@ -1043,7 +1047,7 @@ func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context,
 			nullableTrimmedString(resource.ChannelType), nullableTrimmedString(resource.Region),
 			nullableTrimmedString(resource.InclusionEffect), nullableTrimmedString(resource.LinkType),
 			nullableTrimmedString(resource.PublishRate), nullableTrimmedString(resource.DeliverySpeed),
-			resource.SupplierUpdatedAt, raw).Scan(&resourceID); err != nil {
+			resource.SupplierUpdatedAt, syncedAt, raw).Scan(&resourceID); err != nil {
 			return stats, err
 		}
 		removed, err := s.deletePriceOverridesBelowCost(ctx, tx, resourceID)
@@ -1110,11 +1114,15 @@ func (s *MediaSupplyService) RunMediaResourceSyncOnce(ctx context.Context, model
 		"model_id":                modelID,
 		"synced":                  stats.Synced,
 		"removed_price_overrides": stats.RemovedPriceOverride,
+		"removed_stale_resources": stats.RemovedStale,
 	}, nil
 }
 
 func (s *MediaSupplyService) syncMediaResourceModel(ctx context.Context, modelID, pageSize, maxPages int, pageDelay time.Duration) (mediaResourceSyncStats, error) {
 	var total mediaResourceSyncStats
+	syncedAt := time.Now().UTC()
+	completed := false
+	syncedResourceIDs := make(map[string]struct{})
 	for page := 1; page <= maxPages; page++ {
 		if page > 1 && pageDelay > 0 {
 			select {
@@ -1127,22 +1135,66 @@ func (s *MediaSupplyService) syncMediaResourceModel(ctx context.Context, modelID
 		if err != nil {
 			return total, err
 		}
-		stats, err := s.upsertSyncedResourcesWithStats(ctx, mediaPage.Items)
+		stats, err := s.upsertSyncedResourcesWithStats(ctx, mediaPage.Items, syncedAt)
 		if err != nil {
 			return total, err
 		}
 		total.Synced += stats.Synced
 		total.RemovedPriceOverride += stats.RemovedPriceOverride
-		if len(mediaPage.Items) == 0 {
+		for _, resource := range mediaPage.Items {
+			resourceID := strings.TrimSpace(resource.SupplierResourceID)
+			if resourceID != "" {
+				syncedResourceIDs[resourceID] = struct{}{}
+			}
+		}
+		if mediaPage.RawItemCount == 0 {
+			completed = true
+			break
+		}
+		if pageSize > 0 && mediaPage.RawItemCount < pageSize {
+			completed = true
 			break
 		}
 		if mediaPage.AllPage > 0 && page >= mediaPage.AllPage {
+			completed = true
 			break
 		}
 	}
+	if !completed {
+		return total, fmt.Errorf("meijiequan media resource sync incomplete: reached max pages %d", maxPages)
+	}
+	removed, err := s.markMissingSyncedResourcesDeleted(ctx, mediaSupplySupplierMeijiequan, modelID, syncedResourceIDs)
+	if err != nil {
+		return total, err
+	}
+	total.RemovedStale = removed
 	return total, nil
 }
 
+func (s *MediaSupplyService) markMissingSyncedResourcesDeleted(ctx context.Context, supplier string, modelID int, syncedResourceIDs map[string]struct{}) (int, error) {
+	if s == nil || s.pool == nil || strings.TrimSpace(supplier) == "" || modelID <= 0 {
+		return 0, nil
+	}
+	ids := make([]string, 0, len(syncedResourceIDs))
+	for id := range syncedResourceIDs {
+		if trimmed := strings.TrimSpace(id); trimmed != "" {
+			ids = append(ids, trimmed)
+		}
+	}
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE supplier_media_resources
+		SET deleted_at = NOW(), updated_at = NOW()
+		WHERE supplier = $1
+		  AND model_id = $2
+		  AND deleted_at IS NULL
+		  AND NOT (supplier_resource_id = ANY($3::text[]))
+	`, supplier, modelID, ids)
+	if err != nil {
+		return 0, err
+	}
+	return int(tag.RowsAffected()), nil
+}
+
 func (s *MediaSupplyService) resourceCostForPriceType(ctx context.Context, resourceID int64, priceType string) (int64, error) {
 	var costPrice int64
 	var costPricesRaw []byte

server/internal/tenant/app/media_supply_test.go

@@ -616,6 +616,38 @@ func TestDecodeMeijiequanMediaListAcceptsStringNumbers(t *testing.T) {
 	}
 }
 
+func TestMeijiequanListMediaKeepsRawItemCountForPaging(t *testing.T) {
+	client, cleanup := testMeijiequanClientWithUploadServer(t, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/advSupply/media/getMediaInfo" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{
+			"status": "1",
+			"page": "1",
+			"all_page": "0",
+			"all_num": "0",
+			"data": [
+				{"id":"1","name":"有效媒体","sale_price":"12.5"},
+				{"id":"2","sale_price":"18.0"}
+			]
+		}`))
+	})
+	defer cleanup()
+
+	page, err := client.ListMedia(context.Background(), 1, 1, 2)
+	if err != nil {
+		t.Fatalf("list media: %v", err)
+	}
+	if page.RawItemCount != 2 {
+		t.Fatalf("expected raw count to include invalid upstream rows, got %d", page.RawItemCount)
+	}
+	if len(page.Items) != 1 {
+		t.Fatalf("expected only valid rows to be upserted, got %d", len(page.Items))
+	}
+}
+
 func TestDecodeMeijiequanSearchOptions(t *testing.T) {
 	groups, err := decodeMeijiequanSearchOptions([]byte(`{
 		"pindaoleixing": {"name": "频道类型", "list": ["IT科技", "财经金融"]},

server/internal/tenant/app/meijiequan_client.go

@@ -128,10 +128,11 @@ func (r *meijiequanMediaListResponse) UnmarshalJSON(body []byte) error {
 }
 
 type MeijiequanMediaPage struct {
-	Items   []UpsertSupplierMediaResourceInput
-	Page    int
-	AllPage int
-	AllNum  int
+	Items        []UpsertSupplierMediaResourceInput
+	RawItemCount int
+	Page         int
+	AllPage      int
+	AllNum       int
 }
 
 type meijiequanSearchOptionsCache struct {
@@ -687,6 +688,7 @@ func (c *MeijiequanClient) ListMedia(ctx context.Context, modelID, page, pageSiz
 		result.Page = parsed.Page
 		result.AllPage = parsed.AllPage
 		result.AllNum = parsed.AllNum
+		result.RawItemCount = len(parsed.Data)
 		result.Items = make([]UpsertSupplierMediaResourceInput, 0, len(parsed.Data))
 		for _, item := range parsed.Data {
 			resource, ok := parseMeijiequanMediaItem(modelID, item)