From ecb7209b53dc4ad7a751188156bec56caabba9b6 Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Thu, 1 Oct 2026 11:32:14 +0300
Subject: [PATCH 01/11] add tracing and mock service for local dev

---
 docker-compose.yaml                  |  93 ++++++++++++++++++
 docker/user-data-service/server.js   |  36 +++++++
 src/common/tracing.ts                |  19 +++-
 src/process/models/processManager.ts | 139 +++++++++++++++++----------
 src/process/tracing.ts               |  24 +++++
 5 files changed, 259 insertions(+), 52 deletions(-)
 create mode 100644 docker-compose.yaml
 create mode 100644 docker/user-data-service/server.js
 create mode 100644 src/process/tracing.ts

diff --git a/docker-compose.yaml b/docker-compose.yaml
new file mode 100644
index 0000000..60d9a92
--- /dev/null
+++ b/docker-compose.yaml
@@ -0,0 +1,93 @@
+services:
+  kafka:
+    image: apache/kafka:3.9.1
+    environment:
+      - KAFKA_NODE_ID=1
+      - KAFKA_PROCESS_ROLES=broker,controller
+      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
+      - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
+      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
+      - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
+      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
+    ports:
+      - '9092:9092'
+    volumes:
+      - kafka-data:/var/lib/kafka/data
+    healthcheck:
+      test: ['CMD-SHELL', '/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092']
+      interval: 10s
+      timeout: 10s
+      retries: 15
+
+  elasticsearch:
+    image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+      - ES_JAVA_OPTS=-Xms512m -Xmx512m
+    ports:
+      - '9200:9200'
+    volumes:
+      - elasticsearch-data:/usr/share/elasticsearch/data
+    healthcheck:
+      test: ['CMD-SHELL', "curl -s http://localhost:9200/_cluster/health | grep -q '\"status\"'"]
+      interval: 10s
+      timeout: 5s
+      retries: 15
+
+  kafka-init:
+    image: apache/kafka:3.9.1
+    depends_on:
+      kafka:
+        condition: service_healthy
+    entrypoint: ['/bin/sh', '-c']
+    command:
+      - '/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists --topic topic --partitions 1 --replication-factor 1'
+    restart: 'no'
+
+  user-data-service:
+    image: node:20-alpine
+    working_dir: /usr/src/app
+    volumes:
+      - ./docker/user-data-service:/usr/src/app:ro
+    command: ['node', 'server.js']
+    ports:
+      - '5000:8080'
+
+  jaeger:
+    image: jaegertracing/all-in-one:1.62.0
+    environment:
+      - COLLECTOR_OTLP_ENABLED=true
+    ports:
+      - '16686:16686' # web UI
+      - '4318:4318' # OTLP HTTP receiver
+
+  app:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    restart: on-failure
+    environment:
+      - TELEMETRY_TRACING_ENABLED=true
+      - TELEMETRY_TRACING_URL=http://jaeger:4318/v1/traces
+      - USER_DATA_SERVICE_ENDPOINT=http://user-data-service:8080/user_data
+    ports:
+      - '8080:8080'
+    depends_on:
+      kafka-init:
+        condition: service_completed_successfully
+      elasticsearch:
+        condition: service_healthy
+      user-data-service:
+        condition: service_started
+      jaeger:
+        condition: service_started
+
+volumes:
+  kafka-data:
+  elasticsearch-data:
diff --git a/docker/user-data-service/server.js b/docker/user-data-service/server.js
new file mode 100644
index 0000000..f05d6d0
--- /dev/null
+++ b/docker/user-data-service/server.js
@@ -0,0 +1,36 @@
+// Minimal mock of the external userData service, based on the README's reference implementation.
+// Only intended for local docker-compose use.
+const http = require('http');
+
+const port = process.env.PORT ?? 8080;
+
+const users = {
+  'avi@mapcolonies.net': {
+    firstName: 'avi',
+    lastName: 'map',
+    displayName: 'mapcolonies/avi',
+    mail: 'avi@mapcolonies.net',
+    domains: ['USA', 'FRANCE'],
+  },
+};
+
+const server = http.createServer((req, res) => {
+  const match = /^\/user_data\/([^/?]+)/.exec(req.url ?? '');
+
+  if (req.method !== 'GET' || !match) {
+    res.writeHead(404, { 'Content-Type': 'application/json' });
+    res.end(JSON.stringify({ message: 'not found' }));
+    return;
+  }
+
+  const userId = decodeURIComponent(match[1]);
+  console.log('new request', { userid: userId });
+  const user = users[userId] ?? { [userId]: null };
+
+  res.writeHead(200, { 'Content-Type': 'application/json' });
+  res.end(JSON.stringify(user));
+});
+
+server.listen(port, () => {
+  console.log(`user-data-service mock is running on port ${port}`);
+});
diff --git a/src/common/tracing.ts b/src/common/tracing.ts
index d51d6b6..3444575 100644
--- a/src/common/tracing.ts
+++ b/src/common/tracing.ts
@@ -1,5 +1,8 @@
-import { Tracing } from '@map-colonies/telemetry';
-import { IGNORED_INCOMING_TRACE_ROUTES, IGNORED_OUTGOING_TRACE_ROUTES } from './constants';
+import { asyncCallWithSpan, callWithSpan, Tracing } from '@map-colonies/telemetry';
+import { trace, type Span, type SpanOptions, type Tracer } from '@opentelemetry/api';
+import { IGNORED_INCOMING_TRACE_ROUTES, IGNORED_OUTGOING_TRACE_ROUTES, SERVICE_NAME } from './constants';
+
+const getServiceTracer = (): Tracer => trace.getTracer(SERVICE_NAME);
 
 const tracing = new Tracing(undefined, {
   // eslint-disable-next-line @typescript-eslint/naming-convention
@@ -18,3 +21,15 @@ const tracing = new Tracing(undefined, {
 tracing.start();
 
 export { tracing };
+
+/**
+ * Wraps an async function with a span. When tracing is disabled, fn runs unchanged and span is undefined.
+ */
+export const withSpan = async <T>(spanName: string, spanOptions: SpanOptions, fn: (span?: Span) => Promise<T>): Promise<T> =>
+  asyncCallWithSpan(fn, getServiceTracer(), spanName, spanOptions);
+
+/**
+ * Wraps a sync function with a span. When tracing is disabled, fn runs unchanged and span is undefined.
+ */
+export const withSpanSync = <T>(spanName: string, spanOptions: SpanOptions, fn: (span?: Span) => T): T =>
+  callWithSpan(fn, getServiceTracer(), spanName, spanOptions);
diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts
index 3e0e4aa..dce91a5 100644
--- a/src/process/models/processManager.ts
+++ b/src/process/models/processManager.ts
@@ -2,8 +2,10 @@ import { Logger } from '@map-colonies/js-logger';
 import { center } from '@turf/center';
 import { inject, injectable } from 'tsyringe';
 import { SERVICES } from '../../common/constants';
+import { withSpan } from '../../common/tracing';
 import { EnrichResponse, FeedbackResponse, IApplication } from '../../common/interfaces';
 import { fetchUserDataService } from '../../common/utils';
+import { ProcessSpanName, ProcessAttributes } from '../tracing';
 
 const arabicRegex = /[\u0600-\u06FF]/;
 
@@ -15,29 +17,47 @@ export class ProcessManager {
   ) {}
 
   public async process(feedbackResponse: FeedbackResponse): Promise<EnrichResponse> {
-    const token = JSON.parse(Buffer.from(feedbackResponse.geocodingResponse.apiKey.split('.')[1], 'base64').toString()) as { sub: string };
-    const text = this.getQueryText(feedbackResponse);
-
-    const enrichedResponse: EnrichResponse = {
-      query: {
-        language: arabicRegex.test(text) ? 'ar' : 'he',
-        text,
-      },
-      result: {
-        rank: null,
+    return withSpan(
+      ProcessSpanName.MANAGER_PROCESS,
+      {
+        attributes: {
+          [ProcessAttributes.SITE]: feedbackResponse.geocodingResponse.site,
+          [ProcessAttributes.CHOSEN_RESULT_ID]: feedbackResponse.chosenResultId ?? undefined,
+        },
       },
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-      system: token?.sub,
-      site: feedbackResponse.geocodingResponse.site,
-      // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-      // @ts-expect-error
-      duration: new Date(feedbackResponse.responseTime) - new Date(feedbackResponse.geocodingResponse.respondedAt),
-      timestamp: new Date(),
-    };
-    if (feedbackResponse.chosenResultId === null) {
-      return enrichedResponse;
-    }
-    return this.enrichData(feedbackResponse, enrichedResponse);
+      async (span) => {
+        const token = JSON.parse(Buffer.from(feedbackResponse.geocodingResponse.apiKey.split('.')[1], 'base64').toString()) as { sub: string };
+        const text = this.getQueryText(feedbackResponse);
+
+        const enrichedResponse: EnrichResponse = {
+          query: {
+            language: arabicRegex.test(text) ? 'ar' : 'he',
+            text,
+          },
+          result: {
+            rank: null,
+          },
+          // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+          system: token?.sub,
+          site: feedbackResponse.geocodingResponse.site,
+          // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+          // @ts-expect-error
+          duration: new Date(feedbackResponse.responseTime) - new Date(feedbackResponse.geocodingResponse.respondedAt),
+          timestamp: new Date(),
+        };
+
+        span?.setAttributes({
+          [ProcessAttributes.LANGUAGE]: enrichedResponse.query.language,
+          [ProcessAttributes.SYSTEM]: enrichedResponse.system,
+          [ProcessAttributes.DURATION_MS]: enrichedResponse.duration,
+        });
+
+        if (feedbackResponse.chosenResultId === null) {
+          return enrichedResponse;
+        }
+        return this.enrichData(feedbackResponse, enrichedResponse);
+      }
+    );
   }
 
   public getQueryText(feedbackResponse: FeedbackResponse): string {
@@ -52,33 +72,52 @@ export class ProcessManager {
   }
 
   public async enrichData(feedbackResponse: FeedbackResponse, enrichedResponse: EnrichResponse): Promise<EnrichResponse> {
-    const chosenResult: number = feedbackResponse.chosenResultId as number;
-    const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult];
-
-    const { endpoint, queryParams, headers } = this.appConfig.userDataService;
-    const userId = feedbackResponse.geocodingResponse.userId;
-
-    if (userId !== undefined) {
-      const fetchedUserData = await fetchUserDataService(endpoint, userId, queryParams, headers);
-
-      enrichedResponse.user = {
-        name: userId,
-      };
-
-      const userData = fetchedUserData[userId] ?? fetchedUserData;
-      enrichedResponse.user = { ...enrichedResponse.user, ...userData };
-    }
-
-    enrichedResponse.result = {
-      rank: chosenResult,
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-      score: selectedResponse.properties?._score ?? 0,
-      source: selectedResponse.properties.matches[0].source,
-      layer: selectedResponse.properties.matches[0].layer,
-      name: selectedResponse.properties.names.default,
-      region: selectedResponse.properties.regions[0].region,
-      location: center(selectedResponse),
-    };
-    return enrichedResponse;
+    return withSpan(
+      ProcessSpanName.MANAGER_ENRICH_DATA,
+      { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: feedbackResponse.chosenResultId ?? undefined } },
+      async (span) => {
+        const chosenResult: number = feedbackResponse.chosenResultId as number;
+        const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult];
+
+        const { endpoint, queryParams, headers } = this.appConfig.userDataService;
+        const userId = feedbackResponse.geocodingResponse.userId;
+
+        if (userId !== undefined) {
+          const fetchedUserData = await withSpan(
+            ProcessSpanName.MANAGER_FETCH_USER_DATA,
+            { attributes: { [ProcessAttributes.USER_ID]: userId } },
+            async () => fetchUserDataService(endpoint, userId, queryParams, headers)
+          );
+
+          enrichedResponse.user = {
+            name: userId,
+          };
+
+          const userData = fetchedUserData[userId] ?? fetchedUserData;
+          enrichedResponse.user = { ...enrichedResponse.user, ...userData };
+        }
+
+        enrichedResponse.result = {
+          rank: chosenResult,
+          // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+          score: selectedResponse.properties?._score ?? 0,
+          source: selectedResponse.properties.matches[0].source,
+          layer: selectedResponse.properties.matches[0].layer,
+          name: selectedResponse.properties.names.default,
+          region: selectedResponse.properties.regions[0].region,
+          location: center(selectedResponse),
+        };
+
+        span?.setAttributes({
+          [ProcessAttributes.RESULT_RANK]: chosenResult,
+          [ProcessAttributes.RESULT_SCORE]: enrichedResponse.result.score,
+          [ProcessAttributes.RESULT_SOURCE]: enrichedResponse.result.source,
+          [ProcessAttributes.RESULT_LAYER]: enrichedResponse.result.layer,
+          [ProcessAttributes.RESULT_REGION]: enrichedResponse.result.region,
+        });
+
+        return enrichedResponse;
+      }
+    );
   }
 }
diff --git a/src/process/tracing.ts b/src/process/tracing.ts
new file mode 100644
index 0000000..cc6105b
--- /dev/null
+++ b/src/process/tracing.ts
@@ -0,0 +1,24 @@
+/* eslint-disable @typescript-eslint/naming-convention */
+export const ProcessSpanName = {
+  MANAGER_PROCESS: 'process.manager.process',
+  MANAGER_ENRICH_DATA: 'process.manager.enrich_data',
+  MANAGER_FETCH_USER_DATA: 'process.manager.fetch_user_data',
+} as const;
+
+export type ProcessSpanName = (typeof ProcessSpanName)[keyof typeof ProcessSpanName];
+
+export const ProcessAttributes = {
+  SITE: 'process.site',
+  LANGUAGE: 'process.query.language',
+  SYSTEM: 'process.system',
+  DURATION_MS: 'process.duration_ms',
+  CHOSEN_RESULT_ID: 'process.chosen_result_id',
+  USER_ID: 'process.user.id',
+  RESULT_RANK: 'process.result.rank',
+  RESULT_SCORE: 'process.result.score',
+  RESULT_SOURCE: 'process.result.source',
+  RESULT_LAYER: 'process.result.layer',
+  RESULT_REGION: 'process.result.region',
+} as const;
+
+export type ProcessAttributes = (typeof ProcessAttributes)[keyof typeof ProcessAttributes];

From 193a3df4d25198653fccfe618a1b7987762dcc3e Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Mon, 5 Oct 2026 15:07:51 +0300
Subject: [PATCH 02/11] feat: more metrics

---
 README.md                                     | 12 +++
 scripts/debug-kafka-consumer.js               | 77 +++++++++++++++++++
 .../sample-payloads/feedback-response.json    | 44 +++++++++++
 src/process/models/processManager.ts          | 39 ++++++++--
 src/streamerBuilder.ts                        | 65 ++++++++++++----
 src/streamerTracing.ts                        | 16 ++++
 6 files changed, 228 insertions(+), 25 deletions(-)
 create mode 100755 scripts/debug-kafka-consumer.js
 create mode 100644 scripts/sample-payloads/feedback-response.json
 create mode 100644 src/streamerTracing.ts

diff --git a/README.md b/README.md
index 805cd8c..56586e8 100644
--- a/README.md
+++ b/README.md
@@ -90,6 +90,18 @@ npm run start
 
 ```
 
+## Debugging the Kafka Consumer Locally
+
+To debug `StreamerBuilder`'s message handling without waiting for real traffic, start the local dependencies, run the service, and produce a FeedbackResponse payload shaped like production at it:
+
+```bash
+docker compose up kafka kafka-init elasticsearch user-data-service jaeger
+npm run start:dev
+npm run debug:produce-kafka
+```
+
+`npm run start:dev` already points at the docker-compose stack (Kafka's host-reachable `localhost:9094` listener, Elasticsearch on `localhost:9200`, the mock user-data-service on `localhost:5000`, and tracing exported to Jaeger at `localhost:4318`) — see the `start:dev` script in [package.json](package.json) if you need to point it elsewhere. `docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample. Traces for each run show up in the Jaeger UI at http://localhost:16686.
+
 ## Running Tests
 
 To run tests, run the following command
diff --git a/scripts/debug-kafka-consumer.js b/scripts/debug-kafka-consumer.js
new file mode 100755
index 0000000..73c3c87
--- /dev/null
+++ b/scripts/debug-kafka-consumer.js
@@ -0,0 +1,77 @@
+#!/usr/bin/env node
+/**
+ * Produces a FeedbackResponse message to a local Kafka topic so the consumer
+ * (`StreamerBuilder`, run via `npm run start:dev`) can be stepped through/debugged
+ * against a payload shaped like a real one, instead of waiting for production traffic.
+ *
+ * Prerequisites: `docker compose up kafka kafka-init elasticsearch user-data-service jaeger`,
+ * then run the service on the host with KAFKA_BROKERS=localhost:9094 so it points at the
+ * docker-compose broker's host-reachable listener (the default PLAINTEXT listener only
+ * advertises the in-network hostname "kafka", which the host can't resolve).
+ *
+ * Usage:
+ *   node scripts/debug-kafka-consumer.js [path/to/payload.json] [--new-request-id]
+ *
+ *   path/to/payload.json   A FeedbackResponse JSON payload to replay, e.g. one captured
+ *                          from production. Defaults to scripts/sample-payloads/feedback-response.json.
+ *   --new-request-id       Overwrite requestId with a fresh one, useful for re-sending the
+ *                          same payload repeatedly without colliding on request ids in traces/Elastic.
+ *
+ * Env vars:
+ *   KAFKA_BROKERS   comma separated list, default "localhost:9094"
+ *   KAFKA_TOPIC     default "topic" (matches config/default.json kafkaTopics.input[0])
+ */
+
+const fs = require('fs');
+const path = require('path');
+const { Kafka } = require('kafkajs');
+
+const DEFAULT_PAYLOAD_PATH = path.join(__dirname, 'sample-payloads', 'feedback-response.json');
+
+function parseArgs(argv) {
+  const positional = argv.filter((arg) => !arg.startsWith('--'));
+  return {
+    payloadPath: positional[0] ?? DEFAULT_PAYLOAD_PATH,
+    newRequestId: argv.includes('--new-request-id'),
+  };
+}
+
+function loadPayload(payloadPath, newRequestId) {
+  const raw = fs.readFileSync(payloadPath, 'utf-8');
+  const payload = JSON.parse(raw);
+
+  if (newRequestId) {
+    payload.requestId = `debug-${process.pid}-${process.hrtime.bigint()}`;
+  }
+
+  return payload;
+}
+
+async function main() {
+  const { payloadPath, newRequestId } = parseArgs(process.argv.slice(2));
+  const payload = loadPayload(payloadPath, newRequestId);
+
+  const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(',');
+  const topic = process.env.KAFKA_TOPIC ?? 'topic';
+
+  const kafka = new Kafka({ clientId: 'debug-kafka-producer', brokers });
+  const producer = kafka.producer();
+
+  await producer.connect();
+  try {
+    const result = await producer.send({
+      topic,
+      messages: [{ value: JSON.stringify(payload) }],
+    });
+
+    console.log(`Sent ${payloadPath} to topic "${topic}" on [${brokers.join(', ')}] (requestId: ${payload.requestId})`);
+    console.log(result);
+  } finally {
+    await producer.disconnect();
+  }
+}
+
+main().catch((error) => {
+  console.error('Failed to produce debug message:', error);
+  process.exitCode = 1;
+});
diff --git a/scripts/sample-payloads/feedback-response.json b/scripts/sample-payloads/feedback-response.json
new file mode 100644
index 0000000..15d7e2b
--- /dev/null
+++ b/scripts/sample-payloads/feedback-response.json
@@ -0,0 +1,44 @@
+{
+  "requestId": "debug-local-001",
+  "chosenResultId": 0,
+  "responseTime": "2026-10-05T10:00:03.500Z",
+  "geocodingResponse": {
+    "userId": "avi@mapcolonies.net",
+    "apiKey": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJtYXAtY29sb25pZXMtdGVzdCIsIm5hbWUiOiJKb2huIERvZSIsImlhdCI6MTUxNjIzOTAyMn0.cyKspTifStJI-w2lYk05QaFNyyHYDtUupJddqVj7ZS8",
+    "site": "geocoding-web-client",
+    "respondedAt": "2026-10-05T10:00:03.000Z",
+    "response": {
+      "type": "FeatureCollection",
+      "geocoding": {
+        "version": "1.0",
+        "query": {
+          "query": "דיזנגוף 99, תל אביב"
+        }
+      },
+      "features": [
+        {
+          "type": "Feature",
+          "geometry": { "type": "Point", "coordinates": [34.774302, 32.084041] },
+          "properties": {
+            "type": "Point",
+            "matches": [{ "layer": "address", "source": "osm", "source_id": ["way/123456"] }],
+            "names": { "default": "דיזנגוף 99, תל אביב - יפו" },
+            "regions": [{ "region": "תל אביב - יפו", "sub_region_names": ["מרכז"] }],
+            "_score": 0.97
+          }
+        },
+        {
+          "type": "Feature",
+          "geometry": { "type": "Point", "coordinates": [34.781, 32.09] },
+          "properties": {
+            "type": "Point",
+            "matches": [{ "layer": "address", "source": "osm", "source_id": ["way/654321"] }],
+            "names": { "default": "דיזנגוף 101, תל אביב - יפו" },
+            "regions": [{ "region": "תל אביב - יפו", "sub_region_names": ["מרכז"] }],
+            "_score": 0.81
+          }
+        }
+      ]
+    }
+  }
+}
diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts
index dce91a5..52a2c1b 100644
--- a/src/process/models/processManager.ts
+++ b/src/process/models/processManager.ts
@@ -1,20 +1,36 @@
 import { Logger } from '@map-colonies/js-logger';
 import { center } from '@turf/center';
 import { inject, injectable } from 'tsyringe';
+import axios from 'axios';
+import { Counter, Histogram, Meter } from '@opentelemetry/api';
 import { SERVICES } from '../../common/constants';
 import { withSpan } from '../../common/tracing';
-import { EnrichResponse, FeedbackResponse, IApplication } from '../../common/interfaces';
+import { EnrichResponse, FeedbackResponse, IApplication, UserDataServiceResponse } from '../../common/interfaces';
 import { fetchUserDataService } from '../../common/utils';
 import { ProcessSpanName, ProcessAttributes } from '../tracing';
 
 const arabicRegex = /[\u0600-\u06FF]/;
 
+const getErrorType = (error: unknown): string => {
+  if (axios.isAxiosError(error)) {
+    return error.response !== undefined ? `http_${error.response.status}` : error.code ?? 'network_error';
+  }
+  return error instanceof Error ? error.name : 'unknown_error';
+};
+
 @injectable()
 export class ProcessManager {
+  private readonly processDurationRecorder: Histogram;
+  private readonly userDataServiceErrorsCounter: Counter;
+
   public constructor(
     @inject(SERVICES.LOGGER) private readonly logger: Logger,
-    @inject(SERVICES.APPLICATION) private readonly appConfig: IApplication
-  ) {}
+    @inject(SERVICES.APPLICATION) private readonly appConfig: IApplication,
+    @inject(SERVICES.METER) private readonly meter: Meter
+  ) {
+    this.processDurationRecorder = meter.createHistogram('process_duration_ms', { unit: 'ms' });
+    this.userDataServiceErrorsCounter = meter.createCounter('user_data_service_errors');
+  }
 
   public async process(feedbackResponse: FeedbackResponse): Promise<EnrichResponse> {
     return withSpan(
@@ -51,6 +67,7 @@ export class ProcessManager {
           [ProcessAttributes.SYSTEM]: enrichedResponse.system,
           [ProcessAttributes.DURATION_MS]: enrichedResponse.duration,
         });
+        this.processDurationRecorder.record(enrichedResponse.duration);
 
         if (feedbackResponse.chosenResultId === null) {
           return enrichedResponse;
@@ -83,11 +100,17 @@ export class ProcessManager {
         const userId = feedbackResponse.geocodingResponse.userId;
 
         if (userId !== undefined) {
-          const fetchedUserData = await withSpan(
-            ProcessSpanName.MANAGER_FETCH_USER_DATA,
-            { attributes: { [ProcessAttributes.USER_ID]: userId } },
-            async () => fetchUserDataService(endpoint, userId, queryParams, headers)
-          );
+          let fetchedUserData: UserDataServiceResponse;
+          try {
+            fetchedUserData = await withSpan(
+              ProcessSpanName.MANAGER_FETCH_USER_DATA,
+              { attributes: { [ProcessAttributes.USER_ID]: userId } },
+              async () => fetchUserDataService(endpoint, userId, queryParams, headers)
+            );
+          } catch (error) {
+            this.userDataServiceErrorsCounter.add(1, { type: getErrorType(error) });
+            throw error;
+          }
 
           enrichedResponse.user = {
             name: userId,
diff --git a/src/streamerBuilder.ts b/src/streamerBuilder.ts
index eedfbc1..b1bfc65 100644
--- a/src/streamerBuilder.ts
+++ b/src/streamerBuilder.ts
@@ -4,9 +4,12 @@ import { Logger } from '@map-colonies/js-logger';
 import { Client, ClientOptions } from '@elastic/elasticsearch';
 import { Consumer, ConsumerConfig, Kafka } from 'kafkajs';
 import { CleanupRegistry } from '@map-colonies/cleanup-registry';
+import { Counter, Histogram, Meter } from '@opentelemetry/api';
 import { SERVICES } from './common/constants';
+import { withSpan } from './common/tracing';
 import { FeedbackResponse, IConfig, KafkaOptions } from './common/interfaces';
 import { ProcessManager } from './process/models/processManager';
+import { StreamerAttributes, StreamerSpanName } from './streamerTracing';
 
 interface KafkaTopics {
   input: string[];
@@ -20,13 +23,20 @@ export class StreamerBuilder {
   private readonly kafka: Kafka;
   private readonly consumer: Consumer;
   private readonly elasticClient: Client;
+  private readonly kafkaErrorsCounter: Counter;
+  private readonly recordsIndexedCounter: Counter;
+  private readonly elasticIndexDurationRecorder: Histogram;
 
   public constructor(
     @inject(SERVICES.CONFIG) private readonly config: IConfig,
     @inject(SERVICES.LOGGER) private readonly logger: Logger,
     @inject(SERVICES.CLEANUP_REGISTRY) private readonly cleanupRegistry: CleanupRegistry,
-    @inject(ProcessManager) private readonly manager: ProcessManager
+    @inject(ProcessManager) private readonly manager: ProcessManager,
+    @inject(SERVICES.METER) private readonly meter: Meter
   ) {
+    this.kafkaErrorsCounter = meter.createCounter('kafka_errors');
+    this.recordsIndexedCounter = meter.createCounter('records_indexed');
+    this.elasticIndexDurationRecorder = meter.createHistogram('elastic_index_duration_ms', { unit: 'ms' });
     let kafkaConfig = config.get<KafkaOptions>('kafka');
     if (typeof kafkaConfig.brokers === 'string' || kafkaConfig.brokers instanceof String) {
       kafkaConfig = {
@@ -53,32 +63,53 @@ export class StreamerBuilder {
   public async build(): Promise<void> {
     const { input: inputTopic } = this.config.get<KafkaTopics>('kafkaTopics');
 
-    await this.consumer.connect();
-    await this.consumer.subscribe({ topics: inputTopic });
+    await withSpan(StreamerSpanName.KAFKA_STARTUP, { attributes: { [StreamerAttributes.KAFKA_TOPIC]: inputTopic.join(',') } }, async () => {
+      await this.consumer.connect();
+      await this.consumer.subscribe({ topics: inputTopic });
+    });
     this.logger.info(`Kafka consumer subscribed successfully to ${inputTopic.toString()}`);
 
     await this.consumer.run({
-      eachMessage: async ({ message }) => {
+      eachMessage: async ({ topic, partition, message }) => {
         const value = message.value?.toString();
-        if (value != undefined) {
-          try {
-            const input = JSON.parse(value) as FeedbackResponse;
-            const requestId = input.requestId;
-            const output = await this.manager.process(input);
+        if (value == undefined) {
+          return;
+        }
+
+        try {
+          await withSpan(
+            StreamerSpanName.KAFKA_HANDLE_MESSAGE,
+            {
+              attributes: {
+                [StreamerAttributes.KAFKA_TOPIC]: topic,
+                [StreamerAttributes.KAFKA_PARTITION]: partition,
+                [StreamerAttributes.KAFKA_OFFSET]: message.offset,
+              },
+            },
+            async (span) => {
+              const input = JSON.parse(value) as FeedbackResponse;
+              const requestId = input.requestId;
+              const output = await this.manager.process(input);
+              const index = this.config.get<string>(elasticIndex);
+              span?.setAttribute(StreamerAttributes.ELASTIC_INDEX_NAME, index);
 
-            await this.elasticClient.index({
-              index: this.config.get<string>(elasticIndex),
-              body: output,
-            });
-            this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`);
-          } catch (error) {
-            this.logger.error(`Error: Could not add data to elastic. Reason: ${(error as Error).message}`);
-          }
+              const indexStartTime = Date.now();
+              await this.elasticClient.index({ index, body: output });
+              this.elasticIndexDurationRecorder.record(Date.now() - indexStartTime);
+              this.recordsIndexedCounter.add(1);
+
+              this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`);
+            }
+          );
+        } catch (error) {
+          this.kafkaErrorsCounter.add(1);
+          this.logger.error(`Error: Could not add data to elastic. Reason: ${(error as Error).message}`);
         }
       },
     });
 
     this.consumer.on(this.consumer.events.CRASH, (error) => {
+      this.kafkaErrorsCounter.add(1);
       this.logger.error(error);
       this.logger.error(error.payload.error);
     });
diff --git a/src/streamerTracing.ts b/src/streamerTracing.ts
new file mode 100644
index 0000000..8d9d0d9
--- /dev/null
+++ b/src/streamerTracing.ts
@@ -0,0 +1,16 @@
+/* eslint-disable @typescript-eslint/naming-convention */
+export const StreamerSpanName = {
+  KAFKA_STARTUP: 'streamer.kafka.startup',
+  KAFKA_HANDLE_MESSAGE: 'streamer.kafka.handle_message',
+} as const;
+
+export type StreamerSpanName = (typeof StreamerSpanName)[keyof typeof StreamerSpanName];
+
+export const StreamerAttributes = {
+  KAFKA_TOPIC: 'streamer.kafka.topic',
+  KAFKA_PARTITION: 'streamer.kafka.partition',
+  KAFKA_OFFSET: 'streamer.kafka.offset',
+  ELASTIC_INDEX_NAME: 'streamer.elastic.index_name',
+} as const;
+
+export type StreamerAttributes = (typeof StreamerAttributes)[keyof typeof StreamerAttributes];

From 9a809ef78b0a4f3efd5986c474fdfb36ab4242ff Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Mon, 5 Oct 2026 15:08:22 +0300
Subject: [PATCH 03/11] feat: add docker compose to debug locally

---
 docker-compose.yaml | 28 ++++------------------------
 package.json        |  3 ++-
 2 files changed, 6 insertions(+), 25 deletions(-)

diff --git a/docker-compose.yaml b/docker-compose.yaml
index 60d9a92..1547ded 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -4,10 +4,10 @@ services:
     environment:
       - KAFKA_NODE_ID=1
       - KAFKA_PROCESS_ROLES=broker,controller
-      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
-      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
+      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:9094
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,HOST://localhost:9094
       - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER
-      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
       - KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
       - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
       - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
@@ -16,6 +16,7 @@ services:
       - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
     ports:
       - '9092:9092'
+      - '9094:9094' # host-reachable listener, so the service/scripts can run outside docker while it talks to this broker
     volumes:
       - kafka-data:/var/lib/kafka/data
     healthcheck:
@@ -67,27 +68,6 @@ services:
       - '16686:16686' # web UI
       - '4318:4318' # OTLP HTTP receiver
 
-  app:
-    build:
-      context: .
-      dockerfile: Dockerfile
-    restart: on-failure
-    environment:
-      - TELEMETRY_TRACING_ENABLED=true
-      - TELEMETRY_TRACING_URL=http://jaeger:4318/v1/traces
-      - USER_DATA_SERVICE_ENDPOINT=http://user-data-service:8080/user_data
-    ports:
-      - '8080:8080'
-    depends_on:
-      kafka-init:
-        condition: service_completed_successfully
-      elasticsearch:
-        condition: service_healthy
-      user-data-service:
-        condition: service_started
-      jaeger:
-        condition: service_started
-
 volumes:
   kafka-data:
   elasticsearch-data:
diff --git a/package.json b/package.json
index 2014baf..6edc79e 100644
--- a/package.json
+++ b/package.json
@@ -18,9 +18,10 @@
     "prebuild": "npm run clean",
     "build": "tsc --project tsconfig.build.json  && npm run assets:copy",
     "start": "npm run build && cd dist && node --require ./common/tracing.js ./index.js",
-    "start:dev": "npm run build && cd dist && node --enable-source-maps --require ./common/tracing.js ./index.js",
+    "start:dev": "npm run build && cd dist && KAFKA_BROKERS=localhost:9094 ELASTIC_URL=http://localhost:9200 USER_DATA_SERVICE_ENDPOINT=http://localhost:5000/user_data TELEMETRY_TRACING_ENABLED=true TELEMETRY_TRACING_URL=http://localhost:4318/v1/traces node --enable-source-maps --require ./common/tracing.js ./index.js",
     "assets:copy": "copyfiles -f ./config/* ./dist/config && copyfiles -f ./openapi3.yaml ./dist/ && copyfiles ./package.json dist",
     "clean": "rimraf dist",
+    "debug:produce-kafka": "node scripts/debug-kafka-consumer.js",
     "install": "npx husky install"
   },
   "directories": {

From 769649439e392bd72e7d3e0a9af14ba8c754fb45 Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Mon, 5 Oct 2026 15:28:03 +0300
Subject: [PATCH 04/11] fix: ci errors

---
 README.md           | 27 +++++++++++++++++++--------
 docker-compose.yaml |  2 +-
 2 files changed, 20 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index 56586e8..ebf5e9b 100644
--- a/README.md
+++ b/README.md
@@ -90,17 +90,28 @@ npm run start
 
 ```
 
-## Debugging the Kafka Consumer Locally
+## Debugging Locally
 
-To debug `StreamerBuilder`'s message handling without waiting for real traffic, start the local dependencies, run the service, and produce a FeedbackResponse payload shaped like production at it:
+In production, [feedback-api](https://github.com/MapColonies/feedback-api) is the service that produces the Kafka message this service consumes: it combines the feedback it receives (`request_id`, `chosen_response_id`, `user_id`) with Geocoding's original response and publishes the result to the `kafkaTopics.input` topic. `StreamerBuilder` ([src/streamerBuilder.ts](src/streamerBuilder.ts)) is the consumer on the other end — it's the thing being debugged here.
 
-```bash
-docker compose up kafka kafka-init elasticsearch user-data-service jaeger
-npm run start:dev
-npm run debug:produce-kafka
-```
+To exercise that consumer locally without running feedback-api or waiting for real traffic, [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) stands in for feedback-api: it produces a FeedbackResponse payload, shaped exactly like what feedback-api sends, straight to a local Kafka broker.
+
+1. Start the dependencies (Kafka, Elasticsearch, the userData mock, and Jaeger for tracing):
+   ```bash
+   docker compose up kafka kafka-init elasticsearch user-data-service jaeger
+   ```
+2. Run the service in dev mode. `npm run start:dev` already points at this docker-compose stack (Kafka's host-reachable `localhost:9094` listener, Elasticsearch on `localhost:9200`, the mock user-data-service on `localhost:5000`, and tracing exported to Jaeger at `localhost:4318`) — see the `start:dev` script in [package.json](package.json) if you need to point it elsewhere:
+   ```bash
+   npm run start:dev
+   ```
+3. Produce a message, standing in for feedback-api:
+   ```bash
+   npm run debug:produce-kafka
+   ```
+
+`docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample, or `--new-request-id` to replay the same payload without colliding on request ids.
 
-`npm run start:dev` already points at the docker-compose stack (Kafka's host-reachable `localhost:9094` listener, Elasticsearch on `localhost:9200`, the mock user-data-service on `localhost:5000`, and tracing exported to Jaeger at `localhost:4318`) — see the `start:dev` script in [package.json](package.json) if you need to point it elsewhere. `docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample. Traces for each run show up in the Jaeger UI at http://localhost:16686.
+Once consumed, the enriched record lands in Elasticsearch (`curl localhost:9200/enrich_index/_search`) and the full trace — Kafka handling, the user-data lookup, and the Elasticsearch write — is viewable in the Jaeger UI at http://localhost:16686.
 
 ## Running Tests
 
diff --git a/docker-compose.yaml b/docker-compose.yaml
index 1547ded..f65c3a2 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -36,7 +36,7 @@ services:
     volumes:
       - elasticsearch-data:/usr/share/elasticsearch/data
     healthcheck:
-      test: ['CMD-SHELL', "curl -s http://localhost:9200/_cluster/health | grep -q '\"status\"'"]
+      test: ['CMD-SHELL', 'curl -s http://localhost:9200/_cluster/health | grep -q ''"status"''']
       interval: 10s
       timeout: 5s
       retries: 15

From a4fba357e8e010db600f8f436079dfee5da52fbd Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Mon, 5 Oct 2026 15:44:48 +0300
Subject: [PATCH 05/11] fix: readme for deploy

---
 README.md | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/README.md b/README.md
index ebf5e9b..9876aa3 100644
--- a/README.md
+++ b/README.md
@@ -132,3 +132,17 @@ To only run integration tests:
 ```bash
 npm run test:integration
 ```
+
+## Deploy to azure
+
+```bash
+docker build . --tag acrarolibotnonprod.azurecr.io/geocoding-enrichment:v1.7.0-rc2
+docker push acrarolibotnonprod.azurecr.io/geocoding-enrichment:v1.7.0-rc2
+```
+
+go to https://github.com/MapColonies/vector-helms/tree/master/geocoding-helms/charts/geocoding-enrichment
+update image in values and run:
+
+```
+helm upgrade --install vector-geocoding -f azure.dev.yaml . -n vector-dev
+```

From cc8b9a94a22fce04979ef0dd983cbcbbdef98c01 Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Tue, 6 Oct 2026 14:38:18 +0300
Subject: [PATCH 06/11] feat: add jira sync workflow

---
 .github/workflows/jira-integration.yaml.yml | 14 ++++++++++++++
 1 file changed, 14 insertions(+)
 create mode 100644 .github/workflows/jira-integration.yaml.yml

diff --git a/.github/workflows/jira-integration.yaml.yml b/.github/workflows/jira-integration.yaml.yml
new file mode 100644
index 0000000..34d4ee9
--- /dev/null
+++ b/.github/workflows/jira-integration.yaml.yml
@@ -0,0 +1,14 @@
+name: Jira Integration
+on:
+  pull_request:
+    types: [opened, edited, synchronize]
+
+permissions:
+  statuses: write
+  pull-requests: write
+
+jobs:
+  jira-validation:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: mapcolonies/javascript-github-actions/actions/jira-integration@jira-integration-v1

From d117e0518d51fc7081cc49957bf398a070161427 Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Tue, 6 Oct 2026 14:59:12 +0300
Subject: [PATCH 07/11] feat: move produce script to ts

---
 .dockerignore                                 |  3 +
 ...gration.yaml.yml => jira-integration.yaml} |  0
 docker-compose.yaml                           |  4 +-
 docker/user-data-service/server.js            | 36 ---------
 docker/user-data-service/server.ts            | 41 ++++++++++
 package-lock.json                             |  6 +-
 package.json                                  |  4 +-
 scripts/debug-kafka-consumer.js               | 77 -------------------
 scripts/debug-kafka-consumer.ts               | 63 +++++++++++++++
 src/process/models/processManager.ts          | 19 +++--
 src/streamerBuilder.ts                        |  8 +-
 src/streamerTracing.ts                        |  1 -
 tsconfig.json                                 |  2 +-
 13 files changed, 131 insertions(+), 133 deletions(-)
 rename .github/workflows/{jira-integration.yaml.yml => jira-integration.yaml} (100%)
 delete mode 100644 docker/user-data-service/server.js
 create mode 100644 docker/user-data-service/server.ts
 delete mode 100755 scripts/debug-kafka-consumer.js
 create mode 100755 scripts/debug-kafka-consumer.ts

diff --git a/.dockerignore b/.dockerignore
index bfbdee6..7f806c6 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -4,6 +4,9 @@ tests
 .github
 coverage
 node_modules
+docker
+scripts
+docker-compose.yaml
 .prettier*
 commitlint.config.json
 README.md
diff --git a/.github/workflows/jira-integration.yaml.yml b/.github/workflows/jira-integration.yaml
similarity index 100%
rename from .github/workflows/jira-integration.yaml.yml
rename to .github/workflows/jira-integration.yaml
diff --git a/docker-compose.yaml b/docker-compose.yaml
index f65c3a2..da0afb9 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -55,8 +55,8 @@ services:
     image: node:20-alpine
     working_dir: /usr/src/app
     volumes:
-      - ./docker/user-data-service:/usr/src/app:ro
-    command: ['node', 'server.js']
+      - .:/usr/src/app:ro
+    command: ['npx', 'ts-node', 'docker/user-data-service/server.ts']
     ports:
       - '5000:8080'
 
diff --git a/docker/user-data-service/server.js b/docker/user-data-service/server.js
deleted file mode 100644
index f05d6d0..0000000
--- a/docker/user-data-service/server.js
+++ /dev/null
@@ -1,36 +0,0 @@
-// Minimal mock of the external userData service, based on the README's reference implementation.
-// Only intended for local docker-compose use.
-const http = require('http');
-
-const port = process.env.PORT ?? 8080;
-
-const users = {
-  'avi@mapcolonies.net': {
-    firstName: 'avi',
-    lastName: 'map',
-    displayName: 'mapcolonies/avi',
-    mail: 'avi@mapcolonies.net',
-    domains: ['USA', 'FRANCE'],
-  },
-};
-
-const server = http.createServer((req, res) => {
-  const match = /^\/user_data\/([^/?]+)/.exec(req.url ?? '');
-
-  if (req.method !== 'GET' || !match) {
-    res.writeHead(404, { 'Content-Type': 'application/json' });
-    res.end(JSON.stringify({ message: 'not found' }));
-    return;
-  }
-
-  const userId = decodeURIComponent(match[1]);
-  console.log('new request', { userid: userId });
-  const user = users[userId] ?? { [userId]: null };
-
-  res.writeHead(200, { 'Content-Type': 'application/json' });
-  res.end(JSON.stringify(user));
-});
-
-server.listen(port, () => {
-  console.log(`user-data-service mock is running on port ${port}`);
-});
diff --git a/docker/user-data-service/server.ts b/docker/user-data-service/server.ts
new file mode 100644
index 0000000..84c5e5c
--- /dev/null
+++ b/docker/user-data-service/server.ts
@@ -0,0 +1,41 @@
+import { createServer, IncomingMessage, ServerResponse } from 'http';
+import httpStatus from 'http-status-codes';
+import { UserDataServiceResponse } from '../../src/common/interfaces';
+
+const port = 8080;
+
+// eslint-disable-next-line @typescript-eslint/naming-convention
+const jsonHeaders = { 'Content-Type': 'application/json' };
+
+/* eslint-disable @typescript-eslint/naming-convention */
+const users: UserDataServiceResponse = {
+  'avi@mapcolonies.net': {
+    firstName: 'avi',
+    lastName: 'map',
+    displayName: 'mapcolonies/avi',
+    mail: 'avi@mapcolonies.net',
+    domains: ['USA', 'FRANCE'],
+  },
+};
+/* eslint-enable @typescript-eslint/naming-convention */
+
+const server = createServer((req: IncomingMessage, res: ServerResponse) => {
+  const match = /^\/user_data\/([^/?]+)/.exec(req.url ?? '');
+
+  if (req.method !== 'GET' || !match) {
+    res.writeHead(httpStatus.NOT_FOUND, jsonHeaders);
+    res.end(JSON.stringify({ message: 'not found' }));
+    return;
+  }
+
+  const userId = decodeURIComponent(match[1]);
+  console.log('new request', { userid: userId });
+  const user = users[userId] ?? { [userId]: null };
+
+  res.writeHead(httpStatus.OK, jsonHeaders);
+  res.end(JSON.stringify(user));
+});
+
+server.listen(port, () => {
+  console.log(`user-data-service mock is running on port ${port}`);
+});
diff --git a/package-lock.json b/package-lock.json
index 1f1cfd7..a166933 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -50,6 +50,7 @@
         "@types/jest": "^29.5.2",
         "@types/jsonwebtoken": "^9.0.6",
         "@types/multer": "^1.4.7",
+        "@types/node": "^17.0.36",
         "@types/supertest": "^2.0.12",
         "@types/swagger-ui-express": "^4.1.3",
         "commitlint": "^17.6.6",
@@ -68,6 +69,7 @@
         "standard-version": "^9.5.0",
         "supertest": "^6.3.3",
         "ts-jest": "^29.0.5",
+        "ts-node": "^10.9.1",
         "typescript": "^5.1.3"
       }
     },
@@ -6677,7 +6679,8 @@
     "node_modules/@types/node": {
       "version": "17.0.36",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-17.0.36.tgz",
-      "integrity": "sha512-V3orv+ggDsWVHP99K3JlwtH20R7J4IhI1Kksgc+64q5VxgfRkQG8Ws3MFm/FZOKDYGy9feGFlZ70/HpCNe9QaA=="
+      "integrity": "sha512-V3orv+ggDsWVHP99K3JlwtH20R7J4IhI1Kksgc+64q5VxgfRkQG8Ws3MFm/FZOKDYGy9feGFlZ70/HpCNe9QaA==",
+      "license": "MIT"
     },
     "node_modules/@types/normalize-package-data": {
       "version": "2.4.1",
@@ -17980,6 +17983,7 @@
       "resolved": "https://registry.npmjs.org/ts-node/-/ts-node-10.9.1.tgz",
       "integrity": "sha512-NtVysVPkxxrwFGUUxGYhfux8k78pQB3JqYBXlLRZgdGUqTO5wU/UyHop5p70iEbGhB7q5KmiZiU0Y3KlJrScEw==",
       "dev": true,
+      "license": "MIT",
       "dependencies": {
         "@cspotcode/source-map-support": "^0.8.0",
         "@tsconfig/node10": "^1.0.7",
diff --git a/package.json b/package.json
index 6edc79e..9b82c9c 100644
--- a/package.json
+++ b/package.json
@@ -21,7 +21,7 @@
     "start:dev": "npm run build && cd dist && KAFKA_BROKERS=localhost:9094 ELASTIC_URL=http://localhost:9200 USER_DATA_SERVICE_ENDPOINT=http://localhost:5000/user_data TELEMETRY_TRACING_ENABLED=true TELEMETRY_TRACING_URL=http://localhost:4318/v1/traces node --enable-source-maps --require ./common/tracing.js ./index.js",
     "assets:copy": "copyfiles -f ./config/* ./dist/config && copyfiles -f ./openapi3.yaml ./dist/ && copyfiles ./package.json dist",
     "clean": "rimraf dist",
-    "debug:produce-kafka": "node scripts/debug-kafka-consumer.js",
+    "debug:produce-kafka": "ts-node scripts/debug-kafka-consumer.ts",
     "install": "npx husky install"
   },
   "directories": {
@@ -81,6 +81,7 @@
     "@types/jest": "^29.5.2",
     "@types/jsonwebtoken": "^9.0.6",
     "@types/multer": "^1.4.7",
+    "@types/node": "^17.0.36",
     "@types/supertest": "^2.0.12",
     "@types/swagger-ui-express": "^4.1.3",
     "commitlint": "^17.6.6",
@@ -99,6 +100,7 @@
     "standard-version": "^9.5.0",
     "supertest": "^6.3.3",
     "ts-jest": "^29.0.5",
+    "ts-node": "^10.9.1",
     "typescript": "^5.1.3"
   }
 }
diff --git a/scripts/debug-kafka-consumer.js b/scripts/debug-kafka-consumer.js
deleted file mode 100755
index 73c3c87..0000000
--- a/scripts/debug-kafka-consumer.js
+++ /dev/null
@@ -1,77 +0,0 @@
-#!/usr/bin/env node
-/**
- * Produces a FeedbackResponse message to a local Kafka topic so the consumer
- * (`StreamerBuilder`, run via `npm run start:dev`) can be stepped through/debugged
- * against a payload shaped like a real one, instead of waiting for production traffic.
- *
- * Prerequisites: `docker compose up kafka kafka-init elasticsearch user-data-service jaeger`,
- * then run the service on the host with KAFKA_BROKERS=localhost:9094 so it points at the
- * docker-compose broker's host-reachable listener (the default PLAINTEXT listener only
- * advertises the in-network hostname "kafka", which the host can't resolve).
- *
- * Usage:
- *   node scripts/debug-kafka-consumer.js [path/to/payload.json] [--new-request-id]
- *
- *   path/to/payload.json   A FeedbackResponse JSON payload to replay, e.g. one captured
- *                          from production. Defaults to scripts/sample-payloads/feedback-response.json.
- *   --new-request-id       Overwrite requestId with a fresh one, useful for re-sending the
- *                          same payload repeatedly without colliding on request ids in traces/Elastic.
- *
- * Env vars:
- *   KAFKA_BROKERS   comma separated list, default "localhost:9094"
- *   KAFKA_TOPIC     default "topic" (matches config/default.json kafkaTopics.input[0])
- */
-
-const fs = require('fs');
-const path = require('path');
-const { Kafka } = require('kafkajs');
-
-const DEFAULT_PAYLOAD_PATH = path.join(__dirname, 'sample-payloads', 'feedback-response.json');
-
-function parseArgs(argv) {
-  const positional = argv.filter((arg) => !arg.startsWith('--'));
-  return {
-    payloadPath: positional[0] ?? DEFAULT_PAYLOAD_PATH,
-    newRequestId: argv.includes('--new-request-id'),
-  };
-}
-
-function loadPayload(payloadPath, newRequestId) {
-  const raw = fs.readFileSync(payloadPath, 'utf-8');
-  const payload = JSON.parse(raw);
-
-  if (newRequestId) {
-    payload.requestId = `debug-${process.pid}-${process.hrtime.bigint()}`;
-  }
-
-  return payload;
-}
-
-async function main() {
-  const { payloadPath, newRequestId } = parseArgs(process.argv.slice(2));
-  const payload = loadPayload(payloadPath, newRequestId);
-
-  const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(',');
-  const topic = process.env.KAFKA_TOPIC ?? 'topic';
-
-  const kafka = new Kafka({ clientId: 'debug-kafka-producer', brokers });
-  const producer = kafka.producer();
-
-  await producer.connect();
-  try {
-    const result = await producer.send({
-      topic,
-      messages: [{ value: JSON.stringify(payload) }],
-    });
-
-    console.log(`Sent ${payloadPath} to topic "${topic}" on [${brokers.join(', ')}] (requestId: ${payload.requestId})`);
-    console.log(result);
-  } finally {
-    await producer.disconnect();
-  }
-}
-
-main().catch((error) => {
-  console.error('Failed to produce debug message:', error);
-  process.exitCode = 1;
-});
diff --git a/scripts/debug-kafka-consumer.ts b/scripts/debug-kafka-consumer.ts
new file mode 100755
index 0000000..1e79e0c
--- /dev/null
+++ b/scripts/debug-kafka-consumer.ts
@@ -0,0 +1,63 @@
+#!/usr/bin/env -S npx ts-node
+// See README.md "Debugging Locally" for usage, prerequisites, and env vars.
+
+import { readFileSync } from 'fs';
+import { join } from 'path';
+import { Kafka } from 'kafkajs';
+import { FeedbackResponse } from '../src/common/interfaces';
+
+const DEFAULT_PAYLOAD_PATH = join(__dirname, 'sample-payloads', 'feedback-response.json');
+const CLI_ARGS_OFFSET = 2; // skip `node` and the script path in argv
+
+interface ParsedArgs {
+  payloadPath: string;
+  newRequestId: boolean;
+}
+
+function parseArgs(argv: string[]): ParsedArgs {
+  const positional = argv.filter((arg) => !arg.startsWith('--'));
+  return {
+    payloadPath: positional[0] ?? DEFAULT_PAYLOAD_PATH,
+    newRequestId: argv.includes('--new-request-id'),
+  };
+}
+
+function loadPayload(payloadPath: string, newRequestId: boolean): FeedbackResponse {
+  const raw = readFileSync(payloadPath, 'utf-8');
+  const payload = JSON.parse(raw) as FeedbackResponse;
+
+  if (newRequestId) {
+    payload.requestId = `debug-${process.pid}-${process.hrtime.bigint()}`;
+  }
+
+  return payload;
+}
+
+async function main(): Promise<void> {
+  const { payloadPath, newRequestId } = parseArgs(process.argv.slice(CLI_ARGS_OFFSET));
+  const payload = loadPayload(payloadPath, newRequestId);
+
+  const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(',');
+  const topic = process.env.KAFKA_TOPIC ?? 'topic';
+
+  const kafka = new Kafka({ clientId: 'debug-kafka-producer', brokers });
+  const producer = kafka.producer();
+
+  await producer.connect();
+  try {
+    const result = await producer.send({
+      topic,
+      messages: [{ value: JSON.stringify(payload) }],
+    });
+
+    console.log(`Sent ${payloadPath} to topic "${topic}" on [${brokers.join(', ')}] (requestId: ${payload.requestId})`);
+    console.log(result);
+  } finally {
+    await producer.disconnect();
+  }
+}
+
+main().catch((error: unknown) => {
+  console.error('Failed to produce debug message:', error);
+  process.exitCode = 1;
+});
diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts
index 52a2c1b..9e80c59 100644
--- a/src/process/models/processManager.ts
+++ b/src/process/models/processManager.ts
@@ -53,12 +53,9 @@ export class ProcessManager {
           result: {
             rank: null,
           },
-          // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-          system: token?.sub,
+          system: token.sub,
           site: feedbackResponse.geocodingResponse.site,
-          // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-          // @ts-expect-error
-          duration: new Date(feedbackResponse.responseTime) - new Date(feedbackResponse.geocodingResponse.respondedAt),
+          duration: new Date(feedbackResponse.responseTime).getTime() - new Date(feedbackResponse.geocodingResponse.respondedAt).getTime(),
           timestamp: new Date(),
         };
 
@@ -69,9 +66,6 @@ export class ProcessManager {
         });
         this.processDurationRecorder.record(enrichedResponse.duration);
 
-        if (feedbackResponse.chosenResultId === null) {
-          return enrichedResponse;
-        }
         return this.enrichData(feedbackResponse, enrichedResponse);
       }
     );
@@ -89,11 +83,15 @@ export class ProcessManager {
   }
 
   public async enrichData(feedbackResponse: FeedbackResponse, enrichedResponse: EnrichResponse): Promise<EnrichResponse> {
+    const chosenResult = feedbackResponse.chosenResultId;
+    if (chosenResult === null) {
+      return enrichedResponse;
+    }
+
     return withSpan(
       ProcessSpanName.MANAGER_ENRICH_DATA,
-      { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: feedbackResponse.chosenResultId ?? undefined } },
+      { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: chosenResult } },
       async (span) => {
-        const chosenResult: number = feedbackResponse.chosenResultId as number;
         const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult];
 
         const { endpoint, queryParams, headers } = this.appConfig.userDataService;
@@ -108,6 +106,7 @@ export class ProcessManager {
               async () => fetchUserDataService(endpoint, userId, queryParams, headers)
             );
           } catch (error) {
+            this.logger.error(`Failed to fetch user data for user: ${userId}`, error);
             this.userDataServiceErrorsCounter.add(1, { type: getErrorType(error) });
             throw error;
           }
diff --git a/src/streamerBuilder.ts b/src/streamerBuilder.ts
index b1bfc65..a6ea585 100644
--- a/src/streamerBuilder.ts
+++ b/src/streamerBuilder.ts
@@ -23,6 +23,7 @@ export class StreamerBuilder {
   private readonly kafka: Kafka;
   private readonly consumer: Consumer;
   private readonly elasticClient: Client;
+  private readonly elasticIndexName: string;
   private readonly kafkaErrorsCounter: Counter;
   private readonly recordsIndexedCounter: Counter;
   private readonly elasticIndexDurationRecorder: Histogram;
@@ -57,6 +58,7 @@ export class StreamerBuilder {
     this.consumer = this.kafka.consumer(consumerConfig);
     elasticConfig = config.get<ClientOptions>('elastic');
     this.elasticClient = new Client(elasticConfig);
+    this.elasticIndexName = config.get<string>(elasticIndex);
     this.cleanupRegistry.register({ func: this.consumer.disconnect.bind(this.consumer) });
   }
 
@@ -86,15 +88,13 @@ export class StreamerBuilder {
                 [StreamerAttributes.KAFKA_OFFSET]: message.offset,
               },
             },
-            async (span) => {
+            async () => {
               const input = JSON.parse(value) as FeedbackResponse;
               const requestId = input.requestId;
               const output = await this.manager.process(input);
-              const index = this.config.get<string>(elasticIndex);
-              span?.setAttribute(StreamerAttributes.ELASTIC_INDEX_NAME, index);
 
               const indexStartTime = Date.now();
-              await this.elasticClient.index({ index, body: output });
+              await this.elasticClient.index({ index: this.elasticIndexName, body: output });
               this.elasticIndexDurationRecorder.record(Date.now() - indexStartTime);
               this.recordsIndexedCounter.add(1);
 
diff --git a/src/streamerTracing.ts b/src/streamerTracing.ts
index 8d9d0d9..505ad47 100644
--- a/src/streamerTracing.ts
+++ b/src/streamerTracing.ts
@@ -10,7 +10,6 @@ export const StreamerAttributes = {
   KAFKA_TOPIC: 'streamer.kafka.topic',
   KAFKA_PARTITION: 'streamer.kafka.partition',
   KAFKA_OFFSET: 'streamer.kafka.offset',
-  ELASTIC_INDEX_NAME: 'streamer.elastic.index_name',
 } as const;
 
 export type StreamerAttributes = (typeof StreamerAttributes)[keyof typeof StreamerAttributes];
diff --git a/tsconfig.json b/tsconfig.json
index f5df946..bc2fdd7 100644
--- a/tsconfig.json
+++ b/tsconfig.json
@@ -11,6 +11,6 @@
     "experimentalDecorators": true,
     "emitDecoratorMetadata": true
   },
-  "include": ["src"],
+  "include": ["src", "scripts", "docker"],
   "exclude": ["node_modules", "dist", "tests"]
 }

From 6b615eba771f395eca786efdb773ab9aea904df3 Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Wed, 7 Oct 2026 09:01:58 +0300
Subject: [PATCH 08/11] feat: remove counter

---
 src/process/models/processManager.ts | 22 ++--------------------
 src/streamerBuilder.ts               | 16 +---------------
 2 files changed, 3 insertions(+), 35 deletions(-)

diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts
index 9e80c59..c47dd0e 100644
--- a/src/process/models/processManager.ts
+++ b/src/process/models/processManager.ts
@@ -1,8 +1,6 @@
 import { Logger } from '@map-colonies/js-logger';
 import { center } from '@turf/center';
 import { inject, injectable } from 'tsyringe';
-import axios from 'axios';
-import { Counter, Histogram, Meter } from '@opentelemetry/api';
 import { SERVICES } from '../../common/constants';
 import { withSpan } from '../../common/tracing';
 import { EnrichResponse, FeedbackResponse, IApplication, UserDataServiceResponse } from '../../common/interfaces';
@@ -11,26 +9,12 @@ import { ProcessSpanName, ProcessAttributes } from '../tracing';
 
 const arabicRegex = /[\u0600-\u06FF]/;
 
-const getErrorType = (error: unknown): string => {
-  if (axios.isAxiosError(error)) {
-    return error.response !== undefined ? `http_${error.response.status}` : error.code ?? 'network_error';
-  }
-  return error instanceof Error ? error.name : 'unknown_error';
-};
-
 @injectable()
 export class ProcessManager {
-  private readonly processDurationRecorder: Histogram;
-  private readonly userDataServiceErrorsCounter: Counter;
-
   public constructor(
     @inject(SERVICES.LOGGER) private readonly logger: Logger,
-    @inject(SERVICES.APPLICATION) private readonly appConfig: IApplication,
-    @inject(SERVICES.METER) private readonly meter: Meter
-  ) {
-    this.processDurationRecorder = meter.createHistogram('process_duration_ms', { unit: 'ms' });
-    this.userDataServiceErrorsCounter = meter.createCounter('user_data_service_errors');
-  }
+    @inject(SERVICES.APPLICATION) private readonly appConfig: IApplication
+  ) {}
 
   public async process(feedbackResponse: FeedbackResponse): Promise<EnrichResponse> {
     return withSpan(
@@ -64,7 +48,6 @@ export class ProcessManager {
           [ProcessAttributes.SYSTEM]: enrichedResponse.system,
           [ProcessAttributes.DURATION_MS]: enrichedResponse.duration,
         });
-        this.processDurationRecorder.record(enrichedResponse.duration);
 
         return this.enrichData(feedbackResponse, enrichedResponse);
       }
@@ -107,7 +90,6 @@ export class ProcessManager {
             );
           } catch (error) {
             this.logger.error(`Failed to fetch user data for user: ${userId}`, error);
-            this.userDataServiceErrorsCounter.add(1, { type: getErrorType(error) });
             throw error;
           }
 
diff --git a/src/streamerBuilder.ts b/src/streamerBuilder.ts
index a6ea585..474774b 100644
--- a/src/streamerBuilder.ts
+++ b/src/streamerBuilder.ts
@@ -4,7 +4,6 @@ import { Logger } from '@map-colonies/js-logger';
 import { Client, ClientOptions } from '@elastic/elasticsearch';
 import { Consumer, ConsumerConfig, Kafka } from 'kafkajs';
 import { CleanupRegistry } from '@map-colonies/cleanup-registry';
-import { Counter, Histogram, Meter } from '@opentelemetry/api';
 import { SERVICES } from './common/constants';
 import { withSpan } from './common/tracing';
 import { FeedbackResponse, IConfig, KafkaOptions } from './common/interfaces';
@@ -24,20 +23,13 @@ export class StreamerBuilder {
   private readonly consumer: Consumer;
   private readonly elasticClient: Client;
   private readonly elasticIndexName: string;
-  private readonly kafkaErrorsCounter: Counter;
-  private readonly recordsIndexedCounter: Counter;
-  private readonly elasticIndexDurationRecorder: Histogram;
 
   public constructor(
     @inject(SERVICES.CONFIG) private readonly config: IConfig,
     @inject(SERVICES.LOGGER) private readonly logger: Logger,
     @inject(SERVICES.CLEANUP_REGISTRY) private readonly cleanupRegistry: CleanupRegistry,
-    @inject(ProcessManager) private readonly manager: ProcessManager,
-    @inject(SERVICES.METER) private readonly meter: Meter
+    @inject(ProcessManager) private readonly manager: ProcessManager
   ) {
-    this.kafkaErrorsCounter = meter.createCounter('kafka_errors');
-    this.recordsIndexedCounter = meter.createCounter('records_indexed');
-    this.elasticIndexDurationRecorder = meter.createHistogram('elastic_index_duration_ms', { unit: 'ms' });
     let kafkaConfig = config.get<KafkaOptions>('kafka');
     if (typeof kafkaConfig.brokers === 'string' || kafkaConfig.brokers instanceof String) {
       kafkaConfig = {
@@ -92,24 +84,18 @@ export class StreamerBuilder {
               const input = JSON.parse(value) as FeedbackResponse;
               const requestId = input.requestId;
               const output = await this.manager.process(input);
-
-              const indexStartTime = Date.now();
               await this.elasticClient.index({ index: this.elasticIndexName, body: output });
-              this.elasticIndexDurationRecorder.record(Date.now() - indexStartTime);
-              this.recordsIndexedCounter.add(1);
 
               this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`);
             }
           );
         } catch (error) {
-          this.kafkaErrorsCounter.add(1);
           this.logger.error(`Error: Could not add data to elastic. Reason: ${(error as Error).message}`);
         }
       },
     });
 
     this.consumer.on(this.consumer.events.CRASH, (error) => {
-      this.kafkaErrorsCounter.add(1);
       this.logger.error(error);
       this.logger.error(error.payload.error);
     });

From 6bd14fc542d3bab557df275591693d2edd7cb4ae Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Wed, 7 Oct 2026 09:07:26 +0300
Subject: [PATCH 09/11] feat: format fix

---
 src/process/models/processManager.ts | 90 +++++++++++++---------------
 1 file changed, 43 insertions(+), 47 deletions(-)

diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts
index c47dd0e..dffcee3 100644
--- a/src/process/models/processManager.ts
+++ b/src/process/models/processManager.ts
@@ -71,57 +71,53 @@ export class ProcessManager {
       return enrichedResponse;
     }
 
-    return withSpan(
-      ProcessSpanName.MANAGER_ENRICH_DATA,
-      { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: chosenResult } },
-      async (span) => {
-        const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult];
-
-        const { endpoint, queryParams, headers } = this.appConfig.userDataService;
-        const userId = feedbackResponse.geocodingResponse.userId;
-
-        if (userId !== undefined) {
-          let fetchedUserData: UserDataServiceResponse;
-          try {
-            fetchedUserData = await withSpan(
-              ProcessSpanName.MANAGER_FETCH_USER_DATA,
-              { attributes: { [ProcessAttributes.USER_ID]: userId } },
-              async () => fetchUserDataService(endpoint, userId, queryParams, headers)
-            );
-          } catch (error) {
-            this.logger.error(`Failed to fetch user data for user: ${userId}`, error);
-            throw error;
-          }
-
-          enrichedResponse.user = {
-            name: userId,
-          };
-
-          const userData = fetchedUserData[userId] ?? fetchedUserData;
-          enrichedResponse.user = { ...enrichedResponse.user, ...userData };
+    return withSpan(ProcessSpanName.MANAGER_ENRICH_DATA, { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: chosenResult } }, async (span) => {
+      const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult];
+
+      const { endpoint, queryParams, headers } = this.appConfig.userDataService;
+      const userId = feedbackResponse.geocodingResponse.userId;
+
+      if (userId !== undefined) {
+        let fetchedUserData: UserDataServiceResponse;
+        try {
+          fetchedUserData = await withSpan(
+            ProcessSpanName.MANAGER_FETCH_USER_DATA,
+            { attributes: { [ProcessAttributes.USER_ID]: userId } },
+            async () => fetchUserDataService(endpoint, userId, queryParams, headers)
+          );
+        } catch (error) {
+          this.logger.error(`Failed to fetch user data for user: ${userId}`, error);
+          throw error;
         }
 
-        enrichedResponse.result = {
-          rank: chosenResult,
-          // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-          score: selectedResponse.properties?._score ?? 0,
-          source: selectedResponse.properties.matches[0].source,
-          layer: selectedResponse.properties.matches[0].layer,
-          name: selectedResponse.properties.names.default,
-          region: selectedResponse.properties.regions[0].region,
-          location: center(selectedResponse),
+        enrichedResponse.user = {
+          name: userId,
         };
 
-        span?.setAttributes({
-          [ProcessAttributes.RESULT_RANK]: chosenResult,
-          [ProcessAttributes.RESULT_SCORE]: enrichedResponse.result.score,
-          [ProcessAttributes.RESULT_SOURCE]: enrichedResponse.result.source,
-          [ProcessAttributes.RESULT_LAYER]: enrichedResponse.result.layer,
-          [ProcessAttributes.RESULT_REGION]: enrichedResponse.result.region,
-        });
-
-        return enrichedResponse;
+        const userData = fetchedUserData[userId] ?? fetchedUserData;
+        enrichedResponse.user = { ...enrichedResponse.user, ...userData };
       }
-    );
+
+      enrichedResponse.result = {
+        rank: chosenResult,
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        score: selectedResponse.properties?._score ?? 0,
+        source: selectedResponse.properties.matches[0].source,
+        layer: selectedResponse.properties.matches[0].layer,
+        name: selectedResponse.properties.names.default,
+        region: selectedResponse.properties.regions[0].region,
+        location: center(selectedResponse),
+      };
+
+      span?.setAttributes({
+        [ProcessAttributes.RESULT_RANK]: chosenResult,
+        [ProcessAttributes.RESULT_SCORE]: enrichedResponse.result.score,
+        [ProcessAttributes.RESULT_SOURCE]: enrichedResponse.result.source,
+        [ProcessAttributes.RESULT_LAYER]: enrichedResponse.result.layer,
+        [ProcessAttributes.RESULT_REGION]: enrichedResponse.result.region,
+      });
+
+      return enrichedResponse;
+    });
   }
 }

From c21339caef59a1926172b4b1f8192f8eb3648c3d Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Wed, 7 Oct 2026 10:10:22 +0300
Subject: [PATCH 10/11] feat: fix

---
 README.md                       | 4 ++--
 docker-compose.yaml             | 3 ++-
 package-lock.json               | 1 -
 package.json                    | 3 +--
 scripts/debug-kafka-consumer.ts | 5 ++---
 src/streamerBuilder.ts          | 4 +---
 6 files changed, 8 insertions(+), 12 deletions(-)

diff --git a/README.md b/README.md
index 9876aa3..369b3af 100644
--- a/README.md
+++ b/README.md
@@ -94,7 +94,7 @@ npm run start
 
 In production, [feedback-api](https://github.com/MapColonies/feedback-api) is the service that produces the Kafka message this service consumes: it combines the feedback it receives (`request_id`, `chosen_response_id`, `user_id`) with Geocoding's original response and publishes the result to the `kafkaTopics.input` topic. `StreamerBuilder` ([src/streamerBuilder.ts](src/streamerBuilder.ts)) is the consumer on the other end — it's the thing being debugged here.
 
-To exercise that consumer locally without running feedback-api or waiting for real traffic, [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) stands in for feedback-api: it produces a FeedbackResponse payload, shaped exactly like what feedback-api sends, straight to a local Kafka broker.
+To exercise that consumer locally without running feedback-api or waiting for real traffic, [scripts/debug-kafka-consumer.ts](scripts/debug-kafka-consumer.ts) stands in for feedback-api: it produces a FeedbackResponse payload, shaped exactly like what feedback-api sends, straight to a local Kafka broker.
 
 1. Start the dependencies (Kafka, Elasticsearch, the userData mock, and Jaeger for tracing):
    ```bash
@@ -109,7 +109,7 @@ To exercise that consumer locally without running feedback-api or waiting for re
    npm run debug:produce-kafka
    ```
 
-`docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.js](scripts/debug-kafka-consumer.js) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample, or `--new-request-id` to replay the same payload without colliding on request ids.
+`docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.ts](scripts/debug-kafka-consumer.ts) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample, or `--new-request-id` to replay the same payload without colliding on request ids.
 
 Once consumed, the enriched record lands in Elasticsearch (`curl localhost:9200/enrich_index/_search`) and the full trace — Kafka handling, the user-data lookup, and the Elasticsearch write — is viewable in the Jaeger UI at http://localhost:16686.
 
diff --git a/docker-compose.yaml b/docker-compose.yaml
index da0afb9..75ad91e 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -56,7 +56,8 @@ services:
     working_dir: /usr/src/app
     volumes:
       - .:/usr/src/app:ro
-    command: ['npx', 'ts-node', 'docker/user-data-service/server.ts']
+      - /usr/src/app/dist
+    command: ['sh', '-c', 'npx tsc --project tsconfig.json --skipLibCheck && node dist/docker/user-data-service/server.js']
     ports:
       - '5000:8080'
 
diff --git a/package-lock.json b/package-lock.json
index a166933..df4a481 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -69,7 +69,6 @@
         "standard-version": "^9.5.0",
         "supertest": "^6.3.3",
         "ts-jest": "^29.0.5",
-        "ts-node": "^10.9.1",
         "typescript": "^5.1.3"
       }
     },
diff --git a/package.json b/package.json
index 9b82c9c..b8da2da 100644
--- a/package.json
+++ b/package.json
@@ -21,7 +21,7 @@
     "start:dev": "npm run build && cd dist && KAFKA_BROKERS=localhost:9094 ELASTIC_URL=http://localhost:9200 USER_DATA_SERVICE_ENDPOINT=http://localhost:5000/user_data TELEMETRY_TRACING_ENABLED=true TELEMETRY_TRACING_URL=http://localhost:4318/v1/traces node --enable-source-maps --require ./common/tracing.js ./index.js",
     "assets:copy": "copyfiles -f ./config/* ./dist/config && copyfiles -f ./openapi3.yaml ./dist/ && copyfiles ./package.json dist",
     "clean": "rimraf dist",
-    "debug:produce-kafka": "ts-node scripts/debug-kafka-consumer.ts",
+    "debug:produce-kafka": "tsc --project tsconfig.json --skipLibCheck && node dist/scripts/debug-kafka-consumer.js",
     "install": "npx husky install"
   },
   "directories": {
@@ -100,7 +100,6 @@
     "standard-version": "^9.5.0",
     "supertest": "^6.3.3",
     "ts-jest": "^29.0.5",
-    "ts-node": "^10.9.1",
     "typescript": "^5.1.3"
   }
 }
diff --git a/scripts/debug-kafka-consumer.ts b/scripts/debug-kafka-consumer.ts
index 1e79e0c..2ce312e 100755
--- a/scripts/debug-kafka-consumer.ts
+++ b/scripts/debug-kafka-consumer.ts
@@ -1,12 +1,11 @@
-#!/usr/bin/env -S npx ts-node
-// See README.md "Debugging Locally" for usage, prerequisites, and env vars.
+// Run via `npm run debug:produce-kafka`. See README.md "Debugging Locally" for usage, prerequisites, and env vars.
 
 import { readFileSync } from 'fs';
 import { join } from 'path';
 import { Kafka } from 'kafkajs';
 import { FeedbackResponse } from '../src/common/interfaces';
 
-const DEFAULT_PAYLOAD_PATH = join(__dirname, 'sample-payloads', 'feedback-response.json');
+const DEFAULT_PAYLOAD_PATH = join(process.cwd(), 'scripts', 'sample-payloads', 'feedback-response.json');
 const CLI_ARGS_OFFSET = 2; // skip `node` and the script path in argv
 
 interface ParsedArgs {
diff --git a/src/streamerBuilder.ts b/src/streamerBuilder.ts
index 474774b..2a13059 100644
--- a/src/streamerBuilder.ts
+++ b/src/streamerBuilder.ts
@@ -22,7 +22,6 @@ export class StreamerBuilder {
   private readonly kafka: Kafka;
   private readonly consumer: Consumer;
   private readonly elasticClient: Client;
-  private readonly elasticIndexName: string;
 
   public constructor(
     @inject(SERVICES.CONFIG) private readonly config: IConfig,
@@ -50,7 +49,6 @@ export class StreamerBuilder {
     this.consumer = this.kafka.consumer(consumerConfig);
     elasticConfig = config.get<ClientOptions>('elastic');
     this.elasticClient = new Client(elasticConfig);
-    this.elasticIndexName = config.get<string>(elasticIndex);
     this.cleanupRegistry.register({ func: this.consumer.disconnect.bind(this.consumer) });
   }
 
@@ -84,7 +82,7 @@ export class StreamerBuilder {
               const input = JSON.parse(value) as FeedbackResponse;
               const requestId = input.requestId;
               const output = await this.manager.process(input);
-              await this.elasticClient.index({ index: this.elasticIndexName, body: output });
+              await this.elasticClient.index({ index: this.config.get<string>(elasticIndex), body: output });
 
               this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`);
             }

From 1a39cbc4635a9f23bf90a74ac2b0c0439f97378c Mon Sep 17 00:00:00 2001
From: Yosef Danan <317816049+yosefdanan555@users.noreply.github.com>
Date: Wed, 7 Oct 2026 10:12:56 +0300
Subject: [PATCH 11/11] feat: FIX CUSTOM ENV

---
 config/custom-environment-variables.json | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/config/custom-environment-variables.json b/config/custom-environment-variables.json
index 02d41be..17634bc 100644
--- a/config/custom-environment-variables.json
+++ b/config/custom-environment-variables.json
@@ -49,7 +49,7 @@
     "sslPaths": {
       "ca": "KAFKA_CA_PATH",
       "key": "KAFKA_KEY_PATH",
-      "cert": "KAFKA_CERT_PATkafH"
+      "cert": "KAFKA_CERT_PATH"
     },
     "sasl": {
       "mechanism": "KAFKA_SASL_MECHANISM",