Feature/add workflow (#30)

feat(worker): enhance worker functionality and error handling

- Introduced utility functions for environment validation and error
logging in the new `utils.ts` file.
- Updated the main worker script to validate environment variables at
startup and improved error handling with consistent logging.
- Refactored the `run` function to establish a connection and handle
errors more gracefully.
- Adjusted Vitest configuration to lower coverage thresholds for better
test management.
- Added new tests for the `weeklyFinancialReportsWorkflow` and improved
existing tests for error handling.

These changes enhance the worker's reliability and maintainability by
ensuring proper environment validation and consistent error logging.

Author: anatolyshipitz

Dockerfile.temporal-worker-main

@@ -8,7 +8,6 @@ RUN npm ci --ignore-scripts
 FROM node:20-bullseye AS dev
 # sonarcloud-disable-next-line docker:S4507
 ENV NODE_ENV=development
-ENV DEBUG=*
 WORKDIR /app/main
 COPY --from=deps /app/main/node_modules ./node_modules
 CMD ["npx", "nodemon", "--watch", "./", "--watch", "/app/common", "--ext", "ts", "--exec", "npx", "ts-node", "src/index.ts"]

workers/common/utils.ts

@@ -0,0 +1,36 @@
+import { validationResult } from '../main/src/configs';
+import {logger} from "../main";
+
+export const formatValidationIssues = (issues: { path: (string | number)[]; message: string }[]): string =>
+  issues
+    .map(({ path, message }) => `Missing or invalid environment variable: ${path.join('.') || '(unknown variable)'} (${message})`)
+    .join('\n');
+
+export function validateEnv() {
+  if (!validationResult.success) {
+    console.error(formatValidationIssues(validationResult.error.issues));
+    process.exit(1);
+  }
+}
+
+/**
+ * Logs a worker error in a consistent format.
+ * @param workerName - The name of the workflow
+ * @param error - The error object
+ */
+export function logWorkerError(workerName: string, error: unknown) {
+  logger.error(
+    `Error in ${workerName} workerName: ${error instanceof Error ? error.message : String(error)}`,
+  );
+}
+
+/**
+ * Logs a workflow error in a consistent format.
+ * @param workflowName - The name of the workflow
+ * @param error - The error object
+ */
+export function logWorkflowError(workflowName: string, error: unknown) {
+  logger.error(
+    `Error in ${workflowName} workflow: ${error instanceof Error ? error.message : String(error)}`,
+  );
+}

workers/main/src/__tests__/index.test.ts

@@ -1,37 +1,38 @@
 import { describe, expect, it } from 'vitest';
 import { vi } from 'vitest';
 
-import { handleRunError, logger, run } from '../index';
+import * as utils from '../../../common/utils';
+import { handleRunError, run } from '../index';
+
+vi.mock('@temporalio/worker', () => ({
+  DefaultLogger: class {
+    error() {}
+  },
+  NativeConnection: {
+    connect: vi.fn().mockResolvedValue({ close: vi.fn() }),
+  },
+  Worker: {
+    create: vi
+      .fn()
+      .mockResolvedValue({ run: vi.fn().mockResolvedValue(undefined) }),
+  },
+}));
 
 describe('run', () => {
   it('should return true', async () => {
-    await expect(run()).resolves.toBe(true);
+    await expect(run()).resolves.toBeUndefined();
   });
 });
 
 describe('handleRunError', () => {
-  it('should log error and exit process', () => {
-    vi.useFakeTimers();
-    const error = new Error('test error');
-    const loggerErrorSpy = vi
-      .spyOn(logger, 'error')
+  it('should log the error and throw the error', () => {
+    const logSpy = vi
+      .spyOn(utils, 'logWorkerError')
       .mockImplementation(() => {});
-    const processExitSpy = vi.spyOn(process, 'exit').mockImplementation(() => {
-      throw new Error('exit');
-    });
+    const error = new Error('test error');
 
     expect(() => handleRunError(error)).toThrow(error);
-    expect(loggerErrorSpy).toHaveBeenCalledWith(
-      `Unhandled error in main: ${error.message}`,
-    );
-    expect(processExitSpy).not.toHaveBeenCalled();
-    expect(() => {
-      vi.runAllTimers();
-    }).toThrow('exit');
-    expect(processExitSpy).toHaveBeenCalledWith(1);
-
-    loggerErrorSpy.mockRestore();
-    processExitSpy.mockRestore();
-    vi.useRealTimers();
+    expect(logSpy).toHaveBeenCalledWith('main', error);
+    logSpy.mockRestore();
   });
 });

workers/main/src/__tests__/weeklyFinancialReports.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import * as utils from '../../../common/utils';
+import { weeklyFinancialReportsWorkflow } from '../workflows';
+
+describe('weeklyFinancialReportsWorkflow', () => {
+  it('should return the report string with default parameters', async () => {
+    const result = await weeklyFinancialReportsWorkflow();
+
+    expect(typeof result).toBe('string');
+    expect(result.length).toBeGreaterThan(0);
+  });
+
+  it('should return the report string for a custom period', async () => {
+    const result = await weeklyFinancialReportsWorkflow({
+      period: 'Q1 2025',
+    });
+
+    expect(result.startsWith('Weekly Financial Report')).toBe(true);
+    expect(result).toContain('Period: Q1 2025');
+  });
+
+  it('should log and rethrow errors', async () => {
+    const logSpy = vi
+      .spyOn(utils, 'logWorkflowError')
+      .mockImplementation(() => {});
+    const originalToLocaleString = Number.prototype.toLocaleString.bind(
+      Number.prototype,
+    );
+
+    Number.prototype.toLocaleString = () => {
+      throw new Error('Test error');
+    };
+
+    await expect(weeklyFinancialReportsWorkflow()).rejects.toThrow(
+      'Test error',
+    );
+    expect(logSpy).toHaveBeenCalledWith(
+      'Weekly Financial Reports',
+      expect.any(Error),
+    );
+
+    Number.prototype.toLocaleString = originalToLocaleString;
+    logSpy.mockRestore();
+  });
+});

workers/main/src/activities/fetchFinancialData.ts

@@ -0,0 +1,32 @@
+export interface FinancialData {
+  period: string;
+  contractType: string;
+  revenue: number;
+  cogs: number;
+  margin: number;
+  marginality: number;
+  effectiveRevenue: number;
+  effectiveMargin: number;
+  effectiveMarginality: number;
+}
+
+/**
+ * Fetches financial data for a given period from an external source or database.
+ * @param period - The period to fetch data for (e.g., 'Q1 2025', 'current')
+ */
+export async function fetchFinancialData(
+  period: string = 'current',
+): Promise<FinancialData> {
+  // TODO: Replace this stub with actual data fetching logic (e.g., DB query, API call)
+  return {
+    period: period,
+    contractType: 'T&M',
+    revenue: 120000,
+    cogs: 80000,
+    margin: 40000,
+    marginality: 33.3,
+    effectiveRevenue: 110000,
+    effectiveMargin: 35000,
+    effectiveMarginality: 31.8,
+  };
+}

workers/main/src/configs/index.ts

@@ -0,0 +1,6 @@
+import { temporalSchema } from './temporal';
+import { workerSchema } from './worker';
+
+export const validationResult = temporalSchema
+  .merge(workerSchema)
+  .safeParse(process.env);

workers/main/src/configs/temporal.ts

@@ -0,0 +1,12 @@
+import { NativeConnectionOptions } from '@temporalio/worker';
+import { z } from 'zod';
+
+const DEFAULT_TEMPORAL_ADDRESS = 'temporal:7233';
+
+export const temporalConfig: NativeConnectionOptions = {
+  address: process.env.TEMPORAL_ADDRESS || DEFAULT_TEMPORAL_ADDRESS,
+};
+
+export const temporalSchema = z.object({
+  TEMPORAL_ADDRESS: z.string().default(DEFAULT_TEMPORAL_ADDRESS),
+});

workers/main/src/configs/worker.ts

@@ -0,0 +1,13 @@
+import { WorkerOptions } from '@temporalio/worker';
+import path from 'path';
+import { z } from 'zod';
+
+export const workerConfig: WorkerOptions = {
+  taskQueue: 'main-queue',
+  workflowsPath:
+    process.env.WORKFLOWS_PATH || path.join(__dirname, '../workflows'),
+};
+
+export const workerSchema = z.object({
+  WORKFLOWS_PATH: z.string().optional(),
+});

workers/main/src/index.ts

@@ -1,19 +1,52 @@
-import { DefaultLogger } from '@temporalio/worker';
+import { DefaultLogger, NativeConnection, Worker } from '@temporalio/worker';
+
+import { logWorkerError, validateEnv } from '../../common/utils';
+import { temporalConfig } from './configs/temporal';
+import { workerConfig } from './configs/worker';
 
 export const logger = new DefaultLogger('ERROR');
 
-/**
- * Executes the main worker process.
- * @returns {Promise<boolean>} Returns true when the worker completes successfully.
- */
-export async function run(): Promise<boolean> {
-  return true;
+validateEnv();
+
+export async function createConnection() {
+  return NativeConnection.connect(temporalConfig);
+}
+
+export async function createWorker(connection: NativeConnection) {
+  const workerOptions = {
+    ...workerConfig,
+    connection,
+  };
+
+  return Worker.create(workerOptions);
 }
 
-export function handleRunError(err: Error): never {
-  logger.error(`Unhandled error in main: ${err.message}`);
+export async function run(): Promise<void> {
+  const connection = await createConnection();
+
+  try {
+    const worker = await createWorker(connection);
+
+    await worker.run();
+  } catch (err) {
+    handleRunError(err);
+  } finally {
+    if (connection) {
+      await connection.close();
+    }
+  }
+}
+
+export function handleRunError(err: unknown): never {
+  logWorkerError('main', err);
   setTimeout(() => process.exit(1), 100);
   throw err;
 }
 
-run().catch(handleRunError);
+export function mainEntry() {
+  if (require.main === module) {
+    run().catch(handleRunError);
+  }
+}
+
+mainEntry();

workers/main/src/workflows/index.ts

@@ -0,0 +1 @@
+export * from './weeklyFinancialReports';