feat:slack connector

packages/apps/slack/.babelrc

@@ -0,0 +1,13 @@
+{
+  "plugins": [
+    "@babel/plugin-proposal-class-properties",
+    "@babel/plugin-transform-instanceof",
+    "@babel/plugin-transform-classes"
+  ],
+  "presets": ["@babel/preset-env"],
+  "env": {
+    "test": {
+      "plugins": ["@babel/plugin-transform-runtime"]
+    }
+  }
+} 

packages/apps/slack/.gitignore

@@ -0,0 +1,9 @@
+/lib
+node_modules
+.DS_store
+.env*
+/*.js
+!index.js
+yarn.lock
+
+/dist

packages/apps/slack/.npmignore

@@ -0,0 +1,8 @@
+.DS_store
+src
+dist
+yarn.lock
+.babelrc
+
+.turbo
+.yarn

packages/apps/slack/README.md

@@ -0,0 +1,24 @@
+## 🚀 Integrate Ocular with Slack
+
+### Features
+
+- Fetch Slack Channels and Messages directly into Ocular.
+
+### Installation Steps
+
+1. Open the `ocular/core-config.js` file.
+2. Add the following configuration snippet at the end of the `apps` array:
+
+```js
+const apps = [
+  // Other configurations...
+  {
+    resolve: "slack",
+    options: {
+      client_id: process.env.SLACK_CLIENT_ID,
+      client_secret: process.env.SLACK_CLIENT_SECRET,
+      redirect_uri: `${UI_CORS}/dashboard/marketplace/slack`,
+    },
+  },
+];
+```

packages/apps/slack/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "slack",
+  "version": "0.0.0",
+  "description": "Slack Application for Ocular",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "author": "Yash Khare",
+  "scripts": {
+    "prepare": "cross-env NODE_ENV=production npm run build",
+    "test": "jest --passWithNoTests src",
+    "build": "tsc",
+    "watch": "tsc --watch"
+  },
+  "devDependencies": {
+    "@babel/cli": "^7.24.1",
+    "@babel/core": "^7.24.1",
+    "@babel/preset-env": "^7.24.1",
+    "@babel/preset-react": "^7.23.3"
+  }
+}

packages/apps/slack/src/services/oauth.ts

@@ -0,0 +1,108 @@
+import axios from 'axios';
+import {
+  OauthService,
+  AppNameDefinitions,
+  AppCategoryDefinitions,
+  OAuthToken,
+} from '@ocular/types';
+import { ConfigModule } from '@ocular/ocular/src/types';
+
+class SlackOauth extends OauthService {
+  protected client_id_: string;
+  protected client_secret_: string;
+  protected configModule_: ConfigModule;
+  protected redirect_uri_: string;
+
+  constructor(container, options) {
+    super(arguments[0]);
+    this.client_id_ = options.client_id;
+    this.client_secret_ = options.client_secret;
+    this.redirect_uri_ = options.redirect_uri;
+    this.configModule_ = container.configModule;
+  }
+
+  static getAppDetails(projectConfig, options) {
+    const client_id = options.client_id;
+    const client_secret = options.client_secret;
+    const redirect = options.redirect_uri;
+    return {
+      name: AppNameDefinitions.SLACK,
+      logo: '/slack.svg',
+      description:
+        'Slack is a new way to communicate with your team. Its faster, better organised and more secure than email.',
+      oauth_url: `https://slack.com/oauth/v2/authorize?client_id=${client_id}&scope=app_mentions:read,channels:history,channels:join,channels:manage,channels:read,chat:write.customize,chat:write.public,chat:write,files:read,files:write,groups:history,groups:read,groups:write,im:history,im:read,im:write,links:read,links:write,mpim:history,mpim:read,mpim:write,pins:read,pins:write,reactions:read,reactions:write,reminders:read,reminders:write,team:read,usergroups:read,usergroups:write,users:read,users:write,users.profile:read&user_scope=`,
+      slug: AppNameDefinitions.SLACK,
+      category: AppCategoryDefinitions.PRODUCTIVITY,
+      developer: 'Ocular AI',
+      images: ['/slack.svg'],
+      overview:
+        'Slack is a new way to communicate with your team. Its faster, better organised and more secure than email.',
+      docs: 'https://api.slack.com/docs',
+      website: 'https://slack.com/',
+    };
+  }
+
+  async refreshToken(refresh_token: string): Promise<OAuthToken> {
+    const body = {
+      grant_type: 'refresh_token',
+      client_id: this.client_id_,
+      client_secret: this.client_secret_,
+      refresh_token: refresh_token,
+    };
+
+    const config = {
+      headers: {
+        'content-type': 'application/json',
+      },
+    };
+
+    return axios
+      .post('https:/slack.com/api/oauth.v2.exchange', body, config)
+      .then((res) => {
+        return {
+          token: res.data.access_token,
+          token_expires_at: new Date(Date.now() + res.data.expires_in * 1000),
+          refresh_token: res.data.refresh_token,
+        } as OAuthToken;
+      })
+      .catch((err) => {
+        console.error(err);
+        throw err;
+      });
+  }
+
+  async generateToken(code: string): Promise<OAuthToken> {
+    console.log('***** Generating token from the code:\n');
+
+    const body = {
+      grant_type: 'authorization_code',
+      client_id: `${this.client_id_}`,
+      client_secret: `${this.client_secret_}`,
+      code: code,
+      redirect_uri: `${this.redirect_uri_}`,
+    };
+
+    const config = {
+      headers: {
+        'Content-Type': 'application/json',
+      },
+    };
+
+    return axios
+      .post('https:/slack.com/api/oauth.v2.access', body, config)
+      .then((res) => {
+        return {
+          type: res.data.token_type,
+          token: res.data.access_token,
+          token_expires_at: new Date(Date.now() + res.data.expires_in * 1000),
+          refresh_token: res.data.refresh_token,
+        } as OAuthToken;
+      })
+      .catch((err) => {
+        console.error(err);
+        throw err;
+      });
+  }
+}
+
+export default SlackOauth;

packages/apps/slack/src/services/slack.ts

@@ -0,0 +1,166 @@
+import fs from "fs";
+import axios from "axios";
+import { Readable } from "stream";
+import { App, OAuthService, Organisation } from "@ocular/ocular";
+import {
+  IndexableDocument,
+  TransactionBaseService,
+  Logger,
+  AppNameDefinitions,
+  Section,
+} from "@ocular/types";
+import { ConfigModule } from "@ocular/ocular/src/types";
+import { DocType } from "@ocular/types/src/common";
+
+interface Config {
+  headers: {
+    Authorization: string;
+    Accept: string;
+  };
+}
+
+export default class SlackService extends TransactionBaseService {
+  protected oauthService_: OAuthService;
+  protected logger_: Logger;
+  protected container_: ConfigModule;
+
+  constructor(container) {
+    super(arguments[0]);
+    this.oauthService_ = container.oauthService;
+    this.logger_ = container.logger;
+    this.container_ = container;
+  }
+
+  async getSlackData(org: Organisation) {
+    return Readable.from(this.getSlackChannelsAndConversations(org));
+  }
+
+  async *getSlackChannelsAndConversations(
+    org: Organisation
+  ): AsyncGenerator<IndexableDocument[]> {
+    this.logger_.info(`Starting oculation of Slack for ${org.id} organisation`);
+
+    // Get Slack OAuth for the organisation
+    const oauth = await this.oauthService_.retrieve({
+      id: org.id,
+      app_name: AppNameDefinitions.SLACK,
+    });
+
+    if (!oauth) {
+      this.logger_.error(`No Slack OAuth found for ${org.id} organisation`);
+      return;
+    }
+
+    const config: Config = {
+      headers: {
+        Authorization: `Bearer ${oauth.token}`,
+        Accept: "application/json",
+      },
+    };
+
+    let documents: IndexableDocument[] = [];
+
+    try {
+      const slackChannels = await this.fetchSlackChannels(config)
+      for (const channel of slackChannels) {
+        const conversations = await this.fetchChannelConversations(channel.id,config);
+        for(const conversation of conversations){
+          const thread = await this.fetchThreadForConversation(channel.id,conversation.ts,config)
+
+          const sections: Section[] = thread.map((message, index) => ({
+            offset: index,
+            content: message.text,
+            link: `https://slack.com/api/conversations.replies?channel_id=${channel.id}&ts=${conversation.ts}` 
+          }));
+          const threadDoc : IndexableDocument = {
+            id:conversation.ts, // conversation id
+            organisationId:org.id,
+            source:AppNameDefinitions.SLACK,
+            title:conversation.text, // the main message which lead to conversation
+            metadata:{channel_id:channel.id}, // passing channel id just for top down reference
+            sections:sections, // an array of messages in the specific conversation
+            type: DocType.TEXT,
+            updatedAt: new Date(Date.now())
+          }
+          documents.push(threadDoc);
+          if (documents.length >= 100) {
+            yield documents;
+            documents = [];
+          }
+        }
+      }
+      yield documents;
+      await this.oauthService_.update(oauth.id, {
+        last_sync: new Date(),
+      });
+    } catch (error) {
+      if (error.response && error.response.status === 401) {
+        // Check if it's an unauthorized error
+        this.logger_.info(`Refreshing Slack token for ${org.id} organisation`);
+
+        // Refresh the token
+        const oauthToken = await this.container_["slackOauth"].refreshToken(
+          oauth.refresh_token
+        );
+
+        // Update the OAuth record with the new token
+        await this.oauthService_.update(oauth.id, oauthToken);
+
+        // Retry the request
+        return this.getSlackChannelsAndConversations(org);
+      } else {
+        console.error(error);
+      }
+    }
+
+    this.logger_.info(`Finished oculation of Slack for ${org.id} organisation`);
+  }
+
+  async fetchSlackChannels(config:Config){
+    try{
+      const response = await axios.get(
+        "https://slack.com/api/conversations.list",
+        config
+      )
+
+      if (!response.data)  {
+        return [];
+      }
+
+      const channels = response.data.channels
+      return channels
+    }catch(error){
+      this.logger_.error(
+        "Error fetching Slack Channels in fetchSlackChannels:",
+        error
+      );
+      throw error;
+    }
+  }
+
+  async fetchChannelConversations(channelID:string ,config:Config){
+    try{
+      const conversationsEndpoint = `https://slack.com/api/conversations.history?channel=${channelID}`
+      const response = await axios.get(conversationsEndpoint,config)
+      const conversationsArray = response.data.messages || []
+      const conversations = conversationsArray.map((conversations)=>({
+        id: conversations.ts,
+        text:conversations.text,
+        user: conversations.user
+      }))
+      return conversations
+    }catch(error){
+      throw new Error("Failed to fetch channel conversation.");
+    }
+  }
+
+  async fetchThreadForConversation(channelID:string, tsID:string, config:Config){
+    try{
+      const threadsEndpoint = `https://slack.com/api/conversations.replies?channel_id=${channelID}&ts=${tsID}`
+      const response = await axios.get(threadsEndpoint,config)
+      return response.data.messages
+    }catch(error){
+      throw new Error("Failed to fetch channel conversation.");
+    }
+  }
+}

packages/apps/slack/src/strategies/slack.ts

@@ -0,0 +1,36 @@
+import { BatchJobService, Organisation, EventBusService } from '@ocular/ocular';
+import SlackService from '../services/slack';
+import { INDEX_DOCUMENT_EVENT } from '@ocular/types';
+import { AbstractBatchJobStrategy } from '@ocular/types';
+
+export default class SlackStrategy extends AbstractBatchJobStrategy {
+  static identifier = 'slack-indexing-strategy';
+  static batchType = 'slack';
+  protected batchJobService_: BatchJobService;
+  protected slackService_: SlackService;
+  protected eventBusService_: EventBusService;
+
+  constructor(container) {
+    super(arguments[0]);
+    this.slackService_ = container.slackService;
+    this.batchJobService_ = container.batchJobService;
+    this.eventBusService_ = container.eventBusService;
+  }
+
+  async processJob(batchJobId: string): Promise<void> {
+    const batchJob = await this.batchJobService_.retrieve(batchJobId);
+    const stream = await this.slackService_.getSlackData(
+      batchJob.context?.org as Organisation
+    );
+    stream.on('data', (documents) => {
+      this.eventBusService_.emit(INDEX_DOCUMENT_EVENT, documents);
+    });
+    stream.on('end', () => {
+      console.log('No more data');
+    });
+  }
+
+  buildTemplate(): Promise<string> {
+    throw new Error('Method not implemented.');
+  }
+}