"""Read-only queries backing the console. Every function issues SELECTs
only — the console never mutates the database.

Payment status is computed with the app's own ``paid_status`` so the console
never diverges from what the app considers "paid".
"""
from __future__ import annotations

from datetime import timedelta

from sqlalchemy import func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import utcnow
from app.models import (
    EmailSend,
    PortfolioSync,
    Referral,
    StrategicLogFeedback,
    User,
    UserAcknowledgement,
)
from app.services.access import paid_status


async def overview_stats(session: AsyncSession) -> dict:
    """Headline counts for the landing page: totals, tier split, growth,
    paid-active, engagement."""
    now = utcnow()
    d7 = now - timedelta(days=7)
    d30 = now - timedelta(days=30)

    total = await session.scalar(select(func.count(User.id))) or 0

    tier_rows = (await session.execute(
        select(User.tier, func.count(User.id)).group_by(User.tier)
    )).all()
    tiers = {tier: n for tier, n in tier_rows}

    signups_7d = await session.scalar(
        select(func.count(User.id)).where(User.created_at >= d7)
    ) or 0
    signups_30d = await session.scalar(
        select(func.count(User.id)).where(User.created_at >= d30)
    ) or 0
    active_30d = await session.scalar(
        select(func.count(User.id)).where(User.last_login_at >= d30)
    ) or 0

    # Paid-active = subscribed tier OR a live credit window. The two are
    # counted disjointly (a credit user whose tier is already paid is only
    # counted once, under tier) so the sum can't double-count.
    paid_by_tier = await session.scalar(
        select(func.count(User.id)).where(User.tier.in_(("paid", "enterprise")))
    ) or 0
    credit_active = await session.scalar(
        select(func.count(User.id)).where(
            User.credit_until.is_not(None),
            User.credit_until > now,
            User.tier.not_in(("paid", "enterprise")),
        )
    ) or 0

    sync_enabled = await session.scalar(
        select(func.count(PortfolioSync.user_id))
    ) or 0

    referrals_total = await session.scalar(select(func.count(Referral.id))) or 0
    referrals_converted = await session.scalar(
        select(func.count(Referral.id)).where(Referral.converted_at.is_not(None))
    ) or 0

    return {
        "total_users": total,
        "tiers": tiers,
        "free": tiers.get("free", 0),
        "paid": tiers.get("paid", 0),
        "enterprise": tiers.get("enterprise", 0),
        "paid_active": paid_by_tier + credit_active,
        "credit_active": credit_active,
        "signups_7d": signups_7d,
        "signups_30d": signups_30d,
        "active_30d": active_30d,
        "sync_enabled": sync_enabled,
        "referrals_total": referrals_total,
        "referrals_converted": referrals_converted,
    }


async def list_users(
    session: AsyncSession,
    q: str | None = None,
    limit: int = 200,
    offset: int = 0,
) -> tuple[list[dict], int]:
    """User rows for the table view, newest first. Optional ``q`` filters by
    email substring. Returns (rows, total_matching)."""
    stmt = select(User)
    count_stmt = select(func.count(User.id))
    if q:
        like = f"%{q.strip()}%"
        stmt = stmt.where(User.email.like(like))
        count_stmt = count_stmt.where(User.email.like(like))

    total = await session.scalar(count_stmt) or 0
    users = (await session.execute(
        stmt.order_by(User.created_at.desc()).limit(limit).offset(offset)
    )).scalars().all()

    if not users:
        return [], total

    ids = [u.id for u in users]

    # One grouped query each for the per-user badges, merged in Python —
    # avoids an N+1 without a wide join fan-out.
    sync_ids = set((await session.execute(
        select(PortfolioSync.user_id).where(PortfolioSync.user_id.in_(ids))
    )).scalars().all())

    ref_counts = dict((await session.execute(
        select(Referral.referrer_user_id, func.count(Referral.id))
        .where(Referral.referrer_user_id.in_(ids))
        .group_by(Referral.referrer_user_id)
    )).all())

    rows: list[dict] = []
    for u in users:
        ps = paid_status(u)
        rows.append({
            "id": u.id,
            "email": u.email,
            "tier": u.tier,
            "paid_active": ps.active,
            "paid_source": ps.source,
            "credit_days": ps.days_remaining,
            "created_at": u.created_at,
            "last_login_at": u.last_login_at,
            "lang": u.lang,
            "has_sync": u.id in sync_ids,
            "referrals": ref_counts.get(u.id, 0),
            "on_stripe": bool(u.stripe_customer_id),
            "on_polar": bool(u.polar_customer_id),
            "trialing": u.stripe_trial_end_at is not None,
        })
    return rows, total


