feat: new db schema

10 files changed, 232 insertions, 227 deletions

diff --git a/.helix/languages.toml b/.helix/languages.toml
index 02fab3a..66e6ace 100644
--- a/.helix/languages.toml
+++ b/.helix/languages.toml
@@ -5,5 +5,6 @@ command = "rust-analyzer"
   # check.overrideCommand="cargo check --message-format=json -p luz",
   # check.workspace = false,
   # cargo.target = "wasm32-unknown-unknown",
-# config = { cargo.target = "wasm32-unknown-unknown", cargo.features = ["jid/rusqlite", "stanza/rfc_6121", "stanza/xep_0203", "stanza/rfc_7395", "stanza/xep_0030", "stanza/xep_0060", "stanza/xep_0172", "stanza/xep_0390", "stanza/xep_0128", "stanza/xep_0115", "stanza/xep_0084", "uuid/v4", "tokio/full", "rsasl/provider_base64", "rsasl/plain", "rsasl/config_builder", "rsasl/scram-sha-1", "stanza/xep_0156"] }
+# config = { cargo.target = "wasm32-unknown-unknown" }
+config = { cargo.target = "wasm32-unknown-unknown", cargo.features = ["jid/rusqlite", "stanza/rfc_6121", "stanza/xep_0203", "stanza/rfc_7395", "stanza/xep_0030", "stanza/xep_0060", "stanza/xep_0172", "stanza/xep_0390", "stanza/xep_0128", "stanza/xep_0115", "stanza/xep_0084", "uuid/v4", "tokio/full", "rsasl/provider_base64", "rsasl/plain", "rsasl/config_builder", "rsasl/scram-sha-1", "stanza/xep_0156", "stanza/xep_0297", "stanza/xep_0280", "stanza/xep_0334"] }
 # "sqlx/sqlite", "sqlx/runtime-tokio", "sqlx/uuid", "sqlx/chrono", "jid/sqlx", 
diff --git a/README.md b/README.md
index 29c5242..d684716 100644
--- a/README.md
+++ b/README.md
@@ -168,3 +168,5 @@
 - [ ] polls
 - [ ] privacy lists/circles
 - [ ] pubsub node items filter for each person
+- [ ] unreadification, notify somebody that you unread their message
+      - and extended, settings for this. like if you want to have to manually mark as read, etc.
diff --git a/filamento/migrations/1.sql b/filamento/migrations/1.sql
index 502c5a9..ec574fc 100644
--- a/filamento/migrations/1.sql
+++ b/filamento/migrations/1.sql
@@ -1,9 +1,8 @@
 PRAGMA foreign_keys = on;
 
 -- a user jid will never change, only a chat user will change
