diff --git a/apps/api/plane/bgtasks/data_import_task.py b/apps/api/plane/bgtasks/data_import_task.py index b3f22be88c..ca2718be63 100644 --- a/apps/api/plane/bgtasks/data_import_task.py +++ b/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: diff --git a/apps/silo/package.json b/apps/silo/package.json index dd9bad7206..7ad0382b9f 100644 --- a/apps/silo/package.json +++ b/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", diff --git a/apps/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts b/apps/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts index d2006b77bc..bf9afca200 100644 --- a/apps/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts +++ b/apps/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts @@ -165,6 +165,12 @@ export class ClickUpDataMigrator extends BaseDataMigrator; diff --git a/apps/silo/src/apps/clickup-importer/migrator/transformers/etl.ts b/apps/silo/src/apps/clickup-importer/migrator/transformers/etl.ts index 1c830c777b..7bb7a7c8de 100644 --- a/apps/silo/src/apps/clickup-importer/migrator/transformers/etl.ts +++ b/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 diff --git a/apps/silo/src/etl/base-import-worker.ts b/apps/silo/src/etl/base-import-worker.ts index e046d99964..ca9f4bd829 100644 --- a/apps/silo/src/etl/base-import-worker.ts +++ b/apps/silo/src/etl/base-import-worker.ts @@ -12,7 +12,7 @@ import { migrateToPlane } from "./migrator"; export abstract class BaseDataMigrator 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 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 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"; diff --git a/apps/silo/src/etl/migrator/migrator.ts b/apps/silo/src/etl/migrator/migrator.ts index 80d87e9063..2e60e92529 100644 --- a/apps/silo/src/etl/migrator/migrator.ts +++ b/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}`); diff --git a/packages/etl/src/clickup/services/api.service.ts b/packages/etl/src/clickup/services/api.service.ts index ee8632fafc..b1bf237065 100644 --- a/packages/etl/src/clickup/services/api.service.ts +++ b/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); }