Orquestación de Workflows: NutriLoop Onboarding

from temporalio import workflow
from temporalio.common import RetryPolicy
from dataclasses import dataclass
from datetime import timedelta

# ── Tipos de entrada/salida ──────────────────────────────────────────
@dataclass
class OnboardingInput:
    org_id: str
    plan: str
    seats: int
    admin_email: str
    hubspot_deal_id: str

@workflow.defn
class NutriLoopOnboardingWorkflow:
    """
    Orquesta el onboarding B2B completo.
    Implementa Saga con compensaciones LIFO para rollback atómico.
    """

    @workflow.run
    async def run(self, inp: OnboardingInput) -> dict:
        compensations: list = []

        # ── Compliance gate: aprobación humana si > 1.000 asientos ─────────
        if inp.seats > 1000:
            # Bloquea hasta recibir la señal 'compliance_approved'
            await workflow.wait_condition(
                lambda: self._compliance_ok,
                timeout=timedelta(hours=48)          # escalado automático si no hay respuesta
            )

        try:
            # ── Paso 1: Stripe ──────────────────────────────────────────────────
            sub = await workflow.execute_activity(
                create_stripe_subscription,
                args=[inp.org_id, inp.plan, inp.seats],
                retry_policy=RetryPolicy(
                    maximum_attempts=3,
                    backoff_coefficient=2.0,
                    non_retryable_error_types=["CardDeclinedError"]
                ),
                start_to_close_timeout=timedelta(seconds=30)
            )
            compensations.append((cancel_stripe_subscription, sub["subscription_id"]))

            # ── Paso 2: Provisionar workspace ───────────────────────────────────
            ws = await workflow.execute_activity(
                create_workspace,
                args=[inp.org_id, inp.seats, sub["subscription_id"]],
                retry_policy=RetryPolicy(maximum_attempts=5),
                start_to_close_timeout=timedelta(minutes=2)
            )
            compensations.append((delete_workspace, ws["workspace_id"]))

            # ── Paso 3: Fan-out — HubSpot + Slack + Email en paralelo ───────────
            slack_h, email_h, crm_h = await asyncio.gather(
                workflow.execute_activity(invite_to_slack, args=[inp.admin_email, ws["channel"]]),
                workflow.execute_activity(send_welcome_email, args=[inp.admin_email]),
                workflow.execute_activity(sync_hubspot, args=[inp.hubspot_deal_id, ws]),
            )

            return {"status": "success", "workspace_id": ws["workspace_id"]}

        except Exception as err:
            # ── Rollback LIFO ───────────────────────────────────────────────────
            for fn, arg in reversed(compensations):
                await workflow.execute_activity(fn, args=[arg])
            raise

    @workflow.signal
    def compliance_approved(self, approved_by: str):
        self._compliance_ok = True

    @workflow.query
    def current_state(self) -> str:
        return self._state

from temporalio import activity
import stripe

@activity.defn
async def create_stripe_subscription(org_id: str, plan: str, seats: int) -> dict:
    """
    Actividad idempotente: usa org_id como idempotency_key.
    Si Stripe ya procesó la request, devuelve el objeto existente.
    """
    activity.heartbeat("Creando suscripción Stripe...")         # heartbeat para detección de stall

    try:
        sub = stripe.Subscription.create(
            customer=org_id,
            items=[{"price": price_id_for(plan, seats)}],
            idempotency_key=f"onboard-{org_id}"           # garantía de idempotencia
        )
        return {"subscription_id": sub.id, "status": sub.status}

    except stripe.error.CardError as e:
        # Error NO retryable — Temporal no volverá a intentarlo
        raise CardDeclinedError(str(e)) from e

    except stripe.error.APIConnectionError as e:
        # Error transitorio — Temporal reintentará con backoff exponencial
        raise                                               # re-raise para retry automático

@activity.defn
async def cancel_stripe_subscription(subscription_id: str) -> None:
    """Compensación idempotente: cancelar está bien llamar N veces."""
    try:
        stripe.Subscription.delete(subscription_id)
    except stripe.error.InvalidRequestError:
        pass                                                # ya cancelada → ok