🎉 initial commit

devices-worker/Dockerfile

@@ -0,0 +1,13 @@
+FROM nickblah/lua:5.4-luarocks-alpine
+
+RUN apk add --no-cache gcc musl-dev make libpq git curl procps linux-headers pkgconfig
+
+RUN luarocks install lua-cjson
+RUN luarocks install luasocket
+RUN luarocks install pgmoon
+RUN luarocks install redis-lua
+
+WORKDIR /app
+COPY . /app
+
+CMD ["lua", "worker.lua"]

devices-worker/Makefile

@@ -0,0 +1,4 @@
+.PHONY: run
+
+run:
+	lua worker.lua

devices-worker/db.lua

@@ -0,0 +1,33 @@
+-- PostgreSQL using pgmoon (worker: one connection per process is fine)
+local pgmoon = require("pgmoon")
+
+local DB_HOST = os.getenv("DB_HOST") or "localhost"
+local DB_PORT = tonumber(os.getenv("DB_PORT")) or 5432
+local DB_NAME = os.getenv("DB_NAME") or "handheld_devices"
+local DB_USER = os.getenv("DB_USER") or "devices_user"
+local DB_PASSWORD = os.getenv("DB_PASSWORD") or "devices_password"
+local DB_CONNECT_TIMEOUT_MS = tonumber(os.getenv("DB_CONNECT_TIMEOUT_MS")) or 5000
+
+local config = {
+    host = DB_HOST,
+    port = tostring(DB_PORT),
+    database = DB_NAME,
+    user = DB_USER,
+    password = DB_PASSWORD,
+    socket_type = "luasocket",
+}
+
+local function get_connection()
+    local pg = pgmoon.new(config)
+    pg:settimeout(DB_CONNECT_TIMEOUT_MS)
+    local ok, err = pg:connect()
+    if not ok then
+        return nil, err
+    end
+    return pg
+end
+
+return {
+    get_connection = get_connection,
+    config = config,
+}

devices-worker/handlers/device_handler.lua

