feat: backend of slack context provider integration

ee/tabby-db/Cargo.toml

@@ -27,6 +27,7 @@ tokio = { workspace = true, features = ["fs"] }
 uuid.workspace = true
 cached = { workspace = true, features = ["async"] }
 serde.workspace = true
+serde_json.workspace = true
 
 [dev-dependencies]
 assert_matches.workspace = true

ee/tabby-db/migrations/0039_slack-workspaces.down.sql

@@ -0,0 +1 @@
+DROP TABLE slack_workspaces;

ee/tabby-db/migrations/0039_slack-workspaces.up.sql

@@ -0,0 +1,8 @@
+CREATE TABLE slack_workspaces(
+  id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+  workspace_name VARCHAR(255) NOT NULL,
+  bot_token TEXT NOT NULL,
+  channels TEXT NOT NULL,
+  created_at TIMESTAMP NOT NULL DEFAULT(DATETIME('now')),
+  updated_at TIMESTAMP NOT NULL DEFAULT(DATETIME('now'))
+);

ee/tabby-db/src/lib.rs

@@ -38,6 +38,7 @@ mod provided_repositories;
 mod refresh_tokens;
 mod repositories;
 mod server_setting;
+pub mod slack_workspaces;
 mod threads;
 mod user_completions;
 mod user_events;

ee/tabby-db/src/slack_workspaces.rs

