[Workers Pattern Consolidation]

Overview

This document describes the pattern consolidation refactorings applied to the hoox-setup 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 "@jango-blockchained/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 "@jango-blockchained/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 5-minute cron routine
• 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 "@jango-blockchained/hoox-shared/exchanges";
import type {
  TradeParams,
  OrderResponse,
} from "@jango-blockchained/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 "@jango-blockchained/hoox-shared/queue-handler";
import type { QueueHandlerOptions } from "@jango-blockchained/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 "@jango-blockchained/hoox-shared/cron-handler";
import type { CronHandlerOptions } from "@jango-blockchained/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 "@jango-blockchained/hoox-shared/exchanges";
import type {
  TradeParams,
  OrderResponse,
  Position,
} from "@jango-blockchained/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