"""Statement Period Persister."""

from typing import List, Optional, Tuple

from sqlalchemy import UnaryExpression, and_, func, insert, or_, select, update
from sqlalchemy.orm import aliased
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.functions import coalesce

from collaborator.connectors import mysql
from collaborator.constants import error
from collaborator.constants.statement_period import StatementPeriodStatus
from collaborator.constants.transaction import (
    TYPE_BALANCE_CLEARING,
    TYPE_CREDIT,
    TYPE_DIRECT_PAYMENT,
    TYPE_EXPENSE,
    TYPE_PAYMENT,
    TYPE_PAYMENT_FEES,
    TYPE_REVENUE,
    TYPE_WHT_ALLOCATION,
)
from collaborator.models.ows import ows_royalties
from collaborator.models.rds.collaborator import Collaborator
from collaborator.models.rds.statement_period import StatementPeriod
from collaborator.models.rds.transaction import Transaction
from collaborator.utils.error import OwsError


class StatementPeriodPersister(object):
    """Statement Period persister class."""

    @classmethod
    def create_open_statement_period(
        cls, vendor_id: int, session: Session
    ) -> StatementPeriod:
        """Create open statement period."""
        abacus_statement_period = ows_royalties.get_current_abacus_statement_period()
        new_period = StatementPeriod(
            vendor_id=vendor_id,
            abacus_statement_period_id=abacus_statement_period["statement_period_id"],
        )
        session.add(new_period)
        session.commit()
        session.refresh(new_period)
        return new_period

    @classmethod
    @mysql.db_session
    def get_statement_period_by_id(
        cls, statement_period_id: int, session: Session
    ) -> Optional[StatementPeriod]:
        """Get statement period by id."""
        query = session.query(StatementPeriod).filter_by(
            statement_period_id=statement_period_id
        )

        return query.first()

    @classmethod
    @mysql.db_session
    def get_statement_periods_by_id(
        cls, statement_period_ids: List[int], session: Session
    ):
        """Get statement periods by ID."""
        query = select(
            StatementPeriod.statement_period_id,
            StatementPeriod.name,
            StatementPeriod.vendor_id,
            StatementPeriod.status,
            StatementPeriod.created_date,
            StatementPeriod.updated_date,
        ).where(StatementPeriod.statement_period_id.in_(statement_period_ids))

        return session.execute(query).all()

    @classmethod
    @mysql.db_session
    def get_statement_periods(
        cls,
        vendor_id: int,
        statuses: List[str],
        limit: int,
        offset: int,
        sort_key: Optional[str],
        sort_direction: Optional[str],
        term: Optional[str],
        session: Session,
    ):
        """Get statement periods for a vendor."""
        filters = [
            StatementPeriod.vendor_id == vendor_id,
        ]

        if len(statuses):
            filters.append(StatementPeriod.status.in_(statuses))

        if term:
            filters.append(StatementPeriod.name.ilike(f"%{term}%"))

        base_query = select(StatementPeriod).where(*filters)
        rows_query = base_query

        if limit:
            rows_query = rows_query.limit(limit)

        order = (
            getattr(StatementPeriod, sort_key)
            if sort_key
            else StatementPeriod.statement_period_id
        )
        order = order.desc() if sort_direction == "DESC" else order.asc()
        rows_query = rows_query.order_by(order)

        rows_query = rows_query.offset(offset)
        rows = session.execute(rows_query).scalars().all()

        count_query = base_query.with_only_columns(func.count())
        count = session.execute(count_query).scalars().one()

        return [statement_period.to_dict() for statement_period in rows], count

    @classmethod
    @mysql.db_session
    def get_open_statement_periods(
        cls, account_ids: List[int], session: Session
    ) -> dict[int, StatementPeriod]:
        """Get open statement periods."""
        query = session.query(StatementPeriod).filter(
            StatementPeriod.vendor_id.in_(account_ids),
            StatementPeriod.status == StatementPeriodStatus.OPEN,
        )

        rows = query.all()
        result = {period.vendor_id: period for row in rows for period in (row,)}

        return result

    @classmethod
    @mysql.db_session
    def get_open_statement_period(
        cls, account_id: int, session: Session
    ) -> Optional[StatementPeriod]:
        """Get open statement period."""
        periods_by_account_id = cls.get_open_statement_periods(
            account_ids=[account_id], session=session
        )
        period = next(iter(periods_by_account_id.values()), None)
        return period

    @classmethod
    @mysql.db_session
    def get_or_create_open_statement_period(
        cls, account_id: int, session: Session
    ) -> StatementPeriod:
        """Get or create the open statement period."""
        period = cls.get_open_statement_period(account_id)
        if period is None:
            period = cls.create_open_statement_period(
                vendor_id=account_id, session=session
            )
        return period

    @classmethod
    @mysql.db_session
    def close_statement_period(
        cls, vendor_id: int, period_name: str, session: Session
    ) -> Tuple[dict, dict]:
        """Close a statement period."""
        # Get the period and close it
        period_to_close = (
            session.query(StatementPeriod)
            .filter(
                StatementPeriod.vendor_id == vendor_id,
                StatementPeriod.status == StatementPeriodStatus.OPEN,
            )
            .first()
        )

        if not period_to_close:
            raise OwsError.not_found(
                code=error.ERROR_CODE_STATEMENT_PERIOD_NOT_FOUND,
                message=error.ERROR_MESSAGE_STATEMENT_PERIOD_NOT_FOUND,
            )

        period_to_close.status = StatementPeriodStatus.CLOSED
        period_to_close.name = period_name

        new_open_period = cls.create_open_statement_period(
            vendor_id=vendor_id, session=session
        )
        session.commit()
        return period_to_close.to_dict(), new_open_period.to_dict()

    @classmethod
    @mysql.db_session
    def bulk_close_statement_periods(
        cls,
        period_name: str,
        dp_enabled_vendor_ids: str,
        new_abacus_statement_period_id: int,
        session: Session,
    ):
        """Close all open statement periods."""
        # Get vendors with open statement periods with transactions
        to_close = (
            session.execute(
                select(StatementPeriod)
                .join(
                    Transaction,
                    and_(
                        Transaction.statement_period_id
                        == StatementPeriod.statement_period_id,
                        Transaction.deleted_date == None,  # noqa
                    ),
                )
                .where(
                    StatementPeriod.status == StatementPeriodStatus.OPEN,
                    StatementPeriod.vendor_id.in_(dp_enabled_vendor_ids),
                )
            )
            .scalars()
            .all()
        )

        if len(to_close) == 0:
            return {}

        closed_periods_by_vendor_id = {
            period.vendor_id: period.to_dict() for period in to_close
        }

        # Close open statement periods
        session.execute(
            update(StatementPeriod)
            .values(status=StatementPeriodStatus.CLOSED, name=period_name)
            .where(
                StatementPeriod.statement_period_id.in_(
                    [period["id"] for period in closed_periods_by_vendor_id.values()]
                ),
            )
        )

        # Create new open statement periods.
        # We don't use `create_open_statement_period` here because that function
        # creates the OPEN period with the _current_ Abacus statement period's ID.
        # Bulk-closing collabs periods happens prior to the Abacus period closes,
        # so `new_abacus_statement_period_id` is what the Abacus statement period
        # _will_ be after the "current" one closes.
        session.execute(
            insert(StatementPeriod),
            [
                {
                    "vendor_id": vendor_id,
                    "abacus_statement_period_id": new_abacus_statement_period_id,
                }
                for vendor_id in closed_periods_by_vendor_id.keys()
            ],
        )

        return closed_periods_by_vendor_id

    @classmethod
    def _transactions_sum_subquery(
        cls, transaction_types: List[str], for_collaborator: Optional[bool] = False
    ):
        filters = [
            (Transaction.statement_period_id == StatementPeriod.statement_period_id),
            Transaction.transaction_type.in_(transaction_types),
            Transaction.deleted_date == None,  # noqa
        ]

        if for_collaborator:
            filters.append(Transaction.collaborator_id == Collaborator.collaborator_id)

        return coalesce(
            select(func.sum(Transaction.chargeable_amount))
            .where(*filters)
            .scalar_subquery(),
            0,
        )

    @classmethod
    @mysql.db_session
    def get_vendor_totals_for_statement_periods(
        cls, statement_period_ids: List[int], session: Session
    ):
        """Get vendor-level totals for statement periods by ID."""
        query = select(
            StatementPeriod.statement_period_id,
            StatementPeriod.vendor_id,
            # revenues_total
            cls._transactions_sum_subquery([TYPE_REVENUE]).label("revenues_total"),
            # expenses_total
            cls._transactions_sum_subquery(
                [
                    TYPE_EXPENSE,
                    TYPE_WHT_ALLOCATION,
                    TYPE_BALANCE_CLEARING,
                    TYPE_PAYMENT_FEES,
                ]
            ).label("expenses_total"),
            # payments_total
            cls._transactions_sum_subquery(
                [
                    TYPE_PAYMENT,
                    TYPE_DIRECT_PAYMENT,
                    TYPE_CREDIT,
                ]
            ).label("payments_total"),
            # currency
            select(Transaction.currency.distinct())
            .where(
                Transaction.statement_period_id == StatementPeriod.statement_period_id,
                Transaction.deleted_date == None,  # noqa
            )
            .limit(1)
            .scalar_subquery()
            .label("currency"),
            # currencies_count
            select(func.count(Transaction.currency.distinct()))
            .where(
                Transaction.statement_period_id == StatementPeriod.statement_period_id,
                Transaction.deleted_date == None,  # noqa
            )
            .scalar_subquery()
            .label("currencies_count"),
        ).where(StatementPeriod.statement_period_id.in_(statement_period_ids))

        return session.execute(query).all()

    @classmethod
    @mysql.db_session
    def get_statement_period_participations(
        cls,
        collaborator_id: Optional[int],
        statement_period_id: Optional[int],
        status: Optional[str],
        limit: Optional[int],
        offset: Optional[int],
        from_first_activity: Optional[bool],
        term: Optional[str],
        session: Session,
    ):
        """Get statement period participations."""
        filters = []

        StatementPeriod2 = aliased(StatementPeriod)

        order: UnaryExpression

        if collaborator_id:
            filters.append(Collaborator.collaborator_id == collaborator_id)
            order = StatementPeriod.created_date.desc()

        if statement_period_id:
            filters.append(StatementPeriod.statement_period_id == statement_period_id)
            order = Collaborator.name.asc()

        if status:
            filters.append(StatementPeriod.status == status)

        if term:
            filters.append(Collaborator.name.ilike(f"%{term}%"))

        if from_first_activity and collaborator_id:
            filters.append(
                StatementPeriod.created_date
                >= (
                    select(StatementPeriod.created_date)
                    .join(
                        Transaction,
                        StatementPeriod.statement_period_id
                        == Transaction.statement_period_id,
                    )
                    .where(Transaction.collaborator_id == collaborator_id)
                    .group_by(StatementPeriod.statement_period_id)
                    .order_by(StatementPeriod.created_date.asc())
                    .limit(1)
                    .scalar_subquery()
                )
            )

        rows_query = (
            select(
                StatementPeriod.statement_period_id,
                StatementPeriod.vendor_id,
                Collaborator.collaborator_id,
                # revenues_total
                cls._transactions_sum_subquery(
                    [TYPE_REVENUE],
                    for_collaborator=True,
                ).label("revenues_total"),
                # expenses_total
                cls._transactions_sum_subquery(
                    [
                        TYPE_EXPENSE,
                        TYPE_WHT_ALLOCATION,
                        TYPE_BALANCE_CLEARING,
                        TYPE_PAYMENT_FEES,
                    ],
                    for_collaborator=True,
                ).label("expenses_total"),
                # payments_total
                cls._transactions_sum_subquery(
                    [
                        TYPE_PAYMENT,
                        TYPE_DIRECT_PAYMENT,
                        TYPE_CREDIT,
                    ],
                    for_collaborator=True,
                ).label("payments_total"),
                # opening_balance
                coalesce(
                    select(func.sum(Transaction.chargeable_amount))
                    .where(
                        Transaction.statement_period_id.in_(
                            select(StatementPeriod2.statement_period_id)
                            .where(
                                StatementPeriod2.created_date
                                < StatementPeriod.created_date,
                                StatementPeriod2.vendor_id == StatementPeriod.vendor_id,
                            )
                            .scalar_subquery()
                            .correlate(StatementPeriod),
                        ),
                        Transaction.collaborator_id == Collaborator.collaborator_id,
                        Transaction.deleted_date == None,  # noqa
                    )
                    .scalar_subquery(),
                    0,
                ).label("opening_balance"),
                # closing_balance
                coalesce(
                    select(func.sum(Transaction.chargeable_amount))
                    .where(
                        Transaction.statement_period_id.in_(
                            select(StatementPeriod2.statement_period_id)
                            .where(
                                or_(
                                    and_(
                                        StatementPeriod2.created_date
                                        < StatementPeriod.created_date,
                                        StatementPeriod2.vendor_id
                                        == StatementPeriod.vendor_id,
                                    ),
                                    StatementPeriod2.statement_period_id
                                    == StatementPeriod.statement_period_id,
                                )
                            )
                            .scalar_subquery()
                            .correlate(StatementPeriod),
                        ),
                        Transaction.collaborator_id == Collaborator.collaborator_id,
                        Transaction.deleted_date == None,  # noqa
                    )
                    .scalar_subquery(),
                    0,
                ).label("closing_balance"),
                # currency
                select(Transaction.currency.distinct())
                .where(
                    Transaction.statement_period_id
                    == StatementPeriod.statement_period_id,
                    Transaction.collaborator_id == Collaborator.collaborator_id,
                    Transaction.deleted_date == None,  # noqa: E711
                )
                .limit(1)
                .scalar_subquery()
                .label("currency"),
                # currencies_count
                select(func.count(Transaction.currency.distinct()))
                .where(
                    Transaction.statement_period_id
                    == StatementPeriod.statement_period_id,
                    Transaction.collaborator_id == Collaborator.collaborator_id,
                    Transaction.deleted_date == None,  # noqa
                )
                .scalar_subquery()
                .label("currencies_count"),
            )
            .join(Collaborator, StatementPeriod.vendor_id == Collaborator.vendor_id)
            .where(*filters)
            .order_by(order)
        )

        if limit:
            rows_query = rows_query.limit(limit)

        if offset:
            rows_query = rows_query.offset(offset)

        rows = session.execute(rows_query).all()

        count_query = (
            select(
                StatementPeriod.statement_period_id,
            )
            .join(Collaborator, StatementPeriod.vendor_id == Collaborator.vendor_id)
            .where(*filters)
            .with_only_columns(func.count())
        )

        count = session.execute(count_query).scalars().one()

        return rows, count