CultivaFlow — Arquitectura de Microservicios

Diagrama de arquitectura

CultivaFlow — Vista de servicios

Capa cliente

🌐

React SPA / Mobile App

Clientes web y movil de las agencias

↓

HTTPS / REST

API Gateway

🔀

API Gateway — cultivaflow-gateway:8080

Rate limiting · Auth middleware · Request routing · Circuit breaker global · Aggregation

Gateway

↓

HTTP interno (gRPC planificado)

Servicios de dominio

Core

🔐

AuthService

:8001 — JWT, OAuth, orgs

Core

📢

CampaignService

:8002 — CRUD campanas

Scale

🤖

AIService

:8003 — LLM, prompts, tokens

New

📊

AnalyticsService

:8004 — metricas, clicks

🔔

NotificationService

:8005 — email, webhook

↓↑

eventos asincronos

⚡

Apache Kafka — Event Bus

Comunicacion asincrona entre servicios — persistencia de eventos — replay

campaign.created ai.content.ready analytics.tracked notification.send

Bases de datos (Database-per-service)

🗄️

PostgreSQL

auth_db

🗄️

PostgreSQL

campaign_db

⚡

Redis

ai_cache + queue

🔷

ClickHouse

analytics_db

🗄️

PostgreSQL

notif_db

Flujo Saga — crear campana

📋 OrderFulfillment Pattern — transaccion distribuida atomica con compensacion

Paso 1

Validar Auth

AuthService

→

Paso 2

Crear Campana

CampaignService

→

Paso 3

Generar Copy

AIService

→

Paso 4

Registrar Evento

AnalyticsService

→

Paso 5

Notificar

NotificationService

Compensacion: si AIService falla en paso 3, se ejecuta en orden inverso: cancel_campaign (paso 2), revoke_auth_token (paso 1). Garantia de consistencia eventual.

Patrones clave implementados

Strangler Fig Pattern

Migracion gradual desde monolito Django

Contexto CultivaFlow: El monolito Django sigue sirviendo /admin y /legacy. El proxy enruta /api/v2/campaigns al nuevo CampaignService mientras lo demas permanece en el monolito.

Python
strangler_proxy.py
class StranglerProxy:
    routes = {
        "/api/v2/campaigns": "http://campaign-svc:8002",
        "/api/v2/auth":      "http://auth-svc:8001",
        "/api/v2/ai":        "http://ai-svc:8003",
        "/legacy":           "http://django-monolith:8000",
    }
    async def route(self, request):
        for prefix, target in self.routes.items():
            if request.path.startswith(prefix):
                return await self.proxy(target, request)
        return await self.proxy("http://django-monolith:8000", request)

Circuit Breaker

Proteger CampaignService de fallos de AIService

Contexto CultivaFlow: Si Claude API esta caido, las campanas siguen creandose con copy pendiente (DRAFT). El Circuit Breaker abre tras 5 errores y espera 30s antes de reintentar.

Python
ai_circuit_breaker.py
ai_breaker = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=30,
    success_threshold=2
)
async def generate_copy(brief: str) -> str:
    try:
        return await ai_breaker.call(
            ai_service.generate, brief
        )
    except CircuitBreakerOpenError:
        # Fallback: campana en estado DRAFT
        return CopyResult(status="DRAFT", copy=None)

Saga Pattern (Orquestado)

Transacciones distribuidas con compensacion

Contexto CultivaFlow: Crear una campana requiere coordinar 4 servicios. Si el pago del credito AI falla, se deben cancelar los pasos anteriores automaticamente.

Python
create_campaign_saga.py
saga = CreateCampaignSaga([
    SagaStep("validate_auth",
        action=auth_svc.validate,
        compensation=auth_svc.revoke),
    SagaStep("create_campaign",
        action=campaign_svc.create,
        compensation=campaign_svc.cancel),
    SagaStep("deduct_ai_credits",
        action=billing_svc.deduct,
        compensation=billing_svc.refund),
    SagaStep("generate_copy",
        action=ai_svc.generate,
        compensation=ai_svc.discard),
])
result = await saga.execute(campaign_data)

Bulkhead Pattern

Aislar AIService del resto del sistema

Contexto CultivaFlow: El AIService consume mucha CPU/memoria. Con Bulkhead, los picos de uso de generacion AI no afectan a AuthService ni CampaignService.

Python
bulkhead_config.py
# Thread pools separados por servicio
ai_pool = ThreadPoolExecutor(max_workers=10)
auth_pool = ThreadPoolExecutor(max_workers=20)
campaign_pool = ThreadPoolExecutor(max_workers=15)

# Kubernetes resource limits (YAML)
# ai-service:   cpu: "2", memory: "4Gi"
# auth-service: cpu: "500m", memory: "512Mi"
# campaign-svc: cpu: "1", memory: "1Gi"

@bulkhead(pool=ai_pool, max_concurrent=10)
async def generate_content(brief): ...

Configuracion de resiliencia

🔌

Circuit Breaker — AIService

Abre el circuito cuando Claude API devuelve errores repetidos. Modo degradado: campanas en DRAFT sin copy.

failure_threshold: 5 recovery_timeout: 30s success_threshold: 2 fallback: DRAFT_MODE

🔄

Retry con Backoff — HTTP calls

Reintentos exponenciales para errores transientes en llamadas entre servicios via httpx.

max_attempts: 3 min_wait: 2s max_wait: 10s multiplier: exponential

🏊

Bulkhead — recursos aislados

Pools de threads separados evitan que AIService (intensivo en CPU) sature los servicios criticos.

ai_pool: 10 workers auth_pool: 20 workers campaign_pool: 15 workers queue_size: 100 tasks

Matriz de comunicacion entre servicios

Desde → Hasta	Patron	Protocolo	Razon	Estado
Gateway → AuthService	Sincrono	REST / HTTP	Validar JWT en cada request — baja latencia requerida	✓ Activo
Gateway → CampaignService	Sincrono	REST / HTTP	CRUD campanas — respuesta inmediata al usuario	✓ Activo
CampaignService → AIService	Asincrono	Kafka topic	Generacion de copy puede tardar 5-30s — no bloquear UI	✓ Activo
AIService → AnalyticsService	Asincrono	Kafka topic	Registro de tokens usados — no critico para el flujo	✓ Activo
CampaignService → NotificationService	Asincrono	Kafka topic	Notificacion al agente cuando campana esta lista	⚠ Pendiente
Gateway → AnalyticsService	Sincrono	REST / HTTP	Dashboard en tiempo real — datos frescos requeridos	⚠ En migracion