async def user_detail(session: AsyncSession, user_id: int) -> dict | None:
    """Everything the console shows for one user: the row itself, computed
    payment status, billing linkage, and joined activity history."""
    user = await session.get(User, user_id)
    if user is None:
        return None

    ps = paid_status(user)

    referred_by = None
    if user.referred_by_user_id:
        ref = await session.get(User, user.referred_by_user_id)
        referred_by = {"id": ref.id, "email": ref.email} if ref else None

    # Referrals this user sent (with conversion state).
    sent = (await session.execute(
        select(Referral, User.email)
        .join(User, User.id == Referral.referred_user_id)
        .where(Referral.referrer_user_id == user_id)
        .order_by(Referral.created_at.desc())
    )).all()
    referrals_sent = [{
        "referred_email": email,
        "created_at": r.created_at,
        "converted_at": r.converted_at,
    } for r, email in sent]

    # Digest email delivery log (most recent first, capped).
    email_rows = (await session.execute(
        select(EmailSend).where(EmailSend.user_id == user_id)
        .order_by(EmailSend.sent_at.desc()).limit(25)
    )).scalars().all()
    emails = [{
        "kind": e.kind, "sent_at": e.sent_at,
        "status": e.status, "error": e.error,
    } for e in email_rows]

    # Legal acknowledgements (evidentiary audit trail).
    ack_rows = (await session.execute(
        select(UserAcknowledgement).where(UserAcknowledgement.user_id == user_id)
        .order_by(UserAcknowledgement.accepted_at.desc())
    )).scalars().all()
    acks = [{
        "version": a.version, "lang": a.lang, "accepted_at": a.accepted_at,
    } for a in ack_rows]

    # Feedback votes cast on strategic logs.
    fb_rows = (await session.execute(
        select(StrategicLogFeedback.vote, func.count(StrategicLogFeedback.id))
        .where(StrategicLogFeedback.user_id == user_id)
        .group_by(StrategicLogFeedback.vote)
    )).all()
    feedback = {vote: n for vote, n in fb_rows}

    sync = await session.get(PortfolioSync, user_id)

    return {
        "user": user,
        "paid": ps,
        "referred_by": referred_by,
        "referrals_sent": referrals_sent,
        "emails": emails,
        "acks": acks,
        "feedback": feedback,
        "sync": {
            "enabled": sync is not None,
            "updated_at": sync.updated_at if sync else None,
            "version": sync.version if sync else None,
        },
    }


async def db_stats(session: AsyncSession) -> dict:
    """Per-table size and row estimates for the current schema.

    On MariaDB this reads ``information_schema.tables`` (row counts are the
    engine's estimate for InnoDB, sizes are exact). On sqlite (the test
    backend) there is no such view, so it falls back to real ``COUNT(*)``
    per table with sizes reported as 0.
    """
    dialect = session.bind.dialect.name

    if dialect == "sqlite":
        from app.db import Base
        tables = []
        for tbl in Base.metadata.sorted_tables:
            n = await session.scalar(select(func.count()).select_from(tbl)) or 0
            tables.append({
                "name": tbl.name, "rows": n,
                "data_bytes": 0, "index_bytes": 0, "total_bytes": 0, "pct": 0.0,
            })
        tables.sort(key=lambda t: t["rows"], reverse=True)
        return {"tables": tables, "total_bytes": 0, "total_rows": sum(t["rows"] for t in tables)}

    rows = (await session.execute(text(
        "SELECT table_name, table_rows, data_length, index_length "
        "FROM information_schema.tables "
        "WHERE table_schema = DATABASE() "
        "ORDER BY (data_length + index_length) DESC"
    ))).all()

    tables = []
    total_bytes = 0
    total_rows = 0
    for name, tr, data_len, idx_len in rows:
        data_len = int(data_len or 0)
        idx_len = int(idx_len or 0)
        tot = data_len + idx_len
        total_bytes += tot
        total_rows += int(tr or 0)
        tables.append({
            "name": name, "rows": int(tr or 0),
            "data_bytes": data_len, "index_bytes": idx_len, "total_bytes": tot,
        })
    for t in tables:
        t["pct"] = round(100.0 * t["total_bytes"] / total_bytes, 1) if total_bytes else 0.0

    return {"tables": tables, "total_bytes": total_bytes, "total_rows": total_rows}