# -*- coding: utf-8 -*-
"""
لایه دیتابیس بات — MySQL
"""

import pymysql
import pymysql.cursors
from contextlib import contextmanager
from datetime import datetime

from config import MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DB


def now() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@contextmanager
def get_conn():
    conn = pymysql.connect(
        host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER,
        password=MYSQL_PASSWORD, database=MYSQL_DB,
        charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def _cols(conn, table):
    with conn.cursor() as c:
        c.execute(
            "SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS "
            "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s",
            (MYSQL_DB, table),
        )
        return [r["COLUMN_NAME"] for r in c.fetchall()]


def _add_col(conn, table, col, definition):
    if col not in _cols(conn, table):
        with conn.cursor() as c:
            c.execute(f"ALTER TABLE `{table}` ADD COLUMN `{col}` {definition}")


def _exe(conn, sql, params=()):
    with conn.cursor() as c:
        c.execute(sql, params)
        return c


def _one(conn, sql, params=()):
    with conn.cursor() as c:
        c.execute(sql, params)
        return c.fetchone()


def _all(conn, sql, params=()):
    with conn.cursor() as c:
        c.execute(sql, params)
        return c.fetchall()


def _ins(conn, sql, params=()):
    with conn.cursor() as c:
        c.execute(sql, params)
        return c.lastrowid


# ══════════════════════════════════════
#  init_db
# ══════════════════════════════════════

def init_db():
    with get_conn() as conn:

        _exe(conn, """CREATE TABLE IF NOT EXISTS `users` (
            user_id     BIGINT PRIMARY KEY,
            username    VARCHAR(64),
            full_name   VARCHAR(128),
            balance     BIGINT DEFAULT 0,
            is_blocked  TINYINT DEFAULT 0,
            rules_accepted TINYINT DEFAULT 0,
            kyc_required_pending TINYINT DEFAULT 0,
            kyc_verified TINYINT DEFAULT 0,
            referred_by BIGINT,
            referral_earnings BIGINT DEFAULT 0,
            last_support_at DATETIME,
            joined_at   DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        for col, defn in [
            ("rules_accepted",       "TINYINT DEFAULT 0"),
            ("kyc_required_pending", "TINYINT DEFAULT 0"),
            ("kyc_verified",         "TINYINT DEFAULT 0"),
            ("referred_by",          "BIGINT"),
            ("referral_earnings",    "BIGINT DEFAULT 0"),
            ("last_support_at",      "DATETIME"),
        ]:
            _add_col(conn, "users", col, defn)

        _exe(conn, """CREATE TABLE IF NOT EXISTS `products` (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            category    VARCHAR(32) DEFAULT 'stars',
            title       VARCHAR(255),
            stars_amount INT,
            price       BIGINT,
            price_usdt  DOUBLE,
            gram_amount DOUBLE,
            is_active   TINYINT DEFAULT 1
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        for col, defn in [
            ("category",    "VARCHAR(32) DEFAULT 'stars'"),
            ("title",       "VARCHAR(255)"),
            ("price_usdt",  "DOUBLE"),
            ("gram_amount", "DOUBLE"),
        ]:
            _add_col(conn, "products", col, defn)

        _exe(conn, """CREATE TABLE IF NOT EXISTS `orders` (
            id                  INT AUTO_INCREMENT PRIMARY KEY,
            user_id             BIGINT,
            category            VARCHAR(32) DEFAULT 'stars',
            title               VARCHAR(255),
            stars_amount        INT,
            price               BIGINT,
            recipient_username  VARCHAR(128),
            status              VARCHAR(32) DEFAULT 'pending',
            reseller_id         BIGINT,
            customer_chat_id    BIGINT,
            customer_display    VARCHAR(255),
            discount_code       VARCHAR(64),
            created_at          DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        for col, defn in [
            ("category",          "VARCHAR(32) DEFAULT 'stars'"),
            ("title",             "VARCHAR(255)"),
            ("reseller_id",       "BIGINT"),
            ("customer_chat_id",  "BIGINT"),
            ("customer_display",  "VARCHAR(255)"),
            ("discount_code",     "VARCHAR(64)"),
        ]:
            _add_col(conn, "orders", col, defn)

        _exe(conn, """CREATE TABLE IF NOT EXISTS `deposits` (
            id              INT AUTO_INCREMENT PRIMARY KEY,
            user_id         BIGINT,
            amount          BIGINT,
            receipt_file_id TEXT,
            file_unique_id  VARCHAR(128),
            status          VARCHAR(32) DEFAULT 'pending',
            created_at      DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")
        _add_col(conn, "deposits", "file_unique_id", "VARCHAR(128)")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `transfers` (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            from_user   BIGINT,
            to_user     BIGINT,
            amount      BIGINT,
            created_at  DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `kyc_requests` (
            id              INT AUTO_INCREMENT PRIMARY KEY,
            user_id         BIGINT,
            photo_file_id   TEXT,
            status          VARCHAR(32) DEFAULT 'pending',
            created_at      DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `settings` (
            `key`   VARCHAR(128) PRIMARY KEY,
            `value` MEDIUMTEXT
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `discount_codes` (
            code        VARCHAR(64) PRIMARY KEY,
            type        VARCHAR(16),
            value       INT,
            max_uses    INT,
            used_count  INT DEFAULT 0,
            expires_at  DATE,
            is_active   TINYINT DEFAULT 1,
            created_at  DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `staff` (
            user_id     BIGINT PRIMARY KEY,
            role        VARCHAR(32),
            created_at  DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `order_ratings` (
            order_id    INT PRIMARY KEY,
            user_id     BIGINT,
            rating      INT,
            created_at  DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `resellers` (
            user_id         BIGINT PRIMARY KEY,
            bot_token       VARCHAR(128),
            bot_username    VARCHAR(64),
            admin_id        BIGINT,
            is_active       TINYINT DEFAULT 1,
            profit_percent  DOUBLE DEFAULT 0,
            total_sales     BIGINT DEFAULT 0,
            created_at      DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        for col, defn in [
            ("profit_percent", "DOUBLE DEFAULT 0"),
            ("total_sales",    "BIGINT DEFAULT 0"),
        ]:
            _add_col(conn, "resellers", col, defn)

        _exe(conn, """CREATE TABLE IF NOT EXISTS `reseller_settings` (
            reseller_id BIGINT,
            `key`       VARCHAR(128),
            `value`     MEDIUMTEXT,
            PRIMARY KEY (reseller_id, `key`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `reseller_product_prices` (
            reseller_id BIGINT,
            product_id  INT,
            custom_price BIGINT,
            PRIMARY KEY (reseller_id, product_id)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `reseller_discount_codes` (
            code        VARCHAR(64),
            reseller_id BIGINT,
            type        VARCHAR(16),
            value       INT,
            max_uses    INT,
            used_count  INT DEFAULT 0,
            expires_at  DATE,
            is_active   TINYINT DEFAULT 1,
            created_at  DATETIME,
            PRIMARY KEY (code, reseller_id)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `reseller_balances` (
            reseller_id BIGINT,
            user_id     BIGINT,
            balance     BIGINT DEFAULT 0,
            PRIMARY KEY (reseller_id, user_id)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `reseller_user_deposits` (
            id              INT AUTO_INCREMENT PRIMARY KEY,
            reseller_id     BIGINT,
            user_id         BIGINT,
            amount          BIGINT,
            receipt_file_id TEXT,
            file_unique_id  VARCHAR(128),
            status          VARCHAR(32) DEFAULT 'pending',
            created_at      DATETIME
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        _exe(conn, """CREATE TABLE IF NOT EXISTS `activity_flow` (
            user_id     BIGINT PRIMARY KEY,
            flow        VARCHAR(64),
            started_at  DATETIME,
            reminded    TINYINT DEFAULT 0
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")

        # ─── Default settings ───
        defaults = {
            "card_number": "0000-0000-0000-0000",
            "card_holder": "نام صاحب کارت",
            "rules_text": "📜 قوانین:\n۱) سفارشات پس از تایید پرداخت انجام می‌شود.\n۲) موجودی قابل بازگشت نیست.\n۳) در صورت مشکل از پشتیبانی کمک بگیرید.",
            "support_text": "پیام خود را بفرستید تا به زودی پاسخ دهیم.",
            "deposit_warning_text": "⚖️ قبل از شارژ قوانین را مطالعه کنید.\n⚠️ همیشه از یک کارت ثابت واریز کنید.\nدر صورت قبول روی «✅ تایید قوانین» بزنید.",
            "welcome_text": "⭐ به ربات «استارز و پریمیوم - متا» خوش آمدید.\n\n🛍️ سفارشات استارز و پریمیوم خود را ثبت کنید.\n\n⬇️ یکی از گزینه‌های زیر را انتخاب کنید:",
            "kyc_enabled": "0",
            "sales_channel_id": "",
            "sales_report_template": "✅ یک سفارش جدید انجام شد!\n\n{title}\n💰 {price} تومان\n\n🚀 سفارشات سریع انجام می‌شوند.",
            "admin_group_id": "",
            "btn_order": "🛍 سفارش خدمات",
            "btn_account": "👤 حساب کاربری",
            "btn_deposit": "🟢 افزایش موجودی",
            "btn_transfer": "🔵 انتقال موجودی",
            "btn_rules": "⚖️ قوانین و راهنما",
            "btn_tracking": "🔎 پیگیری سفارشات",
            "btn_support": "☎️ پشتیبانی آنلاین",
            "btn_service_stars": "⭐ خرید استارز تلگرام",
            "btn_service_premium": "🌟 خرید پریمیوم تلگرام",
            "btn_service_gift": "🎁 خرید گیفت استارزی",
            "referral_type": "percent",
            "referral_value": "5",
            "usdt_rate": "0",
            "gram_profit_percent": "10",
            "gram_min_order": "0.1",
            "gram_description": "💎 با خرید گرام، ارز مستقیم به ولت شما واریز می‌شود.\n👛 انتقال مستقیم | 🔒 امنیت بالا | ⚡️ تراکنش سریع",
            "stars_price_per_star": "0",
            "support_cooldown_minutes": "10",
            "abandoned_reminder_minutes": "60",
            "backup_interval_hours": "24",
            "reseller_enabled": "0",
            "reseller_fee": "0",
            "cat_stars_enabled": "1",
            "cat_premium_enabled": "1",
            "cat_gift_enabled": "1",
            "cat_gram_enabled": "1",
        }
        for k, v in defaults.items():
            _exe(conn, "INSERT IGNORE INTO settings (`key`, `value`) VALUES (%s, %s)", (k, v))


# ══════════════════════════════════════
#  Settings
# ══════════════════════════════════════

def get_setting(key: str) -> str:
    with get_conn() as conn:
        row = _one(conn, "SELECT `value` FROM settings WHERE `key`=%s", (key,))
        return row["value"] if row else ""


def set_setting(key: str, value: str):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO settings (`key`,`value`) VALUES (%s,%s) "
             "ON DUPLICATE KEY UPDATE `value`=VALUES(`value`)",
             (key, value))


# ══════════════════════════════════════
#  Users
# ══════════════════════════════════════

def get_or_create_user(user_id: int, username: str, full_name: str):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO users (user_id,username,full_name,balance,is_blocked,joined_at) "
             "VALUES (%s,%s,%s,0,0,%s) "
             "ON DUPLICATE KEY UPDATE username=VALUES(username), full_name=VALUES(full_name)",
             (user_id, username, full_name, now()))
        return dict(_one(conn, "SELECT * FROM users WHERE user_id=%s", (user_id,)))


def get_user(user_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM users WHERE user_id=%s", (user_id,))
        return dict(row) if row else None


def find_user_by_username(username: str):
    username = username.lstrip("@")
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM users WHERE username=%s", (username,))
        return dict(row) if row else None


def update_balance(user_id: int, delta: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET balance=balance+%s WHERE user_id=%s", (delta, user_id))


def set_blocked(user_id: int, blocked: bool):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET is_blocked=%s WHERE user_id=%s", (1 if blocked else 0, user_id))


def set_rules_accepted(user_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET rules_accepted=1 WHERE user_id=%s", (user_id,))


def set_kyc_pending(user_id: int, pending: bool):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET kyc_required_pending=%s WHERE user_id=%s",
             (1 if pending else 0, user_id))


def set_kyc_verified(user_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET kyc_verified=1, kyc_required_pending=0 WHERE user_id=%s", (user_id,))


def all_user_ids():
    with get_conn() as conn:
        return [r["user_id"] for r in _all(conn, "SELECT user_id FROM users")]


def count_users():
    with get_conn() as conn:
        return _one(conn, "SELECT COUNT(*) AS c FROM users")["c"]


def sum_balances():
    with get_conn() as conn:
        row = _one(conn, "SELECT SUM(balance) AS s FROM users")
        return row["s"] or 0


def get_user_order_count(user_id: int) -> int:
    with get_conn() as conn:
        return _one(conn, "SELECT COUNT(*) AS c FROM orders WHERE user_id=%s", (user_id,))["c"]


def get_user_total_spent(user_id: int) -> int:
    with get_conn() as conn:
        row = _one(conn, "SELECT SUM(price) AS s FROM orders WHERE user_id=%s", (user_id,))
        return row["s"] or 0


def get_user_deposits(user_id: int, limit: int = 10):
    with get_conn() as conn:
        rows = _all(conn, "SELECT * FROM deposits WHERE user_id=%s ORDER BY id DESC LIMIT %s",
                    (user_id, limit))
        return [dict(r) for r in rows]


# ── KYC ──────────────────────────────

def create_kyc_request(user_id: int, photo_file_id: str) -> int:
    with get_conn() as conn:
        return _ins(conn,
                    "INSERT INTO kyc_requests (user_id,photo_file_id,status,created_at) VALUES (%s,%s,'pending',%s)",
                    (user_id, photo_file_id, now()))


def get_kyc_request(rid: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM kyc_requests WHERE id=%s", (rid,))
        return dict(row) if row else None


def update_kyc_request_status(rid: int, status: str):
    with get_conn() as conn:
        _exe(conn, "UPDATE kyc_requests SET status=%s WHERE id=%s", (status, rid))


def get_pending_kyc_requests():
    with get_conn() as conn:
        return [dict(r) for r in _all(conn, "SELECT * FROM kyc_requests WHERE status='pending' ORDER BY id")]


# ── Referral ─────────────────────────

def set_referrer(user_id: int, referrer_id: int) -> bool:
    if referrer_id == user_id:
        return False
    with get_conn() as conn:
        row = _one(conn, "SELECT referred_by FROM users WHERE user_id=%s", (user_id,))
        if not row or row["referred_by"] is not None:
            return False
        ref = _one(conn, "SELECT 1 AS e FROM users WHERE user_id=%s", (referrer_id,))
        if not ref:
            return False
        _exe(conn, "UPDATE users SET referred_by=%s WHERE user_id=%s", (referrer_id, user_id))
        return True


def add_referral_earning(user_id: int, amount: int):
    with get_conn() as conn:
        _exe(conn,
             "UPDATE users SET balance=balance+%s, referral_earnings=referral_earnings+%s WHERE user_id=%s",
             (amount, amount, user_id))


def get_referral_count(user_id: int) -> int:
    with get_conn() as conn:
        return _one(conn, "SELECT COUNT(*) AS c FROM users WHERE referred_by=%s", (user_id,))["c"]


def calc_referral_bonus(price: int) -> int:
    rtype = get_setting("referral_type") or "percent"
    try:
        rvalue = float(get_setting("referral_value") or "0")
    except ValueError:
        rvalue = 0
    return int(price * rvalue / 100) if rtype == "percent" else int(rvalue)


# ── Support cooldown ──────────────────

def can_send_support(user_id: int) -> bool:
    try:
        cooldown = int(get_setting("support_cooldown_minutes") or "0")
    except ValueError:
        cooldown = 0
    if cooldown <= 0:
        return True
    u = get_user(user_id)
    if not u or not u["last_support_at"]:
        return True
    last = u["last_support_at"]
    if isinstance(last, str):
        last = datetime.strptime(last, "%Y-%m-%d %H:%M:%S")
    return (datetime.now() - last).total_seconds() >= cooldown * 60


def update_last_support(user_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE users SET last_support_at=%s WHERE user_id=%s", (now(), user_id))


# ══════════════════════════════════════
#  Products
# ══════════════════════════════════════

CATEGORY_LABELS = {
    "stars":   "⭐ استارز تلگرام",
    "premium": "🌟 پریمیوم تلگرام",
    "gift":    "🎁 گیفت استارزی",
    "gram":    "💎 گرام",
}

STAFF_ROLES = ("full", "orders", "products")

RENAMEABLE_BUTTONS = [
    ("btn_order",           "دکمه «سفارش خدمات»"),
    ("btn_account",         "دکمه «حساب کاربری»"),
    ("btn_deposit",         "دکمه «افزایش موجودی»"),
    ("btn_transfer",        "دکمه «انتقال موجودی»"),
    ("btn_rules",           "دکمه «قوانین و راهنما»"),
    ("btn_tracking",        "دکمه «پیگیری سفارشات»"),
    ("btn_support",         "دکمه «پشتیبانی آنلاین»"),
    ("btn_service_stars",   "دکمه «خرید استارز»"),
    ("btn_service_premium", "دکمه «خرید پریمیوم»"),
    ("btn_service_gift",    "دکمه «خرید گیفت»"),
]


def add_product(title: str, price: int, category: str = "stars",
                stars_amount: int = None, gram_amount: float = None):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO products (category,title,stars_amount,gram_amount,price,is_active) "
             "VALUES (%s,%s,%s,%s,%s,1)",
             (category, title, stars_amount, gram_amount, price))


def get_products(active_only: bool = True, category: str = None):
    with get_conn() as conn:
        q = "SELECT * FROM products WHERE 1=1"
        p = []
        if active_only:
            q += " AND is_active=1"
        if category:
            q += " AND category=%s"; p.append(category)
        q += " ORDER BY id ASC"
        return [dict(r) for r in _all(conn, q, p)]


def get_product(product_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM products WHERE id=%s", (product_id,))
        return dict(row) if row else None


def toggle_product(product_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE products SET is_active=1-is_active WHERE id=%s", (product_id,))


def delete_product(product_id: int):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM products WHERE id=%s", (product_id,))


def get_effective_price(product: dict) -> int:
    if product.get("price_usdt"):
        try:
            rate = float(get_setting("usdt_rate") or "0")
        except ValueError:
            rate = 0
        if rate > 0:
            return int(round(product["price_usdt"] * rate))
    return product["price"] or 0


def is_category_enabled(category: str) -> bool:
    return get_setting(f"cat_{category}_enabled") != "0"


def toggle_category(category: str):
    key = f"cat_{category}_enabled"
    set_setting(key, "0" if get_setting(key) != "0" else "1")


def get_gram_product_price(product: dict, gram_usd: float) -> int:
    if product.get("price") and product["price"] > 0:
        return product["price"]
    amount = product.get("gram_amount") or 0
    if amount > 0 and gram_usd > 0:
        profit = float(get_setting("gram_profit_percent") or "0")
        rate   = float(get_setting("usdt_rate") or "0")
        base   = gram_usd * rate * amount
        return int(base * (1 + profit / 100))
    return 0


# ══════════════════════════════════════
#  Orders
# ══════════════════════════════════════

def create_order(user_id: int, title: str, price: int, recipient_username: str,
                 category: str = "stars", stars_amount: int = None, reseller_id: int = None) -> int:
    with get_conn() as conn:
        return _ins(conn,
                    "INSERT INTO orders (user_id,category,title,stars_amount,price,"
                    "recipient_username,status,created_at,reseller_id) "
                    "VALUES (%s,%s,%s,%s,%s,%s,'pending',%s,%s)",
                    (user_id, category, title, stars_amount, price,
                     recipient_username, now(), reseller_id))


def get_order(order_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM orders WHERE id=%s", (order_id,))
        return dict(row) if row else None


def update_order_status(order_id: int, status: str):
    with get_conn() as conn:
        _exe(conn, "UPDATE orders SET status=%s WHERE id=%s", (status, order_id))


def get_user_orders(user_id: int, limit: int = 10):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn, "SELECT * FROM orders WHERE user_id=%s ORDER BY id DESC LIMIT %s",
                     (user_id, limit))]


def get_pending_orders():
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn, "SELECT * FROM orders WHERE status='pending' ORDER BY id ASC")]


def count_orders():
    with get_conn() as conn:
        return _one(conn, "SELECT COUNT(*) AS c FROM orders")["c"]


def sum_completed_revenue():
    with get_conn() as conn:
        row = _one(conn, "SELECT SUM(price) AS s FROM orders WHERE status='done'")
        return row["s"] or 0


def get_reseller_orders_by_user(reseller_id: int, user_id: int, limit: int = 10):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn, "SELECT * FROM orders WHERE reseller_id=%s AND user_id=%s ORDER BY id DESC LIMIT %s",
                     (reseller_id, user_id, limit))]


# ══════════════════════════════════════
#  Deposits
# ══════════════════════════════════════

def create_deposit(user_id: int, amount: int, receipt_file_id: str,
                   file_unique_id: str = None) -> int:
    with get_conn() as conn:
        return _ins(conn,
                    "INSERT INTO deposits (user_id,amount,receipt_file_id,file_unique_id,status,created_at) "
                    "VALUES (%s,%s,%s,%s,'pending',%s)",
                    (user_id, amount, receipt_file_id, file_unique_id, now()))


def get_deposit(deposit_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM deposits WHERE id=%s", (deposit_id,))
        return dict(row) if row else None


def update_deposit_status(deposit_id: int, status: str):
    with get_conn() as conn:
        _exe(conn, "UPDATE deposits SET status=%s WHERE id=%s", (status, deposit_id))


def get_pending_deposits():
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn, "SELECT * FROM deposits WHERE status='pending' ORDER BY id ASC")]


def find_approved_deposit_by_file(file_unique_id: str):
    if not file_unique_id:
        return None
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM deposits WHERE file_unique_id=%s AND status='done' LIMIT 1",
                   (file_unique_id,))
        return dict(row) if row else None


# ══════════════════════════════════════
#  Transfers
# ══════════════════════════════════════

def create_transfer(from_user: int, to_user: int, amount: int):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO transfers (from_user,to_user,amount,created_at) VALUES (%s,%s,%s,%s)",
             (from_user, to_user, amount, now()))


# ══════════════════════════════════════
#  Discount codes
# ══════════════════════════════════════

def create_discount_code(code: str, dtype: str, value: int,
                          max_uses: int = None, expires_at: str = None):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO discount_codes (code,type,value,max_uses,used_count,expires_at,is_active,created_at) "
             "VALUES (%s,%s,%s,%s,0,%s,1,%s) "
             "ON DUPLICATE KEY UPDATE type=VALUES(type),value=VALUES(value),max_uses=VALUES(max_uses),is_active=1",
             (code.upper(), dtype, value, max_uses, expires_at, now()))


def get_discount_code(code: str):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM discount_codes WHERE code=%s", (code.upper(),))
        return dict(row) if row else None


def list_discount_codes():
    with get_conn() as conn:
        return [dict(r) for r in _all(conn, "SELECT * FROM discount_codes ORDER BY created_at DESC")]


def increment_discount_usage(code: str):
    with get_conn() as conn:
        _exe(conn, "UPDATE discount_codes SET used_count=used_count+1 WHERE code=%s", (code.upper(),))


def toggle_discount_code(code: str):
    with get_conn() as conn:
        _exe(conn, "UPDATE discount_codes SET is_active=1-is_active WHERE code=%s", (code.upper(),))


def delete_discount_code(code: str):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM discount_codes WHERE code=%s", (code.upper(),))


def validate_discount_code(code: str):
    row = get_discount_code(code)
    if not row or not row["is_active"]:
        return None
    if row["max_uses"] is not None and row["used_count"] >= row["max_uses"]:
        return None
    if row["expires_at"]:
        try:
            exp = row["expires_at"]
            if isinstance(exp, str):
                exp = datetime.strptime(exp, "%Y-%m-%d")
            if datetime.now() > exp:
                return None
        except Exception:
            pass
    return row


def apply_discount(price: int, code_row: dict) -> int:
    if code_row["type"] == "percent":
        return max(price - int(price * code_row["value"] / 100), 0)
    return max(price - code_row["value"], 0)


# ══════════════════════════════════════
#  Staff
# ══════════════════════════════════════

def add_staff(user_id: int, role: str):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO staff (user_id,role,created_at) VALUES (%s,%s,%s) "
             "ON DUPLICATE KEY UPDATE role=VALUES(role)",
             (user_id, role, now()))


def remove_staff(user_id: int):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM staff WHERE user_id=%s", (user_id,))


def get_staff(user_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM staff WHERE user_id=%s", (user_id,))
        return dict(row) if row else None


def list_staff():
    with get_conn() as conn:
        return [dict(r) for r in _all(conn, "SELECT * FROM staff ORDER BY created_at ASC")]


# ══════════════════════════════════════
#  Ratings
# ══════════════════════════════════════

def create_rating(order_id: int, user_id: int, rating: int):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO order_ratings (order_id,user_id,rating,created_at) VALUES (%s,%s,%s,%s) "
             "ON DUPLICATE KEY UPDATE rating=VALUES(rating)",
             (order_id, user_id, rating, now()))


def get_rating(order_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM order_ratings WHERE order_id=%s", (order_id,))
        return dict(row) if row else None


def get_average_rating():
    with get_conn() as conn:
        row = _one(conn, "SELECT AVG(rating) AS a, COUNT(*) AS c FROM order_ratings")
        return (float(row["a"]) if row["a"] else 0.0), (row["c"] or 0)


# ══════════════════════════════════════
#  Resellers
# ══════════════════════════════════════

def create_reseller(user_id: int, bot_token: str, bot_username: str, admin_id: int):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO resellers (user_id,bot_token,bot_username,admin_id,is_active,created_at) "
             "VALUES (%s,%s,%s,%s,1,%s) "
             "ON DUPLICATE KEY UPDATE bot_token=VALUES(bot_token),bot_username=VALUES(bot_username),admin_id=VALUES(admin_id)",
             (user_id, bot_token, bot_username, admin_id, now()))


def get_reseller(user_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM resellers WHERE user_id=%s", (user_id,))
        return dict(row) if row else None


def list_resellers(active_only: bool = False):
    with get_conn() as conn:
        q = "SELECT * FROM resellers"
        if active_only:
            q += " WHERE is_active=1"
        q += " ORDER BY created_at DESC"
        return [dict(r) for r in _all(conn, q)]


def set_reseller_active(user_id: int, active: bool):
    with get_conn() as conn:
        _exe(conn, "UPDATE resellers SET is_active=%s WHERE user_id=%s",
             (1 if active else 0, user_id))


def delete_reseller(user_id: int):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM resellers WHERE user_id=%s", (user_id,))


def set_reseller_profit(user_id: int, percent: float):
    with get_conn() as conn:
        _exe(conn, "UPDATE resellers SET profit_percent=%s WHERE user_id=%s", (percent, user_id))


def add_reseller_sale(user_id: int, amount: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE resellers SET total_sales=total_sales+%s WHERE user_id=%s",
             (amount, user_id))


def update_reseller_admin(user_id: int, new_admin_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE resellers SET admin_id=%s WHERE user_id=%s", (new_admin_id, user_id))


def get_all_resellers_with_stats():
    with get_conn() as conn:
        return [dict(r) for r in _all(conn, """
            SELECT r.*, COUNT(o.id) AS order_count,
                   COALESCE(SUM(CASE WHEN o.status='done' THEN o.price END),0) AS revenue,
                   u.balance AS credit
            FROM resellers r
            LEFT JOIN orders o ON o.reseller_id=r.user_id
            LEFT JOIN users  u ON u.user_id=r.user_id
            GROUP BY r.user_id ORDER BY r.created_at DESC
        """)]


def get_leaderboard(limit: int = 10):
    with get_conn() as conn:
        return [dict(r) for r in _all(conn, """
            SELECT r.user_id, r.bot_username, r.total_sales, r.profit_percent,
                   COUNT(o.id) AS order_count
            FROM resellers r
            LEFT JOIN orders o ON o.reseller_id=r.user_id AND o.status='done'
            WHERE r.is_active=1
            GROUP BY r.user_id ORDER BY r.total_sales DESC LIMIT %s
        """, (limit,))]


# ── Reseller Settings ─────────────────

def get_reseller_setting(reseller_id: int, key: str, default: str = "") -> str:
    with get_conn() as conn:
        row = _one(conn, "SELECT `value` FROM reseller_settings WHERE reseller_id=%s AND `key`=%s",
                   (reseller_id, key))
        return row["value"] if row else default


def set_reseller_setting(reseller_id: int, key: str, value: str):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO reseller_settings (reseller_id,`key`,`value`) VALUES (%s,%s,%s) "
             "ON DUPLICATE KEY UPDATE `value`=VALUES(`value`)",
             (reseller_id, key, value))


def get_reseller_force_join(reseller_id: int):
    return get_reseller_setting(reseller_id, "force_join_channel")


def set_reseller_force_join(reseller_id: int, channel: str):
    set_reseller_setting(reseller_id, "force_join_channel", channel)


# ── Reseller Product Prices ──────────

def get_reseller_price(reseller_id: int, product: dict) -> int:
    with get_conn() as conn:
        row = _one(conn,
                   "SELECT custom_price FROM reseller_product_prices WHERE reseller_id=%s AND product_id=%s",
                   (reseller_id, product["id"]))
        if row:
            return row["custom_price"]
    reseller = get_reseller(reseller_id)
    base = get_effective_price(product)
    if reseller and reseller["profit_percent"] > 0:
        return int(base * (1 + reseller["profit_percent"] / 100))
    return base


def set_reseller_product_price(reseller_id: int, product_id: int, price: int):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO reseller_product_prices (reseller_id,product_id,custom_price) VALUES (%s,%s,%s) "
             "ON DUPLICATE KEY UPDATE custom_price=VALUES(custom_price)",
             (reseller_id, product_id, price))


def clear_reseller_product_price(reseller_id: int, product_id: int):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM reseller_product_prices WHERE reseller_id=%s AND product_id=%s",
             (reseller_id, product_id))


# ── Reseller Discount Codes ──────────

def create_reseller_discount(reseller_id: int, code: str, dtype: str, value: int,
                              max_uses: int = None, expires_at: str = None):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO reseller_discount_codes "
             "(code,reseller_id,type,value,max_uses,used_count,expires_at,is_active,created_at) "
             "VALUES (%s,%s,%s,%s,%s,0,%s,1,%s) "
             "ON DUPLICATE KEY UPDATE type=VALUES(type),value=VALUES(value)",
             (code.upper(), reseller_id, dtype, value, max_uses, expires_at, now()))


def validate_reseller_discount(reseller_id: int, code: str):
    with get_conn() as conn:
        row = _one(conn,
                   "SELECT * FROM reseller_discount_codes WHERE code=%s AND reseller_id=%s AND is_active=1",
                   (code.upper(), reseller_id))
    if not row:
        return None
    row = dict(row)
    if row["max_uses"] and row["used_count"] >= row["max_uses"]:
        return None
    if row["expires_at"]:
        try:
            exp = row["expires_at"]
            if isinstance(exp, str):
                exp = datetime.strptime(exp, "%Y-%m-%d")
            if datetime.now() > exp:
                return None
        except Exception:
            pass
    return row


def use_reseller_discount(reseller_id: int, code: str):
    with get_conn() as conn:
        _exe(conn,
             "UPDATE reseller_discount_codes SET used_count=used_count+1 WHERE code=%s AND reseller_id=%s",
             (code.upper(), reseller_id))


def list_reseller_discounts(reseller_id: int):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn, "SELECT * FROM reseller_discount_codes WHERE reseller_id=%s ORDER BY created_at DESC",
                     (reseller_id,))]


def toggle_reseller_discount(reseller_id: int, code: str):
    with get_conn() as conn:
        _exe(conn,
             "UPDATE reseller_discount_codes SET is_active=1-is_active WHERE code=%s AND reseller_id=%s",
             (code.upper(), reseller_id))


# ── Reseller Customer Balances ────────

def get_reseller_user_balance(reseller_id: int, user_id: int) -> int:
    with get_conn() as conn:
        row = _one(conn,
                   "SELECT balance FROM reseller_balances WHERE reseller_id=%s AND user_id=%s",
                   (reseller_id, user_id))
        return row["balance"] if row else 0


def update_reseller_user_balance(reseller_id: int, user_id: int, delta: int):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO reseller_balances (reseller_id,user_id,balance) VALUES (%s,%s,GREATEST(0,%s)) "
             "ON DUPLICATE KEY UPDATE balance=GREATEST(0, balance+%s)",
             (reseller_id, user_id, max(0, delta), delta))


# ── Reseller Customer Deposits ────────

def create_reseller_user_deposit(reseller_id: int, user_id: int,
                                  amount: int, file_id: str, file_uid: str = None) -> int:
    with get_conn() as conn:
        return _ins(conn,
                    "INSERT INTO reseller_user_deposits "
                    "(reseller_id,user_id,amount,receipt_file_id,file_unique_id,status,created_at) "
                    "VALUES (%s,%s,%s,%s,%s,'pending',%s)",
                    (reseller_id, user_id, amount, file_id, file_uid, now()))


def get_reseller_user_deposit(dep_id: int):
    with get_conn() as conn:
        row = _one(conn, "SELECT * FROM reseller_user_deposits WHERE id=%s", (dep_id,))
        return dict(row) if row else None


def get_pending_reseller_user_deposits(reseller_id: int):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn,
                     "SELECT * FROM reseller_user_deposits WHERE reseller_id=%s AND status='pending' ORDER BY id",
                     (reseller_id,))]


def approve_reseller_user_deposit(dep_id: int, amount_override: int = None):
    d = get_reseller_user_deposit(dep_id)
    if not d or d["status"] != "pending":
        return False
    amount = amount_override if amount_override else d["amount"]
    with get_conn() as conn:
        _exe(conn, "UPDATE reseller_user_deposits SET status='done' WHERE id=%s", (dep_id,))
    update_reseller_user_balance(d["reseller_id"], d["user_id"], amount)
    return amount


def reject_reseller_user_deposit(dep_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE reseller_user_deposits SET status='rejected' WHERE id=%s", (dep_id,))


def get_reseller_user_deposits(reseller_id: int, user_id: int, limit: int = 8):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn,
                     "SELECT * FROM reseller_user_deposits WHERE reseller_id=%s AND user_id=%s "
                     "ORDER BY id DESC LIMIT %s",
                     (reseller_id, user_id, limit))]


