Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: centralize durable Ticket notification routing

Browse Source
This commit is contained in:
Hare
2026-10-15 02:20:40 +09:00
parent 65ead5fffd
commit 822a084b31
20 changed files with 4219 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ticket/src/assignment_tracking_tests.rs
+1 -1
View File
@@ -408,7 +408,7 @@ fn migration_preserves_original_records_and_only_reconciles_current_fenced_deleg
|r| r.get::<_, i64>(0)
)
.unwrap(),
12
13
);
let value = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
.unwrap()
crates/ticket/src/content_changes.rs
+62
View File
@@ -0,0 +1,62 @@
//! Immutable facts of committed edits, not notification decisions. The caller's write
//! transaction owns both the edit and this INSERT; hosts may capture assignment identity
//! with an INSERT trigger. No text snapshots or processing state are retained here.
use rusqlite::{Connection, params};
use crate::{Result, sqlite_err};
pub(crate) enum ContentChange<'a> {
ContentEdit {
title_changed: bool,
body_event_ref: Option<&'a str>,
targets_changed: bool,
},
CommentEdit {
event_ref: &'a str,
},
}
impl ContentChange<'_> {
pub(crate) fn insert(
self,
conn: &Connection,
workspace_id: &str,
ticket_id: &str,
actor: Option<&str>,
at: &str,
) -> Result<()> {
let (kind, title_changed, source_event_ref, targets_changed) = match self {
Self::ContentEdit {
title_changed,
body_event_ref,
targets_changed,
} => (
"content_edit",
title_changed,
body_event_ref,
targets_changed,
),
Self::CommentEdit { event_ref } => ("comment_edit", false, Some(event_ref), false),
};
conn.execute(
"INSERT INTO typed_ticket_content_changes
(change_id, workspace_id, ticket_id, kind, actor, at,
title_changed, body_changed, targets_changed, source_event_ref)
VALUES (lower(hex(randomblob(16))), ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
workspace_id,
ticket_id,
kind,
actor,
at,
title_changed,
source_event_ref.is_some(),
targets_changed,
source_event_ref
],
)
.map_err(sqlite_err)?;
Ok(())
}
}
crates/ticket/src/content_changes.sql
+20
View File
@@ -0,0 +1,20 @@
-- Ordered component 12 -> 13. Facts have no delivery policy or mutable processing state.
CREATE TABLE typed_ticket_content_changes (
change_id TEXT NOT NULL PRIMARY KEY CHECK (length(trim(change_id)) > 0),
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('content_edit', 'comment_edit')),
actor TEXT CHECK (actor IS NULL OR length(trim(actor)) > 0),
at TEXT NOT NULL,
title_changed INTEGER NOT NULL CHECK (title_changed IN (0, 1)),
body_changed INTEGER NOT NULL CHECK (body_changed IN (0, 1)),
targets_changed INTEGER NOT NULL CHECK (targets_changed IN (0, 1)),
source_event_ref TEXT,
CHECK (title_changed + body_changed + targets_changed > 0),
CHECK ((body_changed = 0 AND source_event_ref IS NULL) OR
(body_changed = 1 AND source_event_ref IS NOT NULL AND length(trim(source_event_ref)) > 0)),
CHECK (kind != 'comment_edit' OR (title_changed = 0 AND body_changed = 1 AND targets_changed = 0)),
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
);
CREATE TRIGGER typed_ticket_content_changes_no_update BEFORE UPDATE ON typed_ticket_content_changes
BEGIN SELECT RAISE(ABORT, 'Ticket content change facts are immutable'); END;
crates/ticket/src/lib.rs
+45 -27
View File
@@ -29,6 +29,7 @@ mod assignment_answer;
pub use assignment_answer::*;
mod assignment_tracking;
pub mod config;
mod content_changes;
mod sqlite_schema;
mod tags;
pub use assignment_tracking::*;
@@ -3116,6 +3117,9 @@ impl TicketBackend for SqliteTicketBackend {
if post.body != edit.expected_body {
return Err(TicketError::Conflict("stale Ticket post body".into()));
}
if post.body == edit.body {
return Ok((previous.clone(), previous));
}
let now = now_utc();
let changed = conn.execute(
"UPDATE typed_ticket_events SET body = ?4, edited_at = ?5, edited_by = ?6 WHERE workspace_id = ?1 AND ticket_id = ?2 AND event_ref = ?3 AND body = ?7 AND kind = 'comment' AND archived = 0",
@@ -3123,6 +3127,16 @@ impl TicketBackend for SqliteTicketBackend {
).map_err(sqlite_err)?;
if changed != 1 { return Err(TicketError::Conflict("stale Ticket post body".into())); }
self.touch_ticket(conn, &ticket_id, &now)?;
let fact = if post.is_initial {
content_changes::ContentChange::ContentEdit {
title_changed: false,
body_event_ref: Some(&post.event_ref),
targets_changed: false,
}
} else {
content_changes::ContentChange::CommentEdit { event_ref: &post.event_ref }
};
fact.insert(conn, &self.workspace_id, &ticket_id, Some(&edit.author), &now)?;
let current = self.load_ticket(conn, &ticket_id)?;
Ok((previous, current))
})
@@ -3315,37 +3329,41 @@ impl TicketBackend for SqliteTicketBackend {
}
let ticket_id = self.resolve_ticket_id(conn, id)?;
let previous = self.load_ticket(conn, &ticket_id)?;
let now = now_utc();
if let Some(title) = edit.title.as_ref() {
conn.execute("UPDATE typed_tickets SET title = ?3, updated_at = ?4 WHERE workspace_id = ?1 AND ticket_id = ?2", params![self.workspace_id, ticket_id, title, now]).map_err(sqlite_err)?;
}
let mut updated_body = None;
if let Some(body) = edit.body.as_ref() {
updated_body = Some(body.clone());
}
let mut updated_body = edit.body.clone();
if let Some(replacement) = edit.body_replacement.as_ref() {
let current_body: String = conn
.query_row(
"SELECT body FROM typed_ticket_events WHERE workspace_id = ?1 AND ticket_id = ?2 AND event_index = -1",
params![self.workspace_id, ticket_id],
|row| row.get(0),
)
.map_err(sqlite_err)?;
let outcome = replacement.apply(current_body.as_str())?;
updated_body = Some(outcome.body);
updated_body = Some(replacement.apply(previous.document.body.as_str())?.body);
}
if let Some(body) = updated_body.as_ref() {
conn.execute("UPDATE typed_ticket_events SET body = ?3, edited_at = ?4, edited_by = ?5 WHERE workspace_id = ?1 AND ticket_id = ?2 AND event_index = -1", params![self.workspace_id, ticket_id, body.as_str(), now, edit.author]).map_err(sqlite_err)?;
self.touch_ticket(conn, &ticket_id, &now)?;
let targets = edit.targets.as_ref().map(|targets| match targets {
TicketTargetsEdit::Set { targets } => targets.as_slice(),
TicketTargetsEdit::Clear => &[],
});
let title_changed = edit.title.as_ref().is_some_and(|title| *title != previous.meta.title);
let body_changed = updated_body.as_ref().is_some_and(|body| *body != previous.document.body);
let targets_changed = targets.is_some_and(|targets| targets != previous.meta.targets);
if !title_changed && !body_changed && !targets_changed {
return Ok((previous.clone(), previous));
}
if let Some(targets) = edit.targets.as_ref() {
let targets = match targets {
TicketTargetsEdit::Set { targets } => targets.as_slice(),
TicketTargetsEdit::Clear => &[],
};
self.replace_ticket_targets(conn, &ticket_id, targets)?;
self.touch_ticket(conn, &ticket_id, &now)?;
let now = now_utc();
if title_changed {
conn.execute("UPDATE typed_tickets SET title = ?3 WHERE workspace_id = ?1 AND ticket_id = ?2", params![self.workspace_id, ticket_id, edit.title]).map_err(sqlite_err)?;
}
let body_event_ref = if body_changed {
let initial = previous.events.iter().find(|event| event.is_initial)
.ok_or_else(|| TicketError::Conflict("Ticket initial post is missing".into()))?;
conn.execute("UPDATE typed_ticket_events SET body = ?3, edited_at = ?4, edited_by = ?5 WHERE workspace_id = ?1 AND ticket_id = ?2 AND event_index = -1", params![self.workspace_id, ticket_id, updated_body.as_ref().expect("changed body").as_str(), now, edit.author]).map_err(sqlite_err)?;
Some(initial.event_ref.as_str())
} else {
None
};
if targets_changed {
self.replace_ticket_targets(conn, &ticket_id, targets.expect("changed targets"))?;
}
self.touch_ticket(conn, &ticket_id, &now)?;
content_changes::ContentChange::ContentEdit {
title_changed,
body_event_ref,
targets_changed,
}.insert(conn, &self.workspace_id, &ticket_id, edit.author.as_deref(), &now)?;
let current = self.load_ticket(conn, &ticket_id)?;
Ok((previous, current))
crates/ticket/src/sqlite_schema.rs
+139 -26
View File
@@ -20,7 +20,7 @@ use crate::{Result, TicketError, sqlite_err};
const MIGRATION_TABLE: &str = "ticket_schema_migrations";
const MAX_SCHEMA_DIAGNOSTICS: usize = 32;
const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 12;
const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 13;
const BASELINE_MIGRATION_NAME: &str = "ticket schema baseline";
#[derive(Clone, Copy)]
@@ -320,11 +320,10 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
}
}
/// Composes retained 6/7, tags-only 8 or cases-only 9 through component 10, then
/// appends the 10→11 assignment and 11→12 post migrations without replacing applied markers.
/// The host owns the surrounding transaction (for example Workspace 91→92).
/// Appends assignment, post, and content-change fact migrations without replacing applied markers.
/// The host owns the surrounding transaction.
/// A savepoint rolls back this component on failure, without committing host work.
/// Current component 12 is verified only; immutable judgments are never reapplied.
/// Current component 13 is verified only; historical edits are never backfilled.
pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> Result<()> {
if connection.is_autocommit() {
return Err(TicketError::Sqlite(
@@ -348,7 +347,7 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
create_latest_ticket_schema(connection)?;
connection
.execute(
"INSERT INTO ticket_schema_migrations VALUES (12, ?1, ?2)",
"INSERT INTO ticket_schema_migrations VALUES (13, ?1, ?2)",
params![BASELINE_MIGRATION_NAME, chrono::Utc::now().to_rfc3339()],
)
.map_err(sqlite_err)?;
@@ -374,7 +373,8 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
TicketSchemaShape::Cases9 => add_tag_schema(connection)?,
TicketSchemaShape::Composed10
| TicketSchemaShape::Tracking11
| TicketSchemaShape::Posts12 => {}
| TicketSchemaShape::Posts12
| TicketSchemaShape::ContentChanges13 => {}
}
if version < 10 {
connection
@@ -392,6 +392,10 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
migrate_posts(connection)?;
connection.execute("INSERT INTO ticket_schema_migrations VALUES (12, 'Canonical Ticket posts', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
}
if version < 13 {
migrate_content_changes(connection)?;
connection.execute("INSERT INTO ticket_schema_migrations VALUES (13, 'Ticket content change facts', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
}
}
verify_sqlite_ticket_schema(connection)
})();
@@ -416,6 +420,7 @@ enum TicketSchemaShape {
Composed10,
Tracking11,
Posts12,
ContentChanges13,
}
impl TicketSchemaShape {
fn from_version(version: i64) -> Self {
@@ -426,6 +431,7 @@ impl TicketSchemaShape {
10 => Self::Composed10,
11 => Self::Tracking11,
12 => Self::Posts12,
13 => Self::ContentChanges13,
_ => unreachable!(),
}
}
@@ -437,25 +443,34 @@ impl TicketSchemaShape {
Self::Composed10 => 10,
Self::Tracking11 => 11,
Self::Posts12 => 12,
Self::ContentChanges13 => 13,
}
}
fn tags(self) -> bool {
matches!(
self,
Self::Tags8 | Self::Composed10 | Self::Tracking11 | Self::Posts12
Self::Tags8
| Self::Composed10
| Self::Tracking11
| Self::Posts12
| Self::ContentChanges13
)
}
fn cases(self) -> bool {
matches!(
self,
Self::Cases9 | Self::Composed10 | Self::Tracking11 | Self::Posts12
Self::Cases9
| Self::Composed10
| Self::Tracking11
| Self::Posts12
| Self::ContentChanges13
)
}
}
/// Verifies the current Ticket-owned SQLite schema without executing DDL.
pub fn verify_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
verify_ticket_schema_shape(connection, TicketSchemaShape::Posts12)
verify_ticket_schema_shape(connection, TicketSchemaShape::ContentChanges13)
}
/// Retained Workspace schema-66 migration produces the pre-classification v7
@@ -569,7 +584,10 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
continue;
}
let mut columns = columns.to_vec();
if matches!(shape, TicketSchemaShape::Posts12) {
if matches!(
shape,
TicketSchemaShape::Posts12 | TicketSchemaShape::ContentChanges13
) {
if table == "typed_tickets" {
columns.retain(|c| c.name != "body");
columns.push(column("created_by", "TEXT", false, 0));
@@ -810,7 +828,9 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
if matches!(
shape,
TicketSchemaShape::Tracking11 | TicketSchemaShape::Posts12
TicketSchemaShape::Tracking11
| TicketSchemaShape::Posts12
| TicketSchemaShape::ContentChanges13
) {
if let Err(error) =
crate::assignment_tracking::verify_assignment_tracking_schema(connection)
@@ -818,7 +838,10 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
push_diagnostic(&mut diagnostics, error.to_string());
}
}
if matches!(shape, TicketSchemaShape::Posts12) {
if matches!(
shape,
TicketSchemaShape::Posts12 | TicketSchemaShape::ContentChanges13
) {
collect_index_diagnostics(
connection,
"typed_ticket_events",
@@ -828,6 +851,9 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
&mut diagnostics,
);
}
if matches!(shape, TicketSchemaShape::ContentChanges13) {
collect_content_change_diagnostics(connection, &mut diagnostics)?;
}
if diagnostics.is_empty() {
Ok(())
} else {
@@ -849,7 +875,90 @@ fn create_latest_ticket_schema(connection: &Connection) -> Result<()> {
.execute_batch(include_str!("latest_schema.sql"))
.map_err(sqlite_err)?;
crate::assignment_tracking::migrate_assignment_tracking(connection)?;
migrate_posts(connection)
migrate_posts(connection)?;
migrate_content_changes(connection)
}
fn migrate_content_changes(connection: &Connection) -> Result<()> {
connection
.execute_batch(include_str!("content_changes.sql"))
.map_err(sqlite_err)
}
fn collect_content_change_diagnostics(
connection: &Connection,
diagnostics: &mut Vec<String>,
) -> Result<()> {
const TABLE: &str = "typed_ticket_content_changes";
collect_table_diagnostics(
connection,
TABLE,
&[
column("change_id", "TEXT", true, 1),
column("workspace_id", "TEXT", true, 0),
column("ticket_id", "TEXT", true, 0),
column("kind", "TEXT", true, 0),
column("actor", "TEXT", false, 0),
column("at", "TEXT", true, 0),
column("title_changed", "INTEGER", true, 0),
column("body_changed", "INTEGER", true, 0),
column("targets_changed", "INTEGER", true, 0),
column("source_event_ref", "TEXT", false, 0),
],
CHILD_FOREIGN_KEYS,
diagnostics,
);
// These new objects are wholly Ticket-owned. Compare their complete definitions,
// including CHECK predicates and the composite FK, not just column affinities.
// Additional host triggers are independent and remain permitted.
let ddl = include_str!("content_changes.sql");
let table_sql = format!(
"CREATE TABLE {}",
ddl.split_once("CREATE TABLE ")
.expect("table DDL")
.1
.split_once(";\n")
.expect("table terminator")
.0
);
let trigger_sql = format!(
"CREATE TRIGGER {}",
ddl.split_once("CREATE TRIGGER ")
.expect("trigger DDL")
.1
.trim()
);
for (kind, name, expected) in [
("table", TABLE, table_sql),
(
"trigger",
"typed_ticket_content_changes_no_update",
trigger_sql,
),
] {
let actual: Option<String> = connection
.query_row(
"SELECT sql FROM sqlite_master WHERE type = ?1 AND name = ?2",
params![kind, name],
|row| row.get(0),
)
.optional()
.map_err(sqlite_err)?;
let normalize = |sql: &str| {
sql.trim()
.trim_end_matches(';')
.split_whitespace()
.collect::<String>()
};
if actual.as_deref().map(normalize) != Some(normalize(&expected)) {
push_diagnostic(
diagnostics,
format!("invalid Ticket content change {kind} {name}"),
);
}
}
collect_foreign_key_check_diagnostics(connection, TABLE, diagnostics);
Ok(())
}
fn migrate_posts(connection: &Connection) -> Result<()> {
@@ -1105,22 +1214,25 @@ fn load_applied_migrations(connection: &Connection) -> Result<BTreeMap<i64, Stri
fn recognized_schema_version(applied: &BTreeMap<i64, String>) -> Result<i64> {
if applied.len() == 1 {
if let Some((&version, name)) = applied.first_key_value() {
if (6..=12).contains(&version) && name == BASELINE_MIGRATION_NAME {
if (6..=13).contains(&version) && name == BASELINE_MIGRATION_NAME {
return Ok(version);
}
}
}
for baseline in [10, 11] {
for baseline in [10, 11, 12] {
let mut expected = BTreeMap::from([(baseline, BASELINE_MIGRATION_NAME.to_owned())]);
if baseline == 10 {
expected.insert(11, "Ticket-owned assignment tracking".to_owned());
for (version, name) in [
(11, "Ticket-owned assignment tracking"),
(12, "Canonical Ticket posts"),
(13, "Ticket content change facts"),
] {
if version <= baseline {
continue;
}
expected.insert(version, name.to_owned());
if applied == &expected {
return Ok(11);
return Ok(version);
}
}
expected.insert(12, "Canonical Ticket posts".to_owned());
if applied == &expected {
return Ok(12);
}
}
Err(TicketError::Sqlite(format!(
@@ -1623,7 +1735,8 @@ mod tests {
BTreeMap::from([
(10, BASELINE_MIGRATION_NAME.to_owned()),
(11, "Ticket-owned assignment tracking".to_owned()),
(12, "Canonical Ticket posts".to_owned())
(12, "Canonical Ticket posts".to_owned()),
(13, "Ticket content change facts".to_owned())
])
);
}
@@ -1766,7 +1879,7 @@ mod tests {
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
assert!(error.to_string().contains(
"migration history must contain a canonical baseline and ordered migrations through 12"
"migration history must contain a canonical baseline and ordered migrations through 13"
));
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 2);
}
@@ -1788,7 +1901,7 @@ mod tests {
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
assert!(error.to_string().contains(
"migration history must contain a canonical baseline and ordered migrations through 12"
"migration history must contain a canonical baseline and ordered migrations through 13"
));
assert!(!table_exists(&connection, "typed_tickets").unwrap());
}
crates/ticket/tests/posts.rs
+7 -1
View File
@@ -1,3 +1,9 @@
//! Real SQLite component tests through TicketBackend, including durable edit facts.
//! Host capture is represented only by test triggers; Workspace routing is not tested here.
#[path = "posts/content_changes.rs"]
mod content_changes;
use rusqlite::{Connection, params};
use tempfile::TempDir;
use ticket::{
@@ -264,7 +270,7 @@ fn schema12_migrates_body_authors_archived_history_and_stable_references_then_re
|r| r.get::<_, i64>(0)
)
.unwrap(),
12
13
);
backend
.add_event("known".into(), comment("Next", "author"))
crates/ticket/tests/posts/content_changes.rs
+609
View File
@@ -0,0 +1,609 @@
use super::*;
use ticket::{TicketBodyReplacement, TicketTarget, TicketTargetAccess, TicketTargetsEdit};
struct Fixture {
_temp: TempDir,
path: std::path::PathBuf,
backend: SqliteTicketBackend,
conn: Connection,
id: String,
}
impl Fixture {
fn new() -> Self {
let temp = TempDir::new().unwrap();
let path = temp.path().join("facts.db");
let backend = SqliteTicketBackend::open(&path, "workspace").unwrap();
let mut input = NewTicket::new("Original title");
input.body = MarkdownText::new("Private original body");
input.author = Some("creator".into());
let id = backend.create(input).unwrap().id;
let conn = Connection::open(&path).unwrap();
conn.execute_batch("PRAGMA foreign_keys=ON").unwrap();
Self {
_temp: temp,
path,
backend,
conn,
id,
}
}
fn ticket(&self) -> ticket::Ticket {
self.backend.show(self.id.clone().into()).unwrap()
}
fn edit(&self, edit: TicketItemEdit) -> ticket::Ticket {
self.backend
.edit_item(self.id.clone().into(), edit)
.unwrap()
}
fn facts(&self) -> Vec<Fact> {
self.conn.prepare("SELECT change_id,workspace_id,ticket_id,kind,actor,at,title_changed,body_changed,targets_changed,source_event_ref FROM typed_ticket_content_changes ORDER BY rowid").unwrap()
.query_map([], |r| Ok(Fact {
id: r.get(0)?, workspace: r.get(1)?, ticket: r.get(2)?, kind: r.get(3)?, actor: r.get(4)?, at: r.get(5)?,
title: r.get(6)?, body: r.get(7)?, targets: r.get(8)?, source: r.get(9)?,
})).unwrap().collect::<rusqlite::Result<_>>().unwrap()
}
}
#[derive(Debug, PartialEq, Eq)]
struct Fact {
id: String,
workspace: String,
ticket: String,
kind: String,
actor: Option<String>,
at: String,
title: bool,
body: bool,
targets: bool,
source: Option<String>,
}
fn targets(names: &[&str]) -> TicketTargetsEdit {
TicketTargetsEdit::Set {
targets: names
.iter()
.map(|name| TicketTarget {
repository_key: (*name).into(),
ref_selector: None,
access: TicketTargetAccess::ReadWrite,
})
.collect(),
}
}
#[test]
fn item_edits_record_actual_fields_and_contemporaneous_actor_once_without_text() {
let f = Fixture::new();
assert!(f.facts().is_empty());
let original = f.ticket();
let edits = [
(
TicketItemEdit {
title: Some("Changed title".into()),
..Default::default()
},
(true, false, false),
),
(
TicketItemEdit {
targets: Some(targets(&["main", "other"])),
..Default::default()
},
(false, false, true),
),
(
TicketItemEdit {
targets: Some(targets(&["other", "main"])),
..Default::default()
},
(false, false, true),
),
(
TicketItemEdit {
title: Some("Combined title".into()),
body: Some(MarkdownText::new("Private replacement body")),
targets: Some(TicketTargetsEdit::Clear),
..Default::default()
},
(true, true, true),
),
(
TicketItemEdit {
title: Some("Combined title".into()),
body_replacement: Some(TicketBodyReplacement {
old_string: "replacement".into(),
new_string: "updated".into(),
replace_all: false,
}),
..Default::default()
},
(false, true, false),
),
];
for (index, (mut edit, flags)) in edits.into_iter().enumerate() {
let actor = format!("editor-{index}");
edit.author = Some(actor.clone());
let current = f.edit(edit);
let facts = f.facts();
assert_eq!(facts.len(), index + 1);
let fact = facts.last().unwrap();
assert_eq!(fact.workspace, "workspace");
assert_eq!(fact.ticket, f.id);
assert_eq!(fact.kind, "content_edit");
assert_eq!(fact.actor.as_deref(), Some(actor.as_str()));
assert_eq!((fact.title, fact.body, fact.targets), flags);
assert_eq!(
fact.source.as_deref(),
flags.1.then_some(original.events[0].event_ref.as_str())
);
assert_eq!(Some(fact.at.as_str()), current.meta.updated_at.as_deref());
assert_eq!(current.events[0].author, original.events[0].author);
}
let facts = f.facts();
assert_eq!(
facts
.iter()
.map(|f| &f.id)
.collect::<std::collections::BTreeSet<_>>()
.len(),
facts.len()
);
// The schema itself excludes content snapshots, not just this particular INSERT.
let columns: Vec<String> = f
.conn
.prepare("SELECT name FROM pragma_table_info('typed_ticket_content_changes')")
.unwrap()
.query_map([], |r| r.get(0))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
assert_eq!(
columns,
[
"change_id",
"workspace_id",
"ticket_id",
"kind",
"actor",
"at",
"title_changed",
"body_changed",
"targets_changed",
"source_event_ref"
]
);
let reopened = SqliteTicketBackend::open(&f.path, "workspace").unwrap();
assert_eq!(reopened.show(f.id.clone().into()).unwrap(), f.ticket());
assert_eq!(f.facts(), facts);
f.edit(TicketItemEdit {
title: Some("Unknown local editor".into()),
..Default::default()
});
assert_eq!(f.facts().last().unwrap().actor, None);
}
#[test]
fn post_edits_distinguish_initial_and_discussion_and_preserve_prior_edit_actors() {
let f = Fixture::new();
f.backend
.add_event(
f.id.clone().into(),
comment("Private discussion", "commenter"),
)
.unwrap();
assert!(f.facts().is_empty());
let before = f.ticket();
for (index, original) in before.events.iter().enumerate() {
let current = f
.backend
.edit_post(
f.id.clone().into(),
TicketPostEdit {
event_ref: original.event_ref.clone(),
expected_body: original.body.clone(),
body: MarkdownText::new(format!("Private edited post {index}")),
author: format!("post-editor-{index}"),
},
)
.unwrap();
let facts = f.facts();
assert_eq!(facts.len(), index + 1);
let fact = facts.last().unwrap();
assert_eq!(
fact.kind,
if original.is_initial {
"content_edit"
} else {
"comment_edit"
}
);
assert_eq!((fact.title, fact.body, fact.targets), (false, true, false));
assert_eq!(fact.source.as_ref(), Some(&original.event_ref));
assert_eq!(
fact.actor.as_deref(),
Some(format!("post-editor-{index}").as_str())
);
assert_eq!(
current.events[index].edited_at.as_deref(),
Some(fact.at.as_str())
);
}
let facts = f.facts();
let post = &f.ticket().events[0];
f.backend
.edit_post(
f.id.clone().into(),
TicketPostEdit {
event_ref: post.event_ref.clone(),
expected_body: post.body.clone(),
body: MarkdownText::new("Another private edit"),
author: "later-editor".into(),
},
)
.unwrap();
assert_eq!(&f.facts()[..2], facts.as_slice());
}
#[test]
fn noops_preserve_metadata_and_do_not_bypass_validation_or_emit_facts() {
let f = Fixture::new();
f.edit(TicketItemEdit {
targets: Some(targets(&["main"])),
author: Some("first".into()),
..Default::default()
});
let before = f.ticket();
let facts = f.facts();
for edit in [
TicketItemEdit {
title: Some(before.meta.title.clone()),
..Default::default()
},
TicketItemEdit {
body: Some(before.document.body.clone()),
..Default::default()
},
TicketItemEdit {
targets: Some(targets(&["main"])),
..Default::default()
},
TicketItemEdit {
title: Some(before.meta.title.clone()),
body: Some(before.document.body.clone()),
targets: Some(targets(&["main"])),
..Default::default()
},
] {
assert_eq!(
f.edit(TicketItemEdit {
author: Some("noop-editor".into()),
..edit
}),
before
);
}
let post = &before.events[0];
let edit = TicketPostEdit {
event_ref: post.event_ref.clone(),
expected_body: post.body.clone(),
body: post.body.clone(),
author: "noop-editor".into(),
};
assert_eq!(
f.backend
.edit_post(f.id.clone().into(), edit.clone())
.unwrap(),
before
);
assert!(
f.backend
.edit_post(
f.id.clone().into(),
TicketPostEdit {
expected_body: MarkdownText::new("stale"),
..edit
}
)
.is_err()
);
assert!(
f.backend
.edit_item(
f.id.clone().into(),
TicketItemEdit {
title: Some(before.meta.title.clone()),
author: Some(" ".into()),
..Default::default()
}
)
.is_err()
);
assert_eq!(f.facts(), facts);
assert_eq!(f.ticket(), before);
f.edit(TicketItemEdit {
targets: Some(TicketTargetsEdit::Clear),
..Default::default()
});
let empty = f.ticket();
let facts = f.facts();
assert_eq!(
f.edit(TicketItemEdit {
targets: Some(TicketTargetsEdit::Clear),
..Default::default()
}),
empty
);
assert_eq!(f.facts(), facts);
}
#[test]
fn failed_host_capture_rolls_back_edits_fact_and_trigger_writes_together() {
let f = Fixture::new();
f.conn.execute_batch("CREATE TABLE host_capture(change_id TEXT, actor TEXT); CREATE TRIGGER capture_fact AFTER INSERT ON typed_ticket_content_changes BEGIN INSERT INTO host_capture VALUES(NEW.change_id, NEW.actor); SELECT RAISE(ABORT, 'capture failed'); END;").unwrap();
let before = f.ticket();
assert!(
f.backend
.edit_item(
f.id.clone().into(),
TicketItemEdit {
title: Some("Changed title".into()),
body: Some(MarkdownText::new("Changed body")),
targets: Some(targets(&["main"])),
author: Some("item-editor".into()),
..Default::default()
}
)
.is_err()
);
let post = &before.events[0];
assert!(
f.backend
.edit_post(
f.id.clone().into(),
TicketPostEdit {
event_ref: post.event_ref.clone(),
expected_body: post.body.clone(),
body: MarkdownText::new("Changed post"),
author: "post-editor".into(),
}
)
.is_err()
);
assert_eq!(f.ticket(), before);
assert!(f.facts().is_empty());
assert_eq!(
f.conn
.query_row("SELECT count(*) FROM host_capture", [], |r| r
.get::<_, i64>(0))
.unwrap(),
0
);
f.conn.execute_batch("DROP TRIGGER capture_fact; CREATE TRIGGER capture_fact AFTER INSERT ON typed_ticket_content_changes BEGIN INSERT INTO host_capture VALUES(NEW.change_id, NEW.actor); END;").unwrap();
f.edit(TicketItemEdit {
title: Some("Successful title".into()),
author: Some("actual-editor".into()),
..Default::default()
});
let fact = f.facts().remove(0);
assert_eq!(
f.conn
.query_row("SELECT change_id, actor FROM host_capture", [], |r| Ok((
r.get::<_, String>(0)?,
r.get::<_, String>(1)?
)))
.unwrap(),
(fact.id, "actual-editor".into())
);
}
fn schema12(conn: &Connection) {
super::schema11(conn);
super::seed_legacy(conn, "retained");
conn.execute_batch(include_str!("../../src/post_migration.sql"))
.unwrap();
conn.execute_batch("INSERT INTO ticket_schema_migrations VALUES(12,'Canonical Ticket posts','original-post-marker');").unwrap();
}
fn schema(conn: &Connection) -> Vec<(String, String, String)> {
conn.prepare("SELECT type,name,sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY type,name")
.unwrap()
.query_map([], |r| {
Ok((
r.get(0)?,
r.get(1)?,
r.get::<_, String>(2)?
.split_whitespace()
.collect::<String>(),
))
})
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap()
}
#[test]
fn content_change_upgrade_matches_fresh_schema_without_backfill_and_preserves_markers() {
let upgraded = Connection::open_in_memory().unwrap();
upgraded
.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);")
.unwrap();
schema12(&upgraded);
migrate_sqlite_ticket_schema(&upgraded).unwrap();
let fresh = Connection::open_in_memory().unwrap();
fresh
.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);")
.unwrap();
migrate_sqlite_ticket_schema(&fresh).unwrap();
let upgraded_schema = schema(&upgraded);
let fresh_schema = schema(&fresh);
assert_eq!(upgraded_schema.len(), fresh_schema.len());
for (upgraded_object, fresh_object) in upgraded_schema.iter().zip(&fresh_schema) {
assert_eq!(upgraded_object, fresh_object);
}
assert_eq!(
upgraded
.query_row(
"SELECT count(*) FROM typed_ticket_content_changes",
[],
|r| r.get::<_, i64>(0)
)
.unwrap(),
0
);
assert_eq!(
upgraded
.query_row(
"SELECT body FROM typed_ticket_events WHERE event_index=-1",
[],
|r| r.get::<_, String>(0)
)
.unwrap(),
"Latest body"
);
assert_eq!(
upgraded
.query_row(
"SELECT applied_at FROM ticket_schema_migrations WHERE version=12",
[],
|r| r.get::<_, String>(0)
)
.unwrap(),
"original-post-marker"
);
let markers: Vec<(i64, String, String)> = upgraded
.prepare("SELECT version,name,applied_at FROM ticket_schema_migrations ORDER BY version")
.unwrap()
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
migrate_sqlite_ticket_schema(&upgraded).unwrap();
let after: Vec<(i64, String, String)> = upgraded
.prepare("SELECT version,name,applied_at FROM ticket_schema_migrations ORDER BY version")
.unwrap()
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
assert_eq!(markers, after);
}
#[test]
fn failed_content_change_migration_rolls_back_ddl_and_marker_without_committing_host() {
let conn = Connection::open_in_memory().unwrap();
schema12(&conn);
conn.execute_batch("CREATE TABLE host_work(value TEXT); CREATE TRIGGER typed_ticket_content_changes_no_update BEFORE UPDATE ON host_work BEGIN SELECT RAISE(ABORT, 'collision'); END;").unwrap();
let before = schema(&conn);
conn.execute_batch("BEGIN IMMEDIATE; INSERT INTO host_work VALUES('pending');")
.unwrap();
assert!(migrate_sqlite_ticket_schema_in_transaction(&conn).is_err());
assert!(!conn.is_autocommit());
assert_eq!(schema(&conn), before);
assert_eq!(
conn.query_row(
"SELECT max(version) FROM ticket_schema_migrations",
[],
|r| r.get::<_, i64>(0)
)
.unwrap(),
12
);
assert_eq!(
conn.query_row("SELECT count(*) FROM host_work", [], |r| r.get::<_, i64>(0))
.unwrap(),
1
);
conn.execute_batch("ROLLBACK; DROP TRIGGER typed_ticket_content_changes_no_update;")
.unwrap();
migrate_sqlite_ticket_schema(&conn).unwrap();
verify_sqlite_ticket_schema(&conn).unwrap();
}
#[test]
fn fact_constraints_reject_invalid_shapes_updates_and_cross_workspace_parent() {
let f = Fixture::new();
let insert = |id: &str,
workspace: &str,
kind: &str,
title: i64,
body: i64,
targets: i64,
source: Option<&str>| {
f.conn.execute("INSERT INTO typed_ticket_content_changes VALUES(?1,?2,?3,?4,'actor','time',?5,?6,?7,?8)", params![id, workspace, f.id, kind, title, body, targets, source])
};
for (kind, title, body, targets, source) in [
("unknown", 1, 0, 0, None),
("content_edit", 0, 0, 0, None),
("content_edit", 2, 0, 0, None),
("content_edit", 0, 2, 0, Some("post")),
("content_edit", 0, 0, 2, None),
("content_edit", 1, 0, 0, Some("post")),
("content_edit", 0, 1, 0, None),
("content_edit", 0, 1, 0, Some(" ")),
("comment_edit", 1, 1, 0, Some("post")),
("comment_edit", 0, 1, 1, Some("post")),
] {
assert!(insert("invalid", "workspace", kind, title, body, targets, source).is_err());
}
assert!(insert("foreign", "other-workspace", "content_edit", 1, 0, 0, None).is_err());
insert("valid", "workspace", "content_edit", 1, 0, 0, None).unwrap();
assert!(insert("valid", "workspace", "content_edit", 1, 0, 0, None).is_err());
assert!(
f.conn
.execute(
"UPDATE typed_ticket_content_changes SET actor='changed'",
[]
)
.is_err()
);
f.conn
.execute("DELETE FROM typed_ticket_content_changes", [])
.unwrap();
assert!(f.facts().is_empty());
}
#[test]
fn verifier_rejects_weakened_fact_constraints_trigger_and_migration_history() {
for (from, to) in [
("CHECK (kind IN ('content_edit', 'comment_edit'))", ""),
("CHECK (title_changed IN (0, 1))", ""),
("CHECK (body_changed IN (0, 1))", ""),
("CHECK (targets_changed IN (0, 1))", ""),
(
"CHECK (title_changed + body_changed + targets_changed > 0),",
"",
),
(
"REFERENCES typed_tickets(workspace_id, ticket_id)",
"REFERENCES typed_tickets(ticket_id, workspace_id)",
),
("ON DELETE CASCADE", "ON DELETE RESTRICT"),
(
"SELECT RAISE(ABORT, 'Ticket content change facts are immutable')",
"SELECT 1",
),
] {
let f = Fixture::new();
f.conn
.execute_batch("DROP TABLE typed_ticket_content_changes")
.unwrap();
let sql = include_str!("../../src/content_changes.sql").replace(from, to);
f.conn.execute_batch(&sql).unwrap();
assert!(verify_sqlite_ticket_schema(&f.conn).is_err(), "{from}");
assert!(migrate_sqlite_ticket_schema(&f.conn).is_err(), "{from}");
}
for mutation in [
"UPDATE ticket_schema_migrations SET name='unrecognized' WHERE version=13",
"INSERT INTO ticket_schema_migrations VALUES(12,'Canonical Ticket posts','invalid extra history')",
"UPDATE ticket_schema_migrations SET version=14 WHERE version=13",
] {
let f = Fixture::new();
f.conn.execute_batch(mutation).unwrap();
assert!(verify_sqlite_ticket_schema(&f.conn).is_err());
assert!(migrate_sqlite_ticket_schema(&f.conn).is_err());
}
}