refactor: rename email-server-worker to email-worker

Rename directory from email-server-worker to email-worker for clarity and brevity. Update all references in CLAUDE.md documentation.

email-worker/src/app.ts

@@ -0,0 +1,34 @@
+import express from 'express';
+import createError from 'http-errors';
+
+import { errorRouter } from './routes/errorRouter';
+import { finalErrorRouter } from './routes/finalErrorRouter';
+import { metricsRouter } from './routes/metricsRouter';
+import { pingRouter } from './routes/pingRouter';
+import { healthcheckRouter } from './routes/healthcheckRouter';
+
+import { SupportedRoutes } from './types/enums/SupportedRoutes';
+
+const app = express();
+
+// u slučaju kada se server vrti iza proxy-a
+// ovaj flag će natjerati Express da informacije poput
+// IP adrese klijenta, protokola uzima iz X-Forward-*
+// HTTP header polja, koja postavlja proxy
+app.set('trust proxy', true);
+
+// prometheus sa ove rute dohvaća zadnje važeću statistiku
+app.use(SupportedRoutes.metricsPath, metricsRouter);
+app.use(SupportedRoutes.ping, pingRouter);
+app.use(SupportedRoutes.healthcheck, healthcheckRouter);
+
+// default handler
+app.use((req, res, next) => next(createError(404)));
+
+// error handler za sve predviđene greške
+app.use(errorRouter);
+
+// error router za nepredviđene greške
+app.use(finalErrorRouter);
+
+export default app;

email-worker/src/entry.ts

@@ -0,0 +1,122 @@
+import app from './app';
+import http from 'http';
+import stoppable  from 'stoppable';
+
+import { createLogger } from './lib/logger';
+import { disposeSyncWorker, startSyncWorker } from './workRunner';
+const logger = createLogger("server:server");
+
+/**
+* Normalize a port into a number, string, or false.
+*/
+const normalizePort = (val:string):string|number|boolean => {
+  const port = parseInt(val, 10);
+
+  if (isNaN(port)) {
+    // named pipe
+    return val;
+  }
+
+  if (port >= 0) {
+    // port number
+    return port;
+  }
+
+  return false;
+};
+
+/**
+* Event listener for HTTP server "error" event.
+*/
+const onError = (error:any):void => {
+  if (error.syscall !== 'listen') {
+    throw error;
+  }
+
+  const bind = typeof port === 'string'
+    ? 'Pipe ' + port
+    : 'Port ' + port;
+
+  // handle specific listen errors with friendly messages
+  switch (error.code) {
+    case 'EACCES':
+      console.error(bind + ' requires elevated privileges');
+      process.exit(1);
+      break;
+    case 'EADDRINUSE':
+      console.error(bind + ' is already in use');
+      process.exit(1);
+      break;
+    default:
+      throw error;
+  }
+};
+
+/**
+* Event listener for HTTP server "listening" event.
+*/
+const onListening = ():void => {
+  const addr = server.address();
+  const bind = typeof addr === 'string'
+    ? 'pipe ' + addr
+    : 'port ' + addr?.port;
+  logger(`⚡️[server]: Server is running at ${bind}`);
+};
+
+/**
+* Get port from environment and store in Express.
+*/
+const port:number|string|boolean = normalizePort(process.env.PORT || '3000');
+
+/**
+* How long should stoppable wait before it starts force-closing connections
+* @description wait max 10 seconds - needs to be shorter than `healthcheck.timeout` (=15sec)
+*/
+const FORCE_STOP_TIMEOUT = 10000;
+
+/**
+* Create HTTP server.
+*/
+const server = stoppable( http.createServer(app), FORCE_STOP_TIMEOUT );
+
+// Listen on provided port, on all network interfaces.
+server.listen(port);
+server.on('error', onError);
+server.on('listening', onListening);
+
+/**
+ * Starting sync worker process
+ */
+startSyncWorker();
+
+// quit on ctrl-c when running docker in terminal
+// (signal neće biti registriran ako je server pokrenuti via `npm` ili `nodemon` - mora biti pokrenuti izravno via Node)
+process.on('SIGINT', () => {
+    logger('Got SIGINT (aka ctrl-c in docker). Graceful shutdown ', new Date().toISOString());
+    shutdown();
+});
+
+// quit properly on docker stop
+// (signal neće biti registriran ako je server pokrenuti via `npm` ili `nodemon` - mora biti pokrenuti izravno via Node)
+process.on('SIGTERM', () => {
+    logger('Got SIGTERM (docker container stop). Graceful shutdown ', new Date().toISOString());
+    shutdown();
+});
+
+// shut down server
+const shutdown = async () => {
+  
+  await disposeSyncWorker();
+
+  // NOTE: server.close is for express based apps
+  // If using hapi, use `server.stop`
+  server.close((err) => {
+    if (err) {
+      console.error(err);
+      process.exitCode = 1;
+    } else {
+      logger('Exiting server process...');
+    }
+    process.exit();
+  });
+};

