aimsora-0010 // processing-worker // Инициализирован worker-сервис: структура TypeScript, конфиги и контейнеризация

minkinad · minkinad · commit 4419cf4ae872 · 2026-03-11T13:00:26.000+03:00
diff --git a/.dockerignore b/.dockerignore
@@ -0,0 +1,4 @@
+node_modules
+dist
+.git
+.github
diff --git a/.env.example b/.env.example
@@ -0,0 +1,8 @@
+NODE_ENV=development
+RABBITMQ_URL=amqp://app:app@localhost:5672
+REDIS_URL=redis://localhost:6379
+API_GRAPHQL_URL=http://localhost:3000/graphql
+
+QUEUE_RAW_EVENT=source.raw.v1
+QUEUE_NORMALIZED_EVENT=source.normalized.v1
+SHARED_CONTRACTS_DIR=../shared-contracts
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -0,0 +1,28 @@
+name: CI
+
+on:
+  push:
+    branches: ["main", "master"]
+  pull_request:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20
+          cache: npm
+
+      - name: Install dependencies
+        run: npm install
+
+      - name: Type check
+        run: npm run check
+
+      - name: Build
+        run: npm run build
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,5 @@
+node_modules/
+dist/
+coverage/
+.env
+.DS_Store
diff --git a/Dockerfile b/Dockerfile
@@ -0,0 +1,17 @@
+FROM node:20-alpine AS deps
+WORKDIR /app
+COPY package.json package-lock.json* ./
+RUN npm install
+
+FROM deps AS build
+WORKDIR /app
+COPY tsconfig.json ./
+COPY src ./src
+RUN npm run build
+
+FROM node:20-alpine AS runtime
+WORKDIR /app
+ENV NODE_ENV=production
+COPY --from=deps /app/node_modules ./node_modules
+COPY --from=build /app/dist ./dist
+CMD ["node", "dist/main.js"]
diff --git a/README.md b/README.md
@@ -0,0 +1,44 @@
+# processing-worker
+
+![CI](https://img.shields.io/badge/CI-GitHub_Actions-2088FF?logo=githubactions&logoColor=white)
+![CD](https://img.shields.io/badge/CD-GitHub_Deploy-2ea44f?logo=github&logoColor=white)
+![Container](https://img.shields.io/badge/Container-GHCR-2496ED?logo=docker&logoColor=white)
+
+Сервис асинхронной обработки и нормализации собранных данных.
+
+## Что делает этот репозиторий
+
+- читает `source.raw.v1` из RabbitMQ;
+- валидирует raw-события;
+- нормализует payload до формата `source.normalized.v1`;
+- публикует нормализованные события и отправляет ingest mutation в `backend-api`.
+
+## Черновая реализация
+
+- consumer RabbitMQ (`src/messaging/queue-client.ts`);
+- нормализатор (`src/normalize.ts`);
+- валидация raw/normalized схем через Ajv;
+- GraphQL-клиент отправки в backend (`src/backend-client.ts`);
+- Dockerfile и CI workflow.
+
+## Локальный запуск
+
+```bash
+cp .env.example .env
+npm install
+npm run start:dev
+```
+
+## Важные переменные
+
+- `RABBITMQ_URL`
+- `QUEUE_RAW_EVENT`
+- `QUEUE_NORMALIZED_EVENT`
+- `API_GRAPHQL_URL`
+- `SHARED_CONTRACTS_DIR`
+
+## Связи с другими репозиториями
+
+- получает события от `scraper-service`;
+- использует схемы из `shared-contracts`;
+- отправляет результаты в `backend-api`.
diff --git a/package.json b/package.json
@@ -0,0 +1,26 @@
+{
+  "name": "processing-worker",
+  "version": "0.1.0",
+  "private": true,
+  "description": "Черновой worker для нормализации и передачи данных в backend-api.",
+  "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"
+  },
+  "dependencies": {
+    "ajv": "^8.17.1",
+    "amqplib": "^0.10.5",
+    "dotenv": "^16.4.7"
+  },
+  "devDependencies": {
+    "@types/amqplib": "^0.10.7",
+    "@types/node": "^22.10.5",
+    "ts-node-dev": "^2.0.0",
+    "typescript": "^5.7.2"
+  },
+  "engines": {
+    "node": ">=20"
+  }
+}
diff --git a/src/backend-client.ts b/src/backend-client.ts
@@ -0,0 +1,51 @@
+import type { NormalizedSourceEvent } from "./types";
+
+const INGEST_MUTATION = `
+mutation IngestNormalizedItem($input: IngestNormalizedItemInput!) {
+  ingestNormalizedItem(input: $input) {
+    accepted
+    idempotencyKey
+  }
+}
+`;
+
+export async function sendToBackend(
+  graphqlUrl: string,
+  event: NormalizedSourceEvent
+): Promise<void> {
+  const response = await fetch(graphqlUrl, {
+    method: "POST",
+    headers: {
+      "content-type": "application/json"
+    },
+    body: JSON.stringify({
+      query: INGEST_MUTATION,
+      variables: {
+        input: {
+          externalId: event.externalId,
+          source: event.source,
+          title: event.title,
+          customer: event.customer,
+          supplier: event.supplier,
+          amount: event.amount,
+          currency: event.currency,
+          publishedAt: event.publishedAt,
+          payloadVersion: event.payloadVersion,
+          rawPayload: { rawRef: event.rawRef, eventId: event.eventId }
+        }
+      }
+    })
+  });
+
+  if (!response.ok) {
+    throw new Error(`Backend API ответил с кодом ${response.status}`);
+  }
+
+  const payload = (await response.json()) as {
+    errors?: Array<{ message: string }>;
+  };
+
+  if (payload.errors?.length) {
+    throw new Error(`Ошибка GraphQL ingest: ${payload.errors.map((e) => e.message).join("; ")}`);
+  }
+}
diff --git a/src/config.ts b/src/config.ts
@@ -0,0 +1,9 @@
+import "dotenv/config";
+
+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"
+};
diff --git a/src/contracts/schema-validator.ts b/src/contracts/schema-validator.ts
@@ -0,0 +1,40 @@
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import Ajv from "ajv";
+import type { NormalizedSourceEvent, RawSourceEvent } from "../types";
+
+export function createSchemaValidators(sharedContractsDir: string) {
+  const rawSchema = JSON.parse(
+    readFileSync(join(sharedContractsDir, "events", "source-raw.v1.json"), "utf-8")
+  );
+  const normalizedSchema = JSON.parse(
+    readFileSync(join(sharedContractsDir, "events", "source-normalized.v1.json"), "utf-8")
+  );
+
+  const ajv = new Ajv({ allErrors: true });
+  const rawValidate = ajv.compile<RawSourceEvent>(rawSchema);
+  const normalizedValidate = ajv.compile<NormalizedSourceEvent>(normalizedSchema);
+
+  return {
+    validateRaw(event: RawSourceEvent): void {
+      if (rawValidate(event)) {
+        return;
+      }
+      throw new Error(
+        `Raw schema validation failed: ${(rawValidate.errors ?? [])
+          .map((err) => `${err.instancePath || "/"} ${err.message}`)
+          .join("; ")}`
+      );
+    },
+    validateNormalized(event: NormalizedSourceEvent): void {
+      if (normalizedValidate(event)) {
+        return;
+      }
+      throw new Error(
+        `Normalized schema validation failed: ${(normalizedValidate.errors ?? [])
+          .map((err) => `${err.instancePath || "/"} ${err.message}`)
+          .join("; ")}`
+      );
+    }
+  };
+}
diff --git a/src/main.ts b/src/main.ts
@@ -0,0 +1,40 @@
+import type { ConsumeMessage } from "amqplib";
+import { sendToBackend } from "./backend-client";
+import { config } from "./config";
+import { createSchemaValidators } from "./contracts/schema-validator";
+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);
+
+  await queue.init();
+  await queue.assertQueue(config.queueRaw);
+  await queue.assertQueue(config.queueNormalized);
+
+  console.log("[processing-worker] запущен", {
+    queueRaw: config.queueRaw,
+    queueNormalized: config.queueNormalized,
+    apiGraphqlUrl: config.apiGraphqlUrl
+  });
+
+  queue.consume(config.queueRaw, async (message: ConsumeMessage) => {
+    const raw = queue.parseMessage<RawSourceEvent>(message);
+    validators.validateRaw(raw);
+
+    const normalized = normalizeRawEvent(raw);
+    validators.validateNormalized(normalized);
+
+    await queue.publish(config.queueNormalized, normalized);
+    await sendToBackend(config.apiGraphqlUrl, normalized);
+
+    console.log(`[processing-worker] обработано событие ${raw.eventId}`);
+  });
+}
+
+void bootstrap().catch((error) => {
+  console.error("[processing-worker] фатальная ошибка", error);
+  process.exit(1);
+});
diff --git a/src/messaging/queue-client.ts b/src/messaging/queue-client.ts
@@ -0,0 +1,55 @@
+import { connect, type Channel, type Connection, type ConsumeMessage } from "amqplib";
+
+export class QueueClient {
+  private connection?: Connection;
+  private channel?: Channel;
+
+  constructor(private readonly rabbitmqUrl: string) {}
+
+  async init(): Promise<void> {
+    this.connection = await connect(this.rabbitmqUrl);
+    this.channel = await this.connection.createChannel();
+  }
+
+  async assertQueue(queue: string): Promise<void> {
+    if (!this.channel) {
+      throw new Error("QueueClient не инициализирован");
+    }
+    await this.channel.assertQueue(queue, { durable: true });
+  }
+
+  async publish(queue: string, payload: unknown): Promise<void> {
+    if (!this.channel) {
+      throw new Error("QueueClient не инициализирован");
+    }
+
+    this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
+      contentType: "application/json",
+      persistent: true
+    });
+  }
+
+  consume(queue: string, handler: (message: ConsumeMessage) => Promise<void>): void {
+    if (!this.channel) {
+      throw new Error("QueueClient не инициализирован");
+    }
+
+    void this.channel.consume(queue, async (msg) => {
+      if (!msg) {
+        return;
+      }
+
+      try {
+        await handler(msg);
+        this.channel?.ack(msg);
+      } catch (error) {
+        console.error("[processing-worker] Ошибка обработки сообщения", error);
+        this.channel?.nack(msg, false, false);
+      }
+    });
+  }
+
+  parseMessage<T>(message: ConsumeMessage): T {
+    return JSON.parse(message.content.toString("utf-8")) as T;
+  }
+}
diff --git a/src/normalize.ts b/src/normalize.ts
@@ -0,0 +1,28 @@
+import type { NormalizedSourceEvent, RawSourceEvent } from "./types";
+
+export function normalizeRawEvent(input: RawSourceEvent): NormalizedSourceEvent {
+  const raw = input.raw;
+
+  const toStringOrUndefined = (value: unknown): string | undefined =>
+    typeof value === "string" && value.trim() ? value : undefined;
+
+  const toNumberOrUndefined = (value: unknown): number | undefined =>
+    typeof value === "number" && Number.isFinite(value) ? value : undefined;
+
+  const title = toStringOrUndefined(raw.title) ?? "Без названия";
+
+  return {
+    eventId: input.eventId,
+    source: input.source,
+    payloadVersion: "v1",
+    externalId: `${input.source}-${input.eventId}`,
+    title,
+    customer: toStringOrUndefined(raw.customer),
+    supplier: toStringOrUndefined(raw.supplier),
+    amount: toNumberOrUndefined(raw.amount),
+    currency: toStringOrUndefined(raw.currency) ?? "RUB",
+    publishedAt: toStringOrUndefined(raw.publishedAt) ?? input.collectedAt,
+    normalizedAt: new Date().toISOString(),
+    rawRef: input.url
+  };
+}
diff --git a/src/types.ts b/src/types.ts
@@ -0,0 +1,23 @@
+export type RawSourceEvent = {
+  eventId: string;
+  source: string;
+  collectedAt: string;
+  url: string;
+  payloadVersion: "v1";
+  raw: Record<string, unknown>;
+};
+
+export type NormalizedSourceEvent = {
+  eventId: string;
+  source: string;
+  payloadVersion: "v1";
+  externalId: string;
+  title: string;
+  customer?: string;
+  supplier?: string;
+  amount?: number;
+  currency?: string;
+  publishedAt?: string;
+  normalizedAt: string;
+  rawRef?: string;
+};
diff --git a/tsconfig.json b/tsconfig.json

-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
 +node_modules
 +dist
 +.git
 +.github