[SILO-389] Clickup additional logs and bug fix for phase 2 not starting (#3671)

* fix status mapping in phase 1 + handle race condn + addn logs

* enhancement: increase wait time for requeuing pages and handle empty paginated batches in BaseDataMigrator

apps/api/plane/bgtasks/data_import_task.py

@@ -87,8 +87,25 @@ def update_job_batch_completion(
         # Refresh the object to get updated values
         job.report.refresh_from_db()
 
+        logger.info(
+            "Updated job batch completion",
+            extra={
+                "jobId": job_id,
+                "isLastBatch": is_last_batch,
+                "phase": phase,
+                "imported_batch_count": imported_batch_count,
+                "total_issues": total_issues,
+                "imported_issues": imported_issues,
+                "completed_batch_count": job.report.completed_batch_count,
+                "total_batch_count": job.report.total_batch_count,
+            },
+        )
+
         # Check if all batches are processed and update job status
-        if job.report.completed_batch_count >= job.report.total_batch_count:
+        if (
+            job.report.completed_batch_count >= job.report.total_batch_count
+            or is_last_batch
+        ):
             """
                 We'll make an api call to silo, such that silo can perform
                 any additional processing along with marking the job as finished
@@ -631,8 +648,14 @@ def import_data(slug, project_id, user_id, job_id, payload):
         job_phase = payload.get("phase", "issues")
         is_last_batch = payload.get("isLastBatch", False)
 
-        print(
-            f"inside import_data task for job {job_id} phase {job_phase} issue_count {len(issue_list or [])} lastBatchData {payload.get('isLastBatch', 'None')}"
+        logger.info(
+            "inside import_data task",
+            extra={
+                "jobId": job_id,
+                "phase": job_phase,
+                "issueCount": len(issue_list or []),
+                "lastBatchData": is_last_batch,
+            },
         )
 
         if issue_list:

apps/silo/package.json

@@ -6,8 +6,7 @@
   "author": "engineering@plane.so",
   "type": "module",
   "scripts": {
-    "dev": "turbo run develop",
-    "develop": "nodemon --config \"./nodemon.json\"/",
+    "dev": "nodemon --config \"./nodemon.json\"/",
     "build": "rm -rf dist && tsc --noEmit && tsup --minify",
     "start": "node dist/start.cjs",
     "test": "jest",

apps/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts

@@ -165,6 +165,12 @@ export class ClickUpDataMigrator extends BaseDataMigrator<TClickUpConfig, TClick
       total_batch_count: batches.length,
     });
 
+    logger.info("[CLICKUP] Batching completed", {
+      jobId: job.id,
+      batchCount: batches.length,
+      paginationContext: currentPageData,
+    });
+
     return finalBatches;
   }
 
@@ -294,6 +300,8 @@ export class ClickUpDataMigrator extends BaseDataMigrator<TClickUpConfig, TClick
 
       // verify availablity of states and plane project update the job config if required
       const isStatesAndProjectAvailable = await clickUpBulkTransformer.verifyStatesAndProject();
+      // clear the cache for the job after updating with states and project
+      await this.store.del(`job:${job.id}:data`);
       if (!isStatesAndProjectAvailable) {
         logger.error(`States and project not available, skipping transformation`, { jobId: job.id });
         return {} as TBatchPullResult<TClickUpEntity>;

apps/silo/src/apps/clickup-importer/migrator/transformers/etl.ts

@@ -63,7 +63,7 @@ export class ClickUpBulkTransformer {
   private readonly teamId: string;
   private readonly spaceId: string;
   private readonly folderId: string;
-  private readonly stateMap: TClickUpStateConfig[];
+  private stateMap: TClickUpStateConfig[];
   private readonly priorityMap: TClickUpPriorityConfig[];
   private readonly planeClient: PlaneClient;
   private readonly clickupService: ClickupAPIService;
@@ -136,6 +136,10 @@ export class ClickUpBulkTransformer {
         existingStates.results
       );
       planeStateMapping = stateMapping as TClickUpStateConfig[];
+      if (planeStateMapping.length !== 0) {
+        this.job.config.state = planeStateMapping;
+        this.stateMap = planeStateMapping;
+      }
     }
 
     // updating the job config

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

@@ -12,7 +12,7 @@ import { migrateToPlane } from "./migrator";
 
 export abstract class BaseDataMigrator<TJobConfig, TSourceEntity> implements TaskHandler {
   private mq: MQ;
-  private store: Store;
+  store: Store;
 
   constructor(mq: MQ, store: Store) {
     this.mq = mq;
@@ -140,7 +140,7 @@ export abstract class BaseDataMigrator<TJobConfig, TSourceEntity> implements Tas
             if (!cachedFirstPagePushed && data?.paginationContext?.page !== 0) {
               // requeue this page, as we are not sure if the first page is pushed or not
               logger.info(`First page not pushed, requeuing the next page`, { jobId: headers.jobId });
-              await wait(5000);
+              await wait(30000);
               await this.pushToQueue(headers, data);
               return true;
             }
@@ -149,6 +149,16 @@ export abstract class BaseDataMigrator<TJobConfig, TSourceEntity> implements Tas
             }
             // eslint-disable-next-line no-case-declarations
             const paginatedBatches = await this.batches(job, data.paginationContext, headers);
+            // if the paginated batches are empty, we need to mark the job as finished
+            if (paginatedBatches.length === 0) {
+              await this.update(headers.jobId, "FINISHED", {
+                total_batch_count: paginatedBatches.length,
+                completed_batch_count: paginatedBatches.length,
+                transformed_batch_count: paginatedBatches.length,
+                end_time: new Date(),
+              });
+              return true;
+            }
             // Push the paginated batches to the queue
             for (const batch of paginatedBatches) {
               headers.type = "transform";

apps/silo/src/etl/migrator/migrator.ts

@@ -442,6 +442,12 @@ export async function migrateToPlane(job: TImportJob, data: PlaneEntities[], met
       uuidv4(),
       "plane.bgtasks.data_import_task.import_data"
     );
+
+    logger.info("[MIGRATOR] Data import task registered", {
+      jobId: job.id,
+      phase: meta.phase,
+      isLastBatch: meta.isLastBatch,
+    });
   } catch (error) {
     const errorMessage = error instanceof Error ? error.message : "Unknown error";
     logger.error(`[${job.id}] Error while migrating the data to Plane: ${errorMessage}`);

packages/etl/src/clickup/services/api.service.ts

@@ -53,6 +53,28 @@ export class ClickupAPIService {
             await new Promise((resolve) => setTimeout(resolve, waitTime));
             return this.client.request(error.config);
           }
+        } else {
+          // retry 2 times with fixed delay
+          const MAX_RETRIES = 1;
+          const RETRY_DELAY = 1000;
+
+          for (let i = 0; i < MAX_RETRIES; i++) {
+            console.log("Retrying ClickUp API", {
+              status: error?.response?.status,
+              retry: i + 1,
+              max_retries: MAX_RETRIES,
+              retry_delay: RETRY_DELAY,
+            });
+            try {
+              return await this.client.request(error.config);
+            } catch (retryError: any) {
+              // If this is the last retry, reject the promise
+              if (i === MAX_RETRIES - 1) {
+                throw retryError;
+              }
+              await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY));
+            }
+          }
         }
         return Promise.reject(error);
       }