email-worker/src/exampleWorker.ts

@@ -0,0 +1,33 @@
+/**
+ * Example worker implementation
+ *
+ * This is a placeholder worker that demonstrates the worker pattern.
+ * Replace this with your actual worker implementation.
+ *
+ * The worker is called periodically by workRunner.ts based on PULL_INTERVAL.
+ *
+ * @throws Error to increment failedRequestCounter in Prometheus
+ * @returns Promise that resolves when work is complete (increments successfulRequestCounter)
+ */
+export const doWork = async () => {
+    const startTime = Date.now();
+
+    // TODO: Implement your periodic worker logic here
+    // Examples:
+    // - Fetch data from external API
+    // - Process queued tasks from database
+    // - Send scheduled emails
+    // - Clean up expired records
+    // - Sync data between systems
+
+    const workDuration = Date.now() - startTime;
+
+    // Log success (only in non-test environments)
+    if (process.env.ENV !== "jest") {
+        const logMessage = `Example worker completed in ${workDuration}ms`;
+        console.log(logMessage);
+    }
+
+    // Note: Throw errors to mark work as failed:
+    // throw new Error("Something went wrong");
+};

email-worker/src/healthcheck.ts

@@ -0,0 +1,29 @@
+import { createLogger } from "./lib/logger";
+
+import http, { IncomingMessage } from "http";
+const logger = createLogger("server:healthcheck");
+
+const options = {
+    host: "localhost",
+    port: "3000",
+    timeout: 2000,
+    path: '/healthcheck/'
+};
+
+const request = http.request(options, (res:IncomingMessage) => {
+
+    logger(`Healthcheck: STATUS ${res.statusCode}`);
+
+    if (res.statusCode == 200) {
+        process.exit(0);
+    } else {
+        process.exit(1);
+    }
+});
+
+request.on("error", function (err:any) {
+    logger("Healthcheck: ERROR");
+    process.exit(1);
+});
+
+request.end();

email-worker/src/lib/initTools.ts

@@ -0,0 +1,8 @@
+
+/**
+ * Za neinicijaliziranu env varijablu vraća default vrijednost
+ * @param value vrijednost env varijable
+ * @param defaultValue default vrijednost
+ * @returns 
+ */
+export const coalesce = (value:string|undefined, defaultValue:string):string => value===undefined ? defaultValue : (value==="" ? defaultValue : value);

email-worker/src/lib/logger.ts

@@ -0,0 +1,21 @@
+import debug from 'debug';
+
+/**
+ * Logs to console / stdout
+ * @param namespace 
+ * @returns instance of Debug
+ */
+export const createLogger = (namespace:string):debug.Debugger => {
+    const dbg = debug(namespace);
+
+    const rx = /nodemon/gi;
+    
+    if(rx.test(process.env?.npm_lifecycle_script ?? "")) {
+        // When started via nodemon:
+        // forcing the use of console insted of stdout
+        // -> nodemon doesn't work with stdout
+        dbg.log = console.log.bind(console);
+    }
+
+    return(dbg);
+};

email-worker/src/lib/metricsCounters.ts

