fix(analytics): keep unknown usage and commitment unknown in the savings rollups

internal/database/postgres/migrations/000147_savings_rollups_propagate_unknown.down.sql

@@ -0,0 +1,73 @@
+-- 000147 down: restore the definitions that SUM/AVG over the nullable cost
+-- columns and so skip unknown rows: monthly_savings_summary as 000078 left
+-- it, provider_savings_summary as 000074 left it, with the plain-column
+-- NULLS NOT DISTINCT unique indexes from 000076.
+DROP MATERIALIZED VIEW IF EXISTS monthly_savings_summary CASCADE;
+
+CREATE MATERIALIZED VIEW monthly_savings_summary AS
+SELECT
+    month,
+    account_id,
+    cloud_account_id,
+    provider,
+    service,
+    AVG(ts_savings) as total_savings,
+    AVG(ts_coverage) as avg_coverage,
+    AVG(ts_commitment) as total_commitment,
+    AVG(ts_usage) as total_usage,
+    SUM(ts_row_count)::BIGINT as snapshot_count,
+    MAX(timestamp) as last_updated
+FROM (
+    SELECT
+        DATE_TRUNC('month', timestamp) as month,
+        timestamp,
+        account_id,
+        cloud_account_id,
+        provider,
+        service,
+        SUM(total_savings) as ts_savings,
+        AVG(coverage_percentage) as ts_coverage,
+        SUM(total_commitment) as ts_commitment,
+        SUM(total_usage) as ts_usage,
+        COUNT(*) as ts_row_count
+    FROM savings_snapshots
+    GROUP BY DATE_TRUNC('month', timestamp), timestamp,
+             account_id, cloud_account_id, provider, service
+) per_ts
+GROUP BY month, account_id, cloud_account_id, provider, service;
+
+CREATE UNIQUE INDEX idx_monthly_savings_summary_unique
+    ON monthly_savings_summary (month, account_id, cloud_account_id, provider, service)
+    NULLS NOT DISTINCT;
+
+DROP MATERIALIZED VIEW IF EXISTS provider_savings_summary CASCADE;
+
+CREATE MATERIALIZED VIEW provider_savings_summary AS
+SELECT
+    provider,
+    account_id,
+    cloud_account_id,
+    MAX(ts_service_count) as service_count,
+    AVG(ts_savings) as total_savings,
+    AVG(ts_commitment) as total_commitment,
+    AVG(ts_coverage) as avg_coverage,
+    MAX(timestamp) as last_updated
+FROM (
+    SELECT
+        provider,
+        account_id,
+        cloud_account_id,
+        timestamp,
+        SUM(total_savings) as ts_savings,
+        SUM(total_commitment) as ts_commitment,
+        AVG(coverage_percentage) as ts_coverage,
+        COUNT(DISTINCT service) as ts_service_count
+    FROM savings_snapshots
+    WHERE timestamp > NOW() - INTERVAL '90 days'
+    GROUP BY provider, account_id, cloud_account_id, timestamp
+) per_ts
+GROUP BY provider, account_id, cloud_account_id;
+
+CREATE UNIQUE INDEX idx_provider_savings_summary_unique
+    ON provider_savings_summary (provider, account_id, cloud_account_id)
+    NULLS NOT DISTINCT;

internal/database/postgres/migrations/000147_savings_rollups_propagate_unknown.up.sql

