[refactor] 0.2.0

1. unused and remove `bytes::Bytes`
2. add Error types

Author: icode

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "yrs-postgres"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2024"
 description = "Simple store layer over Yrs documents"
 license = "MIT"
@@ -11,7 +11,6 @@ repository = "https://github.com/icode/yrs-store"
 readme = "./README.md"
 
 [dependencies]
-bytes = "1.10.1"
 chrono = { version = "0.4", features = ["serde"] }
 futures-util = "0.3.31"
 sqlx = { version = "0.8", default-features = false, features = ["postgres", "chrono", "runtime-tokio"] }
@@ -20,7 +19,7 @@ tokio = { version = "1.44", features = ["macros", "rt-multi-thread"] }
 yrs = { version = "0.23", features = ["sync", "weak"] }
 async-trait = "0.1.88"
 
-yrs-store = { version = "0.1", path = "../yrs-store" }
+yrs-store = { version = "0.2", path = "../yrs-store" }
 
 [dev-dependencies]
 sqlx = { version = "0.8", default-features = false, features = ["macros", "postgres"] }

yrs-postgres/Cargo.toml

@@ -1,6 +1,5 @@
 use async_stream;
 use async_trait::async_trait;
-use bytes::Bytes;
 use chrono::NaiveDateTime;
 use futures_util::stream::BoxStream;
 use futures_util::TryStreamExt;
@@ -33,6 +32,10 @@ impl PostgresStorage {
     }
 }
 
