⚡ Fluxo Padrao de Execucao

// Pipeline de 4 fases
async function pipeline(rawInput: RawMessage): Promise<void> {
  const requestId = crypto.randomUUID();
  const start = Date.now();

  // FASE 1: Entrada
  const validated = validateInput(rawInput);
  const user = await authenticate(validated.chatId);
  await checkRateLimit(user.id);
  log.info({ requestId, chatId: validated.chatId }, 'pipeline.entrada');

  // FASE 2: Decisao
  const intent = classifyIntent(validated.text);
  const agent = selectAgent(intent);
  const context = await buildContext({
    history: await getHistory(validated.chatId, 20),
    systemPrompt: agent.systemPrompt,
    userPrefs: user.preferences,
  });
  log.info({ requestId, agent: agent.id, intent }, 'pipeline.decisao');

  // FASE 3: Execucao
  const result = await executeWithFallback(
    () => agent.run(validated.text, context),
    () => fallbackAgent.run(validated.text, context)
  );
  log.info({ requestId, tokens: result.usage }, 'pipeline.execucao');

  // FASE 4: Saida
  const formatted = formatForChannel(result.text, 'telegram');
  await trackMetrics({ requestId, agent: agent.id, ...result.usage,
    latencyMs: Date.now() - start });
  await saveToHistory(validated.chatId, validated.text, formatted);
  await sendMessage(validated.chatId, formatted);
  log.info({ requestId, latencyMs: Date.now() - start }, 'pipeline.saida');
}

// Middleware pattern - Express style
type Middleware = (ctx: Context, next: () => Promise<void>) => Promise<void>;

// Pre-processamento
const authMiddleware: Middleware = async (ctx, next) => {
  if (!ALLOWED_IDS.includes(ctx.chatId)) {
    ctx.reply('Acesso negado.');
    return; // Para o pipeline
  }
  await next(); // Continua
};

const rateLimitMiddleware: Middleware = async (ctx, next) => {
  const count = await redis.incr(`rate:${ctx.chatId}`);
  await redis.expire(`rate:${ctx.chatId}`, 60);
  if (count > 30) {
    ctx.reply('Rate limit. Aguarde 1 minuto.');
    return;
  }
  await next();
};

const loggingMiddleware: Middleware = async (ctx, next) => {
  const start = Date.now();
  log.info({ requestId: ctx.id, input: ctx.text }, 'request.start');
  await next();
  log.info({ requestId: ctx.id, ms: Date.now() - start }, 'request.end');
};

// Pos-processamento
const formattingMiddleware: Middleware = async (ctx, next) => {
  await next();
  if (ctx.response && ctx.response.length > 4096) {
    ctx.response = ctx.response.slice(0, 4090) + '\n[...]';
  }
};

// Compor a cadeia
const pipeline = compose([
  loggingMiddleware,
  authMiddleware,
  rateLimitMiddleware,
  formattingMiddleware,
  agentHandler,
]);

// Error taxonomy
class AppError extends Error {
  constructor(
    message: string,
    public type: 'user' | 'system' | 'ai',
    public statusCode: number,
    public retryable: boolean = false,
    public context?: Record<string, unknown>
  ) {
    super(message);
  }
}

// Erro de usuario - nao retentar, informar
throw new AppError('Mensagem vazia.', 'user', 400, false);

// Erro de IA - retentar com backoff
throw new AppError('OpenAI 429', 'ai', 429, true, { provider: 'openai' });

// Erro de sistema - logar, alertar, fallback
throw new AppError('DB connection lost', 'system', 500, true);

// Handler centralizado
async function handleError(error: AppError, ctx: Context) {
  log.error({
    requestId: ctx.id,
    type: error.type,
    message: error.message,
    context: error.context,
    stack: error.type === 'system' ? error.stack : undefined,
  });

  switch (error.type) {
    case 'user':
      await ctx.reply(error.message); // Mensagem clara pro usuario
      break;
    case 'ai':
      if (error.retryable) throw error; // Deixa o retry handler cuidar
      await ctx.reply('O servico de IA esta temporariamente indisponivel.');
      break;
    case 'system':
      await ctx.reply('Erro interno. Tente novamente em instantes.');
      await alertAdmin(error); // Notifica admin
      break;
  }
}

