Skip to content

refactor(db): extract message action migration #269

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Andiveli merged 1 commit into from
Aug 26, 2026
Merged

refactor(db): extract message action migration #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 1 addition & 82 deletions src/db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
use std::{
collections::{HashMap, HashSet},
ops::Range,
path::{Path, PathBuf},
sync::{Arc, LazyLock, Mutex},
Expand Down Expand Up @@ -662,8 +661,7 @@ impl DatabaseHandler {

pub fn init(&self) {
let _write_lock = DATABASE_WRITE_LOCK.lock().unwrap();
schema::initialize(&self.db);
self.migrate_message_action_columns();
schema::initialize(&self.db, DELETED_MESSAGE_TEXT);
}

pub fn set_last_read_cursor(
Expand Down Expand Up @@ -721,85 +719,6 @@ impl DatabaseHandler {
.query_map([], |row| Ok((row.get::<_, String>(0)?.into(), row.get(1)?)))?
.collect()
}

/// Migrates the `message_actions` table to the current schema. Any
/// stored replacement body is folded into the message row first, then
/// the now-unused column is removed. Runs under the write lock from
/// `init`, before any handler reads the actions. The steps are
/// idempotent: an interrupted migration resumes safely on the next open.
fn migrate_message_action_columns(&self) {
let has_replacement = self
.db
.prepare(
"SELECT COUNT(*) FROM pragma_table_info('message_actions') WHERE name = 'replacement'",
)
.unwrap()
.query_row([], |row| row.get::<_, i64>(0))
.unwrap()
> 0;
if !has_replacement {
return;
}
let mut effective_bodies: HashMap<String, String> = HashMap::new();
let mut deleted_targets: HashSet<String> = HashSet::new();
{
let mut query = self
.db
.prepare(
"SELECT target_message_id, kind, replacement FROM message_actions ORDER BY occurred_at, arrival_order, action_id",
)
.unwrap();
let rows = query
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, Option<String>>(2)?,
))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
for (target, kind, replacement) in rows {
// kind 0 is Edit; the last edit in the stable order wins.
if kind == 1 {
deleted_targets.insert(target.clone());
} else if kind == 0
&& let Some(body) = replacement
{
effective_bodies.insert(target, body);
}
}
}
for (target, body) in &effective_bodies {
if deleted_targets.contains(target) {
continue;
}
self.db
.execute(
"UPDATE text_messages SET message = ?1 WHERE id = ?2",
rusqlite::params![body, target],
)
.unwrap();
}
for target in deleted_targets {
self.db
.execute(
"UPDATE text_messages SET message = ?1, quote_id = NULL WHERE id = ?2",
rusqlite::params![DELETED_MESSAGE_TEXT, target],
)
.unwrap();
self.db
.execute(
"DELETE FROM file_messages WHERE id = ?1",
rusqlite::params![target],
)
.unwrap();
}
self.db
.execute("ALTER TABLE message_actions DROP COLUMN replacement", [])
.unwrap();
}
}

