Feat: added sync functionality (#144)

* Feat: added sync functionality * Fix: jira and gmail to latest approach * fix: server * fix: jira * Fix: server remove limit * fix: change id to key for jira * Fix: server shouldn't react if no temporal is used --------- Co-authored-by: Manoj K <saimanoj58@gmail.com>
poozlehq · Jul 28, 2023 · 9ba7c11 · 9ba7c11
1 parent 0aed519
commit 9ba7c11
Show file tree

Hide file tree

Showing 150 changed files with 8,116 additions and 800 deletions.
diff --git a/.env b/.env
@@ -1,4 +1,4 @@
-ENGINE_VERSION=0.1.6-alpha
+ENGINE_VERSION=0.1.7-alpha
 
 # POSTGRES
 POSTGRES_USER=docker
@@ -14,6 +14,7 @@ SUPERTOKEN_SCHEMA=supertoken
 
 # Prisma database connection
 DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${DB_HOST}:${DB_PORT}/${POSTGRES_DB}?schema=${DB_SCHEMA}&sslmode=prefer
+SYNC_DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${DB_HOST}:${DB_PORT}/${POSTGRES_DB}
 
 # SuperToken database connection
 SUPERTOKEN_DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${DB_HOST}:${DB_PORT}/${POSTGRES_DB}?schema=${SUPERTOKEN_SCHEMA}&sslmode=prefer
@@ -31,3 +32,12 @@ BACKEND_HOST=http://localhost:3000
 BACKEND_URL=http://server:3000
 
 SUPERTOKEN_CONNECTION_URI=http://supertokens:3567
+
+
+TEMPORAL_ADDRESS=temporal:7233
+CONFIG_PATH=/app/temp
+TEMPORAL_VERSION=1.21.2
+TEMPORAL_UI_VERSION=2.16.1
+
+# Generate the token in UI and add it here
+TOKEN=
diff --git a/docker-compose.yaml b/docker-compose.yaml
@@ -21,19 +21,25 @@ services:
     container_name: poozle-server
     depends_on:
       - db
+      - temporal
+      - worker
+      - supertokens
     restart: unless-stopped
     networks:
       - poozle
     environment:
       - ENGINE_VERSION=${ENGINE_VERSION}
       - DATABASE_URL=${DATABASE_URL}
+      - SYNC_DATABASE_URL=${SYNC_DATABASE_URL}
       - FRONTEND_HOST=${FRONTEND_HOST}
       - PUBLIC_FRONTEND_HOST=${PUBLIC_FRONTEND_HOST}
       - BACKEND_HOST=${BACKEND_HOST}
       - JWT_ACCESS_SECRET=${JWT_ACCESS_SECRET}
       - JWT_REFRESH_SECRET=${JWT_REFRESH_SECRET}
       - PORT=${PORT}
       - SUPERTOKEN_CONNECTION_URI=${SUPERTOKEN_CONNECTION_URI}
+      - MASTER_TOKEN=${TOKEN}
+      - TEMPORAL_ADDRESS=${TEMPORAL_ADDRESS}
 
   webapp:
     image: poozlehq/engine-webapp:${ENGINE_VERSION}
@@ -49,6 +55,50 @@ services:
     ports:
       - 8000:3000
 
+  worker:
+    image: poozlehq/engine-worker:${ENGINE_VERSION}
+    platform: linux/amd64
+    container_name: poozle-worker
+    restart: unless-stopped
+    networks:
+      - poozle
+    environment:
+      - DATABASE_URL=${SYNC_DATABASE_URL}
+      - TEMPORAL_ADDRESS=${TEMPORAL_ADDRESS}
+      - DB_SCHEMA=${DB_SCHEMA}
+      - CONFIG_PATH=${CONFIG_PATH}
+      - TOKEN=${TOKEN}
+      - BACKEND_URL=${BACKEND_URL}
+
+  temporal:
+    image: temporalio/auto-setup:${TEMPORAL_VERSION}
+    container_name: temporal
+    depends_on:
+      - db
+    environment:
+      - DB=postgresql
+      - DB_PORT=${DB_PORT}
+      - POSTGRES_USER=${POSTGRES_USER}
+      - POSTGRES_PWD=${POSTGRES_PASSWORD}
+      - POSTGRES_SEEDS=${DB_HOST}
+    ports:
+      - 7233:7233
+    networks:
+      - poozle
+
+  temporal-ui:
+    container_name: temporal-ui
+    depends_on:
+      - temporal
+    environment:
+      - TEMPORAL_ADDRESS=${TEMPORAL_ADDRESS}
+      - TEMPORAL_CORS_ORIGINS=http://localhost:3000
+    image: temporalio/ui:${TEMPORAL_UI_VERSION}
+    networks:
+      - poozle
+    ports:
+      - 8080:8080
+
   supertokens:
     image: registry.supertokens.io/supertokens/supertokens-postgresql:5.0
     depends_on:

diff --git a/engine-idk/package.json b/engine-idk/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@poozle/engine-idk",
-  "version": "0.1.28",
+  "version": "0.1.29",
   "description": "Used to develop integrations for Poozle",
   "license": "MIT",
   "author": "Poozle <support@poozle.in>",

diff --git a/engine-idk/src/bases/base_integration.ts b/engine-idk/src/bases/base_integration.ts
@@ -55,6 +55,7 @@ export class BaseIntegration implements BaseIntegrationInterface {
           config,
         );
         headers = {
+          ...headers,
           Authorization: `Bearer ${token}`,
         };
       } else {

diff --git a/engine-idk/src/bases/base_path.ts b/engine-idk/src/bases/base_path.ts
@@ -30,14 +30,14 @@ export class BasePath {
   }
 
   async baseRun(method: string, headers: AxiosHeaders, params: Params, config: Config) {
-    const responseFromRun: any = await this.run(method, headers, params, config);
+    try {
+      const responseFromRun: any = await this.run(method, headers, params, config);
 
-    // if this is a request directly to the integration
-    if (params.proxy) {
-      return responseFromRun;
-    }
+      // if this is a request directly to the integration
+      if (params.proxy) {
+        return responseFromRun;
+      }
 
-    try {
       const response: any = {
         data: responseFromRun.data,
       };
@@ -86,7 +86,7 @@ export class BasePath {
     const previous = response.meta ? response.meta.previous ?? '' : '';
 
     return {
-      limit: params.queryParams?.limit ? parseInt(params.queryParams?.limit.toString()) : 10,
+      limit: params.queryParams?.limit as number,
       cursors: {
         previous,
         current,

diff --git a/engine-server/package.json b/engine-server/package.json
@@ -1,6 +1,6 @@
 {
   "name": "engine-server",
-  "version": "0.1.4-alpha",
+  "version": "0.1.7-alpha",
   "description": "Poozle engine",
   "author": "Poozle",
   "license": "MIT",
@@ -26,7 +26,7 @@
     "start:dev": "nest start --watch",
     "start:debug": "nest start --debug --watch",
     "start:prod": "node dist/main",
-    "start:migrate:prod": "yarn migrate:deploy && yarn seed && node dist/main",
+    "start:migrate:prod": "yarn migrate:deploy && node dist/main",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "migrate:dev": "prisma migrate dev --preview-feature",
     "migrate:dev:create": "prisma migrate dev --create-only --preview-feature",
@@ -62,16 +62,19 @@
     "@poozle/engine-idk": "^0.1.20",
     "@prisma/client": "^4.9.0",
     "@segment/analytics-node": "^1.0.0-beta.24",
+    "@temporalio/client": "^1.8.1",
     "@vegardit/prisma-generator-nestjs-dto": "^1.5.1",
     "axios": "^1.4.0",
     "bcrypt": "5.1.0",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.0",
     "cookie-parser": "^1.4.6",
     "date-fns": "^2.30.0",
+    "knex": "^2.5.1",
     "nestjs-prisma": "0.20.0",
     "passport": "0.6.0",
     "passport-jwt": "4.0.1",
+    "pg": "^8.11.1",
     "readline-sync": "^1.4.10",
     "reflect-metadata": "0.1.13",
     "rimraf": "4.1.1",

diff --git a/engine-server/prisma/dbml/schema.dbml b/engine-server/prisma/dbml/schema.dbml
@@ -81,6 +81,8 @@ Table IntegrationAccount {
   createdAt DateTime [default: `now()`, not null]
   updatedAt DateTime [not null]
   accountIdentifier String
+  syncEnabled Boolean [not null, default: false]
+  syncPeriod String
   fromLinks Link
   linkId String
 }
@@ -100,6 +102,25 @@ Table Link {
   IntegrationAccount IntegrationAccount [not null]
 }
 
+Table Job {
+  jobId String [pk]
+  temporalId String [not null]
+  integrationAccountId String [not null]
+  status JobStatus [not null]
+  failedMessage String
+  recordsSynced Int
+  logs String
+  createdAt DateTime [default: `now()`, not null]
+  updatedAt DateTime [not null]
+}
+
+Enum JobStatus {
+  PENDING
+  RUNNING
+  FAILED
+  SUCCEEDED
+}
+
 Enum ReleaseStage {
   ALPHA
   BETA

diff --git a/engine-server/prisma/migrations/20230724073043_added_logs_table/migration.sql b/engine-server/prisma/migrations/20230724073043_added_logs_table/migration.sql
@@ -0,0 +1,19 @@
+-- CreateEnum
+CREATE TYPE "JobStatus" AS ENUM ('PENDING', 'RUNNING', 'FAILED', 'SUCCEEDED');
+
+-- CreateTable
+CREATE TABLE "Job" (
+    "jobId" TEXT NOT NULL,
+    "temporalId" TEXT NOT NULL,
+    "integrationAccountId" TEXT NOT NULL,
+    "status" "JobStatus" NOT NULL,
+    "recordsSynced" INTEGER NOT NULL,
+    "logs" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "Job_pkey" PRIMARY KEY ("jobId")
+);
+
+-- AddForeignKey
+ALTER TABLE "Job" ADD CONSTRAINT "Job_integrationAccountId_fkey" FOREIGN KEY ("integrationAccountId") REFERENCES "IntegrationAccount"("integrationAccountId") ON DELETE RESTRICT ON UPDATE CASCADE;
diff --git a/engine-server/prisma/migrations/20230724073942_make_logs_optional/migration.sql b/engine-server/prisma/migrations/20230724073942_make_logs_optional/migration.sql
@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "Job" ALTER COLUMN "recordsSynced" DROP NOT NULL,
+ALTER COLUMN "logs" DROP NOT NULL;
diff --git a/engine-server/prisma/migrations/20230724174858_add_failed_message/migration.sql b/engine-server/prisma/migrations/20230724174858_add_failed_message/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Job" ADD COLUMN     "failedMessage" TEXT;
diff --git a/engine-server/prisma/migrations/20230725055339_add_sync_params/migration.sql b/engine-server/prisma/migrations/20230725055339_add_sync_params/migration.sql
@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "IntegrationAccount" ADD COLUMN     "syncEnabled" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN     "syncPeriod" TEXT;
diff --git a/engine-server/prisma/migrations/20230725204826_remove_foriegn_check_in_jobs/migration.sql b/engine-server/prisma/migrations/20230725204826_remove_foriegn_check_in_jobs/migration.sql
@@ -0,0 +1,2 @@
+-- DropForeignKey
+ALTER TABLE "Job" DROP CONSTRAINT "Job_integrationAccountId_fkey";
diff --git a/engine-server/prisma/schema.prisma b/engine-server/prisma/schema.prisma
@@ -101,8 +101,13 @@ model IntegrationAccount {
   createdAt                DateTime              @default(now())
   updatedAt                DateTime              @updatedAt
 
+  // to map to the customer organisations
   accountIdentifier String?
 
+  // Sync parameters
+  syncEnabled Boolean @default(false)
+  syncPeriod  String?
+
   fromLinks Link?   @relation(fields: [linkId], references: [linkId])
   linkId    String?
 }
@@ -128,6 +133,30 @@ model Link {
   IntegrationAccount IntegrationAccount[]
 }
 
+model Job {
+  jobId String @id @default(uuid())
+
+  // Used to match the id with in temporal
+  temporalId           String
+  integrationAccountId String
+  status               JobStatus
+
+  failedMessage String?
+
+  recordsSynced Int?
+  logs          String?
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+}
+
+enum JobStatus {
+  PENDING
+  RUNNING
+  FAILED
+  SUCCEEDED
+}
+
 enum ReleaseStage {
   ALPHA
   BETA

diff --git a/engine-server/prisma/seeds/user.ts b/engine-server/prisma/seeds/user.ts
@@ -75,7 +75,6 @@ async function main() {
       const accessToken = response.headers['st-access-token'];
 
       if (!accessToken) {
-        console.log(response.data);
         console.log('User not created');
         return;
       }

diff --git a/engine-server/src/@@generated/integrationAccount/entities/integrationAccount.entity.ts b/engine-server/src/@@generated/integrationAccount/entities/integrationAccount.entity.ts
@@ -34,12 +34,32 @@ export class IntegrationAccount {
    */
   workspaceId: string;
 
+  /**
+   * Name used to fetch data from database
+   */
+  workspaceName?: string;
+
   /**
    * This is used for User experience. You can pass a name
    * to easily identify the account in UI
    */
   integrationAccountName: string;
 
+  /**
+   * This is used to map organisations/user_id fo the customers
+   */
+  accountIdentifier?: string;
+
+  /**
+   * Boolean to check if syncing is enabled
+   */
+  syncEnabled: boolean;
+
+  /**
+   * Internal for sync
+   */
+  syncPeriod?: string;
+
   /**
    * When this account is deleted. If deleted
    */

diff --git a/engine-server/src/@@generated/job/entities/index.ts b/engine-server/src/@@generated/job/entities/index.ts
@@ -0,0 +1 @@
+export * from './job';
diff --git a/engine-server/src/@@generated/job/entities/job.ts b/engine-server/src/@@generated/job/entities/job.ts
@@ -0,0 +1,18 @@
+export enum JobStatus {
+  PENDING = 'PENDING',
+  RUNNING = 'RUNNING',
+  FAILED = 'FAILED',
+  SUCCEEDED = 'SUCCEEDED',
+}
+
+export interface Job {
+  jobId: string;
+  temporalId: string;
+  integrationAccountId: string;
+  status: JobStatus;
+  failedMessage: string | null;
+  recordsSynced: number | null;
+  logs: string | null;
+  createdAt: Date;
+  updatedAt: Date;
+}
diff --git a/engine-server/src/common/interfaces/pagination.interface.ts b/engine-server/src/common/interfaces/pagination.interface.ts
@@ -1,6 +1,7 @@
 /** Copyright (c) 2023, Poozle, all rights reserved. **/
 
 import { ApiProperty } from '@nestjs/swagger';
+import { Transform } from 'class-transformer';
 import { IsNumber, IsOptional, IsString } from 'class-validator';
 
 export class PaginationParams {
@@ -15,12 +16,13 @@ export class PaginationParams {
   cursor?: string;
 
   @ApiProperty({
-    name: 'cursor',
+    name: 'limit',
     type: 'number',
     description:
       'Number of results to return. Minimum 1, Maximum 200, Default 10',
   })
   @IsNumber()
   @IsOptional()
-  limit?: number;
+  @Transform(({ value }) => parseInt(value))
+  limit = 10;
 }
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		-- AlterTable
		ALTER TABLE "Job" ADD COLUMN "failedMessage" TEXT;
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		-- DropForeignKey
		ALTER TABLE "Job" DROP CONSTRAINT "Job_integrationAccountId_fkey";