// Retry com exponential backoff + jitter
async function retryWithBackoff<T>(
  fn: () => Promise<T>,
  opts: { maxRetries: number; baseDelay: number; maxDelay: number }
): Promise<T> {
  for (let attempt = 0; attempt <= opts.maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      if (attempt === opts.maxRetries) throw error;
      const delay = Math.min(
        opts.baseDelay * Math.pow(2, attempt) + Math.random() * 1000,
        opts.maxDelay
      );
      log.warn({ attempt, delay, error: error.message }, 'retry.backoff');
      await sleep(delay);
    }
  }
  throw new Error('Unreachable');
}

// Circuit Breaker
class CircuitBreaker {
  private state: 'closed' | 'open' | 'half-open' = 'closed';
  private failures = 0;
  private lastFailure = 0;

  constructor(
    private threshold: number = 5,
    private resetTimeout: number = 30_000
  ) {}

  async execute<T>(fn: () => Promise<T>, fallback: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailure > this.resetTimeout) {
        this.state = 'half-open';
      } else {
        log.info('circuit.open - using fallback');
        return fallback();
      }
    }
    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      if (this.state === 'open') return fallback();
      throw error;
    }
  }

  private onSuccess() { this.failures = 0; this.state = 'closed'; }
  private onFailure() {
    this.failures++;
    this.lastFailure = Date.now();
    if (this.failures >= this.threshold) this.state = 'open';
  }
}

// Uso combinado
const openaiBreaker = new CircuitBreaker(5, 30_000);
const response = await openaiBreaker.execute(
  () => retryWithBackoff(() => openai.chat(prompt), {
    maxRetries: 3, baseDelay: 1000, maxDelay: 10_000
  }),
  () => ollamaFallback.chat(prompt)
);

// Streaming com OpenAI SDK
import OpenAI from 'openai';

async function streamResponse(prompt: string, onChunk: (text: string) => void) {
  const openai = new OpenAI();
  const stream = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: prompt }],
    stream: true,
  });

  let fullText = '';
  for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta?.content || '';
    fullText += delta;
    onChunk(fullText); // Callback com texto acumulado
  }
  return fullText;
}

// Streaming para Telegram (atualiza mesma mensagem)
async function streamToTelegram(chatId: number, prompt: string) {
  let messageId: number | null = null;
  let lastUpdate = 0;

  await streamResponse(prompt, async (text) => {
    const now = Date.now();
    if (now - lastUpdate < 1500) return; // Throttle: 1.5s entre updates
    lastUpdate = now;

    if (!messageId) {
      const msg = await bot.sendMessage(chatId, text + ' ●');
      messageId = msg.message_id;
    } else {
      await bot.editMessageText(text + ' ●', {
        chat_id: chatId,
        message_id: messageId,
      }).catch(() => {}); // Ignora erro se texto nao mudou
    }
  });

  // Update final sem o indicador de typing
  if (messageId) {
    await bot.editMessageText(fullText, {
      chat_id: chatId,
      message_id: messageId,
    });
  }
}

// SSE para interface web (Next.js API route)
export async function GET(req: Request) {
  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    async start(controller) {
      await streamResponse(prompt, (text) => {
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ text })}\n\n`)
        );
      });
      controller.enqueue(encoder.encode('data: [DONE]\n\n'));
      controller.close();
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

# Estrutura do exercicio
pipeline/
├── pipeline.ts         # Funcao principal que orquestra as 4 fases
├── middleware.ts        # auth, rateLimit, logging middlewares
├── errors.ts           # AppError class + error handler
├── resilience.ts       # retryWithBackoff + CircuitBreaker
├── stream.ts           # Stream consumer + throttled output
└── test.ts             # Testes simulando falhas e sucesso

# Testes para validar
1. Request valido: pipeline completo executa com sucesso, logs corretos
2. Auth falha: middleware bloqueia, usuario recebe "Acesso negado"
3. Rate limit: 21o request em 1 min retorna "Limite atingido"
4. Provider 429: retry com backoff, sucesso na 2a tentativa
5. Provider down: circuit breaker abre, fallback ativado
6. Stream: resposta chega progressivamente, historico salvo completo