llmcost/sync/sync.go

package sync

import (
	"context"
	"log/slog"
	"net/http"
	"strings"

	"github.com/openmeterio/openmeter/openmeter/llmcost"
)

// PriceFilterFunc is called for each source price after normalization.
// Return true to include the price, false to exclude it.
type PriceFilterFunc func(llmcost.SourcePrice) bool

// SyncJob orchestrates fetching prices from external sources,
// normalizing model IDs, and reconciling into global prices.
type SyncJob struct {
	fetchers   []Fetcher
	normalizer ModelIDNormalizer
	reconciler *Reconciler
	filter     PriceFilterFunc
	logger     *slog.Logger
}

// SyncJobConfig contains the dependencies for creating a SyncJob.
type SyncJobConfig struct {
	HTTPClient *http.Client
	Repo       llmcost.Adapter
	Logger     *slog.Logger

	// Fetchers is the list of price fetchers to use.
	// If nil, the default built-in fetchers are used.
	Fetchers []Fetcher

	// MinSourceAgreement is the minimum number of sources that must agree on a price
	// for it to be reconciled. Zero uses DefaultMinSourceAgreement.
	MinSourceAgreement int

	// PriceTolerance is the maximum allowed percentage difference (0.0–1.0) between
	// source prices for them to be considered in agreement. Negative uses DefaultPriceTolerance.
	PriceTolerance float64

	// Filter is an optional function called for each source price after normalization.
	// If set, only prices for which it returns true are included in reconciliation.
	Filter PriceFilterFunc
}

// DefaultFetchers returns the built-in price fetchers.
func DefaultFetchers(client *http.Client) []Fetcher {
	if client == nil {
		client = http.DefaultClient
	}

	return []Fetcher{
		NewModelsDevFetcher(client),
	}
}

// NewSyncJob creates a new sync job with all configured fetchers.
func NewSyncJob(config SyncJobConfig) *SyncJob {
	normalizer := NewDefaultNormalizer()

	fetchers := config.Fetchers
	if fetchers == nil {
		fetchers = DefaultFetchers(config.HTTPClient)
	}

	// Cap minAgreement at the number of fetchers — we can't require more sources
	// to agree than we actually have.
	minAgreement := config.MinSourceAgreement
	if minAgreement <= 0 {
		minAgreement = DefaultMinSourceAgreement
	}

	if numFetchers := len(fetchers); numFetchers > 0 && minAgreement > numFetchers {
		minAgreement = numFetchers
	}

	return &SyncJob{
		fetchers:   fetchers,
		normalizer: normalizer,
		reconciler: NewReconciler(config.Repo, config.Logger, minAgreement, config.PriceTolerance),
		filter:     config.Filter,
		logger:     config.Logger,
	}
}

// sourceModelKey is used to deduplicate prices from the same source after normalization.
type sourceModelKey struct {
	Source   llmcost.PriceSource
	Provider string
	ModelID  string
}

// deduplicateSourcePrices removes duplicate entries that share the same (source, provider, model_id)
// after normalization. When duplicates exist, it prefers entries whose model name does not contain
// a provider prefix (e.g., "GPT-4o" is preferred over "azure/gpt-4o").
func deduplicateSourcePrices(prices []llmcost.SourcePrice) []llmcost.SourcePrice {
	seen := make(map[sourceModelKey]int, len(prices)) // value is index into result
	result := make([]llmcost.SourcePrice, 0, len(prices))

	for _, p := range prices {
		key := sourceModelKey{Source: p.Source, Provider: string(p.Provider), ModelID: p.ModelID}

		if idx, exists := seen[key]; exists {
			existing := result[idx]
			existingHasPrefix := strings.Contains(existing.ModelName, "/")
			newHasPrefix := strings.Contains(p.ModelName, "/")

			// Prefer entries without a provider prefix (e.g., "GPT-4o" over "azure/gpt-4o").
			// When both have the same prefix status, use lexicographic order as a deterministic tie-breaker.
			if existingHasPrefix && !newHasPrefix ||
				existingHasPrefix == newHasPrefix && p.ModelName < existing.ModelName {
				result[idx] = p
			}

			continue
		}

		seen[key] = len(result)
		result = append(result, p)
	}

	return result
}

// Run executes the full sync cycle: fetch → normalize → deduplicate → reconcile.
func (j *SyncJob) Run(ctx context.Context) error {
	var allPrices []llmcost.SourcePrice

	// Phase 1: Fetch from all sources and normalize
	for _, f := range j.fetchers {
		sourceName := f.Source()

		j.logger.Info("fetching prices", "source", sourceName)

		prices, err := f.Fetch(ctx)
		if err != nil {
			j.logger.Error("failed to fetch prices",
				"source", sourceName,
				"error", err)

			continue // Don't fail entire sync if one source is down
		}

		j.logger.Info("fetched prices",
			"source", sourceName,
			"count", len(prices))

		// Normalize model IDs and provider names
		for _, p := range prices {
			provider, modelID := j.normalizer.Normalize(p.ModelID, string(p.Provider))
			p.Provider = llmcost.Provider(provider)
			p.ModelID = modelID
			allPrices = append(allPrices, p)
		}
	}

	// Phase 1.5: Deduplicate within each source after normalization.
	// Provider normalization can collapse multiple raw entries (e.g., azure/gpt-4o and openai/gpt-4o)
	// into the same (source, provider, model_id) key. Without deduplication, these would create
	// false multi-source agreement in the reconciler.
	allPrices = deduplicateSourcePrices(allPrices)

	j.logger.Info("deduplicated prices", "count", len(allPrices))

	// Phase 2: Filter (optional)
	if j.filter != nil {
		filtered := allPrices[:0]
		for _, p := range allPrices {
			if j.filter(p) {
				filtered = append(filtered, p)
			}
		}

		j.logger.Info("filtered prices",
			"before", len(allPrices),
			"after", len(filtered))

		allPrices = filtered
	}

	// Phase 3: Reconcile across sources and upsert global prices
	j.logger.Info("starting reconciliation", "total_prices", len(allPrices))

	return j.reconciler.Reconcile(ctx, allPrices)
}