feat: harden normalization worker

Author: minkinad

package-lock.json

@@ -1,24 +1,29 @@
 {
   "name": "processing-worker",
-  "version": "0.1.0",
+  "version": "1.0.0",
   "private": true,
-  "description": "Черновой worker для нормализации и передачи данных в backend-api.",
+  "description": "Production-ready normalization worker for AIMSORA with retry/DLQ handling and strict contract validation.",
   "scripts": {
     "build": "tsc -p tsconfig.json",
     "start": "node dist/main.js",
     "start:dev": "ts-node-dev --respawn --transpile-only src/main.ts",
-    "check": "tsc --noEmit"
+    "check": "tsc --noEmit",
+    "test": "vitest run"
   },
   "dependencies": {
     "ajv": "^8.17.1",
+    "ajv-formats": "^3.0.1",
     "amqplib": "^0.10.5",
     "dotenv": "^16.4.7"
+    ,"pino": "^9.7.0",
+    "zod": "^3.24.2"
   },
   "devDependencies": {
     "@types/amqplib": "^0.10.7",
     "@types/node": "^22.10.5",
     "ts-node-dev": "^2.0.0",
-    "typescript": "^5.7.2"
+    "typescript": "^5.7.2",
+    "vitest": "^3.1.1"
   },
   "engines": {
     "node": ">=20"

package.json

@@ -5,18 +5,21 @@ mutation IngestNormalizedItem($input: IngestNormalizedItemInput!) {
   ingestNormalizedItem(input: $input) {
     accepted
     idempotencyKey
+    procurementId
   }
 }
 `;
 
 export async function sendToBackend(
   graphqlUrl: string,
+  ingestToken: string,
   event: NormalizedSourceEvent
 ): Promise<void> {
   const response = await fetch(graphqlUrl, {
     method: "POST",
     headers: {
-      "content-type": "application/json"
+      "content-type": "application/json",
+      "x-ingest-token": ingestToken
     },
     body: JSON.stringify({
       query: INGEST_MUTATION,
@@ -25,13 +28,18 @@ export async function sendToBackend(
           externalId: event.externalId,
           source: event.source,
           title: event.title,
+          description: event.description,
           customer: event.customer,
           supplier: event.supplier,
           amount: event.amount,
           currency: event.currency,
           publishedAt: event.publishedAt,
+          deadlineAt: event.deadlineAt,
           payloadVersion: event.payloadVersion,
-          rawPayload: { rawRef: event.rawRef, eventId: event.eventId }
+          sourceUrl: event.sourceUrl,
+          status: event.status,
+          rawPayload: { rawRef: event.rawRef, eventId: event.eventId },
+          rawEvent: event.rawEvent
         }
       }
     })

src/backend-client.ts

@@ -1,9 +1,18 @@
 import "dotenv/config";
+import { z } from "zod";
 
-export const config = {
-  rabbitmqUrl: process.env.RABBITMQ_URL ?? "amqp://app:app@localhost:5672",
-  apiGraphqlUrl: process.env.API_GRAPHQL_URL ?? "http://localhost:3000/graphql",
-  queueRaw: process.env.QUEUE_RAW_EVENT ?? "source.raw.v1",
-  queueNormalized: process.env.QUEUE_NORMALIZED_EVENT ?? "source.normalized.v1",
-  sharedContractsDir: process.env.SHARED_CONTRACTS_DIR ?? "../shared-contracts"
-};
+const envSchema = z.object({
+  RABBITMQ_URL: z.string().default("amqp://app:app@localhost:5672"),
+  API_GRAPHQL_URL: z.string().default("http://localhost:3000/graphql"),
+  API_INGEST_TOKEN: z.string().default("replace_me_ingest_token"),
+  QUEUE_RAW_EVENT: z.string().default("source.raw.v1"),
+  QUEUE_RETRY_EVENT: z.string().default("source.raw.retry.v1"),
+  QUEUE_DEAD_LETTER_EVENT: z.string().default("source.raw.dlq.v1"),
+  QUEUE_NORMALIZED_EVENT: z.string().default("source.normalized.v1"),
+  SHARED_CONTRACTS_DIR: z.string().default("../shared-contracts"),
+  RETRY_ATTEMPTS: z.coerce.number().int().positive().default(5),
+  RETRY_BASE_DELAY_MS: z.coerce.number().int().positive().default(5000),
+  PREFETCH: z.coerce.number().int().positive().default(10)
+});
+
+export const config = envSchema.parse(process.env);

src/config.ts

@@ -1,6 +1,7 @@
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import Ajv from "ajv";
+import addFormats from "ajv-formats";
 import type { NormalizedSourceEvent, RawSourceEvent } from "../types";
 
 export function createSchemaValidators(sharedContractsDir: string) {
@@ -12,6 +13,7 @@ export function createSchemaValidators(sharedContractsDir: string) {
   );
 
   const ajv = new Ajv({ allErrors: true });
+  addFormats(ajv);
   const rawValidate = ajv.compile<RawSourceEvent>(rawSchema);
   const normalizedValidate = ajv.compile<NormalizedSourceEvent>(normalizedSchema);
 

src/contracts/schema-validator.ts

@@ -0,0 +1,6 @@
+import pino from "pino";
+
+export const logger = pino({
+  level: process.env.LOG_LEVEL ?? "info",
+  base: { service: "processing-worker" }
+});

src/logger.ts

@@ -2,39 +2,67 @@ import type { ConsumeMessage } from "amqplib";
 import { sendToBackend } from "./backend-client";
 import { config } from "./config";
 import { createSchemaValidators } from "./contracts/schema-validator";
+import { logger } from "./logger";
 import { QueueClient } from "./messaging/queue-client";
 import { normalizeRawEvent } from "./normalize";
 import type { RawSourceEvent } from "./types";
 
 async function bootstrap(): Promise<void> {
-  const queue = new QueueClient(config.rabbitmqUrl);
-  const validators = createSchemaValidators(config.sharedContractsDir);
+  const queue = new QueueClient(config.RABBITMQ_URL);
+  const validators = createSchemaValidators(config.SHARED_CONTRACTS_DIR);
 
-  await queue.init();
-  await queue.assertQueue(config.queueRaw);
-  await queue.assertQueue(config.queueNormalized);
+  await queue.init(config.PREFETCH);
+  await queue.assertRetryTopology(
+    config.QUEUE_RAW_EVENT,
+    config.QUEUE_RETRY_EVENT,
+    config.QUEUE_DEAD_LETTER_EVENT
+  );
+  await queue.assertQueue(config.QUEUE_NORMALIZED_EVENT);
 
-  console.log("[processing-worker] запущен", {
-    queueRaw: config.queueRaw,
-    queueNormalized: config.queueNormalized,
-    apiGraphqlUrl: config.apiGraphqlUrl
-  });
+  logger.info(
+    {
+      queueRaw: config.QUEUE_RAW_EVENT,
+      queueRetry: config.QUEUE_RETRY_EVENT,
+      queueDeadLetter: config.QUEUE_DEAD_LETTER_EVENT,
+      queueNormalized: config.QUEUE_NORMALIZED_EVENT,
+      apiGraphqlUrl: config.API_GRAPHQL_URL
+    },
+    "processing-worker started"
+  );
+
+  queue.consume(config.QUEUE_RAW_EVENT, async (message: ConsumeMessage) => {
+    const attempt = queue.getRetryCount(message);
+
+    try {
+      const raw = queue.parseMessage<RawSourceEvent>(message);
+      validators.validateRaw(raw);
 
-  queue.consume(config.queueRaw, async (message: ConsumeMessage) => {
-    const raw = queue.parseMessage<RawSourceEvent>(message);
-    validators.validateRaw(raw);
+      const normalized = normalizeRawEvent(raw);
+      validators.validateNormalized(normalized);
 
-    const normalized = normalizeRawEvent(raw);
-    validators.validateNormalized(normalized);
+      await queue.publish(config.QUEUE_NORMALIZED_EVENT, normalized);
+      await sendToBackend(config.API_GRAPHQL_URL, config.API_INGEST_TOKEN, normalized);
+      queue.ack(message);
 
-    await queue.publish(config.queueNormalized, normalized);
-    await sendToBackend(config.apiGraphqlUrl, normalized);
+      logger.info(
+        { eventId: raw.eventId, source: raw.source, externalId: normalized.externalId },
+        "raw event normalized and ingested"
+      );
+    } catch (error) {
+      const reason = error instanceof Error ? error.message : "Unknown worker error";
+      if (attempt < config.RETRY_ATTEMPTS) {
+        await queue.retry(message, config.QUEUE_RETRY_EVENT, attempt + 1, config.RETRY_BASE_DELAY_MS);
+        logger.warn({ err: error, attempt: attempt + 1 }, "message scheduled for retry");
+        return;
+      }
 
-    console.log(`[processing-worker] обработано событие ${raw.eventId}`);
+      await queue.deadLetter(message, config.QUEUE_DEAD_LETTER_EVENT, reason);
+      logger.error({ err: error, attempt }, "message moved to dead-letter queue");
+    }
   });
 }
 
 void bootstrap().catch((error) => {
-  console.error("[processing-worker] фатальная ошибка", error);
+  logger.error({ err: error }, "processing-worker crashed");
   process.exit(1);
 });

src/main.ts

@@ -1,14 +1,15 @@
-import { connect, type Channel, type Connection, type ConsumeMessage } from "amqplib";
+import { connect, type Channel, type ChannelModel, type ConsumeMessage, type Options } from "amqplib";
 
 export class QueueClient {
-  private connection?: Connection;
+  private connection?: ChannelModel;
   private channel?: Channel;
 
   constructor(private readonly rabbitmqUrl: string) {}
 
-  async init(): Promise<void> {
+  async init(prefetch = 10): Promise<void> {
     this.connection = await connect(this.rabbitmqUrl);
     this.channel = await this.connection.createChannel();
+    await this.channel.prefetch(prefetch);
   }
 
   async assertQueue(queue: string): Promise<void> {
@@ -18,14 +19,29 @@ export class QueueClient {
     await this.channel.assertQueue(queue, { durable: true });
   }
 
-  async publish(queue: string, payload: unknown): Promise<void> {
+  async assertRetryTopology(mainQueue: string, retryQueue: string, deadLetterQueue: string): Promise<void> {
+    if (!this.channel) {
+      throw new Error("QueueClient не инициализирован");
+    }
+
+    await this.channel.assertQueue(deadLetterQueue, { durable: true });
+    await this.channel.assertQueue(retryQueue, {
+      durable: true,
+      deadLetterExchange: "",
+      deadLetterRoutingKey: mainQueue
+    });
+    await this.channel.assertQueue(mainQueue, { durable: true });
+  }
+
+  async publish(queue: string, payload: unknown, options?: Options.Publish): Promise<void> {
     if (!this.channel) {
       throw new Error("QueueClient не инициализирован");
     }
 
     this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
       contentType: "application/json",
-      persistent: true
+      persistent: true,
+      ...options
     });
   }
 
@@ -43,13 +59,50 @@ export class QueueClient {
         await handler(msg);
         this.channel?.ack(msg);
       } catch (error) {
-        console.error("[processing-worker] Ошибка обработки сообщения", error);
-        this.channel?.nack(msg, false, false);
+        throw error;
       }
     });
   }
 
   parseMessage<T>(message: ConsumeMessage): T {
     return JSON.parse(message.content.toString("utf-8")) as T;
   }
+
+  ack(message: ConsumeMessage): void {
+    this.channel?.ack(message);
+  }
+
+  async retry(
+    message: ConsumeMessage,
+    retryQueue: string,
+    attempt: number,
+    baseDelayMs: number
+  ): Promise<void> {
+    await this.publish(retryQueue, JSON.parse(message.content.toString("utf-8")), {
+      expiration: String(baseDelayMs * 2 ** attempt),
+      headers: {
+        ...(message.properties.headers ?? {}),
+        "x-retry-count": attempt
+      }
+    });
+    this.ack(message);
+  }
+
+  async deadLetter(
+    message: ConsumeMessage,
+    deadLetterQueue: string,
+    reason: string
+  ): Promise<void> {
+    await this.publish(deadLetterQueue, {
+      reason,
+      failedAt: new Date().toISOString(),
+      payload: JSON.parse(message.content.toString("utf-8"))
+    });
+    this.ack(message);
+  }
+
+  getRetryCount(message: ConsumeMessage): number {
+    const raw = message.properties.headers?.["x-retry-count"];
+    return typeof raw === "number" ? raw : Number(raw ?? 0);
+  }
 }

src/messaging/queue-client.ts

@@ -0,0 +1,58 @@
+import { describe, expect, it } from "vitest";
+import { normalizeRawEvent } from "./normalize";
+import type { RawSourceEvent } from "./types";
+
+describe("normalizeRawEvent", () => {
+  it("normalizes demo-source payload", () => {
+    const input: RawSourceEvent = {
+      eventId: "evt-1",
+      runKey: "demo-run",
+      source: "demo-source",
+      collectedAt: "2026-03-30T10:00:00.000Z",
+      url: "https://example.org",
+      payloadVersion: "v1",
+      artifacts: [],
+      raw: {
+        externalId: "demo-1",
+        title: "Поставка оборудования",
+        customer: "АО Демонстрационная АЭС",
+        amount: 1200
+      }
+    };
+
+    const normalized = normalizeRawEvent(input);
+
+    expect(normalized.externalId).toBe("demo-1");
+    expect(normalized.title).toBe("Поставка оборудования");
+    expect(normalized.customer).toBe("АО Демонстрационная АЭС");
+    expect(normalized.amount).toBe(1200);
+    expect(normalized.sourceUrl).toBe("https://example.org");
+  });
+
+  it("normalizes find-tender payload using ocid", () => {
+    const input: RawSourceEvent = {
+      eventId: "evt-2",
+      runKey: "find-tender-run",
+      source: "find-tender",
+      collectedAt: "2026-03-30T10:00:00.000Z",
+      url: "https://www.find-tender.service.gov.uk/Notice/008889-2026",
+      payloadVersion: "v1",
+      artifacts: [],
+      raw: {
+        ocid: "ocds-h6vhtk-061410",
+        title: "E-Disclosure Services",
+        buyer: "Care Quality Commission",
+        supplier: "KLDISCOVERY LIMITED",
+        amount: 150000,
+        currency: "GBP"
+      }
+    };
+
+    const normalized = normalizeRawEvent(input);
+
+    expect(normalized.externalId).toBe("ocds-h6vhtk-061410");
+    expect(normalized.title).toBe("E-Disclosure Services");
+    expect(normalized.customer).toBe("Care Quality Commission");
+    expect(normalized.sourceUrl).toContain("find-tender.service.gov.uk");
+  });
+});