@@ -0,0 +1,50 @@
+import { Counter, Histogram, register } from 'prom-client';
+import { coalesce } from './initTools';
+
+/** Histogram Buckets */
+const PROMETHEUS_HISTOGRAM_BUCKETS = coalesce(process.env.PROMETHEUS_HISTOGRAM_BUCKETS, "0.1, 0.5, 1, 5, 10");
+
+/** Labela kojom želimo da bude označena metrika prikupljena na ovom web servisu */
+const PROMETHEUS_APP_LABEL = coalesce(process.env.PROMETHEUS_APP_LABEL, 'email-worker');
+
+// na "app" labele ćemo razdvajanje rezultata u Grafani
+register.setDefaultLabels({ app: PROMETHEUS_APP_LABEL });
+
+/**
+ * Broji koliko je ukupno zahtjeva zaprimljeno za obradu
+ */
+ export const totalRequestCounter = new Counter({
+    name: "request_operations_total",
+    help: "ukupan broj zaprimljenih zahtjeva",
+    /** countere razdvajamo po vrsti zahtjeva */
+    labelNames: ['path'],
+});
+
+/**
+ * Broji zahtjeve koji su uspješno obrađeni
+ */
+export const successfulRequestCounter = new Counter({
+    name: "request_operations_ok",
+    help: "broj zahtjeva koji su uspješno obrađeni",
+    /** countere razdvajamo po vrsti zahtjeva */
+    labelNames: ['path'],
+});
+
+/**
+ * Broji zahtjeve kod čije obrade je došlo do greške
+ */
+export const failedRequestCounter = new Counter({
+    name: "request_operations_failed",
+    help: "broj zahtjeva kod čije obrade je došlo do greške",
+    /** countere razdvajamo po vrsti zahtjeva i rezultatu izvođenja */
+    labelNames: ["path", "status"],
+});
+
+/** Histogram mjeri koliko traje obrada pristiglog zahtjeva */
+export const requestDurationHistogram = new Histogram({
+    name: "request_duration_seconds",
+    help: "Trajanje request-a u sekundama",
+    /** countere razdvajamo po vrsti zahtjeva i rezultatu izvođenja */
+    labelNames: ["path", "status"],
+    buckets: PROMETHEUS_HISTOGRAM_BUCKETS?.split(',').map((el) => parseFloat(el))
+});

email-worker/src/lib/serializeError.ts

@@ -0,0 +1,19 @@
+
+
+/**
+ * This function serializes an error object into a string that can be logged
+ * @param ex error object
+ * @returns string
+ * @description SQL Server may generate more than one error for one request so you can access preceding errors with `err.precedingErrors`, while the `ex` itself is a generic error without any useful information
+ */
+export const serializeError = (ex:Error | Error & { precedingErrors?:Error[] }):string => {
+    const { name, message, stack, precedingErrors } = (ex as Error & { precedingErrors?:Error[] });
+
+    // SQL Server may generate more than one error for one request so you can access preceding errors with `ex.precedingErrors`,
+    // while the `ex` itself is a generic error without any useful information
+    if(precedingErrors) {
+        return(serializeError(precedingErrors[0]));
+    }
+
+    return `${name}:${message}`;
+}

email-worker/src/routes/errorRouter.ts

