feat(reccoom): add consumer

README.md

@@ -284,6 +284,14 @@ This project is deployed in accordance to the [DargStack template](https://githu
 
     You cannot access the recommendation service directly.
 
+ - ### `reccoom_consumer`
+
+    You can track the recommender's event streaming consumer using `redpanda-console`.
+
+ - ### `reccoom_migration`
+
+    You cannot access the recommender's database migration service directly.
+
  - ### `reccoom_postgres`
 
     You can access reccoom's database via `adminer`.

src/development/stack.yml

@@ -567,6 +567,50 @@ services:
     volumes:
       - ../../../reccoom/:/srv/app/ #DARGSTACK-REMOVE
       - ./configurations/postgraphile/jwtES256.key.pub:/run/configurations/jwtES256.key.pub:ro
+  reccoom_consumer:
+    # You can track the recommender's event streaming consumer using `redpanda-console`.
+    command: ["uv", "run", "python", "-m", "src.consumer"]
+    environment:
+      KAFKA_BOOTSTRAP_SERVERS: redpanda:9092
+      POSTGRES_HOST: postgres
+      RECCOOM_POSTGRES_HOST: reccoom_postgres
+    healthcheck:
+      test: ["CMD-SHELL", "test -f /run/consumer-alive && [ $$(( $$(date +%s) - $$(date +%s -r /run/consumer-alive) )) -lt 60 ]"]
+      timeout: 5s
+    image: maevsi/reccoom:dev
+    secrets:
+      - source: postgres_db
+        target: /run/environment-variables/POSTGRES_DATABASE
+      - source: postgres_db
+        target: /run/environment-variables/RECCOOM_POSTGRES_DATABASE
+      - source: postgres_role_service_reccoom_username
+        target: /run/environment-variables/POSTGRES_USER
+      - source: postgres_user
+        target: /run/environment-variables/RECCOOM_POSTGRES_USER
+      - source: postgres_role_service_reccoom_password
+        target: /run/environment-variables/POSTGRES_PASSWORD
+      - source: postgres_password
+        target: /run/environment-variables/RECCOOM_POSTGRES_PASSWORD
+    volumes:
+      - ../../../reccoom/:/srv/app/ #DARGSTACK-REMOVE
+  reccoom_migration:
+    # You cannot access the recommender's database migration service directly.
+    command: ["uv", "run", "alembic", "upgrade", "head"]
+    deploy:
+      restart_policy:
+        condition: on-failure
+    environment:
+      RECCOOM_POSTGRES_HOST: reccoom_postgres
+    image: maevsi/reccoom:dev
+    secrets:
+      - source: postgres_db
+        target: /run/environment-variables/RECCOOM_POSTGRES_DATABASE
+      - source: postgres_user
+        target: /run/environment-variables/RECCOOM_POSTGRES_USER
+      - source: postgres_password
+        target: /run/environment-variables/RECCOOM_POSTGRES_PASSWORD
+    volumes:
+      - ../../../reccoom/:/srv/app/ #DARGSTACK-REMOVE
   reccoom_postgres:
     # You can access reccoom's database via `adminer`.
     environment:

src/production/configurations/debezium-postgres-connector/entrypoint.sh

@@ -1,4 +1,5 @@
-#!/bin/bash
+#!/bin/sh
+set -eu
 
 CONNECTOR_NAME="postgres-connector"
 DEBEZIUM_URL="http://debezium:8083/connectors"
@@ -12,36 +13,22 @@ while ! curl --output /dev/null --silent --show-error "$DEBEZIUM_URL"; do
     sleep 5
 done
 
-echo "Debezium is ready. Registering PostgreSQL connector..."
+echo "Debezium is ready. Ensuring PostgreSQL connector is up to date..."
 
-# Check if the connector already exists
-HTTP_STATUS=$(curl --output /dev/null --silent --show-error -w "%{http_code}" "$DEBEZIUM_URL/$CONNECTOR_NAME")
+# PUT is idempotent: creates if missing, updates if config changed,
+# no-op if unchanged. Preserves WAL offsets across config updates.
+curl --fail --output /dev/null --silent --show-error \
+    -X PUT "$DEBEZIUM_URL/$CONNECTOR_NAME/config" \
+    -H "Content-Type: application/json" \
+    -d '{
+    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+    "database.dbname": "'"$POSTGRES_DB"'",
+    "database.hostname": "postgres",
+    "database.password": "'"$POSTGRES_PASSWORD"'",
+    "database.user": "'"$POSTGRES_USER"'",
+    "plugin.name": "pgoutput",
+    "table.include.list": "vibetype.event,vibetype.upload,vibetype_private.notification",
+    "topic.prefix" : "vibetype"
+}'
 
-if [ "$HTTP_STATUS" -eq 200 ]; then
-    echo "Connector '$CONNECTOR_NAME' already exists. Skipping registration."
-elif [ "$HTTP_STATUS" -eq 404 ]; then
-    echo "Connector '$CONNECTOR_NAME' does not exist. Registering..."
-
-    # Register the Debezium PostgreSQL connector
-    curl --output /dev/null --silent --show-error \
-        -X POST $DEBEZIUM_URL \
-        -H "Content-Type: application/json" \
-        -d '{
-        "name": "'"$CONNECTOR_NAME"'",
-        "config": {
-            "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
-            "database.dbname": "'"$POSTGRES_DB"'",
-            "database.hostname": "postgres",
-            "database.password": "'"$POSTGRES_PASSWORD"'",
-            "database.user": "'"$POSTGRES_USER"'",
-            "plugin.name": "pgoutput",
-            "table.include.list": "vibetype.upload,vibetype_private.notification",
-            "topic.prefix" : "vibetype"
-        }
-    }'
-
-    echo "PostgreSQL connector '$CONNECTOR_NAME' registered."
-else
-    echo "Failed to query Debezium API. HTTP status: $HTTP_STATUS"
-    exit 1
-fi
+echo "PostgreSQL connector '$CONNECTOR_NAME' is up to date."