# ══════════════════════════════════════
#  Activity flow
# ══════════════════════════════════════

def set_activity_flow(user_id: int, flow: str):
    with get_conn() as conn:
        _exe(conn,
             "INSERT INTO activity_flow (user_id,flow,started_at,reminded) VALUES (%s,%s,%s,0) "
             "ON DUPLICATE KEY UPDATE flow=VALUES(flow), started_at=VALUES(started_at), reminded=0",
             (user_id, flow, now()))


def clear_activity_flow(user_id: int):
    with get_conn() as conn:
        _exe(conn, "DELETE FROM activity_flow WHERE user_id=%s", (user_id,))


def get_stale_activity_flows(minutes: int):
    with get_conn() as conn:
        return [dict(r) for r in
                _all(conn,
                     "SELECT * FROM activity_flow WHERE reminded=0 "
                     "AND started_at <= DATE_SUB(NOW(), INTERVAL %s MINUTE)",
                     (minutes,))]


def mark_flow_reminded(user_id: int):
    with get_conn() as conn:
        _exe(conn, "UPDATE activity_flow SET reminded=1 WHERE user_id=%s", (user_id,))


# ══════════════════════════════════════
#  Gram price
# ══════════════════════════════════════

async def fetch_gram_price_usd() -> float:
    import aiohttp
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                "https://api.coingecko.com/api/v3/simple/price?ids=the-open-network&vs_currencies=usd",
                timeout=aiohttp.ClientTimeout(total=5),
            ) as resp:
                data = await resp.json()
                return float(data["the-open-network"]["usd"])
    except Exception:
        return 0.0