@@ -0,0 +1,81 @@
+import { ErrorRequestHandler, Request, Response } from "express";
+import createHttpError, { HttpError } from "http-errors";
+import { createLogger } from '../lib/logger';
+import { NgitLocals } from "../types/NgitLocals";
+import { failedRequestCounter } from "../lib/metricsCounters";
+import { SupportedRoutes } from "../types/enums/SupportedRoutes";
+
+const consoleLog = createLogger("server:server");
+
+/**
+ * Router koji se zadnji poziva, a koji sastavlja odgovor u slučaju greške
+ * @param err 
+ * @param req 
+ * @param res 
+ * @param next 
+ */
+export const errorRouter:ErrorRequestHandler = async (err:HttpError, req, res, next) => {
+
+    const requestPath = req.path as SupportedRoutes;
+
+    // kako je ovaj error handler dosta složen, moguće je da negdje baci grešku
+    // > zato je zamotan u try-catch
+    // > na taj način osiguravam da neće srušiti cijeli proces
+    try {
+
+        let { name:errorLogName, message:errorLogText } = err;
+        let responseBody:string = "";
+
+        switch(err.status) {
+            case 400:
+                responseBody = 'bad request';
+                break;
+            case 401:
+                responseBody = 'unauthorized';
+                break;
+            case 403:
+                responseBody = 'forbidden';
+                break;
+            case 404:
+                consoleLog(`page not found ${req.method} ${requestPath}`)
+                responseBody = 'page not found';
+                errorLogText = `page ${requestPath} not found`;
+                break;
+            case 500:
+                responseBody = "internal server error";
+                errorLogText = err.message;
+                break;
+            default:
+                responseBody = err.name;
+                errorLogText = `err.status=${err.status};err.name=${err.name};err.message=${err.message}`;
+        }
+
+        consoleLog(`${errorLogName}:${errorLogText}`);
+        
+        // `headersSent` će biti TRUE ako je router kod kojeg se dogodila greška već poslao header-e
+        // > ako ih probam ponovo postaviti, to će baciti grešku ... a to ovdje mogu izbjeći
+        if(!res.headersSent) {
+            res.status(err.status);
+            res.setHeader('Content-Type', "text/html");
+            res.end(responseBody);
+        } else {
+            // AKO nije pozvan `end` - pozovi ga i završi obradu zahtjeva
+            // ... u suprotnom će konekcija ostati otvorena do timeout-a
+            if(!res.writableEnded) {
+                res.end();
+            }
+        }
+
+    } catch(ex:any) {
+        // ovu grešku će obraditi `finalErrorRouter`
+        next(createHttpError(500, ex));
+    }
+
+    // ne mogu dopustiti da prometheus client sruši server
+    try  {
+        failedRequestCounter.inc({ path: requestPath, status: err.status });
+        (res.locals as NgitLocals).stopPrometheusTimer({ path: req.path, status: err.status });
+    } catch(ex:any) {
+        console.error(ex);
+    }
+};

email-worker/src/routes/finalErrorRouter.ts

@@ -0,0 +1,34 @@
+import { ErrorRequestHandler, Request, Response } from "express";
+import { HttpError } from "http-errors";
+import { createLogger } from '../lib/logger';
+import { NgitLocals } from "../types/NgitLocals";
+
+const consoleLog = createLogger("server:server");
+
+/**
+ * Router koji se izvršava u slučaju grube greške koja nije obrađena nigdje prije
+ * @param err error objekt
+ * @param req express request
+ * @param res express response
+ * @param next 
+ */
+export const finalErrorRouter:ErrorRequestHandler = async (err:HttpError, req, res, next) => {
+
+    const errorLogText:string = JSON.stringify({ message:err.message, name:err.name, stack:err.stack });
+
+    consoleLog(`Server Error ${err.status}\n${errorLogText}`);          
+
+    // `headersSent` će biti TRUE ako je router kod kojeg se dogodila greška već poslao header-e
+    // > ako ih probam ponovo postaviti, to će baciti grešku i u ovom slučaju SRUŠITI SERVER - to ne smijemo dopustiti
+    if(!res.headersSent) {
+        res.status(err.status);
+        res.setHeader('Content-Type', "text/html");
+        res.end(`unhandled server error`);
+    } else {
+        // AKO nije pozvan `end` - pozovi ga i završi obradu zahtjeva
+        // ... u suprotnom će konekcija ostati otvorena do timeout-a
+        if(!res.writableEnded) {
+            res.end();
+        }
+    }
+};

email-worker/src/routes/healthcheckRouter.ts

@@ -0,0 +1,35 @@
+import { RequestHandler, Router } from "express";
+import { workerRunnerInfo } from "../workRunner";
+import { coalesce } from "../lib/initTools";
+
+const PULL_INTERVAL = parseInt(coalesce(process.env.PULL_INTERVAL, "10000"));
+
+/** Maximum time between two worker jobs */
+const MAX_WORKER_LATENCY = PULL_INTERVAL * 2.5;
+
+/**
+ * Router koji se izvršava u slučaju grube greške koja nije obrađena nigdje prije
+ * @param req express request
+ * @param res express response
+ * @param next 
+ */
+export const healthcheckRouter:RequestHandler = async (req, res, next) => {
+    const workerLatency = Date.now() - workerRunnerInfo.lastWorkTime;
+
+    if(workerLatency > MAX_WORKER_LATENCY) {
+        const msg = `No work done in ${workerLatency}ms. Last worker status = "${workerRunnerInfo.status}"`;
+
+        console.warn(msg)
+
+        res.status(500);
+        res.setHeader('Content-Type', 'text/plain');
+        res.end(msg);
+    } else {
+        res.status(200);
+        res.setHeader('Content-Type', 'text/plain');
+        res.end('OK');
+    }
+};
+
+export const pingRouter = Router();
+pingRouter.get('/', healthcheckRouter);

