-- utf8mb4 is required: it stores Urdu text and emoji correctly.
-- utf8mb4_unicode_ci gives case-insensitive, accent-aware comparison.
CREATE DATABASE aibytec_erp
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_unicode_ci;

-- Never let the application connect as root.
-- Create a dedicated user scoped to this one database.
CREATE USER 'erp_app'@'localhost' IDENTIFIED BY 'change_this_password';
GRANT ALL PRIVILEGES ON aibytec_erp.* TO 'erp_app'@'localhost';
FLUSH PRIVILEGES;

fastapi
uvicorn[standard]
sqlalchemy[asyncio]>=2.0
aiomysql              # async MySQL/MariaDB driver — pure Python, no compiler
alembic               # database migrations
pydantic>=2.0
pydantic-settings
python-dotenv
passlib[bcrypt]       # password hashing
pyjwt                 # access tokens
python-multipart      # form parsing for login
pytest
pytest-asyncio
httpx                 # async test client

# Database — XAMPP MariaDB on the default port
DATABASE_URL=mysql+aiomysql://erp_app:change_this_password@localhost:3306/aibytec_erp?charset=utf8mb4

# Security — generate with: python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY=replace_me_with_a_real_random_value
ACCESS_TOKEN_EXPIRE_MINUTES=60

# Application
ENVIRONMENT=development
DEFAULT_CURRENCY=PKR

# Project: AiBytec ERP

## Context
Multi-tenant ERP for trading/distribution SMEs in Pakistan.
Stack: Python 3.12, FastAPI, SQLAlchemy 2.0 async, MariaDB (XAMPP), Alembic.
Layout: app/models, app/schemas, app/repositories, app/services, app/api/v1, tests/

## Hard Rules
1. Every table has: id, tenant_id, created_at, updated_at.
2. Every query filters by tenant_id. No exceptions. Ever.
3. Money is DECIMAL(18,4). Never float. Currency defaults to PKR.
4. Business logic lives in services/, never in api/ route handlers.
5. Route handlers are async and return Pydantic schemas, never ORM objects.
6. Never edit a migration that has already been applied — write a new one.
7. No secrets in code. Everything from environment via app/config.py.

## Conventions
- Endpoints: /api/v1/{plural-noun}, e.g. /api/v1/purchase-orders
- Schemas: XxxCreate, XxxUpdate, XxxRead
- Tests mirror the source tree under tests/

## Commands
- Run:      uvicorn app.main:app --reload
- Migrate:  alembic revision --autogenerate -m "msg" && alembic upgrade head
- Test:     pytest -q

## Do Not
- Do not add a dependency without asking.
- Do not change the ledger posting logic without an explicit instruction.
- Do not use raw SQL where the ORM will do.

"""Verify that Python can reach the XAMPP MariaDB database.

Run this before the first class. If it prints OK, the student is ready.
"""
import asyncio
import os

from dotenv import load_dotenv
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

load_dotenv()


async def main() -> None:
    url = os.getenv("DATABASE_URL")
    if not url:
        raise SystemExit("DATABASE_URL is missing. Copy .env.example to .env first.")

    engine = create_async_engine(url, echo=False)
    try:
        async with engine.connect() as conn:
            version = (await conn.execute(text("SELECT VERSION()"))).scalar()
            dbname = (await conn.execute(text("SELECT DATABASE()"))).scalar()
        print(f"OK  connected to '{dbname}' on {version}")
    except Exception as exc:
        # Fail with an explanation a beginner can act on, not a stack trace.
        print("FAILED to connect.")
        print(f"  {type(exc).__name__}: {exc}")
        print("  Check: XAMPP MySQL is started; the database exists;")
        print("         the user/password in .env match what you granted.")
    finally:
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

"""Shared declarative base and mixins for every ERP table."""
from datetime import datetime

from sqlalchemy import BigInteger, DateTime, Index, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    """Declarative base. All ERP models inherit from this."""

    # InnoDB + utf8mb4 on every table. MyISAM has no transactions,
    # which would silently break double-entry accounting.
    __table_args__ = {
        "mysql_engine": "InnoDB",
        "mysql_charset": "utf8mb4",
        "mysql_collate": "utf8mb4_unicode_ci",
    }


class TimestampMixin:
    """created_at / updated_at, maintained by the database itself."""

    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(), nullable=False
    )


class TenantMixin:
    """Every business table belongs to exactly one tenant (client company).

    Chapter 5 enforces filtering on this column at the repository layer so
    that isolation does not depend on any individual query being written
    correctly.
    """

    tenant_id: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True)


class ERPBase(Base, TimestampMixin, TenantMixin):
    """Convenience base: identity + timestamps + tenant, in one place."""

    __abstract__ = True

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)

from decimal import Decimal
from enum import StrEnum

from sqlalchemy import BigInteger, Boolean, Enum, ForeignKey, Numeric, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import ERPBase


class PartnerType(StrEnum):
    CUSTOMER = "customer"
    SUPPLIER = "supplier"
    BOTH = "both"


class Product(ERPBase):
    __tablename__ = "products"
    __table_args__ = (
        # SKUs must be unique per tenant, not globally — two client
        # companies may legitimately both use the SKU "A-100".
        UniqueConstraint("tenant_id", "sku", name="uq_product_tenant_sku"),
    )

    sku: Mapped[str] = mapped_column(String(64), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    uom: Mapped[str] = mapped_column(String(16), default="PCS", nullable=False)
    # Default prices only. The document line always stores its own price,
    # so historical invoices never change when the price list is updated.
    purchase_price: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    sale_price: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    tax_rate: Mapped[Decimal] = mapped_column(Numeric(5, 2), default=0)  # % e.g. 18.00
    reorder_level: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)


class Partner(ERPBase):
    """Customers and suppliers in one table.

    Many businesses buy from and sell to the same company; two separate
    tables force you to maintain that relationship twice.
    """

    __tablename__ = "partners"

    code: Mapped[str] = mapped_column(String(32), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    partner_type: Mapped[PartnerType] = mapped_column(
        Enum(PartnerType), default=PartnerType.CUSTOMER
    )
    ntn: Mapped[str | None] = mapped_column(String(32))   # FBR tax number
    phone: Mapped[str | None] = mapped_column(String(32))
    address: Mapped[str | None] = mapped_column(String(512))
    credit_limit: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)


class Warehouse(ERPBase):
    __tablename__ = "warehouses"

    code: Mapped[str] = mapped_column(String(32), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

from datetime import datetime
from decimal import Decimal
from enum import StrEnum

from sqlalchemy import BigInteger, DateTime, Enum, ForeignKey, Index, Numeric, String
from sqlalchemy.orm import Mapped, mapped_column

from app.models.base import ERPBase


class MovementType(StrEnum):
    PURCHASE_RECEIPT = "purchase_receipt"   # +
    SALE_DELIVERY = "sale_delivery"         # -
    ADJUSTMENT = "adjustment"               # + or -
    TRANSFER_IN = "transfer_in"             # +
    TRANSFER_OUT = "transfer_out"           # -


class StockMovement(ERPBase):
    """Append-only record of every physical stock change.

    Rows are NEVER updated or deleted. A mistake is corrected by writing an
    opposite movement, which preserves the audit trail. Stock on hand is
    always SUM(quantity) over this table — never a stored column.
    """

    __tablename__ = "stock_movements"
    __table_args__ = (
        # The index that makes stock-on-hand queries fast. Without it,
        # every balance check becomes a full table scan.
        Index("ix_stock_balance", "tenant_id", "product_id", "warehouse_id"),
    )

    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), nullable=False)
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("warehouses.id"), nullable=False)
    movement_type: Mapped[MovementType] = mapped_column(Enum(MovementType), nullable=False)

    # Signed: positive is stock in, negative is stock out.
    # One signed column beats separate in/out columns — balance is a plain SUM.
    quantity: Mapped[Decimal] = mapped_column(Numeric(18, 4), nullable=False)
    unit_cost: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)

    # Which document caused this movement. Stored as a loose pair rather
    # than a foreign key because the source may be any document type.
    source_type: Mapped[str | None] = mapped_column(String(32))
    source_id: Mapped[int | None] = mapped_column(BigInteger)

    moved_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
    note: Mapped[str | None] = mapped_column(String(255))

from datetime import date
from decimal import Decimal
from enum import StrEnum

from sqlalchemy import BigInteger, Date, Enum, ForeignKey, Numeric, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import ERPBase


class AccountType(StrEnum):
    ASSET = "asset"
    LIABILITY = "liability"
    EQUITY = "equity"
    INCOME = "income"
    EXPENSE = "expense"


class Account(ERPBase):
    """A node in the chart of accounts. Self-referencing for grouping."""

    __tablename__ = "accounts"
    __table_args__ = (UniqueConstraint("tenant_id", "code", name="uq_account_tenant_code"),)

    code: Mapped[str] = mapped_column(String(16), nullable=False)   # "1100"
    name: Mapped[str] = mapped_column(String(255), nullable=False)  # "Accounts Receivable"
    account_type: Mapped[AccountType] = mapped_column(Enum(AccountType), nullable=False)
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    # Only leaf accounts accept postings; parents exist to total their children.
    is_postable: Mapped[bool] = mapped_column(default=True)


