test: retry layer api - integration tests

package.json

@@ -53,7 +53,7 @@
     "@map-colonies/mc-priority-queue": "^8.1.0",
     "@map-colonies/mc-utils": "^3.5.1",
     "@map-colonies/openapi-express-viewer": "^3.0.0",
-    "@map-colonies/raster-shared": "^7.6.1-alpha.1",
+    "@map-colonies/raster-shared": "^7.7.0-alpha.1",
     "@map-colonies/read-pkg": "0.0.1",
     "@map-colonies/storage-explorer-middleware": "^1.3.0",
     "@map-colonies/telemetry": "^6.1.0",

src/containerConfig.ts

@@ -5,13 +5,14 @@ import config from 'config';
 import { instancePerContainerCachingFactory } from 'tsyringe';
 import { DependencyContainer } from 'tsyringe/dist/typings/types';
 import xxhashFactory from 'xxhash-wasm';
+import type { HashAlgorithm } from '@map-colonies/raster-shared';
 import { SERVICES, SERVICE_NAME } from './common/constants';
 import { InjectionObject, registerDependencies } from './common/dependencyRegistration';
 import { tracing } from './common/tracing';
 import { INFO_ROUTER_SYMBOL, infoRouterFactory } from './info/routes/infoRouter';
 import { INGESTION_ROUTER_SYMBOL, ingestionRouterFactory } from './ingestion/routes/ingestionRouter';
 import { CHECKSUM_PROCESSOR } from './utils/hash/constants';
-import type { ChecksumProcessor, HashAlgorithm } from './utils/hash/interfaces';
+import type { ChecksumProcessor } from './utils/hash/interfaces';
 import { INGESTION_SCHEMAS_VALIDATOR_SYMBOL, schemasValidationsFactory } from './utils/validation/schemasValidator';
 import { VALIDATE_ROUTER_SYMBOL, validateRouterFactory } from './validate/routes/validateRouter';
 

src/ingestion/controllers/ingestionController.ts

@@ -1,4 +1,4 @@
-import { ConflictError } from '@map-colonies/error-types';
+import { ConflictError, NotFoundError } from '@map-colonies/error-types';
 import { RequestHandler } from 'express';
 import { HttpError } from 'express-openapi-validator/dist/framework/types';
 import { StatusCodes } from 'http-status-codes';
