Embedded mode foundation (Plan 1)

Cargo.toml

@@ -3,6 +3,8 @@ members = [
     "crates/common",
     "crates/api",
     "crates/worker",
+    "crates/client",
+    "crates/embedded-worker",
     "crates/mock-server",
     "crates/dashboard",
 ]

crates/api/src/handlers/workspaces.rs

@@ -21,7 +21,11 @@ pub async fn create(
         .await?
         .ok_or_else(|| AppError::OrgNotFound(format!("Organization {} not found", org_id)))?;
 
-    let schema_name = build_schema_name(&org_id, &body.slug);
+    let schema_name = build_schema_name(
+        &state.config.schema.tenant_schema_prefix,
+        &org_id,
+        &body.slug,
+    );
 
     let workspace =
         db::workspaces::create(&state.pool, &org_id, &body.name, &body.slug, &schema_name)

crates/client/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "kronos-client"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+kronos-common = { path = "../common" }
+sqlx = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+clap = { version = "4", features = ["derive", "env"] }
+anyhow = { workspace = true }
+tracing-subscriber = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["full", "macros"] }

crates/client/src/bin/kronos-migrate.rs

@@ -0,0 +1,45 @@
+//! `kronos-migrate` — render and apply Kronos migrations.
+//!
+//! Designed to drop into the existing `just db-migrate` recipe.
+
+use clap::Parser;
+use kronos_client::{migrate, SchemaConfig};
+use sqlx::PgPool;
+
+#[derive(Parser, Debug)]
+#[command(about = "Render and apply Kronos migrations")]
+struct Args {
+    /// Postgres connection URL.
+    #[arg(long, env = "TE_DATABASE_URL", hide_env_values = true)]
+    database_url: String,
+
+    /// System schema for shared tables (organizations, workspaces).
+    #[arg(long, env = "TE_SYSTEM_SCHEMA", default_value = "public")]
+    system_schema: String,
+
+    /// Prefix prepended to per-workspace schema names.
+    #[arg(long, env = "TE_TENANT_SCHEMA_PREFIX", default_value = "")]
+    tenant_schema_prefix: String,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let args = Args::parse();
+
+    let cfg = SchemaConfig {
+        system_schema: args.system_schema,
+        tenant_schema_prefix: args.tenant_schema_prefix,
+    };
+    cfg.validate()
+        .map_err(|e| anyhow::anyhow!("invalid schema config: {}", e))?;
+
+    let pool = PgPool::connect(&args.database_url).await?;
+    migrate(&pool, &cfg).await?;
+    tracing::info!(
+        system_schema = %cfg.system_schema,
+        tenant_schema_prefix = ?cfg.tenant_schema_prefix,
+        "migrations applied"
+    );
+    Ok(())
+}

crates/client/src/lib.rs

@@ -0,0 +1,7 @@
+//! Kronos library API. Today this crate exposes only the migration entry
+//! point; subsequent plans add the enqueue + CRUD surface.
+
+pub mod migrate;
+
+pub use kronos_common::schema_config::SchemaConfig;
+pub use migrate::{migrate, MigrateError, Migration, MIGRATIONS};

crates/client/src/migrate.rs

@@ -0,0 +1,14 @@
+//! Public migration entry point. Library users call `kronos_client::migrate(&pool, &opts)`
+//! to render and apply all embedded migrations against their database.
+
+pub use kronos_common::migrations::{apply, MigrateError, Migration, MIGRATIONS};
+pub use kronos_common::schema_config::SchemaConfig;
+
+use sqlx::PgPool;
+
+/// Convenience alias matching the spec's API shape.
+///
+/// Equivalent to `kronos_common::migrations::apply(pool, opts)`.
+pub async fn migrate(pool: &PgPool, opts: &SchemaConfig) -> Result<(), MigrateError> {
+    apply(pool, opts).await
+}

crates/client/tests/migrate_kronos_namespace.rs

