because I got bored of customising my CV for every job
feat(worker): add NestJS PDF render worker with heartbeat strategy pattern
+392
+39
apps/worker/package.json
+39
apps/worker/package.json
···
1
+
{
2
+
"name": "@cv/worker",
3
+
"version": "0.0.0",
4
+
"private": true,
5
+
"type": "commonjs",
6
+
"scripts": {
7
+
"build": "tsc -p tsconfig.build.json",
8
+
"start": "node dist/main.js project-q:work async",
9
+
"dev": "nodemon --watch src -e ts --exec \"ts-node src/main.ts project-q:work async\"",
10
+
"typecheck": "tsc -p tsconfig.build.json --noEmit",
11
+
"lint": "biome check .",
12
+
"lint:fix": "biome check --write ."
13
+
},
14
+
"dependencies": {
15
+
"@cv/system": "*",
16
+
"@nestjs/common": "^10.4.7",
17
+
"@nestjs/config": "^3.2.0",
18
+
"@nestjs/core": "^10.4.7",
19
+
"@riotbyte/project-q-core": "link:/Users/niels/Developer/riotbyte/project-q/packages/core",
20
+
"@riotbyte/project-q-nestjs": "link:/Users/niels/Developer/riotbyte/project-q/packages/framework/nest",
21
+
"@riotbyte/project-q-prisma": "link:/Users/niels/Developer/riotbyte/project-q/packages/transport/prisma",
22
+
"@riotbyte/nest-service-locator": "link:/Users/niels/Developer/riotbyte/nest-service-locator",
23
+
"@nestjs/event-emitter": "^3.0.1",
24
+
"nest-commander": "^3.16.0",
25
+
"eventemitter2": "^6.4.9",
26
+
"zod": "^4.3.6",
27
+
"playwright": "^1.52.0",
28
+
"playwright-core": "^1.52.0",
29
+
"reflect-metadata": "^0.2.2",
30
+
"rxjs": "^7.8.1"
31
+
},
32
+
"devDependencies": {
33
+
"@cv/tsconfig": "*",
34
+
"@types/node": "^22.7.5",
35
+
"nodemon": "^3.1.7",
36
+
"ts-node": "^10.9.2",
37
+
"typescript": "^5.6.3"
38
+
}
39
+
}
+23
apps/worker/src/config.ts
+23
apps/worker/src/config.ts
···
1
+
const requireEnv = (key: string): string => {
2
+
const value = process.env[key];
3
+
if (!value) {
4
+
throw new Error(`Missing required env var: ${key}`);
5
+
}
6
+
return value;
7
+
};
8
+
9
+
export type WorkerConfig = typeof config;
10
+
11
+
export const config = {
12
+
get databaseUrl() {
13
+
return requireEnv("DATABASE_URL");
14
+
},
15
+
queueSchema: process.env["QUEUE_SCHEMA"] ?? "queue",
16
+
queueName: process.env["QUEUE_NAME"] ?? "default",
17
+
pollIntervalMs: Number(process.env["POLL_INTERVAL_MS"] ?? "1000"),
18
+
pdfOutputDir: process.env["PDF_OUTPUT_DIR"] ?? "./pdf-output",
19
+
pdfTimeoutMs: Number(process.env["PDF_TIMEOUT_MS"] ?? "30000"),
20
+
heartbeatFilePath:
21
+
process.env["HEARTBEAT_FILE_PATH"] ?? "/tmp/worker-heartbeat",
22
+
heartbeatDbIntervalMs: Number(process.env["HEARTBEAT_DB_INTERVAL_MS"] ?? "0"),
23
+
} as const;
+41
apps/worker/src/handlers/render-pdf.handler.ts
+41
apps/worker/src/handlers/render-pdf.handler.ts
···
1
+
import * as fs from "node:fs/promises";
2
+
import * as path from "node:path";
3
+
import { Inject, Injectable, Logger } from "@nestjs/common";
4
+
import type { Envelope, Handler } from "@riotbyte/project-q-core";
5
+
import { HandlerTag } from "@riotbyte/project-q-nestjs";
6
+
import type { WorkerConfig } from "../config";
7
+
import { HtmlToPdfService } from "../pdf/html-to-pdf.service";
8
+
import { WORKER_CONFIG } from "../worker.module";
9
+
10
+
type RenderPdfData = {
11
+
cvId: string;
12
+
html: string;
13
+
requestedBy: string;
14
+
};
15
+
16
+
@Injectable()
17
+
@HandlerTag.decorator({ handles: "render-pdf" })
18
+
export class RenderPdfHandler implements Handler {
19
+
private readonly logger = new Logger(RenderPdfHandler.name);
20
+
21
+
constructor(
22
+
private readonly pdfService: HtmlToPdfService,
23
+
@Inject(WORKER_CONFIG) private readonly config: WorkerConfig,
24
+
) {}
25
+
26
+
async handle(envelope: Envelope): Promise<void> {
27
+
const { cvId, html, requestedBy } = envelope.message.data as RenderPdfData;
28
+
29
+
this.logger.log(
30
+
`Rendering PDF for CV ${cvId} (requested by ${requestedBy})`,
31
+
);
32
+
33
+
const pdf = await this.pdfService.convert(html, this.config.pdfTimeoutMs);
34
+
35
+
await fs.mkdir(this.config.pdfOutputDir, { recursive: true });
36
+
const outputPath = path.join(this.config.pdfOutputDir, `${cvId}.pdf`);
37
+
await fs.writeFile(outputPath, pdf);
38
+
39
+
this.logger.log(`PDF written to ${outputPath} (${pdf.length} bytes)`);
40
+
}
41
+
}
+48
apps/worker/src/heartbeat/db-heartbeat.strategy.ts
+48
apps/worker/src/heartbeat/db-heartbeat.strategy.ts
···
1
+
import pg from "pg";
2
+
import type { HeartbeatStrategy } from "./heartbeat.strategy";
3
+
4
+
/** Writes periodic heartbeats to a Postgres table for distributed monitoring. */
5
+
export class DbHeartbeatStrategy implements HeartbeatStrategy {
6
+
private pool: pg.Pool | null = null;
7
+
private workerId: string | null = null;
8
+
private lastWriteAt = 0;
9
+
10
+
constructor(
11
+
private readonly databaseUrl: string,
12
+
private readonly intervalMs: number,
13
+
) {}
14
+
15
+
async onStart(workerId: string): Promise<void> {
16
+
this.pool = new pg.Pool({ connectionString: this.databaseUrl });
17
+
this.workerId = workerId;
18
+
}
19
+
20
+
async onRunning(): Promise<void> {
21
+
if (!(this.pool && this.workerId)) {
22
+
return;
23
+
}
24
+
25
+
const now = Date.now();
26
+
if (now - this.lastWriteAt < this.intervalMs) {
27
+
return;
28
+
}
29
+
30
+
this.lastWriteAt = now;
31
+
32
+
await this.pool.query(
33
+
`INSERT INTO queue.worker_heartbeats (worker_id, last_seen_at)
34
+
VALUES ($1, now())
35
+
ON CONFLICT (worker_id)
36
+
DO UPDATE SET last_seen_at = now()`,
37
+
[this.workerId],
38
+
);
39
+
}
40
+
41
+
async onStop(): Promise<void> {
42
+
if (!this.pool) {
43
+
return;
44
+
}
45
+
await this.pool.end();
46
+
this.pool = null;
47
+
}
48
+
}
+18
apps/worker/src/heartbeat/file-heartbeat.strategy.ts
+18
apps/worker/src/heartbeat/file-heartbeat.strategy.ts
···
1
+
import * as fs from "node:fs/promises";
2
+
import type { HeartbeatStrategy } from "./heartbeat.strategy";
3
+
4
+
/** Touches a file on each heartbeat tick for Docker/orchestrator health checks. */
5
+
export class FileHeartbeatStrategy implements HeartbeatStrategy {
6
+
constructor(private readonly filePath: string) {}
7
+
8
+
async onStart(): Promise<void> {
9
+
await fs.writeFile(this.filePath, "");
10
+
}
11
+
12
+
async onRunning(): Promise<void> {
13
+
const now = new Date();
14
+
await fs.utimes(this.filePath, now, now);
15
+
}
16
+
17
+
async onStop(): Promise<void> {}
18
+
}
+32
apps/worker/src/heartbeat/heartbeat.listener.ts
+32
apps/worker/src/heartbeat/heartbeat.listener.ts
···
1
+
import { Inject, Injectable, type OnModuleInit } from "@nestjs/common";
2
+
import {
3
+
WorkerRunningEvent,
4
+
WorkerStartedEvent,
5
+
WorkerStoppedEvent,
6
+
} from "@riotbyte/project-q-core";
7
+
import { EventEmitter2 } from "eventemitter2";
8
+
import type { HeartbeatStrategy } from "./heartbeat.strategy";
9
+
import { HEARTBEAT_STRATEGIES } from "./heartbeat.strategy";
10
+
11
+
@Injectable()
12
+
export class HeartbeatListener implements OnModuleInit {
13
+
constructor(
14
+
private readonly emitter: EventEmitter2,
15
+
@Inject(HEARTBEAT_STRATEGIES)
16
+
private readonly strategies: HeartbeatStrategy[],
17
+
) {}
18
+
19
+
onModuleInit(): void {
20
+
this.emitter.on(WorkerStartedEvent.name, async (event: WorkerStartedEvent) => {
21
+
await Promise.all(this.strategies.map((s) => s.onStart(event.worker.id)));
22
+
});
23
+
24
+
this.emitter.on(WorkerRunningEvent.name, async () => {
25
+
await Promise.all(this.strategies.map((s) => s.onRunning()));
26
+
});
27
+
28
+
this.emitter.on(WorkerStoppedEvent.name, async () => {
29
+
await Promise.all(this.strategies.map((s) => s.onStop()));
30
+
});
31
+
}
32
+
}
+37
apps/worker/src/heartbeat/heartbeat.module.ts
+37
apps/worker/src/heartbeat/heartbeat.module.ts
···
1
+
import type { DynamicModule } from "@nestjs/common";
2
+
import { Module } from "@nestjs/common";
3
+
import type { WorkerConfig } from "../config";
4
+
import { DbHeartbeatStrategy } from "./db-heartbeat.strategy";
5
+
import { FileHeartbeatStrategy } from "./file-heartbeat.strategy";
6
+
import { HeartbeatListener } from "./heartbeat.listener";
7
+
import type { HeartbeatStrategy } from "./heartbeat.strategy";
8
+
import { HEARTBEAT_STRATEGIES } from "./heartbeat.strategy";
9
+
10
+
@Module({})
11
+
export class HeartbeatModule {
12
+
static register(config: WorkerConfig): DynamicModule {
13
+
const strategies: HeartbeatStrategy[] = [];
14
+
15
+
if (config.heartbeatFilePath) {
16
+
strategies.push(new FileHeartbeatStrategy(config.heartbeatFilePath));
17
+
}
18
+
19
+
if (config.heartbeatDbIntervalMs > 0) {
20
+
strategies.push(
21
+
new DbHeartbeatStrategy(
22
+
config.databaseUrl,
23
+
config.heartbeatDbIntervalMs,
24
+
),
25
+
);
26
+
}
27
+
28
+
return {
29
+
module: HeartbeatModule,
30
+
providers: [
31
+
{ provide: HEARTBEAT_STRATEGIES, useValue: strategies },
32
+
HeartbeatListener,
33
+
],
34
+
exports: [HEARTBEAT_STRATEGIES],
35
+
};
36
+
}
37
+
}
+8
apps/worker/src/heartbeat/heartbeat.strategy.ts
+8
apps/worker/src/heartbeat/heartbeat.strategy.ts
···
1
+
export const HEARTBEAT_STRATEGIES = Symbol("HEARTBEAT_STRATEGIES");
2
+
3
+
/** Lifecycle-aware heartbeat strategy for worker health monitoring. */
4
+
export interface HeartbeatStrategy {
5
+
onStart(workerId: string): Promise<void>;
6
+
onRunning(): Promise<void>;
7
+
onStop(): Promise<void>;
8
+
}
+18
apps/worker/src/logger.provider.ts
+18
apps/worker/src/logger.provider.ts
···
1
+
import { Logger as NestLogger } from "@nestjs/common";
2
+
import type { Logger } from "@riotbyte/project-q-core";
3
+
4
+
export class NestProjectQLogger implements Logger {
5
+
private readonly logger = new NestLogger("ProjectQ");
6
+
7
+
info(message: string): void {
8
+
this.logger.log(message);
9
+
}
10
+
11
+
error(message: string): void {
12
+
this.logger.error(message);
13
+
}
14
+
15
+
debug(message: string): void {
16
+
this.logger.debug(message);
17
+
}
18
+
}
+18
apps/worker/src/main.ts
+18
apps/worker/src/main.ts
···
1
+
import "reflect-metadata";
2
+
import { Logger } from "@nestjs/common";
3
+
import { CommandFactory } from "nest-commander";
4
+
import { WorkerModule } from "./worker.module";
5
+
6
+
const logger = new Logger("Bootstrap");
7
+
8
+
async function bootstrap(): Promise<void> {
9
+
await CommandFactory.run(WorkerModule, {
10
+
logger: ["log", "error", "warn", "debug"],
11
+
cliName: "worker",
12
+
});
13
+
}
14
+
15
+
bootstrap().catch((error) => {
16
+
logger.error("Worker crashed", error);
17
+
process.exit(1);
18
+
});
+50
apps/worker/src/pdf/html-to-pdf.service.ts
+50
apps/worker/src/pdf/html-to-pdf.service.ts
···
1
+
import { Injectable, Logger, OnModuleDestroy } from "@nestjs/common";
2
+
import { chromium } from "playwright";
3
+
import type { Browser } from "playwright-core";
4
+
5
+
/** Manages a Playwright browser singleton for HTML-to-PDF conversion. */
6
+
@Injectable()
7
+
export class HtmlToPdfService implements OnModuleDestroy {
8
+
private readonly logger = new Logger(HtmlToPdfService.name);
9
+
private browser: Browser | null = null;
10
+
11
+
private async ensureBrowser(): Promise<Browser> {
12
+
if (this.browser) {
13
+
return this.browser;
14
+
}
15
+
16
+
this.logger.log("Launching Chromium");
17
+
this.browser = await chromium.launch({
18
+
args: ["--no-sandbox", "--disable-dev-shm-usage"],
19
+
});
20
+
21
+
return this.browser;
22
+
}
23
+
24
+
/** Renders HTML to an A4 PDF buffer. */
25
+
async convert(html: string, timeoutMs = 30_000): Promise<Buffer> {
26
+
const browser = await this.ensureBrowser();
27
+
const page = await browser.newPage();
28
+
page.setDefaultTimeout(timeoutMs);
29
+
30
+
try {
31
+
await page.setContent(html, {
32
+
waitUntil: "networkidle",
33
+
timeout: timeoutMs,
34
+
});
35
+
36
+
return await page.pdf({ format: "A4", printBackground: true });
37
+
} finally {
38
+
await page.close();
39
+
}
40
+
}
41
+
42
+
async onModuleDestroy(): Promise<void> {
43
+
if (!this.browser) {
44
+
return;
45
+
}
46
+
this.logger.log("Closing Chromium");
47
+
await this.browser.close();
48
+
this.browser = null;
49
+
}
50
+
}
+41
apps/worker/src/worker.module.ts
+41
apps/worker/src/worker.module.ts
···
1
+
import { Module } from "@nestjs/common";
2
+
import { ConfigModule } from "@nestjs/config";
3
+
import { EventEmitterModule } from "@nestjs/event-emitter";
4
+
import { DatabaseModule, PrismaService } from "@cv/system";
5
+
import { LoggerProvider, MessengerModule } from "@riotbyte/project-q-nestjs";
6
+
import { PrismaClientToken, PrismaTransportFactory } from "@riotbyte/project-q-prisma";
7
+
import type { WorkerConfig } from "./config";
8
+
import { config } from "./config";
9
+
import { RenderPdfHandler } from "./handlers/render-pdf.handler";
10
+
import { HeartbeatModule } from "./heartbeat/heartbeat.module";
11
+
import { NestProjectQLogger } from "./logger.provider";
12
+
import { HtmlToPdfService } from "./pdf/html-to-pdf.service";
13
+
14
+
export const WORKER_CONFIG = Symbol("WORKER_CONFIG");
15
+
16
+
@Module({
17
+
imports: [
18
+
ConfigModule.forRoot({
19
+
isGlobal: true,
20
+
load: [() => ({ DATABASE_URL: config.databaseUrl })],
21
+
}),
22
+
DatabaseModule,
23
+
EventEmitterModule.forRoot(),
24
+
HeartbeatModule.register(config),
25
+
MessengerModule.forRoot({
26
+
transports: {
27
+
async: { dsn: `prisma://?queue=${config.queueName}`, retry: true },
28
+
},
29
+
routing: new Map<string, "async">([["render-pdf", "async"]]),
30
+
}),
31
+
],
32
+
providers: [
33
+
{ provide: WORKER_CONFIG, useValue: config satisfies WorkerConfig },
34
+
{ provide: PrismaClientToken, useExisting: PrismaService },
35
+
{ provide: LoggerProvider, useClass: NestProjectQLogger },
36
+
PrismaTransportFactory,
37
+
RenderPdfHandler,
38
+
HtmlToPdfService,
39
+
],
40
+
})
41
+
export class WorkerModule {}
+8
apps/worker/tsconfig.build.json
+8
apps/worker/tsconfig.build.json
+10
apps/worker/tsconfig.json
+10
apps/worker/tsconfig.json