class JournalEntry(ERPBase):
    """The header of a balanced accounting transaction."""

    __tablename__ = "journal_entries"

    entry_date: Mapped[date] = mapped_column(Date, nullable=False)
    narration: Mapped[str] = mapped_column(String(512), nullable=False)
    source_type: Mapped[str | None] = mapped_column(String(32))
    source_id: Mapped[int | None] = mapped_column(BigInteger)

    lines: Mapped[list["JournalLine"]] = relationship(
        back_populates="entry", cascade="all, delete-orphan", lazy="selectin"
    )


class JournalLine(ERPBase):
    """One side of a double entry. Debit and credit are separate,
    non-negative columns — a signed column makes trial balances harder
    to read and reconcile against printed accounting reports."""

    __tablename__ = "journal_lines"

    entry_id: Mapped[int] = mapped_column(ForeignKey("journal_entries.id"), nullable=False)
    account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"), nullable=False)
    debit: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    credit: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)

    entry: Mapped["JournalEntry"] = relationship(back_populates="lines")

"""The single most important test in an ERP: every journal entry balances."""
from decimal import Decimal

import pytest
from sqlalchemy import select

from app.models.accounting import JournalEntry


@pytest.mark.asyncio
async def test_every_journal_entry_balances(session, seeded_transactions):
    """Debits must equal credits for every entry, to the paisa.

    Run this after any change to posting logic. If it fails, stop and fix
    it before writing another line — an unbalanced ledger silently corrupts
    every financial report built on top of it.
    """
    entries = (await session.execute(select(JournalEntry))).scalars().all()
    assert entries, "no entries seeded — the fixture is not doing its job"

    for entry in entries:
        debits = sum(line.debit for line in entry.lines)
        credits = sum(line.credit for line in entry.lines)
        assert debits == credits, (
            f"Entry {entry.id} ({entry.narration}) is out of balance: "
            f"debits {debits} != credits {credits}"
        )
        assert debits > Decimal("0"), f"Entry {entry.id} posts nothing"

aibytec-erp/
├── CLAUDE.md              # agent conventions (Chapter 2)
├── SPEC.md                # domain specification (Chapter 3)
├── .env.example
├── requirements.txt
├── alembic/               # migrations
├── app/
│   ├── main.py            # FastAPI app, router registration
│   ├── config.py          # settings from environment
│   ├── database.py        # engine, session factory
│   ├── deps.py            # shared dependencies (session, current user)
│   ├── models/            # SQLAlchemy tables
│   ├── schemas/           # Pydantic request/response models
│   ├── repositories/      # tenant-scoped queries
│   ├── services/          # business logic and transactions
│   └── api/v1/            # routers
└── tests/                 # mirrors app/

"""Application settings, loaded once from the environment."""
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str
    secret_key: str
    access_token_expire_minutes: int = 60
    environment: str = "development"
    default_currency: str = "PKR"

    @property
    def is_production(self) -> bool:
        return self.environment == "production"


@lru_cache
def get_settings() -> Settings:
    """Cached so the .env file is parsed once per process, not per request."""
    return Settings()

"""Async engine and session factory for MariaDB/MySQL."""
from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import get_settings

settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=not settings.is_production,   # log SQL while learning; silence in prod
    pool_pre_ping=True,                # XAMPP drops idle connections; this reconnects
    pool_recycle=3600,                 # stay under MySQL's 8-hour wait_timeout
    pool_size=10,
    max_overflow=20,
)

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,   # keep objects usable after commit, for response building
)


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """FastAPI dependency. One session per request, always closed."""
    async with SessionLocal() as session:
        yield session

from decimal import Decimal

from pydantic import BaseModel, ConfigDict, Field, field_validator


class ProductBase(BaseModel):
    sku: str = Field(min_length=1, max_length=64)
    name: str = Field(min_length=1, max_length=255)
    uom: str = Field(default="PCS", max_length=16)
    sale_price: Decimal = Field(default=Decimal("0"), ge=0)
    tax_rate: Decimal = Field(default=Decimal("0"), ge=0, le=100)
    reorder_level: Decimal = Field(default=Decimal("0"), ge=0)

    @field_validator("sku")
    @classmethod
    def normalise_sku(cls, v: str) -> str:
        """Uppercase and strip so 'a-100 ' and 'A-100' cannot both exist."""
        return v.strip().upper()


class ProductCreate(ProductBase):
    """What a client may send when creating. Note the absence of tenant_id:
    it comes from the authenticated token, never from the request body."""

    purchase_price: Decimal = Field(default=Decimal("0"), ge=0)


class ProductUpdate(BaseModel):
    """All fields optional — this is a PATCH payload."""

    name: str | None = Field(default=None, max_length=255)
    sale_price: Decimal | None = Field(default=None, ge=0)
    reorder_level: Decimal | None = Field(default=None, ge=0)
    is_active: bool | None = None


class ProductRead(ProductBase):
    """What we return. purchase_price is deliberately excluded — cost price
    is not something every role should see."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    is_active: bool

"""Generic tenant-scoped repository.

Every query in the application goes through this class. Because tenant_id
is applied inside the base methods, a developer (or an agent) cannot write
a query that forgets it without deliberately bypassing the repository.
"""
from typing import Any, Generic, TypeVar

from sqlalchemy import Select, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.base import ERPBase

ModelT = TypeVar("ModelT", bound=ERPBase)


class TenantRepository(Generic[ModelT]):
    model: type[ModelT]

    def __init__(self, session: AsyncSession, tenant_id: int) -> None:
        self.session = session
        self.tenant_id = tenant_id

    def _base_query(self) -> Select:
        """Every query starts here. This is the isolation boundary."""
        return select(self.model).where(self.model.tenant_id == self.tenant_id)

    async def get(self, obj_id: int) -> ModelT | None:
        """Fetch by id, scoped to the tenant.

        A request for another tenant's id returns None rather than raising,
        so the caller responds 404 and does not confirm the row exists.
        """
        result = await self.session.execute(
            self._base_query().where(self.model.id == obj_id)
        )
        return result.scalar_one_or_none()

    async def list(
        self, *, limit: int = 50, offset: int = 0, **filters: Any
    ) -> list[ModelT]:
        query = self._base_query()
        for field, value in filters.items():
            if value is not None:
                query = query.where(getattr(self.model, field) == value)
        result = await self.session.execute(query.limit(limit).offset(offset))
        return list(result.scalars().all())

    async def count(self, **filters: Any) -> int:
        query = select(func.count()).select_from(self.model).where(
            self.model.tenant_id == self.tenant_id
        )
        for field, value in filters.items():
            if value is not None:
                query = query.where(getattr(self.model, field) == value)
        return (await self.session.execute(query)).scalar_one()

    async def create(self, **data: Any) -> ModelT:
        """tenant_id is injected here, never accepted from the caller."""
        obj = self.model(**data, tenant_id=self.tenant_id)
        self.session.add(obj)
        await self.session.flush()   # populate obj.id without committing
        return obj

from app.repositories.product import ProductRepository
from app.schemas.product import ProductCreate


class DuplicateSKUError(Exception):
    """Raised when a SKU already exists for this tenant.

    A domain exception, not an HTTPException — the service layer must stay
    free of HTTP concepts so it can be reused by the CLI, the MCP server
    (Chapter 10), and background jobs.
    """


class ProductService:
    def __init__(self, repo: ProductRepository) -> None:
        self.repo = repo

    async def create_product(self, payload: ProductCreate):
        existing = await self.repo.get_by_sku(payload.sku)
        if existing is not None:
            raise DuplicateSKUError(f"SKU {payload.sku} already exists")

        product = await self.repo.create(**payload.model_dump())
        await self.repo.session.commit()   # the service owns the transaction
        return product

from fastapi import APIRouter, Depends, HTTPException, Query, status

from app.deps import get_product_service
from app.schemas.product import ProductCreate, ProductRead
from app.services.product_service import DuplicateSKUError, ProductService

router = APIRouter(prefix="/products", tags=["Products"])


@router.post("", response_model=ProductRead, status_code=status.HTTP_201_CREATED)
async def create_product(
    payload: ProductCreate,
    service: ProductService = Depends(get_product_service),
):
    """Create a product for the authenticated user's tenant."""
    try:
        return await service.create_product(payload)
    except DuplicateSKUError as exc:
        # Translate the domain error into HTTP here, and only here.
        raise HTTPException(status.HTTP_409_CONFLICT, str(exc)) from exc


@router.get("", response_model=list[ProductRead])
async def list_products(
    q: str | None = Query(None, description="Search SKU or name"),
    limit: int = Query(50, le=200),   # cap it: never let a client request everything
    offset: int = Query(0, ge=0),
    service: ProductService = Depends(get_product_service),
):
    return await service.search(q=q, limit=limit, offset=offset)

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from app.api.v1 import accounts, auth, partners, products, purchases, sales, stock

app = FastAPI(
    title="AiBytec ERP",
    version="1.0.0",
    description="Custom ERP for trading and distribution businesses.",
)


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    """Turn Pydantic's nested errors into a flat, readable list.

    Front-end developers and demo audiences can act on this; they cannot
    act on the default payload.
    """
    problems = [
        {
            "field": ".".join(str(p) for p in err["loc"] if p != "body"),
            "message": err["msg"],
        }
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": "Validation failed", "problems": problems},
    )


