use regionId

Author: chronark

pkg/db/deployment_topology_list_by_versions.sql_generated.go

@@ -9,6 +9,6 @@ FROM `deployment_topology` dt
 INNER JOIN `deployments` d ON dt.deployment_id = d.id
 INNER JOIN `workspaces` w ON d.workspace_id = w.id
 INNER JOIN `regions` r ON dt.region_id = r.id
-WHERE r.name = sqlc.arg(region) AND dt.version > sqlc.arg(afterVersion)
+WHERE r.id = sqlc.arg(region_id) AND dt.version > sqlc.arg(afterVersion)
 ORDER BY dt.version ASC
 LIMIT ?;

pkg/db/instances_find_by_deployment_id.sql_generated.go

@@ -533,15 +533,14 @@ CREATE TABLE `deployment_topology` (
 	`pk` bigint unsigned AUTO_INCREMENT NOT NULL,
 	`workspace_id` varchar(64) NOT NULL,
 	`deployment_id` varchar(64) NOT NULL,
-	`region` varchar(64),
+	`region` varchar(255) DEFAULT 'DELETE ME',
 	`region_id` varchar(64) NOT NULL,
 	`desired_replicas` int NOT NULL,
 	`version` bigint unsigned NOT NULL,
 	`desired_status` enum('stopped','running') NOT NULL,
 	`created_at` bigint NOT NULL,
 	`updated_at` bigint,
-	CONSTRAINT `deployment_topology_pk` PRIMARY KEY(`pk`),
-	CONSTRAINT `unique_version_per_region` UNIQUE(`region_id`,`version`)
+	CONSTRAINT `deployment_topology_pk` PRIMARY KEY(`pk`)
 );
 
 CREATE TABLE `acme_users` (
@@ -605,7 +604,7 @@ CREATE TABLE `sentinels` (
 	`environment_id` varchar(255) NOT NULL,
 	`k8s_name` varchar(64) NOT NULL,
 	`k8s_address` varchar(255) NOT NULL,
-	`region` varchar(64),
+	`region` varchar(255) DEFAULT 'DELETE ME',
 	`region_id` varchar(255) NOT NULL DEFAULT 'TODO',
 	`image` varchar(255) NOT NULL,
 	`desired_state` enum('running','standby','archived') NOT NULL DEFAULT 'running',
@@ -620,8 +619,7 @@ CREATE TABLE `sentinels` (
 	CONSTRAINT `sentinels_pk` PRIMARY KEY(`pk`),
 	CONSTRAINT `sentinels_id_unique` UNIQUE(`id`),
 	CONSTRAINT `sentinels_k8s_name_unique` UNIQUE(`k8s_name`),
-	CONSTRAINT `sentinels_k8s_address_unique` UNIQUE(`k8s_address`),
-	CONSTRAINT `unique_version_per_region` UNIQUE(`region_id`,`version`)
+	CONSTRAINT `sentinels_k8s_address_unique` UNIQUE(`k8s_address`)
 );
 
 CREATE TABLE `instances` (
@@ -631,6 +629,7 @@ CREATE TABLE `instances` (
 	`workspace_id` varchar(255) NOT NULL,
 	`project_id` varchar(255) NOT NULL,
 	`app_id` varchar(64) NOT NULL,
+	`region` varchar(255) DEFAULT 'DELETE ME',
 	`region_id` varchar(64) NOT NULL DEFAULT 'TODO',
 	`k8s_name` varchar(255) NOT NULL,
 	`address` varchar(255) NOT NULL,
@@ -706,16 +705,14 @@ CREATE TABLE `cilium_network_policies` (
 	`deployment_id` varchar(128) NOT NULL,
 	`k8s_name` varchar(64) NOT NULL,
 	`k8s_namespace` varchar(255) NOT NULL,
-	`region` varchar(255),
+	`region` varchar(255) DEFAULT 'DELETE ME',
 	`region_id` varchar(64) NOT NULL DEFAULT 'TODO',
 	`policy` json NOT NULL,
 	`version` bigint unsigned NOT NULL,
 	`created_at` bigint NOT NULL,
 	`updated_at` bigint,
 	CONSTRAINT `cilium_network_policies_pk` PRIMARY KEY(`pk`),
-	CONSTRAINT `cilium_network_policies_id_unique` UNIQUE(`id`),
-	CONSTRAINT `one_deployment_per_region` UNIQUE(`deployment_id`,`region_id`,`k8s_name`),
-	CONSTRAINT `unique_version_per_region` UNIQUE(`region_id`,`version`)
+	CONSTRAINT `cilium_network_policies_id_unique` UNIQUE(`id`)
 );
 
 CREATE TABLE `clusters` (

pkg/db/instances_find_by_deployment_id_and_region_id.sql_generated.go

@@ -2,7 +2,6 @@ package cluster
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"connectrpc.com/connect"
@@ -37,9 +36,7 @@ func (s *Service) Heartbeat(ctx context.Context, req *connect.Request[ctrlv1.Hea
 	now := time.Now().UnixMilli()
 
 	err := db.Query.UpsertRegion(ctx, s.db.RW(), db.UpsertRegionParams{
-		// using a readable id here to make debugging significantly easier
-		// do not rely on this schema though. treat ids as opaque strings.
-		ID:       fmt.Sprintf("%s::%s", platform, regionName),
+		ID:       uid.New(uid.RegionPrefix),
 		Name:     regionName,
 		Platform: platform,
 	})

pkg/db/instances_find_by_pod_name.sql_generated.go

@@ -37,15 +37,28 @@ func (s *Service) WatchDeployments(
 		return err
 	}
 
-	region := req.Header().Get("X-Krane-Region")
+	regionName := req.Header().Get("X-Krane-Region")
 	platform := req.Header().Get("X-Krane-Platform")
 	if err := assert.All(
-		assert.NotEmpty(region, "region is required"),
+		assert.NotEmpty(regionName, "region is required"),
 		assert.NotEmpty(platform, "platform is required"),
 	); err != nil {
 		return connect.NewError(connect.CodeInvalidArgument, err)
 	}
 
+	logger.Info("starting WatchDeployments stream", "region_name", regionName, "platform", platform)
+
+	region, err := db.Query.FindRegionByNameAndPlatform(ctx, s.db.RO(), db.FindRegionByNameAndPlatformParams{
+		Name:     regionName,
+		Platform: platform,
+	})
+	if err != nil {
+		logger.Error("failed to find region for WatchDeployments", "error", err, "region_name", regionName, "platform", platform)
+		return connect.NewError(connect.CodeInternal, err)
+	}
+
+	logger.Info("found region for WatchDeployments", "region_id", region.ID)
+
 	versionCursor := req.Msg.GetVersionLastSeen()
 
 	for {
@@ -55,7 +68,7 @@ func (s *Service) WatchDeployments(
 		default:
 		}
 
-		states, err := s.fetchDeploymentStates(ctx, region, versionCursor)
+		states, err := s.fetchDeploymentStates(ctx, region.ID, versionCursor)
 		if err != nil {
 			logger.Error("failed to fetch deployment states", "error", err)
 			return connect.NewError(connect.CodeInternal, err)
@@ -79,9 +92,9 @@ func (s *Service) WatchDeployments(
 // fetchDeploymentStates queries the database for deployment topologies in the given region
 // with versions greater than afterVersion, returning up to 100 results. Rows that fail
 // conversion are logged and skipped rather than failing the entire batch.
-func (s *Service) fetchDeploymentStates(ctx context.Context, region string, afterVersion uint64) ([]*ctrlv1.DeploymentState, error) {
+func (s *Service) fetchDeploymentStates(ctx context.Context, regionID string, afterVersion uint64) ([]*ctrlv1.DeploymentState, error) {
 	rows, err := db.Query.ListDeploymentTopologyByRegion(ctx, s.db.RO(), db.ListDeploymentTopologyByRegionParams{
-		Region:       region,
+		RegionID:     regionID,
 		Afterversion: afterVersion,
 		Limit:        100,
 	})
@@ -99,6 +112,8 @@ func (s *Service) fetchDeploymentStates(ctx context.Context, region string, afte
 		states = append(states, state)
 	}
 
+	logger.Info("fetched deployment states", "count", len(states), "region_id", regionID, "after_version", afterVersion)
+
 	return states, nil
 }