diff --git a/crates/cli/src/config.rs b/crates/cli/src/config.rs
index abcac98fbaa..3ed0fd15acd 100644
--- a/crates/cli/src/config.rs
+++ b/crates/cli/src/config.rs
@@ -12,6 +12,7 @@ use toml_edit::ArrayOfTables;
 const DEFAULT_SERVER_KEY: &str = "default_server";
 const WEB_SESSION_TOKEN_KEY: &str = "web_session_token";
 const SPACETIMEDB_TOKEN_KEY: &str = "spacetimedb_token";
+const ROOT_DATABASE_NAMESPACE_KEY: &str = "root_database_namespace";
 const SERVER_CONFIGS_KEY: &str = "server_configs";
 const NICKNAME_KEY: &str = "nickname";
 const HOST_KEY: &str = "host";
@@ -124,6 +125,7 @@ pub struct RawConfig {
     // TODO: Move these IDs/tokens out of config so we're no longer storing sensitive tokens in a human-edited file.
     web_session_token: Option<String>,
     spacetimedb_token: Option<String>,
+    root_database_namespace: Option<String>,
 }
 
 #[derive(Debug, Clone)]
@@ -173,6 +175,7 @@ impl RawConfig {
             server_configs: vec![maincloud, local],
             web_session_token: None,
             spacetimedb_token: None,
+            root_database_namespace: None,
         }
     }
 
@@ -448,9 +451,14 @@ Fetch the server's fingerprint with:
         self.spacetimedb_token = Some(token);
     }
 
+    pub fn set_root_database_namespace(&mut self, namespace: Option<String>) {
+        self.root_database_namespace = namespace;
+    }
+
     pub fn clear_login_tokens(&mut self) {
         self.web_session_token = None;
         self.spacetimedb_token = None;
+        self.root_database_namespace = None;
     }
 }
 
@@ -461,6 +469,7 @@ impl TryFrom<&toml_edit::DocumentMut> for RawConfig {
         let default_server = read_opt_str(value, DEFAULT_SERVER_KEY)?;
         let web_session_token = read_opt_str(value, WEB_SESSION_TOKEN_KEY)?;
         let spacetimedb_token = read_opt_str(value, SPACETIMEDB_TOKEN_KEY)?;
+        let root_database_namespace = read_opt_str(value, ROOT_DATABASE_NAMESPACE_KEY)?;
 
         let mut server_configs = Vec::new();
         if let Some(arr) = read_table(value, SERVER_CONFIGS_KEY)? {
@@ -474,6 +483,7 @@ impl TryFrom<&toml_edit::DocumentMut> for RawConfig {
             server_configs,
             web_session_token,
             spacetimedb_token,
+            root_database_namespace,
         })
     }
 }
@@ -654,11 +664,13 @@ impl Config {
             server_configs: old_server_configs,
             web_session_token,
             spacetimedb_token,
+            root_database_namespace,
         } = &self.home;
 
         set_value(DEFAULT_SERVER_KEY, default_server.as_deref());
         set_value(WEB_SESSION_TOKEN_KEY, web_session_token.as_deref());
         set_value(SPACETIMEDB_TOKEN_KEY, spacetimedb_token.as_deref());
+        set_value(ROOT_DATABASE_NAMESPACE_KEY, root_database_namespace.as_deref());
 
         // Short-circuit if there are no servers.
         if old_server_configs.is_empty() {
@@ -808,6 +820,10 @@ Update the server's fingerprint with:
         self.home.set_spacetimedb_token(token);
     }
 
+    pub fn set_root_database_namespace(&mut self, namespace: Option<String>) {
+        self.home.set_root_database_namespace(namespace);
+    }
+
     pub fn clear_login_tokens(&mut self) {
         self.home.clear_login_tokens();
     }
@@ -819,6 +835,13 @@ Update the server's fingerprint with:
     pub fn spacetimedb_token(&self) -> Option<&String> {
         self.home.spacetimedb_token.as_ref()
     }
+
+    pub fn root_database_namespace(&self) -> Option<&str> {
+        self.home
+            .root_database_namespace
+            .as_deref()
+            .and_then(|value| (!value.is_empty()).then_some(value))
+    }
 }
 
 /// Update the value of a key in a `TOML` document, preserving the formatting and comments of the original value.
diff --git a/crates/cli/src/subcommands/dns.rs b/crates/cli/src/subcommands/dns.rs
index 59e049b038b..acdd386d268 100644
--- a/crates/cli/src/subcommands/dns.rs
+++ b/crates/cli/src/subcommands/dns.rs
@@ -1,10 +1,12 @@
 use crate::common_args;
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, get_auth_header, ResponseExt};
+use crate::util::{add_auth_header_opt, get_auth_header, prepend_root_database_namespace, ResponseExt};
+use anyhow::ensure;
 use clap::ArgMatches;
 use clap::{Arg, Command};
+use reqwest::StatusCode;
 