@app.get("/health", tags=["System"])
async def health():
    return {"status": "ok"}


for module in (auth, products, partners, stock, purchases, sales, accounts):
    app.include_router(module.router, prefix="/api/v1")

"""Shared fixtures. Tests run against a separate database, never your dev one."""
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.database import get_session
from app.main import app
from app.models.base import Base

TEST_URL = (
    "mysql+aiomysql://erp_app:change_this_password@localhost:3306/"
    "aibytec_erp_test?charset=utf8mb4"
)


@pytest_asyncio.fixture
async def session():
    """Fresh schema per test. Slower than transactional rollback, but far
    easier for students to reason about when a test fails."""
    engine = create_async_engine(TEST_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as s:
        yield s
    await engine.dispose()


@pytest_asyncio.fixture
async def client(session):
    """HTTP client with the real app, but the test database injected."""
    app.dependency_overrides[get_session] = lambda: session
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

from sqlalchemy import BigInteger, Boolean, Column, ForeignKey, String, Table, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base, ERPBase, TimestampMixin

# Association tables. No business columns, so plain Core tables are fine.
role_permissions = Table(
    "role_permissions", Base.metadata,
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
    Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
)

user_roles = Table(
    "user_roles", Base.metadata,
    Column("user_id", ForeignKey("users.id"), primary_key=True),
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
)


class Tenant(Base, TimestampMixin):
    """A client company. The root of every isolation boundary."""

    __tablename__ = "tenants"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    ntn: Mapped[str | None] = mapped_column(String(32))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)


class User(ERPBase):
    __tablename__ = "users"
    __table_args__ = (
        # Email is unique per tenant, not globally: the same person may
        # legitimately hold accounts at two client companies.
        UniqueConstraint("tenant_id", "email", name="uq_user_tenant_email"),
    )

    email: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str] = mapped_column(String(255), nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

    roles: Mapped[list["Role"]] = relationship(
        secondary=user_roles, lazy="selectin"   # avoid N+1 on every request
    )


class Permission(Base, TimestampMixin):
    """Global catalogue, e.g. 'invoice:create'. Not tenant-scoped —
    the vocabulary of actions is the same for every client."""

    __tablename__ = "permissions"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    code: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    description: Mapped[str] = mapped_column(String(255), default="")


class Role(ERPBase):
    """Tenant-scoped, so each client can define its own roles."""

    __tablename__ = "roles"

    name: Mapped[str] = mapped_column(String(64), nullable=False)
    permissions: Mapped[list[Permission]] = relationship(
        secondary=role_permissions, lazy="selectin"
    )

"""Password hashing and JWT issuing. The only module that touches either."""
from datetime import UTC, datetime, timedelta

import jwt
from passlib.context import CryptContext

from app.config import get_settings

settings = get_settings()

# bcrypt: deliberately slow, which is the point. Never MD5 or SHA for passwords.
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

ALGORITHM = "HS256"


def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def create_access_token(user_id: int, tenant_id: int, permissions: list[str]) -> str:
    """Embed tenant_id in the token itself.

    This is the security keystone: the tenant is never read from a request
    body, a query parameter, or a header the client controls. It comes only
    from a token the server signed.
    """
    now = datetime.now(UTC)
    payload = {
        "sub": str(user_id),
        "tenant_id": tenant_id,
        "perms": permissions,
        "iat": now,
        "exp": now + timedelta(minutes=settings.access_token_expire_minutes),
    }
    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)


def decode_access_token(token: str) -> dict:
    """Raises jwt.PyJWTError on tamper, expiry, or bad signature."""
    return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])

"""Shared FastAPI dependencies: current user, tenant context, permissions."""
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_session
from app.models.auth import User
from app.security import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


class CurrentUser:
    """Everything a request needs to know about who is calling."""

    def __init__(self, user_id: int, tenant_id: int, permissions: set[str]) -> None:
        self.user_id = user_id
        self.tenant_id = tenant_id
        self.permissions = permissions


async def get_current_user(token: str = Depends(oauth2_scheme)) -> CurrentUser:
    try:
        payload = decode_access_token(token)
    except jwt.PyJWTError:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            "Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return CurrentUser(
        user_id=int(payload["sub"]),
        tenant_id=int(payload["tenant_id"]),
        permissions=set(payload.get("perms", [])),
    )


def require(*permissions: str):
    """Dependency factory for permission checks.

        @router.post("", dependencies=[Depends(require("product:create"))])

    Requires ALL listed permissions. Returns 403, not 404 — the caller is
    authenticated, so hiding the endpoint's existence buys nothing.
    """

    async def checker(user: CurrentUser = Depends(get_current_user)) -> CurrentUser:
        missing = set(permissions) - user.permissions
        if missing:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                f"Missing permission(s): {', '.join(sorted(missing))}",
            )
        return user

    return checker


async def get_product_service(
    session: AsyncSession = Depends(get_session),
    user: CurrentUser = Depends(get_current_user),
):
    """Note how tenant_id flows from the token into the repository.

    A route handler never sees or passes a tenant id. It cannot get it wrong.
    """
    from app.repositories.product import ProductRepository
    from app.services.product_service import ProductService

    return ProductService(ProductRepository(session, user.tenant_id))

from datetime import datetime

from sqlalchemy import JSON, BigInteger, DateTime, Index, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.models.base import ERPBase


class AuditLog(ERPBase):
    """Append-only record of who did what. Never updated, never deleted.

    Retain for at least as long as the client's statutory audit period —
    in Pakistan, six years is a safe default.
    """

    __tablename__ = "audit_logs"
    __table_args__ = (Index("ix_audit_lookup", "tenant_id", "entity", "entity_id"),)

    user_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    action: Mapped[str] = mapped_column(String(32), nullable=False)    # create/update/delete
    entity: Mapped[str] = mapped_column(String(64), nullable=False)    # "product"
    entity_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    # Only what changed, as {"field": {"old": ..., "new": ...}}.
    # Storing whole rows makes the table enormous and the diff unreadable.
    changes: Mapped[dict] = mapped_column(JSON, default=dict)
    ip_address: Mapped[str | None] = mapped_column(String(45))         # IPv6-safe
    occurred_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

"""Proof that tenants cannot see each other's data.

If this test ever fails, stop all other work. Nothing else in the codebase
matters more than this passing.
"""
import pytest


@pytest.mark.asyncio
async def test_tenant_cannot_read_other_tenants_product(client, two_tenants):
    tenant_a, tenant_b = two_tenants

    created = await client.post(
        "/api/v1/products",
        json={"sku": "A-100", "name": "Tenant A Widget"},
        headers={"Authorization": f"Bearer {tenant_a.token}"},
    )
    assert created.status_code == 201
    product_id = created.json()["id"]

    # Tenant B asks for tenant A's product by its real id.
    stolen = await client.get(
        f"/api/v1/products/{product_id}",
        headers={"Authorization": f"Bearer {tenant_b.token}"},
    )
    # 404, not 403: confirming the row exists is itself an information leak.
    assert stolen.status_code == 404

    # And it must not appear in any listing.
    listing = await client.get(
        "/api/v1/products",
        headers={"Authorization": f"Bearer {tenant_b.token}"},
    )
    assert all(p["id"] != product_id for p in listing.json())


@pytest.mark.asyncio
async def test_tenant_id_in_body_is_ignored(client, two_tenants):
    """A forged tenant_id in the payload must have no effect at all."""
    tenant_a, tenant_b = two_tenants

    response = await client.post(
        "/api/v1/products",
        json={"sku": "FORGED", "name": "Attempt", "tenant_id": tenant_b.id},
        headers={"Authorization": f"Bearer {tenant_a.token}"},
    )
    assert response.status_code == 201

    # It must have landed in tenant A, whose token was used.
    seen_by_b = await client.get(
        "/api/v1/products",
        headers={"Authorization": f"Bearer {tenant_b.token}"},
    )
    assert all(p["sku"] != "FORGED" for p in seen_by_b.json())

from datetime import date
from decimal import Decimal
from enum import StrEnum

from sqlalchemy import Date, Enum, ForeignKey, Numeric, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import ERPBase


class DocumentStatus(StrEnum):
    DRAFT = "draft"           # editable
    CONFIRMED = "confirmed"   # committed to the supplier; locked
    PARTIAL = "partial"       # some quantity received
    COMPLETED = "completed"   # fully received
    CANCELLED = "cancelled"


class PurchaseOrder(ERPBase):
    __tablename__ = "purchase_orders"

    po_number: Mapped[str] = mapped_column(String(32), nullable=False)
    supplier_id: Mapped[int] = mapped_column(ForeignKey("partners.id"), nullable=False)
    order_date: Mapped[date] = mapped_column(Date, nullable=False)
    expected_date: Mapped[date | None] = mapped_column(Date)
    status: Mapped[DocumentStatus] = mapped_column(
        Enum(DocumentStatus), default=DocumentStatus.DRAFT
    )
    notes: Mapped[str | None] = mapped_column(String(512))

    lines: Mapped[list["PurchaseOrderLine"]] = relationship(
        back_populates="order", cascade="all, delete-orphan", lazy="selectin"
    )

    @property
    def total(self) -> Decimal:
        """Derived, never stored — a stored total drifts from its lines."""
        return sum((line.line_total for line in self.lines), Decimal("0"))