@@ -0,0 +1,76 @@
+//! Smoke test for non-default schema namespacing.
+//!
+//! Applies migrations with `system_schema = "kronos_smoke"` against a fresh
+//! database, then asserts the expected tables exist in the configured schema.
+//! This proves the parameterized migration path lands tables in the requested
+//! namespace; it does not assert the absence of tables in `public`, since
+//! sibling integration tests legitimately populate `public` under the default
+//! schema config.
+//!
+//! Requires Postgres at `TE_DATABASE_URL` (default postgres://kronos:kronos@localhost:5432/taskexecutor).
+//! Run with: `cargo test -p kronos-client --test migrate_kronos_namespace -- --ignored`
+
+use kronos_client::{migrate, SchemaConfig};
+use sqlx::PgPool;
+
+fn db_url() -> String {
+    std::env::var("TE_DATABASE_URL").unwrap_or_else(|_| {
+        "postgres://kronos:kronos@localhost:5432/taskexecutor".to_string()
+    })
+}
+
+#[tokio::test]
+#[ignore]
+async fn migrations_create_tables_in_configured_schema_only() {
+    let pool = PgPool::connect(&db_url()).await.unwrap();
+
+    // Wipe any prior state from this test schema, but DO NOT touch `public`
+    // (the integration tests in the same DB rely on `public`).
+    sqlx::query("DROP SCHEMA IF EXISTS kronos_smoke CASCADE")
+        .execute(&pool)
+        .await
+        .unwrap();
+
+    let cfg = SchemaConfig {
+        system_schema: "kronos_smoke".to_string(),
+        tenant_schema_prefix: "kronos_".to_string(),
+    };
+
+    migrate(&pool, &cfg).await.unwrap();
+
+    // organizations and workspaces are in kronos_smoke
+    let orgs_in_kronos: (bool,) = sqlx::query_as(
+        "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
+         WHERE table_schema = 'kronos_smoke' AND table_name = 'organizations')",
+    )
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert!(orgs_in_kronos.0, "organizations should exist in kronos_smoke");
+
+    let ws_in_kronos: (bool,) = sqlx::query_as(
+        "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
+         WHERE table_schema = 'kronos_smoke' AND table_name = 'workspaces')",
+    )
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert!(ws_in_kronos.0, "workspaces should exist in kronos_smoke");
+
+    // The migrations create exactly two system-level tables: organizations and
+    // workspaces. Verify the count in kronos_smoke matches.
+    let table_count: (i64,) = sqlx::query_as(
+        "SELECT COUNT(*) FROM information_schema.tables \
+         WHERE table_schema = 'kronos_smoke' AND table_name IN ('organizations', 'workspaces')",
+    )
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(table_count.0, 2, "expected both organizations and workspaces in kronos_smoke");
+
+    // Cleanup so re-running the test is clean.
+    sqlx::query("DROP SCHEMA IF EXISTS kronos_smoke CASCADE")
+        .execute(&pool)
+        .await
+        .unwrap();
+}

crates/common/src/config.rs

