feat: implementing new job init- merge tiles tasks creation(MAPCO-4348)

package.json

@@ -43,7 +43,7 @@
     "@map-colonies/error-express-handler": "^2.1.0",
     "@map-colonies/express-access-log-middleware": "^2.0.1",
     "@map-colonies/js-logger": "^1.0.1",
-    "@map-colonies/mc-model-types": "^17.0.4",
+    "@map-colonies/mc-model-types": "^17.1.6",
     "@map-colonies/mc-priority-queue": "^8.1.1",
     "@map-colonies/mc-utils": "^3.1.0",
     "@map-colonies/read-pkg": "0.0.1",
@@ -58,6 +58,7 @@
     "http-status-codes": "^2.2.0",
     "lodash.get": "^4.4.2",
     "lodash.has": "^4.5.2",
+    "pino": "^9.4.0",
     "prom-client": "^15.1.2",
     "reflect-metadata": "^0.1.13",
     "tsyringe": "^4.8.0"

src/common/interfaces.ts

@@ -1,7 +1,7 @@
 import { IJobResponse, ITaskResponse } from '@map-colonies/mc-priority-queue';
 import { InputFiles, NewRasterLayerMetadata, PolygonPart, TileOutputFormat } from '@map-colonies/mc-model-types';
 import { TilesMimeFormat } from '@map-colonies/types';
-import { BBox, GeoJSON } from 'geojson';
+import { BBox, Polygon } from 'geojson';
 import { Footprint, ITileRange } from '@map-colonies/mc-utils';
 
 //#region config interfaces
@@ -48,11 +48,7 @@ export interface IngestionConfig {
   maxTaskAttempts: number;
 }
 //#endregion config