class PurchaseOrderLine(ERPBase):
    __tablename__ = "purchase_order_lines"

    order_id: Mapped[int] = mapped_column(ForeignKey("purchase_orders.id"), nullable=False)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), nullable=False)

    quantity_ordered: Mapped[Decimal] = mapped_column(Numeric(18, 4), nullable=False)
    # Accumulates as goods receipts are posted against this line.
    quantity_received: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    # Copied from the product at confirmation time, then frozen forever.
    unit_price: Mapped[Decimal] = mapped_column(Numeric(18, 4), nullable=False)

    order: Mapped["PurchaseOrder"] = relationship(back_populates="lines")

    @property
    def line_total(self) -> Decimal:
        return self.quantity_ordered * self.unit_price

    @property
    def quantity_pending(self) -> Decimal:
        return self.quantity_ordered - self.quantity_received

"""Goods receipt posting — the canonical atomic ERP transaction."""
from datetime import datetime
from decimal import Decimal

from app.models.purchasing import DocumentStatus
from app.models.stock import MovementType, StockMovement


class ReceiptError(Exception):
    """Domain error: the receipt is not valid business-wise."""


class ReceiptService:
    def __init__(self, session, tenant_id: int, user_id: int) -> None:
        self.session = session
        self.tenant_id = tenant_id
        self.user_id = user_id

    async def post_receipt(
        self, po_id: int, warehouse_id: int, lines: list[dict]
    ) -> dict:
        """Receive goods against a purchase order.

        lines: [{"po_line_id": int, "quantity": Decimal}]

        Writes stock movements, updates received quantities, and advances
        the PO status — all inside ONE transaction. A failure at any point
        rolls the whole thing back, so stock can never increase without the
        PO recording it.
        """
        po = await self._get_po(po_id)
        if po is None:
            raise ReceiptError(f"Purchase order {po_id} not found")
        if po.status in (DocumentStatus.DRAFT, DocumentStatus.CANCELLED):
            raise ReceiptError(f"Cannot receive against a {po.status} order")

        lines_by_id = {line.id: line for line in po.lines}
        now = datetime.now()

        try:
            for item in lines:
                po_line = lines_by_id.get(item["po_line_id"])
                if po_line is None:
                    raise ReceiptError(f"Line {item['po_line_id']} not on this order")

                qty = Decimal(str(item["quantity"]))
                if qty <= 0:
                    raise ReceiptError("Received quantity must be positive")

                # Over-receipt is a real warehouse event, but it must be a
                # deliberate decision, not a silent one.
                if qty > po_line.quantity_pending:
                    raise ReceiptError(
                        f"Line {po_line.id}: receiving {qty} exceeds the "
                        f"{po_line.quantity_pending} still pending"
                    )

                self.session.add(
                    StockMovement(
                        tenant_id=self.tenant_id,
                        product_id=po_line.product_id,
                        warehouse_id=warehouse_id,
                        movement_type=MovementType.PURCHASE_RECEIPT,
                        quantity=qty,                    # positive: stock in
                        unit_cost=po_line.unit_price,
                        source_type="purchase_order",
                        source_id=po.id,
                        moved_at=now,
                    )
                )
                po_line.quantity_received += qty

            po.status = (
                DocumentStatus.COMPLETED
                if all(l.quantity_pending == 0 for l in po.lines)
                else DocumentStatus.PARTIAL
            )

            await self.session.commit()
            return {"po_id": po.id, "status": po.status, "received_at": now}

        except Exception:
            # Explicit rollback so a half-written receipt never persists.
            await self.session.rollback()
            raise

from decimal import Decimal

from sqlalchemy import func, select

from app.models.master import Product
from app.models.stock import StockMovement
from app.repositories.base import TenantRepository


class StockRepository(TenantRepository[StockMovement]):
    model = StockMovement

    async def on_hand(self, product_id: int, warehouse_id: int | None = None) -> Decimal:
        """Current stock = SUM of all signed movements.

        Backed by ix_stock_balance, this stays fast into the millions of rows.
        When it eventually does not, add a periodic snapshot table and sum
        only movements since the last snapshot — do not add a mutable column.
        """
        query = select(func.coalesce(func.sum(StockMovement.quantity), 0)).where(
            StockMovement.tenant_id == self.tenant_id,
            StockMovement.product_id == product_id,
        )
        if warehouse_id is not None:
            query = query.where(StockMovement.warehouse_id == warehouse_id)
        return Decimal(str((await self.session.execute(query)).scalar_one()))

    async def below_reorder_level(self) -> list[dict]:
        """Products whose total stock has fallen to or below reorder level.

        This one query drives the purchasing screen the client will use most.
        """
        balance = func.coalesce(func.sum(StockMovement.quantity), 0).label("on_hand")
        query = (
            select(Product.id, Product.sku, Product.name,
                   balance, Product.reorder_level)
            .outerjoin(StockMovement, StockMovement.product_id == Product.id)
            .where(Product.tenant_id == self.tenant_id, Product.is_active.is_(True))
            .group_by(Product.id)
            .having(balance <= Product.reorder_level)
        )
        rows = await self.session.execute(query)
        return [
            {"product_id": r.id, "sku": r.sku, "name": r.name,
             "on_hand": r.on_hand, "reorder_level": r.reorder_level}
            for r in rows
        ]

from datetime import datetime
from decimal import Decimal

from sqlalchemy import select

from app.models.master import Product
from app.models.stock import MovementType, StockMovement


class InsufficientStockError(Exception):
    pass


class StockService:
    def __init__(self, session, tenant_id: int) -> None:
        self.session = session
        self.tenant_id = tenant_id

    async def reserve_and_issue(
        self, product_id: int, warehouse_id: int, quantity: Decimal, source: tuple[str, int]
    ) -> None:
        """Issue stock out, safely, under concurrent access.

        SELECT ... FOR UPDATE takes an exclusive lock on the product row.
        A second transaction attempting the same product blocks until this
        one commits, then re-reads the balance and sees the true figure.

        Note it is the PRODUCT row we lock, not the movement rows: locking
        the aggregate's anchor is what serialises the read-then-write.
        """
        locked = await self.session.execute(
            select(Product)
            .where(Product.id == product_id, Product.tenant_id == self.tenant_id)
            .with_for_update()
        )
        product = locked.scalar_one_or_none()
        if product is None:
            raise InsufficientStockError(f"Product {product_id} not found")

        available = await self._on_hand(product_id, warehouse_id)
        if available < quantity:
            raise InsufficientStockError(
                f"{product.sku}: {available} available, {quantity} requested"
            )

        self.session.add(
            StockMovement(
                tenant_id=self.tenant_id,
                product_id=product_id,
                warehouse_id=warehouse_id,
                movement_type=MovementType.SALE_DELIVERY,
                quantity=-quantity,          # negative: stock out
                source_type=source[0],
                source_id=source[1],
                moved_at=datetime.now(),
            )
        )
        # Caller commits — the lock releases then, not before.

"""Prove that two simultaneous issues cannot oversell."""
import asyncio
from decimal import Decimal

import pytest

from app.services.stock_service import InsufficientStockError


@pytest.mark.asyncio
async def test_cannot_oversell_last_unit(session_factory, seeded_one_unit):
    """One unit in stock, two concurrent requests: exactly one must win."""
    product_id, warehouse_id = seeded_one_unit

    async def try_issue():
        async with session_factory() as s:
            service = StockService(s, tenant_id=1)
            await service.reserve_and_issue(
                product_id, warehouse_id, Decimal("1"), ("sale_order", 1)
            )
            await s.commit()

    results = await asyncio.gather(try_issue(), try_issue(), return_exceptions=True)

    succeeded = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, InsufficientStockError)]

    assert len(succeeded) == 1, "both issues succeeded — the row lock is missing"
    assert len(failed) == 1, "the losing transaction did not fail cleanly"

"""Invoice arithmetic. Small, pure, and heavily tested — this code decides
what a client charges their customers, and errors here are visible."""
from decimal import ROUND_HALF_UP, Decimal

TWO_PLACES = Decimal("0.01")


def money(value: Decimal) -> Decimal:
    """Round to 2 decimals, half-up.

    Banker's rounding (Python's default for round()) is NOT what invoices
    use. 2.5 must become 3, not 2.
    """
    return value.quantize(TWO_PLACES, rounding=ROUND_HALF_UP)


def calculate_line(
    quantity: Decimal, unit_price: Decimal,
    discount_pct: Decimal = Decimal("0"), tax_rate: Decimal = Decimal("0"),
) -> dict:
    """Compute one invoice line. Order matters: discount before tax,
    because tax is charged on the discounted amount."""
    gross = money(quantity * unit_price)
    discount = money(gross * discount_pct / Decimal("100"))
    net = money(gross - discount)
    tax = money(net * tax_rate / Decimal("100"))
    return {
        "gross": gross, "discount": discount,
        "net": net, "tax": tax, "total": money(net + tax),
    }


def calculate_invoice(lines: list[dict]) -> dict:
    """Sum already-rounded lines so the printed figures reconcile exactly."""
    computed = [calculate_line(**line) for line in lines]
    return {
        "lines": computed,
        "subtotal": sum((c["net"] for c in computed), Decimal("0")),
        "tax_total": sum((c["tax"] for c in computed), Decimal("0")),
        "grand_total": sum((c["total"] for c in computed), Decimal("0")),
    }