@@ -72,6 +72,9 @@ export class IngestionController {
       if (error instanceof ValidationError) {
         (error as HttpError).status = StatusCodes.BAD_REQUEST; //400
       }
+      if (error instanceof NotFoundError) {
+        (error as HttpError).status = StatusCodes.NOT_FOUND; //404
+      }
       if (error instanceof ConflictError) {
         (error as HttpError).status = StatusCodes.CONFLICT; //409
       }

src/ingestion/errors/ingestionErrors.ts

@@ -1,14 +1,14 @@
 import { OperationStatus } from '@map-colonies/mc-priority-queue';
 import { Logger } from '@map-colonies/js-logger';
-import { BadRequestError } from '@map-colonies/error-types';
+import { BadRequestError, NotFoundError } from '@map-colonies/error-types';
 import { Span } from '@opentelemetry/api';
 
 export class UnsupportedEntityError extends Error {
   public constructor(message: string) {
     super(message);
   }
 }
-export class FileNotFoundError extends UnsupportedEntityError {
+export class FileNotFoundError extends NotFoundError {
   public constructor(fileName: string);
   public constructor(fileName: string[]);
   public constructor(fileName: string, path: string);

src/ingestion/interfaces.ts

@@ -1,8 +1,7 @@
 /* eslint-disable @typescript-eslint/no-magic-numbers */
 import { ICreateJobResponse } from '@map-colonies/mc-priority-queue';
-import { ingestionBaseJobParamsSchema, ingestionValidationTaskParamsSchema } from '@map-colonies/raster-shared';
+import { ingestionBaseJobParamsSchema } from '@map-colonies/raster-shared';
 import z from 'zod';
-import { checksumSchema, type Checksum } from '../utils/hash/interfaces';
 
 export interface SourcesValidationResponse {
   isValid: boolean;
@@ -54,23 +53,3 @@ export interface TileSize {
 }
 
 export type IngestionBaseJobParams = z.infer<typeof ingestionBaseJobParamsSchema>;
-
-export type BaseValidationTaskParams = z.infer<typeof ingestionValidationTaskParamsSchema>;
-
-export interface ChecksumValidationParameters {
-  checksums: Checksum[];
-}
-
-export interface ValidationTaskParameters extends BaseValidationTaskParams, ChecksumValidationParameters {}
-
-export interface ValidationTaskParametersPartial extends Omit<ValidationTaskParameters, 'isValid'> {
-  isValid?: boolean;
-}
-
-export const validationTaskParametersSchema = ingestionValidationTaskParamsSchema.extend({
-  checksums: z.array(checksumSchema),
-});
-
-export const validationTaskParametersSchemaPartial = validationTaskParametersSchema.partial({ isValid: true });
-
-export type TaskValidationParametersPartial = z.infer<typeof validationTaskParametersSchemaPartial>;

src/ingestion/models/ingestionManager.ts

@@ -1,3 +1,4 @@
+import { relative } from 'node:path';
 import { ConflictError, NotFoundError } from '@map-colonies/error-types';
 import { Logger } from '@map-colonies/js-logger';
 import {
@@ -13,6 +14,9 @@ import {
   inputFilesSchema,
   rasterProductTypeSchema,
   resourceIdSchema,
+  ingestionValidationTaskParamsSchema,
+  type IngestionValidationTaskParams,
+  type Checksum as IChecksum,
   type FileMetadata,
   type IngestionNewJobParams,
   type IngestionSwapUpdateJobParams,
@@ -30,20 +34,12 @@ import { CatalogClient } from '../../serviceClients/catalogClient';
 import { JobManagerWrapper } from '../../serviceClients/jobManagerWrapper';
 import { MapProxyClient } from '../../serviceClients/mapProxyClient';
 import { Checksum } from '../../utils/hash/checksum';
-import { Checksum as IChecksum } from '../../utils/hash/interfaces';
 import { getAbsolutePathInputFiles } from '../../utils/paths';
 import { getShapefileFiles } from '../../utils/shapefile';
 import { ZodValidator } from '../../utils/validation/zodValidator';
 import { ValidateManager } from '../../validate/models/validateManager';
 import { ChecksumError, throwInvalidJobStatusError } from '../errors/ingestionErrors';
-import type {
-  ChecksumValidationParameters,
-  IngestionBaseJobParams,
-  ResponseId,
-  TaskValidationParametersPartial,
-  ValidationTaskParametersPartial,
-} from '../interfaces';
-import { validationTaskParametersSchemaPartial } from '../interfaces';
+import type { IngestionBaseJobParams, ResponseId } from '../interfaces';
 import type { RasterLayerMetadata } from '../schemas/layerCatalogSchema';
 import type { IngestionNewLayer } from '../schemas/newLayerSchema';
 import type { IngestionUpdateLayer } from '../schemas/updateLayerSchema';
@@ -213,9 +209,9 @@ export class IngestionManager {
       throwInvalidJobStatusError(jobId, retryJob.status, this.logger, activeSpan);
     }
 
-    const validationTask: ITaskResponse<TaskValidationParametersPartial> = await this.getValidationTask(jobId, logCtx);
+    const validationTask: ITaskResponse<IngestionValidationTaskParams> = await this.getValidationTask(jobId, logCtx);
     const { resourceId, productType } = this.parseAndValidateJobIdentifiers(retryJob.resourceId, retryJob.productType);
-    await this.zodValidator.validate(validationTaskParametersSchemaPartial, validationTask.parameters);
+    await this.zodValidator.validate(ingestionValidationTaskParamsSchema, validationTask.parameters);
     await this.polygonPartsManagerClient.deleteValidationEntity(resourceId, productType);
 
     if (validationTask.parameters.isValid === true) {
@@ -237,8 +233,8 @@ export class IngestionManager {
   }
 
   @withSpanAsyncV4
-  private async getValidationTask(jobId: string, logCtx: LogContext): Promise<ITaskResponse<TaskValidationParametersPartial>> {
-    const tasks = await this.jobManagerWrapper.getTasksForJob<TaskValidationParametersPartial>(jobId);
+  private async getValidationTask(jobId: string, logCtx: LogContext): Promise<ITaskResponse<IngestionValidationTaskParams>> {
+    const tasks = await this.jobManagerWrapper.getTasksForJob<IngestionValidationTaskParams>(jobId);
 
     const validationTask = tasks.find((task) => task.type === this.validationTaskType);
 
@@ -264,7 +260,7 @@ export class IngestionManager {
   @withSpanAsyncV4
   private async hardReset(
     retryJob: IJobResponse<IngestionBaseJobParams, unknown>,
-    validationTask: ITaskResponse<TaskValidationParametersPartial>,
+    validationTask: ITaskResponse<IngestionValidationTaskParams>,
     shouldConsiderChecksumChanges: boolean,
     logCtx: LogContext
   ): Promise<void> {
@@ -279,7 +275,7 @@ export class IngestionManager {
     const absoluteInputFilesPaths = await this.validateAndGetAbsoluteInputFiles(retryJob.parameters.inputFiles);
     const { metadataShapefilePath } = absoluteInputFilesPaths;
 
-    const newChecksums = await this.getFilesChecksum(metadataShapefilePath);
+    const newChecksums = await this.getChecksum(metadataShapefilePath);
 
     let updatedChecksums = validationTask.parameters.checksums;
 
@@ -291,12 +287,12 @@ export class IngestionManager {
         trace.getActiveSpan()?.setAttribute('exception.type', error.status);
         throw error;
       }
-      updatedChecksums = this.buildUpdatedChecksums(validationTask.parameters.checksums, newChecksums, logCtx);
+      updatedChecksums = this.buildUpdatedChecksums(validationTask.parameters.checksums, this.convertChecksumsToRelativePaths(newChecksums), logCtx);
     }
 
     const reportToSet: FileMetadata | undefined = validationTask.parameters.report ?? undefined;
 
-    const updatedParameters: ValidationTaskParametersPartial = {
+    const updatedParameters: IngestionValidationTaskParams = {
       isValid: validationTask.parameters.isValid,
       report: reportToSet,
       checksums: updatedChecksums,
@@ -367,18 +363,18 @@ export class IngestionManager {
   }
 
   @withSpanAsyncV4
-  private async manualResetJobAndTask(jobId: string, taskId: string, parameters: ValidationTaskParametersPartial, logCtx: LogContext): Promise<void> {
+  private async manualResetJobAndTask(jobId: string, taskId: string, parameters: IngestionValidationTaskParams, logCtx: LogContext): Promise<void> {
     this.logger.debug({ msg: 'manually updating validation task and job status to PENDING', logContext: logCtx, jobId, taskId });
 
-    const taskParameters: IUpdateTaskBody<ValidationTaskParametersPartial> = {
+    const taskParameters: IUpdateTaskBody<IngestionValidationTaskParams> = {
       parameters,
       status: OperationStatus.PENDING,
       attempts: 0,
       percentage: 0,
       reason: '',
     };
 
-    await this.jobManagerWrapper.updateTask<ValidationTaskParametersPartial>(jobId, taskId, taskParameters);
+    await this.jobManagerWrapper.updateTask<IngestionValidationTaskParams>(jobId, taskId, taskParameters);
     await this.jobManagerWrapper.updateJob(jobId, { status: OperationStatus.PENDING, reason: '' });
     this.logger.debug({ msg: 'validation task and job status updated to PENDING successfully', logContext: logCtx, jobId, taskId });
   }
@@ -564,9 +560,10 @@ export class IngestionManager {
   @withSpanAsyncV4
   private async newLayerJobPayload(
     newLayer: EnhancedIngestionNewLayer
-  ): Promise<ICreateJobBody<IngestionNewJobParams, ChecksumValidationParameters>> {
-    const checksums = await this.getFilesChecksum(newLayer.inputFiles.metadataShapefilePath.absolute);
-    const taskParameters: ChecksumValidationParameters = { checksums };
+  ): Promise<ICreateJobBody<IngestionNewJobParams, IngestionValidationTaskParams>> {
+    const checksums = await this.getChecksum(newLayer.inputFiles.metadataShapefilePath.absolute);
+    const relativeChecksums = this.convertChecksumsToRelativePaths(checksums);
+    const taskParameters: IngestionValidationTaskParams = { checksums: relativeChecksums };
 
     const newLayerRelative = {
       ...newLayer,
@@ -602,15 +599,16 @@ export class IngestionManager {
   private async updateLayerJobPayload(
     rasterLayerMetadata: RasterLayerMetadata,
     updateLayer: EnhancedIngestionUpdateLayer
-  ): Promise<ICreateJobBody<IngestionUpdateJobParams | IngestionSwapUpdateJobParams, ChecksumValidationParameters>> {
+  ): Promise<ICreateJobBody<IngestionUpdateJobParams | IngestionSwapUpdateJobParams, IngestionValidationTaskParams>> {
     const { displayPath, id, productId, productType, productVersion, tileOutputFormat, productName, productSubType } = rasterLayerMetadata;
     const isSwapUpdate = this.supportedIngestionSwapTypes.find((supportedSwapObj) => {
       return supportedSwapObj.productType === productType && supportedSwapObj.productSubType === productSubType;
     });
     const updateJobAction = isSwapUpdate ? this.swapUpdateJobType : this.updateJobType;
 
-    const checksums = await this.getFilesChecksum(updateLayer.inputFiles.metadataShapefilePath.absolute);
-    const taskParameters: ChecksumValidationParameters = { checksums };
+    const checksums = await this.getChecksum(updateLayer.inputFiles.metadataShapefilePath.absolute);
+    const relativeChecksums = this.convertChecksumsToRelativePaths(checksums);
+    const taskParameters: IngestionValidationTaskParams = { checksums: relativeChecksums };
 
     const updateLayerRelative = {
       ...updateLayer,
@@ -647,7 +645,7 @@ export class IngestionManager {
   }
 
   @withSpanAsyncV4
-  private async getFilesChecksum(shapefilePath: string): Promise<IChecksum[]> {
+  private async getChecksum(shapefilePath: string): Promise<IChecksum[]> {
     const checksums = await Promise.all(getShapefileFiles(shapefilePath).map(async (fileName) => this.getFileChecksum(fileName)));
     return checksums;
   }
@@ -685,4 +683,11 @@ export class IngestionManager {
     const validStatuses = [OperationStatus.FAILED, OperationStatus.SUSPENDED];
     return validStatuses.includes(status);
   }
+
+  private convertChecksumsToRelativePaths(checksums: IChecksum[]): IChecksum[] {
+    return checksums.map((checksum) => ({
+      ...checksum,
+      fileName: relative(this.sourceMount, checksum.fileName),
+    }));
+  }
 }

src/serviceClients/jobManagerWrapper.ts

@@ -1,13 +1,17 @@
 import { Logger } from '@map-colonies/js-logger';
 import { ICreateJobBody, ICreateJobResponse, JobManagerClient } from '@map-colonies/mc-priority-queue';
 import { IHttpRetryConfig } from '@map-colonies/mc-utils';
-import type { IngestionNewJobParams, IngestionSwapUpdateJobParams, IngestionUpdateJobParams } from '@map-colonies/raster-shared';
+import type {
+  IngestionNewJobParams,
+  IngestionSwapUpdateJobParams,
+  IngestionUpdateJobParams,
+  IngestionValidationTaskParams,
+} from '@map-colonies/raster-shared';
 import { withSpanAsyncV4 } from '@map-colonies/telemetry';
 import { trace, Tracer } from '@opentelemetry/api';
 import { inject, injectable } from 'tsyringe';
 import { SERVICES } from '../common/constants';
 import type { IConfig } from '../common/interfaces';
-import { ChecksumValidationParameters } from '../ingestion/interfaces';
 
 @injectable()
 export class JobManagerWrapper extends JobManagerClient {
@@ -27,7 +31,7 @@ export class JobManagerWrapper extends JobManagerClient {
 
   @withSpanAsyncV4
   public async createIngestionJob(
-    payload: ICreateJobBody<IngestionNewJobParams | IngestionUpdateJobParams | IngestionSwapUpdateJobParams, ChecksumValidationParameters>
+    payload: ICreateJobBody<IngestionNewJobParams | IngestionUpdateJobParams | IngestionSwapUpdateJobParams, IngestionValidationTaskParams>
   ): Promise<ICreateJobResponse> {
     const activeSpan = trace.getActiveSpan();
     activeSpan?.updateName('jobManagerWrapper.createJobWrapper');

src/utils/hash/checksum.ts

@@ -4,11 +4,12 @@ import type { Logger } from '@map-colonies/js-logger';
 import { withSpanAsyncV4 } from '@map-colonies/telemetry';
 import { trace, type Tracer } from '@opentelemetry/api';
 import { inject, injectable } from 'tsyringe';
+import type { Checksum as IChecksum } from '@map-colonies/raster-shared';
 import { SERVICES } from '../../common/constants';
 import type { LogContext } from '../../common/interfaces';
 import { ChecksumError } from '../../ingestion/errors/ingestionErrors';
 import { CHECKSUM_PROCESSOR } from './constants';
-import type { ChecksumProcessor, Checksum as IChecksum } from './interfaces';
+import type { ChecksumProcessor } from './interfaces';
 
 @injectable()
 export class Checksum {

src/utils/hash/interfaces.ts

@@ -1,5 +1,4 @@
-import z from 'zod';
-import { HASH_ALGORITHMS } from './constants';
+import { HashAlgorithm } from '@map-colonies/raster-shared';
 
 /**
  * Interface describing a hash processor instance.
@@ -25,20 +24,6 @@ interface HashProcessor {
   reset?: () => void;
 }
 
-export type HashAlgorithm = (typeof HASH_ALGORITHMS)[number];
-
-export interface Checksum {
-  algorithm: HashAlgorithm;
-  checksum: string;
-  fileName: string;
-}
-
-export const checksumSchema = z.object({
-  algorithm: z.enum(HASH_ALGORITHMS),
-  checksum: z.string(),
-  fileName: z.string(),
-});
-
 /**
  * Interface describing a checksum processor instance.
  * Provides a consistent API for calculating checksums.

tests/configurations/integration/jest.config.js

@@ -17,6 +17,7 @@ module.exports = {
     '!<rootDir>/src/ingestion/schemas/constants.ts',
     '!<rootDir>/src/**/interfaces.ts',
     '!<rootDir>/src/utils/hash/constants.ts',
+    '!<rootDir>/src/ingestion/schemas/layerCatalogSchema.ts', // currently unused - to be covered once we will start using it
   ],
   coverageDirectory: '<rootDir>/coverage',
   rootDir: '../../../.',
@@ -35,10 +36,10 @@ module.exports = {
   testEnvironment: 'node',
   coverageThreshold: {
     global: {
-      branches: 73,
+      branches: 75,
       functions: 80,
       lines: 80,
-      statements: -46,
+      statements: -27,
     },
   },
 };

tests/configurations/unit/jest.config.js

@@ -33,7 +33,7 @@ module.exports = {
       branches: 60,
       functions: 70,
       lines: 80,
-      statements: -126,
+      statements: -120,
     },
   },
 };

tests/integration/ingestion/helpers/ingestionRequestSender.ts

@@ -12,4 +12,8 @@ export class IngestionRequestSender {
   public async updateLayer(id: string, body: IngestionUpdateLayer): Promise<supertest.Response> {
     return supertest.agent(this.app).put(`/ingestion/${id}`).set('Content-Type', 'application/json').send(body);
   }
+
+  public async retryIngestion(jobId: string): Promise<supertest.Response> {
+    return supertest.agent(this.app).put(`/ingestion/${jobId}/retry`).set('Content-Type', 'application/json');
+  }
 }