root/sase.tr
1
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: add OpenTelemetry observability, Faro frontend monitoring, remove legacy Next.js app

Browse Source 
- Add OpenTelemetry SDK with tracing, metrics, and OTLP export for API and worker
- Integrate Grafana Faro for frontend real-user monitoring
- Instrument health checks, database, Bull queues, and HTTP exception filter
- Add Grafana dashboard JSON for service overview
- Remove deprecated apps/web-nj (Next.js) — fully replaced by Vite+React frontend
- Update nginx config with OTEL collector proxy
- Minor UI fixes in schema viewer, category components, and subscription flow

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sase Dev
2026-02-16 17:20:49 +00:00
parent 87d8eea298
commit dcbeb83ccc
113 changed files with 2745 additions and 7351 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
apps/api/package.json
View File

@@ -20,6 +20,7 @@
},
"dependencies": {
"@aws-sdk/client-s3": "^3.700.0",
"@kubiks/otel-drizzle": "^2.1.0",
"@nestjs/common": "^10.4.0",
"@nestjs/config": "^3.3.0",
"@nestjs/core": "^10.4.0",
@@ -27,10 +28,26 @@
"@nestjs/schedule": "^4.1.0",
"@nestjs/swagger": "^8.1.0",
"@nestjs/throttler": "^6.3.0",
"@opentelemetry/api": "^1.9.0",
"@opentelemetry/exporter-logs-otlp-http": "^0.212.0",
"@opentelemetry/exporter-metrics-otlp-http": "^0.212.0",
"@opentelemetry/exporter-trace-otlp-http": "^0.212.0",
"@opentelemetry/instrumentation": "^0.212.0",
"@opentelemetry/instrumentation-express": "^0.59.0",
"@opentelemetry/instrumentation-http": "^0.212.0",
"@opentelemetry/instrumentation-ioredis": "^0.59.0",
"@opentelemetry/instrumentation-nestjs-core": "^0.57.0",
"@opentelemetry/resources": "^2.5.1",
"@opentelemetry/sdk-logs": "^0.212.0",
"@opentelemetry/sdk-metrics": "^2.5.1",
"@opentelemetry/sdk-node": "^0.212.0",
"@opentelemetry/sdk-trace-base": "^2.5.1",
"@opentelemetry/semantic-conventions": "^1.39.0",
"@sase/config": "workspace:*",
"@sase/shared": "workspace:*",
"better-auth": "^1.2.0",
"bullmq": "^5.30.0",
"bullmq-otel": "^1.2.0",
"dotenv": "^16.4.0",
"drizzle-orm": "^0.41.0",
"helmet": "^8.1.0",

14
apps/api/src/common/filters/http-exception.filter.ts
View File

@@ -1,5 +1,6 @@
import { ExceptionFilter, Catch, ArgumentsHost, HttpException, HttpStatus, Logger } from "@nestjs/common";
import { Response } from "express";
import { trace, SpanStatusCode } from "@opentelemetry/api";
@Catch()
export class HttpExceptionFilter implements ExceptionFilter {
@@ -29,6 +30,19 @@ export class HttpExceptionFilter implements ExceptionFilter {
this.logger.error(`Unhandled error: ${exception.message}`, exception.stack);
}
// Record error on active OTel span
const span = trace.getActiveSpan();
if (span) {
span.setAttribute("error.code", code);
span.setAttribute("http.status_code", status);
if (exception instanceof Error) {
span.recordException(exception);
}
if (status >= 500) {
span.setStatus({ code: SpanStatusCode.ERROR, message });
}
}
response.status(status).json({
success: false,
error: {

14
apps/api/src/common/interceptors/logging.interceptor.ts
View File

@@ -1,5 +1,6 @@
import { Injectable, NestInterceptor, ExecutionContext, CallHandler, Logger } from "@nestjs/common";
import { Observable, tap } from "rxjs";
import { trace } from "@opentelemetry/api";
@Injectable()
export class LoggingInterceptor implements NestInterceptor {
@@ -14,7 +15,18 @@ export class LoggingInterceptor implements NestInterceptor {
tap(() => {
const response = context.switchToHttp().getResponse();
const elapsed = Date.now() - now;
this.logger.log(`${method} ${url} ${response.statusCode} ${elapsed}ms`);
const span = trace.getActiveSpan();
if (span) {
span.setAttribute("http.response_time_ms", elapsed);
if (request.user?.id) {
span.setAttribute("user.id", request.user.id);
}
}
const traceId = span?.spanContext().traceId;
const traceTag = traceId ? ` [trace:${traceId}]` : "";
this.logger.log(`${method} ${url} ${response.statusCode} ${elapsed}ms${traceTag}`);
}),
);
}

6
apps/api/src/config/configuration.ts
View File

@@ -40,4 +40,10 @@ export default () => ({
username: process.env.EMEX_USERNAME,
password: process.env.EMEX_PASSWORD,
},
otel: {
enabled: process.env.OTEL_ENABLED === "true",
endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
serviceName: process.env.OTEL_SERVICE_NAME || "sase-api",
sampleRate: parseFloat(process.env.OTEL_TRACE_SAMPLE_RATE || "1.0"),
},
});

13
apps/api/src/database/database.provider.ts
View File

@@ -6,6 +6,7 @@ import * as core from "./schema/core";
import * as pl24 from "./schema/pl24";
import * as emex from "./schema/emex";
import * as relations from "./schema/relations";
import { isOtelEnabled } from "../telemetry";
export const DATABASE = "DATABASE";
@@ -23,10 +24,20 @@ export const DatabaseProvider: Provider = {
connect_timeout: 10,
});
const db = drizzle(client, {
let db: Database = drizzle(client, {
schema: { ...core, ...pl24, ...emex, ...relations },
});
if (isOtelEnabled) {
try {
const { instrumentDrizzle } = require("@kubiks/otel-drizzle");
db = instrumentDrizzle(db);
console.log("[otel] Drizzle ORM instrumented");
} catch (err) {
console.warn("[otel] Drizzle instrumentation unavailable:", (err as Error).message);
}
}
console.log("Database connected");
return db;
},

49
apps/api/src/health.controller.ts
View File

@@ -1,11 +1,54 @@
import { Controller, Get } from "@nestjs/common";
import { Controller, Get, Inject } from "@nestjs/common";
import { Public } from "./common/decorators/public.decorator";
import { DATABASE, Database } from "./database/database.provider";
import { RedisService } from "./redis/redis.service";
import { isOtelEnabled } from "./telemetry";
import { sql } from "drizzle-orm";
function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
return Promise.race([
promise,
new Promise<never>((_, reject) => setTimeout(() => reject(new Error("timeout")), ms)),
]);
}
@Controller("health")
export class HealthController {
constructor(
@Inject(DATABASE) private readonly db: Database,
private readonly redis: RedisService,
) {}
@Get()
@Public()
check() {
return { status: "ok", timestamp: new Date().toISOString() };
async check() {
const checks: Record<string, string> = {};
// Database check (3s timeout)
try {
await withTimeout(this.db.execute(sql`SELECT 1`), 3000);
checks.database = "ok";
} catch {
checks.database = "error";
}
// Redis check (3s timeout)
try {
await withTimeout(this.redis.getClient().ping(), 3000);
checks.redis = "ok";
} catch {
checks.redis = "error";
}
// Telemetry status
checks.telemetry = isOtelEnabled ? "enabled" : "disabled";
const status = checks.database === "ok" && checks.redis === "ok" ? "ok" : "degraded";
return {
status,
timestamp: new Date().toISOString(),
checks,
};
}
}

12
apps/api/src/jobs/bull.config.ts
View File

@@ -1,4 +1,5 @@
import { ConnectionOptions } from "bullmq";
import { isOtelEnabled } from "../telemetry";
export function getBullConnection(): ConnectionOptions {
return {
@@ -8,6 +9,17 @@ export function getBullConnection(): ConnectionOptions {
};
}
export function getBullTelemetry() {
if (!isOtelEnabled) return undefined;
try {
const { BullMQOtel } = require("bullmq-otel");
return new BullMQOtel("sase");
} catch (err) {
console.warn("[otel] BullMQ telemetry unavailable:", (err as Error).message);
return undefined;
}
}
export const QUEUE_NAMES = {
EMEX_SCRAPE: "emex-scrape",
SUBSCRIPTION_EXPIRY: "subscription-expiry",

4
apps/api/src/jobs/queues/emex-scrape.queue.ts
View File

@@ -1,14 +1,16 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, QUEUE_NAMES } from "../bull.config";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
export const EMEX_SCRAPE_QUEUE = "EMEX_SCRAPE_QUEUE";
export const EmexScrapeQueueProvider: Provider = {
provide: EMEX_SCRAPE_QUEUE,
useFactory: () => {
const telemetry = getBullTelemetry();
return new Queue(QUEUE_NAMES.EMEX_SCRAPE, {
connection: getBullConnection(),
...(telemetry ? { telemetry } : {}),
defaultJobOptions: {
attempts: 3,
backoff: {

4
apps/api/src/jobs/queues/query-cleanup.queue.ts
View File

@@ -1,14 +1,16 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, QUEUE_NAMES } from "../bull.config";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
export const QUERY_CLEANUP_QUEUE = "QUERY_CLEANUP_QUEUE";
export const QueryCleanupQueueProvider: Provider = {
provide: QUERY_CLEANUP_QUEUE,
useFactory: () => {
const telemetry = getBullTelemetry();
return new Queue(QUEUE_NAMES.QUERY_CLEANUP, {
connection: getBullConnection(),
...(telemetry ? { telemetry } : {}),
defaultJobOptions: {
attempts: 2,
backoff: {

4
apps/api/src/jobs/queues/subscription-expiry.queue.ts
View File

@@ -1,14 +1,16 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, QUEUE_NAMES } from "../bull.config";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
export const SUBSCRIPTION_EXPIRY_QUEUE = "SUBSCRIPTION_EXPIRY_QUEUE";
export const SubscriptionExpiryQueueProvider: Provider = {
provide: SUBSCRIPTION_EXPIRY_QUEUE,
useFactory: () => {
const telemetry = getBullTelemetry();
return new Queue(QUEUE_NAMES.SUBSCRIPTION_EXPIRY, {
connection: getBullConnection(),
...(telemetry ? { telemetry } : {}),
defaultJobOptions: {
attempts: 3,
backoff: {

4
apps/api/src/main.ts
View File

@@ -1,3 +1,5 @@
import "./telemetry/tracing"; // MUST be first — instruments modules before they load
import { NestFactory } from "@nestjs/core";
import { ConfigService } from "@nestjs/config";
import helmet from "helmet";
@@ -21,7 +23,7 @@ async function bootstrap() {
origin: corsOrigins,
credentials: true,
methods: ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
allowedHeaders: ["Content-Type", "Authorization", "Cookie"],
allowedHeaders: ["Content-Type", "Authorization", "Cookie", "traceparent", "tracestate"],
exposedHeaders: ["set-cookie"],
maxAge: 86400,
});

97
apps/api/src/telemetry/__tests__/telemetry.spec.ts Normal file
View File

@@ -0,0 +1,97 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
describe("Telemetry module", () => {
const originalEnv = process.env;
beforeEach(() => {
process.env = { ...originalEnv };
vi.resetModules();
});
afterEach(() => {
process.env = originalEnv;
});
describe("isOtelEnabled", () => {
it("should be false when OTEL_ENABLED is not set", async () => {
delete process.env.OTEL_ENABLED;
const { isOtelEnabled } = await import("../index");
expect(isOtelEnabled).toBe(false);
});
it("should be false when OTEL_ENABLED is 'false'", async () => {
process.env.OTEL_ENABLED = "false";
const { isOtelEnabled } = await import("../index");
expect(isOtelEnabled).toBe(false);
});
it("should be true when OTEL_ENABLED is 'true'", async () => {
process.env.OTEL_ENABLED = "true";
const { isOtelEnabled } = await import("../index");
expect(isOtelEnabled).toBe(true);
});
});
describe("getTracer", () => {
it("should return a tracer without errors when OTel is disabled", async () => {
process.env.OTEL_ENABLED = "false";
const { getTracer } = await import("../index");
const tracer = getTracer("test");
expect(tracer).toBeDefined();
});
});
describe("getMeter", () => {
it("should return a meter without errors when OTel is disabled", async () => {
process.env.OTEL_ENABLED = "false";
const { getMeter } = await import("../index");
const meter = getMeter("test");
expect(meter).toBeDefined();
});
});
describe("sdk-factory", () => {
it("should not throw when OTel is enabled but endpoint is missing", async () => {
process.env.OTEL_ENABLED = "true";
delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
const { createNodeSDK } = await import("../sdk-factory");
expect(() =>
createNodeSDK({
serviceName: "test",
instrumentations: [],
}),
).not.toThrow();
});
it("should return null when endpoint is not configured", async () => {
delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
const { createNodeSDK } = await import("../sdk-factory");
const sdk = createNodeSDK({
serviceName: "test",
instrumentations: [],
});
expect(sdk).toBeNull();
});
});
describe("metrics", () => {
it("should create metrics without errors", async () => {
process.env.OTEL_ENABLED = "false";
const { getAppMetrics } = await import("../metrics");
const metrics = getAppMetrics();
expect(metrics).toBeDefined();
expect(metrics.vinDecodeCounter).toBeDefined();
expect(metrics.vinDecodeDuration).toBeDefined();
expect(metrics.activeSubscriptions).toBeDefined();
expect(metrics.paymentCounter).toBeDefined();
expect(metrics.jobCounter).toBeDefined();
});
it("should return same instance on multiple calls", async () => {
const { getAppMetrics } = await import("../metrics");
const a = getAppMetrics();
const b = getAppMetrics();
expect(a).toBe(b);
});
});
});

28
apps/api/src/telemetry/index.ts Normal file
View File

@@ -0,0 +1,28 @@
import { trace, metrics, SpanStatusCode, type Span } from "@opentelemetry/api";
export const isOtelEnabled = process.env.OTEL_ENABLED === "true";
export function getTracer(name = "sase-api") {
return trace.getTracer(name);
}
export function getMeter(name = "sase-api") {
return metrics.getMeter(name);
}
export function recordSpanError(error: Error, span?: Span) {
const activeSpan = span || trace.getActiveSpan();
if (activeSpan) {
activeSpan.recordException(error);
activeSpan.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
}
}
export function setSpanAttributes(attributes: Record<string, string | number | boolean>) {
const activeSpan = trace.getActiveSpan();
if (activeSpan) {
activeSpan.setAttributes(attributes);
}
}
export { trace, metrics, SpanStatusCode };

33
apps/api/src/telemetry/metrics.ts Normal file
View File

@@ -0,0 +1,33 @@
import { getMeter } from "./index";
let _metrics: ReturnType<typeof createMetrics> | null = null;
function createMetrics() {
const meter = getMeter("sase-api");
return {
vinDecodeCounter: meter.createCounter("sase.vin_decode.total", {
description: "Total VIN decode attempts",
}),
vinDecodeDuration: meter.createHistogram("sase.vin_decode.duration_ms", {
description: "VIN decode duration in milliseconds",
unit: "ms",
}),
activeSubscriptions: meter.createObservableGauge("sase.subscriptions.active", {
description: "Number of active subscriptions",
}),
paymentCounter: meter.createCounter("sase.payments.total", {
description: "Total payment attempts",
}),
jobCounter: meter.createCounter("sase.jobs.total", {
description: "Total background jobs processed",
}),
};
}
export function getAppMetrics() {
if (!_metrics) {
_metrics = createMetrics();
}
return _metrics;
}

81
apps/api/src/telemetry/sdk-factory.ts Normal file
View File

@@ -0,0 +1,81 @@
import { NodeSDK } from "@opentelemetry/sdk-node";
import { resourceFromAttributes } from "@opentelemetry/resources";
import {
ATTR_SERVICE_NAME,
SEMRESATTRS_DEPLOYMENT_ENVIRONMENT,
} from "@opentelemetry/semantic-conventions";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { BatchSpanProcessor, TraceIdRatioBasedSampler } from "@opentelemetry/sdk-trace-base";
import { BatchLogRecordProcessor } from "@opentelemetry/sdk-logs";
import type { Instrumentation } from "@opentelemetry/instrumentation";
export interface SDKConfig {
serviceName: string;
instrumentations: Instrumentation[];
}
function parseHeaders(headerString: string): Record<string, string> {
if (!headerString) return {};
const headers: Record<string, string> = {};
for (const pair of headerString.split(",")) {
const [key, ...valueParts] = pair.split("=");
if (key && valueParts.length > 0) {
headers[key.trim()] = valueParts.join("=").trim();
}
}
return headers;
}
export function createNodeSDK(config: SDKConfig): NodeSDK | null {
const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
if (!endpoint) {
console.log(`[otel] No OTLP endpoint configured — telemetry will not be exported`);
return null;
}
const headers = parseHeaders(process.env.OTEL_EXPORTER_OTLP_HEADERS || "");
const sampleRate = parseFloat(process.env.OTEL_TRACE_SAMPLE_RATE || "1.0");
const resource = resourceFromAttributes({
[ATTR_SERVICE_NAME]: config.serviceName,
"service.version": process.env.npm_package_version || "0.1.0",
[SEMRESATTRS_DEPLOYMENT_ENVIRONMENT]: process.env.NODE_ENV || "development",
});
const traceExporter = new OTLPTraceExporter({
url: `${endpoint}/v1/traces`,
headers,
});
const metricExporter = new OTLPMetricExporter({
url: `${endpoint}/v1/metrics`,
headers,
});
const logExporter = new OTLPLogExporter({
url: `${endpoint}/v1/logs`,
headers,
});
const sdk = new NodeSDK({
resource,
spanProcessors: [new BatchSpanProcessor(traceExporter)],
metricReader: new PeriodicExportingMetricReader({
exporter: metricExporter,
exportIntervalMillis: 30_000,
}),
logRecordProcessors: [new BatchLogRecordProcessor(logExporter)],
instrumentations: config.instrumentations,
sampler: sampleRate < 1.0 ? new TraceIdRatioBasedSampler(sampleRate) : undefined,
});
sdk.start();
console.log(
`[otel] Telemetry enabled — service: ${config.serviceName}, endpoint: ${endpoint}, sample: ${sampleRate}`,
);
return sdk;
}

42
apps/api/src/telemetry/tracing.ts Normal file
View File

@@ -0,0 +1,42 @@
import "dotenv/config"; // Load env vars before checking OTEL_ENABLED
import { isOtelEnabled } from "./index";
if (isOtelEnabled) {
// Dynamic imports to avoid loading OTel modules when disabled
const { createNodeSDK } = require("./sdk-factory") as typeof import("./sdk-factory");
const { HttpInstrumentation } = require("@opentelemetry/instrumentation-http");
const { ExpressInstrumentation } = require("@opentelemetry/instrumentation-express");
const { NestInstrumentation } = require("@opentelemetry/instrumentation-nestjs-core");
const { IORedisInstrumentation } = require("@opentelemetry/instrumentation-ioredis");
const sdk = createNodeSDK({
serviceName: process.env.OTEL_SERVICE_NAME || "sase-api",
instrumentations: [
new HttpInstrumentation({
ignoreIncomingRequestHook: (req: { url?: string }) => {
const url = req.url || "";
return url === "/api/health" || url.startsWith("/assets/");
},
}),
new ExpressInstrumentation(),
new NestInstrumentation(),
new IORedisInstrumentation(),
],
});
if (sdk) {
const shutdown = async () => {
try {
await sdk.shutdown();
console.log("[otel] Telemetry shut down cleanly");
} catch (err) {
console.error("[otel] Error during shutdown:", err);
}
};
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
}
} else {
console.log("[otel] Telemetry disabled (OTEL_ENABLED !== 'true')");
}

20
apps/api/src/telemetry/worker-tracing.ts Normal file
View File

@@ -0,0 +1,20 @@
import "dotenv/config"; // Load env vars before checking OTEL_ENABLED
import { isOtelEnabled } from "./index";
import type { NodeSDK } from "@opentelemetry/sdk-node";
let sdk: NodeSDK | null = null;
if (isOtelEnabled) {
const { createNodeSDK } = require("./sdk-factory") as typeof import("./sdk-factory");
const { HttpInstrumentation } = require("@opentelemetry/instrumentation-http");
const { IORedisInstrumentation } = require("@opentelemetry/instrumentation-ioredis");
sdk = createNodeSDK({
serviceName: process.env.OTEL_SERVICE_NAME || "sase-worker",
instrumentations: [new HttpInstrumentation(), new IORedisInstrumentation()],
});
} else {
console.log("[worker][otel] Telemetry disabled (OTEL_ENABLED !== 'true')");
}
export { sdk };

35
apps/api/src/worker.ts
View File

@@ -1,11 +1,15 @@
import "./telemetry/worker-tracing"; // MUST be first — instruments modules before they load
import "dotenv/config";
import postgres from "postgres";
import { drizzle } from "drizzle-orm/postgres-js";
import { Worker } from "bullmq";
import { getBullConnection, QUEUE_NAMES } from "./jobs/bull.config";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "./jobs/bull.config";
import { processEmexScrape } from "./jobs/processors/emex-scrape.processor";
import { processSubscriptionExpiry } from "./jobs/processors/subscription-expiry.processor";
import { processQueryCleanup } from "./jobs/processors/query-cleanup.processor";
import { isOtelEnabled } from "./telemetry";
import { sdk } from "./telemetry/worker-tracing";
// ─── Database connection (standalone, no NestJS) ──────
const databaseUrl = process.env.DATABASE_URL;
@@ -20,10 +24,22 @@ const sql = postgres(databaseUrl, {
connect_timeout: 10,
});
const db = drizzle(sql);
let db = drizzle(sql);
// Conditionally instrument Drizzle with OTel
if (isOtelEnabled) {
try {
const { instrumentDrizzle } = require("@kubiks/otel-drizzle");
db = instrumentDrizzle(db);
console.log("[worker][otel] Drizzle ORM instrumented");
} catch (err) {
console.warn("[worker][otel] Drizzle instrumentation unavailable:", (err as Error).message);
}
}
// ─── BullMQ connection config ─────────────────────────
const connection = getBullConnection();
const telemetry = getBullTelemetry();
// ─── Workers ──────────────────────────────────────────
const workers: Worker[] = [];
@@ -37,6 +53,7 @@ const emexScrapeWorker = new Worker(
{
connection,
concurrency: 3,
...(telemetry ? { telemetry } : {}),
limiter: {
max: 10,
duration: 60000,
@@ -65,6 +82,7 @@ const subscriptionExpiryWorker = new Worker(
{
connection,
concurrency: 1,
...(telemetry ? { telemetry } : {}),
},
);
@@ -87,6 +105,7 @@ const queryCleanupWorker = new Worker(
{
connection,
concurrency: 1,
...(telemetry ? { telemetry } : {}),
},
);
@@ -105,16 +124,22 @@ console.log("[worker] BullMQ workers started");
console.log(`[worker] Listening on queues: ${Object.values(QUEUE_NAMES).join(", ")}`);
console.log(`[worker] Redis: ${process.env.REDIS_HOST || "localhost"}:${process.env.REDIS_PORT || 6379}`);
// ─── Graceful shutdown ────────────────────────────────
// ─── Graceful shutdown (OTel → Workers → DB) ─────────
async function shutdown(signal: string) {
console.log(`[worker] Received ${signal}, shutting down gracefully...`);
try {
// Close all workers (wait for current jobs to finish)
// 1. Shut down OTel SDK (flush remaining spans)
if (sdk) {
await sdk.shutdown();
console.log("[worker][otel] Telemetry shut down");
}
// 2. Close all workers (wait for current jobs to finish)
await Promise.all(workers.map((w) => w.close()));
console.log("[worker] All workers closed");
// Close database connection
// 3. Close database connection
await sql.end();
console.log("[worker] Database connection closed");