Workshop 2: Production Patterns for AI Integrations

import {
  ElevenLabsProvider,
  RunComfyProvider,
  CivitaiProvider,
} from '@oshun/ai-providers';

// Initialize all providers
const elevenlabs = new ElevenLabsProvider({
  apiKey: process.env.ELEVENLABS_API_KEY!,
  maxRetries: 3,
});

const runcomfy = new RunComfyProvider({
  apiKey: process.env.RUNCOMFY_API_KEY!,
  maxConcurrentJobs: 5,
});

const civitai = new CivitaiProvider({
  apiKey: process.env.CIVITAI_API_KEY,
});

[Wait 5s for full audio] → [Play audio]

[Start playing immediately] → [Continue as chunks arrive]

// Start a streaming session
const session = await elevenlabs.startStreamingSession({
  voiceId: 'EXAVITQu4vr4xnSDxMaL',
  modelId: 'eleven_flash_v2_5', // Optimized for streaming
  optimizeStreamingLatency: 4,   // Maximum optimization
});

console.log(`Session started: ${session.sessionId}`);

// Collect audio chunks
const chunks: Buffer[] = [];

session.on('audio', (chunk: Buffer) => {
  chunks.push(chunk);
  console.log(`Received chunk: ${chunk.length} bytes`);
});

session.on('end', () => {
  const totalAudio = Buffer.concat(chunks);
  console.log(`Total audio: ${totalAudio.length} bytes`);
});

// Send text
await session.sendText('Hello! This is streaming synthesis.');
await session.sendText('Each sentence is processed separately.');
await session.flush();
await session.close();

async function interactiveAssistant(
  responses: string[]
): Promise<void> {
  const session = await elevenlabs.startStreamingSession({
    voiceId: 'EXAVITQu4vr4xnSDxMaL',
    modelId: 'eleven_flash_v2_5',
    optimizeStreamingLatency: 4,
  });

  for (const response of responses) {
    console.log(`Speaking: "${response}"`);

    // Send and wait for this response to complete
    await session.sendText(response);
    await session.flush();

    // Simulate thinking time
    await new Promise(r => setTimeout(r, 500));
  }

  await session.close();
}

// Usage
await interactiveAssistant([
  'Welcome to our AI assistant.',
  'How can I help you today?',
  'I can answer questions about anything.',
]);

class AudioBuffer {
  private chunks: Buffer[] = [];
  private totalBytes = 0;

  addChunk(chunk: Buffer): void {
    this.chunks.push(chunk);
    this.totalBytes += chunk.length;
  }

  getBuffer(): Buffer {
    return Buffer.concat(this.chunks);
  }

  getTotalBytes(): number {
    return this.totalBytes;
  }

  // Estimate duration (44100Hz, 16-bit, mono)
  getEstimatedDuration(): number {
    return this.totalBytes / (44100 * 2);
  }

  clear(): void {
    this.chunks = [];
    this.totalBytes = 0;
  }
}

// Use it with streaming
const buffer = new AudioBuffer();

session.on('audio', (chunk) => {
  buffer.addChunk(chunk);
  console.log(`Duration so far: ${buffer.getEstimatedDuration().toFixed(2)}s`);
});

┌───────────────────────────────────────────────┐
│  Items: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]       │
│                                               │
│  Concurrency = 3:                             │
│  ┌─────┐ ┌─────┐ ┌─────┐                     │
│  │  1  │ │  2  │ │  3  │ ← Processing         │
│  └─────┘ └─────┘ └─────┘                     │
│  [4, 5, 6, 7, 8, 9, 10] ← Waiting            │
└───────────────────────────────────────────────┘

async function processBatch<T, R>(
  items: T[],
  processor: (item: T) => Promise<R>,
  concurrency: number
): Promise<R[]> {
  const results: R[] = [];

  for (let i = 0; i < items.length; i += concurrency) {
    const batch = items.slice(i, i + concurrency);
    const batchResults = await Promise.all(
      batch.map(item => processor(item))
    );
    results.push(...batchResults);
    console.log(`Processed ${results.length}/${items.length}`);
  }

  return results;
}

// Usage: Generate multiple images
const prompts = [
  'A red dragon',
  'A blue phoenix',
  'A green griffin',
  'A golden unicorn',
];

const results = await processBatch(
  prompts,
  async (prompt) => {
    const job = await runcomfy.executeWorkflow({
      workflowId: 'txt2img-sdxl-v1',
      inputs: { prompt, width: 1024, height: 1024 },
    });
    return runcomfy.waitForCompletion(job.id);
  },
  2 // Process 2 at a time
);

async function rateLimitedProcess<T, R>(
  items: T[],
  processor: (item: T) => Promise<R>,
  requestsPerSecond: number
): Promise<R[]> {
  const results: R[] = [];
  const delayMs = 1000 / requestsPerSecond;

  for (const item of items) {
    const start = Date.now();
    results.push(await processor(item));

    // Maintain rate limit
    const elapsed = Date.now() - start;
    if (elapsed < delayMs) {
      await new Promise(r => setTimeout(r, delayMs - elapsed));
    }
  }

  return results;
}

