diff --git a/crates/merge-request/src/lib.rs b/crates/merge-request/src/lib.rs index ca8821be..adeebebb 100644 --- a/crates/merge-request/src/lib.rs +++ b/crates/merge-request/src/lib.rs @@ -11,6 +11,13 @@ use uuid::Uuid; const SCHEMA_VERSION: i64 = 12; const PREVIOUS_SCHEMA_VERSION: i64 = 11; const MAX_BODY_BYTES: usize = 16 * 1024; +const DOMAIN_TABLES: [&str; 5] = [ + "merge_requests", + "merge_request_ticket_relations", + "merge_request_thread_events", + "merge_request_review_grants", + "merge_request_reviewer_child_sessions", +]; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -1143,27 +1150,87 @@ fn ticket_event( } pub fn migrate(c: &Connection) -> Result<(), MergeRequestError> { - match schema_version(c)? { - None => fresh(c), - Some(SCHEMA_VERSION) => verify(c), - Some(PREVIOUS_SCHEMA_VERSION) => from_v11(c), - Some(v) => Err(MergeRequestError::Operation(format!( + match schema_state(c)? { + SchemaState::Fresh => fresh(c), + SchemaState::Current(SCHEMA_VERSION) => verify(c), + SchemaState::Current(PREVIOUS_SCHEMA_VERSION) => from_v11(c, PreviousSchemaMarker::Current), + SchemaState::Legacy(PREVIOUS_SCHEMA_VERSION) => from_v11(c, PreviousSchemaMarker::Legacy), + SchemaState::Current(v) => Err(MergeRequestError::Operation(format!( "unsupported schema {v}" ))), + SchemaState::Legacy(v) => Err(MergeRequestError::Operation(format!( + "unsupported legacy schema {v}" + ))), } } -fn schema_version(c: &Connection) -> Result, MergeRequestError> { - let e:bool=c.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema')",[],|r|r.get(0))?; - if !e { - return Ok(None); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SchemaState { + Fresh, + Current(i64), + Legacy(i64), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum PreviousSchemaMarker { + Current, + Legacy, +} + +fn schema_state(c: &Connection) -> Result { + let (current, legacy): (bool, bool) = c.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema'),EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema_migrations')", + [], + |r| Ok((r.get(0)?, r.get(1)?)), + )?; + if current && legacy { + return Err(MergeRequestError::Corrupt( + "both current and legacy schema markers exist".into(), + )); } - c.query_row( - "SELECT version FROM merge_request_schema WHERE singleton=1", + if current { + let (count, singleton, version): (i64, Option, Option) = c.query_row( + "SELECT COUNT(*),MIN(singleton),MAX(version) FROM merge_request_schema", + [], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + )?; + if count != 1 || singleton != Some(1) { + return Err(MergeRequestError::Corrupt( + "current schema marker must contain exactly singleton 1".into(), + )); + } + let version = version.ok_or_else(|| { + MergeRequestError::Corrupt("current schema marker version is null".into()) + })?; + return Ok(SchemaState::Current(version)); + } + if legacy { + let (count, version): (i64, Option) = c.query_row( + "SELECT COUNT(*),MAX(version) FROM merge_request_schema_migrations", + [], + |r| Ok((r.get(0)?, r.get(1)?)), + )?; + if count != 1 { + return Err(MergeRequestError::Corrupt( + "legacy schema marker must contain exactly one version".into(), + )); + } + let version = version.ok_or_else(|| { + MergeRequestError::Corrupt("legacy schema marker version is null".into()) + })?; + return Ok(SchemaState::Legacy(version)); + } + let domain_tables: bool = c.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name GLOB 'merge_request*')", [], |r| r.get(0), - ) - .optional() - .map_err(Into::into) + )?; + if domain_tables { + return Err(MergeRequestError::Corrupt( + "merge request tables exist without a schema marker".into(), + )); + } + Ok(SchemaState::Fresh) } fn fresh(c: &Connection) -> Result<(), MergeRequestError> { let t = c.unchecked_transaction()?; @@ -1180,13 +1247,26 @@ fn tables(t: &Transaction<'_>, marker: bool) -> Result<(), MergeRequestError> { t.execute_batch("CREATE TABLE merge_requests(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,repository_id TEXT NOT NULL,state TEXT NOT NULL CHECK(state IN('open','merged','closed')),selector_from TEXT,selector_to TEXT NOT NULL,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id),FOREIGN KEY(workspace_id,repository_id)REFERENCES repositories(workspace_id,repository_id));CREATE TABLE merge_request_ticket_relations(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,ticket_id TEXT NOT NULL,relation_kind TEXT NOT NULL CHECK(relation_kind='implements'),created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,ticket_id),FOREIGN KEY(workspace_id,merge_request_id)REFERENCES merge_requests(workspace_id,merge_request_id)ON DELETE CASCADE,FOREIGN KEY(workspace_id,ticket_id)REFERENCES typed_tickets(workspace_id,ticket_id)ON DELETE CASCADE);CREATE TABLE merge_request_thread_events(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,event_id TEXT NOT NULL,sequence INTEGER NOT NULL,kind TEXT NOT NULL CHECK(kind IN('review_requested','review','review_revoked','review_cancelled','comment','merge')),payload_json TEXT NOT NULL,operation_id TEXT,created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,event_id),UNIQUE(workspace_id,merge_request_id,sequence),FOREIGN KEY(workspace_id,merge_request_id)REFERENCES merge_requests(workspace_id,merge_request_id)ON DELETE CASCADE);CREATE UNIQUE INDEX merge_request_merge_operations ON merge_request_thread_events(workspace_id,operation_id)WHERE operation_id IS NOT NULL;CREATE TABLE merge_request_review_grants(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,request_event_id TEXT NOT NULL,subject_ref TEXT NOT NULL,reviewer_runtime_id TEXT NOT NULL,reviewer_worker_id TEXT NOT NULL,capability_token TEXT PRIMARY KEY,issued_at TEXT NOT NULL,consumed_at TEXT,revoked_at TEXT,status TEXT NOT NULL CHECK(status IN('issued','consumed','revoked')),FOREIGN KEY(workspace_id,merge_request_id,request_event_id)REFERENCES merge_request_thread_events(workspace_id,merge_request_id,event_id)ON DELETE CASCADE);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT NOT NULL,child_session_id TEXT NOT NULL,parent_runtime_id TEXT NOT NULL,parent_worker_id TEXT NOT NULL,reviewer_profile TEXT NOT NULL,registered_at TEXT NOT NULL,status TEXT NOT NULL CHECK(status IN('active','consumed')),PRIMARY KEY(workspace_id,child_session_id));")?; Ok(()) } -fn from_v11(c: &Connection) -> Result<(), MergeRequestError> { +fn from_v11( + c: &Connection, + previous_marker: PreviousSchemaMarker, +) -> Result<(), MergeRequestError> { let t = c.unchecked_transaction()?; + if previous_marker == PreviousSchemaMarker::Legacy { + t.execute_batch("CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY CHECK(singleton=1),version INTEGER NOT NULL);")?; + t.execute( + "INSERT INTO merge_request_schema VALUES(1,?1)", + params![PREVIOUS_SCHEMA_VERSION], + )?; + } t.execute_batch("ALTER TABLE merge_requests RENAME TO merge_requests_v11;ALTER TABLE merge_request_ticket_relations RENAME TO merge_request_ticket_relations_v11;ALTER TABLE merge_request_revisions RENAME TO merge_request_revisions_v11;ALTER TABLE merge_request_revision_paths RENAME TO merge_request_revision_paths_v11;ALTER TABLE merge_request_reviewer_child_sessions RENAME TO merge_request_reviewer_child_sessions_v11;ALTER TABLE merge_request_review_attempts RENAME TO merge_request_review_attempts_v11;ALTER TABLE merge_request_reviews RENAME TO merge_request_reviews_v11;ALTER TABLE merge_request_review_findings RENAME TO merge_request_review_findings_v11;ALTER TABLE merge_request_completion_operations RENAME TO merge_request_completion_operations_v11;")?; tables(&t, false)?; t.execute("INSERT INTO merge_requests SELECT workspace_id,merge_request_id,repository_id,CASE state WHEN 'draft'THEN'open'ELSE state END,NULL,target_ref_selector,created_at,updated_at FROM merge_requests_v11",[])?; t.execute("INSERT INTO merge_request_ticket_relations SELECT * FROM merge_request_ticket_relations_v11",[])?; migrate_events(&t)?; + if previous_marker == PreviousSchemaMarker::Legacy { + t.execute("DROP TABLE merge_request_schema_migrations", [])?; + } t.execute_batch("DROP TABLE merge_request_review_findings_v11;DROP TABLE merge_request_reviews_v11;DROP TABLE merge_request_review_attempts_v11;DROP TABLE merge_request_reviewer_child_sessions_v11;DROP TABLE merge_request_revision_paths_v11;DROP TABLE merge_request_revisions_v11;DROP TABLE merge_request_completion_operations_v11;DROP TABLE merge_request_ticket_relations_v11;DROP TABLE merge_requests_v11;UPDATE merge_request_schema SET version=12 WHERE singleton=1;")?; fk(&t)?; t.commit()?; @@ -1363,12 +1443,7 @@ fn migrate_events(t: &Transaction<'_>) -> Result<(), MergeRequestError> { Ok(()) } fn verify(c: &Connection) -> Result<(), MergeRequestError> { - for n in [ - "merge_requests", - "merge_request_ticket_relations", - "merge_request_thread_events", - "merge_request_review_grants", - ] { + for n in DOMAIN_TABLES { let e: bool = c.query_row( "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table'AND name=?1)", params![n], @@ -1381,15 +1456,16 @@ fn verify(c: &Connection) -> Result<(), MergeRequestError> { fk(c) } fn fk(c: &Connection) -> Result<(), MergeRequestError> { - let v: Option<(String, i64)> = c - .query_row("PRAGMA foreign_key_check", [], |r| { - Ok((r.get(0)?, r.get(1)?)) - }) - .optional()?; - if let Some((t, r)) = v { - return Err(MergeRequestError::Corrupt(format!( - "foreign key violation {t}:{r}" - ))); + for table in DOMAIN_TABLES { + let sql = format!("PRAGMA foreign_key_check('{table}')"); + let violation: Option<(String, Option, String)> = c + .query_row(&sql, [], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) + .optional()?; + if let Some((child, row, parent)) = violation { + return Err(MergeRequestError::Corrupt(format!( + "foreign key violation in `{child}` row {row:?}, parent `{parent}`" + ))); + } } Ok(()) } diff --git a/crates/merge-request/tests/store.rs b/crates/merge-request/tests/store.rs index 33968cad..1f4ec9bf 100644 --- a/crates/merge-request/tests/store.rs +++ b/crates/merge-request/tests/store.rs @@ -188,10 +188,37 @@ fn review_revocation_invalidates_readiness() { } #[test] -fn v11_migration_preserves_review_events_and_requires_selector_repair() { +fn legacy_v11_migration_preserves_review_events_and_replaces_marker() { let c = Connection::open_in_memory().unwrap(); - c.execute_batch("CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,PRIMARY KEY(workspace_id,ticket_id));INSERT INTO repositories VALUES('W','R');INSERT INTO typed_tickets VALUES('W','T');CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY,version INTEGER);INSERT INTO merge_request_schema VALUES(1,11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');INSERT INTO merge_request_review_attempts VALUES('W','PENDING','MR','T','V',1,'A','runtime','coder','pending-child','builtin:reviewer','pending-token','registered','2026-07-26T12:00:02Z',NULL);").unwrap(); + c.execute_batch("CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,PRIMARY KEY(workspace_id,ticket_id));INSERT INTO repositories VALUES('W','R');INSERT INTO typed_tickets VALUES('W','T');CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);INSERT INTO merge_request_schema_migrations(version) VALUES(11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');INSERT INTO merge_request_review_attempts VALUES('W','PENDING','MR','T','V',1,'A','runtime','coder','pending-child','builtin:reviewer','pending-token','registered','2026-07-26T12:00:02Z',NULL);").unwrap(); + c.execute_batch( + "CREATE TABLE unrelated_parent(left_id TEXT,right_id TEXT,PRIMARY KEY(left_id,right_id));CREATE TABLE unrelated_child(left_id TEXT REFERENCES unrelated_parent(left_id));", + ) + .unwrap(); + let unrelated_mismatch = c + .query_row("PRAGMA foreign_key_check", [], |_| Ok(())) + .unwrap_err(); + assert!( + unrelated_mismatch + .to_string() + .contains("foreign key mismatch") + ); merge_request::migrate(&c).unwrap(); + assert_eq!( + c.query_row("SELECT version FROM merge_request_schema", [], |r| { + r.get::<_, i64>(0) + }) + .unwrap(), + 12 + ); + let legacy_marker: bool = c + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema_migrations')", + [], + |r| r.get(0), + ) + .unwrap(); + assert!(!legacy_marker); let selector: Option = c .query_row("SELECT selector_from FROM merge_requests", [], |r| r.get(0)) .unwrap(); @@ -217,6 +244,35 @@ fn v11_migration_preserves_review_events_and_requires_selector_repair() { assert!(!old); } +#[test] +fn failed_legacy_v11_migration_rolls_back_marker_bridge() { + let c = Connection::open_in_memory().unwrap(); + c.execute_batch( + "CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);INSERT INTO merge_request_schema_migrations(version) VALUES(11);CREATE TABLE merge_requests(merge_request_id TEXT);", + ) + .unwrap(); + + assert!(merge_request::migrate(&c).is_err()); + for table in ["merge_request_schema_migrations", "merge_requests"] { + let exists: bool = c + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1)", + [table], + |r| r.get(0), + ) + .unwrap(); + assert!(exists, "{table} was not rolled back"); + } + let current_marker: bool = c + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema')", + [], + |r| r.get(0), + ) + .unwrap(); + assert!(!current_marker); +} + #[test] fn authority_reads_full_thread_while_public_pages_remain_bounded() { let (_d, store) = fixture();