impl DatabaseHandler {
Expand Down
122 changes: 118 additions & 4 deletions src/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
use std::path::Path;
use std::time::Duration;
use std::{
collections::{HashMap, HashSet},
path::Path,
};

use rusqlite::Connection;
use strum::IntoEnumIterator;
Expand Down Expand Up @@ -87,7 +90,7 @@ pub(crate) fn prepare_legacy_forwarding_schema(db: &Connection) {

/// Creates the complete message schema and applies its legacy migrations.
/// Callers hold the process-wide database write lock.
pub(crate) fn initialize(db: &Connection) {
pub(crate) fn initialize(db: &Connection, deleted_message_text: &str) {
db.execute(
"CREATE TABLE IF NOT EXISTS chats (jid TEXT PRIMARY KEY)",
[],
Expand Down Expand Up @@ -143,6 +146,7 @@ pub(crate) fn initialize(db: &Connection) {
)
.unwrap();
prepare_legacy_message_schema(db);
migrate_message_action_columns(db, deleted_message_text);
}

fn ensure_read_receipt_schema(db: &Connection) -> rusqlite::Result<()> {
Expand Down Expand Up @@ -205,15 +209,75 @@ fn ensure_mention_columns(db: &Connection) {
}
}

/// Migrates legacy action replacement bodies into message rows, then removes
/// the obsolete action column. The caller holds the database write lock.
fn migrate_message_action_columns(db: &Connection, deleted_message_text: &str) {
let has_replacement = db
.prepare(
"SELECT COUNT(*) FROM pragma_table_info('message_actions') WHERE name = 'replacement'",
)
.unwrap()
.query_row([], |row| row.get::<_, i64>(0))
.unwrap()
> 0;
if !has_replacement {
return;
}
let mut effective_bodies = HashMap::new();
let mut deleted_targets = HashSet::new();
let rows = db
.prepare("SELECT target_message_id, kind, replacement FROM message_actions ORDER BY occurred_at, arrival_order, action_id")
.unwrap()
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?, row.get::<_, Option<String>>(2)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
for (target, kind, replacement) in rows {
// kind 0 is Edit; the last edit in stable order wins.
if kind == 1 {
deleted_targets.insert(target);
} else if kind == 0
&& let Some(body) = replacement
{
effective_bodies.insert(target, body);
}
}
for (target, body) in &effective_bodies {
if !deleted_targets.contains(target) {
db.execute(
"UPDATE text_messages SET message = ?1 WHERE id = ?2",
rusqlite::params![body, target],
)
.unwrap();
}
}
for target in deleted_targets {
db.execute(
"UPDATE text_messages SET message = ?1, quote_id = NULL WHERE id = ?2",
rusqlite::params![deleted_message_text, target],
)
.unwrap();
db.execute(
"DELETE FROM file_messages WHERE id = ?1",
rusqlite::params![target],
)
.unwrap();
}
db.execute("ALTER TABLE message_actions DROP COLUMN replacement", [])
.unwrap();
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn message_schema_initialization_is_idempotent() {
let db = Connection::open_in_memory().unwrap();
initialize(&db);
initialize(&db);
initialize(&db, "[deleted]");
initialize(&db, "[deleted]");
assert_eq!(columns(&db, "text_messages").len(), 12);
assert_eq!(columns(&db, "file_messages").len(), 15);
}
Expand All @@ -233,4 +297,54 @@ mod tests {
assert!(columns(&db, "text_messages").contains(&"mentions_self".into()));
assert!(columns(&db, "file_messages").contains(&"forwarding_score".into()));
}

#[test]
fn legacy_action_replacements_migrate_in_order_and_are_idempotent() {
let db = Connection::open_in_memory().unwrap();
prepare(&db);
db.execute_batch(
"CREATE TABLE text_messages (id TEXT PRIMARY KEY, chat_jid TEXT, sender_jid TEXT, timestamp INTEGER, quote_id TEXT, is_from_me INTEGER, read INTEGER, message TEXT);
CREATE TABLE file_messages (id TEXT PRIMARY KEY, chat_jid TEXT, sender_jid TEXT, timestamp INTEGER, quote_id TEXT, is_from_me INTEGER, read INTEGER, kind INTEGER, path TEXT, file_id TEXT, caption TEXT);
INSERT INTO text_messages VALUES ('target', 'chat', 'sender', 1, 'quote', 0, 0, 'old');
ALTER TABLE message_actions ADD COLUMN replacement TEXT;
INSERT INTO message_actions (action_id, target_message_id, chat_jid, sender_jid, kind, replacement, occurred_at, arrival_order) VALUES ('edit-1', 'target', 'chat', 'sender', 0, 'first', 1, 1);
INSERT INTO message_actions (action_id, target_message_id, chat_jid, sender_jid, kind, replacement, occurred_at, arrival_order) VALUES ('edit-2', 'target', 'chat', 'sender', 0, 'last', 2, 1);
INSERT INTO message_actions (action_id, target_message_id, chat_jid, sender_jid, kind, replacement, occurred_at, arrival_order) VALUES ('delete-1', 'target', 'chat', 'sender', 1, NULL, 3, 1);",
)
.unwrap();

initialize(&db, "[deleted]");
initialize(&db, "[deleted]");

assert_eq!(
db.query_row(
"SELECT message FROM text_messages WHERE id = 'target'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"[deleted]"
);
assert_eq!(
db.query_row(
"SELECT message, quote_id FROM text_messages WHERE id = 'target'",
[],
|row| { Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?)) }
)
.unwrap(),
("[deleted]".to_owned(), None)
);
// Persisted edits intentionally load as empty replacement markers; the
// migrated message row is the source of the displayed body.
assert_eq!(
db.query_row(
"SELECT group_concat(kind, ',') FROM (SELECT kind FROM message_actions WHERE target_message_id = 'target' ORDER BY occurred_at, arrival_order, action_id)",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"0,0,1"
);
assert!(!columns(&db, "message_actions").contains(&"replacement".into()));
}
}
Loading