[Refactoring]

Workers Pattern Consolidation - Complete Reference

Overview

This document describes the pattern consolidation refactorings applied to the hoox workers monorepo. These changes eliminate code duplication and establish standardized patterns for common worker patterns including queue handling, cron scheduling, and exchange client implementations.

Status: ✅ Production-Ready
Last Updated: June 4, 2026
Version: 1.0

---

What Changed

1. Queue Handler Factory ✅

Location: packages/shared/src/queue-handler.ts

What: A reusable factory for handling CloudFlare Queues with retry logic and exponential backoff.

Before: Each queue-based worker had inline retry/backoff logic (60+ lines per worker)

After: Single factory used across all workers (30 lines per worker)

Code Reduction:

• trade-worker: 61 lines → 30 lines (51% reduction)
• Consistent retry behavior across all queue-based workers

Usage Example:

import { createQueueHandler } from "@hoox-sh/hoox-shared/queue-handler";

export default {
  async queue(batch, env, ctx) {
    const handler = createQueueHandler({
      maxRetries: 5,
      backoffDelays: [0, 30, 60, 300, 900],
      logger,
      onMessage: async (message, attemptNumber) => {
        // Your message processing logic
        const result = await processMessage(message, env, ctx);
        if (!result.success) throw new Error(result.error);
      },
      onRetry: (_message, _attemptNumber, _error, _delaySeconds) => {
        // Logging is handled by createQueueHandler internally
      },
      onDLQ: async (message, attemptNumber, error) => {
        // Handle dead-lettering
        await logFailedMessage(message, error);
      },
    });

    return await handler(batch);
  },
};

Key Features:

• ✅ Generic message type support (<T>)
• ✅ Exponential backoff with configurable delays
• ✅ Automatic retry logic with attempt tracking
• ✅ Dead-letter queue (DLQ) handling
• ✅ Built-in logging with optional logger
• ✅ Type-safe with TypeScript

Benefits:

• ✅ Reduced code duplication (51% reduction in trade-worker)
• ✅ Consistent retry behavior across workers
• ✅ Type-safe with generic message types
• ✅ Built-in logging with logger option
• ✅ Easier to test and maintain

Updated Workers:

• workers/trade-worker - Uses factory for queue message handling

---

2. Cron Handler ✅

Location: packages/shared/src/cron-handler.ts

What: Standardized handler for scheduled/cron-triggered workers with consistent logging and error handling.

Before: Custom logging in each scheduled worker, inconsistent patterns

After: Standardized logging with automatic duration measurement

Usage Example:

import { createCronHandler } from "@hoox-sh/hoox-shared/cron-handler";

export default {
  async scheduled(event, env, ctx) {
    const handler = createCronHandler({
      name: "my-worker",
      logger,
      handler: async (event, env, ctx) => {
        // Your cron logic here
        await runCronTask(env, ctx);
      },
    });

    return await handler(event, env, ctx);
  },
};

Logging Output:

my-worker cron triggered { cron: "0 * * * *", scheduledTime: 1234567890 }
my-worker cron handler completed successfully { cron: "0 * * * *", durationMs: 145 }

Error Logging:

my-worker cron handler failed { cron: "0 * * * *", durationMs: 245, error: "Task failed" }

Key Features:

• ✅ Automatic execution time measurement
• ✅ Structured logging with event details
• ✅ Consistent error handling and logging
• ✅ Generic environment type support (<Env>)
• ✅ Optional logger instance

Benefits:

• ✅ Consistent logging across all scheduled workers
• ✅ Automatic duration measurement
• ✅ Structured error logging with context
• ✅ Cleaner handler code
• ✅ Easier debugging with standardized logs

Updated Workers:

• workers/agent-worker - Uses factory for configurable cron routine (1–1440 min)
• workers/report-worker - Uses factory for report generation cron

---

3. Exchange Client Base Class ✅

Location: packages/shared/src/exchanges/

What: Abstract base class for exchange implementations (Binance, Bybit, MEXC) with common functionality.

Files:

• base-exchange-client.ts - Abstract class with 6 abstract methods
• types.ts - TypeScript types for exchanges
• index.ts - Barrel exports