@@ -0,0 +1,44 @@
+local db = require("db")
+local log = require("log")
+
+local DeviceHandler = {}
+
+function DeviceHandler.handle(event)
+    local conn, err = db.get_connection()
+    if not conn then
+        error("Database connection failed: " .. tostring(err))
+    end
+
+    local ok, handler_err = pcall(function()
+        conn:query([[
+            CREATE TABLE IF NOT EXISTS device_events (
+                id SERIAL PRIMARY KEY,
+                device_id INTEGER,
+                device_name VARCHAR(255),
+                event_type VARCHAR(100),
+                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        ]])
+
+        conn:query(
+            "INSERT INTO device_events (device_id, device_name, event_type) VALUES ($1, $2, $3)",
+            tonumber(event.device_id) or 0,
+            event.device_name or "",
+            event.event_type
+        )
+    end)
+
+    conn:disconnect()
+
+    if not ok then
+        error(handler_err)
+    end
+
+    log.info("Device event logged", {
+        component = "device_handler",
+        device_name = event.device_name,
+        request_id = event.request_id,
+    })
+end
+
+return DeviceHandler

devices-worker/handlers/rating_handler.lua

@@ -0,0 +1,45 @@
+local db = require("db")
+local log = require("log")
+
+local RatingHandler = {}
+
+function RatingHandler.handle(event)
+    local conn, err = db.get_connection()
+    if not conn then
+        error("Database connection failed: " .. tostring(err))
+    end
+
+    local ok, handler_err = pcall(function()
+        conn:query([[
+            CREATE TABLE IF NOT EXISTS rating_events (
+                id SERIAL PRIMARY KEY,
+                device_id INTEGER,
+                user_id VARCHAR(255),
+                score INTEGER,
+                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        ]])
+
+        conn:query(
+            "INSERT INTO rating_events (device_id, user_id, score) VALUES ($1, $2, $3)",
+            tonumber(event.device_id) or 0,
+            event.user_id or "",
+            tonumber(event.score) or 0
+        )
+    end)
+
+    conn:disconnect()
+
+    if not ok then
+        error(handler_err)
+    end
+
+    log.info("Rating event logged", {
+        component = "rating_handler",
+        device_id = event.device_id,
+        score = event.score,
+        request_id = event.request_id,
+    })
+end
+
+return RatingHandler

devices-worker/handlers/review_handler.lua

@@ -0,0 +1,42 @@
+local db = require("db")
+local log = require("log")
+
+local ReviewHandler = {}
+
+function ReviewHandler.handle(event)
+    local conn, err = db.get_connection()
+    if not conn then
+        error("Database connection failed: " .. tostring(err))
+    end
+
+    local ok, handler_err = pcall(function()
+        conn:query([[
+            CREATE TABLE IF NOT EXISTS review_events (
+                id SERIAL PRIMARY KEY,
+                device_id INTEGER,
+                user_id VARCHAR(255),
+                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        ]])
+
+        conn:query(
+            "INSERT INTO review_events (device_id, user_id) VALUES ($1, $2)",
+            tonumber(event.device_id) or 0,
+            event.user_id or ""
+        )
+    end)
+
+    conn:disconnect()
+
+    if not ok then
+        error(handler_err)
+    end
+
+    log.info("Review event logged", {
+        component = "review_handler",
+        device_id = event.device_id,
+        request_id = event.request_id,
+    })
+end
+
+return ReviewHandler

devices-worker/log.lua

@@ -0,0 +1,15 @@
+local cjson = require("cjson")
+
+local function log(level, msg, fields)
+    fields = fields or {}
+    fields.level = level
+    fields.msg = msg
+    fields.time = os.date("!%Y-%m-%dT%H:%M:%SZ")
+    print(cjson.encode(fields))
+end
+
+return {
+    info = function(msg, fields) log("info", msg, fields) end,
+    warn = function(msg, fields) log("warn", msg, fields) end,
+    error = function(msg, fields) log("error", msg, fields) end,
+}

devices-worker/worker.lua

@@ -0,0 +1,111 @@
+local redis = require("redis")
+local cjson = require("cjson")
+local socket = require("socket")
+local log = require("log")
+
+local REDIS_HOST = os.getenv("REDIS_HOST") or "127.0.0.1"
+local REDIS_PORT = tonumber(os.getenv("REDIS_PORT")) or 6379
+local MAX_RETRIES = tonumber(os.getenv("WORKER_MAX_RETRIES")) or 3
+local DLQ_KEY = "devices:events:dlq"
+
+local device_handler = require("handlers.device_handler")
+local rating_handler = require("handlers.rating_handler")
+local review_handler = require("handlers.review_handler")
+
+local handlers = {
+    DevicePublished = device_handler.handle,
+    DeviceDeleted = device_handler.handle,
+    DeviceUpdated = device_handler.handle,
+    RatingPublished = rating_handler.handle,
+    ReviewPublished = review_handler.handle,
+    UserCreated = function(event)
+        log.info("User created", { component = "worker", user_id = event.user_id })
+    end,
+}
+
+local function move_to_dlq(red, event_json, reason)
+    red:lpush(DLQ_KEY, cjson.encode({
+        event = event_json,
+        reason = reason,
+        failed_at = os.time(),
+    }))
+    log.warn("Event moved to DLQ", { reason = reason, dlq_key = DLQ_KEY })
+end
+
+local function process_event(red, event_json)
+    local ok_decode, event = pcall(cjson.decode, event_json)
+    if not ok_decode or not event then
+        return false, "decode_failed"
+    end
+
+    local handler = handlers[event.event_type]
+    if not handler then
+        return true, nil -- Acknowledge unknown event types
+    end
+
+    local attempt = 0
+    local last_err
+
+    while attempt < MAX_RETRIES do
+        attempt = attempt + 1
+        local success, handler_err = pcall(handler, event)
+        if success then
+            return true, nil
+        end
+        last_err = tostring(handler_err)
+        if attempt < MAX_RETRIES then
+            local delay = math.min(2 ^ attempt * 0.5, 10)
+            log.warn("Handler failed, retrying", {
+                component = "worker",
+                attempt = attempt,
+                max_retries = MAX_RETRIES,
+                delay = delay,
+                err = last_err,
+            })
+            socket.sleep(delay)
+        end
+    end
+
+    return false, last_err
+end
+
+local function run_worker()
+    while true do
+        local ok, err = pcall(function()
+            log.info("Connecting to Redis", { component = "worker" })
+
+            local red = redis.connect(REDIS_HOST, REDIS_PORT)
+            log.info("Connected to Redis, waiting for events", { component = "worker" })
+
+            while true do
+                local event_json = red:brpoplpush("devices:events:queue", "devices:events:processing", 0)
+
+                if event_json then
+                    local success, reason = process_event(red, event_json)
+
+                    if success then
+                        red:lrem("devices:events:processing", 1, event_json)
+                        log.info("Event processed", { component = "worker" })
+                    else
+                        -- Move to DLQ and remove from processing
+                        move_to_dlq(red, event_json, reason)
+                        red:lrem("devices:events:processing", 1, event_json)
+                    end
+                end
+            end
+        end)
+
+        if not ok then
+            log.error("Worker error", { component = "worker", err = tostring(err) })
+            log.info("Reconnecting in 5 seconds", { component = "worker" })
+            socket.sleep(5)
+        end
+    end
+end
+
+log.info("Handheld Devices Worker starting", {
+    component = "worker",
+    redis = REDIS_HOST .. ":" .. REDIS_PORT,
+})
+
+run_worker()