From 0153b794b1424ec14b7c1a4cc1157fa2b37568ad Mon Sep 17 00:00:00 2001
From: Brian Bischoff <brian.bischoff@globetracker.com>
Date: Tue, 8 Oct 2024 10:37:15 +0100
Subject: [PATCH] feat: 1st prod ready release

---
 .env.test                                |   6 +----
 README.md                                |  32 ++++++++++++++++-------
 bun.lockb                                | Bin 29594 -> 29594 bytes
 services/docker-compose.yaml             |  30 +++++++++++++++++++++
 src/env/index.ts                         |   4 ++-
 src/server/routes/transform/transform.ts |  27 +++++++++++++++----
 6 files changed, 78 insertions(+), 21 deletions(-)
 create mode 100644 services/docker-compose.yaml

diff --git a/.env.test b/.env.test
index 697209c..809bbe1 100644
--- a/.env.test
+++ b/.env.test
@@ -7,9 +7,5 @@ LOG_LEVEL=error
 # Set this one to true if you want pretty logs. Set to empty (not false) if you want to disable it.
 # In production, you should set this to empty (not false)
 LOG_PRETTY=true
-
 # Service
-SERVICE_PORT=8888
-
-
-
+SERVICE_PORT=3000
diff --git a/README.md b/README.md
index 939a49a..9e54253 100644
--- a/README.md
+++ b/README.md
@@ -1,17 +1,28 @@
-# Generic Kafka Forward 2
-
+# Generic Kafka Forward
 
