leocrm/app/services/sequence_service.py

"""Sequence service — get_next_number with SELECT FOR UPDATE locking, CRUD, delete."""

from __future__ import annotations

import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.audit import log_audit
from app.core.visibility import apply_visibility_filter, check_single_entity_access
from app.models.sequence import Sequence


def _sequence_to_dict(s: Sequence) -> dict[str, Any]:
    """Serialize a Sequence ORM object to dict."""
    return {
        "id": str(s.id),
        "name": s.name,
        "prefix": s.prefix,
        "next_number": s.next_number,
        "padding": s.padding,
        "created_at": s.created_at.isoformat() if s.created_at else None,
        "updated_at": s.updated_at.isoformat() if s.updated_at else None,
    }


async def get_next_number(
    db: AsyncSession,
    name: str,
    tenant_id: uuid.UUID,
) -> str:
    """Get the next formatted number for a sequence.

    Uses SELECT FOR UPDATE to lock the row, increment next_number,
    and return the formatted string (prefix + zero-padded number).

    Raises ValueError if the sequence does not exist.
    """
    q = (
        select(Sequence)
        .where(
            Sequence.name == name,
            Sequence.tenant_id == tenant_id,
            Sequence.deleted_at.is_(None),
        )
        .with_for_update()
    )
    result = await db.execute(q)
    seq = result.scalar_one_or_none()
    if seq is None:
        raise ValueError(f"Sequence '{name}' not found for tenant {tenant_id}")

    current_number = seq.next_number
    formatted = f"{seq.prefix}{str(current_number).zfill(seq.padding)}"
    seq.next_number = current_number + 1

    await db.flush()
    return formatted


async def create_sequence(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    data: dict[str, Any],
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """Create a new sequence."""
    sequence = Sequence(
        tenant_id=tenant_id,
        name=data["name"],
        prefix=data.get("prefix", ""),
        next_number=1,
        padding=data.get("padding", 4),
        owner_id=user_id,
    )
    db.add(sequence)
    await db.flush()
    await db.refresh(sequence)
    await log_audit(
        db, tenant_id, user_id, "create", "sequence", sequence.id,
        changes={"name": sequence.name, "prefix": sequence.prefix},
    )
    return _sequence_to_dict(sequence)


async def list_sequences(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """List all sequences for a tenant."""
    q = select(Sequence).where(
        Sequence.tenant_id == tenant_id,
        Sequence.deleted_at.is_(None),
    ).order_by(Sequence.name.asc())
    if user_id and not is_system_admin:
        q = await apply_visibility_filter(
            db, q, "sequence", Sequence, user_id, tenant_id, is_system_admin
        )
    result = await db.execute(q)
    sequences = result.scalars().all()
    return {
        "items": [_sequence_to_dict(s) for s in sequences],
        "total": len(sequences),
    }


async def update_sequence(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    sequence_id: uuid.UUID,
    data: dict[str, Any],
    is_system_admin: bool = False,
) -> dict[str, Any] | None:
    """Update a sequence (name, prefix, padding only — NOT next_number)."""
    q = select(Sequence).where(
        Sequence.id == sequence_id,
        Sequence.tenant_id == tenant_id,
        Sequence.deleted_at.is_(None),
    )
    result = await db.execute(q)
    sequence = result.scalar_one_or_none()
    if sequence is None:
        return None

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "sequence", sequence.id, user_id, tenant_id, "write", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    changes: dict[str, Any] = {}
    for field in ("name", "prefix", "padding"):
        if field in data and data[field] is not None:
            old_val = getattr(sequence, field)
            changes[field] = {"old": old_val, "new": data[field]}
            setattr(sequence, field, data[field])

    await db.flush()
    await db.refresh(sequence)
    await log_audit(db, tenant_id, user_id, "update", "sequence", sequence_id, changes=changes)
    return _sequence_to_dict(sequence)


async def delete_sequence(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    sequence_id: uuid.UUID,
    is_system_admin: bool = False,
) -> bool:
    """Soft-delete a sequence."""
    q = select(Sequence).where(
        Sequence.id == sequence_id,
        Sequence.tenant_id == tenant_id,
        Sequence.deleted_at.is_(None),
    )
    result = await db.execute(q)
    sequence = result.scalar_one_or_none()
    if sequence is None:
        return False

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "sequence", sequence.id, user_id, tenant_id, "admin", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    sequence.deleted_at = datetime.now(UTC)
    await db.flush()
    await log_audit(
        db, tenant_id, user_id, "delete", "sequence", sequence_id,
        changes={"name": sequence.name},
    )
    return True