Merge branch 'main' into reverse-proxy-allow-match-or

management/server/migration/migration.go

@@ -683,3 +683,81 @@ func BackfillPublicIDs[T any](ctx context.Context, db *gorm.DB) error {
 	log.WithContext(ctx).Infof("Backfill of empty public_id in table %s completed", tableName)
 	return nil
 }
+
+// FoldCostAggregatesIntoBuckets migrates a per-request cost table from the old
+// "stored aggregate" shape (cost_usd + cache_cost_usd columns) to the per-bucket
+// breakdown, where the total and cache portion are derived on read instead.
+//
+// The fold preserves both aggregates exactly for historical rows: the cache
+// total moves into cached_input_cost_usd and the remainder into
+// input_cost_usd, so a row's derived total and cache cost still match what it
+// reported before the upgrade. The finer split is genuinely unknown for those
+// rows — the old schema never recorded a read/write or input/output division —
+// so it is lumped rather than guessed; only rows written after the upgrade
+// carry a true four-way split.
+//
+// Dropping the columns before folding would zero every historical row's cost,
+// so the update runs first and the drop only happens once it succeeds. A table
+// with no cost_usd column has already been migrated (or was created fresh) and
+// is skipped.
+func FoldCostAggregatesIntoBuckets[T any](ctx context.Context, db *gorm.DB) error {
+	var model T
+
+	if !db.Migrator().HasTable(&model) {
+		log.WithContext(ctx).Debugf("table for %T does not exist, no cost-bucket migration needed", model)
+		return nil
+	}
+	if !db.Migrator().HasColumn(&model, "cost_usd") {
+		log.WithContext(ctx).Debugf("table for %T has no cost_usd column, cost buckets already migrated", model)
+		return nil
+	}
+
+	stmt := &gorm.Statement{DB: db}
+	if err := stmt.Parse(&model); err != nil {
+		return fmt.Errorf("parse model schema: %w", err)
+	}
+	tableName := stmt.Schema.Table
+
+	// COALESCE guards rows whose new columns were added as NULL by an earlier
+	// AutoMigrate run that predates the NOT NULL default.
+	hasCacheColumn := db.Migrator().HasColumn(&model, "cache_cost_usd")
+	cacheExpr := "0"
+	if hasCacheColumn {
+		cacheExpr = "COALESCE(cache_cost_usd, 0)"
+	}
+
+	if err := db.Transaction(func(tx *gorm.DB) error {
+		// Only touch rows that carry a legacy total and no breakdown yet, so
+		// the migration is idempotent and never overwrites a true split.
+		update := fmt.Sprintf(`UPDATE %s
+			SET input_cost_usd = COALESCE(cost_usd, 0) - %s,
+			    cached_input_cost_usd = %s,
+			    cache_creation_cost_usd = 0,
+			    output_cost_usd = 0
+			WHERE COALESCE(cost_usd, 0) <> 0
+			  AND COALESCE(input_cost_usd, 0) = 0
+			  AND COALESCE(cached_input_cost_usd, 0) = 0
+			  AND COALESCE(cache_creation_cost_usd, 0) = 0
+			  AND COALESCE(output_cost_usd, 0) = 0`, tableName, cacheExpr, cacheExpr)
+		res := tx.Exec(update)
+		if res.Error != nil {
+			return fmt.Errorf("fold legacy cost aggregates in %s: %w", tableName, res.Error)
+		}
+		log.WithContext(ctx).Infof("folded legacy cost aggregates into per-bucket columns for %d rows in table %s", res.RowsAffected, tableName)
+
+		if err := tx.Migrator().DropColumn(&model, "cost_usd"); err != nil {
+			return fmt.Errorf("drop cost_usd from %s: %w", tableName, err)
+		}
+		if hasCacheColumn {
+			if err := tx.Migrator().DropColumn(&model, "cache_cost_usd"); err != nil {
+				return fmt.Errorf("drop cache_cost_usd from %s: %w", tableName, err)
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	log.WithContext(ctx).Infof("migration of stored cost aggregates to per-bucket columns in table %s completed", tableName)
+	return nil
+}

management/server/migration/migration_test.go

@@ -16,6 +16,7 @@ import (
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 
+	agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
 	"github.com/netbirdio/netbird/management/server/migration"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
 	"github.com/netbirdio/netbird/management/server/testutil"
@@ -639,3 +640,99 @@ func TestCleanupOrphanedResources_SkipsWhenForeignKeyExists(t *testing.T) {
 	db.Model(&testChildWithFK{}).Count(&count)
 	assert.Equal(t, int64(2), count, "Both rows should survive — migration must skip when FK constraint exists")
 }
+
+// legacyCostRow is the pre-breakdown shape of the usage table: cost was stored
+// as a total plus a cache portion, with no per-bucket columns. Used to build a
+// realistic pre-upgrade table for the fold migration to run against.
+type legacyCostRow struct {
+	ID           string `gorm:"primaryKey"`
+	AccountID    string
+	Model        string
+	CostUSD      float64
+	CacheCostUSD float64
+}
+
+func (legacyCostRow) TableName() string { return "agent_network_request_usage" }
+
+// TestFoldCostAggregatesIntoBuckets_PreservesHistoricalCost covers the upgrade
+// path: a table written under the old schema must come out with its per-row
+// total and cache cost unchanged, because dropping cost_usd without folding it
+// forward would silently zero every historical row's spend.
+func TestFoldCostAggregatesIntoBuckets_PreservesHistoricalCost(t *testing.T) {
+	ctx := context.Background()
+	db := setupDatabase(t)
+	// setupDatabase hands back a process-shared database, so start from a clean
+	// table rather than inheriting rows from another test.
+	require.NoError(t, db.Migrator().DropTable(&agentNetworkTypes.AgentNetworkUsage{}))
+
+	require.NoError(t, db.AutoMigrate(&legacyCostRow{}), "legacy table must be created")
+	require.NoError(t, db.Create(&legacyCostRow{
+		ID: "u1", AccountID: "acct-1", Model: "claude-sonnet-4-6", CostUSD: 0.0123, CacheCostUSD: 0.0029,
+	}).Error)
+	require.NoError(t, db.Create(&legacyCostRow{
+		ID: "u2", AccountID: "acct-1", Model: "gpt-4o", CostUSD: 0.5, CacheCostUSD: 0,
+	}).Error)
+	// A zero-cost row (denied / unpriced request) must stay zero, not be touched.
+	require.NoError(t, db.Create(&legacyCostRow{ID: "u3", AccountID: "acct-1", Model: "gw/unpriced"}).Error)
+
+	// AutoMigrate adds the per-bucket columns alongside the legacy ones, exactly
+	// as a real upgrade does before the post-auto migrations run.
+	require.NoError(t, db.AutoMigrate(&agentNetworkTypes.AgentNetworkUsage{}), "new columns must be added")
+
+	require.NoError(t, migration.FoldCostAggregatesIntoBuckets[agentNetworkTypes.AgentNetworkUsage](ctx, db))
+
+	assert.False(t, db.Migrator().HasColumn(&agentNetworkTypes.AgentNetworkUsage{}, "cost_usd"),
+		"legacy cost_usd column must be dropped once folded")
+	assert.False(t, db.Migrator().HasColumn(&agentNetworkTypes.AgentNetworkUsage{}, "cache_cost_usd"),
+		"legacy cache_cost_usd column must be dropped once folded")
+
+	var rows []*agentNetworkTypes.AgentNetworkUsage
+	require.NoError(t, db.Order("id").Find(&rows).Error)
+	require.Len(t, rows, 3)
+
+	// u1: total and cache portion both preserved; the read/write and
+	// input/output splits are unknowable for a legacy row, so the cache total
+	// lands on cached_input and the remainder on input.
+	assert.InDelta(t, 0.0123, rows[0].TotalCostUSD(), 1e-9, "historical total must survive the fold")
+	assert.InDelta(t, 0.0029, rows[0].CacheCostUSD(), 1e-9, "historical cache cost must survive the fold")
+	assert.InDelta(t, 0.0094, rows[0].InputCostUSD, 1e-9, "non-cache remainder lands on input")
+	assert.InDelta(t, 0.0029, rows[0].CachedInputCostUSD, 1e-9, "legacy cache total lands on cached input")
+	assert.Zero(t, rows[0].CacheCreationCostUSD, "legacy rows carry no read/write split to recover")
+	assert.Zero(t, rows[0].OutputCostUSD, "legacy rows carry no input/output split to recover")
+
+	// u2: no cache spend — the whole total is the non-cache remainder.
+	assert.InDelta(t, 0.5, rows[1].TotalCostUSD(), 1e-9, "cache-free historical total must survive")
+	assert.Zero(t, rows[1].CacheCostUSD(), "a cache-free row must stay cache-free")
+
+	// u3: zero stays zero rather than being rewritten.
+	assert.Zero(t, rows[2].TotalCostUSD(), "an unpriced row must remain unpriced")
+}
+
+// TestFoldCostAggregatesIntoBuckets_SkipsAlreadyMigrated proves the migration is
+// safe to re-run: with no legacy column present it is a no-op that leaves a
+// true four-way split untouched.
+func TestFoldCostAggregatesIntoBuckets_SkipsAlreadyMigrated(t *testing.T) {
+	ctx := context.Background()
+	db := setupDatabase(t)
+	require.NoError(t, db.Migrator().DropTable(&agentNetworkTypes.AgentNetworkUsage{}))
+
+	require.NoError(t, db.AutoMigrate(&agentNetworkTypes.AgentNetworkUsage{}))
+	// Timestamp must be set explicitly: a zero time.Time serialises as
+	// '0000-00-00 00:00:00', which MySQL rejects under strict mode.
+	require.NoError(t, db.Create(&agentNetworkTypes.AgentNetworkUsage{
+		ID: "u1", AccountID: "acct-1", Model: "claude-sonnet-4-6",
+		Timestamp:    time.Date(2026, 5, 5, 9, 0, 0, 0, time.UTC),
+		InputCostUSD: 0.001, CachedInputCostUSD: 0.002, CacheCreationCostUSD: 0.003, OutputCostUSD: 0.004,
+	}).Error)
+
+	require.NoError(t, migration.FoldCostAggregatesIntoBuckets[agentNetworkTypes.AgentNetworkUsage](ctx, db),
+		"running against an already-migrated table must be a no-op, not an error")
+
+	var row agentNetworkTypes.AgentNetworkUsage
+	require.NoError(t, db.First(&row, "id = ?", "u1").Error)
+	assert.InDelta(t, 0.001, row.InputCostUSD, 1e-9, "a true split must not be rewritten")
+	assert.InDelta(t, 0.002, row.CachedInputCostUSD, 1e-9)
+	assert.InDelta(t, 0.003, row.CacheCreationCostUSD, 1e-9)
+	assert.InDelta(t, 0.004, row.OutputCostUSD, 1e-9)
+	assert.InDelta(t, 0.01, row.TotalCostUSD(), 1e-9, "derived total sums the four buckets")
+}

management/server/store/sql_store_agentnetwork.go

@@ -71,7 +71,7 @@ func (s *SqlStore) GetAgentNetworkMetrics(ctx context.Context) (AgentNetworkMetr
 	usageRow := db.Model(&agentNetworkTypes.AgentNetworkUsage{}).
 		Select("COALESCE(SUM(input_tokens), 0) AS input_tokens, " +
 			"COALESCE(SUM(output_tokens), 0) AS output_tokens, " +
-			"COALESCE(SUM(cost_usd), 0) AS cost_usd").Row()
+			"COALESCE(SUM" + agentNetworkTypes.CostUSDSQLExpr + ", 0) AS cost_usd").Row()
 	if err := usageRow.Scan(&m.InputTokens, &m.OutputTokens, &m.CostUSD); err != nil {
 		return AgentNetworkMetrics{}, fmt.Errorf("scan agent network usage metrics: %w", err)
 	}

management/server/store/sql_store_agentnetwork_accesslog_test.go

@@ -37,7 +37,7 @@ func TestAgentNetworkUsage_RealStore_RoundTrip(t *testing.T) {
 		InputTokens:        1200,
 		OutputTokens:       640,
 		TotalTokens:        1840,
-		CostUSD:            0.0231,
+		InputCostUSD:       0.0231,
 	}
 	usageGroups := []agentNetworkTypes.AgentNetworkUsageGroup{
 		{UsageID: usage.ID, GroupID: "grp-eng", AccountID: accountID},
@@ -71,7 +71,7 @@ func TestAgentNetworkUsage_RealStore_RoundTrip(t *testing.T) {
 		InputTokens:  1200,
 		OutputTokens: 640,
 		TotalTokens:  1840,
-		CostUSD:      0.0231,
+		InputCostUSD: 0.0231,
 	}
 	entryGroups := []agentNetworkTypes.AgentNetworkAccessLogGroup{
 		{LogID: entry.ID, GroupID: "grp-eng", AccountID: accountID},
@@ -127,7 +127,7 @@ func TestAgentNetworkUsageOverview_DailyAggregation(t *testing.T) {
 	mk := func(id string, ts time.Time, model string, in, out int64, cost float64) *agentNetworkTypes.AgentNetworkUsage {
 		return &agentNetworkTypes.AgentNetworkUsage{
 			ID: id, AccountID: accountID, Timestamp: ts, Model: model,
-			InputTokens: in, OutputTokens: out, TotalTokens: in + out, CostUSD: cost,
+			InputTokens: in, OutputTokens: out, TotalTokens: in + out, InputCostUSD: cost,
 		}
 	}
 	require.NoError(t, s.CreateAgentNetworkUsage(ctx, mk("u1", day1, "gpt-4o", 100, 50, 0.10), nil))
@@ -143,7 +143,7 @@ func TestAgentNetworkUsageOverview_DailyAggregation(t *testing.T) {
 	assert.Equal(t, "2026-05-05", buckets[0].PeriodStart, "oldest-first ordering")
 	assert.Equal(t, int64(300), buckets[0].InputTokens, "same-day input tokens summed")
 	assert.Equal(t, int64(130), buckets[0].OutputTokens)
-	assert.InDelta(t, 0.30, buckets[0].CostUSD, 1e-9, "same-day cost summed")
+	assert.InDelta(t, 0.30, buckets[0].TotalCostUSD(), 1e-9, "same-day cost summed")
 	assert.Equal(t, "2026-05-06", buckets[1].PeriodStart)
 	assert.Equal(t, int64(15), buckets[1].TotalTokens)
 
@@ -174,7 +174,7 @@ func TestAgentNetworkAccessLogSessions_RealStore(t *testing.T) {
 			ID: id, AccountID: accountID, ServiceID: "svc", Timestamp: ts,
 			UserID: user, StatusCode: 200, Provider: provider, Model: model,
 			SessionID: session, Decision: decision,
-			InputTokens: 100, OutputTokens: 50, TotalTokens: 150, CostUSD: cost,
+			InputTokens: 100, OutputTokens: 50, TotalTokens: 150, InputCostUSD: cost,
 		}
 	}
 
@@ -207,7 +207,7 @@ func TestAgentNetworkAccessLogSessions_RealStore(t *testing.T) {
 	s1 := sessions[2]
 	assert.Equal(t, 2, s1.RequestCount, "s1 has two requests")
 	assert.Equal(t, int64(300), s1.TotalTokens, "tokens summed across the session")
-	assert.InDelta(t, 0.30, s1.CostUSD, 1e-9, "cost summed across the session")
+	assert.InDelta(t, 0.30, s1.TotalCostUSD(), 1e-9, "cost summed across the session")
 	assert.Equal(t, "alice", s1.UserID)
 	assert.Equal(t, "allow", s1.Decision)
 	// SQLite hands times back in time.Local; normalise to UTC so the instant is

management/server/store/store.go

@@ -650,6 +650,14 @@ func getMigrationsPostAuto(ctx context.Context) []migrationFunc {
 		func(db *gorm.DB) error {
 			return migration.DropIndex[proxy.Proxy](ctx, db, "idx_proxy_account_id_unique")
 		},
+		// Post-auto so the per-bucket cost columns already exist when the legacy
+		// aggregates are folded into them and dropped.
+		func(db *gorm.DB) error {
+			return migration.FoldCostAggregatesIntoBuckets[agentNetworkTypes.AgentNetworkAccessLog](ctx, db)
+		},
+		func(db *gorm.DB) error {
+			return migration.FoldCostAggregatesIntoBuckets[agentNetworkTypes.AgentNetworkUsage](ctx, db)
+		},
 	}
 }