Before: Base class duplicated in trade-worker, no shared interface

After: Single shared base class in packages/shared with consistent interface

Usage Example:

import { BaseExchangeClient } from "@hoox-sh/hoox-shared/exchanges";
import type {
  TradeParams,
  OrderResponse,
} from "@hoox-sh/hoox-shared/exchanges";

export class BinanceClient extends BaseExchangeClient {
  async validateApiCredentials(): Promise<boolean> {
    // Implementation
    const response = await this.makeRequest("GET", "/api/v3/account");
    return !!response.balances;
  }

  async executeTrade(params: TradeParams): Promise<OrderResponse> {
    // Implementation
    const response = await this.makeRequest("POST", "/api/v3/order", {
      symbol: params.symbol,
      side: params.side,
      quantity: params.quantity,
      price: params.price,
    });
    return response;
  }

  async getMarkPrice(symbol: string): Promise<number> {
    // Implementation
  }

  async getOpenPositions(symbol?: string): Promise<Position[]> {
    // Implementation
  }

  async closePosition(
    symbol: string,
    positionId?: string
  ): Promise<OrderResponse> {
    // Implementation
  }

  async getWalletBalance(): Promise<number> {
    // Implementation
  }
}

Abstract Methods:

• validateApiCredentials(): Promise<boolean> - Verify API credentials are valid
• executeTrade(params: TradeParams): Promise<OrderResponse> - Execute a trade
• getMarkPrice(symbol: string): Promise<number> - Get current mark price
• getOpenPositions(symbol?: string): Promise<Position[]> - Get open positions
• closePosition(symbol: string, positionId?: string): Promise<OrderResponse> - Close a position
• getWalletBalance(): Promise<number> - Get wallet balance

Protected Helper Methods:

• makeRequest(method: string, path: string, data?: any): Promise<any> - Make HTTP request (subclass implements)
• getErrorMessagePrefix(): string - Get error message prefix for logging

Key Features:

• ✅ Type-safe configuration and responses
• ✅ Consistent interface across exchanges
• ✅ Helper methods for common operations
• ✅ Crypto signature support (HMAC-SHA256)
• ✅ Testnet/sandbox support

Benefits:

• ✅ Consistent interface for all exchange implementations
• ✅ Type-safe configuration and responses
• ✅ Reusable across multiple workers
• ✅ Helper methods for URL building, signing, etc.
• ✅ Easier to add new exchanges

Updated Workers:

• workers/trade-worker - Exchange clients (Binance, Bybit, MEXC) extend shared base class

---

Migration Guide

For New Workers with Queues

If you're building a new worker that needs queue handling:

import { createQueueHandler } from "@hoox-sh/hoox-shared/queue-handler";
import type { QueueHandlerOptions } from "@hoox-sh/hoox-shared/queue-handler";

interface MyQueueMessage {
  requestId: string;
  action: string;
  data: Record<string, any>;
}

const MAX_RETRIES = 5;
const BACKOFF_DELAYS = [0, 30, 60, 300, 900];

export default {
  async queue(batch, env, ctx) {
    const handler = createQueueHandler<MyQueueMessage>({
      maxRetries: MAX_RETRIES,
      backoffDelays: BACKOFF_DELAYS,
      logger,
      onMessage: async (message, attemptNumber) => {
        // Process message
        const result = await processMessage(message, env, ctx);
        if (!result.success) throw new Error(result.error);
      },
      onRetry: (_message, _attemptNumber, _error, _delaySeconds) => {
        // Logging is handled by createQueueHandler internally
      },
      onDLQ: async (message, attempt, error) => {
        // Log to dead-letter queue
        await logToDatabase(message, error);
      },
    });

    return await handler(batch);
  },
};

Key Points:

• Define your message type as a TypeScript interface
• Set maxRetries and backoffDelays as constants
• Implement onMessage with your processing logic
• Implement onDLQ for dead-letter handling
• The factory handles all retry logic and logging

---

For New Scheduled Workers

If you're building a new worker with scheduled/cron triggers:

import { createCronHandler } from "@hoox-sh/hoox-shared/cron-handler";
import type { CronHandlerOptions } from "@hoox-sh/hoox-shared/cron-handler";