email-worker/src/routes/metricsRouter.ts

@@ -0,0 +1,19 @@
+import { Router, NextFunction, Request, Response } from "express";
+import createError from 'http-errors';
+import { register } from 'prom-client';
+import { createLogger } from '../lib/logger';
+
+const logger = createLogger("server:metrics");
+
+export const metricsRouter = Router();
+
+metricsRouter.get('/', async (req:Request, res:Response, next:NextFunction) => {
+    // ne mogu dopustiti da prometheus client sruši server
+    try {
+        logger(`⚡️[server]: GET /metrics`);
+        res.set('Content-Type', register.contentType);
+        res.end(await register.metrics());
+    } catch(ex:any) {
+        next(createError(500, (ex as Error).message));
+    }
+});

email-worker/src/routes/pingRouter.ts

@@ -0,0 +1,16 @@
+import { RequestHandler, Router } from "express";
+
+/**
+ * Router koji se izvršava u slučaju grube greške koja nije obrađena nigdje prije
+ * @param req express request
+ * @param res express response
+ * @param next 
+ */
+export const pingRequestHandler:RequestHandler = async (req, res, next) => {
+    res.status(200);
+    res.setHeader('Content-Type', 'text/plain');
+    res.end('PONG');
+};
+
+export const pingRouter = Router();
+pingRouter.get('/', pingRequestHandler);

email-worker/src/types/NgitLocals.ts

@@ -0,0 +1,7 @@
+import { LabelValues } from "prom-client";
+
+/** data assignet to `express.response.locals` */
+export type NgitLocals = {
+    /** Prometheus client timer */
+    stopPrometheusTimer: (labels?: LabelValues<"path"|"status">) => number,
+};

email-worker/src/types/enums/SupportedRoutes.ts

@@ -0,0 +1,5 @@
+export enum SupportedRoutes {
+    metricsPath='/metrics',
+    ping='/ping',
+    healthcheck='/healthcheck',
+}

email-worker/src/types/environment.d.ts

@@ -0,0 +1,29 @@
+
+declare global {
+    namespace NodeJS {
+        interface ProcessEnv  {
+            /**
+             * (optional) environment u kojem se proces vrti
+             * @default undefined
+             * */
+            ENV?:"dev"|"jest"
+            /**
+             * (optional) App label to be used in Prometheus (Grafana)
+             * @default "email-worker"
+             * */
+            PROMETHEUS_APP_LABEL?: string
+            /**
+             * (optional) Prometheus histogram bucket sizes (grafana)
+             * @default "0.1, 0.5, 1, 5, 10"
+             * */
+            PROMETHEUS_HISTOGRAM_BUCKETS?: string
+            /**
+             * (required) Pull interval in milliseconds - how often should worker cycle run
+             * @default "10000"
+             * */
+            PULL_INTERVAL:string
+        }
+    }
+}
+
+export {}

email-worker/src/workRunner.ts

