IMN 707 - IVASS Certified attributes importer (#963)

.github/workflows/build-push.yaml

@@ -81,6 +81,8 @@ jobs:
             dockerfile_path: packages/producer-keychain-readmodel-writer
           - image_name: anac-certified-attributes-importer
             dockerfile_path: packages/anac-certified-attributes-importer
+          - image_name: ivass-certified-attributes-importer
+            dockerfile_path: packages/ivass-certified-attributes-importer
           - image_name: one-trust-notices
             dockerfile_path: packages/one-trust-notices
 

package.json

@@ -25,6 +25,7 @@
     "start:bff": "turbo start --filter pagopa-interop-backend-for-frontend",
     "start:api-gateway": "turbo start --filter pagopa-interop-api-gateway",
     "start:anac-certified-attributes-importer": "turbo start --filter pagopa-interop-anac-certified-attributes-importer",
+    "start:ivass-certified-attributes-importer": "turbo start --filter pagopa-interop-ivass-certified-attributes-importer",
     "start:one-trust-notices": "turbo start --filter pagopa-interop-one-trust-notices",
     "test": "turbo test",
     "build": "turbo build",

packages/ivass-certified-attributes-importer/.env

@@ -0,0 +1,21 @@
+LOG_LEVEL=info
+
+READMODEL_DB_HOST="localhost"
+READMODEL_DB_NAME="readmodel"
+READMODEL_DB_USERNAME="root"
+READMODEL_DB_PASSWORD="example"
+READMODEL_DB_PORT=27017
+
+INTERNAL_JWT_KID=ffcc9b5b-4612-49b1-9374-9d203a3834f2
+INTERNAL_JWT_SUBJECT="dev-refactor.interop-eservice-descriptors-archiver"
+INTERNAL_JWT_ISSUER="dev-refactor.interop.pagopa.it"
+INTERNAL_JWT_AUDIENCE="refactor.dev.interop.pagopa.it/internal"
+INTERNAL_JWT_SECONDS_DURATION=60
+
+TENANT_PROCESS_URL="http://localhost:3500"
+RECORDS_PROCESS_BATCH_SIZE=5
+IVASS_TENANT_ID=""
+SOURCE_URL=""
+HISTORY_BUCKET_NAME=""
+
+AWS_CONFIG_FILE=aws.config.local

packages/ivass-certified-attributes-importer/Dockerfile

@@ -0,0 +1,44 @@
+FROM node:20.14.0-slim@sha256:5e8ac65a0231d76a388683d07ca36a9769ab019a85d85169fe28e206f7a3208e as build
+
+RUN corepack enable
+
+WORKDIR /app
+COPY package.json /app/
+COPY pnpm-lock.yaml /app/
+COPY pnpm-workspace.yaml /app/
+
+COPY ./packages/ivass-certified-attributes-importer/package.json /app/packages/ivass-certified-attributes-importer/package.json
+COPY ./packages/commons/package.json /app/packages/commons/package.json
+COPY ./packages/models/package.json /app/packages/models/package.json
+COPY ./packages/api-clients/package.json /app/packages/api-clients/package.json
+
+RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile
+
+COPY tsconfig.json /app/
+COPY turbo.json /app/
+COPY ./packages/ivass-certified-attributes-importer /app/packages/ivass-certified-attributes-importer/
+COPY ./packages/commons /app/packages/commons
+COPY ./packages/models /app/packages/models
+COPY ./packages/api-clients /app/packages/api-clients
+
+RUN pnpm build && \
+    rm -rf /app/node_modules/.modules.yaml && \
+    rm -rf /app/node_modules/.cache && \
+    mkdir /out && \
+    cp -a --parents -t /out \
+      node_modules packages/ivass-certified-attributes-importer/node_modules \
+      package*.json packages/ivass-certified-attributes-importer/package*.json \
+      packages/commons/ \
+      packages/models/ \
+      packages/api-clients  \
+      packages/ivass-certified-attributes-importer/dist && \
+    find /out -exec touch -h --date=@0 {} \;
+
+FROM node:20.14.0-slim@sha256:5e8ac65a0231d76a388683d07ca36a9769ab019a85d85169fe28e206f7a3208e as final 
+
+COPY --from=build /out /app
+
+WORKDIR /app/packages/ivass-certified-attributes-importer
+EXPOSE 3000
+
+CMD [ "node", "." ]

packages/ivass-certified-attributes-importer/README.md

@@ -0,0 +1,3 @@
+# IVASS certified attributes importer
+
+This job assigns the IVASS certified attributes to the authorized tenants

packages/ivass-certified-attributes-importer/aws.config.local

@@ -0,0 +1,4 @@
+[default]
+aws_access_key_id=testawskey
+aws_secret_access_key=testawssecret
+region=eu-central-1

packages/ivass-certified-attributes-importer/package.json

@@ -0,0 +1,45 @@
+{
+  "name": "pagopa-interop-ivass-certified-attributes-importer",
+  "private": true,
+  "version": "1.0.0",
+  "description": "PagoPA Interoperability ivass-certified-attributes-importer job",
+  "main": "dist",
+  "type": "module",
+  "scripts": {
+    "test": "vitest",
+    "lint": "eslint . --ext .ts,.tsx",
+    "lint:autofix": "eslint . --ext .ts,.tsx --fix",
+    "format:check": "prettier --check src",
+    "format:write": "prettier --write src",
+    "start": "node --loader ts-node/esm -r 'dotenv-flow/config' --watch ./src/index.ts",
+    "build": "tsc",
+    "check": "tsc --project tsconfig.check.json"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "Apache-2.0",
+  "devDependencies": {
+    "@pagopa/eslint-config": "3.0.0",
+    "@types/node": "20.14.6",
+    "@types/ssh2-sftp-client": "9.0.4",
+    "@types/adm-zip": "0.5.5",
+    "pagopa-interop-commons-test": "workspace:*",
+    "prettier": "2.8.8",
+    "testcontainers": "10.9.0",
+    "ts-node": "10.9.2",
+    "typescript": "5.4.5",
+    "vitest": "1.6.0"
+  },
+  "dependencies": {
+    "@aws-sdk/client-s3": "3.387.0",
+    "adm-zip": "0.5.15",
+    "axios": "1.5.0",
+    "csv": "6.3.2",
+    "ts-pattern": "5.2.0",
+    "dotenv-flow": "4.1.0",
+    "pagopa-interop-commons": "workspace:*",
+    "pagopa-interop-models": "workspace:*",
+    "pagopa-interop-api-clients": "workspace:*",
+    "zod": "3.23.8"
+  }
+}

packages/ivass-certified-attributes-importer/src/config/config.ts

@@ -0,0 +1,38 @@
+import {
+  APIEndpoint,
+  FileManagerConfig,
+  LoggerConfig,
+  ReadModelDbConfig,
+  TokenGenerationConfig,
+} from "pagopa-interop-commons";
+import { z } from "zod";
+
+const IvassCertifiedAttributesImporterConfig = LoggerConfig.and(
+  FileManagerConfig
+)
+  .and(ReadModelDbConfig)
+  .and(TokenGenerationConfig)
+  .and(
+    z
+      .object({
+        SOURCE_URL: z.string(),
+        HISTORY_BUCKET_NAME: z.string(),
+        TENANT_PROCESS_URL: APIEndpoint,
+        RECORDS_PROCESS_BATCH_SIZE: z.number(),
+        IVASS_TENANT_ID: z.string(),
+      })
+      .transform((c) => ({
+        sourceUrl: c.SOURCE_URL,
+        historyBucketName: c.HISTORY_BUCKET_NAME,
+        tenantProcessUrl: c.TENANT_PROCESS_URL,
+        recordsProcessBatchSize: c.RECORDS_PROCESS_BATCH_SIZE,
+        ivassTenantId: c.IVASS_TENANT_ID,
+      }))
+  );
+
+export type IvassCertifiedAttributesImporterConfig = z.infer<
+  typeof IvassCertifiedAttributesImporterConfig
+>;
+
+export const config: IvassCertifiedAttributesImporterConfig =
+  IvassCertifiedAttributesImporterConfig.parse(process.env);

packages/ivass-certified-attributes-importer/src/config/constants.ts

@@ -0,0 +1 @@
+export const IVASS_INSURANCES_ATTRIBUTE_CODE = "insurances";

packages/ivass-certified-attributes-importer/src/index.ts

@@ -0,0 +1,47 @@
+import crypto from "crypto";
+import {
+  InteropTokenGenerator,
+  ReadModelRepository,
+  RefreshableInteropToken,
+  initFileManager,
+  logger,
+} from "pagopa-interop-commons";
+import { config } from "./config/config.js";
+import { TenantProcessService } from "./service/tenantProcessService.js";
+import { importAttributes } from "./service/processor.js";
+import { downloadCSV } from "./service/fileDownloader.js";
+import { ReadModelQueries } from "./service/readModelQueriesService.js";
+
+const loggerInstance = logger({
+  serviceName: "ivass-certified-attributes-importer",
+  correlationId: crypto.randomUUID(),
+});
+
+const fileManager = initFileManager(config);
+
+const csvDownloader = (): Promise<string> =>
+  downloadCSV(
+    config.sourceUrl,
+    fileManager,
+    config.historyBucketName,
+    loggerInstance
+  );
+
+const readModelClient = ReadModelRepository.init(config);
+const readModelQueries: ReadModelQueries = new ReadModelQueries(
+  readModelClient
+);
+
+const tokenGenerator = new InteropTokenGenerator(config);
+const refreshableToken = new RefreshableInteropToken(tokenGenerator);
+const tenantProcess = new TenantProcessService(config.tenantProcessUrl);
+
+await importAttributes(
+  csvDownloader,
+  readModelQueries,
+  tenantProcess,
+  refreshableToken,
+  config.recordsProcessBatchSize,
+  config.ivassTenantId,
+  loggerInstance
+);

packages/ivass-certified-attributes-importer/src/model/csvRowModel.ts

@@ -0,0 +1,32 @@
+import { z } from "zod";
+
+const FISCAL_CODE_LENGTH = 11;
+
+export const RawCsvRow = z.object({
+  CODICE_IVASS: z.string(),
+  TIPO_ALBO: z.string().optional(),
+  DATA_ISCRIZIONE_ALBO_ELENCO: z.string().transform((arg) => new Date(arg)),
+  DATA_CANCELLAZIONE_ALBO_ELENCO: z.string().transform((arg) => new Date(arg)),
+  DENOMINAZIONE_IMPRESA: z.string(),
+  CODICE_FISCALE: z
+    .string()
+    .transform((s) => s.substring(s.length - FISCAL_CODE_LENGTH))
+    .optional(),
+  CLASSIFICAZIONE: z.string().optional(),
+  INDIRIZZO_SEDE_LEGALE_RAPPRESENTANZA_IN_ITALIA: z.string().optional(),
+  INDIRIZZO_DIREZIONE_GENERALE: z.string().optional(),
+  INDIRIZZO_CASA_MADRE: z.string().optional(),
+  TIPO_LAVORO: z.string().optional(),
+  PEC: z.string().optional(),
+});
+
+export type RawCsvRow = z.infer<typeof RawCsvRow>;
+
+export const CsvRow = z.object({
+  CODICE_IVASS: z.string(),
+  DATA_ISCRIZIONE_ALBO_ELENCO: z.date(),
+  DATA_CANCELLAZIONE_ALBO_ELENCO: z.date(),
+  CODICE_FISCALE: z.string().optional(),
+});
+
+export type CsvRow = z.infer<typeof CsvRow>;

packages/ivass-certified-attributes-importer/src/model/interopContextModel.ts

@@ -0,0 +1,8 @@
+import { z } from "zod";
+
+export const InteropContext = z.object({
+  correlationId: z.string(),
+  bearerToken: z.string(),
+});
+
+export type InteropContext = z.infer<typeof InteropContext>;

packages/ivass-certified-attributes-importer/src/model/processorModel.ts

@@ -0,0 +1,16 @@
+import { ExternalId } from "pagopa-interop-models";
+import { CsvRow } from "./csvRowModel.js";
+
+export type BatchParseResult = {
+  processedRecordsCount: number;
+  records: CsvRow[];
+};
+
+export type AttributeIdentifiers = {
+  id: string;
+  externalId: ExternalId;
+};
+
+export type IvassAttributes = {
+  ivassInsurances: AttributeIdentifiers;
+};

packages/ivass-certified-attributes-importer/src/service/fileDownloader.ts

@@ -0,0 +1,87 @@
+import axios from "axios";
+import AmdZip from "adm-zip";
+import { FileManager, Logger } from "pagopa-interop-commons";
+
+const unzipFile = async (zipBlob: Blob): Promise<Buffer> => {
+  const entries = new AmdZip(
+    Buffer.from(await zipBlob.arrayBuffer())
+  ).getEntries();
+  const csvEntries = entries.filter((entry) =>
+    entry.entryName.endsWith(".csv")
+  );
+
+  if (csvEntries.length === 0) {
+    throw new Error("The archive does not contain csv files");
+  }
+
+  if (csvEntries.length > 1) {
+    throw new Error("The archive contains multiple csv files");
+  }
+
+  const entry = entries[0];
+
+  if (!entry.getData) {
+    throw new Error("Unexpected error: getData method is undefined");
+  }
+
+  return entry.getData();
+};
+
+async function downloadFile(
+  url: string
+): Promise<{ filename: string; blob: Blob }> {
+  /**
+   * We need first to get the cookies from the first request;
+   * When we call the url without the cookies, the server will make infinite redirects
+   * until we reach the maxRedirects limit (?why?).
+   *
+   * So we need to make a first request (limiting the redirects) to get the cookies and then make another request with the cookies
+   * set in the headers to get the file.
+   */
+  const redirectRes = await axios.get(url, {
+    maxRedirects: 0,
+    validateStatus(status) {
+      return status >= 200 && status < 303;
+    },
+  });
+
+  const CookieHeader = redirectRes.headers["set-cookie"]?.join("; ");
+
+  const dataRes = await axios.get(url, {
+    headers: {
+      Cookie: CookieHeader,
+    },
+    responseType: "arraybuffer",
+  });
+
+  const filename = dataRes.headers["content-disposition"]
+    .split("filename=")[1]
+    .replace(/"/g, "");
+  const blob = new Blob([dataRes.data], { type: "application/zip" });
+
+  return {
+    blob,
+    filename,
+  };
+}
+
+export const downloadCSV = async (
+  sourceUrl: string,
+  fileManager: FileManager,
+  bucket: string,
+  logger: Logger
+): Promise<string> => {
+  const { blob, filename } = await downloadFile(sourceUrl);
+
+  const zipFile = Buffer.from(await blob.arrayBuffer());
+  await fileManager.storeBytesByKey(
+    bucket,
+    `organizations/${filename}`,
+    zipFile,
+    logger
+  );
+
+  const unzippedFile = await unzipFile(blob);
+
+  return unzippedFile.toString();
+};