From c8b57a6a5d74faa05a81df8e84f051799fa4c1ae Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 13 Aug 2026 22:25:55 +0900 Subject: [PATCH] server: persist evaluated virtual config trees --- Cargo.lock | 1 + crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/config_source.rs | 463 +++++++++++++++++++ crates/workspace-server/src/lib.rs | 3 + crates/workspace-server/src/server.rs | 118 ++++- crates/workspace-server/src/store.rs | 48 ++ 6 files changed, 631 insertions(+), 3 deletions(-) create mode 100644 crates/workspace-server/src/config_source.rs diff --git a/Cargo.lock b/Cargo.lock index d25198cf..cdc5792b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6227,6 +6227,7 @@ dependencies = [ "async-trait", "axum", "chrono", + "config-source", "flow", "futures", "manifest", diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 4f7f7e4f..b2008cfb 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -19,6 +19,7 @@ async-trait.workspace = true axum = { workspace = true, features = ["ws"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } futures.workspace = true +config-source.workspace = true flow = { path = "../flow" } manifest.workspace = true protocol = { workspace = true } diff --git a/crates/workspace-server/src/config_source.rs b/crates/workspace-server/src/config_source.rs new file mode 100644 index 00000000..232c86d2 --- /dev/null +++ b/crates/workspace-server/src/config_source.rs @@ -0,0 +1,463 @@ +use chrono::{SecondsFormat, Utc}; +use config_source::{ + ConfigContentType, ConfigEntry, ConfigTreeChange, ConfigTreeSnapshot, DECODAL_VERSION, + DEFAULT_IMPORT_POLICY_VERSION, DEFAULT_SCHEMA_VERSION, EvaluationResult, SnapshotEnvironment, + ToolchainContract, VirtualPath, +}; +use rusqlite::{OptionalExtension, TransactionBehavior, params}; +use serde::{Deserialize, Serialize}; + +use crate::{Error, Result, SqliteWorkspaceStore}; + +pub const DEFAULT_CONFIG_ENTRYPOINT: &str = "workspace.dcdl"; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkspaceConfigState { + pub snapshot: ConfigTreeSnapshot, + pub contract: ToolchainContract, + pub projection_digest: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluatedConfigCandidate { + pub base_revision: u64, + pub base_digest: String, + pub snapshot: ConfigTreeSnapshot, + pub contract: ToolchainContract, + pub evaluation: EvaluationResult, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ConfigCommitRequest { + pub base_revision: u64, + pub base_digest: String, + pub changes: Vec, + pub entrypoints: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ConfigPreviewRequest { + pub changes: Vec, + pub entrypoints: Vec, +} + +impl SqliteWorkspaceStore { + pub fn load_workspace_config( + &self, + workspace_id: &str, + ) -> Result> { + self.with_conn(|conn| load_state(conn, workspace_id)) + } + + pub fn evaluate_workspace_config_candidate( + &self, + workspace_id: &str, + request: &ConfigCommitRequest, + ) -> Result { + let current = self + .load_workspace_config(workspace_id)? + .unwrap_or_else(empty_state); + if current.snapshot.revision != request.base_revision + || current.snapshot.digest != request.base_digest + { + return Err(config_conflict(format!( + "base revision/digest mismatch; current revision is {}", + current.snapshot.revision + ))); + } + evaluate_candidate(current, &request.changes, request.entrypoints.clone()) + } + + pub fn preview_workspace_config( + &self, + workspace_id: &str, + request: &ConfigPreviewRequest, + ) -> Result { + let current = self + .load_workspace_config(workspace_id)? + .unwrap_or_else(empty_state); + evaluate_candidate(current, &request.changes, request.entrypoints.clone()) + } + + pub fn commit_evaluated_workspace_config( + &self, + workspace_id: &str, + candidate: &EvaluatedConfigCandidate, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let workspace_exists: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM workspaces WHERE workspace_id = ?1)", + [workspace_id], + |row| row.get(0), + )?; + if !workspace_exists { + return Err(Error::WorkspaceIdMismatch); + } + let current = load_state(&tx, workspace_id)?.unwrap_or_else(empty_state); + if current.snapshot.revision != candidate.base_revision + || current.snapshot.digest != candidate.base_digest + { + return Err(config_conflict(format!( + "base revision/digest mismatch; current revision is {}", + current.snapshot.revision + ))); + } + let next_revision = current.snapshot.revision + 1; + let mut snapshot = candidate.snapshot.clone(); + snapshot.revision = next_revision; + let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true); + tx.execute( + r#"INSERT INTO workspace_config_trees ( + workspace_id, revision, tree_digest, schema_version, entrypoints_json, + decodal_version, import_policy_version, toolchain_fingerprint, + projection_digest, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) + ON CONFLICT(workspace_id) DO UPDATE SET + revision = excluded.revision, + tree_digest = excluded.tree_digest, + schema_version = excluded.schema_version, + decodal_version = excluded.decodal_version, + import_policy_version = excluded.import_policy_version, + toolchain_fingerprint = excluded.toolchain_fingerprint, + projection_digest = excluded.projection_digest, + updated_at = excluded.updated_at"#, + params![ + workspace_id, + next_revision as i64, + snapshot.digest, + candidate.contract.schema_version, + serde_json::to_string(&candidate.contract.entrypoints) + .map_err(|error| Error::Store(error.to_string()))?, + candidate.contract.decodal_version, + candidate.contract.import_policy_version, + candidate.contract.fingerprint, + candidate.evaluation.projection_digest, + now, + ], + )?; + tx.execute( + "DELETE FROM workspace_config_entries WHERE workspace_id = ?1", + [workspace_id], + )?; + for entry in snapshot.entries.values() { + tx.execute( + r#"INSERT INTO workspace_config_entries ( + workspace_id, path, content_type, content, content_digest + ) VALUES (?1, ?2, ?3, ?4, ?5)"#, + params![ + workspace_id, + entry.path.as_str(), + content_type_label(entry.content_type), + entry.content, + entry.content_digest, + ], + )?; + } + let manifest_json = serde_json::to_string(&snapshot.entries) + .map_err(|error| Error::Store(error.to_string()))?; + tx.execute( + r#"INSERT INTO workspace_config_tree_revisions ( + workspace_id, revision, tree_digest, toolchain_fingerprint, + projection_digest, manifest_json, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#, + params![ + workspace_id, + next_revision as i64, + snapshot.digest, + candidate.contract.fingerprint, + candidate.evaluation.projection_digest, + manifest_json, + now, + ], + )?; + tx.commit()?; + Ok(WorkspaceConfigState { + snapshot, + contract: candidate.contract.clone(), + projection_digest: candidate.evaluation.projection_digest.clone(), + }) + }) + } + + pub fn evaluate_and_commit_workspace_config( + &self, + workspace_id: &str, + request: &ConfigCommitRequest, + ) -> Result { + let candidate = self.evaluate_workspace_config_candidate(workspace_id, request)?; + self.commit_evaluated_workspace_config(workspace_id, &candidate) + } +} + +fn evaluate_candidate( + current: WorkspaceConfigState, + changes: &[ConfigTreeChange], + entrypoints: Vec, +) -> Result { + let snapshot = current.snapshot.apply(changes).map_err(config_error)?; + let contract = ToolchainContract::new( + DEFAULT_SCHEMA_VERSION, + entrypoints, + DEFAULT_IMPORT_POLICY_VERSION, + ); + let evaluation = SnapshotEnvironment::new(snapshot.clone()) + .evaluate_contract(&contract) + .map_err(|diagnostics| { + Error::InvalidInput( + serde_json::to_string(&diagnostics) + .unwrap_or_else(|_| "virtual config evaluation failed".to_string()), + ) + })?; + Ok(EvaluatedConfigCandidate { + base_revision: current.snapshot.revision, + base_digest: current.snapshot.digest, + snapshot, + contract, + evaluation, + }) +} + +fn load_state( + conn: &rusqlite::Connection, + workspace_id: &str, +) -> Result> { + let header = conn + .query_row( + r#"SELECT revision, tree_digest, schema_version, entrypoints_json, + decodal_version, import_policy_version, toolchain_fingerprint, projection_digest + FROM workspace_config_trees WHERE workspace_id = ?1"#, + [workspace_id], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, u32>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, u32>(5)?, + row.get::<_, String>(6)?, + row.get::<_, String>(7)?, + )) + }, + ) + .optional()?; + let Some(( + revision, + stored_digest, + schema_version, + entrypoints_json, + decodal_version, + import_policy_version, + fingerprint, + projection_digest, + )) = header + else { + return Ok(None); + }; + let mut statement = conn.prepare( + r#"SELECT path, content_type, content, content_digest + FROM workspace_config_entries WHERE workspace_id = ?1 ORDER BY path"#, + )?; + let entries = statement + .query_map([workspace_id], |row| { + let path = row.get::<_, String>(0)?; + let content_type = row.get::<_, String>(1)?; + let content = row.get::<_, String>(2)?; + let stored_entry_digest = row.get::<_, String>(3)?; + Ok((path, content_type, content, stored_entry_digest)) + })? + .collect::, _>>()? + .into_iter() + .map(|(path, content_type, content, stored_entry_digest)| { + let path = VirtualPath::parse(path).map_err(config_error)?; + let entry = ConfigEntry::new(path, parse_content_type(&content_type)?, content) + .map_err(config_error)?; + if entry.content_digest != stored_entry_digest { + return Err(Error::RegistryInconsistency(format!( + "virtual config entry digest mismatch for {}", + entry.path + ))); + } + Ok(entry) + }) + .collect::>>()?; + let snapshot = + ConfigTreeSnapshot::from_entries(revision as u64, entries).map_err(config_error)?; + if snapshot.digest != stored_digest { + return Err(Error::RegistryInconsistency(format!( + "virtual config tree digest mismatch for Workspace {workspace_id}" + ))); + } + let entrypoints: Vec = serde_json::from_str(&entrypoints_json) + .map_err(|error| Error::RegistryInconsistency(error.to_string()))?; + let contract = ToolchainContract::new(schema_version, entrypoints, import_policy_version); + if decodal_version != DECODAL_VERSION || contract.fingerprint != fingerprint { + return Err(Error::RegistryInconsistency(format!( + "virtual config toolchain metadata mismatch for Workspace {workspace_id}" + ))); + } + Ok(Some(WorkspaceConfigState { + snapshot, + contract, + projection_digest, + })) +} + +fn empty_state() -> WorkspaceConfigState { + WorkspaceConfigState { + snapshot: ConfigTreeSnapshot::empty(), + contract: ToolchainContract::new( + DEFAULT_SCHEMA_VERSION, + Vec::new(), + DEFAULT_IMPORT_POLICY_VERSION, + ), + projection_digest: config_source::digest_bytes(b"[]"), + } +} + +fn content_type_label(value: ConfigContentType) -> &'static str { + match value { + ConfigContentType::Decodal => "decodal", + ConfigContentType::Text => "text", + } +} + +fn parse_content_type(value: &str) -> Result { + match value { + "decodal" => Ok(ConfigContentType::Decodal), + "text" => Ok(ConfigContentType::Text), + _ => Err(Error::RegistryInconsistency(format!( + "unknown virtual config content type {value:?}" + ))), + } +} + +fn config_error(error: impl std::fmt::Display) -> Error { + Error::InvalidInput(error.to_string()) +} + +fn config_conflict(message: impl Into) -> Error { + Error::WorkspaceConfigConflict(message.into()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ControlPlaneStore, WorkspaceRecord}; + + fn workspace() -> WorkspaceRecord { + WorkspaceRecord { + workspace_id: "w-config".into(), + owner_account_id: None, + display_name: "Config".into(), + state: "active".into(), + created_at: "2026-08-13T00:00:00Z".into(), + updated_at: "2026-08-13T00:00:00Z".into(), + } + } + + fn path(value: &str) -> VirtualPath { + VirtualPath::parse(value).unwrap() + } + + #[tokio::test] + async fn invalid_candidate_is_never_persisted() { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + store.upsert_workspace(&workspace()).await.unwrap(); + let current = ConfigTreeSnapshot::empty(); + let error = store + .evaluate_and_commit_workspace_config( + "w-config", + &ConfigCommitRequest { + base_revision: 0, + base_digest: current.digest, + changes: vec![ConfigTreeChange::Create { + path: path(DEFAULT_CONFIG_ENTRYPOINT), + content_type: ConfigContentType::Decodal, + content: "{ broken = ; }".into(), + }], + entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)], + }, + ) + .unwrap_err(); + assert!(matches!(error, Error::InvalidInput(_))); + assert!(store.load_workspace_config("w-config").unwrap().is_none()); + } + + #[tokio::test] + async fn valid_candidate_commits_snapshot_revision_and_provenance_atomically() { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + store.upsert_workspace(&workspace()).await.unwrap(); + let empty = ConfigTreeSnapshot::empty(); + let committed = store + .evaluate_and_commit_workspace_config( + "w-config", + &ConfigCommitRequest { + base_revision: 0, + base_digest: empty.digest, + changes: vec![ConfigTreeChange::Create { + path: path(DEFAULT_CONFIG_ENTRYPOINT), + content_type: ConfigContentType::Decodal, + content: "{ answer = 42; }".into(), + }], + entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)], + }, + ) + .unwrap(); + assert_eq!(committed.snapshot.revision, 1); + assert_eq!(committed.contract.decodal_version, DECODAL_VERSION); + assert!(!committed.projection_digest.is_empty()); + let reread = store.load_workspace_config("w-config").unwrap().unwrap(); + assert_eq!(reread.snapshot, committed.snapshot); + } + + #[tokio::test] + async fn stale_cas_cannot_overwrite_newer_tree() { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + store.upsert_workspace(&workspace()).await.unwrap(); + let empty = ConfigTreeSnapshot::empty(); + let request = ConfigCommitRequest { + base_revision: 0, + base_digest: empty.digest, + changes: vec![ConfigTreeChange::Create { + path: path(DEFAULT_CONFIG_ENTRYPOINT), + content_type: ConfigContentType::Decodal, + content: "{ answer = 42; }".into(), + }], + entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)], + }; + let candidate = store + .evaluate_workspace_config_candidate("w-config", &request) + .unwrap(); + store + .commit_evaluated_workspace_config("w-config", &candidate) + .unwrap(); + let error = store + .commit_evaluated_workspace_config("w-config", &candidate) + .unwrap_err(); + assert!(matches!(error, Error::WorkspaceConfigConflict(_))); + } + + #[test] + fn migration_creates_config_authority_without_changing_applied_migrations() { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + store + .with_conn(|conn| { + for table in [ + "workspace_config_trees", + "workspace_config_entries", + "workspace_config_tree_revisions", + ] { + let exists: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)", + [table], + |row| row.get(0), + )?; + assert!(exists, "missing {table}"); + } + Ok(()) + }) + .unwrap(); + } +} diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index daa9f66e..811a3f0a 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -8,6 +8,7 @@ pub mod auth; pub mod authority; pub mod companion; pub mod config; +pub mod config_source; pub mod hosts; pub mod identity; pub mod memory_backend; @@ -106,6 +107,8 @@ pub enum Error { TicketAssignmentConflict(String), #[error("Workdir attachment conflict: {0}")] WorkdirAttachmentConflict(String), + #[error("Workspace config update conflict: {0}")] + WorkspaceConfigConflict(String), #[error("Registry inconsistency: {0}")] RegistryInconsistency(String), #[error("Worker source identity is invalid: {0}")] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 8a0aac22..afeede77 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -12,6 +12,7 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, patch, post, put}; use axum::{Json, Router}; use chrono::{Duration, SecondsFormat, Utc}; +use config_source::ConfigTreeSnapshot; use flow::{FlowSourceKind, FlowSourceResolveRequest, ResolvedFlowSource}; use futures::{SinkExt, StreamExt}; use memory::backend::{ @@ -61,6 +62,7 @@ use crate::companion::{ CompanionStatusResponse, CompanionTranscriptProjection, }; use crate::config::{BackendRuntimesConfigFile, RemoteRuntimeConfigFile, resolve_remote_runtime}; +use crate::config_source::{ConfigCommitRequest, ConfigPreviewRequest}; use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EMBEDDED_RUNTIME_ID, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, @@ -252,6 +254,7 @@ const ORCHESTRATOR_ATTENTION_PROMPT: &str = include_str!(concat!( pub struct WorkspaceApi { pub(crate) config: ServerConfig, pub(crate) store: Arc, + config_store: Arc, authority: SqliteWorkspaceAuthority, runtime: Arc, companion: Arc, @@ -741,7 +744,11 @@ impl WorkspaceApi { let runtime = Arc::new(runtime); let companion = Arc::new(CompanionConsole::disabled()); let observation_proxy = BackendObservationProxy::new(config.runtime_event_sources.clone()); + let config_store = Arc::new(crate::SqliteWorkspaceStore::open( + config.database_path.clone(), + )?); let api = Self { + config_store, authority: SqliteWorkspaceAuthority::new( config.database_path.clone(), config.workspace_id.clone(), @@ -1132,6 +1139,22 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/settings/workspace", get(scoped_get_workspace_settings).put(scoped_update_workspace_settings), ) + .route( + "/api/w/{workspace_id}/config/source-tree", + get(scoped_get_workspace_config_tree), + ) + .route( + "/api/w/{workspace_id}/config/source-tree/preview", + post(scoped_preview_workspace_config_tree), + ) + .route( + "/api/w/{workspace_id}/config/source-tree/commit", + post(scoped_commit_workspace_config_tree), + ) + .route( + "/api/w/{workspace_id}/config/source-tree/entries/{*path}", + get(scoped_get_workspace_config_entry), + ) .route( "/api/w/{workspace_id}/settings/profiles", get(scoped_get_profile_settings).post(scoped_create_profile_source), @@ -2418,6 +2441,95 @@ async fn scoped_update_workspace_settings( )) } +#[derive(Debug, Deserialize)] +struct WorkspaceConfigEntryPath { + workspace_id: String, + path: String, +} + +#[derive(Debug, Serialize)] +struct WorkspaceConfigTreeResponse { + snapshot: ConfigTreeSnapshot, + contract: config_source::ToolchainContract, + projection_digest: String, +} + +async fn scoped_get_workspace_config_tree( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let state = api + .config_store + .load_workspace_config(&path.workspace_id)? + .unwrap_or_else(|| crate::config_source::WorkspaceConfigState { + snapshot: ConfigTreeSnapshot::empty(), + contract: config_source::ToolchainContract::new( + config_source::DEFAULT_SCHEMA_VERSION, + Vec::new(), + config_source::DEFAULT_IMPORT_POLICY_VERSION, + ), + projection_digest: config_source::digest_bytes(b"[]"), + }); + Ok(Json(WorkspaceConfigTreeResponse { + snapshot: state.snapshot, + contract: state.contract, + projection_digest: state.projection_digest, + })) +} + +async fn scoped_get_workspace_config_entry( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let virtual_path = config_source::VirtualPath::parse(&path.path) + .map_err(|error| ApiError::from(Error::InvalidInput(error.to_string())))?; + let state = api + .config_store + .load_workspace_config(&path.workspace_id)? + .ok_or_else(|| { + ApiError::from(Error::InvalidRecordId("virtual config source tree".into())) + })?; + let entry = state + .snapshot + .get(&virtual_path) + .cloned() + .ok_or_else(|| ApiError::from(Error::InvalidRecordId(path.path)))?; + Ok(Json(entry)) +} + +async fn scoped_preview_workspace_config_tree( + State(api): State, + AxumPath(path): AxumPath, + Json(request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + Ok(Json( + api.config_store + .preview_workspace_config(&path.workspace_id, &request)?, + )) +} + +async fn scoped_commit_workspace_config_tree( + State(api): State, + AxumPath(path): AxumPath, + Json(request): Json, +) -> ApiResult<(StatusCode, Json)> { + validate_workspace_scope(&api, &path.workspace_id)?; + let state = api + .config_store + .evaluate_and_commit_workspace_config(&path.workspace_id, &request)?; + Ok(( + StatusCode::CREATED, + Json(WorkspaceConfigTreeResponse { + snapshot: state.snapshot, + contract: state.contract, + projection_digest: state.projection_digest, + }), + )) +} + async fn scoped_get_profile_settings( State(api): State, AxumPath(path): AxumPath, @@ -11255,9 +11367,9 @@ impl IntoResponse for ApiError { Error::BrowserMergeConfirmationRequired | Error::BrowserReopenConfirmationRequired => { StatusCode::FORBIDDEN } - Error::TicketAssignmentConflict(_) | Error::WorkdirAttachmentConflict(_) => { - StatusCode::CONFLICT - } + Error::TicketAssignmentConflict(_) + | Error::WorkdirAttachmentConflict(_) + | Error::WorkspaceConfigConflict(_) => StatusCode::CONFLICT, Error::WorkerSourceIdentity(_) | Error::InvalidInput(_) => StatusCode::BAD_REQUEST, Error::InvalidRuntimeIdentifier { .. } | Error::ReservedWorkerName(_) => { StatusCode::BAD_REQUEST diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 23236931..019c47f5 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -166,6 +166,11 @@ const MIGRATIONS: &[Migration] = &[ name: "create Worker mutation source proof replay guard", apply: create_worker_mutation_source_proof_replay_guard, }, + Migration { + version: 30, + name: "create Workspace virtual config source authority", + apply: create_workspace_config_source_authority, + }, ]; struct Migration { @@ -4527,6 +4532,49 @@ fn current_schema_version(conn: &Connection) -> Result { .map_err(Error::from) } +fn create_workspace_config_source_authority(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" + CREATE TABLE workspace_config_trees ( + workspace_id TEXT PRIMARY KEY, + revision INTEGER NOT NULL CHECK (revision >= 0), + tree_digest TEXT NOT NULL, + schema_version INTEGER NOT NULL, + entrypoints_json TEXT NOT NULL, + decodal_version TEXT NOT NULL, + import_policy_version INTEGER NOT NULL, + toolchain_fingerprint TEXT NOT NULL, + projection_digest TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + CREATE TABLE workspace_config_entries ( + workspace_id TEXT NOT NULL, + path TEXT NOT NULL, + content_type TEXT NOT NULL, + content TEXT NOT NULL, + content_digest TEXT NOT NULL, + PRIMARY KEY (workspace_id, path), + FOREIGN KEY (workspace_id) REFERENCES workspace_config_trees(workspace_id) ON DELETE CASCADE + ); + CREATE INDEX idx_workspace_config_entries_prefix + ON workspace_config_entries(workspace_id, path); + CREATE TABLE workspace_config_tree_revisions ( + workspace_id TEXT NOT NULL, + revision INTEGER NOT NULL, + tree_digest TEXT NOT NULL, + toolchain_fingerprint TEXT NOT NULL, + projection_digest TEXT NOT NULL, + manifest_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, revision), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + "#, + )?; + Ok(()) +} + fn create_worker_mutation_source_proof_replay_guard(conn: &Connection) -> Result<()> { conn.execute_batch( r#"