1 files changed, 169 insertions, 0 deletions
diff --git a/crates/chat-server/src/store.rs b/crates/chat-server/src/store.rs
new file mode 100644
index 0000000..fce3bb3
--- /dev/null
+++ b/crates/chat-server/src/store.rs
@@ -0,0 +1,169 @@
+// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai>
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+use anyhow::{Context, Result, ensure};
+use rusqlite::{Connection, OptionalExtension, params};
+use serde_json::{Value, json};
+use std::path::Path;
+
+pub(crate) struct Store {
+    connection: Connection,
+}
+
+impl Store {
+    pub(crate) fn open(path: &Path) -> Result<Self> {
+        let connection = Connection::open(path)?;
+        connection.execute_batch("PRAGMA foreign_keys=ON; PRAGMA journal_mode=DELETE; PRAGMA synchronous=FULL; PRAGMA secure_delete=ON; CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS conversations (id TEXT PRIMARY KEY, cert TEXT NOT NULL, profile TEXT); CREATE TABLE IF NOT EXISTS messages (seq INTEGER PRIMARY KEY AUTOINCREMENT, conversation TEXT NOT NULL REFERENCES conversations(id), id TEXT NOT NULL UNIQUE, sender TEXT NOT NULL, ciphertext TEXT NOT NULL);")?;
+        Ok(Self { connection })
+    }
+
+    pub(crate) fn owner(&self) -> Result<Option<String>> {
+        Ok(self
+            .connection
+            .query_row("SELECT value FROM metadata WHERE key='owner'", [], |row| {
+                row.get(0)
+            })
+            .optional()?)
+    }
+
+    pub(crate) fn install_owner(&self, cert: &str) -> Result<()> {
+        ensure!(self.owner()?.is_none(), "Already initialized");
+        self.connection
+            .execute("INSERT INTO metadata VALUES ('owner', ?1)", [cert])?;
+        Ok(())
+    }
+
+    pub(crate) fn register(&self, id: &str, cert: &str) -> Result<()> {
+        if self.connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM conversations WHERE id=?1)",
+            [id],
+            |row| row.get::<_, bool>(0),
+        )? {
+            return Ok(());
+        }
+        let count: i64 =
+            self.connection
+                .query_row("SELECT COUNT(*) FROM conversations", [], |row| row.get(0))?;
+        ensure!(count < 1000, "Conversation limit reached");
+        self.connection.execute(
+            "INSERT INTO conversations (id,cert) VALUES (?1,?2)",
+            params![id, cert],
+        )?;
+        Ok(())
+    }
+
+    pub(crate) fn conversation(&self, id: &str) -> Result<Value> {
+        let (cert, profile): (String, Option<String>) = self.connection.query_row(
+            "SELECT cert,profile FROM conversations WHERE id=?1",
+            [id],
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )?;
+        let (total, visitor): (i64, i64) = self.connection.query_row(
+            "SELECT COUNT(*),COALESCE(SUM(sender=?1),0) FROM messages WHERE conversation=?1",
+            [id],
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )?;
+        Ok(
+            json!({"id":id,"public":cert,"profile":profile,"total":total,"visitor_sent":visitor,"owner_sent":total.checked_sub(visitor).context("Counter invariant")?}),
+        )
+    }
+
+    pub(crate) fn inbox(&self) -> Result<Value> {
+        let mut statement = self
+            .connection
+            .prepare("SELECT id FROM conversations ORDER BY id")?;
+        let ids = statement
+            .query_map([], |row| row.get::<_, String>(0))?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        let conversations = ids
+            .iter()
+            .map(|id| self.conversation(id))
+            .collect::<Result<Vec<_>>>()?;
+        Ok(json!({"conversations":conversations}))
+    }
+
+    pub(crate) fn profile(&self, id: &str, ciphertext: &str) -> Result<()> {
+        self.connection.execute(
+            "UPDATE conversations SET profile=?2 WHERE id=?1 AND profile IS NULL",
+            params![id, ciphertext],
+        )?;
+        Ok(())
+    }
+
+    pub(crate) fn messages(&self, id: &str, count: u64, before: u64) -> Result<Value> {
+        ensure!(
+            count <= 50 && before <= i64::MAX.unsigned_abs(),
+            "Invalid history window"
+        );
+        let mut statement = self.connection.prepare("SELECT seq,id,sender,ciphertext FROM messages WHERE conversation=?1 AND (?3=0 OR seq<?3) ORDER BY seq DESC LIMIT ?2")?;
+        let messages = statement.query_map(params![id,i64::try_from(count)?,i64::try_from(before)?], |row| Ok(json!({"seq":row.get::<_,i64>(0)?,"id":row.get::<_,String>(1)?,"sender":row.get::<_,String>(2)?,"ciphertext":row.get::<_,String>(3)?})))?.collect::<rusqlite::Result<Vec<_>>>()?;
+        Ok(json!({"conversation":self.conversation(id)?,"messages":messages}))
+    }
+
+    pub(crate) fn send(
+        &mut self,
+        conversation: &str,
+        id: &str,
+        sender: &str,
+        ciphertext: &str,
+    ) -> Result<Value> {
+        let transaction = self.connection.transaction()?;
+        let existing: Option<(i64, String, String, String)> = transaction
+            .query_row(
+                "SELECT seq,conversation,sender,ciphertext FROM messages WHERE id=?1",
+                [id],
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+            )
+            .optional()?;
+        if let Some((seq, previous_conversation, previous_sender, previous_ciphertext)) = existing {
+            ensure!(
+                previous_conversation == conversation
+                    && previous_sender == sender
+                    && previous_ciphertext == ciphertext,
+                "Conflicting message identifier"
+            );
+            transaction.commit()?;
+            return Ok(json!({"seq":seq,"conversation":self.conversation(conversation)?}));
+        }
+        let count: i64 = transaction.query_row(
+            "SELECT COUNT(*) FROM messages WHERE conversation=?1",
+            [conversation],
+            |row| row.get(0),
+        )?;
+        let used: i64 = transaction.query_row(
+            "SELECT COALESCE(SUM(LENGTH(ciphertext)),0) FROM messages",
+            [],
+            |row| row.get(0),
+        )?;
+        ensure!(
+            count < 10_000 && used < 1_000_000_000,
+            "Storage quota reached"
+        );
+        transaction.execute(
+            "INSERT INTO messages (conversation,id,sender,ciphertext) VALUES (?1,?2,?3,?4)",
+            params![conversation, id, sender, ciphertext],
+        )?;
+        let sequence = transaction.last_insert_rowid();
+        transaction.commit()?;
+        Ok(json!({"seq":sequence,"conversation":self.conversation(conversation)?}))
+    }
+
+    pub(crate) fn snapshot(&self, destination: &Path) -> Result<()> {
+        ensure!(!destination.exists(), "Snapshot destination already exists");
+        self.connection
+            .backup("main", destination, None::<fn(rusqlite::backup::Progress)>)?;
+        Ok(())
+    }
+}