AgentFlow — Diseño del Event Store

-- ============================================================
-- AgentFlow Event Store — PostgreSQL / Supabase
-- Cumplimiento GDPR: anonimizacion sin borrar eventos
-- ============================================================

-- Tabla principal de eventos (inmutable, append-only)
CREATE TABLE agentflow_events (
    id               UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    stream_id        VARCHAR(255) NOT NULL,   -- e.g. "Agent-550e8400"
    stream_type      VARCHAR(64)  NOT NULL,   -- Agent | Campaign | Interaction
    event_type       VARCHAR(128) NOT NULL,   -- AgentCreated, LeadConverted...
    event_data       JSONB        NOT NULL,   -- payload del evento
    metadata         JSONB        DEFAULT '{}', -- correlation_id, user_id, trace_id
    version          BIGINT       NOT NULL,   -- version dentro del stream
    global_position  BIGSERIAL,               -- posicion global (orden total)
    created_at       TIMESTAMPTZ  DEFAULT NOW(),
    -- GDPR: reemplazar PII sin mutar el evento
    anonymized_at    TIMESTAMPTZ  DEFAULT NULL,
    anonymized_by    UUID         DEFAULT NULL,

    CONSTRAINT uq_stream_version UNIQUE (stream_id, version),
    -- Evita doble escritura con el mismo event_id
    CONSTRAINT uq_event_id UNIQUE (id)
);

-- Indices criticos para AgentFlow
CREATE INDEX idx_stream          ON agentflow_events(stream_id, version);
CREATE INDEX idx_global_position ON agentflow_events(global_position);
CREATE INDEX idx_event_type      ON agentflow_events(event_type);
CREATE INDEX idx_stream_type     ON agentflow_events(stream_type, global_position);
CREATE INDEX idx_created_at      ON agentflow_events(created_at);

-- Snapshots — para agentes con muchos eventos (>200)
CREATE TABLE agentflow_snapshots (
    stream_id     VARCHAR(255) PRIMARY KEY,
    stream_type   VARCHAR(64)  NOT NULL,
    snapshot_data JSONB        NOT NULL,
    version       BIGINT       NOT NULL,
    created_at    TIMESTAMPTZ  DEFAULT NOW()
);

-- Checkpoints para subscriptores (projections, read models)
CREATE TABLE agentflow_checkpoints (
    subscription_id VARCHAR(128) PRIMARY KEY,
    last_position   BIGINT       NOT NULL DEFAULT 0,
    updated_at      TIMESTAMPTZ  DEFAULT NOW()
);

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List
from uuid import UUID, uuid4
import json, asyncpg


@dataclass
class AgentFlowEvent:
    stream_id:       str
    event_type:      str                   # AgentCreated, LeadConverted...
    data:            dict
    metadata:        dict    = field(default_factory=dict)
    event_id:        UUID    = field(default_factory=uuid4)
    version:         Optional[int] = None
    global_position: Optional[int] = None
    created_at:      datetime = field(default_factory=datetime.utcnow)


class AgentFlowEventStore:
    """Event Store para AgentFlow. append-only, inmutable, GDPR-ready."""

    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def append(
        self,
        stream_id:        str,              # "Agent-550e8400-e29b..."
        stream_type:      str,              # "Agent" | "Campaign" | "Interaction"
        events:           List[AgentFlowEvent],
        expected_version: Optional[int] = None
    ) -> List[AgentFlowEvent]:
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # Optimistic concurrency check
                if expected_version is not None:
                    current = await conn.fetchval(
                        "SELECT COALESCE(MAX(version),0) FROM agentflow_events WHERE stream_id=$1",
                        stream_id
                    )
                    if current != expected_version:
                        raise ConcurrencyError(
                            f"Stream {stream_id}: expected v{expected_version}, got v{current}"
                        )

                start = await conn.fetchval(
                    "SELECT COALESCE(MAX(version),0)+1 FROM agentflow_events WHERE stream_id=$1",
                    stream_id
                )

                saved = []
                for i, ev in enumerate(events):
                    ev.version = start + i
                    row = await conn.fetchrow(
                        """INSERT INTO agentflow_events
                           (id, stream_id, stream_type, event_type,
                            event_data, metadata, version, created_at)
                           VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
                           RETURNING global_position""",
                        ev.event_id, stream_id, stream_type, ev.event_type,
                        json.dumps(ev.data), json.dumps(ev.metadata),
                        ev.version, ev.created_at
                    )
                    ev.global_position = row['global_position']
                    saved.append(ev)
                return saved

    async def read_stream(
        self, stream_id: str, from_version: int = 0
    ) -> List[AgentFlowEvent]:
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """SELECT * FROM agentflow_events
                   WHERE stream_id=$1 AND version>=$2
                   ORDER BY version""",
                stream_id, from_version
            )
            return [self._from_row(r) for r in rows]

    async def anonymize(self, stream_id: str, by_user: UUID):
        """GDPR: reemplaza PII en event_data sin borrar el evento."""
        async with self.pool.acquire() as conn:
            await conn.execute(
                """UPDATE agentflow_events
                   SET event_data = event_data - 'email' - 'phone' - 'name'
                       || '{"_anonymized": true}'::jsonb,
                       anonymized_at = NOW(), anonymized_by = $2
                   WHERE stream_id=$1""",
                stream_id, by_user
            )


class ConcurrencyError(Exception): pass