Horizontal Scaling

services:
  worker:
    build: .
    command: ["bun", "run", "src/worker.ts"]
    
    environment:
      # Enable queue mode
      - USE_JOB_QUEUE=true
      
      # External Redis
      - REDIS_URL=${REDIS_URL}
      
      # Database
      - SUPABASE_URL=${SUPABASE_URL}
      - SUPABASE_ANON_KEY=${SUPABASE_ANON_KEY}
      
      # LLM API Keys
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - GOOGLE_API_KEY=${GOOGLE_API_KEY}
      
      # Worker concurrency
      - CHAT_QUEUE_CONCURRENCY=${CHAT_QUEUE_CONCURRENCY:-5}
      - DEEP_RESEARCH_QUEUE_CONCURRENCY=${DEEP_RESEARCH_QUEUE_CONCURRENCY:-3}
      
      # Production
      - NODE_ENV=production
    
    restart: unless-stopped
    
    # Allow long-running jobs to complete
    stop_grace_period: 8h
    
    # Resource limits
    deploy:
      resources:
        limits:
          memory: 2G
        reservations:
          memory: 512M

import redis
import subprocess
import time

REDIS_URL = "redis://your-redis-host:6379"
MIN_WORKERS = 2
MAX_WORKERS = 10
SCALE_UP_THRESHOLD = 20
SCALE_DOWN_THRESHOLD = 5

def get_queue_depth():
    r = redis.from_url(REDIS_URL)
    chat_waiting = r.llen("bull:chat:waiting")
    research_waiting = r.llen("bull:deep-research:waiting")
    return chat_waiting + research_waiting

def get_current_workers():
    result = subprocess.run(
        ["docker-compose", "-f", "docker-compose.worker.yml", "ps", "-q"],
        capture_output=True,
        text=True
    )
    return len(result.stdout.strip().split("\n"))

def scale_workers(count):
    count = max(MIN_WORKERS, min(MAX_WORKERS, count))
    subprocess.run([
        "docker-compose", "-f", "docker-compose.worker.yml",
        "up", "-d", "--scale", f"worker={count}"
    ])
    print(f"Scaled to {count} workers")

while True:
    depth = get_queue_depth()
    current = get_current_workers()
    
    if depth > SCALE_UP_THRESHOLD:
        scale_workers(current + 2)
    elif depth < SCALE_DOWN_THRESHOLD and current > MIN_WORKERS:
        scale_workers(current - 1)
    
    time.sleep(60)  # Check every minute

const client = require('prom-client');
const { getChatQueue, getDeepResearchQueue } = require('./queue/queues');

const queueDepthGauge = new client.Gauge({
  name: 'bioagents_queue_depth',
  help: 'Number of jobs waiting in queue',
  labelNames: ['queue', 'state']
});

async function updateMetrics() {
  const chatQueue = getChatQueue();
  const researchQueue = getDeepResearchQueue();
  
  const chatCounts = await chatQueue.getJobCounts();
  const researchCounts = await researchQueue.getJobCounts();
  
  queueDepthGauge.set({ queue: 'chat', state: 'waiting' }, chatCounts.waiting);
  queueDepthGauge.set({ queue: 'chat', state: 'active' }, chatCounts.active);
  queueDepthGauge.set({ queue: 'deep-research', state: 'waiting' }, researchCounts.waiting);
  queueDepthGauge.set({ queue: 'deep-research', state: 'active' }, researchCounts.active);
}

setInterval(updateMetrics, 10000); // Every 10 seconds