@@ -0,0 +1,165 @@
+import { failedRequestCounter, requestDurationHistogram, successfulRequestCounter, totalRequestCounter } from "./lib/metricsCounters";
+import { coalesce } from "./lib/initTools";
+import { createLogger } from "./lib/logger";
+import { serializeError } from "./lib/serializeError";
+import { doWork } from "./exampleWorker";
+
+/** time between two pull operations */
+const PULL_INTERVAL = parseInt(coalesce(process.env.PULL_INTERVAL, "10000"));
+const consoleLog = createLogger("server:server");
+
+/** Writes entry to log */
+const logWrite = (logTitle:string, logMessage:string) => {
+    consoleLog(`${logTitle}: ${logMessage}}`);
+}
+
+/** Writes error to log */
+const logError = (ex: any) => 
+    logWrite(serializeError(ex), "error");
+
+/**
+ * zastavica za zaustavljanje sinhronizacije
+ */
+let disposed:boolean = false;
+/** is worker started - prevents multiple starts */
+let workerStarted:boolean = false;
+/** Promise which is resolved once the pending work in progress is completed */
+let pendingWork:Promise<void>|undefined;
+/** Worker re-run timeout */
+let pendingTimeout:NodeJS.Timeout|undefined;
+
+/** Enumeracija pojedinih statusa obrade jednog work-a */
+export enum WorkerRunnerStatus {
+    init="init",
+    disposed="disposed",
+    beginWork="beginWork",
+    updatedStats1="updatedStats1",
+    updatedStats2="updatedStats2",
+    stoppedStatTimer="stoppedStatTimer",
+    workDone="workDone",
+    newIntervalScheduled="newIntervalScheduled",
+    currentWorkResolved="currentWorkResolved",
+}
+
+/** Info o statusu workera */
+export type WorkerRunnerInfo = {
+    /** zadnje izvršena readnja */
+    status: WorkerRunnerStatus,
+    /** vrijeme kada je worker zadnji puta pokrenut */
+    lastWorkTime: number,
+}
+
+/** Info o statusu workera, koji koristi healthcheck kako bi vidio da li stvar funkcionira */
+export const workerRunnerInfo:WorkerRunnerInfo = {
+    status: WorkerRunnerStatus.init,
+    lastWorkTime: Date.now()
+}
+
+export const workRunner = async () => {
+
+    pendingTimeout = undefined;
+    workerRunnerInfo.lastWorkTime = Date.now();
+    workerRunnerInfo.status = WorkerRunnerStatus.beginWork;
+
+    // AKO je modul zaustavljen
+    // -> nemoj se pokrenuti
+    if(disposed) {
+        workerRunnerInfo.status = WorkerRunnerStatus.disposed;
+        return;
+    }
+
+    // kreiram Promise koji omogućuje da dispose zna
+    // pričekati da worker završi sa poslom (ako je u tom trenutku aktivan)
+    pendingWork = new Promise(async (resolve) => {
+
+        try {
+            totalRequestCounter.inc();
+
+            const stopPrometheusTimer = requestDurationHistogram.startTimer();
+
+            workerRunnerInfo.status = WorkerRunnerStatus.updatedStats1;
+            
+            try {
+                // ne dopuštam da stvar sruši worker
+                await doWork();
+
+                workerRunnerInfo.status = WorkerRunnerStatus.workDone;
+
+                // ažuriram statistiku
+                successfulRequestCounter.inc();
+
+                workerRunnerInfo.status = WorkerRunnerStatus.updatedStats2;
+            } catch(ex:any) {
+
+                // ažuriram statistiku
+                failedRequestCounter.inc();
+                logError(ex);
+            }
+
+            stopPrometheusTimer();
+
+            workerRunnerInfo.status = WorkerRunnerStatus.stoppedStatTimer;
+        } catch(ex:any) {
+            logError(ex);
+        }
+    
+        // nemoj pokrenuti timer ako je worker u međuvremenu disposed
+        if(!disposed) {
+            // pull again after timeout
+            pendingTimeout = setTimeout(workRunner, PULL_INTERVAL);
+            workerRunnerInfo.status = WorkerRunnerStatus.newIntervalScheduled;
+        } else {
+            logWrite("Info", "... exiting worker loop");
+        }
+
+        resolve();
+
+        workerRunnerInfo.status = WorkerRunnerStatus.currentWorkResolved;
+
+        pendingWork = undefined;
+    });
+    
+    // this is an async function which must return a promise
+    // > so return the promise which will be resolved once the work is done
+    return(pendingWork);
+};
+
+/**
+ * Starts the worker
+ */
+export const startSyncWorker = () => {
+    if(!workerStarted && !disposed) {
+        workerStarted = true;
+        workRunner();
+        logWrite("Info", "Worker Started");
+    };
+}
+
+/**
+ * Stops and disposes the worker
+ */
+export const disposeSyncWorker = async () => {
+    logWrite("Info", "Disposing worker ...");
+
+    disposed = true;
+
+    // preventing timer from trigger another work cycle
+    if(pendingTimeout) {
+        clearTimeout(pendingTimeout);
+    }
+
+    // IF no work is currently in progress
+    // > return a resolved promise
+    if(!pendingWork) {
+        return(Promise.resolve());
+    }
+
+    await pendingWork;
+
+    logWrite("Info", "Worker disposed!");
+}
+
+/** Ovo se koristi samo za Unit Testing */
+export const reset_dispose = () => {
+    disposed = false;
+}