"""Double-entry posting engine.

Every financial event in the ERP calls post_entry(). Centralising it means
there is exactly one place where the balance rule is enforced, and exactly
one place to audit when the trial balance is wrong.
"""
from datetime import date
from decimal import Decimal

from app.models.accounting import JournalEntry, JournalLine


class UnbalancedEntryError(Exception):
    """Debits did not equal credits. This must never reach the database."""


async def post_entry(
    session, *, tenant_id: int, entry_date: date, narration: str,
    lines: list[dict], source: tuple[str, int] | None = None,
) -> JournalEntry:
    """Write a balanced journal entry.

    lines: [{"account_id": int, "debit": Decimal, "credit": Decimal}]

    Does NOT commit — the caller owns the transaction, so a sale can post
    stock movements and journal lines together, atomically.
    """
    total_debit = sum((Decimal(str(l.get("debit", 0))) for l in lines), Decimal("0"))
    total_credit = sum((Decimal(str(l.get("credit", 0))) for l in lines), Decimal("0"))

    if total_debit != total_credit:
        raise UnbalancedEntryError(
            f"'{narration}': debits {total_debit} != credits {total_credit}"
        )
    if total_debit == 0:
        raise UnbalancedEntryError(f"'{narration}': entry posts nothing")

    entry = JournalEntry(
        tenant_id=tenant_id, entry_date=entry_date, narration=narration,
        source_type=source[0] if source else None,
        source_id=source[1] if source else None,
    )
    for line in lines:
        entry.lines.append(
            JournalLine(
                tenant_id=tenant_id,
                account_id=line["account_id"],
                debit=Decimal(str(line.get("debit", 0))),
                credit=Decimal(str(line.get("credit", 0))),
            )
        )

    session.add(entry)
    await session.flush()
    return entry

from datetime import date
from decimal import Decimal

from app.services.invoice_calc import calculate_invoice
from app.services.ledger import post_entry


class SalesService:
    """Posts a delivery + invoice as a single atomic business event."""

    def __init__(self, session, tenant_id: int, accounts: dict[str, int]) -> None:
        self.session = session
        self.tenant_id = tenant_id
        # Account ids resolved once from the chart of accounts, e.g.
        # {"receivable": 12, "revenue": 40, "tax_payable": 21,
        #  "cogs": 50, "inventory": 13}
        self.accounts = accounts

    async def post_sale(self, order_id: int, invoice_date: date) -> dict:
        """Deliver goods and invoice them.

        Four things happen together, or none of them do:
          1. stock leaves the warehouse (with a row lock per product)
          2. an invoice document is created
          3. receivable + revenue + tax are posted
          4. COGS + inventory reduction are posted

        Any exception rolls back all four. That atomicity IS the ERP.
        """
        order = await self._get_order(order_id)
        stock_service = StockService(self.session, self.tenant_id)

        try:
            totals = calculate_invoice([
                {
                    "quantity": line.quantity,
                    "unit_price": line.unit_price,
                    "discount_pct": line.discount_pct,
                    "tax_rate": line.tax_rate,
                }
                for line in order.lines
            ])

            # 1 — issue stock and accumulate its cost
            cost_of_goods = Decimal("0")
            for line in order.lines:
                await stock_service.reserve_and_issue(
                    line.product_id, order.warehouse_id,
                    line.quantity, ("sales_order", order.id),
                )
                cost_of_goods += await stock_service.moving_average_cost(
                    line.product_id
                ) * line.quantity

            # 2 — the invoice document
            invoice = await self._create_invoice(order, totals, invoice_date)

            # 3 — revenue side
            await post_entry(
                self.session, tenant_id=self.tenant_id, entry_date=invoice_date,
                narration=f"Sales invoice {invoice.invoice_number}",
                source=("invoice", invoice.id),
                lines=[
                    {"account_id": self.accounts["receivable"],
                     "debit": totals["grand_total"]},
                    {"account_id": self.accounts["revenue"],
                     "credit": totals["subtotal"]},
                    {"account_id": self.accounts["tax_payable"],
                     "credit": totals["tax_total"]},
                ],
            )

            # 4 — cost side (skipped when cost is unknown, e.g. opening stock)
            if cost_of_goods > 0:
                await post_entry(
                    self.session, tenant_id=self.tenant_id, entry_date=invoice_date,
                    narration=f"COGS for invoice {invoice.invoice_number}",
                    source=("invoice", invoice.id),
                    lines=[
                        {"account_id": self.accounts["cogs"], "debit": cost_of_goods},
                        {"account_id": self.accounts["inventory"], "credit": cost_of_goods},
                    ],
                )

            await self.session.commit()
            return {"invoice_id": invoice.id, "total": totals["grand_total"]}

        except Exception:
            await self.session.rollback()
            raise

from datetime import date

from sqlalchemy import func, select

from app.models.accounting import Account, AccountType, JournalEntry, JournalLine


class ReportRepository:
    def __init__(self, session, tenant_id: int) -> None:
        self.session = session
        self.tenant_id = tenant_id

    async def trial_balance(self, as_of: date) -> list[dict]:
        """Every account with its debit and credit totals.

        The grand totals MUST match. If they do not, something wrote journal
        lines without going through post_entry() — find it immediately.
        """
        query = (
            select(
                Account.code, Account.name, Account.account_type,
                func.coalesce(func.sum(JournalLine.debit), 0).label("debit"),
                func.coalesce(func.sum(JournalLine.credit), 0).label("credit"),
            )
            .join(JournalLine, JournalLine.account_id == Account.id)
            .join(JournalEntry, JournalEntry.id == JournalLine.entry_id)
            .where(
                Account.tenant_id == self.tenant_id,
                JournalEntry.entry_date <= as_of,
            )
            .group_by(Account.id)
            .order_by(Account.code)
        )
        rows = await self.session.execute(query)
        return [dict(r._mapping) for r in rows]

    async def profit_and_loss(self, start: date, end: date) -> dict:
        """Income less expenses for a period.

        Income accounts carry credit balances and expenses carry debit
        balances, so each is netted in its natural direction.
        """
        query = (
            select(
                Account.account_type,
                func.coalesce(func.sum(JournalLine.credit - JournalLine.debit), 0)
                .label("net"),
            )
            .join(JournalLine, JournalLine.account_id == Account.id)
            .join(JournalEntry, JournalEntry.id == JournalLine.entry_id)
            .where(
                Account.tenant_id == self.tenant_id,
                Account.account_type.in_([AccountType.INCOME, AccountType.EXPENSE]),
                JournalEntry.entry_date.between(start, end),
            )
            .group_by(Account.account_type)
        )
        results = {r.account_type: r.net for r in await self.session.execute(query)}

        income = results.get(AccountType.INCOME, 0)
        expenses = -results.get(AccountType.EXPENSE, 0)   # flip to positive
        return {
            "period": {"from": start, "to": end},
            "income": income,
            "expenses": expenses,
            "net_profit": income - expenses,
        }

"""Render an invoice to PDF.

Approach: render Jinja2 HTML, then convert. HTML is far easier to restyle
per client than laying out a PDF by coordinates, and every client wants
their own logo, colours and footer text.
"""
from pathlib import Path

from jinja2 import Environment, FileSystemLoader

env = Environment(
    loader=FileSystemLoader("app/templates"),
    autoescape=True,   # customer names may contain & or <
)


def render_invoice_html(invoice: dict, tenant: dict) -> str:
    template = env.get_template("invoice.html")
    return template.render(invoice=invoice, tenant=tenant)


def write_invoice_pdf(invoice: dict, tenant: dict, out_path: Path) -> Path:
    """Convert the rendered HTML to PDF.

    WeasyPrint is pure Python and handles CSS well, but needs GTK on
    Windows. If that becomes a classroom obstacle, ship the HTML and let
    the browser's Print-to-PDF do the work — for an SME invoice that is a
    perfectly acceptable v1.
    """
    from weasyprint import HTML

    html = render_invoice_html(invoice, tenant)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    HTML(string=html).write_pdf(str(out_path))
    return out_path

from datetime import date
from decimal import Decimal
from enum import StrEnum

from sqlalchemy import BigInteger, Boolean, Date, Enum, ForeignKey, Numeric, String
from sqlalchemy.orm import Mapped, mapped_column

from app.models.base import ERPBase


class AttendanceStatus(StrEnum):
    PRESENT = "present"
    ABSENT = "absent"
    LEAVE = "leave"
    HALF_DAY = "half_day"
    HOLIDAY = "holiday"


class Employee(ERPBase):
    __tablename__ = "employees"

    employee_code: Mapped[str] = mapped_column(String(32), nullable=False)
    full_name: Mapped[str] = mapped_column(String(255), nullable=False)
    cnic: Mapped[str | None] = mapped_column(String(15))   # 00000-0000000-0
    department: Mapped[str | None] = mapped_column(String(64))
    designation: Mapped[str | None] = mapped_column(String(64))
    joined_on: Mapped[date] = mapped_column(Date, nullable=False)
    left_on: Mapped[date | None] = mapped_column(Date)

    basic_salary: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    # Optional link — most employees never get a login.
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)


