LeadFlow — Arquitectura CQRS

from dataclasses import dataclass, field
from datetime import datetime
import uuid

@dataclass
class Command:
    command_id: str = field(
        default_factory=lambda: str(uuid.uuid4())
    )
    timestamp: datetime = field(
        default_factory=datetime.utcnow
    )

@dataclass
class CaptureLeadCommand(Command):
    nombre: str = ""
    empresa: str = ""
    email: str = ""
    fuente: str = ""  # "web" | "linkedin" | "referido"

@dataclass
class AssignLeadCommand(Command):
    lead_id: str = ""
    comercial_id: str = ""

@dataclass
class QualifyLeadCommand(Command):
    lead_id: str = ""
    puntuacion: int = 0   # 0–100
    etapa: str = ""        # "calificado" | "propuesta"

@dataclass
class CloseLeadCommand(Command):
    lead_id: str = ""
    resultado: str = ""  # "ganado" | "perdido"
    motivo: str = ""

from abc import ABC, abstractmethod

class CommandHandler(ABC):
    @abstractmethod
    async def handle(self, command): ...

class CaptureLeadHandler(CommandHandler):
    def __init__(self, repo, event_store):
        self.repo = repo
        self.event_store = event_store

    async def handle(self, cmd: CaptureLeadCommand) -> str:
        # 1. Validar
        if not cmd.email:
            raise ValueError("Email requerido")
        if await self.repo.email_exists(cmd.email):
            raise ValueError("Lead duplicado")

        # 2. Crear agregado
        lead = Lead.create(
            nombre=cmd.nombre,
            empresa=cmd.empresa,
            email=cmd.email,
            fuente=cmd.fuente,
        )

        # 3. Persistir eventos
        await self.event_store.append(
            stream_id=f"Lead-{lead.id}",
            events=lead.uncommitted_events
        )

        # 4. Retornar ID
        return lead.id


class CommandBus:
    def __init__(self):
        self._map = {}

    def register(self, cmd_cls, handler):
        self._map[cmd_cls] = handler

    async def dispatch(self, command):
        h = self._map.get(type(command))
        if not h:
            raise ValueError("Sin handler")
        return await h.handle(command)

from dataclasses import dataclass
from typing import Optional, List

@dataclass
class GetLeadDashboardQuery:
    """Resumen ejecutivo para el manager."""
    periodo_dias: int = 30

@dataclass
class GetLeadsByComercialQuery:
    comercial_id: str
    etapa: Optional[str] = None
    page: int = 1
    page_size: int = 20

@dataclass
class GetConversionFunnelQuery:
    """Embudos de conversión por etapa."""
    equipo_id: Optional[str] = None
    periodo_dias: int = 30

# ──── DTOs de resultado ────
@dataclass
class DashboardView:
    leads_nuevos: int
    leads_calificados: int
    leads_cerrados_ganados: int
    tasa_conversion: float
    valor_pipeline: float

@dataclass
class LeadView:
    lead_id: str
    nombre: str
    empresa: str
    etapa: str
    puntuacion: int
    comercial: str
    dias_en_etapa: int

@dataclass
class FunnelStageView:
    etapa: str
    total: int
    conversion_pct: float
    tiempo_medio_dias: float

class LeadDashboardHandler:
    def __init__(self, read_db):
        self.db = read_db

    async def handle(
        self, q: GetLeadDashboardQuery
    ) -> DashboardView:
        # SELECT directo sobre vista desnormalizada
        # — sin tocar el write model —
        row = await self.db.fetchrow("""
            SELECT
                COUNT(*) FILTER (
                    WHERE etapa = 'nuevo'
                    AND created_at > now() - $1::interval
                ) AS nuevos,
                COUNT(*) FILTER (
                    WHERE etapa = 'calificado'
                ) AS calificados,
                COUNT(*) FILTER (
                    WHERE resultado = 'ganado'
                    AND closed_at > now() - $1::interval
                ) AS ganados,
                SUM(valor_estimado) AS pipeline
            FROM lead_views
        """, f"{q.periodo_dias} days")
        total = row['nuevos'] + row['calificados']
        tasa = (
            row['ganados'] / total * 100
            if total > 0 else 0.0
        )
        return DashboardView(
            leads_nuevos=row['nuevos'],
            leads_calificados=row['calificados'],
            leads_cerrados_ganados=row['ganados'],
            tasa_conversion=round(tasa, 1),
            valor_pipeline=row['pipeline'] or 0.0
        )

-- Agregados y eventos; escrituras rápidas
CREATE TABLE leads (
    id          UUID PRIMARY KEY,
    email       TEXT UNIQUE NOT NULL,
    estado      TEXT NOT NULL DEFAULT 'nuevo',
    version     INT NOT NULL DEFAULT 1,
    created_at  TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE events (
    global_pos  BIGSERIAL PRIMARY KEY,
    stream_id   TEXT NOT NULL,       -- "Lead-{uuid}"
    event_type  TEXT NOT NULL,       -- "LeadCaptured"
    payload     JSONB NOT NULL,
    version     INT NOT NULL,
    occurred_at TIMESTAMPTZ DEFAULT now(),
    UNIQUE (stream_id, version)      -- optimistic lock
);

CREATE INDEX ON events(stream_id);
CREATE INDEX ON events(global_pos);

-- Vista materializada para queries rápidos
CREATE TABLE lead_views (
    lead_id           UUID PRIMARY KEY,
    nombre            TEXT,
    empresa           TEXT,
    email             TEXT,
    fuente            TEXT,
    etapa             TEXT,          -- pipeline actual
    puntuacion        SMALLINT,
    comercial_nombre  TEXT,          -- desnormalizado
    comercial_equipo  TEXT,          -- desnormalizado
    valor_estimado    NUMERIC(12,2),
    resultado         TEXT,          -- "ganado"|"perdido"|NULL
    dias_en_etapa     INT,
    created_at        TIMESTAMPTZ,
    updated_at        TIMESTAMPTZ,
    last_event_ver    INT NOT NULL DEFAULT 0
);

-- Índices para los filtros más comunes
CREATE INDEX ON lead_views(comercial_nombre, etapa);
CREATE INDEX ON lead_views(etapa, created_at DESC);
CREATE INDEX ON lead_views(puntuacion DESC);

Hacer

Evitar