[SILO-452] feat: Queue multiple job runs in importers (#3945)

* modify: queue multiple import jobs and bug fixes

* moved queue job logic inside migrator + code refactoring

* fix: code improvement

Author: Saurabhkmr98

apps/api/plane/ee/serializers/api/issue_property.py

@@ -12,7 +12,7 @@
 
 class IssuePropertyAPISerializer(BaseSerializer):
     relation_type = serializers.ChoiceField(
-        choices=RelationTypeEnum.choices, required=False
+        choices=RelationTypeEnum.choices, required=False, allow_null=True
     )
 
     class Meta:

apps/api/plane/ee/views/api/job/base.py

@@ -54,19 +54,28 @@ def post(self, request):
     def get(self, request, pk=None):
         if not pk:
             filters = {}
+            order_by = "-created_at"
             # Handle source parameter
             if "source" in request.query_params:
                 filters["source"] = request.query_params["source"]
 
+            # Handle status parameter
+            if "statuses" in request.query_params:
+                filters["status__in"] = request.query_params["statuses"].split(",")
+
             # Handle workspace_id/workspaceId parameter
             if "workspace_id" in request.query_params:
                 filters["workspace_id"] = request.query_params["workspace_id"]
             elif "workspaceId" in request.query_params:
                 filters["workspace_id"] = request.query_params["workspaceId"]
 
+            # Handle order_by parameter
+            if "order_by" in request.query_params:
+                order_by = request.query_params["order_by"]
+
             import_jobs = (
                 ImportJob.objects.filter(**filters)
-                .order_by("-created_at")
+                .order_by(order_by)
                 .select_related("workspace", "report")
             )
             serializer = ImportJobAPISerializer(import_jobs, many=True)

apps/silo/src/apps/engine/controllers/job.controller.ts

@@ -8,15 +8,16 @@
 
 import { Request, Response } from "express";
 
-import { TImporterKeys, TIntegrationKeys } from "@plane/etl/core";
-import { resetJobIfStarted } from "@/helpers/job";
+import { E_JOB_STATUS, TImporterKeys, TIntegrationKeys } from "@plane/etl/core";
 import { responseHandler } from "@/helpers/response-handler";
-import { Controller, Get, Post, Put, useValidateUserAuthentication } from "@/lib";
+import { APIError, Controller, Get, Post, Put, useValidateUserAuthentication } from "@/lib";
 import { logger } from "@/logger";
 import { getAPIClient } from "@/services/client";
 import { importTaskManger } from "@/worker";
+import { JobService } from "../services/job.service";
 
 const client = getAPIClient();
+const jobService = new JobService();
 
 @Controller("/api/jobs")
 export class JobController {
@@ -120,13 +121,13 @@ export class JobController {
 
       // Get the job from the given job id
       const job = await client.importJob.getImportJob(body.jobId);
-      if ((job.status && job.status === "FINISHED") || job.status === "ERROR") {
+      if ((job.status && job.status === E_JOB_STATUS.FINISHED) || job.status === E_JOB_STATUS.ERROR) {
         res.status(400).json({ message: "Job already finished or errored out, can't cancel" });
         return;
       }
 
       await client.importJob.updateImportJob(body.jobId, {
-        status: "CANCELLED",
+        status: E_JOB_STATUS.CANCELLED,
         cancelled_at: new Date().toISOString(),
       });
       res.status(200).json({ message: "Job cancelled successfully" });
@@ -157,6 +158,7 @@ export class JobController {
         },
         { phase, isLastBatch }
       );
+
       res.status(200).json({ message: "Job updated successfully" });
     } catch (error: any) {
       responseHandler(res, 500, error);
@@ -182,48 +184,14 @@ export class JobController {
         return;
       }
 
-      // Get the job from the given job id
-      const job = await client.importJob.getImportJob(body.jobId);
-      // If the job is not finished or error, just send 400 OK, and don't do
-      // anything
-      if (
-        job.status &&
-        job.status != "CREATED" &&
-        job.status != "FINISHED" &&
-        job.status != "ERROR" &&
-        job.status != "CANCELLED"
-      ) {
-        res.status(400).json({ message: "Job already in progress, can't instantiate again" });
-        return;
-      }
-      // Check if the config is already present, for the particular job or not
-      if (!job.config || job.source == null) {
-        res.status(400).json({
-          message: "Config for the requested job is not found, make sure to create a config before initiating a job",
-        });
-        return;
-      }
-      logger.info(`[${job.id}] Initiating job ${job.source}}`);
-
-      await client.importJob.updateImportJob(job.id, {
-        status: "CREATED",
-        cancelled_at: null,
-        error_metadata: {},
-      });
-
-      await resetJobIfStarted(job);
-
-      await importTaskManger.registerTask(
-        {
-          route: job.source.toLowerCase(),
-          jobId: job.id,
-          type: "initiate",
-        },
-        {}
-      );
+      await jobService.runJob(body.jobId);
 
       res.status(200).json({ message: "Job initiated successfully" });
     } catch (error: any) {
+      if (error instanceof APIError) {
+        res.status(error.statusCode).json({ message: error.message });
+        return;
+      }
       responseHandler(res, 500, error);
     }
   }

apps/silo/src/apps/engine/services/job.service.ts

@@ -0,0 +1,74 @@
+import { E_JOB_STATUS } from "@plane/etl/core";
+import { resetJobIfStarted } from "@/helpers/job";
+import { APIError } from "@/lib";
+import { logger } from "@/logger";
+import { getAPIClient } from "@/services/client";
+import { importTaskManger } from "@/worker";
+
+const JOB_IN_PROGRESS_STATUSES = [
+  E_JOB_STATUS.INITIATED,
+  E_JOB_STATUS.PULLED,
+  E_JOB_STATUS.PULLING,
+  E_JOB_STATUS.PROGRESSING,
+  E_JOB_STATUS.TRANSFORMING,
+  E_JOB_STATUS.TRANSFORMED,
+  E_JOB_STATUS.PUSHING,
+];
+
+const JOB_ALLOWED_FOR_RE_RUN_STATUSES = [
+  E_JOB_STATUS.CREATED,
+  E_JOB_STATUS.QUEUED,
+  E_JOB_STATUS.CANCELLED,
+  E_JOB_STATUS.ERROR,
+  E_JOB_STATUS.FINISHED,
+];
+
+export class JobService {
+  private readonly client = getAPIClient();
+
+  async runJob(jobId: string) {
+    // Get the job from the given job id
+    const job = await this.client.importJob.getImportJob(jobId);
+    const jobsInProgress = await this.client.importJob.listImportJobs({
+      workspace_id: job.workspace_id,
+      source: job.source,
+      statuses: JOB_IN_PROGRESS_STATUSES.join(","),
+    });
+    const isAnyJobsInProgress = jobsInProgress.length > 0;
+    // If the job is not finished or error, just send 400 OK, and don't do anything
+    if (job.status && !JOB_ALLOWED_FOR_RE_RUN_STATUSES.includes(job.status as E_JOB_STATUS)) {
+      throw new APIError("Job is not in a valid status to run, can't instantiate again", 400);
+    }
+    // Check if the config is already present, for the particular job or not
+    if (!job.config || job.source == null) {
+      throw new APIError(
+        "Config for the requested job is not found, make sure to create a config before initiating a job",
+        400
+      );
+    }
+
+    // Update the job status to initiated or queued if there are any jobs started
+    await this.client.importJob.updateImportJob(job.id, {
+      status: !isAnyJobsInProgress ? E_JOB_STATUS.INITIATED : E_JOB_STATUS.QUEUED,
+      cancelled_at: null,
+      error_metadata: {},
+    });
+
+    // If there are no jobs created, then we need to register the task to initiate the job
+    if (!isAnyJobsInProgress) {
+      logger.info(`Initiating import job`, { jobId: job.id, source: job.source });
+      // Reset the job if it has already started
+      await resetJobIfStarted(job);
+      await importTaskManger.registerTask(
+        {
+          route: job.source.toLowerCase(),
+          jobId: job.id,
+          type: "initiate",
+        },
+        {}
+      );
+    }
+
+    return job;
+  }
+}

apps/silo/src/etl/base-import-worker.ts

@@ -1,5 +1,6 @@
-import { TJobStatus, PlaneEntities } from "@plane/etl/core";
+import { TJobStatus, PlaneEntities, E_JOB_STATUS } from "@plane/etl/core";
 import { TImportJob } from "@plane/types";
+import { JobService } from "@/apps/engine/services/job.service";
 import { IMPORT_JOB_FIRST_PAGE_PUSHED_CACHE_KEY } from "@/helpers/cache-keys";
 import { wait } from "@/helpers/delay";
 import { updateJobWithReport } from "@/helpers/job";
@@ -67,6 +68,22 @@ export abstract class BaseDataMigrator<TJobConfig, TSourceEntity> implements Tas
         completed_batch_count: report.total_batch_count,
       }),
     ]);
