feat: insert database row into selected database

Cargo.toml

@@ -47,7 +47,7 @@ chrono = { version = "0.4.37", features = [
 derive_more = { version = "0.99" }
 secrecy.workspace = true
 rand = { version = "0.8", features = ["std_rng"] }
-anyhow = "1.0.79"
+anyhow.workspace = true
 thiserror = "1.0.56"
 reqwest = { workspace = true, features = [
   "json",
@@ -248,7 +248,7 @@ serde = { version = "1.0.195", features = ["derive"] }
 bytes = "1.5.0"
 workspace-template = { path = "libs/workspace-template" }
 uuid = { version = "1.6.1", features = ["v4", "v5"] }
-anyhow = "1.0.79"
+anyhow = "1.0.94"
 actix = "0.13.3"
 actix-web = { version = "4.5.1", default-features = false, features = [
   "openssl",
@@ -312,13 +312,13 @@ lto = false          # Disable Link-Time Optimization
 [patch.crates-io]
 # It's diffcult to resovle different version with the same crate used in AppFlowy Frontend and the Client-API crate.
 # So using patch to workaround this issue.
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
-collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "c4e23b6d6b1ba320c74b40e557069e99dd0377e7" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
+collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "128f3a81ea86a58e355615b1c00edbf861867886" }
 
 [features]
 history = []

admin_frontend/Cargo.toml

@@ -12,7 +12,7 @@ gotrue-entity = { path = "../libs/gotrue-entity" }
 database-entity = { path = "../libs/database-entity" }
 shared-entity = { path = "../libs/shared-entity" }
 
-anyhow = "1.0.79"
+anyhow.workspace = true
 axum = { version = "0.7", features = ["json"] }
 tokio = { version = "1.36", features = ["rt-multi-thread", "macros"] }
 askama = "0.12"

libs/client-api/src/http_chat.rs

@@ -320,7 +320,7 @@ impl Stream for QuestionStream {
   fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
     let this = self.project();
 
-    return match ready!(this.stream.as_mut().poll_next(cx)) {
+    match ready!(this.stream.as_mut().poll_next(cx)) {
       Some(Ok(value)) => match value {
         Value::Object(mut value) => {
           if let Some(metadata) = value.remove(STREAM_METADATA_KEY) {
@@ -344,6 +344,6 @@ impl Stream for QuestionStream {
       },
       Some(Err(err)) => Poll::Ready(Some(Err(err))),
       None => Poll::Ready(None),
-    };
+    }
   }
 }

libs/client-api/src/http_collab.rs

@@ -4,8 +4,8 @@ use app_error::AppError;
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use client_api_entity::workspace_dto::{
-  AFDatabase, AFDatabaseField, AFDatabaseRow, AFDatabaseRowDetail, DatabaseRowUpdatedItem,
-  ListDatabaseRowDetailParam, ListDatabaseRowUpdatedParam,
+  AFDatabase, AFDatabaseField, AFDatabaseRow, AFDatabaseRowDetail, AFInsertDatabaseField,
+  DatabaseRowUpdatedItem, ListDatabaseRowDetailParam, ListDatabaseRowUpdatedParam,
 };
 use client_api_entity::{
   BatchQueryCollabParams, BatchQueryCollabResult, CollabParams, CreateCollabParams,
@@ -210,6 +210,28 @@ impl Client {
     AppResponse::from_response(resp).await?.into_data()
   }
 
+  // Adds a database field to the specified database.
+  // Returns the field id of the newly created field.
+  pub async fn add_database_field(
+    &self,
+    workspace_id: &str,
+    database_id: &str,
+    insert_field: &AFInsertDatabaseField,
+  ) -> Result<String, AppResponseError> {
+    let url = format!(
+      "{}/api/workspace/{}/database/{}/fields",
+      self.base_url, workspace_id, database_id
+    );
+    let resp = self
+      .http_client_with_auth(Method::POST, &url)
+      .await?
+      .json(insert_field)
+      .send()
+      .await?;
+    log_request_id(&resp);
+    AppResponse::from_response(resp).await?.into_data()
+  }
+
   pub async fn list_database_row_ids_updated(
     &self,
     workspace_id: &str,
@@ -250,6 +272,32 @@ impl Client {
     AppResponse::from_response(resp).await?.into_data()
   }
 
+  /// Example payload:
+  /// {
+  ///   "Name": "some_data",        # using column name
+  ///   "_pIkG": "some other data"  # using field_id (can be obtained from [get_database_fields])
+  /// }
+  /// Upon success, returns the row id for the newly created row.
+  pub async fn add_database_item(
+    &self,
+    workspace_id: &str,
+    database_id: &str,
+    payload: &serde_json::Value,
+  ) -> Result<String, AppResponseError> {
+    let url = format!(
+      "{}/api/workspace/{}/database/{}/row",
+      self.base_url, workspace_id, database_id
+    );
+    let resp = self
+      .http_client_with_auth(Method::POST, &url)
+      .await?
+      .json(&payload)
+      .send()
+      .await?;
+    log_request_id(&resp);
+    AppResponse::from_response(resp).await?.into_data()
+  }
+
   #[instrument(level = "debug", skip_all, err)]
   pub async fn post_realtime_msg(
     &self,

libs/client-api/src/http_file.rs

@@ -290,7 +290,7 @@ impl WSClientConnectURLProvider for Client {
 ///
 /// # Returns
 /// A `Result` containing the base64-encoded MD5 hash on success, or an error if the file cannot be read.
-
+///
 /// Asynchronously calculates the MD5 hash of a file using efficient buffer handling and returns it as a base64-encoded string.
 ///
 /// # Arguments

libs/collab-stream/src/pubsub.rs

@@ -50,7 +50,7 @@ impl CollabStreamPub {
 
   #[instrument(level = "debug", skip_all, err)]
   pub async fn publish(&mut self, message: PubSubMessage) -> Result<(), StreamError> {
-    self.conn.publish(ACTIVE_COLLAB_CHANNEL, message).await?;
+    let () = self.conn.publish(ACTIVE_COLLAB_CHANNEL, message).await?;
     Ok(())
   }
 }

libs/collab-stream/src/stream.rs

@@ -2,7 +2,7 @@ use crate::error::StreamError;
 use crate::model::{MessageId, StreamBinary, StreamMessage, StreamMessageByStreamKey};
 use redis::aio::ConnectionManager;
 use redis::streams::{StreamMaxlen, StreamReadOptions};
-use redis::{pipe, AsyncCommands, RedisError};
+use redis::{pipe, AsyncCommands, Pipeline, RedisError};
 
 pub struct CollabStream {
   connection_manager: ConnectionManager,
@@ -34,9 +34,9 @@ impl CollabStream {
     let mut pipe = pipe();
     for message in messages {
       let tuple = message.into_tuple_array();
-      pipe.xadd(&self.stream_key, "*", tuple.as_slice());
+      let _: &mut Pipeline = pipe.xadd(&self.stream_key, "*", tuple.as_slice());
     }
-    pipe.query_async(&mut self.connection_manager).await?;
+    let () = pipe.query_async(&mut self.connection_manager).await?;
     Ok(())
   }
 
@@ -90,7 +90,7 @@ impl CollabStream {
   }
 
   pub async fn clear(&mut self) -> Result<(), RedisError> {
-    self
+    let () = self
       .connection_manager
       .xtrim(&self.stream_key, StreamMaxlen::Equals(0))
       .await?;

libs/collab-stream/src/stream_group.rs

@@ -7,7 +7,7 @@ use redis::streams::{
   StreamClaimOptions, StreamClaimReply, StreamMaxlen, StreamPendingData, StreamPendingReply,
   StreamReadOptions,
 };
-use redis::{pipe, AsyncCommands, ErrorKind, RedisResult};
+use redis::{pipe, AsyncCommands, ErrorKind, Pipeline, RedisResult};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, trace, warn};
 
@@ -119,7 +119,7 @@ impl StreamGroup {
       .into_iter()
       .map(|m| m.to_string())
       .collect::<Vec<String>>();
-    self
+    let () = self
       .connection_manager
       .xack(&self.stream_key, &self.group_name, &message_ids)
       .await?;
@@ -164,7 +164,7 @@ impl StreamGroup {
       let message = message.into();
       let tuple = message.into_tuple_array();
       if let Some(len) = self.config.max_len {
-        pipe
+        let _: &mut Pipeline = pipe
           .cmd("XADD")
           .arg(&self.stream_key)
           .arg("MAXLEN")
@@ -177,7 +177,7 @@ impl StreamGroup {
       }
     }
 
-    pipe.query_async(&mut self.connection_manager).await?;
+    let () = pipe.query_async(&mut self.connection_manager).await?;
     if let Err(err) = self.set_expiration().await {
       error!("set expiration fail: {:?}", err);
     }
@@ -197,13 +197,13 @@ impl StreamGroup {
     let tuple = message.into_tuple_array();
     match self.config.max_len {
       Some(max_len) => {
-        self
+        let () = self
           .connection_manager
           .xadd_maxlen(&self.stream_key, StreamMaxlen::Approx(max_len), "*", &tuple)
           .await?;
       },
       None => {
-        self
+        let () = self
           .connection_manager
           .xadd(&self.stream_key, "*", tuple.as_slice())
           .await?;
@@ -369,7 +369,7 @@ impl StreamGroup {
   /// Use the `XTRIM` command to truncate the Redis stream to a maximum length of zero, effectively
   /// removing all entries from the stream.
   pub async fn clear(&mut self) -> Result<(), StreamError> {
-    self
+    let () = self
       .connection_manager
       .xtrim(&self.stream_key, StreamMaxlen::Equals(0))
       .await?;
@@ -393,7 +393,7 @@ impl StreamGroup {
       };
 
       if should_set_expiration {
-        self
+        let () = self
           .connection_manager
           .expire(&self.stream_key, expire_time)
           .await?;

libs/database/src/collab/collab_db_ops.rs

@@ -43,7 +43,6 @@ use uuid::Uuid;
 /// * There's a database operation failure.
 /// * There's an attempt to insert a row with an existing `object_id` but a different `workspace_id`.
 ///
-
 #[inline]
 #[instrument(level = "trace", skip(tx, params), fields(oid=%params.object_id), err)]
 pub async fn insert_into_af_collab(

libs/shared-entity/src/dto/workspace_dto.rs

@@ -376,3 +376,10 @@ pub struct AFDatabaseField {
   pub type_option: HashMap<String, serde_json::Value>,
   pub is_primary: bool,
 }
+
+#[derive(Default, Debug, Clone, Serialize, Deserialize)]
+pub struct AFInsertDatabaseField {
+  pub name: String,
+  pub field_type: i64,                             // FieldType ID
+  pub type_option_data: Option<serde_json::Value>, // TypeOptionData
+}

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.78.0"
+channel = "1.83.0"

services/appflowy-collaborate/src/collab/cache/mem_cache.rs

@@ -32,7 +32,7 @@ impl CollabMemCache {
   pub async fn insert_collab_meta(&self, meta: CollabMetadata) -> Result<(), AppError> {
     let key = collab_meta_key(&meta.object_id);
     let value = serde_json::to_string(&meta)?;
-    self
+    let () = self
       .connection_manager
       .clone()
       .set_ex(key, value, ONE_MONTH)
@@ -188,7 +188,7 @@ impl CollabMemCache {
       // for executing a subsequent transaction (with MULTI/EXEC). If any of the watched keys are
       // altered by another client before the current client executes EXEC, the transaction will be
       // aborted by Redis (the EXEC will return nil indicating the transaction was not processed).
-      redis::cmd("WATCH")
+      let () = redis::cmd("WATCH")
         .arg(&cache_object_id)
         .query_async::<_, ()>(&mut conn)
         .await?;
@@ -228,7 +228,7 @@ impl CollabMemCache {
             .ignore()
             .expire(&cache_object_id, expiration_seconds.unwrap_or(SEVEN_DAYS) as i64) // Setting the expiration to 7 days
             .ignore();
-        pipeline.query_async(&mut conn).await?;
+        let () = pipeline.query_async(&mut conn).await?;
       }
       Ok::<(), redis::RedisError>(())
     }

services/appflowy-collaborate/src/group/manager.rs

@@ -179,7 +179,7 @@ where
       self.edit_state_max_secs,
       indexer,
     )?);
-    self.state.insert_group(object_id, group.clone());
+    self.state.insert_group(object_id, group);
     Ok(())
   }
 }