-use spacetimedb_client_api_messages::name::{DomainName, SetDomainsResult};
+use spacetimedb_client_api_messages::name::{parse_database_name, parse_domain_name, DomainName, SetDomainsResult};
 
 pub fn cli() -> Command {
     Command::new("rename")
@@ -26,21 +28,22 @@ pub fn cli() -> Command {
 }
 
 pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::Error> {
-    let domain = args.get_one::<String>("new-name").unwrap();
+    let requested_name = args.get_one::<String>("new-name").unwrap();
     let database_identity = args.get_one::<String>("database-identity").unwrap();
     let server = args.get_one::<String>("server").map(|s| s.as_ref());
     let force = args.get_flag("force");
     let auth_header = get_auth_header(&mut config, false, server, !force).await?;
+    let root_database_namespace = config.root_database_namespace();
+    let host_url = config.get_host_url(server)?;
+    let client = reqwest::Client::new();
 
-    let domain: DomainName = domain.parse()?;
+    let current_default_name = get_default_name(&client, &host_url, database_identity, &auth_header).await?;
+    let name = resolve_name_for_rename(requested_name, current_default_name.as_ref(), root_database_namespace)?;
 
-    let builder = reqwest::Client::new()
-        .put(format!(
-            "{}/v1/database/{database_identity}/names",
-            config.get_host_url(server)?
-        ))
-        .header(reqwest::header::CONTENT_TYPE, "application/json")
-        .body(serde_json::to_string(&[&domain])?);
+    let builder = client
+        .post(format!("{host_url}/v1/database/default-name/{database_identity}"))
+        .header(reqwest::header::CONTENT_TYPE, "text/plain; charset=utf-8")
+        .body(name.to_string());
     let builder = add_auth_header_opt(builder, &auth_header);
 
     let response = builder.send().await?;
@@ -50,9 +53,9 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     if !status.is_success() {
         anyhow::bail!(match result {
             SetDomainsResult::Success => "".to_string(),
-            SetDomainsResult::PermissionDenied { domain } => format!("Permission denied for domain: {domain}"),
+            SetDomainsResult::PermissionDenied { domain } => format!("Permission denied for database name: {domain}"),
             SetDomainsResult::PermissionDeniedOnAny { domains } =>
-                format!("Permission denied for domains: {domains:?}"),
+                format!("Permission denied for database names: {domains:?}"),
             SetDomainsResult::DatabaseNotFound => format!("Database {database_identity} not found"),
             SetDomainsResult::NotYourDatabase { .. } =>
                 format!("You cannot rename {database_identity} because it is owned by another identity."),
@@ -60,7 +63,51 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
         });
     }
 
-    println!("Name set to {domain} for identity {database_identity}.");
+    println!("Name set to {name} for identity {database_identity}.");
 
     Ok(())
 }
+
+async fn get_default_name(
+    client: &reqwest::Client,
+    host_url: &str,
+    database_identity: &str,
+    auth_header: &crate::util::AuthHeader,
+) -> Result<Option<DomainName>, anyhow::Error> {
+    let builder = client.get(format!("{host_url}/v1/database/default-name/{database_identity}"));
+    let builder = add_auth_header_opt(builder, auth_header);
+    let response = builder.send().await?;
+    if response.status() == StatusCode::NOT_FOUND {
+        return Ok(None);
+    }
+
+    response.json_or_error().await.map(Some)
+}
+
+fn resolve_name_for_rename(
+    requested_name: &str,
+    current_default_name: Option<&DomainName>,
+    root_database_namespace: Option<&str>,
+) -> Result<DomainName, anyhow::Error> {
+    if let Some(current_default_name) = current_default_name {
+        if current_default_name
+            .sub_domain()
+            .is_some_and(|sub_domain| sub_domain.contains('/'))
+        {
+            ensure!(
+                !requested_name.contains('/'),
+                "Child database rename target cannot contain `/`"
+            );
+            parse_database_name(requested_name)?;
+            let parent = current_default_name
+                .as_ref()
+                .rsplit_once('/')
+                .map(|(parent, _)| parent)
+                .ok_or_else(|| anyhow::anyhow!("Failed to determine parent database path"))?;
+            return parse_domain_name(format!("{parent}/{requested_name}")).map_err(Into::into);
+        }
+    }
+
+    let qualified = prepend_root_database_namespace(requested_name, root_database_namespace);
+    parse_domain_name(qualified).map_err(Into::into)
+}
diff --git a/crates/cli/src/subcommands/list.rs b/crates/cli/src/subcommands/list.rs
index d650e98f52d..19fefca74a9 100644
--- a/crates/cli/src/subcommands/list.rs
+++ b/crates/cli/src/subcommands/list.rs
@@ -24,13 +24,19 @@ pub fn cli() -> Command {
 
 #[derive(Deserialize)]
 struct DatabasesResult {
-    pub identities: Vec<IdentityRow>,
+    pub identities: Vec<IdentityOnlyRow>,
 }
 
-#[derive(Tabled, Deserialize)]
+#[derive(Deserialize)]
 #[serde(transparent)]
+struct IdentityOnlyRow {
+    pub db_identity: Identity,
+}
+
+#[derive(Tabled)]
 struct IdentityRow {
     pub db_identity: Identity,
+    pub default_name: String,
 }
 
 pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::Error> {
@@ -58,11 +64,26 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
         .context("unable to retrieve databases for identity")?;
 
     if !result.identities.is_empty() {
-        let mut table = Table::new(result.identities);
+        let mut rows = Vec::with_capacity(result.identities.len());
+        for row in result.identities {
+            let default_name = util::spacetime_reverse_dns(&config, &row.db_identity.to_string(), server)
+                .await
+                .with_context(|| format!("unable to retrieve database names for {}", row.db_identity))?
+                .names
+                .first()
+                .map(ToString::to_string)
+                .unwrap_or_else(|| "<unnamed>".to_owned());
+            rows.push(IdentityRow {
+                db_identity: row.db_identity,
+                default_name,
+            });
+        }
+
+        let mut table = Table::new(rows);
         table
             .with(Style::psql())
             .with(Modify::new(Columns::first()).with(Alignment::left()));
-        println!("Associated database identities for {identity}:\n");
+        println!("Associated databases for {identity}:\n");
         println!("{table}");
     } else {
         println!("No databases found for {identity}.");
diff --git a/crates/cli/src/subcommands/login.rs b/crates/cli/src/subcommands/login.rs
index ef696d06a81..f9caa73b4c4 100644
--- a/crates/cli/src/subcommands/login.rs
+++ b/crates/cli/src/subcommands/login.rs
@@ -1,4 +1,4 @@
-use crate::util::decode_identity;
+use crate::util::{decode_identity, decode_root_database_namespace};
 use crate::Config;
 use clap::{Arg, ArgAction, ArgGroup, ArgMatches, Command};
 use reqwest::Url;
@@ -64,6 +64,7 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
 
     if let Some(token) = spacetimedb_token {
         config.set_spacetimedb_token(token.clone());
+        config.set_root_database_namespace(Some(String::new()));
         config.save();
         return Ok(());
     }
@@ -138,6 +139,18 @@ pub async fn spacetimedb_login_force(
         spacetimedb_login(host, &session_token).await?
     };
     config.set_spacetimedb_token(token.clone());
+    let root_database_namespace = if direct_login {
+        Some(String::new())
+    } else {
+        match decode_root_database_namespace(&token) {
+            Ok(namespace) => Some(namespace.unwrap_or_default()),
+            Err(err) => {
+                eprintln!("WARNING: failed to extract root database namespace from token: {err}");
+                Some(String::new())
+            }
+        }
+    };
+    config.set_root_database_namespace(root_database_namespace);
     config.save();
 
     Ok(token)
diff --git a/crates/cli/src/subcommands/publish.rs b/crates/cli/src/subcommands/publish.rs
index 4c8a77f2fcb..1bf89d311d5 100644
--- a/crates/cli/src/subcommands/publish.rs
+++ b/crates/cli/src/subcommands/publish.rs
@@ -1,16 +1,16 @@
-use anyhow::{ensure, Context};
+use anyhow::Context;
 use clap::Arg;
 use clap::ArgAction::{Set, SetTrue};
 use clap::ArgMatches;
 use reqwest::{StatusCode, Url};
-use spacetimedb_client_api_messages::name::{is_identity, parse_database_name, PublishResult};
-use spacetimedb_client_api_messages::name::{DatabaseNameError, PrePublishResult, PrettyPrintStyle, PublishOp};
+use spacetimedb_client_api_messages::name::{is_identity, parse_domain_name, PublishResult};
+use spacetimedb_client_api_messages::name::{PrePublishResult, PrettyPrintStyle, PublishOp};
 use std::path::PathBuf;
 use std::{env, fs};
 
 use crate::common_args::ClearMode;
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, get_auth_header, AuthHeader, ResponseExt};
+use crate::util::{add_auth_header_opt, get_auth_header, prepend_root_database_namespace, AuthHeader, ResponseExt};
 use crate::util::{decode_identity, y_or_n};
 use crate::{build, common_args};
 
@@ -75,10 +75,10 @@ pub fn cli() -> clap::Command {
         )
         .arg(
             Arg::new("parent")
-            .help("Domain or identity of a parent for this database")
+            .help("Database name or identity of a parent for this database")
             .long("parent")
             .long_help(
-"A valid domain or identity of an existing database that should be the parent of this database.
+"A valid database name or identity of an existing database that should be the parent of this database.
 
 If a parent is given, the new database inherits the team permissions from the parent.
 A parent can only be set when a database is created, not when it is updated."
@@ -98,12 +98,12 @@ An organization can only be set when a database is created, not when it is updat
         )
         .arg(
             Arg::new("name|identity")
-                .help("A valid domain or identity for this database")
+                .help("A valid database name or identity for this database")
                 .long_help(
-"A valid domain or identity for this database.
+"A valid database name or identity for this database.
 
-Database names must match the regex `/^[a-z0-9]+(-[a-z0-9]+)*$/`,
-i.e. only lowercase ASCII letters and numbers, separated by dashes."),
+Database names may include a root namespace and child path segments,
+for example: `@user/my-db` or `@user/game/region-1`."),
         )
         .arg(common_args::server()
                 .help("The nickname, domain name or URL of the server to host the database."),
@@ -180,8 +180,12 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     //  easily create a new identity with an email
     let auth_header = get_auth_header(&mut config, anon_identity, server, !force).await?;
 
-    let (name_or_identity, parent) =
-        validate_name_and_parent(name_or_identity.map(String::as_str), parent.map(String::as_str))?;
+    let root_database_namespace = config.root_database_namespace();
+    let (name_or_identity, parent) = normalize_name_and_parent(
+        name_or_identity.map(String::as_str),
+        parent.map(String::as_str),
+        root_database_namespace,
+    )?;
 
     if !path_to_project.exists() {
         return Err(anyhow::anyhow!(
@@ -224,7 +228,7 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     let client = reqwest::Client::new();
     // If a name was given, ensure to percent-encode it.
     // We also use PUT with a name or identity, and POST otherwise.
-    let mut builder = if let Some(name_or_identity) = name_or_identity {
+    let mut builder = if let Some(name_or_identity) = name_or_identity.as_deref() {
         let encode_set = const { &percent_encoding::NON_ALPHANUMERIC.remove(b'_').remove(b'-') };
         let domain = percent_encoding::percent_encode(name_or_identity.as_bytes(), encode_set);
         let mut builder = client.put(format!("{database_host}/v1/database/{domain}"));
@@ -293,6 +297,7 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
             };
             if let Some(domain) = domain {
                 println!("{op} database with name: {domain}, identity: {database_identity}");
+                println!("Connection database name: {domain}");
             } else {
                 println!("{op} database with identity: {database_identity}");
             }
@@ -304,13 +309,14 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
             // If we're not in the `anon_identity` case, then we have already forced the user to log in above (using `get_auth_header`), so this should be safe to unwrap.
             let token = config.spacetimedb_token().unwrap();
             let identity = decode_identity(token)?;
-            //TODO(jdetter): Have a nice name generator here, instead of using some abstract characters
-            // we should perhaps generate fun names like 'green-fire-dragon' instead
-            let suggested_tld: String = identity.chars().take(12).collect();
+            let suggested_namespace = config
+                .root_database_namespace()
+                .map(str::to_owned)
+                .unwrap_or_else(|| format!("@{}", identity.chars().take(12).collect::<String>()));
             return Err(anyhow::anyhow!(
-                "The database {name} is not registered to the identity you provided.\n\
-                We suggest you push to either a domain owned by you, or a new domain like:\n\
-                \tspacetime publish {suggested_tld}\n",
+                "The database name {name} is not registered to the identity you provided.\n\
+                Publish under a root namespace that you own, for example:\n\
+                \tspacetime publish {suggested_namespace}/my-database\n",
             ));
         }
     }
@@ -318,11 +324,13 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     Ok(())
 }
 
-fn validate_name_or_identity(name_or_identity: &str) -> Result<(), DatabaseNameError> {
+fn validate_name_or_identity(name_or_identity: &str) -> anyhow::Result<()> {
     if is_identity(name_or_identity) {
         Ok(())
     } else {
-        parse_database_name(name_or_identity).map(drop)
+        parse_domain_name(name_or_identity)
+            .map(drop)
+            .map_err(anyhow::Error::from)
     }
 }
 
@@ -330,33 +338,39 @@ fn invalid_parent_name(name: &str) -> String {
     format!("invalid parent database name `{name}`")
 }
 
-fn validate_name_and_parent<'a>(
-    name: Option<&'a str>,
-    parent: Option<&'a str>,
-) -> anyhow::Result<(Option<&'a str>, Option<&'a str>)> {
-    if let Some(parent) = parent.as_ref() {
-        validate_name_or_identity(parent).with_context(|| invalid_parent_name(parent))?;
+fn normalize_name_and_parent(
+    name: Option<&str>,
+    parent: Option<&str>,
+    root_database_namespace: Option<&str>,
+) -> anyhow::Result<(Option<String>, Option<String>)> {
+    let mut name = name.map(str::to_owned);
+    let mut parent = parent.map(str::to_owned);
+
+    if let Some(parent_name) = parent.as_deref() {
+        validate_name_or_identity(parent_name).with_context(|| invalid_parent_name(parent_name))?;
     }
 
-    match name {
-        Some(name) => match name.split_once('/') {
-            Some((parent_alt, child)) => {
-                ensure!(
-                    parent.is_none() || parent.is_some_and(|parent| parent == parent_alt),
-                    "cannot specify both --parent and <parent>/<child>"
-                );
-                validate_name_or_identity(parent_alt).with_context(|| invalid_parent_name(parent_alt))?;
-                validate_name_or_identity(child)?;
-
-                Ok((Some(child), Some(parent_alt)))
-            }
-            None => {
-                validate_name_or_identity(name)?;
-                Ok((Some(name), parent))
-            }
-        },
-        None => Ok((None, parent)),
+    if parent.is_some()
+        && name
+            .as_deref()
+            .is_some_and(|raw_name| !is_identity(raw_name) && raw_name.contains('/'))
+    {
+        anyhow::bail!("child database name cannot contain `/` when --parent is set");
+    }
+
+    if let Some(parent_name) = parent.as_mut() {
+        *parent_name = prepend_root_database_namespace(parent_name, root_database_namespace);
+        validate_name_or_identity(parent_name).with_context(|| invalid_parent_name(parent_name))?;
+    }
+
+    if let Some(name_or_identity) = name.as_mut() {
+        if parent.is_none() {
+            *name_or_identity = prepend_root_database_namespace(name_or_identity, root_database_namespace);
+        }
+        validate_name_or_identity(name_or_identity)?;
     }
+
+    Ok((name, parent))
 }
 
 /// Determine the pretty print style based on the NO_COLOR environment variable.
@@ -487,47 +501,56 @@ mod tests {
 
     #[test]
     fn validate_none_arguments_returns_none_values() {
-        assert_matches!(validate_name_and_parent(None, None), Ok((None, None)));
-        assert_matches!(validate_name_and_parent(Some("foo"), None), Ok((Some(_), None)));
-        assert_matches!(validate_name_and_parent(None, Some("foo")), Ok((None, Some(_))));
+        assert_matches!(normalize_name_and_parent(None, None, None), Ok((None, None)));
+        assert_matches!(normalize_name_and_parent(Some("foo"), None, None), Ok((Some(_), None)));
+        assert_matches!(normalize_name_and_parent(None, Some("foo"), None), Ok((None, Some(_))));
     }
 
     #[test]
     fn validate_valid_arguments_returns_arguments() {
         let name = "child";
         let parent = "parent";
-        let result = (Some(name), Some(parent));
+        let result = (Some(name.to_owned()), Some(parent.to_owned()));
         assert_matches!(
-            validate_name_and_parent(Some(name), Some(parent)),
+            normalize_name_and_parent(Some(name), Some(parent), None),
             Ok(val) if val == result
         );
     }
 
     #[test]
-    fn validate_parent_and_path_name_returns_error_unless_parent_equal() {
+    fn validate_path_name_is_rejected_when_parent_is_set() {
+        assert_matches!(
+            normalize_name_and_parent(Some("parent/child"), Some("parent"), None),
+            Err(_)
+        );
         assert_matches!(
-            validate_name_and_parent(Some("parent/child"), Some("parent")),
-            Ok((Some("child"), Some("parent")))
+            normalize_name_and_parent(Some("parent/child"), Some("cousin"), None),
+            Err(_)
         );
-        assert_matches!(validate_name_and_parent(Some("parent/child"), Some("cousin")), Err(_));
     }
 
     #[test]
-    fn validate_more_than_two_path_segments_are_an_error() {
-        assert_matches!(validate_name_and_parent(Some("proc/net/tcp"), None), Err(_));
-        assert_matches!(validate_name_and_parent(Some("proc//net"), None), Err(_));
+    fn validate_more_than_two_path_segments_are_supported() {
+        assert_matches!(
+            normalize_name_and_parent(Some("proc/net/tcp"), None, None),
+            Ok((Some(name), None)) if name == "proc/net/tcp"
+        );
+        assert_matches!(normalize_name_and_parent(Some("proc//net"), None, None), Err(_));
     }
 
     #[test]
     fn validate_trailing_slash_is_an_error() {
-        assert_matches!(validate_name_and_parent(Some("foo//"), None), Err(_));
-        assert_matches!(validate_name_and_parent(Some("foo/bar/"), None), Err(_));
+        assert_matches!(normalize_name_and_parent(Some("foo//"), None, None), Err(_));
+        assert_matches!(normalize_name_and_parent(Some("foo/bar/"), None, None), Err(_));
     }
 
     #[test]
-    fn validate_parent_cant_have_slash() {
-        assert_matches!(validate_name_and_parent(Some("child"), Some("par/ent")), Err(_));
-        assert_matches!(validate_name_and_parent(Some("child"), Some("parent/")), Err(_));
+    fn validate_parent_can_have_path_segments() {
+        assert_matches!(
+            normalize_name_and_parent(Some("child"), Some("par/ent"), None),
+            Ok((Some(name), Some(parent))) if name == "child" && parent == "par/ent"
+        );
+        assert_matches!(normalize_name_and_parent(Some("child"), Some("parent/"), None), Err(_));
     }
 
     #[test]
@@ -536,8 +559,37 @@ mod tests {
         let child = Identity::ONE.to_string();
 
         assert_matches!(
-            validate_name_and_parent(Some(&child), Some(&parent)),
-            Ok(res) if res == (Some(&child), Some(&parent))
+            normalize_name_and_parent(Some(&child), Some(&parent), None),
+            Ok((Some(name), Some(parent_name))) if name == child && parent_name == parent
+        );
+    }
+
+    #[test]
+    fn prepend_root_namespace_to_unqualified_name_and_parent() {
+        assert_matches!(
+            normalize_name_and_parent(Some("my-db"), None, Some("@alice")),
+            Ok((Some(name), None)) if name == "@alice/my-db"
+        );
+        assert_matches!(
+            normalize_name_and_parent(Some("child"), Some("parent/leaf"), Some("@alice")),
+            Ok((Some(name), Some(parent))) if name == "child" && parent == "@alice/parent/leaf"
+        );
+        assert_matches!(
+            normalize_name_and_parent(Some("parent/leaf/child"), None, Some("@alice")),
+            Ok((Some(name), None)) if name == "@alice/parent/leaf/child"
+        );
+    }
+
+    #[test]
+    fn dont_prepend_root_namespace_when_already_qualified_or_identity() {
+        let identity = Identity::ZERO.to_string();
+        assert_matches!(
+            normalize_name_and_parent(Some("@bob/my-db"), None, Some("@alice")),
+            Ok((Some(name), None)) if name == "@bob/my-db"
+        );
+        assert_matches!(
+            normalize_name_and_parent(Some(&identity), None, Some("@alice")),
+            Ok((Some(name), None)) if name == identity
         );
     }
 }
diff --git a/crates/cli/src/util.rs b/crates/cli/src/util.rs
index fd5de755629..a8888912504 100644
--- a/crates/cli/src/util.rs
+++ b/crates/cli/src/util.rs
@@ -2,7 +2,7 @@ use anyhow::Context;
 use base64::{engine::general_purpose::STANDARD_NO_PAD as BASE_64_STD_NO_PAD, Engine as _};
 use reqwest::{RequestBuilder, Url};
 use spacetimedb_auth::identity::{IncomingClaims, SpacetimeIdentityClaims};
-use spacetimedb_client_api_messages::name::GetNamesResponse;
+use spacetimedb_client_api_messages::name::{is_identity, GetNamesResponse};
 use spacetimedb_lib::Identity;
 use std::io::Write;
 use std::path::{Path, PathBuf};
@@ -113,7 +113,7 @@ pub async fn spacetime_dns(
     server: Option<&str>,
 ) -> Result<Option<Identity>, anyhow::Error> {
     let client = reqwest::Client::new();
-    let url = format!("{}/v1/database/{}/identity", config.get_host_url(server)?, domain);
+    let url = format!("{}/v1/database/identity/{}", config.get_host_url(server)?, domain);
     let Some(res) = client.get(url).send().await?.found() else {
         return Ok(None);
     };
@@ -137,7 +137,7 @@ pub async fn spacetime_reverse_dns(
     server: Option<&str>,
 ) -> Result<GetNamesResponse, anyhow::Error> {
     let client = reqwest::Client::new();
-    let url = format!("{}/v1/database/{}/names", config.get_host_url(server)?, identity);
+    let url = format!("{}/v1/database/names/{}", config.get_host_url(server)?, identity);
     client.get(url).send().await?.json_or_error().await
 }
 
@@ -286,6 +286,11 @@ pub fn y_or_n(force: bool, prompt: &str) -> anyhow::Result<bool> {
 }
 
 pub fn decode_identity(token: &String) -> anyhow::Result<String> {
+    let claims_data = decode_claims(token)?;
+    Ok(claims_data.identity.to_string())
+}
+
+fn decode_claims(token: &str) -> anyhow::Result<SpacetimeIdentityClaims> {
     // Here, we manually extract and decode the claims from the json web token.
     // We do this without using the `jsonwebtoken` crate because it doesn't seem to have a way to skip signature verification.
     // But signature verification would require getting the public key from a server, and we don't necessarily want to do that.
@@ -297,9 +302,70 @@ pub fn decode_identity(token: &String) -> anyhow::Result<String> {
     let decoded_string = String::from_utf8(decoded_bytes)?;
 
     let claims_data: IncomingClaims = serde_json::from_str(decoded_string.as_str())?;
-    let claims_data: SpacetimeIdentityClaims = claims_data.try_into()?;
+    claims_data.try_into()
+}
 
-    Ok(claims_data.identity.to_string())
+pub fn decode_root_database_namespace(token: &str) -> anyhow::Result<Option<String>> {
+    let claims = decode_claims(token)?;
+    let Some(extra) = claims.extra else {
+        return Ok(None);
+    };
+
+    let find_extra_claim = |keys: &[&str]| {
+        keys.iter().find_map(|key| {
+            extra
+                .iter()
+                .find_map(|(claim_key, value)| (claim_key.as_ref() == *key).then_some(value))
+                .and_then(serde_json::Value::as_str)
+        })
+    };
+
+    let root = find_extra_claim(&["root_database_namespace", "rootDatabaseNamespace", "root_namespace"])
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_owned);
+    if root.is_some() {
+        return Ok(root);
+    }
+
+    let username = find_extra_claim(&["username", "user_name", "preferred_username", "login"])
+        .map(str::trim)
+        .filter(|value| !value.is_empty());
+    Ok(username.map(|username| {
+        if username.starts_with('@') {
+            username.to_owned()
+        } else {
+            format!("@{username}")
+        }
+    }))
+}
+
+pub fn prepend_root_database_namespace(name: &str, root_namespace: Option<&str>) -> String {
+    if is_identity(name) {
+        return name.to_owned();
+    }
+
+    let Some(root_namespace) = root_namespace else {
+        return name.to_owned();
+    };
+    let root_namespace = root_namespace.trim().trim_end_matches('/');
+    if root_namespace.is_empty() {
+        return name.to_owned();
+    }
+
+    if name == root_namespace
+        || name
+            .strip_prefix(root_namespace)
+            .is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
+    {
+        return name.to_owned();
+    }
+
+    if name.split('/').next().is_some_and(|first| first.starts_with('@')) {
+        return name.to_owned();
+    }
+
+    format!("{root_namespace}/{name}")
 }
 
 pub async fn get_login_token_or_log_in(
diff --git a/crates/client-api-messages/src/name.rs b/crates/client-api-messages/src/name.rs
index b835078ac4d..2f221fdc33f 100644
--- a/crates/client-api-messages/src/name.rs
+++ b/crates/client-api-messages/src/name.rs
@@ -86,7 +86,7 @@ pub enum PublishResult {
         /// In other words, this echoes back a domain name if one was given. If
         /// the database name given was in fact a database identity, this will be
         /// `None`.
-        domain: Option<DatabaseName>,
+        domain: Option<DomainName>,
         /// The identity of the published database.
         ///
         /// Always set, regardless of whether publish resolved a domain name first
@@ -103,7 +103,7 @@ pub enum PublishResult {
     /// If you were trying to insert this database name, but the tld `clockworklabs` is
     /// owned by an identity other than the identity that you provided, then you will receive
     /// this error.
-    PermissionDenied { name: DatabaseName },
+    PermissionDenied { name: DomainName },
 }
 
 #[derive(serde::Serialize, serde::Deserialize, Debug, Default)]
@@ -589,7 +589,7 @@ mod serde_impls {
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct GetNamesResponse {
-    pub names: Vec<DatabaseName>,
+    pub names: Vec<DomainName>,
 }
 
 /// Returns whether a hex string is a valid identity.
diff --git a/crates/client-api/src/lib.rs b/crates/client-api/src/lib.rs
index 01b78da9b00..527ab969a9c 100644
--- a/crates/client-api/src/lib.rs
+++ b/crates/client-api/src/lib.rs
@@ -265,7 +265,9 @@ pub trait ControlStateReadAccess {
     // DNS
     async fn lookup_database_identity(&self, domain: &str) -> anyhow::Result<Option<Identity>>;
     async fn reverse_lookup(&self, database_identity: &Identity) -> anyhow::Result<Vec<DomainName>>;
+    async fn lookup_database_default_name(&self, database_identity: &Identity) -> anyhow::Result<Option<DomainName>>;
     async fn lookup_namespace_owner(&self, name: &str) -> anyhow::Result<Option<Identity>>;
+    async fn allow_register_tld_on_publish(&self) -> anyhow::Result<bool>;
 }
 
 /// Write operations on the SpacetimeDB control plane.
@@ -374,9 +376,17 @@ impl<T: ControlStateReadAccess + Send + Sync + Sync + ?Sized> ControlStateReadAc
         (**self).reverse_lookup(database_identity).await
     }
 
+    async fn lookup_database_default_name(&self, database_identity: &Identity) -> anyhow::Result<Option<DomainName>> {
+        (**self).lookup_database_default_name(database_identity).await
+    }
+
     async fn lookup_namespace_owner(&self, name: &str) -> anyhow::Result<Option<Identity>> {
         (**self).lookup_namespace_owner(name).await
     }
+
+    async fn allow_register_tld_on_publish(&self) -> anyhow::Result<bool> {
+        (**self).allow_register_tld_on_publish().await
+    }
 }
 
 #[async_trait]
diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs
index eab3e0a9262..17056cbfc13 100644
--- a/crates/client-api/src/routes/database.rs
+++ b/crates/client-api/src/routes/database.rs
@@ -1,6 +1,5 @@
 use std::borrow::Cow;
 use std::num::NonZeroU8;
-use std::str::FromStr;
 use std::time::Duration;
 use std::{env, io};
 
@@ -20,6 +19,7 @@ use axum::response::{ErrorResponse, IntoResponse};
 use axum::routing::MethodRouter;
 use axum::Extension;
 use axum_extra::TypedHeader;
+use bytestring::ByteString;
 use futures::TryStreamExt;
 use http::StatusCode;
 use log::{info, warn};
@@ -34,7 +34,7 @@ use spacetimedb::identity::Identity;
 use spacetimedb::messages::control_db::{Database, HostType};
 use spacetimedb_client_api_messages::http::SqlStmtResult;
 use spacetimedb_client_api_messages::name::{
-    self, DatabaseName, DomainName, MigrationPolicy, PrePublishAutoMigrateResult, PrePublishManualMigrateResult,
+    self, parse_domain_name, DomainName, MigrationPolicy, PrePublishAutoMigrateResult, PrePublishManualMigrateResult,
     PrePublishResult, PrettyPrintStyle, PublishOp, PublishResult,
 };
 use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9;
@@ -83,6 +83,11 @@ pub struct CallParams {
     reducer: String,
 }
 
+#[derive(Deserialize)]
+pub struct CallTailParams {
+    name_or_identity_and_reducer: String,
+}
+
 pub const NO_SUCH_DATABASE: (StatusCode, &str) = (StatusCode::NOT_FOUND, "No such database.");
 const MISDIRECTED: (StatusCode, &str) = (StatusCode::NOT_FOUND, "Database is not scheduled on this host");
 
@@ -135,6 +140,56 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
     }): Path<CallParams>,
     TypedHeader(content_type): TypedHeader<headers::ContentType>,
     ByteStringBody(body): ByteStringBody,
+) -> axum::response::Result<impl IntoResponse> {
+    call_inner(worker_ctx, auth, name_or_identity, reducer, content_type, body).await
+}
+
+/// Call a reducer/procedure where the path uses the form:
+/// `/database/call/*name_or_identity_and_reducer`.
+pub async fn call_with_tail<S: ControlStateDelegate + NodeDelegate>(
+    State(worker_ctx): State<S>,
+    Extension(auth): Extension<SpacetimeAuth>,
+    Path(CallTailParams {
+        name_or_identity_and_reducer,
+    }): Path<CallTailParams>,
+    TypedHeader(content_type): TypedHeader<headers::ContentType>,
+    ByteStringBody(body): ByteStringBody,
+) -> axum::response::Result<impl IntoResponse> {
+    let Some((raw_name_or_identity, reducer)) = name_or_identity_and_reducer.rsplit_once('/') else {
+        return Err((StatusCode::BAD_REQUEST, "missing reducer name in route").into());
+    };
+    if reducer.is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "missing reducer name in route").into());
+    }
+
+    let name_or_identity =
+        parse_name_or_identity(raw_name_or_identity).map_err(|(status, msg)| ErrorResponse::from((status, msg)))?;
+    call_inner(
+        worker_ctx,
+        auth,
+        name_or_identity,
+        reducer.to_string(),
+        content_type,
+        body,
+    )
+    .await
+}
+
+fn parse_name_or_identity(raw: &str) -> Result<NameOrIdentity, (StatusCode, String)> {
+    if let Ok(identity) = Identity::from_hex(raw) {
+        return Ok(NameOrIdentity::Identity(identity.into()));
+    }
+    let name = parse_domain_name(raw).map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))?;
+    Ok(NameOrIdentity::Name(name))
+}
+
+async fn call_inner<S: ControlStateDelegate + NodeDelegate>(
+    worker_ctx: S,
+    auth: SpacetimeAuth,
+    name_or_identity: NameOrIdentity,
+    reducer: String,
+    content_type: headers::ContentType,
+    body: ByteString,
 ) -> axum::response::Result<impl IntoResponse> {
     assert_content_type_json(content_type)?;
 
@@ -568,19 +623,87 @@ pub async fn get_names<S: ControlStateDelegate>(
     Path(ReverseDNSParams { name_or_identity }): Path<ReverseDNSParams>,
 ) -> axum::response::Result<impl IntoResponse> {
     let database_identity = name_or_identity.resolve(&ctx).await?;
-
-    let names = ctx
-        .reverse_lookup(&database_identity)
+    let mut names = ctx.reverse_lookup(&database_identity).await.map_err(log_and_500)?;
+    if let Some(default_name) = ctx
+        .lookup_database_default_name(&database_identity)
         .await
         .map_err(log_and_500)?
-        .into_iter()
-        .filter_map(|x| String::from(x).try_into().ok())
-        .collect();
+    {
+        if let Some(ix) = names.iter().position(|name| name == &default_name) {
+            let default = names.remove(ix);
+            names.insert(0, default);
+        } else {
+            names.insert(0, default_name);
+        }
+    }
 
     let response = name::GetNamesResponse { names };
     Ok(axum::Json(response))
 }
 
+#[derive(Deserialize)]
+pub struct DefaultNameParams {
+    name_or_identity: NameOrIdentity,
+}
+
+pub async fn get_default_name<S: ControlStateDelegate>(
+    State(ctx): State<S>,
+    Path(DefaultNameParams { name_or_identity }): Path<DefaultNameParams>,
+) -> axum::response::Result<impl IntoResponse> {
+    let database_identity = name_or_identity.resolve(&ctx).await?;
+    let Some(default_name) = ctx
+        .lookup_database_default_name(&database_identity)
+        .await
+        .map_err(log_and_500)?
+    else {
+        return Err((StatusCode::NOT_FOUND, "database does not have a default name").into());
+    };
+
+    Ok(axum::Json(default_name))
+}
+
+pub async fn set_default_name<S: ControlStateDelegate + Authorization>(
+    State(ctx): State<S>,
+    Path(DefaultNameParams { name_or_identity }): Path<DefaultNameParams>,
+    Extension(auth): Extension<SpacetimeAuth>,
+    name: String,
+) -> axum::response::Result<impl IntoResponse> {
+    let requested_name = parse_domain_name(name).map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))?;
+    validate_database_name_path(&ctx, &requested_name).await?;
+    let database_identity = name_or_identity.resolve(&ctx).await?;
+
+    let database = ctx
+        .get_database_by_identity(&database_identity)
+        .await
+        .map_err(log_and_500)?
+        .ok_or(NO_SUCH_DATABASE)?;
+
+    ctx.authorize_action(auth.claims.identity, database.database_identity, Action::RenameDatabase)
+        .await?;
+
+    let mut names = ctx.reverse_lookup(&database_identity).await.map_err(log_and_500)?;
+    if !names.iter().any(|existing| existing == &requested_name) {
+        names.push(requested_name.clone());
+    }
+    names.retain(|existing| existing != &requested_name);
+    names.insert(0, requested_name);
+
+    let response = ctx
+        .replace_dns_records(&database_identity, &database.owner_identity, &names)
+        .await
+        .map_err(log_and_500)?;
+    let status = match response {
+        name::SetDomainsResult::Success => StatusCode::OK,
+        name::SetDomainsResult::PermissionDenied { .. }
+        | name::SetDomainsResult::PermissionDeniedOnAny { .. }
+        | name::SetDomainsResult::NotYourDatabase { .. } => StatusCode::UNAUTHORIZED,
+        name::SetDomainsResult::DatabaseNotFound => StatusCode::NOT_FOUND,
+        name::SetDomainsResult::OtherError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+    };
+
+    Ok((status, axum::Json(response)))
+}
+
 #[derive(Deserialize)]
 pub struct ResetDatabaseParams {
     name_or_identity: NameOrIdentity,
@@ -657,7 +780,7 @@ pub struct PublishDatabaseQueryParams {
 
 pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
-    Path(PublishDatabaseParams { name_or_identity }): Path<PublishDatabaseParams>,
+    Path(PublishDatabaseParams { mut name_or_identity }): Path<PublishDatabaseParams>,
     Query(PublishDatabaseQueryParams {
         clear,
         num_replicas,
@@ -708,6 +831,56 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
         }
     }
 
+    let mut parent = parent;
+
+    if let Some(NameOrIdentity::Name(child_name)) = name_or_identity.clone() {
+        if let Some(parent_noi) = parent.clone() {
+            if child_name.sub_domain().is_some() {
+                return Err(bad_request(
+                    "child database name cannot contain `/` when `parent` is set".into(),
+                ));
+            }
+            let full_parent_name = match parent_noi {
+                NameOrIdentity::Name(parent_name) => Some(parent_name),
+                NameOrIdentity::Identity(parent_identity) => {
+                    let parent_identity = Identity::from(parent_identity);
+                    let default_name = ctx
+                        .lookup_database_default_name(&parent_identity)
+                        .await
+                        .map_err(log_and_500)?;
+                    let Some(default_name) = default_name else {
+                        return Err(bad_request(
+                            format!(
+                                "Can't publish a named child \"{}\" of unnamed database {}",
+                                child_name, parent_identity
+                            )
+                            .into(),
+                        ));
+                    };
+                    Some(default_name)
+                }
+            };
+            let Some(full_parent_name) = full_parent_name else {
+                unreachable!("all parent variants are handled above");
+            };
+            let full_child_name = parse_domain_name(format!("{full_parent_name}/{child_name}"))
+                .map_err(|err| bad_request(err.to_string().into()))?;
+            name_or_identity = Some(NameOrIdentity::Name(full_child_name));
+        }
+    }
+
+    if parent.is_none() {
+        if let Some(NameOrIdentity::Name(name)) = name_or_identity.as_ref() {
+            if let Some(parent_name) = infer_parent_name(name)? {
+                parent = Some(NameOrIdentity::Name(parent_name));
+            }
+        }
+    }
+
+    if let Some(NameOrIdentity::Name(name)) = name_or_identity.as_ref() {
+        validate_database_name_path(&ctx, name).await?;
+    }
+
     let (database_identity, db_name) = get_or_create_identity_and_name(&ctx, &auth, name_or_identity.as_ref()).await?;
     let maybe_parent_database_identity = match parent.as_ref() {
         None => None,
@@ -801,7 +974,7 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
     }
 }
 
-/// Try to resolve `name_or_identity` to an [Identity] and [DatabaseName].
+/// Try to resolve `name_or_identity` to an [Identity] and [DomainName].
 ///
 /// - If the database exists and has a name registered for it, return that.
 /// - If the database does not exist, but `name_or_identity` is a name,
@@ -813,7 +986,7 @@ async fn get_or_create_identity_and_name<'a>(
     ctx: &(impl ControlStateDelegate + NodeDelegate),
     auth: &SpacetimeAuth,
     name_or_identity: Option<&'a NameOrIdentity>,
-) -> axum::response::Result<(Identity, Option<&'a DatabaseName>)> {
+) -> axum::response::Result<(Identity, Option<&'a DomainName>)> {
     match name_or_identity {
         Some(noi) => match noi.try_resolve(ctx).await.map_err(log_and_500)? {
             Ok(resolved) => Ok((resolved, noi.name())),
@@ -840,36 +1013,109 @@ async fn create_name(
     ctx: &(impl NodeDelegate + ControlStateDelegate),
     auth: &SpacetimeAuth,
     database_identity: &Identity,
-    name: &DatabaseName,
+    name: &DomainName,
 ) -> axum::response::Result<()> {
-    let tld: name::Tld = name.clone().into();
-    let tld = match ctx
-        .register_tld(&auth.claims.identity, tld)
-        .await
-        .map_err(log_and_500)?
-    {
-        name::RegisterTldResult::Success { domain } | name::RegisterTldResult::AlreadyRegistered { domain } => domain,
-        name::RegisterTldResult::Unauthorized { .. } => {
+    validate_database_name_path(ctx, name).await?;
+
+    let tld: name::Tld = name.to_tld();
+    let namespace_owner = ctx.lookup_namespace_owner(tld.as_str()).await.map_err(log_and_500)?;
+    match namespace_owner {
+        Some(owner) if owner != auth.claims.identity => {
             return Err((
                 StatusCode::UNAUTHORIZED,
                 axum::Json(PublishResult::PermissionDenied { name: name.clone() }),
             )
                 .into())
         }
+        Some(_) => {}
+        None => {
+            if !ctx.allow_register_tld_on_publish().await.map_err(log_and_500)? {
+                return Err((
+                    StatusCode::FORBIDDEN,
+                    format!(
+                        "{} does not own root namespace `{}`. You should publish under a root namespace that you own.",
+                        auth.claims.identity, tld
+                    ),
+                )
+                    .into());
+            }
+            match ctx
+                .register_tld(&auth.claims.identity, tld.clone())
+                .await
+                .map_err(log_and_500)?
+            {
+                name::RegisterTldResult::Success { .. } | name::RegisterTldResult::AlreadyRegistered { .. } => {}
+                name::RegisterTldResult::Unauthorized { .. } => {
+                    return Err((
+                        StatusCode::UNAUTHORIZED,
+                        axum::Json(PublishResult::PermissionDenied { name: name.clone() }),
+                    )
+                        .into())
+                }
+            };
+        }
     };
+
     let res = ctx
-        .create_dns_record(&auth.claims.identity, &tld.into(), database_identity)
+        .create_dns_record(&auth.claims.identity, name, database_identity)
         .await
         .map_err(log_and_500)?;
     match res {
         name::InsertDomainResult::Success { .. } => Ok(()),
         name::InsertDomainResult::TldNotRegistered { .. } | name::InsertDomainResult::PermissionDenied { .. } => {
-            Err(log_and_500("impossible: we just registered the tld"))
+            Err((StatusCode::FORBIDDEN, "permission denied to set database name").into())
         }
         name::InsertDomainResult::OtherError(e) => Err(log_and_500(e)),
     }
 }
 
+async fn validate_database_name_path(
+    ctx: &(impl ControlStateDelegate + ?Sized),
+    name: &DomainName,
+) -> axum::response::Result<()> {
+    let Some(sub_domain) = name.sub_domain() else {
+        return Ok(());
+    };
+
+    let mut parts = sub_domain.split('/').collect::<Vec<_>>();
+    if parts.len() <= 1 {
+        return Ok(());
+    }
+
+    // The last path segment is the database name itself; all preceding
+    // segments must already exist.
+    parts.pop();
+    let mut current = name.tld().to_string();
+    for part in parts {
+        current.push('/');
+        current.push_str(part);
+        if ctx
+            .lookup_database_identity(&current)
+            .await
+            .map_err(log_and_500)?
+            .is_none()
+        {
+            return Err(bad_request(format!("Parent database {current} not found").into()));
+        }
+    }
+
+    Ok(())
+}
+
+fn infer_parent_name(name: &DomainName) -> axum::response::Result<Option<DomainName>> {
+    let Some(sub_domain) = name.sub_domain() else {
+        return Ok(None);
+    };
+    let mut parts = sub_domain.split('/').collect::<Vec<_>>();
+    if parts.len() < 2 {
+        return Ok(None);
+    }
+    parts.pop();
+    let parent = parse_domain_name(format!("{}/{}", name.tld(), parts.join("/")))
+        .map_err(|err| bad_request(err.to_string().into()))?;
+    Ok(Some(parent))
+}
+
 fn schema_migration_policy(
     policy: MigrationPolicy,
     token: Option<Hash>,
@@ -1029,11 +1275,12 @@ pub async fn add_name<S: ControlStateDelegate>(
     Extension(auth): Extension<SpacetimeAuth>,
     name: String,
 ) -> axum::response::Result<impl IntoResponse> {
-    let name = DatabaseName::try_from(name).map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))?;
+    let name = parse_domain_name(name).map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))?;
+    validate_database_name_path(&ctx, &name).await?;
     let database_identity = name_or_identity.resolve(&ctx).await?;
 
     let response = ctx
-        .create_dns_record(&auth.claims.identity, &name.into(), &database_identity)
+        .create_dns_record(&auth.claims.identity, &name, &database_identity)
         .await
         // TODO: better error code handling
         .map_err(log_and_500)?;
@@ -1062,9 +1309,12 @@ pub async fn set_names<S: ControlStateDelegate + Authorization>(
     let validated_names = names
         .0
         .into_iter()
-        .map(|s| DatabaseName::from_str(&s).map(DomainName::from).map_err(|e| (s, e)))
+        .map(|s| parse_domain_name(s.as_str()).map_err(|e| (s, e)))
         .collect::<Result<Vec<_>, _>>()
         .map_err(|(input, e)| (StatusCode::BAD_REQUEST, format!("Error parsing `{input}`: {e}")))?;
+    for name in &validated_names {
+        validate_database_name_path(&ctx, name).await?;
+    }
 
     let database_identity = name_or_identity.resolve(&ctx).await?;
 
@@ -1093,7 +1343,10 @@ pub async fn set_names<S: ControlStateDelegate + Authorization>(
         })?;
 
     for name in &validated_names {
-        if ctx.lookup_database_identity(name.as_str()).await.unwrap().is_some() {
+        if let Some(existing_identity) = ctx.lookup_database_identity(name.as_str()).await.map_err(log_and_500)? {
+            if existing_identity == database_identity {
+                continue;
+            }
             return Ok((
                 StatusCode::BAD_REQUEST,
                 axum::Json(name::SetDomainsResult::OtherError(format!(
@@ -1163,12 +1416,18 @@ pub struct DatabaseRoutes<S> {
     pub names_post: MethodRouter<S>,
     /// PUT: /database/:name_or_identity/names
     pub names_put: MethodRouter<S>,
+    /// GET: /database/:name_or_identity/default-name
+    pub default_name_get: MethodRouter<S>,
+    /// POST: /database/:name_or_identity/default-name
+    pub default_name_post: MethodRouter<S>,
     /// GET: /database/:name_or_identity/identity
     pub identity_get: MethodRouter<S>,
     /// GET: /database/:name_or_identity/subscribe
     pub subscribe_get: MethodRouter<S>,
     /// POST: /database/:name_or_identity/call/:reducer
     pub call_reducer_procedure_post: MethodRouter<S>,
+    /// POST: /database/call/*name_or_identity_and_reducer
+    pub call_reducer_procedure_post_tail: MethodRouter<S>,
     /// GET: /database/:name_or_identity/schema
     pub schema_get: MethodRouter<S>,
     /// GET: /database/:name_or_identity/logs
@@ -1197,9 +1456,12 @@ where
             names_get: get(get_names::<S>),
             names_post: post(add_name::<S>),
             names_put: put(set_names::<S>),
+            default_name_get: get(get_default_name::<S>),
+            default_name_post: post(set_default_name::<S>),
             identity_get: get(get_identity::<S>),
             subscribe_get: get(handle_websocket::<S>),
             call_reducer_procedure_post: post(call::<S>),
+            call_reducer_procedure_post_tail: post(call_with_tail::<S>),
             schema_get: get(schema::<S>),
             logs_get: get(logs::<S>),
             sql_post: post(sql::<S>),
@@ -1215,6 +1477,21 @@ where
     S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static,
 {
     pub fn into_router(self, ctx: S) -> axum::Router<S> {
+        let identity_get_tail = self.identity_get.clone();
+        let subscribe_get_tail = self.subscribe_get.clone();
+        let call_tail = self.call_reducer_procedure_post_tail;
+        let schema_get_tail = self.schema_get.clone();
+        let logs_get_tail = self.logs_get.clone();
+        let sql_post_tail = self.sql_post.clone();
+        let names_get_tail = self.names_get.clone();
+        let names_post_tail = self.names_post.clone();
+        let names_put_tail = self.names_put.clone();
+        let default_name_get_tail = self.default_name_get.clone();
+        let default_name_post_tail = self.default_name_post.clone();
+        let pre_publish_tail = self.pre_publish.clone();
+        let reset_tail = self.db_reset.clone();
+        let timestamp_tail = self.timestamp_get.clone();
+
         let db_router = axum::Router::<S>::new()
             .route("/", self.db_put)
             .route("/", self.db_get)
@@ -1222,6 +1499,8 @@ where
             .route("/names", self.names_get)
             .route("/names", self.names_post)
             .route("/names", self.names_put)
+            .route("/default-name", self.default_name_get)
+            .route("/default-name", self.default_name_post)
             .route("/identity", self.identity_get)
             .route("/subscribe", self.subscribe_get)
             .route("/call/:reducer", self.call_reducer_procedure_post)
@@ -1234,6 +1513,20 @@ where
 
         axum::Router::new()
             .route("/", self.root_post)
+            .route("/identity/*name_or_identity", identity_get_tail)
+            .route("/subscribe/*name_or_identity", subscribe_get_tail)
+            .route("/call/*name_or_identity_and_reducer", call_tail)
+            .route("/schema/*name_or_identity", schema_get_tail)
+            .route("/logs/*name_or_identity", logs_get_tail)
+            .route("/sql/*name_or_identity", sql_post_tail)
+            .route("/names/*name_or_identity", names_get_tail)
+            .route("/names/*name_or_identity", names_post_tail)
+            .route("/names/*name_or_identity", names_put_tail)
+            .route("/default-name/*name_or_identity", default_name_get_tail)
+            .route("/default-name/*name_or_identity", default_name_post_tail)
+            .route("/pre_publish/*name_or_identity", pre_publish_tail)
+            .route("/reset/*name_or_identity", reset_tail)
+            .route("/unstable/timestamp/*name_or_identity", timestamp_tail)
             .nest("/:name_or_identity", db_router)
             .route_layer(axum::middleware::from_fn_with_state(ctx, anon_auth_middleware::<S>))
     }
diff --git a/crates/client-api/src/util.rs b/crates/client-api/src/util.rs
index adc1f632bc4..21027da7fb5 100644
--- a/crates/client-api/src/util.rs
+++ b/crates/client-api/src/util.rs
@@ -14,7 +14,7 @@ use http::{HeaderName, HeaderValue, StatusCode};
 
 use hyper::body::Body;
 use spacetimedb::Identity;
-use spacetimedb_client_api_messages::name::DatabaseName;
+use spacetimedb_client_api_messages::name::{parse_domain_name, DomainName};
 
 use crate::routes::identity::IdentityForUrl;
 use crate::{log_and_500, ControlStateReadAccess};
@@ -62,7 +62,7 @@ impl headers::Header for XForwardedFor {
 #[derive(Clone, Debug)]
 pub enum NameOrIdentity {
     Identity(IdentityForUrl),
-    Name(DatabaseName),
+    Name(DomainName),
 }
 
 impl NameOrIdentity {
@@ -73,7 +73,7 @@ impl NameOrIdentity {
         }
     }
 
-    pub fn name(&self) -> Option<&DatabaseName> {
+    pub fn name(&self) -> Option<&DomainName> {
         if let Self::Name(name) = self {
             Some(name)
         } else {
@@ -87,17 +87,17 @@ impl NameOrIdentity {
     /// returned directly.
     ///
     /// Otherwise, if `self` is a [`NameOrIdentity::Name`], the [`Identity`] is
-    /// looked up by that name in the SpacetimeDB DNS and returned.
+    /// looked up by that database name and returned.
     ///
-    /// Errors are returned if the DNS lookup fails.
+    /// Errors are returned if name lookup fails.
     ///
-    /// An `Ok` result is itself a [`Result`], which is `Err(DatabaseName)` if the
+    /// An `Ok` result is itself a [`Result`], which is `Err(DomainName)` if the
     /// given [`NameOrIdentity::Name`] is not registered in the SpacetimeDB DNS,
     /// i.e. no corresponding [`Identity`] exists.
     pub async fn try_resolve(
         &self,
         ctx: &(impl ControlStateReadAccess + ?Sized),
-    ) -> anyhow::Result<Result<Identity, &DatabaseName>> {
+    ) -> anyhow::Result<Result<Identity, &DomainName>> {
         Ok(match self {
             Self::Identity(identity) => Ok(Identity::from(*identity)),
             Self::Name(name) => ctx.lookup_database_identity(name.as_ref()).await?.ok_or(name),
@@ -106,12 +106,12 @@ impl NameOrIdentity {
 
     /// A variant of [`Self::try_resolve()`] which maps to a 404 (Not Found)
     /// response if `self` is a [`NameOrIdentity::Name`] for which no
-    /// corresponding [`Identity`] is found in the SpacetimeDB DNS.
+    /// corresponding [`Identity`] is found.
     pub async fn resolve(&self, ctx: &(impl ControlStateReadAccess + ?Sized)) -> axum::response::Result<Identity> {
         self.try_resolve(ctx)
             .await
             .map_err(log_and_500)?
-            .map_err(|name| (StatusCode::NOT_FOUND, format!("`{name}` not found")).into())
+            .map_err(|name| (StatusCode::NOT_FOUND, format!("database name `{name}` not found")).into())
     }
 
     /// If `self` is a [`NameOrIdentity::Name`], looks up the name in the
@@ -131,7 +131,7 @@ impl NameOrIdentity {
         match self {
             Self::Identity(identity) => Ok(Identity::from(*identity)),
             Self::Name(name) => ctx
-                .lookup_namespace_owner(name.as_ref())
+                .lookup_namespace_owner(name.tld().as_str())
                 .await
                 .map_err(log_and_500)?
                 .ok_or_else(|| (StatusCode::NOT_FOUND, format!("`{name}` not found")).into()),
@@ -148,7 +148,7 @@ impl<'de> ::serde::Deserialize<'de> for NameOrIdentity {
         if let Ok(addr) = Identity::from_hex(&s) {
             Ok(NameOrIdentity::Identity(IdentityForUrl::from(addr)))
         } else {
-            let name: DatabaseName = s.try_into().map_err(::serde::de::Error::custom)?;
+            let name: DomainName = parse_domain_name(s).map_err(::serde::de::Error::custom)?;
             Ok(NameOrIdentity::Name(name))
         }
     }
diff --git a/crates/pg/src/pg_server.rs b/crates/pg/src/pg_server.rs
index 860df156f25..4993826a616 100644
--- a/crates/pg/src/pg_server.rs
+++ b/crates/pg/src/pg_server.rs
@@ -27,7 +27,7 @@ use spacetimedb_client_api::routes::database;
 use spacetimedb_client_api::routes::database::{SqlParams, SqlQueryParams};
 use spacetimedb_client_api::{Authorization, ControlStateReadAccess, ControlStateWriteAccess, NodeDelegate};
 use spacetimedb_client_api_messages::http::SqlStmtResult;
-use spacetimedb_client_api_messages::name::DatabaseName;
+use spacetimedb_client_api_messages::name::parse_domain_name;
 use spacetimedb_lib::sats::satn::{PsqlClient, TypedSerializer};
 use spacetimedb_lib::sats::{satn, Serialize, Typespace};
 use spacetimedb_lib::version::spacetimedb_lib_version;
@@ -154,7 +154,9 @@ where
     async fn exe_sql(&self, query: String) -> PgWireResult<Vec<Response>> {
         let params = self.cached.lock().await.clone().unwrap();
         let db = SqlParams {
-            name_or_identity: database::NameOrIdentity::Name(DatabaseName(params.database.clone())),
+            name_or_identity: database::NameOrIdentity::Name(
+                parse_domain_name(params.database.as_str()).map_err(|e| PgError::Sql(e.to_string()))?,
+            ),
         };
 
         let sql = match response(
@@ -252,7 +254,9 @@ impl<T: Sync + Send + ControlStateReadAccess + ControlStateWriteAccess + NodeDel
                     log::info!("PG: Connecting to database: {database}");
                 }
 
-                let name = database::NameOrIdentity::Name(DatabaseName(database.clone()));
+                let name = database::NameOrIdentity::Name(
+                    parse_domain_name(database.as_str()).map_err(|e| PgError::Sql(e.to_string()))?,
+                );
                 match response(name.resolve(&self.ctx).await, &database).await {
                     Ok(identity) => identity,
                     Err(PgError::Pg(PgWireError::UserError(err))) => {
diff --git a/crates/standalone/src/lib.rs b/crates/standalone/src/lib.rs
index 6c4e61dc9c8..73518ebb80f 100644
--- a/crates/standalone/src/lib.rs
+++ b/crates/standalone/src/lib.rs
@@ -24,7 +24,7 @@ use spacetimedb_client_api::auth::{self, LOCALHOST};
 use spacetimedb_client_api::routes::subscribe::{HasWebSocketOptions, WebSocketOptions};
 use spacetimedb_client_api::{ControlStateReadAccess, DatabaseResetDef, Host, NodeDelegate};
 use spacetimedb_client_api_messages::name::{
-    DatabaseName, DomainName, InsertDomainResult, RegisterTldResult, SetDomainsResult, Tld,
+    parse_domain_name, DomainName, InsertDomainResult, RegisterTldResult, SetDomainsResult, Tld,
 };
 use spacetimedb_datastore::db_metrics::data_size::DATA_SIZE_METRICS;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
@@ -250,9 +250,21 @@ impl spacetimedb_client_api::ControlStateReadAccess for StandaloneEnv {
         Ok(self.control_db.spacetime_reverse_dns(database_identity)?)
     }
 
+    async fn lookup_database_default_name(&self, database_identity: &Identity) -> anyhow::Result<Option<DomainName>> {
+        Ok(self
+            .control_db
+            .spacetime_reverse_dns(database_identity)?
+            .into_iter()
+            .next())
+    }
+
     async fn lookup_namespace_owner(&self, name: &str) -> anyhow::Result<Option<Identity>> {
-        let name: DatabaseName = name.parse()?;
-        Ok(self.control_db.spacetime_lookup_tld(Tld::from(name))?)
+        let domain = parse_domain_name(name)?;
+        Ok(self.control_db.spacetime_lookup_tld(domain.tld())?)
+    }
+
+    async fn allow_register_tld_on_publish(&self) -> anyhow::Result<bool> {
+        Ok(true)
     }
 }