+
+    // start the queued jobs
+    const queuedJobs = await client.importJob.listImportJobs({
+      workspace_id: job.workspace_id,
+      source: job.source,
+      statuses: E_JOB_STATUS.QUEUED,
+      order_by: "created_at",
+    });
+    const nextJob = queuedJobs[0];
+    const isAnyJobsQueued = queuedJobs.length > 0;
+
+    if (isAnyJobsQueued) {
+      const jobService = new JobService();
+      logger.info(`Starting next job for ${job.source}`, { jobId: job.id, nextJobId: nextJob.id });
+      await jobService.runJob(nextJob.id);
+    }
   }
 
   async handleTask(headers: TaskHeaders, data: any): Promise<boolean> {

apps/silo/src/lib/errors.ts

@@ -1,6 +1,6 @@
 import { AxiosError, RawAxiosResponseHeaders } from "axios";
 import { wait } from "@/helpers/delay";
-import { logger } from "@/logger";
+import { captureException, logger } from "@/logger";
 
 export type APIRatelimitResponse = {
   error_code: number;
@@ -26,11 +26,6 @@ interface RateLimitHeaders {
   "x-ratelimit-reset": string;
 }
 
-interface APIRateLimitError extends Error {
-  response?: {
-    headers?: Partial<RateLimitHeaders>;
-  };
-}
 
 function isRateLimitHeaders(headers: any): headers is RateLimitHeaders {
   return (
@@ -65,6 +60,7 @@ export async function protect<T>(fn: (...args: any[]) => Promise<T>, ...args: an
     try {
       return await fn(...args);
     } catch (error) {
+      captureException(error as Error);
       // Check if the error is an Axios error with status 429
       if (error instanceof AxiosError && error.response?.status === 429) {
         logger.info("Rate limit exceeded ====== in protect");

apps/silo/src/services/job/import-job.service.ts

@@ -4,6 +4,12 @@ import { logger } from "@/logger";
 import { APIService } from "@/services/api.service";
 // types
 import { ClientOptions } from "@/types";
+
+export type TImportJobListQueryParams = Partial<Record<keyof TImportJob, string | boolean | number>> & {
+  statuses?: string;
+  order_by?: "created_at" | "updated_at" | "status";
+};
+
 export class ImportJobAPIService<TJobConfig = object> extends APIService {
   constructor(options: ClientOptions) {
     super(options);
@@ -38,7 +44,7 @@ export class ImportJobAPIService<TJobConfig = object> extends APIService {
       });
   }
 
-  async listImportJobs(params?: Partial<Record<keyof TImportJob, string | boolean | number>>): Promise<TImportJob[]> {
+  async listImportJobs(params?: TImportJobListQueryParams): Promise<TImportJob[]> {
     params = removeUndefinedFromObject(params);
     return this.get(`/api/v1/import-jobs/`, { params: params })
       .then((response) => response.data)

apps/web/ee/components/importers/common/dashboard/base-dashboard.tsx

@@ -97,7 +97,7 @@ export const BaseDashboard = observer(<T,>(props: IBaseDashboardProps<T>) => {
   const isReRunDisabled = (job: any) => {
     if (!job || !job?.status) return true;
 
-    return ![E_JOB_STATUS.CREATED, E_JOB_STATUS.FINISHED, E_JOB_STATUS.ERROR, E_JOB_STATUS.CANCELLED].includes(
+    return ![E_JOB_STATUS.CREATED, E_JOB_STATUS.FINISHED, E_JOB_STATUS.ERROR, E_JOB_STATUS.CANCELLED, E_JOB_STATUS.QUEUED].includes(
       job?.status as E_JOB_STATUS
     );
   };

apps/web/ee/components/importers/common/dashboard/status.tsx

@@ -13,6 +13,7 @@ type TSyncJobStatusProps = {
 
 const STATUS_CLASSNAMES: { [key in TJobStatus]: string } = {
   [E_JOB_STATUS.CREATED]: "text-gray-500 border border-gray-500 bg-gray-500/10",
+  [E_JOB_STATUS.QUEUED]: "text-gray-500 border border-gray-500 bg-gray-500/10",
   [E_JOB_STATUS.INITIATED]: "text-gray-500 border border-gray-500 bg-gray-500/10",
   [E_JOB_STATUS.PULLING]: "text-yellow-500 border border-yellow-500 bg-yellow-500/10",
   [E_JOB_STATUS.PULLED]: "text-yellow-500 border border-yellow-500 bg-yellow-500/10",

apps/web/ee/store/importers/clickup/root.store.ts

@@ -41,6 +41,11 @@ const defaultImporterData: TImporterClickUpDataPayload = {
   },
 };
 
+const defaultConfigData: Partial<TClickUpConfig> = {
+  skipAdditionalDataImport: true,
+  skipUserImport: false,
+};
+
 export interface IClickUpStore extends IImporterBaseStore {
   // observables
   dashboardView: boolean;
@@ -71,7 +76,7 @@ export class ClickUpStore extends ImporterBaseStore implements IClickUpStore {
   dashboardView: boolean = true;
   stepper: TClickUpImporterStepKeys = E_CLICKUP_IMPORTER_STEPS.CONFIGURE_CLICKUP;
   importerData: TImporterClickUpDataPayload = defaultImporterData;
-  configData: Partial<TClickUpConfig> = {};
+  configData: Partial<TClickUpConfig> = defaultConfigData;
   // store instances
   job: IImporterJobStore<TClickUpConfig>;
   auth: IClickUpAuthStore;
@@ -182,6 +187,6 @@ export class ClickUpStore extends ImporterBaseStore implements IClickUpStore {
     this.dashboardView = true;
     this.stepper = E_CLICKUP_IMPORTER_STEPS.CONFIGURE_CLICKUP;
     this.importerData = defaultImporterData;
-    this.configData = {};
+    this.configData = defaultConfigData;
   };
 }