+fn map_sqlx_err(error: sqlx::Error) -> StoreError {
+    StoreError::StorageError(Box::new(error))
+}
+
 #[async_trait]
 impl Store for PostgresStorage {
     async fn delete(&self) -> Result<(), StoreError> {
@@ -44,22 +47,22 @@ impl Store for PostgresStorage {
         .execute(&self.pool)
         .await
         .map(|_| ())
-        .map_err(StoreError::SqlxError)
+        .map_err(map_sqlx_err)
     }
 
-    async fn write(&self, update: &Bytes) -> Result<(), StoreError> {
+    async fn write(&self, update: &Vec<u8>) -> Result<(), StoreError> {
         let document_id = self.document_id.clone();
         let now = chrono::Utc::now().naive_utc();
         let query_result = sqlx::query(&format!(
             "INSERT INTO {} (document_id, payload, timestamp) VALUES ($1, $2, $3)",
             self.table_name
         ))
         .bind(document_id)
-        .bind(update.as_ref())
+        .bind(update)
         .bind(now)
         .execute(&self.pool)
         .await
-        .map_err(StoreError::SqlxError)?;
+        .map_err(map_sqlx_err)?;
 
         let rows_affected = query_result.rows_affected();
 
@@ -74,7 +77,7 @@ impl Store for PostgresStorage {
         Ok(())
     }
 
-    async fn read(&self) -> Result<BoxStream<Result<(Bytes, i64), StoreError>>, StoreError> {
+    async fn read(&self) -> Result<BoxStream<Result<(Vec<u8>, i64), StoreError>>, StoreError> {
         let document_id = self.document_id;
         let table_name = self.table_name.clone();
         let pool = self.pool.clone();
@@ -89,9 +92,8 @@ impl Store for PostgresStorage {
                 .bind(document_id)
                 .fetch(&pool);
 
-            while let Some(row) = rows.try_next().await.map_err(StoreError::SqlxError)? {
-                let payload_vec: Vec<u8> = row.get("payload");
-                let payload = payload_vec.into();
+            while let Some(row) = rows.try_next().await.map_err(map_sqlx_err)? {
+                let payload: Vec<u8> = row.get("payload");
                 let timestamp_ndt: NaiveDateTime = row.get("timestamp");
                 let timestamp_ms = timestamp_ndt.and_utc().timestamp_millis();
                 yield Ok((payload, timestamp_ms));
@@ -101,7 +103,7 @@ impl Store for PostgresStorage {
         Ok(Box::pin(stream))
     }
 
-    async fn read_payloads(&self) -> Result<BoxStream<Result<Bytes, StoreError>>, StoreError> {
+    async fn read_payloads(&self) -> Result<BoxStream<Result<Vec<u8>, StoreError>>, StoreError> {
         let document_id = self.document_id;
         let table_name = self.table_name.clone();
         let pool = self.pool.clone();
@@ -116,9 +118,8 @@ impl Store for PostgresStorage {
                 .bind(document_id)
                 .fetch(&pool);
 
-            while let Some(row) = rows.try_next().await.map_err(StoreError::SqlxError)? {
-                let payload_vec: Vec<u8> = row.get("payload");
-                let payload = payload_vec.into();
+            while let Some(row) = rows.try_next().await.map_err(map_sqlx_err)? {
+                let payload: Vec<u8> = row.get("payload");
                 yield Ok(payload);
             }
         };
@@ -129,28 +130,28 @@ impl Store for PostgresStorage {
     async fn squash(&self) -> Result<(), StoreError> {
         let doc = Doc::new();
         self.load(&doc).await?;
-        let tx = self.pool.begin().await.map_err(StoreError::SqlxError)?;
+        let tx = self.pool.begin().await.map_err(map_sqlx_err)?;
         sqlx::query(&format!(
             "DELETE FROM {} WHERE document_id = $1",
             self.table_name
         ))
         .bind(self.document_id)
         .execute(&self.pool)
         .await
-        .map_err(StoreError::SqlxError)?;
+        .map_err(map_sqlx_err)?;
 
         let squashed_update = doc.get_update();
         self.write(&squashed_update).await?;
 
         // 如果在事务超出范围之前都未调用，rollback则自动调用
-        tx.commit().await.map_err(StoreError::SqlxError)?;
+        tx.commit().await.map_err(map_sqlx_err)?;
 
         if self.run_vacuum {
             // 回收死行占据的存储空间
             sqlx::query(&format!("VACUUM {}", self.table_name))
                 .execute(&self.pool)
                 .await
-                .map_err(StoreError::SqlxError)?;
+                .map_err(map_sqlx_err)?;
         }
         Ok(())
     }

yrs-postgres/src/lib.rs

@@ -1,6 +1,6 @@
 [package]
 name = "yrs-store"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2024"
 description = "Simple store layer over Yrs documents"
 license = "MIT"
@@ -11,8 +11,6 @@ repository = "https://github.com/icode/yrs-store"
 readme = "./README.md"
 
 [dependencies]
-bytes = "1.10.1"
 futures-util = "0.3.31"
-sqlx = { version = "0.8", default-features = false }
 yrs = { version = "0.23", features = ["sync", "weak"] }
 async-trait = "0.1.88"

yrs-store/Cargo.toml

@@ -1,4 +1,3 @@
-use bytes::Bytes;
 use std::collections::HashMap;
 use yrs::error::{Error, UpdateError};
 use yrs::updates::decoder::Decode;
@@ -8,29 +7,29 @@ use yrs::{
 };
 
 pub trait ForStore {
-    fn get_update(&self) -> Bytes;
-    fn diff_state_update(&self, state: &Bytes) -> Result<Bytes, Error>;
-    fn apply_update(&self, txn: &mut TransactionMut, update: &Bytes) -> Result<(), UpdateError>;
+    fn get_update(&self) -> Vec<u8>;
+    fn diff_state_update(&self, state: &Vec<u8>) -> Result<Vec<u8>, Error>;
+    fn apply_update(&self, txn: &mut TransactionMut, update: &Vec<u8>) -> Result<(), UpdateError>;
     fn roots(&self, txn: &Transaction<'static>) -> HashMap<String, Out>;
     fn observe<F>(&self, f: F) -> Result<Subscription, TransactionAcqError>
     where
         F: Fn(&TransactionMut, &TransactionCleanupEvent) + Send + Sync + 'static;
 }
 
 impl ForStore for Doc {
-    fn get_update(&self) -> Bytes {
+    fn get_update(&self) -> Vec<u8> {
         self.transact()
             .encode_state_as_update_v1(&StateVector::default())
             .into()
     }
 
-    fn diff_state_update(&self, state: &Bytes) -> Result<Bytes, Error> {
+    fn diff_state_update(&self, state: &Vec<u8>) -> Result<Vec<u8>, Error> {
         let state_vector = StateVector::decode_v1(&state.as_ref())?;
         let update = self.transact().encode_state_as_update_v1(&state_vector);
         Ok(update.into())
     }
 
-    fn apply_update(&self, txn: &mut TransactionMut, update: &Bytes) -> Result<(), UpdateError> {
+    fn apply_update(&self, txn: &mut TransactionMut, update: &Vec<u8>) -> Result<(), UpdateError> {
         let u = Update::decode_v1(update.as_ref()).unwrap();
         txn.apply_update(u)
     }

yrs-store/src/doc.rs

@@ -1,3 +1,4 @@
+use std::error::Error;
 use std::fmt;
 use yrs::error::UpdateError;
 
@@ -6,18 +7,20 @@ pub enum StoreError {
     UpdateError(UpdateError),
     YrsError(yrs::error::Error),
     WriteError(String),
-    SqlxError(sqlx::Error),
+    Error(Box<dyn Error + Send + Sync + 'static>),
+    StorageError(Box<dyn Error + Send + Sync + 'static>),
 }
 
 impl fmt::Display for StoreError {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
             StoreError::UpdateError(e) => e.fmt(f),
-            StoreError::SqlxError(e) => e.fmt(f),
+            StoreError::Error(e) => e.fmt(f),
+            StoreError::StorageError(e) => e.fmt(f),
             StoreError::YrsError(e) => e.fmt(f),
             StoreError::WriteError(msg) => write!(f, "Write error: {}", msg),
         }
     }
 }
 
-impl std::error::Error for StoreError {}
+impl Error for StoreError {}

yrs-store/src/errors.rs

@@ -4,10 +4,8 @@ pub mod errors;
 use crate::doc::ForStore;
 use crate::errors::StoreError;
 use async_trait::async_trait;
-use bytes::Bytes;
 use futures_util::stream::BoxStream;
 use futures_util::{StreamExt};
-use std::pin::pin;
 use yrs::{Doc, Transact};
 
 #[async_trait]
@@ -19,9 +17,9 @@ pub trait Store: Send + Sync {
         Ok(())
     }
     async fn delete(&self) -> Result<(), StoreError>;
-    async fn write(&self, update: &Bytes) -> Result<(), StoreError>;
-    async fn read(&self) -> Result<BoxStream<Result<(Bytes, i64), StoreError>>, StoreError>;
-    async fn read_payloads(&self) -> Result<BoxStream<Result<Bytes, StoreError>>, StoreError>;
+    async fn write(&self, update: &Vec<u8>) -> Result<(), StoreError>;
+    async fn read(&self) -> Result<BoxStream<Result<(Vec<u8>, i64), StoreError>>, StoreError>;
+    async fn read_payloads(&self) -> Result<BoxStream<Result<Vec<u8>, StoreError>>, StoreError>;
     async fn squash(&self) -> Result<(), StoreError>;
     /// save a YDoc updates
     /// # Arguments
@@ -36,7 +34,6 @@ pub trait Store: Send + Sync {
     async fn load(&self, doc: &Doc) -> Result<(), StoreError> {
         let mut txn = doc.transact_mut();
         let mut streams = self.read_payloads().await?;
-        let mut streams = pin!(streams);
         while let Some(result) = streams.next().await {
             let update = result?;
             doc.apply_update(&mut txn, &update)