feat: rename a project safely with sqt rename-project

crates/sqltide-cli/src/argv/command_table.json

@@ -3373,6 +3373,162 @@
       ]
      }
     },
+    {
+     "name": "rename-project",
+     "parser": {
+      "prog": "sqt rename-project",
+      "description": "Rename the project while keeping its Kafka consumer groups, manifest history and project role grants. Dry-run by default; use --apply to rename.",
+      "groups": [
+       {
+        "title": "positional arguments"
+       },
+       {
+        "title": "options"
+       }
+      ],
+      "actions": [
+       {
+        "kind": "help",
+        "options": [
+         "-h",
+         "--help"
+        ],
+        "dest": "help",
+        "nargs": 0,
+        "default": {
+         "suppress": true
+        },
+        "help": "show this help message and exit",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [],
+        "dest": "new_name",
+        "default": null,
+        "required": true,
+        "help": "New project name",
+        "group": 0
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--project-dir"
+        ],
+        "dest": "project_dir",
+        "default": null,
+        "type": "Path",
+        "help": "Path to the project root containing pipelines/",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--host"
+        ],
+        "dest": "host",
+        "default": null,
+        "help": "ClickHouse host",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--port"
+        ],
+        "dest": "port",
+        "default": null,
+        "type": "int",
+        "help": "ClickHouse HTTP port",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--username"
+        ],
+        "dest": "username",
+        "default": null,
+        "help": "ClickHouse username",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--password"
+        ],
+        "dest": "password",
+        "default": null,
+        "help": "ClickHouse password",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--database"
+        ],
+        "dest": "database",
+        "default": null,
+        "help": "Target ClickHouse database",
+        "group": 1
+       },
+       {
+        "kind": "store_true",
+        "options": [
+         "--apply"
+        ],
+        "dest": "apply",
+        "nargs": 0,
+        "const": true,
+        "default": false,
+        "help": "Apply the rename (dry-run by default)",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--from"
+        ],
+        "dest": "from_name",
+        "default": null,
+        "help": "Previous project name; complete a rename on this target after sqltide_project.toml already carries the new name",
+        "metavar": "OLD_NAME",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--control-store-url"
+        ],
+        "dest": "control_store_url",
+        "default": null,
+        "help": "SQLite/PostgreSQL account control-store URL",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--target"
+        ],
+        "dest": "target",
+        "default": null,
+        "help": "Named project target to use",
+        "group": 1
+       },
+       {
+        "kind": "store",
+        "options": [
+         "--vars"
+        ],
+        "dest": "vars",
+        "default": {},
+        "type": "parse_cli_vars",
+        "help": "Project variable overrides as one JSON object",
+        "group": 1
+       }
+      ]
+     }
+    },
     {
      "name": "janitor",
      "parser": {
@@ -3929,6 +4085,10 @@
      "metavar": "reconcile",
      "help": "Reconcile live metadata baseline"
     },
+    {
+     "metavar": "rename-project",
+     "help": "Rename the project and keep its live state"
+    },
     {
      "metavar": "janitor",
      "help": "Clean up stale deployment artifacts"

crates/sqltide-clickhouse/src/metadata.rs

@@ -715,6 +715,38 @@ pub fn manifest_history_query(
     )
 }
 
+/// Copy one project target's manifest history from `from_identity` to `to_identity`, row for
+/// row, skipping manifests the new identity already holds, so applying it twice is a no-op.
+#[must_use]
+pub fn manifest_identity_copy(
+    database: &str,
+    from_identity: &str,
+    to_identity: &str,
+    target_name: &str,
+    target_database: &str,
+) -> String {
+    let table: &str = MANIFESTS_TABLE.0;
+    let target: String = format!(
+        "manifest.target_name = {} AND manifest.target_database = {}",
+        text_literal(target_name),
+        text_literal(target_database)
+    );
+    format!(
+        "INSERT INTO {database}.{table} (manifest_id, invocation_id, project_identity, \
+         target_name, target_database, is_production, project_revision, manifest_fingerprint, \
+         manifest_version, pipelines, resources, tool_version, published_at) SELECT \
+         manifest.manifest_id, manifest.invocation_id, {to}, manifest.target_name, \
+         manifest.target_database, manifest.is_production, manifest.project_revision, \
+         manifest.manifest_fingerprint, manifest.manifest_version, manifest.pipelines, \
+         manifest.resources, manifest.tool_version, manifest.published_at FROM \
+         {database}.{table} AS manifest WHERE manifest.project_identity = {from} AND {target} \
+         AND manifest.manifest_id NOT IN (SELECT manifest.manifest_id FROM {database}.{table} \
+         AS manifest WHERE manifest.project_identity = {to} AND {target});",
+        from = text_literal(from_identity),
+        to = text_literal(to_identity),
+    )
+}
+
 /// One append containing a complete immutable manifest.
 #[must_use]
 pub fn manifest_publication(database: &str, manifest: &Manifest) -> String {

crates/sqltide-clickhouse/src/realization.rs

@@ -49,6 +49,9 @@ pub struct ManagedSourceRequest<'a> {
     pub format: &'a str,
     pub consumer_group: Option<&'a str>,
     pub project_name: Option<&'a str>,
+    /// The project's `consumer_group_identity`, which replaces `project_name` in the default
+    /// consumer group when set.
+    pub consumer_group_identity: Option<&'a str>,
     pub target_name: Option<&'a str>,
     pub setting_keys: Vec<&'a str>,
 }
@@ -99,18 +102,11 @@ pub fn realize_managed_source(
     let kafka_table_name: String = format!("{KAFKA_TABLE_PREFIX}{}", request.logical_name);
     let consumer_group: String = match request.consumer_group {
         Some(group) if !group.is_empty() => group.to_owned(),
-        // The pre-rename group, so existing sources resume from their committed offsets.
-        _ => {
-            let segments: Vec<&str> = [
-                request.project_name,
-                request.target_name,
-                Some(request.logical_name),
-            ]
-            .into_iter()
-            .flatten()
-            .collect();
-            format!("streambuild_{}", segments.join("_"))
-        }
+        _ => default_consumer_group(
+            request.consumer_group_identity.or(request.project_name),
+            request.target_name,
+            request.logical_name,
+        ),
     };
     Ok(ManagedSourceRealization {
         raw_table_name: format!("{RAW_TABLE_PREFIX}{}", request.logical_name),
@@ -121,6 +117,22 @@ pub fn realize_managed_source(
     })
 }
 
+/// The consumer group a Kafka source uses when it declares none: the project identity (its
+/// `consumer_group_identity`, else its name), target and logical name. The `streambuild_`
+/// prefix is the pre-rename one, so existing sources resume from their committed offsets.
+#[must_use]
+pub fn default_consumer_group(
+    identity: Option<&str>,
+    target_name: Option<&str>,
+    logical_name: &str,
+) -> String {
+    let segments: Vec<&str> = [identity, target_name, Some(logical_name)]
+        .into_iter()
+        .flatten()
+        .collect();
+    format!("streambuild_{}", segments.join("_"))
+}
+
 fn landing_query(kafka_table_name: &str) -> String {
     format!(
         "SELECT\n    _key AS kafka_key,\n    message AS kafka_value,\n    _topic AS kafka_topic,\n    \
@@ -183,6 +195,7 @@ mod tests {
             format: "JSONAsString",
             consumer_group,
             project_name: Some("shop"),
+            consumer_group_identity: None,
             target_name: None,
             setting_keys: vec!["kafka_num_consumers"],
         }
@@ -204,6 +217,31 @@ mod tests {
         );
     }
 
+    #[test]
+    fn consumer_group_identity_replaces_the_project_name_in_the_default_group_only() {
+        let mut renamed = request(None);
+        renamed.project_name = Some("storefront");
+        renamed.target_name = Some("dev");
+        assert_eq!(
+            realize_managed_source(&renamed).unwrap().consumer_group,
+            "streambuild_storefront_dev_orders"
+        );
+        renamed.consumer_group_identity = Some("shop");
+        assert_eq!(
+            realize_managed_source(&renamed).unwrap().consumer_group,
+            "streambuild_shop_dev_orders"
+        );
+        renamed.consumer_group = Some("orders-reader");
+        assert_eq!(
+            realize_managed_source(&renamed).unwrap().consumer_group,
+            "orders-reader"
+        );
+        assert_eq!(
+            default_consumer_group(None, None, "orders"),
+            "streambuild_orders"
+        );
+    }
+
     #[test]
     fn rejects_unsupported_kinds_formats_and_typed_setting_overrides() {
         let mut bad = request(None);

crates/sqltide-config/src/lib.rs

@@ -7,6 +7,7 @@ pub mod errors;
 pub mod models;
 pub mod pipelines;
 pub mod project;
+pub mod project_rename;
 pub mod toml;
 pub mod yaml;