From 3f702084d20b3c3a3ea916f3110e99b22bda60b3 Mon Sep 17 00:00:00 2001
From: Chia <Chia@93.nz>
Date: Thu, 6 Aug 2026 15:58:57 +1200
Subject: feat: complete commercial developer workflows

Add tenant-safe usage observability, prepaid billing controls, API key lifecycle management, Embeddings metering, configurable billing alerts, and resilient provider health propagation. Harden Stripe failure handling, migrations, readiness, and the authenticated control-plane UI with end-to-end verification evidence.
---
 internal/controlplane/store.go | 48 ++++++++++++++++++++++++++++++------------
 1 file changed, 35 insertions(+), 13 deletions(-)

(limited to 'internal/controlplane/store.go')

diff --git a/internal/controlplane/store.go b/internal/controlplane/store.go
index c4d7016..82c87ae 100644
--- a/internal/controlplane/store.go
+++ b/internal/controlplane/store.go
@@ -23,7 +23,10 @@ var schemaSQL string
 
 var ErrRedisDisabled = errors.New("Redis propagation is disabled")
 
-const migrationVersion int64 = 2026080605
+const (
+	migrationVersion int64 = 2026080610
+	migrationLockID  int64 = 0x41494757 // "AIGW"; stable across migration versions.
+)
 
 type Options struct {
 	DatabaseURL            string
@@ -82,6 +85,13 @@ func (s *Store) RedisEnabled() bool {
 	return s.redis != nil
 }
 
+func (s *Store) PingRedis(ctx context.Context) error {
+	if s.redis == nil {
+		return ErrRedisDisabled
+	}
+	return s.redis.Ping(ctx).Err()
+}
+
 func (s *Store) Ping(ctx context.Context) error { return s.db.Ping(ctx) }
 
 func (s *Store) Migrate(ctx context.Context) error {
@@ -112,28 +122,40 @@ func MigrationStatusDatabase(ctx context.Context, databaseURL string) (Migration
 }
 
 func applySchema(ctx context.Context, db *pgxpool.Pool) error {
+	hash := sha256.Sum256([]byte(schemaSQL))
+	checksum := hex.EncodeToString(hash[:])
 	tx, err := db.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("begin migration: %w", err)
 	}
 	defer tx.Rollback(ctx)
-	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, migrationVersion); err != nil {
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, migrationLockID); err != nil {
 		return err
 	}
-	if _, err := tx.Exec(ctx, schemaSQL); err != nil {
-		return fmt.Errorf("apply control-plane schema: %w", err)
+	var migrationsExist bool
+	if err := tx.QueryRow(ctx, `SELECT to_regclass('schema_migrations') IS NOT NULL`).Scan(&migrationsExist); err != nil {
+		return fmt.Errorf("inspect migration table: %w", err)
 	}
-	hash := sha256.Sum256([]byte(schemaSQL))
-	checksum := hex.EncodeToString(hash[:])
-	var existing string
-	err = tx.QueryRow(ctx, `SELECT checksum FROM schema_migrations WHERE version=$1`, migrationVersion).Scan(&existing)
-	if err == nil && existing != checksum {
-		return fmt.Errorf("migration %d checksum changed; deploy an explicit new migration version", migrationVersion)
+	if migrationsExist {
+		var existing string
+		err = tx.QueryRow(ctx, `SELECT checksum FROM schema_migrations WHERE version=$1`, migrationVersion).Scan(&existing)
+		if err == nil {
+			if existing != checksum {
+				return fmt.Errorf("migration %d checksum changed; deploy an explicit new migration version", migrationVersion)
+			}
+			if err := tx.Commit(ctx); err != nil {
+				return fmt.Errorf("commit migration check: %w", err)
+			}
+			return nil
+		}
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return fmt.Errorf("read migration checksum: %w", err)
+		}
 	}
-	if !errors.Is(err, pgx.ErrNoRows) && err != nil {
-		return err
+	if _, err := tx.Exec(ctx, schemaSQL); err != nil {
+		return fmt.Errorf("apply control-plane schema: %w", err)
 	}
-	if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations(version,name,checksum) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING`, migrationVersion, "tenant-billing-profiles", checksum); err != nil {
+	if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations(version,name,checksum) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING`, migrationVersion, "tenant-spend-alert-preferences", checksum); err != nil {
 		return err
 	}
 	if err := tx.Commit(ctx); err != nil {
-- 
cgit v1.2.3