--- TODO: avatar, nick, etc.
 create table if not exists users(
-    -- TODO: enforce bare jid
+    -- bare jid
     jid text primary key not null,
     nick text,
     avatar text,
@@ -12,28 +11,26 @@ create table if not exists users(
     -- TODO: last_seen
 );
 
--- -- links to messages, jabber users, stores jid history, etc.
--- create table identities(
---     id text primary key not null
--- );
-
--- create table identities_users(
---     id text not null,
---     jid text not null,
---     -- whichever has the newest timestamp is the active one.
---     -- what to do when somebody moves, but then the old jid is used again without having explicitly moved back? create new identity to assign ownership to?
---     -- merging of identities?
---     activated_timestamp not null,
---     foreign key(id) references identities(id),
---     foreign key(jid) references users(jid),
---     primary key(activated timestamp, id, jid)
--- );
-
-create table if not exists resources(
-    bare_jid text not null,
-    resource text not null,
-    foreign key(bare_jid) references users(jid),
-    primary key(bare_jid, resource)
+-- links to messages, jabber users, stores jid history, etc.
+-- every identity requires a user, so imported messages must be assigned to an existing jabber account.
+create table if not exists identities(
+    id text primary key not null,
+    -- each time a user moves, their identity's primary jid is updated, and the new jid is added to the identities log
+    primary_jid text not null unique,
+    foreign key(primary_jid) references users(jid)
+);
+-- this also allows e.g. mucs with imported histories to assign messages temporarily to muc-generated temporary users, then send user move events to assign message ownership once that person joins the new group chat with a jabber account.
+
+-- identities log basically.
+create table if not exists identities_users(
+    id text not null,
+    jid text not null,
+    -- what to do when somebody moves, but then the old jid is used again without having explicitly moved back? create new identity to assign ownership to?
+    -- merging of identities?
+    activated_timestamp not null,
+    foreign key(id) references identities(id),
+    foreign key(jid) references users(jid),
+    primary key(id, jid)
 );
 
 -- enum for subscription state
@@ -71,7 +68,8 @@ create table if not exists chats (
     id text primary key not null,
     have_chatted bool not null,
     correspondent text not null unique,
-    foreign key(correspondent) references users(jid)
+    -- TODO: how many messages to store locally/rely on mam, whether to auto-download/store media too.
+    foreign key(correspondent) references identities(id)
 );
 
 -- enum for subscription state
@@ -92,27 +90,17 @@ create table if not exists messages (
     -- channel_id uuid,
     -- check ((chat_id == null) <> (channel_id == null)),
     -- check ((chat_id == null) or (channel_id == null)),
-    -- user is the current "owner" of the message
     -- TODO: queued messages offline
-    -- TODO: timestamp
     timestamp text not null,
 
-    -- TODO: icky
-    -- the user to show it coming from (not necessarily the original sender)
-    -- from_identity text not null,
-    -- original sender details (only from jabber supported for now)
-    from_jid text not null,
-    -- resource can be null
-    from_resource text,
-    -- check (from_jid != original_sender),
+    -- the identity to show it coming from (not necessarily the original sender)
+    from_identity text not null,
+
+    -- TODO: unread message count, read bool not null,
 
     foreign key(delivery) references delivery(state),
-    -- TODO: from can be either a jid, a moved jid (for when a contact moves, save original sender jid/user but link to new user), or imported (from another service (save details), linked to new user)
-    -- TODO: read bool not null,
     foreign key(chat_id) references chats(id) on delete cascade,
-    -- foreign key(from_identity) references identities(id),
-    foreign key(from_jid) references users(jid),
-    foreign key(from_jid, from_resource) references resources(bare_jid, resource)
+    foreign key(from_identity) references identities(id)
 );
 
 -- enum for subscription state
@@ -139,3 +127,17 @@ create table if not exists capability_hash_nodes (
 );
 
 insert into capability_hash_nodes ( node, capabilities ) values ('https://bunny.garden/filamento#mSavc/SLnHm8zazs5RlcbD/iXoc=', 'aHR0cDovL2phYmJlci5vcmcvcHJvdG9jb2wvY2Fwcx9odHRwOi8vamFiYmVyLm9yZy9wcm90b2NvbC9kaXNjbyNpbmZvH2h0dHA6Ly9qYWJiZXIub3JnL3Byb3RvY29sL2Rpc2NvI2l0ZW1zH2h0dHA6Ly9qYWJiZXIub3JnL3Byb3RvY29sL25pY2sfaHR0cDovL2phYmJlci5vcmcvcHJvdG9jb2wvbmljaytub3RpZnkfHGNsaWVudB9wYx8fZmlsYW1lbnRvIDAuMS4wHx4cHA==') on conflict do nothing;
+
+-- TODO: later maybe just a full stanza log period?
+create table if not exists message_stanzas (
+    id integer primary key not null,
+    raw text not null,
+    timestamp text not null,
+    -- should be 1 to many as the same original stanza could technically have multiple envelopes...(e.g. message sync, mam), also makes making dangling stanzas not possible easier when modeling the relationship this way.
+    envelope_of integer,
+    linked_message text,
+    -- check that stanza is either linked to another stanza, or a message (no dangling stanzas)
+    check ((envelope_of == null) <> (linked_message == null)),
+    foreign key(linked_message) references messages(id),
+    foreign key(envelope_of) references message_stanzas(id)
+)
diff --git a/filamento/src/chat.rs b/filamento/src/chat.rs
index c02654f..07577fd 100644
--- a/filamento/src/chat.rs
+++ b/filamento/src/chat.rs
@@ -13,17 +13,43 @@ use uuid::Uuid;
 #[cfg_attr(feature = "reactive_stores", derive(reactive_stores::Store))]
 pub struct Message {
     pub id: Uuid,
-    // does not contain full user information
-    // bare jid (for now)
+    /// jid of user currently tied to the original sender, updated by jid move event. original sender can be found within the source data.
     pub from: BareJID,
     pub delivery: Option<Delivery>,
     pub timestamp: DateTime<Utc>,
-    // TODO: originally_from
-    // TODO: message edits
-    // TODO: message timestamp
+    // TODO: message edits. message edits will need to include the edit and the source stanza that caused the edit.
+    /// original message sources (XMPP, imported, etc.). cannot change, but may be deleted or have information redacted. can be multiple because a message may be updated, have reactions appended to it, delivery receipt, user moved, etc.
+    pub source: Vec<Source>,
     pub body: Body,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Source {
+    XMPP(XMPPMessage),
+    // TODO: imported information
+    Imported,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct XMPPMessage {
+    // TODO: upon software updates, upgrade all message and xmppmessage structs with missed data from old XMPP raw messages. maybe just don't store parsed message in xmpp message, have it be a method
+    /// the raw data received from the xml stream
+    pub raw: String,
+    /// the timestamp the client received the stanza, or associated with the <delay> in the envelope.
+    pub timestamp: DateTime<Utc>,
+    /// if message was received in a carbon envolope, forwarded, or in an encrypted envelope, etc., the full xmpp message it came from is included here, linked. there could technically be multiple envelopes (same stanza received through multiple envelopes).
+    pub envelopes: Vec<XMPPMessage>,
+}
+
+impl XMPPMessage {
+    // TODO: syncify
+    // pub async fn parsed(&self) -> stanza::client::message::Message {
+    //     let reader = peanuts::Reader::new(peanuts::ReadableString(self.raw.to_string()));
+    //     let message = reader.read().await?;
+    //     message
+    // }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub enum Delivery {
diff --git a/filamento/src/db.rs b/filamento/src/db.rs
index 298d54a..79fdd9a 100644
--- a/filamento/src/db.rs
+++ b/filamento/src/db.rs
@@ -341,22 +341,6 @@ impl Db {
         result
     }
 
-    pub(crate) async fn update_chat_correspondent(
-        &self,
-        old_chat: Chat,
-        new_correspondent: BareJID,
-    ) -> Result<Chat, Error> {
-        let (result, recv) = oneshot::channel();
-        let command = DbCommand::UpdateChatCorrespondent {
-            old_chat,
-            new_correspondent,
-            result,
-        };
-        self.sender.send(command);
-        let result = recv.await?;
-        result
-    }
-
     // pub(crate) async fn update_chat
 
     pub(crate) async fn delete_chat(&self, chat: BareJID) -> Result<(), Error> {
@@ -416,7 +400,7 @@ impl Db {
         &self,
         message: Message,
         chat: BareJID,
-        from: FullJID,
+        from: BareJID,
     ) -> Result<(), Error> {
         let (result, recv) = oneshot::channel();
         let command = DbCommand::CreateMessage {
@@ -438,28 +422,6 @@ impl Db {
         result
     }
 
-    /// create direct message from incoming. MUST upsert chat and user
-    #[tracing::instrument]
-    pub(crate) async fn create_message_with_user_resource(
-        &self,
-        message: Message,
-        // TODO: enforce two kinds of jid. bare and full
-        chat: BareJID,
-        from: FullJID,
-    ) -> Result<(), Error> {
-        tracing::info!("MSGDEBUG create_message_with_user_resource exists");
-        let (result, recv) = oneshot::channel();
-        let command = DbCommand::CreateMessageWithUserResource {
-            message,
-            chat,
-            from,
-            result,
-        };
-        self.sender.send(command);
-        let result = recv.await?;
-        result
-    }
-
     pub(crate) async fn update_message_delivery(
         &self,
         message: Uuid,
@@ -600,11 +562,6 @@ pub enum DbCommand {
         chat: BareJID,
         result: oneshot::Sender<Result<(), Error>>,
     },
-    UpdateChatCorrespondent {
-        old_chat: Chat,
-        new_correspondent: BareJID,
-        result: oneshot::Sender<Result<Chat, Error>>,
-    },
     DeleteChat {
         chat: BareJID,
         result: oneshot::Sender<Result<(), Error>>,
@@ -632,19 +589,13 @@ pub enum DbCommand {
     CreateMessage {
         message: Message,
         chat: BareJID,
-        from: FullJID,
+        from: BareJID,
         result: oneshot::Sender<Result<(), Error>>,
     },
     UpsertChatAndUser {
         chat: BareJID,
         result: oneshot::Sender<Result<bool, Error>>,
     },
-    CreateMessageWithUserResource {
-        message: Message,
-        chat: BareJID,
-        from: FullJID,
-        result: oneshot::Sender<Result<(), Error>>,
-    },
     UpdateMessageDelivery {
         message: Uuid,
         delivery: Delivery,
@@ -710,7 +661,6 @@ impl Display for DbCommand {
             DbCommand::CreateChat { chat, result } => "CreateChat",
             DbCommand::ReadChat { chat, result } => "ReadChat",
             DbCommand::MarkChatAsChatted { chat, result } => "MarkChatAsChatted",
-            DbCommand::UpdateChatCorrespondent { old_chat, new_correspondent, result } => "UpdateChatCorrespondent",
             DbCommand::DeleteChat { chat, result } => "DeleteChat",
             DbCommand::ReadChats { result } => "ReadChats",
             DbCommand::ReadChatsOrdered { result } => "ReadChatsOrdered",
@@ -718,7 +668,6 @@ impl Display for DbCommand {
             DbCommand::ReadChatsOrderedWithLatestMessagesAndUsers { result } => "ReadChatsOrderedWithLatestMessagesAndUsers",
             DbCommand::CreateMessage { message, chat, from, result } => "CreateMessage",
             DbCommand::UpsertChatAndUser { chat, result } => "UpsertChatAndUser",
-            DbCommand::CreateMessageWithUserResource { message, chat, from, result } => "CreateMessageWithUserResource",
             DbCommand::UpdateMessageDelivery { message, delivery, result } => "UpdateMessageDelivery",
             DbCommand::DeleteMessage { message, result } => "DeleteMessage",
             DbCommand::ReadMessage { message, result } => "ReadMessage",
@@ -865,13 +814,6 @@ impl DbActor {
                 DbCommand::MarkChatAsChatted { chat, result } => {
                     result.send(self.mark_chat_as_chatted(chat));
                 }
-                DbCommand::UpdateChatCorrespondent {
-                    old_chat,
-                    new_correspondent,
-                    result,
-                } => {
-                    result.send(self.update_chat_correspondent(old_chat, new_correspondent));
-                }
                 DbCommand::DeleteChat { chat, result } => {
                     result.send(self.delete_chat(chat));
                 }
@@ -898,14 +840,6 @@ impl DbActor {
                 DbCommand::UpsertChatAndUser { chat, result } => {
                     result.send(self.upsert_chat_and_user(&chat));
                 }
-                DbCommand::CreateMessageWithUserResource {
-                    message,
-                    chat,
-                    from,
-                    result,
-                } => {
-                    result.send(self.create_message_with_user_resource(message, chat, from));
-                }
                 DbCommand::UpdateMessageDelivery {
                     message,
                     delivery,
@@ -1059,7 +993,7 @@ impl DbActor {
         }
     }
 
-    // TODO: use references everywhere
+    // TODO: use references everywhere?
     pub(crate) fn update_user(&self, user: User) -> Result<(), Error> {
         self.db.execute(
             "update users set nick = ?1, avatar = ?2 where user_jid = ?1",
@@ -1260,23 +1194,25 @@ impl DbActor {
     }
 
     pub(crate) fn create_chat(&self, chat: Chat) -> Result<(), Error> {
-        let id = Uuid::new_v4();
         let jid = chat.correspondent();
+        debug!("aick: before user identity upsert {jid}");
+        let identity = self.upsert_user_identity(jid)?;
+        debug!("aick: chat user identity: {identity}");
+        let id = Uuid::new_v4();
+        debug!("aick: chat uuid: {id}");
         self.db.execute(
             "insert into chats (id, correspondent, have_chatted) values (?1, ?2, ?3)",
-            (id, jid, chat.have_chatted),
+            (id, identity, chat.have_chatted),
         )?;
         Ok(())
     }
 
-    // TODO: what happens if a correspondent changes from a user to a contact? maybe just have correspondent be a user, then have the client make the user show up as a contact in ui if they are in the loaded roster.
-
     /// TODO: this is NOT a read
     pub(crate) fn read_chat(&self, chat: BareJID) -> Result<Chat, Error> {
         let chat_opt = self
             .db
             .query_row(
-                "select correspondent, have_chatted from chats where correspondent = ?1",
+                "select primary_jid, have_chatted from chats join identities on correspondent = identities.id where primary_jid = ?1",
                 [&chat],
                 |row| {
                     Ok(Chat {
@@ -1302,7 +1238,7 @@ impl DbActor {
     pub(crate) fn read_chat_and_user(&self, chat: BareJID) -> Result<(Chat, User), Error> {
         let user = self.read_user(chat.clone())?;
         let chat_opt = self.db.query_row(
-            "select correspondent, have_chatted, jid, nick, avatar from chats join users on correspondent = jid where correspondent = ?1",
+            "select primary_jid, have_chatted, jid, nick, avatar from chats join identities i on chats.correspondent = i.id join users on jid = primary_jid where primary_jid = ?1",
             [&chat],
             |row| {
                 Ok((
@@ -1325,7 +1261,9 @@ impl DbActor {
                     correspondent: chat,
                     have_chatted: false,
                 };
+                debug!("aick: creating chat");
                 self.create_chat(chat.clone())?;
+                debug!("aick: created chat");
                 Ok((chat, user))
             }
         }
@@ -1333,35 +1271,15 @@ impl DbActor {
 
     pub(crate) fn mark_chat_as_chatted(&self, chat: BareJID) -> Result<(), Error> {
         self.db.execute(
-            "update chats set have_chatted = true where correspondent = ?1",
+            "update chats set have_chatted = true where correspondent = (select id from identities where primary_jid = ?1)",
             [chat],
         )?;
         Ok(())
     }
 
-    pub(crate) fn update_chat_correspondent(
-        &self,
-        old_chat: Chat,
-        new_correspondent: BareJID,
-    ) -> Result<Chat, Error> {
-        let new_jid = &new_correspondent;
-        let old_jid = old_chat.correspondent();
-        let chat = self.db.query_row(
-            "update chats set correspondent = ?1 where correspondent = ?2 returning correspondent, have_chatted",
-            [new_jid, old_jid],
-            |row| Ok(Chat {
-                correspondent: row.get(0)?,
-                have_chatted: row.get(1)?,
-            })
-        )?;
-        Ok(chat)
-    }
-
-    // pub(crate) fn update_chat
-
     pub(crate) fn delete_chat(&self, chat: BareJID) -> Result<(), Error> {
         self.db
-            .execute("delete from chats where correspondent = ?1", [chat])?;
+            .execute("delete from chats where correspondent = (select id from identities where primary_jid = ?1)", [chat])?;
         Ok(())
     }
 
@@ -1369,7 +1287,7 @@ impl DbActor {
     pub(crate) fn read_chats(&self) -> Result<Vec<Chat>, Error> {
         let chats = self
             .db
-            .prepare("select correspondent, have_chatted from chats")?
+            .prepare("select primary_jid, have_chatted from chats join identities on correspondent = identities.id")?
             .query_map([], |row| {
                 Ok(Chat {
                     correspondent: row.get(0)?,
@@ -1385,7 +1303,7 @@ impl DbActor {
     pub(crate) fn read_chats_ordered(&self) -> Result<Vec<Chat>, Error> {
         let chats = self
             .db
-            .prepare("select c.correspondent, c.have_chatted, m.* from chats c join (select chat_id, max(timestamp) max_timestamp from messages group by chat_id) max_timestamps on c.id = max_timestamps.chat_id join messages m on max_timestamps.chat_id = m.chat_id and max_timestamps.max_timestamp = m.timestamp order by m.timestamp desc")?
+            .prepare("select i.primary_jid, c.have_chatted, m.* from chats c join identities i on c.correspondent = i.id join (select chat_id, max(timestamp) max_timestamp from messages group by chat_id) max_timestamps on c.id = max_timestamps.chat_id join messages m on max_timestamps.chat_id = m.chat_id and max_timestamps.max_timestamp = m.timestamp order by m.timestamp desc")?
             .query_map([], |row| {
                 Ok(Chat {
                     correspondent: row.get(0)?,
@@ -1401,9 +1319,32 @@ impl DbActor {
     pub(crate) fn read_chats_ordered_with_latest_messages(
         &self,
     ) -> Result<Vec<(Chat, Message)>, Error> {
-        let chats = self
+        let mut chats = self
             .db
-            .prepare("select c.correspondent, c.have_chatted, m.id, m.from_jid, m.delivery, m.timestamp, m.body from chats c join (select chat_id, max(timestamp) max_timestamp from messages group by chat_id) max_timestamps on c.id = max_timestamps.chat_id join messages m on max_timestamps.chat_id = m.chat_id and max_timestamps.max_timestamp = m.timestamp order by m.timestamp desc")?
+            .prepare(
+                "
+SELECT ci.primary_jid,
+       c.have_chatted,
+       m.id,
+       ui.primary_jid,
+       m.delivery,
+       m.timestamp,
+       m.body
+FROM   chats c
+       JOIN (SELECT chat_id,
+                    Max(timestamp) max_timestamp
+             FROM   messages
+             GROUP  BY chat_id) max_timestamps
+         ON c.id = max_timestamps.chat_id
+       JOIN messages m
+         ON max_timestamps.chat_id = m.chat_id
+            AND max_timestamps.max_timestamp = m.timestamp
+       JOIN identities AS ci
+         ON ci.id = c.correspondent
+       JOIN identities AS ui
+         ON ui.id = m.from_identity
+ORDER  BY m.timestamp DESC",
+            )?
             .query_map([], |row| {
                 Ok((
                     Chat {
@@ -1415,10 +1356,10 @@ impl DbActor {
                         from: row.get(3)?,
                         delivery: row.get(4)?,
                         timestamp: row.get(5)?,
-                        body: Body {
-                            body: row.get(6)?,
-                        },
-                    }
+                        body: Body { body: row.get(6)? },
+                        // TODO: query raw sources.
+                        source: Vec::new(),
+                    },
                 ))
             })?
             .collect::<Result<Vec<_>, _>>()?;
@@ -1432,7 +1373,41 @@ impl DbActor {
     ) -> Result<Vec<((Chat, User), (Message, User))>, Error> {
         let chats = self
             .db
-            .prepare("select c.id as chat_id, c.correspondent as chat_correspondent, c.have_chatted as chat_have_chatted, m.id as message_id, m.body as message_body, m.delivery as message_delivery, m.timestamp as message_timestamp, m.from_jid as message_from_jid, cu.jid as chat_user_jid, cu.nick as chat_user_nick, cu.avatar as chat_user_avatar, mu.jid as message_user_jid, mu.nick as message_user_nick, mu.avatar as message_user_avatar from chats c join (select chat_id, max(timestamp) max_timestamp from messages group by chat_id) max_timestamps on c.id = max_timestamps.chat_id join messages m on max_timestamps.chat_id = m.chat_id and max_timestamps.max_timestamp = m.timestamp join users as cu on cu.jid = c.correspondent join users as mu on mu.jid = m.from_jid order by m.timestamp desc")?
+            .prepare(
+                "
+SELECT c.id            AS chat_id,
+       ci.primary_jid  AS chat_correspondent,
+       c.have_chatted  AS chat_have_chatted,
+       m.id            AS message_id,
+       m.body          AS message_body,
+       m.delivery      AS message_delivery,
+       m.timestamp     AS message_timestamp,
+       ui.primary_jid  AS message_from_jid,
+       cu.jid          AS chat_user_jid,
+       cu.nick         AS chat_user_nick,
+       cu.avatar       AS chat_user_avatar,
+       mu.jid          AS message_user_jid,
+       mu.nick         AS message_user_nick,
+       mu.avatar       AS message_user_avatar
+FROM   chats c
+       JOIN (SELECT chat_id,
+                    Max(timestamp) max_timestamp
+             FROM   messages
+             GROUP  BY chat_id) max_timestamps
+         ON c.id = max_timestamps.chat_id
+       JOIN messages m
+         ON max_timestamps.chat_id = m.chat_id
+            AND max_timestamps.max_timestamp = m.timestamp
+       JOIN identities AS ci
+         ON ci.id = c.correspondent
+       JOIN identities AS ui
+         ON ui.id = m.from_identity
+       JOIN users AS cu
+         ON cu.jid = ci.primary_jid
+       JOIN users AS mu
+         ON mu.jid = ui.primary_jid
+ORDER  BY m.timestamp DESC",
+            )?
             .query_map([], |row| {
                 Ok((
                     (
@@ -1444,7 +1419,7 @@ impl DbActor {
                             jid: row.get("chat_user_jid")?,
                             nick: row.get("chat_user_nick")?,
                             avatar: row.get("chat_user_avatar")?,
-                        }
+                        },
                     ),
                     (
                         Message {
@@ -1455,12 +1430,14 @@ impl DbActor {
                             body: Body {
                                 body: row.get("message_body")?,
                             },
+                            // TODO: query raw sources.
+                            source: Vec::new(),
                         },
                         User {
                             jid: row.get("message_user_jid")?,
                             nick: row.get("message_user_nick")?,
                             avatar: row.get("message_user_avatar")?,
-                        }
+                        },
                     ),
                 ))
             })?
@@ -1471,18 +1448,18 @@ impl DbActor {
     #[tracing::instrument]
     fn read_chat_id(&self, chat: BareJID) -> Result<Uuid, Error> {
         let chat_id = self.db.query_row(
-            "select id from chats where correspondent = ?1",
+            "select id from chats where correspondent = (select id from identities where primary_jid = ?1)",
             [chat],
             |row| Ok(row.get(0)?),
         )?;
         Ok(chat_id)
     }
 
-    fn read_chat_id_opt(&self, chat: JID) -> Result<Option<Uuid>, Error> {
+    fn read_chat_id_opt(&self, chat: BareJID) -> Result<Option<Uuid>, Error> {
         let chat_id = self
             .db
             .query_row(
-                "select id from chats where correspondent = ?1",
+                "select id from chats where correspondent = (select id from identities where primary_jid = ?1)",
                 [chat],
                 |row| Ok(row.get(0)?),
             )
@@ -1491,57 +1468,60 @@ impl DbActor {
     }
 
     /// if the chat doesn't already exist, it must be created by calling create_chat() before running this function.
+    /// create direct message from incoming. MUST upsert user w/ identity, and chat w/identity
     #[tracing::instrument]
     pub(crate) fn create_message(
         &self,
         message: Message,
         chat: BareJID,
-        from: FullJID,
+        from: BareJID,
     ) -> Result<(), Error> {
-        let from_jid = from.as_bare();
-        let chat_id = self.read_chat_id(chat)?;
+        debug!("oomla: 1");
+        let chat_identity = self.upsert_user_identity(&chat)?;
+        debug!("oomla: upserted chat user and got identity {chat_identity}");
+        let (chat_id, _) = self.upsert_chat(chat_identity)?;
+        debug!("oomla: upserted chat and got chat id {chat_id}");
+        let from_identity = self.upsert_user_identity(&from)?;
+        debug!("oomla: upserted from user and got user identity {from_identity}");
         tracing::debug!("creating message");
-        self.db.execute("insert into messages (id, body, chat_id, from_jid, from_resource, timestamp, delivery) values (?1, ?2, ?3, ?4, ?5, ?6, ?7)", (&message.id, &message.body.body, &chat_id, &from_jid, &from.resourcepart, &message.timestamp, &message.delivery))?;
+        self.db.execute("insert into messages (id, body, chat_id, from_identity, timestamp, delivery) values (?1, ?2, ?3, ?4, ?5, ?6)", (&message.id, &message.body.body, &chat_id, &from_identity, &message.timestamp, &message.delivery))?;
         Ok(())
     }
 
-    pub(crate) fn upsert_chat_and_user(&self, chat: &BareJID) -> Result<bool, Error> {
-        let db = &self.db;
-        db.execute(
+    // returns the user identity
+    pub(crate) fn upsert_user_identity(&self, chat: &BareJID) -> Result<Uuid, Error> {
+        self.db.execute(
             "insert into users (jid) values (?1) on conflict do nothing",
             [&chat],
         )?;
-        let id = Uuid::new_v4();
-        db.execute("insert into chats (id, correspondent, have_chatted) values (?1, ?2, ?3) on conflict do nothing", (id, &chat, false))?;
-        let chat = db.query_row(
-            "select correspondent, have_chatted from chats where correspondent = ?1",
+        let identity = Uuid::new_v4();
+        self.db.execute(
+            "insert into identities (id, primary_jid) values (?1, ?2) on conflict do nothing",
+            (identity, &chat),
+        )?;
+        let identity = self.db.query_row(
+            "select id from identities where primary_jid = ?1",
             [&chat],
-            |row| {
-                Ok(Chat {
-                    correspondent: row.get(0)?,
-                    have_chatted: row.get(1)?,
-                })
-            },
+            |row| Ok(row.get(0)?),
         )?;
-        Ok(chat.have_chatted)
+        Ok(identity)
     }
 
-    /// create direct message from incoming. MUST upsert chat and user
-    #[tracing::instrument]
-    pub(crate) fn create_message_with_user_resource(
-        &self,
-        message: Message,
-        chat: BareJID,
-        from: FullJID,
-    ) -> Result<(), Error> {
-        let from_jid = from.as_bare();
-        tracing::debug!("creating resource");
-        self.db.execute(
-            "insert into resources (bare_jid, resource) values (?1, ?2) on conflict do nothing",
-            (&from_jid, &from.resourcepart),
+    pub(crate) fn upsert_chat(&self, identity: Uuid) -> Result<(Uuid, bool), Error> {
+        let chat_id = Uuid::new_v4();
+        self.db.execute("insert into chats (id, correspondent, have_chatted) values (?1, ?2, ?3) on conflict do nothing", (chat_id, &identity, false))?;
+        let (chat_id, have_chatted) = self.db.query_row(
+            "select id, have_chatted from chats where correspondent = ?1",
+            [identity],
+            |row| Ok((row.get(0)?, row.get(1)?)),
         )?;
-        self.create_message(message, chat, from)?;
-        Ok(())
+        Ok((chat_id, have_chatted))
+    }
+
+    pub(crate) fn upsert_chat_and_user(&self, chat: &BareJID) -> Result<bool, Error> {
+        let chat_identity = self.upsert_user_identity(&chat)?;
+        let (_chat_id, have_chatted) = self.upsert_chat(chat_identity)?;
+        Ok(have_chatted)
     }
 
     pub(crate) fn update_message_delivery(
@@ -1576,7 +1556,7 @@ impl DbActor {
 
     pub(crate) fn read_message(&self, message: Uuid) -> Result<Message, Error> {
         let message = self.db.query_row(
-            "select id, from_jid, delivery, timestamp, body from messages where id = ?1",
+            "select id, primary_jid, delivery, timestamp, body from messages join identities on identities.id = messages.from_identity where messages.id = ?1",
             [&message],
             |row| {
                 Ok(Message {
@@ -1586,6 +1566,8 @@ impl DbActor {
                     delivery: row.get(2)?,
                     timestamp: row.get(3)?,
                     body: Body { body: row.get(4)? },
+                    // TODO: query raw sources
+                    source: Vec::new(),
                 })
             },
         )?;
@@ -1598,16 +1580,17 @@ impl DbActor {
         let messages = self
             .db
             .prepare(
-                "select id, from_jid, delivery, timestamp, body from messages where chat_id = ?1",
+                "select id, primary_jid, delivery, timestamp, body from messages join identities on identities.id = messages.from_identity where chat_id = (select id from identities where primary_jid = ?1)",
             )?
             .query_map([chat_id], |row| {
                 Ok(Message {
                     id: row.get(0)?,
-                    // TODO: full from
                     from: row.get(1)?,
                     delivery: row.get(2)?,
                     timestamp: row.get(3)?,
                     body: Body { body: row.get(4)? },
+                    // TODO: query raw sources
+                    source: Vec::new(),
                 })
             })?
             .collect::<Result<Vec<_>, _>>()?;
@@ -1622,7 +1605,7 @@ impl DbActor {
         let messages = self
             .db
             .prepare(
-                "select id, from_jid, delivery, timestamp, body, jid, nick, avatar from messages join users on jid = from_jid where chat_id = ? order by timestamp asc",
+                "select id, primary_jid, delivery, timestamp, body, jid, nick, avatar from messages join users on jid = (select primary_jid from identities where id = from_identity) where chat_id = ? order by timestamp asc",
             )?
             .query_map([chat_id], |row| {
                 Ok((
@@ -1633,6 +1616,8 @@ impl DbActor {
                         delivery: row.get(2)?,
                         timestamp: row.get(3)?,
                         body: Body { body: row.get(4)? },
+                        // TODO: query raw sources
+                        source: Vec::new(),
                     },
                     User {
                         jid: row.get(5)?,
diff --git a/filamento/src/lib.rs b/filamento/src/lib.rs
index d3033b7..cb3d820 100644
--- a/filamento/src/lib.rs
+++ b/filamento/src/lib.rs
@@ -803,13 +803,12 @@ mod tests {
 
     use crate::chat;
     use crate::files::FilesMem;
-    use std::path::Path;
     use tokio_with_wasm::alias as tokio;
 
     #[wasm_bindgen_test]
     async fn login() -> () {
         tracing_wasm::set_as_global_default();
-        let db = Db::create_connect_and_migrate(Path::new("./filamento.db"))
+        let db = Db::create_connect_and_migrate("./filamento.db")
             .await
             .unwrap();
         let (client, mut recv) = Client::new(
diff --git a/filamento/src/logic/offline.rs b/filamento/src/logic/offline.rs
index aa84f3d..3b0d1c9 100644
--- a/filamento/src/logic/offline.rs
+++ b/filamento/src/logic/offline.rs
@@ -163,21 +163,14 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
                 delivery: Some(Delivery::Failed),
                 timestamp,
                 body,
+                source: Vec::new(),
             };
             // try to store in message history that there is a new message that is sending. if client is quit mid-send then can mark as failed and re-send
             // TODO: mark these as potentially failed upon client launch
             if let Err(e) = logic
                 .db()
                 // TODO: can create message without a resource....
-                .create_message_with_user_resource(
-                    message.clone(),
-                    jid.clone(),
-                    // TODO: when message is queued and sent, the from must also be updated with the correct resource
-                    FullJID {
-                        bare_jid: logic.jid.clone(),
-                        resourcepart: "unsent".to_string(),
-                    },
-                )
+                .create_message(message.clone(), jid.clone(), logic.jid.clone())
                 .await
             {
                 // TODO: should these really be handle_error or just the error macro?
diff --git a/filamento/src/logic/online.rs b/filamento/src/logic/online.rs
index b36f9a9..2368eff 100644
--- a/filamento/src/logic/online.rs
+++ b/filamento/src/logic/online.rs
@@ -520,13 +520,15 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
         body: body.clone(),
         timestamp,
         delivery: Some(Delivery::Sending),
+        // TODO: raw stanza logging
+        source: Vec::new(),
     };
 
     // try to store in message history that there is a new message that is sending. if client is quit mid-send then can mark as failed and re-send
     // TODO: mark these as potentially failed upon client launch
     if let Err(e) = logic
         .db()
-        .create_message_with_user_resource(message.clone(), jid.clone(), connection.jid().clone())
+        .create_message(message.clone(), jid.clone(), connection.jid().to_bare())
         .await
     {
         // TODO: should these really be handle_error or just the error macro?
@@ -586,6 +588,7 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
     match result {
         Ok(_) => {
             info!("sent message: {:?}", message_stanza);
+            // TODO: raw stanza
             if let Err(e) = logic.db().update_message_delivery(id, Delivery::Written).await {
                 error!("updating message delivery: {}", e);
             }
diff --git a/filamento/src/logic/process_stanza.rs b/filamento/src/logic/process_stanza.rs
index 67b0d3f..5f180b9 100644
--- a/filamento/src/logic/process_stanza.rs
+++ b/filamento/src/logic/process_stanza.rs
@@ -76,31 +76,19 @@ pub async fn recv_message<Fs: FileStore + Clone>(
                     body: body.body.unwrap_or_default(),
                 },
                 delivery: None,
+                // TODO: log raw stanza
+                source: Vec::new(),
             };
             // TODO: process message type="error"
 
             // save the message to the database
-            match logic.db().upsert_chat_and_user(from.to_bare()).await {
-                Ok(_) => match from.as_full() {
-                    Ok(from) => {
-                        if let Err(e) = logic
-                            .db()
-                            .create_message_with_user_resource(
-                                message.clone(),
-                                from.to_bare(),
-                                from.clone(),
-                            )
-                            .await
-                        {
-                            error!("failed to create message: {}", e);
-                        }
-                    }
-                    Err(e) => error!("failed to create message: {}", e),
-                },
-                Err(e) => {
-                    error!("failed to upsert chat and user: {}", e);
-                }
-            };
+            if let Err(e) = logic
+                .db()
+                .create_message(message.clone(), from.to_bare(), from.to_bare())
+                .await
+            {
+                error!("failed to create message: {}", e);
+            }
 
             let from_user = match logic.db().read_user(from.to_bare()).await {
                 Ok(u) => u,
diff --git a/stanza/src/xep_0297.rs b/stanza/src/xep_0297.rs
index 4dc8a26..533a171 100644
--- a/stanza/src/xep_0297.rs
+++ b/stanza/src/xep_0297.rs
@@ -13,6 +13,12 @@ pub struct Forwarded {
     stanza: Option<Box<Stanza>>,
 }
 
+// TODO: raw stanza source option
+pub enum ForwardedStanza {
+    Parsed(Box<Stanza>),
+    Raw(String),
+}
+
 impl FromElement for Forwarded {
     fn from_element(mut element: Element) -> peanuts::DeserializeResult<Self> {
         element.check_name("forwarded")?;