class Attendance(ERPBase):
    """One row per employee per day. Append-only in spirit: corrections are
    made by an authorised edit that writes an audit log entry."""

    __tablename__ = "attendance"

    employee_id: Mapped[int] = mapped_column(ForeignKey("employees.id"), nullable=False)
    attendance_date: Mapped[date] = mapped_column(Date, nullable=False)
    status: Mapped[AttendanceStatus] = mapped_column(
        Enum(AttendanceStatus), default=AttendanceStatus.PRESENT
    )
    hours_worked: Mapped[Decimal] = mapped_column(Numeric(6, 2), default=8)
    note: Mapped[str | None] = mapped_column(String(255))


class PayrollRun(ERPBase):
    """A month's payroll. Locked once posted so figures cannot drift after
    salaries have been paid and the ledger entry written."""

    __tablename__ = "payroll_runs"

    period_year: Mapped[int] = mapped_column(nullable=False)
    period_month: Mapped[int] = mapped_column(nullable=False)   # 1-12
    is_posted: Mapped[bool] = mapped_column(Boolean, default=False)
    posted_at: Mapped[date | None] = mapped_column(Date)


class Payslip(ERPBase):
    __tablename__ = "payslips"

    run_id: Mapped[int] = mapped_column(ForeignKey("payroll_runs.id"), nullable=False)
    employee_id: Mapped[int] = mapped_column(ForeignKey("employees.id"), nullable=False)

    basic: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    allowances: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    deductions: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    tax: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    net_pay: Mapped[Decimal] = mapped_column(Numeric(18, 4), default=0)
    days_present: Mapped[Decimal] = mapped_column(Numeric(5, 2), default=0)

"""Payroll calculation and posting.

Deliberately simple: basic salary pro-rated by attendance, plus allowances,
less deductions. Income tax slabs change with each Finance Act, so tax is
isolated in one function you can update annually without touching anything
else.
"""
from calendar import monthrange
from datetime import date
from decimal import Decimal

from app.services.invoice_calc import money
from app.services.ledger import post_entry


class PayrollLocked(Exception):
    """The run is already posted; salaries have been paid against it."""


class PayrollService:
    def __init__(self, session, tenant_id: int, accounts: dict[str, int]) -> None:
        self.session = session
        self.tenant_id = tenant_id
        self.accounts = accounts

    async def calculate_run(self, year: int, month: int) -> list[dict]:
        """Produce a DRAFT payroll. Nothing is posted; nothing is locked.

        The client reviews this, fixes attendance mistakes, and only then
        asks for it to be posted.
        """
        working_days = Decimal(str(monthrange(year, month)[1]))
        employees = await self._active_employees(year, month)

        draft = []
        for emp in employees:
            present = await self._days_present(emp.id, year, month)

            # Pro-rate on attendance. Whether unpaid absence should reduce
            # pay is a POLICY decision — confirm it with the client in
            # writing before implementing it.
            earned_basic = money(emp.basic_salary * present / working_days)
            allowances = await self._allowances(emp.id, year, month)
            gross = earned_basic + allowances
            tax = self._income_tax(gross)
            deductions = await self._deductions(emp.id, year, month)

            draft.append({
                "employee_id": emp.id,
                "employee_name": emp.full_name,
                "days_present": present,
                "basic": earned_basic,
                "allowances": allowances,
                "deductions": deductions,
                "tax": tax,
                "net_pay": money(gross - tax - deductions),
            })
        return draft

    def _income_tax(self, monthly_gross: Decimal) -> Decimal:
        """Monthly withholding tax.

        WARNING: slabs change with every Finance Act. This function is a
        placeholder to be replaced with the current year's rates, verified
        against FBR's published slabs, before any client uses it for real
        salaries. Do not let an agent invent tax rates.
        """
        return Decimal("0")   # implement per current Finance Act

    async def post_run(self, run_id: int, posting_date: date) -> dict:
        """Lock the run and write the accounting entry.

        Debit Salary Expense (gross), credit Salaries Payable (net) and
        Tax Payable (withheld). Balances by construction.
        """
        run = await self._get_run(run_id)
        if run.is_posted:
            raise PayrollLocked(f"Payroll {run.period_month}/{run.period_year} is posted")

        slips = await self._slips_for(run_id)
        gross = sum((s.basic + s.allowances for s in slips), Decimal("0"))
        tax = sum((s.tax for s in slips), Decimal("0"))
        net = sum((s.net_pay for s in slips), Decimal("0"))
        deductions = gross - tax - net

        try:
            lines = [
                {"account_id": self.accounts["salary_expense"], "debit": gross},
                {"account_id": self.accounts["salaries_payable"], "credit": net},
                {"account_id": self.accounts["tax_payable"], "credit": tax},
            ]
            if deductions > 0:
                lines.append(
                    {"account_id": self.accounts["other_payable"], "credit": deductions}
                )

            await post_entry(
                self.session, tenant_id=self.tenant_id, entry_date=posting_date,
                narration=f"Payroll {run.period_month:02d}/{run.period_year}",
                source=("payroll_run", run.id), lines=lines,
            )

            run.is_posted = True
            run.posted_at = posting_date
            await self.session.commit()
            return {"run_id": run.id, "gross": gross, "net": net, "employees": len(slips)}

        except Exception:
            await self.session.rollback()
            raise

# WRONG — loads every row into memory, then sums in Python.
# Works fine with 500 invoices; falls over at 50,000.
invoices = (await session.execute(select(Invoice))).scalars().all()
total = sum(i.grand_total for i in invoices)

# RIGHT — the database does the arithmetic and returns one number.
total = (
    await session.execute(
        select(func.coalesce(func.sum(Invoice.grand_total), 0))
        .where(Invoice.tenant_id == tenant_id,
               Invoice.invoice_date.between(start, end))
    )
).scalar_one()

class Invoice(ERPBase):
    __tablename__ = "invoices"
    __table_args__ = (
        # Column order matters: equality columns first, then the range
        # column. MySQL can use tenant_id for lookup and invoice_date for
        # the range scan only in this order.
        Index("ix_invoice_reporting", "tenant_id", "invoice_date"),
        Index("ix_invoice_customer", "tenant_id", "partner_id", "invoice_date"),
    )

from datetime import date, timedelta

from sqlalchemy import func, select


class DashboardRepository:
    def __init__(self, session, tenant_id: int) -> None:
        self.session = session
        self.tenant_id = tenant_id

    async def summary(self, on: date | None = None) -> dict:
        """Headline figures for the owner's home screen.

        Each is a single aggregate query. Running them concurrently with
        asyncio.gather() would be faster, but they share one session, and a
        SQLAlchemy AsyncSession is not safe for concurrent use. If this ever
        gets slow, give each query its own session — do not gather on one.
        """
        today = on or date.today()
        month_start = today.replace(day=1)
        week_start = today - timedelta(days=today.weekday())

        return {
            "as_of": today,
            "sales_this_month": await self._sales_between(month_start, today),
            "sales_this_week": await self._sales_between(week_start, today),
            "receivables_outstanding": await self._outstanding_receivables(),
            "overdue_invoices": await self._overdue_count(today),
            "low_stock_items": await self._low_stock_count(),
            "top_products": await self._top_products(month_start, today, limit=5),
        }

    async def _top_products(self, start: date, end: date, limit: int) -> list[dict]:
        from app.models.master import Product
        from app.models.sales import Invoice, InvoiceLine

        revenue = func.sum(InvoiceLine.net_amount).label("revenue")
        query = (
            select(Product.sku, Product.name, revenue,
                   func.sum(InvoiceLine.quantity).label("units"))
            .join(InvoiceLine, InvoiceLine.product_id == Product.id)
            .join(Invoice, Invoice.id == InvoiceLine.invoice_id)
            .where(Invoice.tenant_id == self.tenant_id,
                   Invoice.invoice_date.between(start, end))
            .group_by(Product.id)
            .order_by(revenue.desc())
            .limit(limit)
        )
        return [dict(r._mapping) for r in await self.session.execute(query)]

"""Stream query results to Excel without buffering the whole file."""
from io import BytesIO

from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill
from openpyxl.utils import get_column_letter


def rows_to_xlsx(rows: list[dict], sheet_name: str = "Report") -> BytesIO:
    """Turn a list of dicts into a formatted worksheet.

    write_only mode keeps memory flat for large exports, which matters when
    the accountant asks for three years of transactions.
    """
    wb = Workbook()
    ws = wb.active
    ws.title = sheet_name[:31]   # Excel's hard limit on sheet names

    if not rows:
        ws.append(["No data for the selected period"])
        buffer = BytesIO()
        wb.save(buffer)
        buffer.seek(0)
        return buffer

    headers = list(rows[0].keys())
    ws.append([h.replace("_", " ").title() for h in headers])

    header_fill = PatternFill("solid", fgColor="003135")   # AIBYTEC teal
    for cell in ws[1]:
        cell.font = Font(bold=True, color="FFFFFF")
        cell.fill = header_fill

    for row in rows:
        ws.append([row.get(h) for h in headers])

    # Width by content, capped so one long address does not stretch a column.
    for i, header in enumerate(headers, start=1):
        longest = max([len(str(header))] + [len(str(r.get(header, ""))) for r in rows])
        ws.column_dimensions[get_column_letter(i)].width = min(longest + 2, 40)

    ws.freeze_panes = "A2"   # header stays visible while scrolling

    buffer = BytesIO()
    wb.save(buffer)
    buffer.seek(0)
    return buffer

