Fix Database Schema migration related changes

Author: VirajSalaka

gateway/gateway-controller/pkg/storage/gateway-controller-db.sql

@@ -150,7 +150,7 @@ CREATE INDEX IF NOT EXISTS idx_api_key_status ON api_keys(status);
 CREATE INDEX IF NOT EXISTS idx_api_key_expiry ON api_keys(expires_at);
 CREATE INDEX IF NOT EXISTS idx_created_by ON api_keys(created_by);
 
--- EventHub: Organization States Table (added in schema version 9)
+-- EventHub: Organization States Table (added in schema version 6)
 -- Tracks version information per organization for change detection
 CREATE TABLE IF NOT EXISTS organization_states (
     organization TEXT PRIMARY KEY,
@@ -160,7 +160,7 @@ CREATE TABLE IF NOT EXISTS organization_states (
 
 CREATE INDEX IF NOT EXISTS idx_organization_states_updated ON organization_states(updated_at);
 
--- EventHub: Unified Events Table (added in schema version 9)
+-- EventHub: Unified Events Table (added in schema version 6)
 -- Stores all entity change events (APIs, certificates, LLM templates, etc.)
 CREATE TABLE IF NOT EXISTS events (
     organization_id TEXT NOT NULL,
@@ -177,5 +177,5 @@ CREATE TABLE IF NOT EXISTS events (
 CREATE INDEX IF NOT EXISTS idx_events_lookup ON events(organization_id, processed_timestamp);
 CREATE INDEX IF NOT EXISTS idx_events_type ON events(event_type);
 
--- Set schema version to 9
-PRAGMA user_version = 9;
+-- Set schema version to 6
+PRAGMA user_version = 6;

gateway/gateway-controller/pkg/storage/sqlite.go

@@ -215,120 +215,10 @@ func (s *SQLiteStorage) initSchema() error {
 		}
 
 		if version == 5 {
-			// Add event tables for EventHub
-			if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS api_events (
-				id INTEGER PRIMARY KEY AUTOINCREMENT,
-				organization_id TEXT NOT NULL,
-				processed_timestamp TIMESTAMP NOT NULL,
-				originated_timestamp TIMESTAMP NOT NULL,
-				event_data TEXT NOT NULL
-			);`); err != nil {
-				return fmt.Errorf("failed to migrate schema to version 6 (api_events): %w", err)
-			}
-			if _, err := s.db.Exec(`CREATE INDEX IF NOT EXISTS idx_api_events_lookup ON api_events(organization_id, processed_timestamp);`); err != nil {
-				return fmt.Errorf("failed to create api_events index: %w", err)
-			}
-
-			if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS certificate_events (
-				id INTEGER PRIMARY KEY AUTOINCREMENT,
-				organization_id TEXT NOT NULL,
-				processed_timestamp TIMESTAMP NOT NULL,
-				originated_timestamp TIMESTAMP NOT NULL,
-				event_data TEXT NOT NULL
-			);`); err != nil {
-				return fmt.Errorf("failed to migrate schema to version 6 (certificate_events): %w", err)
-			}
-			if _, err := s.db.Exec(`CREATE INDEX IF NOT EXISTS idx_cert_events_lookup ON certificate_events(organization_id, processed_timestamp);`); err != nil {
-				return fmt.Errorf("failed to create certificate_events index: %w", err)
-			}
-
-			if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS llm_template_events (
-				id INTEGER PRIMARY KEY AUTOINCREMENT,
-				organization_id TEXT NOT NULL,
-				processed_timestamp TIMESTAMP NOT NULL,
-				originated_timestamp TIMESTAMP NOT NULL,
-				event_data TEXT NOT NULL
-			);`); err != nil {
-				return fmt.Errorf("failed to migrate schema to version 6 (llm_template_events): %w", err)
-			}
-			if _, err := s.db.Exec(`CREATE INDEX IF NOT EXISTS idx_llm_events_lookup ON llm_template_events(organization_id, processed_timestamp);`); err != nil {
-				return fmt.Errorf("failed to create llm_template_events index: %w", err)
-			}
-
-			if _, err := s.db.Exec("PRAGMA user_version = 6"); err != nil {
-				return fmt.Errorf("failed to set schema version to 6: %w", err)
-			}
-			s.logger.Info("Schema migrated to version 6 (event tables)")
-			version = 6
-		}
-
-		if version == 6 {
-			// Add topic_states table
-			if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS topic_states (
-				topic_name TEXT PRIMARY KEY,
-				version_id TEXT NOT NULL DEFAULT '',
-				updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
-			);`); err != nil {
-				return fmt.Errorf("failed to migrate schema to version 7 (topic_states): %w", err)
-			}
-			if _, err := s.db.Exec(`CREATE INDEX IF NOT EXISTS idx_topic_states_updated ON topic_states(updated_at);`); err != nil {
-				return fmt.Errorf("failed to create topic_states index: %w", err)
-			}
-			if _, err := s.db.Exec("PRAGMA user_version = 7"); err != nil {
-				return fmt.Errorf("failed to set schema version to 7: %w", err)
-			}
-			s.logger.Info("Schema migrated to version 7 (topic_states table)")
-			version = 7
-		}
-
-		if version == 7 {
-			// Migrate topic_states to include organization as part of primary key
-			s.logger.Info("Migrating topic_states table to include organization (version 8)")
-
-			// Step 1: Rename old table
-			if _, err := s.db.Exec(`ALTER TABLE topic_states RENAME TO topic_states_old;`); err != nil {
-				return fmt.Errorf("failed to rename topic_states table: %w", err)
-			}
-
-			// Step 2: Create new table with organization
-			if _, err := s.db.Exec(`CREATE TABLE topic_states (
-				organization TEXT NOT NULL,
-				topic_name TEXT NOT NULL,
-				version_id TEXT NOT NULL DEFAULT '',
-				updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
-				PRIMARY KEY (organization, topic_name)
-			);`); err != nil {
-				return fmt.Errorf("failed to create new topic_states table: %w", err)
-			}
-
-			// Step 3: Migrate data (set organization to empty string for existing rows)
-			if _, err := s.db.Exec(`INSERT INTO topic_states (organization, topic_name, version_id, updated_at)
-				SELECT '', topic_name, version_id, updated_at FROM topic_states_old;`); err != nil {
-				return fmt.Errorf("failed to migrate topic_states data: %w", err)
-			}
-
-			// Step 4: Drop old table
-			if _, err := s.db.Exec(`DROP TABLE topic_states_old;`); err != nil {
-				return fmt.Errorf("failed to drop old topic_states table: %w", err)
-			}
-
-			// Step 5: Recreate index
-			if _, err := s.db.Exec(`CREATE INDEX IF NOT EXISTS idx_topic_states_updated ON topic_states(updated_at);`); err != nil {
-				return fmt.Errorf("failed to create topic_states index: %w", err)
-			}
-
-			if _, err := s.db.Exec("PRAGMA user_version = 8"); err != nil {
-				return fmt.Errorf("failed to set schema version to 8: %w", err)
-			}
-			s.logger.Info("Schema migrated to version 8 (topic_states with organization)")
-			version = 8
-		}
-
-		if version == 8 {
-			// Migrate to organization-centric Event Hub architecture
-			s.logger.Info("Migrating to organization-centric Event Hub (version 9)")
+			// Add EventHub tables for multi-replica synchronization
+			s.logger.Info("Migrating to EventHub schema (version 6)")
 
-			// Step 1: Create new organization_states table
+			// Create organization_states table
 			if _, err := s.db.Exec(`CREATE TABLE organization_states (
 				organization TEXT PRIMARY KEY,
 				version_id TEXT NOT NULL DEFAULT '',
@@ -341,15 +231,7 @@ func (s *SQLiteStorage) initSchema() error {
 				return fmt.Errorf("failed to create organization_states index: %w", err)
 			}
 
-			// Step 2: Migrate state data (consolidate per organization)
-			if _, err := s.db.Exec(`INSERT INTO organization_states (organization, version_id, updated_at)
-				SELECT organization, MAX(version_id), MAX(updated_at)
-				FROM topic_states
-				GROUP BY organization;`); err != nil {
-				return fmt.Errorf("failed to migrate state data: %w", err)
-			}
-
-			// Step 3: Create unified events table
+			// Create unified events table
 			if _, err := s.db.Exec(`CREATE TABLE events (
 				organization_id TEXT NOT NULL,
 				processed_timestamp TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
@@ -372,17 +254,12 @@ func (s *SQLiteStorage) initSchema() error {
 				return fmt.Errorf("failed to create events type index: %w", err)
 			}
 
-			// Step 4: Drop old topic_states table (data already migrated)
-			if _, err := s.db.Exec(`DROP TABLE topic_states;`); err != nil {
-				return fmt.Errorf("failed to drop topic_states table: %w", err)
-			}
-
-			if _, err := s.db.Exec("PRAGMA user_version = 9"); err != nil {
-				return fmt.Errorf("failed to set schema version to 9: %w", err)
+			if _, err := s.db.Exec("PRAGMA user_version = 6"); err != nil {
+				return fmt.Errorf("failed to set schema version to 6: %w", err)
 			}
 
-			s.logger.Info("Schema migrated to version 9 (organization-centric Event Hub)")
-			version = 9
+			s.logger.Info("Schema migrated to version 6 (EventHub tables)")
+			version = 6
 		}
 
 		s.logger.Info("Database schema up to date", zap.Int("version", version))

gateway/gateway-controller/tests/integration/schema_test.go

@@ -95,7 +95,7 @@ func TestSchemaInitialization(t *testing.T) {
 		var version int
 		err := rawDB.QueryRow("PRAGMA user_version").Scan(&version)
 		assert.NoError(t, err)
-		assert.Equal(t, 9, version, "Schema version should be 5")
+		assert.Equal(t, 6, version, "Schema version should be 6")
 	})
 
 	// Verify deployments table exists