async function advancedBatch<T, R>(
  items: T[],
  processor: (item: T) => Promise<R>,
  options: {
    concurrency: number;
    signal?: AbortSignal;
    onProgress?: (completed: number, total: number) => void;
  }
): Promise<Array<{ item: T; result?: R; error?: Error }>> {
  const results: Array<{ item: T; result?: R; error?: Error }> = [];
  let completed = 0;

  for (let i = 0; i < items.length; i += options.concurrency) {
    // Check for cancellation
    if (options.signal?.aborted) {
      for (let j = i; j < items.length; j++) {
        results.push({ item: items[j], error: new Error('Cancelled') });
      }
      break;
    }

    const batch = items.slice(i, i + options.concurrency);

    const batchResults = await Promise.all(
      batch.map(async (item) => {
        try {
          const result = await processor(item);
          return { item, result };
        } catch (error) {
          return { item, error: error as Error };
        }
      })
    );

    results.push(...batchResults);
    completed += batch.length;
    options.onProgress?.(completed, items.length);
  }

  return results;
}

// Usage with AbortController
const controller = new AbortController();

// Cancel after 30 seconds
setTimeout(() => controller.abort(), 30000);

const results = await advancedBatch(
  prompts,
  processor,
  {
    concurrency: 3,
    signal: controller.signal,
    onProgress: (done, total) => {
      console.log(`Progress: ${done}/${total} (${(done/total*100).toFixed(1)}%)`);
    },
  }
);

async function retryWithBackoff<T>(
  operation: () => Promise<T>,
  options: {
    maxRetries: number;
    initialDelayMs: number;
    maxDelayMs: number;
  }
): Promise<T> {
  let delay = options.initialDelayMs;
  let lastError: Error | undefined;

  for (let attempt = 0; attempt <= options.maxRetries; attempt++) {
    try {
      return await operation();
    } catch (error) {
      lastError = error as Error;

      // Check if retryable
      const retryable = (error as any).retryable !== false;
      if (!retryable || attempt === options.maxRetries) {
        throw lastError;
      }

      // Add jitter (0-25%)
      const jitter = delay * Math.random() * 0.25;
      const waitTime = Math.min(delay + jitter, options.maxDelayMs);

      console.log(`Attempt ${attempt + 1} failed. Retrying in ${waitTime.toFixed(0)}ms...`);
      await new Promise(r => setTimeout(r, waitTime));

      delay *= 2; // Exponential
    }
  }

  throw lastError!;
}

// Usage
const result = await retryWithBackoff(
  () => elevenlabs.synthesize({ text: 'Hello', voiceId: '...' }),
  { maxRetries: 3, initialDelayMs: 1000, maxDelayMs: 30000 }
);

class CircuitBreaker {
  private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
  private failures = 0;
  private lastFailure = 0;

  constructor(
    private readonly threshold: number = 5,
    private readonly resetTimeout: number = 30000
  ) {}

  async execute<T>(operation: () => Promise<T>): Promise<T> {
    this.updateState();

    if (this.state === 'OPEN') {
      throw new Error('Circuit breaker is OPEN');
    }

    try {
      const result = await operation();
      this.recordSuccess();
      return result;
    } catch (error) {
      this.recordFailure();
      throw error;
    }
  }

  private updateState(): void {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailure >= this.resetTimeout) {
        this.state = 'HALF_OPEN';
      }
    }
  }

  private recordSuccess(): void {
    this.failures = 0;
    this.state = 'CLOSED';
  }

  private recordFailure(): void {
    this.failures++;
    this.lastFailure = Date.now();
    if (this.failures >= this.threshold) {
      this.state = 'OPEN';
    }
  }

  getState(): string {
    this.updateState();
    return this.state;
  }
}

// Usage
const breaker = new CircuitBreaker(5, 30000);

try {
  const result = await breaker.execute(() =>
    elevenlabs.synthesize({ text: 'Test', voiceId: '...' })
  );
} catch (error) {
  if (error.message === 'Circuit breaker is OPEN') {
    console.log('Service is temporarily unavailable');
  }
}

async function withFallback<T>(
  primary: () => Promise<T>,
  fallbacks: Array<() => Promise<T>>,
  options?: { timeout?: number }
): Promise<T> {
  const providers = [primary, ...fallbacks];
  const errors: Error[] = [];

  for (const provider of providers) {
    try {
      if (options?.timeout) {
        const result = await Promise.race([
          provider(),
          new Promise<never>((_, reject) =>
            setTimeout(() => reject(new Error('Timeout')), options.timeout)
          ),
        ]);
        return result;
      }
      return await provider();
    } catch (error) {
      errors.push(error as Error);
    }
  }

  throw new Error(`All providers failed: ${errors.map(e => e.message).join(', ')}`);
}

