wechat_profile/

writer.rs

//! Narrow incoming plain-text projection into an initialized official profile.
//! Callers must persist an application intent before entry and recover it before
//! advancing a native sync checkpoint. Message/session transactions are separate;
//! retries reconcile the same server identity rather than resend anything.
use crate::{
    keys::verify_first_page,
    schema::{inspect, quote_identifier},
    snapshot::configure_cipher,
    KeyFile, ProfileError, ProfileLocation, ReadOptions, Result, SourceMessageId,
};
use md5::{Digest, Md5};
use rusqlite::{params, Connection, OptionalExtension};
use serde::{Deserialize, Serialize};
use std::{fs, io::Read, path::Path, time::Duration};

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct IncomingText {
    pub peer: String,
    pub own_username: String,
    pub server_message_id: String,
    pub server_sequence: i64,
    pub create_time: i64,
    pub content: String,
    pub source_xml: String,
    pub unread: bool,
}
#[derive(Clone, Debug)]
pub struct WriteOptions {
    pub exclusive_ownership_confirmed: bool,
    pub allow_plaintext: bool,
    /// Reference to the caller's already durable apply intent, never a SyncKey.
    pub durable_intent_id: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AppliedText {
    pub source_id: SourceMessageId,
    pub message_inserted: bool,
    pub session_projected: bool,
    pub cross_database_atomic: bool,
}

pub(crate) fn open_write(
    location: &ProfileLocation,
    rel: &str,
    keys: &KeyFile,
    allow_plaintext: bool,
) -> Result<Connection> {
    if !crate::keys::valid_relative_path(rel) {
        return Err(ProfileError::InvalidQuery);
    }
    let path = location.storage.join(rel);
    let canonical = fs::canonicalize(&path)?;
    if !canonical.starts_with(&location.storage) || canonical != path {
        return Err(ProfileError::UnsupportedFile);
    }
    let m = fs::symlink_metadata(&path)?;
    if !m.is_file() || wechat_platform::is_link(&m) {
        return Err(ProfileError::UnsupportedFile);
    }
    let mut page = [0u8; 4096];
    fs::File::open(&path)?.read_exact(&mut page)?;
    let c = Connection::open_with_flags(
        &path,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )?;
    c.busy_timeout(Duration::from_secs(2))?;
    if &page[..16] == b"SQLite format 3\0" {
        if !allow_plaintext {
            return Err(ProfileError::KeyUnavailable(rel.into()));
        }
    } else {
        let k = keys
            .databases
            .get(rel)
            .ok_or_else(|| ProfileError::KeyUnavailable(rel.into()))?;
        if !verify_first_page(&page, k.raw_page_key.expose(), &k.salt) {
            return Err(ProfileError::KeyMismatch(rel.into()));
        }
        configure_cipher(&c, &k.raw_page_key, rel)?;
    }
    c.query_row("SELECT count(*) FROM sqlite_master", [], |r| {
        r.get::<_, i64>(0)
    })
    .map_err(|_| ProfileError::KeyMismatch(rel.into()))?;
    c.execute_batch(
        "PRAGMA trusted_schema=OFF; PRAGMA synchronous=FULL; PRAGMA temp_store=MEMORY;",
    )?;
    Ok(c)
}
pub(crate) const MESSAGE_COLUMNS: &[&str] = &[
    "local_id",
    "server_id",
    "local_type",
    "sort_seq",
    "real_sender_id",
    "create_time",
    "status",
    "upload_status",
    "download_status",
    "server_seq",
    "origin_source",
    "source",
    "message_content",
    "compress_content",
    "packed_info_data",
    "WCDB_CT_message_content",
    "WCDB_CT_source",
];
const SESSION_COLUMNS: &[&str] = &[
    "username",
    "type",
    "unread_count",
    "unread_first_msg_srv_id",
    "unread_first_pat_msg_local_id",
    "unread_first_pat_msg_sort_seq",
    "is_hidden",
    "summary",
    "draft",
    "status",
    "last_timestamp",
    "sort_timestamp",
    "last_clear_unread_timestamp",
    "last_msg_locald_id",
    "last_msg_type",
    "last_msg_sub_type",
    "last_msg_sender",
    "last_sender_display_name",
    "last_msg_ext_type",
];
pub(crate) fn exact_columns(
    c: &Connection,
    db: &str,
    table: &str,
    expected: &[&str],
) -> Result<crate::TableSchema> {
    let report = inspect(c, db)?;
    let t = report
        .tables
        .into_iter()
        .find(|t| t.name == table)
        .ok_or_else(|| ProfileError::UnsupportedSchema(db.into()))?;
    let names = t
        .columns
        .iter()
        .map(|v| v.name.as_str())
        .collect::<Vec<_>>();
    let named_columns =
        names.len() == expected.len() && names.iter().all(|name| expected.contains(name));
    if !named_columns || !t.triggers.is_empty() {
        return Err(ProfileError::UnsupportedSchema(db.into()));
    }
    for col in &t.columns {
        let ty = if [
            "username",
            "user_name",
            "source",
            "message_content",
            "compress_content",
            "summary",
            "draft",
            "last_msg_sender",
            "last_sender_display_name",
        ]
        .contains(&col.name.as_str())
        {
            "TEXT"
        } else if col.name == "packed_info_data" {
            "BLOB"
        } else {
            "INTEGER"
        };
        let pk = if col.name == "local_id" && crate::schema::message_table(table)
            || col.name == "user_name" && table == "Name2Id"
            || col.name == "username" && table == "SessionTable"
            || col.name == "username_id" && table == "SessionUnreadListTable_1"
        {
            1
        } else if col.name == "server_id" && table == "SessionUnreadListTable_1" {
            2
        } else {
            0
        };
        let default = if col.name.starts_with("WCDB_CT_") {
            Some("NULL")
        } else {
            None
        };
        if !col.declared_type.eq_ignore_ascii_case(ty)
            || col.primary_key_position != pk
            || col.not_null
            || col.hidden != 0
            || col
                .default_sql
                .as_deref()
                .map(str::to_ascii_uppercase)
                .as_deref()
                != default
        {
            return Err(ProfileError::UnsupportedSchema(db.into()));
        }
    }
    Ok(t)
}
fn var(mut n: u64, out: &mut Vec<u8>) {
    while n >= 128 {
        out.push((n as u8 & 127) | 128);
        n >>= 7;
    }
    out.push(n as u8);
}
fn packed(from: i64, to: i64) -> Result<Vec<u8>> {
    if from <= 0 || to <= 0 {
        return Err(ProfileError::InvalidMetadata);
    }
    let mut v = vec![8];
    var(from as u64, &mut v);
    v.push(16);
    var(to as u64, &mut v);
    v.extend([88, 0]);
    Ok(v)
}
pub(crate) fn find_database(location: &ProfileLocation, basename: &str) -> Result<String> {
    fn walk(
        root: &Path,
        here: &Path,
        name: &str,
        out: &mut Vec<String>,
        depth: usize,
    ) -> Result<()> {
        if depth > 16 {
            return Err(ProfileError::ResourceLimit);
        }
        for e in fs::read_dir(here)? {
            let e = e?;
            let m = fs::symlink_metadata(e.path())?;
            if wechat_platform::is_link(&m) {
                return Err(ProfileError::UnsupportedFile);
            }
            if m.is_dir() {
                walk(root, &e.path(), name, out, depth + 1)?;
            } else if e.file_name() == name {
                out.push(
                    e.path()
                        .strip_prefix(root)
                        .map_err(|_| ProfileError::InvalidProfile)?
                        .to_str()
                        .ok_or(ProfileError::InvalidProfile)?
                        .replace('\\', "/"),
                );
            }
        }
        Ok(())
    }
    let mut found = Vec::new();
    walk(
        &location.storage,
        &location.storage,
        basename,
        &mut found,
        0,
    )?;
    if found.len() != 1 {
        return Err(ProfileError::UnsupportedSchema(basename.into()));
    }
    Ok(found.remove(0))
}
/// Refuse implicit rotation/routing when another ordinary shard is present.
pub(crate) fn require_single_message_shard(location: &ProfileLocation) -> Result<()> {
    fn visit(path: &Path, depth: usize, count: &mut usize, entries: &mut usize) -> Result<()> {
        if depth > 16 {
            return Err(ProfileError::ResourceLimit);
        }
        for entry in fs::read_dir(path)? {
            let entry = entry?;
            *entries += 1;
            if *entries > 16384 {
                return Err(ProfileError::ResourceLimit);
            }
            let metadata = fs::symlink_metadata(entry.path())?;
            if wechat_platform::is_link(&metadata) {
                return Err(ProfileError::UnsupportedFile);
            }
            if metadata.is_dir() {
                visit(&entry.path(), depth + 1, count, entries)?;
            } else if let Some(name) = entry.file_name().to_str() {
                if name
                    .strip_prefix("message_")
                    .and_then(|s| s.strip_suffix(".db"))
                    .is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()))
                {
                    *count += 1;
                }
            }
        }
        Ok(())
    }
    let (mut count, mut entries) = (0, 0);
    visit(&location.storage, 0, &mut count, &mut entries)?;
    if count == 1 {
        Ok(())
    } else {
        Err(ProfileError::WritesUnsupported)
    }
}
/// Supports only the verified existing direct-chat incoming text family. It does
/// not allocate new chats/maps, infer sort collisions, update FTS or project
/// unknown media/system commands. A partial error may follow a committed message
/// transaction; recover by retrying the same durable input.
pub fn apply_incoming_text(
    location: &ProfileLocation,
    keys: &KeyFile,
    input: &IncomingText,
    options: &WriteOptions,
) -> Result<AppliedText> {
    if !options.exclusive_ownership_confirmed {
        return Err(ProfileError::SnapshotRequired);
    }
    if options.durable_intent_id.is_empty() || options.durable_intent_id.len() > 256 {
        return Err(ProfileError::InvalidQuery);
    }
    if keys.account_binding != location.account {
        return Err(ProfileError::AccountMismatch);
    }
    if input.peer.is_empty()
        || input.peer == input.own_username
        || input.peer.ends_with("@chatroom")
        || input.peer.len() > 512
        || input.own_username.is_empty()
        || input.own_username.len() > 512
        || input.content.len() > 1024 * 1024
        || input.source_xml.len() > 1024 * 1024
        || input.content.contains('\0')
        || input.source_xml.contains('\0')
        || input.create_time <= 0
        || input.server_sequence < 0
    {
        return Err(ProfileError::WritesUnsupported);
    }
    let sid = input
        .server_message_id
        .parse::<u64>()
        .map_err(|_| ProfileError::InvalidQuery)?;
    if sid == 0 {
        return Err(ProfileError::InvalidQuery);
    }
    let signed = sid as i64;
    let sort = input
        .create_time
        .checked_mul(1000)
        .ok_or(ProfileError::InvalidQuery)?;
    let message_db = find_database(location, "message_0.db")?;
    let session_db = find_database(location, "session.db")?;
    // Other shards make write routing unknown; do not silently always use zero.
    let view = crate::ProfileReader::open(
        location.clone(),
        keys.clone(),
        ReadOptions {
            stopped_or_consistent_snapshot: true,
            allow_plaintext: options.allow_plaintext,
            ..Default::default()
        },
    )?;
    if view
        .inventory()
        .iter()
        .filter(|d| {
            let n = d.relative_path.rsplit('/').next().unwrap_or("");
            n.starts_with("message_")
                && n.trim_start_matches("message_")
                    .strip_suffix(".db")
                    .is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()))
        })
        .count()
        != 1
    {
        return Err(ProfileError::WritesUnsupported);
    }
    drop(view);
    let mut messages = open_write(location, &message_db, keys, options.allow_plaintext)?;
    let mut sessions = open_write(location, &session_db, keys, options.allow_plaintext)?;
    let table = format!("Msg_{:x}", Md5::digest(input.peer.as_bytes()));
    let table_schema = exact_columns(&messages, &message_db, &table, MESSAGE_COLUMNS)?;
    let expected_indexes = [
        ("_SERVERID", vec!["server_id"]),
        ("_SORTSEQ", vec!["sort_seq"]),
        ("_TYPE_SEQ", vec!["local_type", "sort_seq"]),
        ("_SENDERID", vec!["real_sender_id"]),
    ];
    if table_schema.indexes.len() != 4
        || !expected_indexes.iter().all(|(suffix, columns)| {
            table_schema.indexes.iter().any(|i| {
                i.name == format!("{table}{suffix}")
                    && !i.unique
                    && !i.partial
                    && i.columns.iter().map(|s| s.as_deref()).collect::<Vec<_>>()
                        == columns.iter().map(|s| Some(*s)).collect::<Vec<_>>()
            })
        })
    {
        return Err(ProfileError::UnsupportedSchema(message_db));
    }
    let local_id = table_schema
        .columns
        .iter()
        .find(|column| column.name == "local_id")
        .ok_or_else(|| ProfileError::UnsupportedSchema(message_db.clone()))?;
    if local_id.declared_type.to_uppercase() != "INTEGER"
        || local_id.primary_key_position != 1
        || !table_schema
            .definition
            .as_ref()
            .is_some_and(|s| s.to_uppercase().contains("AUTOINCREMENT"))
    {
        return Err(ProfileError::UnsupportedSchema(message_db));
    }
    exact_columns(&sessions, &session_db, "SessionTable", SESSION_COLUMNS)?;
    exact_columns(
        &sessions,
        &session_db,
        "SessionUnreadListTable_1",
        &["username_id", "server_id", "local_id", "create_time"],
    )?;
    exact_columns(
        &messages,
        &message_db,
        "Name2Id",
        &["user_name", "is_session"],
    )?;
    exact_columns(&sessions, &session_db, "Name2Id", &["user_name"])?;
    let sender = messages
        .query_row(
            "SELECT rowid FROM Name2Id WHERE user_name=?1",
            [&input.peer],
            |r| r.get::<_, i64>(0),
        )
        .optional()?
        .ok_or(ProfileError::WritesUnsupported)?;
    let recipient = messages
        .query_row(
            "SELECT rowid FROM Name2Id WHERE user_name=?1",
            [&input.own_username],
            |r| r.get::<_, i64>(0),
        )
        .optional()?
        .ok_or(ProfileError::WritesUnsupported)?;
    let session_name = sessions
        .query_row(
            "SELECT rowid FROM Name2Id WHERE user_name=?1",
            [&input.peer],
            |r| r.get::<_, i64>(0),
        )
        .optional()?
        .ok_or(ProfileError::WritesUnsupported)?;
    let current=sessions.query_row("SELECT unread_count,last_msg_sender,last_sender_display_name,last_timestamp,sort_timestamp FROM SessionTable WHERE username=?1",[&input.peer],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,i64>(3)?,r.get::<_,i64>(4)?))).optional()?.ok_or(ProfileError::WritesUnsupported)?;
    if current.0 < 0 || !current.1.is_empty() || !current.2.is_empty() {
        return Err(ProfileError::WritesUnsupported);
    }
    let packed = packed(sender, recipient)?;
    let source = zstd::stream::encode_all(input.source_xml.as_bytes(), 3)
        .map_err(|_| ProfileError::InvalidMetadata)?;
    let q = quote_identifier(&table);
    let tx = messages.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
    let existing=tx.query_row(&format!("SELECT local_id,local_type,sort_seq,real_sender_id,create_time,status,upload_status,download_status,server_seq,origin_source,source,message_content,compress_content,packed_info_data,WCDB_CT_message_content,WCDB_CT_source FROM {q} WHERE server_id=?1"),[signed],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?,r.get::<_,i64>(2)?,r.get::<_,i64>(3)?,r.get::<_,i64>(4)?,r.get::<_,i64>(5)?,r.get::<_,i64>(6)?,r.get::<_,i64>(7)?,r.get::<_,i64>(8)?,r.get::<_,i64>(9)?,r.get::<_,Vec<u8>>(10)?,r.get::<_,String>(11)?,r.get::<_,String>(12)?,r.get::<_,Vec<u8>>(13)?,r.get::<_,i64>(14)?,r.get::<_,i64>(15)?))).optional()?;
    let (local, inserted) = if let Some(r) = existing {
        let duplicates: i64 = tx.query_row(
            &format!("SELECT count(*) FROM {q} WHERE server_id=?1"),
            [signed],
            |r| r.get(0),
        )?;
        if duplicates != 1
            || (r.1, r.2, r.3, r.4, r.5, r.6, r.7, r.8, r.9)
                != (
                    1,
                    sort,
                    sender,
                    input.create_time,
                    3,
                    0,
                    0,
                    input.server_sequence,
                    2,
                )
            || r.10 != source
            || r.11 != input.content
            || !r.12.is_empty()
            || r.13 != packed
            || r.14 != 0
            || r.15 != 4
        {
            return Err(ProfileError::InvalidMetadata);
        }
        (r.0, false)
    } else {
        let collisions: i64 = tx.query_row(
            &format!("SELECT count(*) FROM {q} WHERE sort_seq=?1"),
            [sort],
            |r| r.get(0),
        )?;
        if collisions != 0 {
            return Err(ProfileError::WritesUnsupported);
        }
        tx.execute(&format!("INSERT INTO {q}(server_id,local_type,sort_seq,real_sender_id,create_time,status,upload_status,download_status,server_seq,origin_source,source,message_content,compress_content,packed_info_data,WCDB_CT_message_content,WCDB_CT_source)VALUES(?1,1,?2,?3,?4,3,0,0,?5,2,?6,?7,'',?8,0,4)"),params![signed,sort,sender,input.create_time,input.server_sequence,source,input.content,packed])?;
        (tx.last_insert_rowid(), true)
    };
    tx.commit()?;
    let tx = sessions.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
    let unread_existing=tx.query_row("SELECT local_id,create_time FROM SessionUnreadListTable_1 WHERE username_id=?1 AND server_id=?2",params![session_name,signed],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?))).optional()?;
    if unread_existing.is_some_and(|(l, t)| l != local || t != input.create_time) {
        return Err(ProfileError::InvalidMetadata);
    }
    // Never resurrect an unread entry already consumed by a later official read.
    // The summary ID is the durable evidence that this exact projection completed.
    let already_projected=tx.query_row("SELECT last_msg_locald_id,last_timestamp,last_clear_unread_timestamp FROM SessionTable WHERE username=?1",[&input.peer],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?,r.get::<_,i64>(2)?)))?;
    if input.unread
        && unread_existing.is_none()
        && !(already_projected.0 == local && already_projected.1 == input.create_time)
    {
        if already_projected.1 > input.create_time || already_projected.2 >= input.create_time {
            return Err(ProfileError::WritesUnsupported);
        }
        tx.execute("INSERT INTO SessionUnreadListTable_1(username_id,server_id,local_id,create_time)VALUES(?1,?2,?3,?4)",params![session_name,signed,local,input.create_time])?;
        tx.execute("UPDATE SessionTable SET unread_first_msg_srv_id=CASE WHEN unread_count=0 THEN ?1 ELSE unread_first_msg_srv_id END,unread_count=unread_count+1 WHERE username=?2",params![signed,input.peer])?;
    }
    if current.3 <= input.create_time && current.4 <= input.create_time {
        tx.execute("UPDATE SessionTable SET summary=?1,last_timestamp=?2,sort_timestamp=?2,last_msg_locald_id=?3,last_msg_type=1,last_msg_sub_type=0,last_msg_ext_type=0 WHERE username=?4",params![input.content,input.create_time,local,input.peer])?;
    }
    tx.commit()?;
    Ok(AppliedText {
        source_id: SourceMessageId {
            database: message_db,
            table,
            local_id: local.to_string(),
        },
        message_inserted: inserted,
        session_projected: true,
        cross_database_atomic: false,
    })
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ReadProjection {
    pub removed_unread_rows: usize,
    pub unread_count_before: i64,
    pub unread_count_after: i64,
    pub exact_ids_only: bool,
}
/// Project only the selected source-qualified incoming observations. This is
/// desktop-local state: it neither acknowledges consumer events nor sends a read
/// receipt to the phone. Counter/list discrepancies already present are retained
/// by subtracting only removed rows rather than rebuilding from list cardinality.
pub fn mark_read_messages(
    location: &ProfileLocation,
    keys: &KeyFile,
    thread: &str,
    ids: &[SourceMessageId],
    options: &WriteOptions,
) -> Result<ReadProjection> {
    if !options.exclusive_ownership_confirmed {
        return Err(ProfileError::SnapshotRequired);
    }
    if options.durable_intent_id.is_empty()
        || ids.is_empty()
        || ids.len() > 200
        || thread.is_empty()
        || thread.len() > 512
    {
        return Err(ProfileError::InvalidQuery);
    }
    if keys.account_binding != location.account {
        return Err(ProfileError::AccountMismatch);
    }
    let target = format!("Msg_{:x}", Md5::digest(thread.as_bytes()));
    let mut identities = std::collections::BTreeSet::new();
    let mut seen = std::collections::BTreeSet::new();
    for id in ids {
        if !seen.insert(id.clone()) {
            continue;
        }
        if id.table != target {
            return Err(ProfileError::InvalidQuery);
        }
        let local = id
            .local_id
            .parse::<i64>()
            .map_err(|_| ProfileError::InvalidQuery)?;
        let c = open_write(location, &id.database, keys, options.allow_plaintext)?;
        exact_columns(&c, &id.database, &target, MESSAGE_COLUMNS)?;
        let row = c
            .query_row(
                &format!(
                    "SELECT server_id,create_time FROM {} WHERE local_id=?1",
                    quote_identifier(&target)
                ),
                [local],
                |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)),
            )
            .optional()?
            .ok_or(ProfileError::InvalidQuery)?;
        if row.0 == 0 {
            return Err(ProfileError::WritesUnsupported);
        }
        identities.insert((row.0, local, row.1));
    }
    let rel = find_database(location, "session.db")?;
    let mut c = open_write(location, &rel, keys, options.allow_plaintext)?;
    exact_columns(&c, &rel, "SessionTable", SESSION_COLUMNS)?;
    exact_columns(
        &c,
        &rel,
        "SessionUnreadListTable_1",
        &["username_id", "server_id", "local_id", "create_time"],
    )?;
    exact_columns(&c, &rel, "Name2Id", &["user_name"])?;
    let tx = c.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
    let nameid = tx
        .query_row(
            "SELECT rowid FROM Name2Id WHERE user_name=?1",
            [thread],
            |r| r.get::<_, i64>(0),
        )
        .optional()?
        .ok_or(ProfileError::WritesUnsupported)?;
    let (before, first) = tx
        .query_row(
            "SELECT unread_count,unread_first_msg_srv_id FROM SessionTable WHERE username=?1",
            [thread],
            |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)),
        )
        .optional()?
        .ok_or(ProfileError::WritesUnsupported)?;
    if before < 0 {
        return Err(ProfileError::WritesUnsupported);
    }
    let mut removed = 0;
    let mut removed_first = false;
    for (server, local, time) in identities {
        let existing=tx.query_row("SELECT local_id,create_time FROM SessionUnreadListTable_1 WHERE username_id=?1 AND server_id=?2",params![nameid,server],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?))).optional()?;
        if let Some((l, t)) = existing {
            if (l, t) != (local, time) {
                return Err(ProfileError::WritesUnsupported);
            }
            removed+=tx.execute("DELETE FROM SessionUnreadListTable_1 WHERE username_id=?1 AND server_id=?2 AND local_id=?3 AND create_time=?4",params![nameid,server,local,time])?;
            removed_first |= server == first;
        }
    }
    let after = before.saturating_sub(removed as i64).max(0);
    if removed > 0 {
        let next = if removed_first {
            if after == 0 {
                0
            } else {
                tx.query_row("SELECT server_id FROM SessionUnreadListTable_1 WHERE username_id=?1 ORDER BY create_time,local_id,server_id LIMIT 1",[nameid],|r|r.get::<_,i64>(0)).optional()?.unwrap_or(0)
            }
        } else {
            first
        };
        tx.execute(
            "UPDATE SessionTable SET unread_count=?1,unread_first_msg_srv_id=?2 WHERE username=?3",
            params![after, next, thread],
        )?;
    }
    tx.commit()?;
    Ok(ReadProjection {
        removed_unread_rows: removed,
        unread_count_before: before,
        unread_count_after: after,
        exact_ids_only: true,
    })
}

#[cfg(test)]
mod database_path_tests {
    use super::*;
    #[test]
    fn discovered_databases_use_the_key_manifests_portable_names() {
        let temporary = tempfile::tempdir().unwrap();
        let account = temporary.path().join("account");
        let database = account.join("db_storage/nested/session.db");
        fs::create_dir_all(database.parent().unwrap()).unwrap();
        fs::write(&database, b"fixture").unwrap();
        let location = ProfileLocation::new(account).unwrap();
        assert_eq!(
            find_database(&location, "session.db").unwrap(),
            "nested/session.db"
        );
    }
}