@@ -0,0 +1,96 @@
+-- 000147: keep an unknown cost unknown through the savings rollups (#151).
+--
+-- savings_snapshots.total_usage (000067, H2) and total_commitment (000101)
+-- are nullable on purpose: the collector writes NULL when no provider figure
+-- exists (for example an all-upfront purchase has no recurring cost to derive
+-- usage from), because a placeholder 0 would read as a real $0. SUM and AVG
+-- skip NULL, so the rollups below turned "us-east-1 $4000 + eu-west-1
+-- unknown" into a definite $4000 and dropped the unknown region without any
+-- signal.
+--
+-- Each rollup now yields NULL when any row it covers is NULL, in the same
+-- COUNT(col) = COUNT(*) form the history totals use for upfront_cost
+-- (internal/api/analytics_postgres.go). The inner query makes a timestamp's
+-- total unknown if one of its rows is unknown; the outer query makes the
+-- period's average unknown if one of its timestamps is, since averaging only
+-- the known instants would understate a period in which a purchase with no
+-- figure became active. A known 0 is still 0. total_savings is NOT NULL and
+-- coverage_percentage is a percentage averaged over the rows that have one,
+-- so both keep their current definitions.
+--
+-- daily_savings_trend aggregates no nullable cost column and is unchanged.
+--
+-- DROP + CREATE because a materialized view's query cannot be altered in
+-- place. Both views are created populated, and their unique indexes are
+-- recreated as 000076 left them, so refresh_savings_materialized_views() can
+-- keep refreshing them CONCURRENTLY. The column names and types are
+-- unchanged (the CASE has no ELSE, so it keeps the numeric type of SUM/AVG).
+DROP MATERIALIZED VIEW IF EXISTS monthly_savings_summary CASCADE;
+
+CREATE MATERIALIZED VIEW monthly_savings_summary AS
+SELECT
+    month,
+    account_id,
+    cloud_account_id,
+    provider,
+    service,
+    AVG(ts_savings) as total_savings,
+    AVG(ts_coverage) as avg_coverage,
+    CASE WHEN COUNT(ts_commitment) = COUNT(*) THEN AVG(ts_commitment) END as total_commitment,
+    CASE WHEN COUNT(ts_usage) = COUNT(*) THEN AVG(ts_usage) END as total_usage,
+    SUM(ts_row_count)::BIGINT as snapshot_count,
+    MAX(timestamp) as last_updated
+FROM (
+    SELECT
+        DATE_TRUNC('month', timestamp) as month,
+        timestamp,
+        account_id,
+        cloud_account_id,
+        provider,
+        service,
+        SUM(total_savings) as ts_savings,
+        AVG(coverage_percentage) as ts_coverage,
+        CASE WHEN COUNT(total_commitment) = COUNT(*) THEN SUM(total_commitment) END as ts_commitment,
+        CASE WHEN COUNT(total_usage) = COUNT(*) THEN SUM(total_usage) END as ts_usage,
+        COUNT(*) as ts_row_count
+    FROM savings_snapshots
+    GROUP BY DATE_TRUNC('month', timestamp), timestamp,
+             account_id, cloud_account_id, provider, service
+) per_ts
+GROUP BY month, account_id, cloud_account_id, provider, service;
+
+CREATE UNIQUE INDEX idx_monthly_savings_summary_unique
+    ON monthly_savings_summary (month, account_id, cloud_account_id, provider, service)
+    NULLS NOT DISTINCT;
+
+DROP MATERIALIZED VIEW IF EXISTS provider_savings_summary CASCADE;
+
+CREATE MATERIALIZED VIEW provider_savings_summary AS
+SELECT
+    provider,
+    account_id,
+    cloud_account_id,
+    MAX(ts_service_count) as service_count,
+    AVG(ts_savings) as total_savings,
+    CASE WHEN COUNT(ts_commitment) = COUNT(*) THEN AVG(ts_commitment) END as total_commitment,
+    AVG(ts_coverage) as avg_coverage,
+    MAX(timestamp) as last_updated
+FROM (
+    SELECT
+        provider,
+        account_id,
+        cloud_account_id,
+        timestamp,
+        SUM(total_savings) as ts_savings,
+        CASE WHEN COUNT(total_commitment) = COUNT(*) THEN SUM(total_commitment) END as ts_commitment,
+        AVG(coverage_percentage) as ts_coverage,
+        COUNT(DISTINCT service) as ts_service_count
+    FROM savings_snapshots
+    WHERE timestamp > NOW() - INTERVAL '90 days'
+    GROUP BY provider, account_id, cloud_account_id, timestamp
+) per_ts
+GROUP BY provider, account_id, cloud_account_id;
+
+CREATE UNIQUE INDEX idx_provider_savings_summary_unique
+    ON provider_savings_summary (provider, account_id, cloud_account_id)
+    NULLS NOT DISTINCT;

internal/database/postgres/migrations/000147_savings_rollups_propagate_unknown_test.go

@@ -0,0 +1,198 @@
+//go:build integration
+
+package migrations_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const rollupsUnknownVersion = 147
+
+// snapshotRow147 is one savings_snapshots row; a nil cost is an unknown one.
+type snapshotRow147 struct {
+	account    string
+	at         time.Time
+	region     string
+	usage      *float64
+	commitment *float64
+}
+
+// rollup147 is what the two views report for one account.
+type rollup147 struct {
+	monthlyUsage, monthlyCommitment, providerCommitment *float64
+}
+
+func cost147(v float64) *float64 { return &v }
+
+func insertSnapshots147(ctx context.Context, t *testing.T, pool *pgxpool.Pool, rows []snapshotRow147) {
+	t.Helper()
+	for _, r := range rows {
+		_, err := pool.Exec(ctx, `INSERT INTO savings_snapshots
+			(account_id, timestamp, provider, service, region, commitment_type,
+			 total_commitment, total_usage, total_savings)
+			VALUES ($1, $2, 'aws', 'ec2', $3, 'RI', $4, $5, 10)`,
+			r.account, r.at, r.region, r.commitment, r.usage)
+		require.NoError(t, err)
+	}
+}
+
+func readRollup147(ctx context.Context, t *testing.T, pool *pgxpool.Pool, account string) rollup147 {
+	t.Helper()
+	var got rollup147
+	require.NoError(t, pool.QueryRow(ctx, `SELECT total_usage::float8, total_commitment::float8
+		FROM monthly_savings_summary WHERE account_id = $1`, account).
+		Scan(&got.monthlyUsage, &got.monthlyCommitment), "monthly row for %s", account)
+	require.NoError(t, pool.QueryRow(ctx, `SELECT total_commitment::float8
+		FROM provider_savings_summary WHERE account_id = $1`, account).
+		Scan(&got.providerCommitment), "provider row for %s", account)
+	return got
+}
+
+// requireRollup147 checks every column against want, where nil means the view
+// must report NULL (unknown) and a value means that exact known figure.
+func requireRollup147(t *testing.T, label string, want *float64, got rollup147) {
+	t.Helper()
+	for name, v := range map[string]*float64{
+		"monthly total_usage":       got.monthlyUsage,
+		"monthly total_commitment":  got.monthlyCommitment,
+		"provider total_commitment": got.providerCommitment,
+	} {
+		if want == nil {
+			assert.Nil(t, v, "%s: %s must be unknown (NULL)", label, name)
+			continue
+		}
+		if assert.NotNil(t, v, "%s: %s must be known", label, name) {
+			assert.InDelta(t, *want, *v, 0.001, "%s: %s", label, name)
+		}
+	}
+}
+
+// TestMigration_SavingsRollupsPropagateUnknown locks down migration 000147
+// (issue #151): monthly_savings_summary and provider_savings_summary report an
+// unknown total_usage / total_commitment as NULL instead of summing or
+// averaging past it, keep a known zero as 0, keep the H5 sum-then-average
+// result for fully known data, and the down migration restores the previous
+// view definitions exactly.
+func TestMigration_SavingsRollupsPropagateUnknown(t *testing.T) {
+	ctx := context.Background()
+	container := testhelpers.RequirePostgresContainer(ctx, t)
+	t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) })
+	pool, path := container.DB.Pool(), getMigrationsPath()
+	prev := previousMigrationVersion(t, rollupsUnknownVersion)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev))
+	prevDefs := viewDefs147(ctx, t, pool)
+
+	// Two collection instants inside the current month, whose partition the
+	// migrations create, and inside provider_savings_summary's 90-day window.
+	now := time.Now().UTC()
+	t1 := time.Date(now.Year(), now.Month(), 1, 1, 0, 0, 0, time.UTC)
+	t2 := t1.Add(time.Hour)
+	insertSnapshots147(ctx, t, pool, []snapshotRow147{
+		// One region known, one with no provider figure at the same instant.
+		{"acct-mixed", t1, "us-east-1", cost147(4000), cost147(4000)},
+		{"acct-mixed", t1, "eu-west-1", nil, nil},
+		// A real, known zero in every row.
+		{"acct-zero", t1, "us-east-1", cost147(0), cost147(0)},
+		{"acct-zero", t1, "eu-west-1", cost147(0), cost147(0)},
+		// Fully known: AVG(100+50, 200+100) = 225.
+		{"acct-known", t1, "us-east-1", cost147(100), cost147(100)},
+		{"acct-known", t1, "eu-west-1", cost147(50), cost147(50)},
+		{"acct-known", t2, "us-east-1", cost147(200), cost147(200)},
+		{"acct-known", t2, "eu-west-1", cost147(100), cost147(100)},
+		// Known at the first instant, unknown at the second.
+		{"acct-later", t1, "us-east-1", cost147(100), cost147(100)},
+		{"acct-later", t2, "us-east-1", nil, nil},
+	})
+
+	// The previous definitions report the unknown rows as if they were absent.
+	_, err := pool.Exec(ctx, `REFRESH MATERIALIZED VIEW monthly_savings_summary;
+		REFRESH MATERIALIZED VIEW provider_savings_summary`)
+	require.NoError(t, err)
+	requireRollup147(t, "before, mixed", cost147(4000), readRollup147(ctx, t, pool, "acct-mixed"))
+	requireRollup147(t, "before, later", cost147(100), readRollup147(ctx, t, pool, "acct-later"))
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, rollupsUnknownVersion))
+	// The production refresh path (CONCURRENTLY) still works on the new views.
+	_, err = pool.Exec(ctx, `SELECT refresh_savings_materialized_views()`)
+	require.NoError(t, err)
+
+	requireRollup147(t, "mixed", nil, readRollup147(ctx, t, pool, "acct-mixed"))
+	requireRollup147(t, "later", nil, readRollup147(ctx, t, pool, "acct-later"))
+	requireRollup147(t, "zero", cost147(0), readRollup147(ctx, t, pool, "acct-zero"))
+	requireRollup147(t, "known", cost147(225), readRollup147(ctx, t, pool, "acct-known"))
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev))
+	assert.Equal(t, prevDefs, viewDefs147(ctx, t, pool), "down must restore the previous view definitions")
+	requireRollup147(t, "after down, mixed", cost147(4000), readRollup147(ctx, t, pool, "acct-mixed"))
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, rollupsUnknownVersion))
+	requireRollup147(t, "up again, mixed", nil, readRollup147(ctx, t, pool, "acct-mixed"))
+}
+
+// viewDefs147 returns the two views' definitions and unique indexes.
+func viewDefs147(ctx context.Context, t *testing.T, pool *pgxpool.Pool) map[string]string {
+	t.Helper()
+	defs := map[string]string{}
+	for _, view := range []string{"monthly_savings_summary", "provider_savings_summary"} {
+		var def, idx string
+		require.NoError(t, pool.QueryRow(ctx, `SELECT pg_get_viewdef($1::text::regclass),
+			(SELECT string_agg(indexdef, ';' ORDER BY indexname) FROM pg_indexes WHERE tablename = $1::text)`,
+			view).Scan(&def, &idx))
+		defs[view] = def + "\n" + idx
+	}
+	return defs
+}
+
+// TestMigration_SavingsRollupsPropagateUnknown_ConcurrentRefresh covers the
+// deploy-time race: the collector's refresh_savings_materialized_views() runs
+// while 000147 is dropping and recreating the views. The refresh must wait for
+// the migration and then refresh the new views, not fail on the dropped ones.
+func TestMigration_SavingsRollupsPropagateUnknown_ConcurrentRefresh(t *testing.T) {
+	ctx := context.Background()
+	container := testhelpers.RequirePostgresContainer(ctx, t)
+	t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) })
+	pool, path := container.DB.Pool(), getMigrationsPath()
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousMigrationVersion(t, rollupsUnknownVersion)))
+
+	now := time.Now().UTC()
+	t1 := time.Date(now.Year(), now.Month(), 1, 1, 0, 0, 0, time.UTC)
+	insertSnapshots147(ctx, t, pool, []snapshotRow147{
+		{"acct-mixed", t1, "us-east-1", cost147(4000), cost147(4000)},
+		{"acct-mixed", t1, "eu-west-1", nil, nil},
+	})
+
+	// A reader holding provider_savings_summary stops the migration after it
+	// has replaced monthly_savings_summary, with that view still locked.
+	blocker, err := pool.Begin(ctx)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = blocker.Rollback(context.Background()) })
+	_, err = blocker.Exec(ctx, `SELECT count(*) FROM provider_savings_summary`)
+	require.NoError(t, err)
+
+	migrated := make(chan error, 1)
+	go func() { migrated <- migrations.MigrateToVersion(ctx, pool, path, rollupsUnknownVersion) }()
+	// pg_stat_activity keeps only the start of the migration's text, so match
+	// on its header comment.
+	testhelpers.WaitForLockWait(ctx, t, pool, "-- 000147:%")
+
+	refreshed := make(chan error, 1)
+	go func() {
+		_, err := pool.Exec(ctx, `SELECT refresh_savings_materialized_views()`)
+		refreshed <- err
+	}()
+	testhelpers.WaitForLockWait(ctx, t, pool, "SELECT refresh_savings_materialized_views()")
+
+	require.NoError(t, blocker.Rollback(ctx))
+	require.NoError(t, testhelpers.Await(t, migrated))
+	require.NoError(t, testhelpers.Await(t, refreshed), "a refresh queued behind the migration must succeed")
+	requireRollup147(t, "after racing refresh", nil, readRollup147(ctx, t, pool, "acct-mixed"))
+}