Especificación Técnica: log_sesiones

Esquema: core.identidad
Base de Datos: core
Servicio: svc-identidad
Propósito: Bitácora transaccional de intentos y eventos de sesión (login/logout/fallos).

---

1. Justificación y Mejoras de Arquitectura

• Registro liviano orientado a escritura intensiva (audit created).
• Capacidad de correlación por token_jti e IP en diagnósticos.
• RLS por empresa actual mantiene la confidencialidad de la bitácora.

---

2. Definiciones de Implementación

DBMLSQLSQLAlchemyAlembic

Table core.identidad.log_sesiones {
    id                         uuid          [pk, default: `gen_random_uuid()`]
    empresa_id                 bigint        [not null]
    usuario_id                 uuid          [not null, ref: > mae_usuarios.id]
    accion                     varchar(20)  
    token_jti                  uuid         
    ip                         varchar(45)  
    user_agent                 varchar(300) 
    exitoso                    boolean      
    razon_fallo                varchar(100) 

    // Auditoría General
    created_at              timestamptz   [not null, default: `now()`, note: 'Fecha de creación']

    // RLS por empresa actual
}

CREATE TABLE IF NOT EXISTS core.identidad.log_sesiones (
    id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
    empresa_id BIGINT NOT NULL,
    usuario_id UUID NOT NULL,
    accion VARCHAR(20),
    token_jti UUID,
    ip VARCHAR(45),
    user_agent VARCHAR(300),
    exitoso BOOLEAN,
    razon_fallo VARCHAR(100),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT fk_log_sesiones_usuario FOREIGN KEY (usuario_id) REFERENCES core.identidad.mae_usuarios (id)
);

-- Row-Level Security
ALTER TABLE core.identidad.log_sesiones ENABLE ROW LEVEL SECURITY;
CREATE POLICY aislamiento_empresa ON core.identidad.log_sesiones
    FOR ALL
    USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);

-- Comentarios
COMMENT ON COLUMN core.identidad.log_sesiones.empresa_id IS 'Empresa de la sesión';
COMMENT ON COLUMN core.identidad.log_sesiones.usuario_id IS 'Usuario involucrado';
COMMENT ON COLUMN core.identidad.log_sesiones.accion IS 'Acción registrada (login, logout, fallo)';
COMMENT ON COLUMN core.identidad.log_sesiones.token_jti IS 'Identificador del JWT emitido';
COMMENT ON COLUMN core.identidad.log_sesiones.ip IS 'IP origen';
COMMENT ON COLUMN core.identidad.log_sesiones.user_agent IS 'User-Agent del cliente';
COMMENT ON COLUMN core.identidad.log_sesiones.exitoso IS 'Resultado de la acción';
COMMENT ON COLUMN core.identidad.log_sesiones.razon_fallo IS 'Motivo del fallo';

from datetime import datetime, date
from typing import Optional
import uuid

from sqlalchemy import (
    BigInteger,
    Boolean,
    CheckConstraint,
    Date,
    DateTime,
    ForeignKey,
    Identity,
    Index,
    Integer,
    SmallInteger,
    String,
    Text,
    UniqueConstraint,
    text,
)
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass

class LogSesion(Base):
    __tablename__ = 'log_sesiones'
    __table_args__ =         {"schema": 'core.identidad'},

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )
    empresa_id: Mapped[int] = mapped_column(
        BigInteger,
        nullable=False,
        comment='Empresa de la sesión',
    )
    usuario_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        ForeignKey("mae_usuarios.id"),
        comment='Usuario involucrado',
    )
    accion: Mapped[str | None] = mapped_column(
        String(20),
        comment='Acción registrada (login, logout, fallo)',
    )
    token_jti: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True),
        comment='Identificador del JWT emitido',
    )
    ip: Mapped[str | None] = mapped_column(
        String(45),
        comment='IP origen',
    )
    user_agent: Mapped[str | None] = mapped_column(
        String(300),
        comment='User-Agent del cliente',
    )
    exitoso: Mapped[bool | None] = mapped_column(
        Boolean,
        comment='Resultado de la acción',
    )
    razon_fallo: Mapped[str | None] = mapped_column(
        String(100),
        comment='Motivo del fallo',
    )

    # Auditoría General
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=text("now()"),
        comment='Fecha de creación',
    )

"""revision: core_id_0012
create table core.identidad.log_sesiones
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql


revision: str = 'core_id_0012'
down_revision: str | None = 'core_id_0011'
branch_labels: str | None = None
depends_on: str | None = None


def upgrade() -> None:
    op.create_table(
        'log_sesiones',
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("empresa_id", sa.BigInteger(), nullable=False),
        sa.Column("usuario_id", postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column("accion", sa.String(length=20)),
        sa.Column("token_jti", postgresql.UUID(as_uuid=True)),
        sa.Column("ip", sa.String(length=45)),
        sa.Column("user_agent", sa.String(length=300)),
        sa.Column("exitoso", sa.Boolean()),
        sa.Column("razon_fallo", sa.String(length=100)),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
        sa.ForeignKeyConstraint(['usuario_id'], ['core.identidad.mae_usuarios.id'], name='fk_log_sesiones_usuario'),
        schema='core.identidad',
    )

    op.execute(
        """
        ALTER TABLE core.identidad.log_sesiones ENABLE ROW LEVEL SECURITY;
        CREATE POLICY aislamiento_empresa
            ON core.identidad.log_sesiones
            FOR ALL
            USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);
        """
    )


def downgrade() -> None:
    op.drop_table('log_sesiones', schema='core.identidad')