-export interface LogContext {
-  fileName: string;
-  class?: string;
-  function?: string;
-}
+
 export interface IJobHandler {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   handleJobInit: (job: IJobResponse<any, any>, taskId: string) => Promise<void>;
@@ -79,7 +75,7 @@ export interface ExtendedRasterLayerMetadata extends NewRasterLayerMetadata {
 export interface MergeTilesTaskParams {
   inputFiles: InputFiles;
   taskMetadata: MergeTilesMetadata;
-  partData: PolygonPart[];
+  partsData: PolygonPart[];
 }
 
 export interface MergeTilesMetadata {
@@ -93,16 +89,16 @@ export enum Grid {
   TWO_ON_ONE = '2x1',
 }
 //#region task
-export interface ILayerMergeData {
+export interface IPartSourceContext {
   fileName: string;
   tilesPath: string;
-  footprint?: GeoJSON;
+  footprint: Polygon;
   extent: BBox;
   maxZoom: number;
 }
 
 export interface IMergeParameters {
-  layers: ILayerMergeData[];
+  parts: IPartSourceContext[];
   destPath: string;
   maxZoom: number;
   grid: Grid;
@@ -124,13 +120,13 @@ export interface IMergeTaskParameters {
   batches: ITileRange[];
 }
 
-export interface IMergeOverlaps {
-  layers: ILayerMergeData[];
+export interface IPartsIntersection {
+  parts: IPartSourceContext[];
   intersection: Footprint;
 }
 
-export interface OverlapProcessingState {
-  accumulatedOverlap: Footprint | null;
+export interface IntersectionState {
+  accumulatedIntersection: Footprint | null;
   currentIntersection: Footprint | null;
 }
 //#endregion task

src/job/models/jobProcessor.ts

@@ -4,13 +4,12 @@ import { inject, injectable } from 'tsyringe';
 import { IJobResponse, OperationStatus, TaskHandler as QueueClient } from '@map-colonies/mc-priority-queue';
 import { getAvailableJobTypes } from '../../utils/configUtil';
 import { SERVICES } from '../../common/constants';
-import { IConfig, IngestionConfig, JobAndTaskResponse, LogContext, TaskResponse } from '../../common/interfaces';
+import { IConfig, IngestionConfig, JobAndTaskResponse, TaskResponse } from '../../common/interfaces';
 import { JOB_HANDLER_FACTORY_SYMBOL, JobHandlerFactory } from './jobHandlerFactory';
 
 @injectable()
 export class JobProcessor {
   private readonly dequeueIntervalMs: number;
-  private readonly logContext: LogContext;
   private readonly jobTypes: string[];
   private readonly pollingTaskTypes: string[];
   private readonly ingestionConfig: IngestionConfig;
@@ -26,33 +25,26 @@ export class JobProcessor {
     const { jobs, pollingTasks } = this.ingestionConfig;
     this.jobTypes = getAvailableJobTypes(jobs);
     this.pollingTaskTypes = [pollingTasks.init, pollingTasks.finalize];
-    this.logContext = {
-      fileName: __filename,
-      class: JobProcessor.name,
-    };
   }
 
   public async start(): Promise<void> {
-    const logCtx: LogContext = { ...this.logContext, function: this.start.name };
-    this.logger.info({ msg: 'starting polling', logContext: logCtx });
+    this.logger.info({ msg: 'starting polling' });
     while (this.isRunning) {
       await this.consumeAndProcess();
     }
   }
 
   public stop(): void {
-    const logCtx: LogContext = { ...this.logContext, function: this.stop.name };
-    this.logger.info({ msg: 'stopping polling', logContext: logCtx });
+    this.logger.info({ msg: 'stopping polling' });
     this.isRunning = false;
   }
 
   private async consumeAndProcess(): Promise<void> {
-    const logger = this.logger.child({ logContext: { ...this.logContext, function: this.consumeAndProcess.name } });
     let jobAndTask: JobAndTaskResponse | undefined = undefined;
     try {
       jobAndTask = await this.getJobAndTaskResponse();
       if (!jobAndTask) {
-        logger.debug({ msg: 'waiting for next dequeue', metadata: { dequeueIntervalMs: this.dequeueIntervalMs } });
+        this.logger.debug({ msg: 'waiting for next dequeue', dequeueIntervalMs: this.dequeueIntervalMs });
         await setTimeoutPromise(this.dequeueIntervalMs);
         return;
       }
@@ -61,17 +53,16 @@ export class JobProcessor {
     } catch (error) {
       if (error instanceof Error && jobAndTask) {
         const { job, task } = jobAndTask;
-        logger.error({ msg: 'rejecting task', error, metadata: { job, task } });
+        this.logger.error({ msg: 'rejecting task', error, job, task });
         await this.queueClient.reject(job.id, task.id, true, error.message);
       }
-      logger.debug({ msg: 'waiting for next dequeue', metadata: { dequeueIntervalMs: this.dequeueIntervalMs } });
+      this.logger.debug({ msg: 'waiting for next dequeue', dequeueIntervalMs: this.dequeueIntervalMs });
       await setTimeoutPromise(this.dequeueIntervalMs);
     }
   }
 
-  private async processJob(jobAndTaskType: JobAndTaskResponse): Promise<void> {
-    const logger = this.logger.child({ logContext: { ...this.logContext, function: this.processJob.name } });
-    const { job, task } = jobAndTaskType;
+  private async processJob(jobAndTask: JobAndTaskResponse): Promise<void> {
+    const { job, task } = jobAndTask;
     try {
       const taskTypes = this.ingestionConfig.pollingTasks;
       const jobHandler = this.jobHandlerFactory(job.type);
@@ -86,40 +77,39 @@ export class JobProcessor {
       }
     } catch (error) {
       if (error instanceof Error) {
-        logger.error({ msg: `failed processing the job: ${error.message}`, error });
+        this.logger.error({ msg: `failed processing the job: ${error.message}`, error });
       }
       throw error;
     }
   }
 
   private async getJobAndTaskResponse(): Promise<JobAndTaskResponse | undefined> {
-    const logger = this.logger.child({ logContext: { ...this.logContext, function: this.getJobAndTaskResponse.name } });
     try {
       for (const taskType of this.pollingTaskTypes) {
         for (const jobType of this.jobTypes) {
           const { task, shouldSkipTask } = await this.getTask(jobType, taskType);
 
           if (shouldSkipTask) {
-            logger.debug({ msg: `skipping task of type "${taskType}" and job of type "${jobType}"` });
+            this.logger.debug({ msg: `skipping task of type "${taskType}" and job of type "${jobType}"` });
             continue;
           }
 
           const job = await this.getJob(task.jobId);
-          logger.info({ msg: `got job and task response`, jobId: job.id, jobType: job.type, taskId: task.id, taskType: task.type });
+          this.logger.info({ msg: `got job and task response`, jobId: job.id, jobType: job.type, taskId: task.id, taskType: task.type });
 
           return { job, task };
         }
       }
     } catch (error) {
       if (error instanceof Error) {
-        logger.error({ msg: `Failed to get job and task response: ${error.message}`, error });
+        this.logger.error({ msg: `Failed to get job and task response: ${error.message}`, error });
       }
       throw error;
     }
   }
 
   private async getJob(jobId: string): Promise<IJobResponse<unknown, unknown>> {
-    const logger = this.logger.child({ logContext: { ...this.logContext, function: this.getJob.name }, jobId });
+    const logger = this.logger.child({ jobId });
 
     logger.info({ msg: `updating job status to ${OperationStatus.IN_PROGRESS}` });
     await this.queueClient.jobManagerClient.updateJob(jobId, { status: OperationStatus.IN_PROGRESS });
@@ -131,7 +121,8 @@ export class JobProcessor {
   }
 
   private async getTask(jobType: string, taskType: string): Promise<TaskResponse<unknown>> {
-    const logger = this.logger.child({ logContext: { ...this.logContext, function: this.getTask.name }, jobType, taskType });
+    const logger = this.logger.child({ jobType, taskType });
+
     logger.debug({ msg: `trying to dequeue task of type "${taskType}" and job of type "${jobType}"` });
     const task = await this.queueClient.dequeue(jobType, taskType);
 
@@ -141,7 +132,6 @@ export class JobProcessor {
     }
     if (task.attempts >= this.ingestionConfig.maxTaskAttempts) {
       const message = `${taskType} task ${task.id} reached max attempts, rejects as unrecoverable`;
-
       logger.warn({ msg: message, taskId: task.id, attempts: task.attempts });
       await this.queueClient.reject(task.jobId, task.id, false);