# AiBytec ERP — Frontend

## Stack
Next.js (App Router) + TypeScript (strict) + Tailwind + TanStack Query.
No other UI or state library without asking.

## Non-negotiable
1. Every API call goes through lib/api.ts. No fetch() in components.
2. Server state = TanStack Query. Local state = useState. No global store.
3. All money rendered via formatPKR() from lib/format.ts. Never toFixed().
4. Every list screen: loading state, empty state, error state. All three.
5. Tables use components/DataTable.tsx. Do not write a new <table>.
6. Buttons use components/Button.tsx variants. Do not restyle inline.
7. No `any`. If a type is unknown, generate it from the OpenAPI schema.

## Structure
app/(dashboard)/{module}/page.tsx      list
app/(dashboard)/{module}/[id]/page.tsx detail
components/                            shared, presentational only
lib/api.ts                             typed API client
lib/format.ts                          money, date, number formatting

## Permissions
Hide actions the user cannot perform, using usePermission('invoice:create').
Hiding is cosmetic — the API enforces it for real.

/**
 * Single entry point for every API call.
 *
 * Centralising this means auth headers, error shape, and the base URL are
 * defined once. Rule 1 in CLAUDE.md exists to protect this file's monopoly.
 */
import type { paths } from "./api-types";

const BASE = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:8000";

export class ApiError extends Error {
  constructor(
    public status: number,
    message: string,
    public problems?: { field: string; message: string }[],
  ) {
    super(message);
  }
}

async function request<T>(path: string, init: RequestInit = {}): Promise<T> {
  const token = typeof window !== "undefined"
    ? window.sessionStorage.getItem("erp_token")
    : null;

  const res = await fetch(`${BASE}${path}`, {
    ...init,
    headers: {
      "Content-Type": "application/json",
      ...(token ? { Authorization: `Bearer ${token}` } : {}),
      ...init.headers,
    },
  });

  if (!res.ok) {
    // Matches the flattened validation shape from Chapter 4.
    const body = await res.json().catch(() => ({ detail: res.statusText }));
    throw new ApiError(res.status, body.detail ?? "Request failed", body.problems);
  }

  return res.status === 204 ? (undefined as T) : ((await res.json()) as T);
}

export const api = {
  get: <T>(p: string) => request<T>(p),
  post: <T>(p: string, body: unknown) =>
    request<T>(p, { method: "POST", body: JSON.stringify(body) }),
  patch: <T>(p: string, body: unknown) =>
    request<T>(p, { method: "PATCH", body: JSON.stringify(body) }),
  delete: <T>(p: string) => request<T>(p, { method: "DELETE" }),
};

/**
 * Formatting helpers. Rule 3 of CLAUDE.md: all money goes through formatPKR.
 *
 * Without this, half the screens show "15000", a quarter show "15,000.00",
 * and one shows "15000.000000000002" because someone used a float.
 */
export function formatPKR(value: number | string): string {
  const n = typeof value === "string" ? Number.parseFloat(value) : value;
  if (!Number.isFinite(n)) return "—";
  return new Intl.NumberFormat("en-PK", {
    style: "currency",
    currency: "PKR",
    minimumFractionDigits: 0,
    maximumFractionDigits: 2,
  }).format(n);
}

export function formatDate(iso: string): string {
  return new Intl.DateTimeFormat("en-PK", {
    day: "2-digit", month: "short", year: "numeric",
  }).format(new Date(iso));
}

"use client";

import { useQuery } from "@tanstack/react-query";
import { useState } from "react";

import { DataTable } from "@/components/DataTable";
import { Button } from "@/components/Button";
import { api } from "@/lib/api";
import { formatPKR } from "@/lib/format";
import { usePermission } from "@/lib/permissions";

type Product = {
  id: number; sku: string; name: string;
  uom: string; sale_price: string; is_active: boolean;
};

export default function ProductsPage() {
  const [search, setSearch] = useState("");
  const canCreate = usePermission("product:create");

  const { data, isLoading, error } = useQuery({
    queryKey: ["products", search],
    queryFn: () =>
      api.get<Product[]>(`/api/v1/products?q=${encodeURIComponent(search)}`),
  });

  return (
    <div className="space-y-4">
      <div className="flex items-center justify-between">
        <h1 className="text-2xl font-semibold text-slate-900">Products</h1>
        {/* Cosmetic only — the API enforces this permission for real. */}
        {canCreate && <Button href="/products/new">New Product</Button>}
      </div>

      <input
        value={search}
        onChange={(e) => setSearch(e.target.value)}
        placeholder="Search SKU or name…"
        className="w-full max-w-sm rounded-lg border border-slate-300 px-3 py-2"
      />

      {/* Rule 4: all three states, every time. */}
      <DataTable
        rows={data ?? []}
        isLoading={isLoading}
        error={error}
        emptyMessage="No products yet. Add your first one to get started."
        columns={[
          { key: "sku", header: "SKU" },
          { key: "name", header: "Name" },
          { key: "uom", header: "Unit" },
          {
            key: "sale_price",
            header: "Sale Price",
            align: "right",
            render: (p: Product) => formatPKR(p.sale_price),
          },
        ]}
        rowHref={(p: Product) => `/products/${p.id}`}
      />
    </div>
  );
}

---
name: erp-module
description: Scaffold a complete ERP module — model, schema, repository,
  service, router, tests, and the Next.js list and detail screens. Use
  whenever the user asks to add a new module or entity to the ERP.
---

# ERP Module Scaffold

## Before writing anything
1. Read app/models/master.py, app/api/v1/products.py and
   web/app/(dashboard)/products/page.tsx as the reference implementations.
2. List your assumptions about fields, relationships and permissions.
3. Wait for approval.

## Backend order
1. `app/models/{module}.py` — inherit ERPBase. Never add tenant_id by hand.
2. `app/schemas/{module}.py` — Create / Update / Read. Read excludes cost fields.
3. `app/repositories/{module}.py` — extend TenantRepository. Never bypass
   `_base_query()`.
4. `app/services/{module}_service.py` — business rules. Raise domain errors,
   never HTTPException. The service owns the commit.
5. `app/api/v1/{module}.py` — thin handlers. Translate domain errors to HTTP.
   Add `Depends(require("{module}:action"))` to every mutating endpoint.
6. `tests/test_{module}.py` — must include a tenant-isolation test.
7. Register the router in `app/main.py`.
8. `alembic revision --autogenerate` — then STOP and show me the migration.

## Frontend order
9.  Regenerate `web/lib/api-types.ts` from the OpenAPI schema.
10. List page at `web/app/(dashboard)/{module}/page.tsx` using DataTable.
11. Detail page at `.../[id]/page.tsx`.
12. All money through formatPKR(). All three states on every list.

## Definition of done
- `pytest -q` passes, including the isolation test
- `tsc --noEmit` passes with no `any`
- The migration has been read by a human
- `grep -rn "fetch(" web/app/` returns nothing

"""MCP server exposing AiBytec ERP as tools.

Reuses the service layer from Chapters 4-8 directly. Because services never
knew about HTTP, they work unchanged here — that separation, decided in
Chapter 4, is what makes this file short.
"""
import os
from datetime import date, timedelta
from decimal import Decimal

from fastmcp import FastMCP

from app.database import SessionLocal
from app.repositories.dashboard import DashboardRepository
from app.repositories.stock import StockRepository

mcp = FastMCP("aibytec-erp")

# The tenant comes from the server's environment, exactly as it comes from
# the JWT in the API. The model cannot choose which company it is querying.
TENANT_ID = int(os.environ["ERP_TENANT_ID"])


@mcp.tool
async def get_stock_level(sku: str, warehouse: str | None = None) -> dict:
    """Current stock on hand for a product.

    Args:
        sku: Product SKU, e.g. "A-100". Case-insensitive.
        warehouse: Optional warehouse code. Omit for all warehouses.

    Returns exact figures from the stock ledger — never an estimate.
    """
    async with SessionLocal() as session:
        repo = StockRepository(session, TENANT_ID)
        product = await repo.product_by_sku(sku.strip().upper())
        if product is None:
            # A clear, actionable failure. The model relays this to the user
            # rather than inventing a plausible number.
            return {"error": f"No product with SKU '{sku}'"}

        wh_id = await repo.warehouse_id_by_code(warehouse) if warehouse else None
        qty = await repo.on_hand(product.id, wh_id)
        return {
            "sku": product.sku,
            "name": product.name,
            "on_hand": str(qty),
            "uom": product.uom,
            "warehouse": warehouse or "all",
            "reorder_level": str(product.reorder_level),
            "below_reorder": qty <= product.reorder_level,
        }


@mcp.tool
async def list_low_stock() -> list[dict]:
    """Products at or below their reorder level. Use this to answer
    'what do we need to order?' or 'what is running out?'"""
    async with SessionLocal() as session:
        rows = await StockRepository(session, TENANT_ID).below_reorder_level()
        return [{k: str(v) for k, v in row.items()} for row in rows]