@@ -168,6 +168,38 @@ impl MetricsEnv {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct SchemaEnv {
+    pub system_schema: String,
+    pub tenant_schema_prefix: String,
+}
+
+impl SchemaEnv {
+    fn new() -> Self {
+        Self {
+            system_schema: get_from_env_or_default(
+                "TE_SYSTEM_SCHEMA",
+                "public".to_string(),
+            ),
+            tenant_schema_prefix: get_from_env_or_default(
+                "TE_TENANT_SCHEMA_PREFIX",
+                String::new(),
+            ),
+        }
+    }
+
+    /// Convert env-loaded values into a [`SchemaConfig`] value.
+    ///
+    /// Callers should call [`SchemaConfig::validate`] before using the result;
+    /// [`AppConfig::from_env`] already does this on startup.
+    pub fn to_schema_config(&self) -> crate::schema_config::SchemaConfig {
+        crate::schema_config::SchemaConfig {
+            system_schema: self.system_schema.clone(),
+            tenant_schema_prefix: self.tenant_schema_prefix.clone(),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Top-level config
 // ---------------------------------------------------------------------------
@@ -179,6 +211,7 @@ pub struct AppConfig {
     pub worker: WorkerEnv,
     pub crypto: CryptoEnv,
     pub metrics: MetricsEnv,
+    pub schema: SchemaEnv,
 }
 
 impl AppConfig {
@@ -213,13 +246,67 @@ impl AppConfig {
             .await
             .map_err(|e| anyhow::anyhow!(e))?;
         let metrics = MetricsEnv::new();
+        let schema = SchemaEnv::new();
+
+        // Validate schema config early so misconfiguration fails fast at startup.
+        schema
+            .to_schema_config()
+            .validate()
+            .map_err(|e| anyhow::anyhow!("invalid schema config: {}", e))?;
 
         Ok(Self {
             db,
             server,
             worker,
             crypto,
             metrics,
+            schema,
         })
     }
 }
+
+#[cfg(test)]
+mod schema_env_tests {
+    use super::*;
+
+    #[test]
+    fn defaults_match_today() {
+        let saved_sys = std::env::var("TE_SYSTEM_SCHEMA").ok();
+        let saved_prefix = std::env::var("TE_TENANT_SCHEMA_PREFIX").ok();
+        std::env::remove_var("TE_SYSTEM_SCHEMA");
+        std::env::remove_var("TE_TENANT_SCHEMA_PREFIX");
+
+        let s = SchemaEnv::new();
+        assert_eq!(s.system_schema, "public");
+        assert_eq!(s.tenant_schema_prefix, "");
+
+        if let Some(v) = saved_sys {
+            std::env::set_var("TE_SYSTEM_SCHEMA", v);
+        }
+        if let Some(v) = saved_prefix {
+            std::env::set_var("TE_TENANT_SCHEMA_PREFIX", v);
+        }
+    }
+
+    #[test]
+    fn picks_up_non_default_values() {
+        let saved_sys = std::env::var("TE_SYSTEM_SCHEMA").ok();
+        let saved_prefix = std::env::var("TE_TENANT_SCHEMA_PREFIX").ok();
+
+        std::env::set_var("TE_SYSTEM_SCHEMA", "kronos_test_env");
+        std::env::set_var("TE_TENANT_SCHEMA_PREFIX", "k_");
+
+        let s = SchemaEnv::new();
+        assert_eq!(s.system_schema, "kronos_test_env");
+        assert_eq!(s.tenant_schema_prefix, "k_");
+
+        match saved_sys {
+            Some(v) => std::env::set_var("TE_SYSTEM_SCHEMA", v),
+            None => std::env::remove_var("TE_SYSTEM_SCHEMA"),
+        }
+        match saved_prefix {
+            Some(v) => std::env::set_var("TE_TENANT_SCHEMA_PREFIX", v),
+            None => std::env::remove_var("TE_TENANT_SCHEMA_PREFIX"),
+        }
+    }
+}

crates/common/src/lib.rs

@@ -7,7 +7,9 @@ pub mod error;
 #[cfg(feature = "kms")]
 pub mod kms;
 pub mod metrics;
+pub mod migrations;
 pub mod models;
 pub mod pagination;
+pub mod schema_config;
 pub mod template;
 pub mod tenant;

crates/common/src/migrations/embedded.rs

@@ -0,0 +1,26 @@
+//! Migration template files embedded at compile time. The order in
+//! `MIGRATIONS` matches the order they must be applied.
+
+pub struct Migration {
+    pub name: &'static str,
+    pub template: &'static str,
+}
+
+pub const MIGRATIONS: &[Migration] = &[
+    Migration {
+        name: "20260317000000_initial",
+        template: include_str!("../../../../migrations/20260317000000_initial.sql"),
+    },
+    Migration {
+        name: "20260318000000_multi_tenancy",
+        template: include_str!("../../../../migrations/20260318000000_multi_tenancy.sql"),
+    },
+    Migration {
+        name: "20260322000000_txn_based_pickup",
+        template: include_str!("../../../../migrations/20260322000000_txn_based_pickup.sql"),
+    },
+    Migration {
+        name: "20260322000001_pg_cron",
+        template: include_str!("../../../../migrations/20260322000001_pg_cron.sql"),
+    },
+];