@@ -0,0 +1,169 @@
+use anyhow::{anyhow, Result};
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sqlx::{prelude::FromRow, query, query_as};
+use tabby_db_macros::query_paged_as;
+
+use crate::DbConn;
+
+#[derive(Debug, FromRow, Serialize, Deserialize, Clone)]
+pub struct SlackWorkspaceDAO {
+    pub id: i64,
+    pub workspace_name: String,
+    pub bot_token: String,
+    pub channels: serde_json::Value,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+impl SlackWorkspaceDAO {
+    pub fn get_channels(&self) -> Result<Vec<String>, serde_json::Error> {
+        serde_json::from_value(self.channels.clone())
+    }
+}
+
+impl DbConn {
+    pub async fn list_slack_workspace(
+        &self,
+        ids: Option<Vec<i64>>,
+        limit: Option<usize>,
+        skip_id: Option<i32>,
+        backwards: bool,
+    ) -> Result<Vec<SlackWorkspaceDAO>> {
+        let mut conditions = vec![];
+        if let Some(ids) = ids {
+            let ids: Vec<String> = ids.iter().map(i64::to_string).collect();
+            let ids = ids.join(", ");
+            conditions.push(format!("id in ({ids})"));
+        }
+        let condition = (!conditions.is_empty()).then_some(conditions.join(" AND "));
+        let integrations = query_paged_as!(
+            SlackWorkspaceDAO,
+            "slack_workspaces",
+            [
+                "id",
+                "workspace_name",
+                "bot_token",
+                "channels",
+                "created_at" as "created_at: DateTime<Utc>",
+                "updated_at" as "updated_at: DateTime<Utc>"
+            ],
+            limit,
+            skip_id,
+            backwards,
+            condition
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        Ok(integrations)
+    }
+
+    pub async fn create_slack_workspace(
+        &self,
+        workspace_name: String,
+        bot_token: String,
+        channels: Option<Vec<String>>,
+    ) -> Result<i64> {
+        let channels_json = serde_json::to_value(channels.unwrap_or_default())?;
+
+        let res = query!(
+            "INSERT INTO slack_workspaces(workspace_name, bot_token, channels) VALUES (?, ?, ?);",
+            workspace_name,
+            bot_token,
+            channels_json
+        )
+        .execute(&self.pool)
+        .await?;
+
+        Ok(res.last_insert_rowid())
+    }
+
+    pub async fn delete_slack_workspace(&self, id: i64) -> Result<bool> {
+        query!("DELETE FROM slack_workspaces WHERE id = ?;", id)
+            .execute(&self.pool)
+            .await?;
+        Ok(true)
+    }
+    pub async fn get_slack_workspace(&self, id: i64) -> Result<Option<SlackWorkspaceDAO>> {
+        let integration = query_as!(
+            SlackWorkspaceDAO,
+            r#"SELECT 
+                id, 
+                workspace_name, 
+                bot_token, 
+                channels as "channels: Value",
+                created_at as "created_at!: DateTime<Utc>",
+                updated_at as "updated_at!: DateTime<Utc>"
+            FROM slack_workspaces 
+            WHERE id = ?"#,
+            id
+        )
+        .fetch_optional(&self.pool)
+        .await?;
+
+        Ok(integration)
+    }
+
+    pub async fn update_slack_workspace(
+        &self,
+        id: i64,
+        workspace_name: String,
+        bot_token: String,
+        channels: Option<Vec<String>>,
+    ) -> Result<()> {
+        let channels_json = serde_json::to_value(channels.unwrap_or_default())?;
+        let rows = query!(
+            "UPDATE slack_workspaces 
+             SET workspace_name = ?, bot_token = ?, channels = ? 
+             WHERE id = ?",
+            workspace_name,
+            bot_token,
+            channels_json,
+            id
+        )
+        .execute(&self.pool)
+        .await?;
+
+        if rows.rows_affected() == 1 {
+            Ok(())
+        } else {
+            Err(anyhow!("failed to update: slack workspace not found"))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use crate::DbConn;
+
+    #[tokio::test]
+    async fn test_update_slack_workspace() {
+        let conn = DbConn::new_in_memory().await.unwrap();
+
+        let channels = Some(vec!["channel1".to_string(), "channel2".to_string()]);
+        let id = conn
+            .create_slack_workspace("test_workspace".into(), "xoxb-test-token".into(), channels)
+            .await
+            .unwrap();
+
+        let workspace = conn.get_slack_workspace(id).await.unwrap().unwrap();
+        assert_eq!(workspace.workspace_name, "test_workspace");
+
+        let new_channels = Some(vec!["new_channel1".to_string(), "new_channel2".to_string()]);
+        conn.update_slack_workspace(
+            id,
+            "updated_workspace".into(),
+            "xoxb-new-token".into(),
+            new_channels,
+        )
+        .await
+        .unwrap();
+
+        let updated_workspace = conn.get_slack_workspace(id).await.unwrap().unwrap();
+        assert_eq!(updated_workspace.workspace_name, "updated_workspace");
+        assert_eq!(updated_workspace.bot_token, "xoxb-new-token");
+    }
+}

ee/tabby-schema/graphql/schema.graphql

@@ -126,6 +126,12 @@ input CreateMessageInput {
   attachments: MessageAttachmentInput
 }
 
+input CreateSlackWorkspaceInput {
+  workspaceName: String!
+  botToken: String!
+  channelIds: [String!]
+}
+
 input CreateThreadAndRunInput {
   thread: CreateThreadInput!
   options: ThreadRunOptionsInput! = {codeQuery: null, debugOptions: null, docQuery: null, generateRelevantQuestions: false}
@@ -572,6 +578,8 @@ type Mutation {
   deleteUserGroupMembership(userGroupId: ID!, userId: ID!): Boolean!
   grantSourceIdReadAccess(sourceId: String!, userGroupId: ID!): Boolean!
   revokeSourceIdReadAccess(sourceId: String!, userGroupId: ID!): Boolean!
+  createSlackWorkspace(input: CreateSlackWorkspaceInput!): Boolean!
+  deleteSlackWorkspaceIntegration(id: ID!): Boolean!
 }
 
 type NetworkSetting {
@@ -693,6 +701,8 @@ type Query {
   "List user groups."
   userGroups: [UserGroup!]!
   sourceIdAccessPolicies(sourceId: String!): SourceIdAccessPolicy!
+  slackWorkspaces(ids: [ID!], after: String, before: String, first: Int, last: Int): SlackWorkspaceConnection!
+  slackChannels(botToken: String!): [SlackChannel!]!
 }
 
 type RefreshTokenResponse {
@@ -746,6 +756,31 @@ type ServerInfo {
   isDemoMode: Boolean!
 }
 
+type SlackChannel {
+  id: String!
+  name: String!
+}
+
+type SlackWorkspace {
+  id: ID!
+  botToken: String!
+  workspaceName: String!
+  channels: [String!]
+  createdAt: DateTime!
+  updatedAt: DateTime!
+  jobInfo: JobInfo!
+}
+
+type SlackWorkspaceConnection {
+  edges: [SlackWorkspaceEdge!]!
+  pageInfo: PageInfo!
+}
+
+type SlackWorkspaceEdge {
+  node: SlackWorkspace!
+  cursor: String!
+}
+
 type SourceIdAccessPolicy {
   sourceId: String!
   read: [UserGroup!]!

ee/tabby-schema/src/schema/mod.rs

@@ -10,12 +10,12 @@ pub mod job;
 pub mod license;
 pub mod repository;
 pub mod setting;
+pub mod slack_workspaces;
 pub mod thread;
 pub mod user_event;
 pub mod user_group;
 pub mod web_documents;
 pub mod worker;
-
 use std::sync::Arc;
 
 use access_policy::{AccessPolicyService, SourceIdAccessPolicy};
@@ -33,6 +33,7 @@ use juniper::{
     Object, RootNode, ScalarValue, Value, ID,
 };
 use repository::RepositoryGrepOutput;
+use slack_workspaces::{CreateSlackWorkspaceInput, SlackChannel, SlackWorkspaceService};
 use tabby_common::api::{code::CodeSearch, event::EventLogger};
 use thread::{CreateThreadAndRunInput, CreateThreadRunInput, ThreadRunStream, ThreadService};
 use tracing::{error, warn};
@@ -86,6 +87,7 @@ pub trait ServiceLocator: Send + Sync {
     fn context(&self) -> Arc<dyn ContextService>;
     fn user_group(&self) -> Arc<dyn UserGroupService>;
     fn access_policy(&self) -> Arc<dyn AccessPolicyService>;
+    fn slack(&self) -> Arc<dyn SlackWorkspaceService>;
 }
 
 pub struct Context {
@@ -658,6 +660,36 @@ impl Query {
 
         Ok(SourceIdAccessPolicy { source_id, read })
     }
+
+    //list all slack workspace with selected channel
+    pub async fn slack_workspaces(
+        ctx: &Context,
+        ids: Option<Vec<ID>>,
+        after: Option<String>,
+        before: Option<String>,
+        first: Option<i32>,
+        last: Option<i32>,
+    ) -> Result<relay::Connection<slack_workspaces::SlackWorkspace>> {
+        relay::query_async(
+            after,
+            before,
+            first,
+            last,
+            |after, before, first, last| async move {
+                ctx.locator
+                    .slack()
+                    .list(ids, after, before, first, last)
+                    .await
+            },
+        )
+        .await
+    }
+
+    pub async fn slack_channels(ctx: &Context, bot_token: String) -> Result<Vec<SlackChannel>> {
+        check_admin(ctx).await?;
+        let res = ctx.locator.slack().list_visible_channels(bot_token).await?;
+        Ok(res)
+    }
 }
 
 #[derive(GraphQLObject)]
@@ -1147,6 +1179,25 @@ impl Mutation {
             .await?;
         Ok(true)
     }
+
+    async fn create_slack_workspace(
+        ctx: &Context,
+        input: CreateSlackWorkspaceInput,
+    ) -> Result<bool> {
+        check_admin(ctx).await?;
+        input.validate()?;
+        ctx.locator
+            .slack()
+            .create(input.workspace_name, input.bot_token, input.channel_ids)
+            .await?;
+        Ok(true)
+    }
+
+    async fn delete_slack_workspace_integration(ctx: &Context, id: ID) -> Result<bool> {
+        check_admin(ctx).await?;
+        ctx.locator.slack().delete(id).await?;
+        Ok(true)
+    }
 }
 
 fn from_validation_errors<S: ScalarValue>(error: ValidationErrors) -> FieldError<S> {