// Usage: Multiple providers for redundancy
const audio = await withFallback(
  () => provider1.synthesize({ text, voiceId }),
  [
    () => provider2.synthesize({ text, voiceId }),
    () => localTTS.synthesize({ text }), // Local fallback
  ],
  { timeout: 10000 }
);

┌─────────────────────────────────────────┐
│           Connection Pool               │
│  ┌───┐ ┌───┐ ┌───┐ ┌───┐ ┌───┐        │
│  │ 1 │ │ 2 │ │ 3 │ │ 4 │ │ 5 │        │
│  └─┬─┘ └─┬─┘ └─┬─┘ └───┘ └───┘        │
│    │     │     │   (available)         │
│    ▼     ▼     ▼                       │
│  Task   Task  Task (in use)            │
└─────────────────────────────────────────┘

class ConnectionPool<T extends { shutdown: () => Promise<void> }> {
  private available: T[] = [];
  private inUse = new Set<T>();
  private waiting: Array<(conn: T) => void> = [];

  constructor(
    private factory: () => Promise<T>,
    private maxSize: number
  ) {}

  async acquire(): Promise<T> {
    // Return available connection
    if (this.available.length > 0) {
      const conn = this.available.pop()!;
      this.inUse.add(conn);
      return conn;
    }

    // Create new if under limit
    if (this.inUse.size < this.maxSize) {
      const conn = await this.factory();
      this.inUse.add(conn);
      return conn;
    }

    // Wait for available
    return new Promise(resolve => {
      this.waiting.push(resolve);
    });
  }

  release(conn: T): void {
    this.inUse.delete(conn);

    if (this.waiting.length > 0) {
      const waiter = this.waiting.shift()!;
      this.inUse.add(conn);
      waiter(conn);
    } else {
      this.available.push(conn);
    }
  }

  async shutdown(): Promise<void> {
    const all = [...this.available, ...this.inUse];
    await Promise.all(all.map(c => c.shutdown()));
    this.available = [];
    this.inUse.clear();
  }
}

class CleanupManager {
  private resources: Array<{
    name: string;
    cleanup: () => Promise<void>;
  }> = [];

  register(name: string, cleanup: () => Promise<void>): void {
    this.resources.push({ name, cleanup });
  }

  async cleanup(): Promise<void> {
    // Cleanup in reverse order (LIFO)
    const reversed = [...this.resources].reverse();

    for (const { name, cleanup } of reversed) {
      try {
        await cleanup();
        console.log(`Cleaned up: ${name}`);
      } catch (error) {
        console.error(`Failed to cleanup ${name}:`, error);
      }
    }

    this.resources = [];
  }
}

// Usage
const manager = new CleanupManager();

manager.register('ElevenLabs', () => elevenlabs.shutdown());
manager.register('RunComfy', () => runcomfy.shutdown());

// Cleanup on exit
process.on('SIGINT', async () => {
  console.log('Shutting down...');
  await manager.cleanup();
  process.exit(0);
});

class MetricsCollector {
  private requests: Array<{
    timestamp: number;
    endpoint: string;
    success: boolean;
    latencyMs: number;
  }> = [];

  record(endpoint: string, success: boolean, latencyMs: number): void {
    this.requests.push({
      timestamp: Date.now(),
      endpoint,
      success,
      latencyMs,
    });
  }

  getStats(windowMs = 60000): {
    total: number;
    success: number;
    failure: number;
    avgLatency: number;
    p95Latency: number;
  } {
    const cutoff = Date.now() - windowMs;
    const recent = this.requests.filter(r => r.timestamp >= cutoff);

    if (recent.length === 0) {
      return { total: 0, success: 0, failure: 0, avgLatency: 0, p95Latency: 0 };
    }

    const latencies = recent.map(r => r.latencyMs).sort((a, b) => a - b);

    return {
      total: recent.length,
      success: recent.filter(r => r.success).length,
      failure: recent.filter(r => !r.success).length,
      avgLatency: latencies.reduce((a, b) => a + b, 0) / latencies.length,
      p95Latency: latencies[Math.floor(latencies.length * 0.95)],
    };
  }
}

// Usage
const metrics = new MetricsCollector();

// Wrap operations
async function trackedSynthesize(text: string, voiceId: string) {
  const start = Date.now();
  try {
    const result = await elevenlabs.synthesize({ text, voiceId });
    metrics.record('synthesize', true, Date.now() - start);
    return result;
  } catch (error) {
    metrics.record('synthesize', false, Date.now() - start);
    throw error;
  }
}

// Report stats
setInterval(() => {
  const stats = metrics.getStats();
  console.log(`Stats: ${stats.total} requests, ${stats.avgLatency.toFixed(0)}ms avg`);
}, 10000);