+Forwards events from (flowcore)[https://www.flowcore.io] to Kafka.
 
 ### Environment Variables
 
-| Environment Variable            | Description                         |   Type   | Default Value | Required |
-|---------------------------------|-------------------------------------|:--------:|---------------|:--------:|
-| LOG_LEVEL                       | The log level                       | `string` | `info`        |          |
-| LOG_PRETTY                      | Whether to log in pretty format     | `int`    | `0`           |          |
-| SERVICE_PORT                    | The port the service will listen on | `int`    | `3000`        |          |
-
-
-# Development
+| Environment Variable             | Description                            |   Type    | Default Value                             | Required |
+|----------------------------------|----------------------------------------|:---------:|-------------------------------------------|:--------:|
+| NODE_ENV                         | The environment the app is running in  | `string`  | `production`                              |          |
+| LOG_LEVEL                        | The log level                          | `string`  | `info`                                    |          |
+| LOG_PRETTY                       | Whether to log in pretty format        | `boolean` | `false`                                   |          |
+| KAFKA_BROKER                     | Kafka broker URL                       | `string`  |                                           |    ✓     |
+| KAFKA_TOPIC                      | Kafka topic                            | `string`  |                                           |    ✓     |
+| KAFKA_GROUP_ID                   | Kafka group ID                         | `string`  | `flowcore-generic-kafka-forwarder`        |          |
+| KAFKA_CLIENT_ID                  | Kafka client ID                        | `string`  | `flowcore-generic-kafka-forwarder-client` |          |
+| KAFKA_SSL                        | Use SSL for Kafka connection           | `boolean` | `false`                                   |          |
+| KAFKA_SASL_MECHANISM             | Kafka SASL mechanism                   | `string`  |                                           |          |
+| KAFKA_USERNAME                   | Kafka username                         | `string`  | `""`                                      |          |
+| KAFKA_PASSWORD                   | Kafka password                         | `string`  | `""`                                      |          |
+| KAFKAJS_NO_PARTITIONER_WARNING   | Disable partitioner warning            | `boolean` | `true`                                    |          |
+| KAFKA_AWS_AUTHORIZATION_IDENTITY | Kafka AWS authorization identity       | `string`  | `""`                                      |          |
+| KAFKA_KEY_PATH                   | Kafka key path                         | `string`  | `key`                                     |          |
+| KAFKA_IGNORE_EMPTY_KEY           | Ignore empty Kafka key                 | `boolean` | `false`                                   |          |
+| KAFKA_ADD_FLOWCORE_HEADERS       | Add Flowcore headers to Kafka messages | `boolean` | `false`                                   |          |
+| SERVICE_PORT                     | The port the service will listen on    |   `int`   | `3000`                                    |          |
 
 ## App setup
 
@@ -24,6 +35,7 @@ cp .env.example .env
 
 # You can create a new api key in flowcore and set the `FLOWCORE_API_KEY` environment variable in the `.env` file.
 ```
+
 ## Building
 
 ```bash
diff --git a/bun.lockb b/bun.lockb
index c1d2e4e6ad64a46d5cc28d927ab32a47e6e0478c..f10cd067d4b913ab80b83f3d9710fe71b6c82e53 100755
GIT binary patch
delta 24
fcmbRBoN?B3#tr@foQ!cMdWPnDh6bA>0-TfqaJmQ8

delta 24
acmbRBoN?B3#tr@foJ<T*usI^YNf`iN*ajZ}

diff --git a/services/docker-compose.yaml b/services/docker-compose.yaml
new file mode 100644
index 0000000..8b38228
--- /dev/null
+++ b/services/docker-compose.yaml
@@ -0,0 +1,30 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.0
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "2181:2181"
+
+  kafka:
+    image: confluentinc/cp-kafka:7.4.0
+    depends_on:
+      - zookeeper
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+    ports:
+      - "9092:9092"
+
+  kafdrop:
+    image: obsidiandynamics/kafdrop:latest
+    depends_on:
+      - kafka
+    environment:
+      KAFKA_BROKERCONNECT: kafka:9092
+      JVM_OPTS: -Xms32M -Xmx64M
+    ports:
+      - "9000:9000"
diff --git a/src/env/index.ts b/src/env/index.ts
index 2d79aa8..a45cddc 100644
--- a/src/env/index.ts
+++ b/src/env/index.ts
@@ -15,6 +15,8 @@ const envSchema = z.object({
   KAFKAJS_NO_PARTITIONER_WARNING: z.coerce.boolean().default(true),
   KAFKA_AWS_AUTHORIZATION_IDENTITY: z.coerce.string().default(""),
   KAFKA_KEY_PATH: z.string().default("key"),
+  KAFKA_IGNORE_EMPTY_KEY: z.coerce.boolean().default(false),
+  KAFKA_ADD_FLOWCORE_HEADERS: z.coerce.boolean().default(false),
 
   PORT: z.coerce.number().default(3000),
 })
@@ -22,7 +24,7 @@ const envSchema = z.object({
 const env = envSchema.safeParse(process.env)
 
 if (!env.success) {
-  console.error("Missing environment variables", env.error.flatten().fieldErrors)
+  console.error("Missing environment variables.", env.error.flatten().fieldErrors)
   process.exit(1)
 }
 
diff --git a/src/server/routes/transform/transform.ts b/src/server/routes/transform/transform.ts
index 70db7a6..decf5a5 100644
--- a/src/server/routes/transform/transform.ts
+++ b/src/server/routes/transform/transform.ts
@@ -27,13 +27,30 @@ export const transform = new Elysia({
       // const result = Value.Parse(ExamplePayloadDto, payload)
       // Extract the key from the payload using the configurable path
       const keyPath = env.KAFKA_KEY_PATH
-      const key = get(event, keyPath)
-      console.log("transform", event)
+      const key_value = get(payload, keyPath)
+
+      if (!key_value) {
+        if (env.KAFKA_IGNORE_EMPTY_KEY) {
+          logger.warn(`No key found in the payload, using path: '${keyPath}'`)
+          return
+        }
+      }
+
+      const headers: {
+        [p: string]: string
+      } = {}
+
+      if (env.KAFKA_ADD_FLOWCORE_HEADERS) {
+        headers["x-flowcore-event-source"] = `${event.dataCore}/${event.aggregator}/${event.eventType}/${event.eventId}`
+        headers["x-flowcore-timeBucket"] = event.timeBucket
+      }
+
+      logger.debug(`Sending message to Kafka topic: ${env.KAFKA_TOPIC} with key: ${key_value}`)
 
       // Send message to Kafka
-      await producer.send({
-        topic: "your-topic",
-        messages: [{ key: key, value: JSON.stringify(payload) }],
+      return await producer.send({
+        topic: env.KAFKA_TOPIC,
+        messages: [{ key: key_value, value: JSON.stringify(payload), headers }],
       })
     },
     {