interface Env {
  // Your environment bindings
  DB: D1Database;
  KV: KVNamespace;
}

export default {
  async scheduled(event, env, ctx) {
    const handler = createCronHandler<Env>({
      name: "my-scheduled-worker",
      logger,
      handler: async (event, env, ctx) => {
        // Your cron logic
        await runRoutine(env, ctx);
      },
    });

    return await handler(event, env, ctx);
  },
};

Key Points:

• Define your Env type with all bindings
• Set name to your worker name for logging
• Implement handler with your cron logic
• The factory handles logging and duration measurement
• Errors are logged with context and re-thrown

---

For New Exchange Clients

If you're adding a new exchange:

import { BaseExchangeClient } from "@hoox-sh/hoox-shared/exchanges";
import type {
  TradeParams,
  OrderResponse,
  Position,
} from "@hoox-sh/hoox-shared/exchanges";

export class MyExchangeClient extends BaseExchangeClient {
  constructor(apiKey: string, apiSecret: string) {
    super(apiKey, apiSecret);
  }

  async validateApiCredentials(): Promise<boolean> {
    try {
      const response = await this.makeRequest("GET", "/api/account");
      return !!response.id;
    } catch {
      return false;
    }
  }

  async executeTrade(params: TradeParams): Promise<OrderResponse> {
    try {
      const response = await this.makeRequest("POST", "/api/orders", {
        symbol: params.symbol,
        side: params.side,
        quantity: params.quantity,
        price: params.price,
      });
      return {
        orderId: response.id,
        symbol: params.symbol,
        status: response.status,
      };
    } catch (e) {
      throw new Error(`Trade execution failed: ${String(e)}`);
    }
  }

  async getMarkPrice(symbol: string): Promise<number> {
    const response = await this.makeRequest("GET", `/api/ticker/${symbol}`);
    return parseFloat(response.price);
  }

  async getOpenPositions(symbol?: string): Promise<Position[]> {
    const response = await this.makeRequest("GET", "/api/positions");
    return response.positions.map((p: any) => ({
      symbol: p.symbol,
      side: p.side.toLowerCase() as "long" | "short",
      quantity: p.quantity,
      entryPrice: p.entryPrice,
      unrealizedPnl: p.pnl,
    }));
  }

  async closePosition(
    symbol: string,
    positionId?: string
  ): Promise<OrderResponse> {
    const response = await this.makeRequest("POST", "/api/positions/close", {
      symbol,
      positionId,
    });
    return {
      orderId: response.id,
      symbol,
      status: response.status,
    };
  }

  async getWalletBalance(): Promise<number> {
    const response = await this.makeRequest("GET", "/api/account/balance");
    return response.total;
  }

  protected async makeRequest(
    method: string,
    path: string,
    data?: any
  ): Promise<any> {
    // Implement your exchange-specific HTTP request logic
    // This is where you handle authentication, signing, etc.
    const url = `${this.baseUrl}${path}`;
    const response = await fetch(url, {
      method,
      headers: {
        "X-API-Key": this.apiKey,
        // Add other headers as needed
      },
      body: data ? JSON.stringify(data) : undefined,
    });

    if (!response.ok) {
      throw new Error(`API error: ${response.statusText}`);
    }

    return response.json();
  }
}

Key Points:

• Extend BaseExchangeClient with your exchange name
• Implement all 6 abstract methods
• Implement makeRequest with your exchange-specific logic
• Use the base class constructor to initialize credentials
• Return typed responses for type safety

---

Testing

All refactored patterns are fully tested:

Test Files

• packages/shared/src/__tests__/queue-handler.test.ts - 5 test cases
• packages/shared/src/__tests__/cron-handler.test.ts - 5 test cases
• packages/shared/src/__tests__/exchanges.test.ts - 3 test cases

Running Tests

# Test shared package patterns
bun test packages/shared/

# Test specific pattern
bun test packages/shared/src/__tests__/queue-handler.test.ts

# Test all workers
bun test workers/

# Test specific worker
bun test workers/trade-worker/

Test Coverage

All patterns have comprehensive test coverage:

---

Files Modified

Created Files