@mcp.tool
async def sales_summary(days: int = 30) -> dict:
    """Sales totals for the last N days, with the top-selling products.

    Args:
        days: Look-back window. Capped at 365 to protect the database.
    """
    days = max(1, min(days, 365))
    end = date.today()
    start = end - timedelta(days=days)

    async with SessionLocal() as session:
        repo = DashboardRepository(session, TENANT_ID)
        return {
            "period": f"{start} to {end}",
            "total_sales_pkr": str(await repo._sales_between(start, end)),
            "top_products": await repo._top_products(start, end, limit=5),
        }


@mcp.tool
async def draft_purchase_order(sku: str, quantity: float, supplier_code: str) -> dict:
    """Create a DRAFT purchase order. It is NOT sent to the supplier.

    A human must review and confirm it in the ERP before anything is
    ordered. This tool deliberately cannot confirm — write actions with
    real-world consequences always stop at draft.
    """
    async with SessionLocal() as session:
        from app.services.purchase_service import PurchaseService

        service = PurchaseService(session, TENANT_ID, user_id=0)
        po = await service.create_draft(
            sku=sku.strip().upper(),
            quantity=Decimal(str(quantity)),
            supplier_code=supplier_code,
        )
        return {
            "po_number": po.po_number,
            "status": "draft",
            "message": "Draft created. Open the ERP to review and confirm it.",
        }


if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "aibytec-erp": {
      "command": "python",
      "args": ["-m", "mcp_server.erp_server"],
      "env": {
        "ERP_TENANT_ID": "1",
        "DATABASE_URL": "mysql+aiomysql://erp_app:pass@localhost:3306/aibytec_erp"
      }
    }
  }
}

"""RAG over client documents — policies, contracts, supplier terms.

Design decisions that matter:
  * tenant_id is a payload filter on EVERY search. A tenant must never
    retrieve another tenant's contract.
  * Retrieved chunks are returned to the caller and shown in the UI, so a
    user can see WHERE an answer came from. In a teaching build this is
    essential; in a client build it is what earns trust.
"""
from dataclasses import dataclass

from openai import AsyncOpenAI
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import (
    Distance, FieldCondition, Filter, MatchValue, PointStruct, VectorParams,
)

COLLECTION = "erp_documents"
EMBED_MODEL = "text-embedding-3-small"   # 1536 dims, inexpensive
VECTOR_SIZE = 1536

openai = AsyncOpenAI()
qdrant = AsyncQdrantClient(url="http://localhost:6333")


@dataclass
class Chunk:
    text: str
    source: str
    score: float


async def ensure_collection() -> None:
    existing = {c.name for c in (await qdrant.get_collections()).collections}
    if COLLECTION not in existing:
        await qdrant.create_collection(
            collection_name=COLLECTION,
            vectors_config=VectorParams(size=VECTOR_SIZE, distance=Distance.COSINE),
        )


def chunk_text(text: str, size: int = 900, overlap: int = 150) -> list[str]:
    """Split on paragraph boundaries where possible.

    Overlap prevents an answer being cut in half at a chunk boundary, which
    is the most common cause of 'the document says it but RAG cannot find it'.
    """
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks: list[str] = []
    current = ""

    for para in paragraphs:
        if len(current) + len(para) < size:
            current += ("\n\n" if current else "") + para
        else:
            if current:
                chunks.append(current)
            current = (current[-overlap:] + "\n\n" + para) if current else para
    if current:
        chunks.append(current)
    return chunks


async def index_document(tenant_id: int, source: str, text: str) -> int:
    """Embed and store one document. Returns the number of chunks stored."""
    await ensure_collection()
    chunks = chunk_text(text)

    # Batch the embedding call — one request per chunk is slow and costly.
    response = await openai.embeddings.create(model=EMBED_MODEL, input=chunks)

    points = [
        PointStruct(
            id=abs(hash((tenant_id, source, i))) % (10**18),
            vector=item.embedding,
            payload={"tenant_id": tenant_id, "source": source,
                     "chunk_index": i, "text": chunk},
        )
        for i, (chunk, item) in enumerate(zip(chunks, response.data))
    ]
    await qdrant.upsert(collection_name=COLLECTION, points=points)
    return len(points)


async def search(tenant_id: int, query: str, top_k: int = 4) -> list[Chunk]:
    """Retrieve the most relevant chunks for this tenant only."""
    embedded = await openai.embeddings.create(model=EMBED_MODEL, input=[query])

    hits = await qdrant.query_points(
        collection_name=COLLECTION,
        query=embedded.data[0].embedding,
        limit=top_k,
        # The isolation boundary, repeated at the vector store.
        query_filter=Filter(
            must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))]
        ),
    )
    return [
        Chunk(text=p.payload["text"], source=p.payload["source"], score=p.score)
        for p in hits.points
    ]

"""The 'Ask your ERP' assistant.

Grounding rules, in priority order:
  1. Numbers come from tools (SQL). Never from retrieved text.
  2. Policy answers come from retrieved chunks, and cite their source.
  3. If neither has the answer, say so. Do not guess.
"""
from openai import AsyncOpenAI

from app.ai.knowledge_base import search

openai = AsyncOpenAI()

SYSTEM_PROMPT = """You are the assistant for a Pakistani trading company's ERP.

Rules:
1. For stock, sales, or money figures, ALWAYS use a tool. Never estimate,
   and never take a number from a document excerpt.
2. For policy or procedure questions, use only the provided document
   excerpts, and name the source file in your answer.
3. If the tools and excerpts do not contain the answer, say exactly that.
   A wrong number in an ERP is worse than no number.
4. Reply in the SAME language as the question. If the user writes in Urdu
   or Roman Urdu, reply in Urdu.
5. Format amounts as PKR with thousands separators.
6. Be brief. Users are busy.
"""


async def ask(tenant_id: int, question: str, tools: list[dict]) -> dict:
    """Answer a question using ERP tools plus document retrieval.

    Returns the answer AND the retrieved chunks, so the UI can show what
    the answer was based on. Showing retrieval is what makes users trust it.
    """
    chunks = await search(tenant_id, question, top_k=4)

    context = "\n\n---\n\n".join(
        f"[source: {c.source}]\n{c.text}" for c in chunks
    ) or "No relevant documents found."

    response = await openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "system", "content": f"Document excerpts:\n{context}"},
            {"role": "user", "content": question},
        ],
        tools=tools,
        temperature=0,   # deterministic: an ERP must not vary its answers
    )

    return {
        "answer": response.choices[0].message.content,
        "retrieved": [
            {"source": c.source, "score": round(c.score, 3),
             "preview": c.text[:200]}
            for c in chunks
        ],
    }

services:
  db:
    image: mysql:8.4
    restart: unless-stopped
    environment:
      MYSQL_DATABASE: aibytec_erp
      MYSQL_USER: erp_app
      MYSQL_PASSWORD_FILE: /run/secrets/db_password
      MYSQL_ROOT_PASSWORD_FILE: /run/secrets/db_root_password
    command:
      - --character-set-server=utf8mb4
      - --collation-server=utf8mb4_unicode_ci
      - --innodb-buffer-pool-size=1G      # tune to ~60% of available RAM
    volumes:
      - db_data:/var/lib/mysql
    secrets: [db_password, db_root_password]
    # NOTE: no ports section. The database is reachable only from the
    # compose network, never from the host or the internet.
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
      interval: 10s
      timeout: 5s
      retries: 10

  api:
    build: .
    restart: unless-stopped
    depends_on:
      db: { condition: service_healthy }   # wait for real readiness
    environment:
      DATABASE_URL: mysql+aiomysql://erp_app:${DB_PASSWORD}@db:3306/aibytec_erp?charset=utf8mb4
      ENVIRONMENT: production
    ports: ["8000:8000"]
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request;
              urllib.request.urlopen('http://localhost:8000/health')"]
      interval: 30s

  qdrant:
    image: qdrant/qdrant:latest
    restart: unless-stopped
    volumes: [qdrant_data:/qdrant/storage]

volumes:
  db_data:
  qdrant_data:

secrets:
  db_password:
    file: ./secrets/db_password.txt
  db_root_password:
    file: ./secrets/db_root_password.txt

FROM python:3.12-slim

# Build dependencies for bcrypt and friends, removed in the same layer.
RUN apt-get update \
 && apt-get install -y --no-install-recommends build-essential \
 && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Copy requirements first so pip's layer caches across code changes.
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Never run the application as root.
RUN useradd --create-home appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

# Migrations run at startup so a deploy can never leave code and schema
# out of step. Keep migrations backwards-compatible for one release.
CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"]

name: CI

on:
  push: { branches: [main] }
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      mysql:
        image: mysql:8.4
        env:
          MYSQL_ROOT_PASSWORD: testpass
          MYSQL_DATABASE: aibytec_erp_test
        ports: ["3306:3306"]
        options: >-
          --health-cmd="mysqladmin ping -h localhost"
          --health-interval=10s --health-retries=10

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12", cache: pip }

      - run: pip install -r requirements.txt

      - name: Run tests
        env:
          DATABASE_URL: mysql+aiomysql://root:testpass@127.0.0.1:3306/aibytec_erp_test
          SECRET_KEY: ci-only-not-a-real-secret
        run: pytest -q

      # These two run separately so a failure names the actual problem
      # rather than being buried in the general test output.
      - name: Tenant isolation must pass
        run: pytest tests/test_tenant_isolation.py -v

      - name: Audit dependencies
        run: pip install pip-audit && pip-audit