def calc_gram_price_toman(gram_usd: float, amount: float) -> int:
    try:
        rate   = float(get_setting("usdt_rate") or "0")
        profit = float(get_setting("gram_profit_percent") or "0")
    except ValueError:
        rate, profit = 0, 0
    if rate <= 0 or gram_usd <= 0:
        return 0
    return int(gram_usd * rate * amount * (1 + profit / 100))


# ══════════════════════════════════════
#  Stats
# ══════════════════════════════════════

def orders_count_since(since_str: str) -> int:
    with get_conn() as conn:
        return _one(conn, "SELECT COUNT(*) AS c FROM orders WHERE created_at>=%s", (since_str,))["c"]


def revenue_since(since_str: str) -> int:
    with get_conn() as conn:
        row = _one(conn, "SELECT SUM(price) AS s FROM orders WHERE status='done' AND created_at>=%s",
                   (since_str,))
        return row["s"] or 0


def get_full_stats():
    from datetime import timedelta
    n     = datetime.now()
    today = n.strftime("%Y-%m-%d 00:00:00")
    week  = (n - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")
    month = (n - timedelta(days=30)).strftime("%Y-%m-%d %H:%M:%S")
    avg_rating, rating_count = get_average_rating()
    with get_conn() as conn:
        top_users = _all(conn,
                         "SELECT user_id,username,balance FROM users ORDER BY balance DESC LIMIT 5")
        blocked   = _one(conn, "SELECT COUNT(*) AS c FROM users WHERE is_blocked=1")["c"]
        active_rs = _one(conn, "SELECT COUNT(*) AS c FROM resellers WHERE is_active=1")["c"]
    return {
        "total_users": count_users(), "blocked_users": blocked, "total_balance": sum_balances(),
        "orders_today": orders_count_since(today),
        "orders_week":  orders_count_since(week),
        "orders_month": orders_count_since(month),
        "orders_total": count_orders(),
        "revenue_today": revenue_since(today),
        "revenue_week":  revenue_since(week),
        "revenue_month": revenue_since(month),
        "revenue_total": sum_completed_revenue(),
        "pending_orders":   len(get_pending_orders()),
        "pending_deposits": len(get_pending_deposits()),
        "pending_kyc":      len(get_pending_kyc_requests()),
        "avg_rating": avg_rating, "rating_count": rating_count,
        "active_resellers": active_rs,
        "top_users": [dict(r) for r in top_users],
    }