"""Data access for the billing domain. **Only this module may touch bl_* tables.**

架构文档 §3.2 硬性规则 1: 每个域只能通过对方暴露的 service 接口调用，禁止直接
查对方的表。import-linter enforces it (contract "no cross-domain repository access").
"""

from __future__ import annotations

from datetime import datetime

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from packages.shared.models import Ledger, Quota, QuotaRule, RpFeedback, RpReport


async def get_rule(session: AsyncSession, quota_key: str) -> QuotaRule | None:
    return await session.get(QuotaRule, quota_key)


async def lock_or_create_quota(
    session: AsyncSession,
    *,
    user_id: int,
    quota_key: str,
    period_key: str,
    limit_value: int,
) -> Quota:
    """Get the period row with a row lock, creating it on first use.

    ``ON CONFLICT DO NOTHING`` + ``SELECT ... FOR UPDATE`` makes concurrent first
    requests for the same period safe without an advisory lock.
    """
    await session.execute(
        pg_insert(Quota)
        .values(
            user_id=user_id,
            quota_key=quota_key,
            period_key=period_key,
            used=0,
            limit_value=limit_value,
            bonus_granted=0,
        )
        .on_conflict_do_nothing(index_elements=["user_id", "quota_key", "period_key"])
    )
    result = await session.execute(
        select(Quota)
        .where(
            Quota.user_id == user_id,
            Quota.quota_key == quota_key,
            Quota.period_key == period_key,
        )
        .with_for_update()
    )
    row = result.scalar_one()
    return row


async def get_ledger_by_idempotency_key(session: AsyncSession, key: str) -> Ledger | None:
    result = await session.execute(select(Ledger).where(Ledger.idempotency_key == key))
    return result.scalar_one_or_none()


async def get_ledger(session: AsyncSession, tx_id: int) -> Ledger | None:
    return await session.get(Ledger, tx_id)


async def list_ledger(session: AsyncSession, user_id: int, limit: int = 50) -> list[Ledger]:
    result = await session.execute(
        select(Ledger)
        .where(Ledger.user_id == user_id)
        .order_by(Ledger.created_at.desc(), Ledger.tx_id.desc())
        .limit(limit)
    )
    return list(result.scalars().all())


def add_ledger(session: AsyncSession, ledger: Ledger) -> Ledger:
    session.add(ledger)
    return ledger


# -- 任务完成凭据探测（T2b，rp_* 只读） --------------------------------------
# 分层契约（pyproject import-linter layers）禁止 packages.billing 依赖
# packages.report，而 claim 完成校验又必须读 rp_feedback/rp_report 的服务端
# 可信信号 —— 直接查 shared.models 的只读 exists 探测是唯一合规通道。
# 纪律：billing 对 rp_* 永远只读，绝不写入；「完成判定」语义仍归 report 域。

#: rp_report.status 的「已完成」取值：与 packages/report/repo.py 的
#: STATUS_SUCCEEDED 对齐（同上，分层契约不允许 import 该常量，故复制字面量）。
_REPORT_STATUS_SUCCEEDED = "succeeded"


async def has_feedback_between(
    session: AsyncSession, *, user_id: int, start: datetime, end: datetime
) -> bool:
    """``[start, end)`` 内该用户是否存在反馈行（report_feedback 完成凭据）。"""
    result = await session.execute(
        select(RpFeedback.feedback_id)
        .where(
            RpFeedback.user_id == user_id,
            RpFeedback.created_at >= start,
            RpFeedback.created_at < end,
        )
        .limit(1)
    )
    return result.first() is not None


async def has_completed_report(session: AsyncSession, *, user_id: int) -> bool:
    """该用户是否存在已完成报告（first_report 完成凭据）。"""
    result = await session.execute(
        select(RpReport.report_id)
        .where(RpReport.user_id == user_id, RpReport.status == _REPORT_STATUS_SUCCEEDED)
        .limit(1)
    )
    return result.first() is not None