• packages/shared/src/queue-handler.ts - Queue handler factory (92 lines)
• packages/shared/src/cron-handler.ts - Cron handler factory (77 lines)
• packages/shared/src/exchanges/base-exchange-client.ts - Base class (209 lines)
• packages/shared/src/exchanges/types.ts - Exchange types (20 lines)
• packages/shared/src/exchanges/index.ts - Barrel exports
• packages/shared/src/__tests__/queue-handler.test.ts - Queue tests
• packages/shared/src/__tests__/cron-handler.test.ts - Cron tests
• packages/shared/src/__tests__/exchanges.test.ts - Exchange tests

Modified Files

• packages/shared/src/index.ts - Added exports for new utilities
• workers/trade-worker/src/index.ts - Updated to use createQueueHandler
• workers/agent-worker/src/index.ts - Updated to use createCronHandler
• workers/report-worker/src/index.ts - Updated to use createCronHandler
• workers/trade-worker/src/mexc-client.ts - Updated imports
• workers/trade-worker/src/binance-client.ts - Updated imports
• workers/trade-worker/src/bybit-client.ts - Updated imports

---

Impact Summary

---

Best Practices Going Forward

1. Use createQueueHandler for all queue-based workers

• No manual retry/backoff logic
• Consistent error handling and logging
• Type-safe with generic message types

const handler = createQueueHandler<YourMessageType>({
  maxRetries: 5,
  backoffDelays: [0, 30, 60, 300, 900],
  logger,
  onMessage: async (msg) => {
    /* ... */
  },
  onDLQ: async (msg, attempt, error) => {
    /* ... */
  },
});

2. Use createCronHandler for all scheduled workers

• Standardized logging format
• Automatic duration measurement
• Consistent error handling

const handler = createCronHandler<Env>({
  name: "my-worker",
  logger,
  handler: async (event, env, ctx) => {
    /* ... */
  },
});

3. Extend BaseExchangeClient for new exchanges

• Type-safe configuration
• Consistent interface across exchanges
• Reusable helper methods

export class MyExchangeClient extends BaseExchangeClient {
  async validateApiCredentials(): Promise<boolean> {
    /* ... */
  }
  async executeTrade(params: TradeParams): Promise<OrderResponse> {
    /* ... */
  }
  // ... implement other abstract methods
}

4. Keep helper/utility functions organized

• Factories in packages/shared/src/ for framework concerns (queues, cron)
• Helpers in logic/ directories for domain logic (trade execution, notifications)
• Types in types.ts or exchanges/types.ts for shared types

5. Test all patterns before committing

• All pattern tests are in packages/shared/__tests__/
• Worker tests cover integration
• Run bun test before committing

---

Troubleshooting

Queue Handler Issues

Problem: Messages not being retried

• Check maxRetries is greater than 0
• Verify backoffDelays array has enough entries
• Ensure onMessage throws error on failure

Problem: DLQ not being called

• Verify onDLQ is implemented
• Check that maxRetries is being exceeded
• Ensure message attempts counter is working

Cron Handler Issues

Problem: Logs not appearing

• Verify logger is passed to handler
• Check logger implementation has info and error methods
• Ensure logger is properly initialized

Problem: Duration not measured

• Duration is always measured automatically
• Check logger output for durationMs field
• Verify handler is actually being called

Exchange Client Issues

Problem: API credentials validation failing

• Verify apiKey and apiSecret are correct
• Check exchange API endpoint is accessible
• Ensure makeRequest is properly implemented

Problem: Trade execution failing

• Verify TradeParams are correct for exchange
• Check exchange API documentation for parameter names
• Ensure proper error handling in executeTrade

---

Related Documentation

• Architecture Overview
• Service Bindings
• API Endpoint Directory
• Development Testing

---

Changelog

Version 1.0 (June 4, 2026)

• ✅ Queue handler factory implemented
• ✅ Cron handler factory implemented
• ✅ Exchange client base class extracted
• ✅ All workers refactored to use new patterns
• ✅ Comprehensive test coverage added
• ✅ Documentation completed

---

Status: Production-Ready
Last Updated: June 4, 2026
Maintainer: Hoox Development Team