mfinder/db/settings_sql.py

import threading
from sqlalchemy import create_engine, Column, TEXT, Boolean, Numeric, BigInteger, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool
from sqlalchemy.orm.exc import NoResultFound
from mfinder import DB_URL, LOGGER

BASE = declarative_base()


def merge_env_shortener_into_list(shorteners_json, env_site, env_api, env_download):
    import json
    try:
        shorteners = json.loads(shorteners_json) if shorteners_json else []
    except Exception:
        shorteners = []
    
    parsed_site = None
    if env_site:
        ss = env_site.strip()
        parsed_site = None if ss.lower() in ("off", "disable", "disabled", "none") else ss
        
    parsed_api = env_api.strip() if env_api else None
    
    parsed_hd = None
    if env_download:
        hd = env_download.strip()
        parsed_hd = None if hd.lower() in ("off", "disable", "disabled", "none") else hd

    if not parsed_site or not parsed_api:
        return shorteners_json, False
        
    exists = False
    updated = False
    for s in shorteners:
        if s.get("site") == parsed_site:
            if s.get("api") != parsed_api:
                s["api"] = parsed_api
                updated = True
            if parsed_hd is not None and s.get("how_to_download") != parsed_hd:
                s["how_to_download"] = parsed_hd
                updated = True
            exists = True
            break
            
    if not exists:
        shorteners.append({
            "site": parsed_site,
            "api": parsed_api,
            "how_to_download": parsed_hd,
            "status": True
        })
        updated = True
        
    return json.dumps(shorteners), updated

class AdminSettings(BASE):
    __tablename__ = "admin_settings"
    setting_name = Column(TEXT, primary_key=True)
    auto_delete = Column(Numeric)
    custom_caption = Column(TEXT)
    fsub_channel = Column(Numeric)
    channel_link = Column(TEXT)
    caption_uname = Column(TEXT)
    repair_mode = Column(Boolean)
    shortener_status = Column(Boolean)
    admins_list = Column(TEXT)
    db_channels_list = Column(TEXT)
    token_shortener_enabled = Column(Boolean)
    token_timeout = Column(Numeric)
    log_channel = Column(Numeric)
    log_channel_access_hash = Column(Numeric)
    fsub_channel_access_hash = Column(Numeric)
    fsub_channels_list = Column(TEXT)
    fsub_channels_access_hashes = Column(TEXT)
    shorteners_list = Column(TEXT)
    smart_rotator = Column(Boolean, default=False)
    newsletter_days = Column(TEXT)
    newsletter_time = Column(TEXT)
    newsletter_count = Column(Integer)
    newsletter_enabled = Column(Boolean, default=False)
    newsletter_target = Column(TEXT)
    announcement_enabled = Column(Boolean, default=False)
    announcement_channel = Column(Numeric)
    announcement_channel_access_hash = Column(Numeric)

    def __init__(self, setting_name="default"):
        self.setting_name = setting_name
        self.auto_delete = 0
        self.custom_caption = None
        self.fsub_channel = None
        self.channel_link = None
        self.caption_uname = None
        self.repair_mode = False
        self.shortener_site = None
        self.shortener_api = None
        self.shortener_status = True
        self.admins_list = None
        self.db_channels_list = None
        self.token_shortener_enabled = False
        self.token_timeout = 3600
        self.log_channel = None
        self.how_to_download = None
        self.log_channel_access_hash = None
        self.fsub_channel_access_hash = None
        self.fsub_channels_list = None
        self.fsub_channels_access_hashes = "{}"
        self.shorteners_list = "[]"
        self.smart_rotator = False
        self.newsletter_days = "Friday"
        self.newsletter_time = "20:00"
        self.newsletter_count = 10
        self.newsletter_enabled = False
        self.newsletter_target = "subscribed"
        self.newsletter_days = "Friday"
        self.newsletter_time = "20:00"
        self.newsletter_count = 10
        self.newsletter_enabled = False

class PendingFsubRequests(BASE):
    __tablename__ = "pending_fsub_requests"
    user_id = Column(Numeric, primary_key=True)
    search_query = Column(TEXT)
    target_file_id = Column(TEXT)
    created_at = Column(Numeric)

    def __init__(self, user_id, search_query=None, target_file_id=None):
        self.user_id = user_id
        self.search_query = search_query
        self.target_file_id = target_file_id
        import time
        self.created_at = time.time()

class Settings(BASE):
    __tablename__ = "settings"
    user_id = Column(BigInteger, primary_key=True)
    precise_mode = Column(Boolean)
    button_mode = Column(Boolean)
    link_mode = Column(Boolean)
    list_mode = Column(Boolean)

    def __init__(self, user_id, precise_mode, button_mode, link_mode, list_mode):
        self.user_id = user_id
        self.precise_mode = precise_mode
        self.button_mode = button_mode
        self.link_mode = link_mode
        self.list_mode = list_mode

class UserTokens(BASE):
    __tablename__ = "user_tokens"
    user_id = Column(BigInteger, primary_key=True)
    token = Column(TEXT)
    created_at = Column(Numeric)
    expires_at = Column(Numeric)
    status = Column(TEXT)
    search_query = Column(TEXT)
    target_file_id = Column(TEXT)
    req_msg_id = Column(Numeric, nullable=True)

    def __init__(self, user_id, token, created_at, expires_at=0, status="pending", search_query=None, target_file_id=None, req_msg_id=None):
        self.user_id = user_id
        self.token = token
        self.created_at = created_at
        self.expires_at = expires_at
        self.status = status
        self.search_query = search_query
        self.target_file_id = target_file_id
        self.req_msg_id = req_msg_id

class UserShortenerUsage(BASE):
    __tablename__ = "user_shortener_usage"
    user_id = Column(BigInteger, primary_key=True)
    last_index = Column(Integer, default=0)
    last_date = Column(TEXT)  # Format: YYYY-MM-DD

    def __init__(self, user_id, last_index=0, last_date=""):
        self.user_id = user_id
        self.last_index = last_index
        self.last_date = last_date

# SQLite configuration for auto-delete queue to save Neon compute hours
SQLITE_DB_URL = "sqlite:///mfinder_autodelete.db"
sqlite_engine = create_engine(SQLITE_DB_URL, pool_pre_ping=True)
sqlite_session_maker = sessionmaker(bind=sqlite_engine, autoflush=False)
SQLITE_SESSION = scoped_session(sqlite_session_maker)
SQLITE_BASE = declarative_base()

class AutoDeleteMessage(SQLITE_BASE):
    __tablename__ = "auto_delete_messages"
    id = Column(Integer, primary_key=True, autoincrement=True)
    chat_id = Column(Numeric)
    message_id = Column(Integer)
    delete_at = Column(Numeric)  # Unix timestamp

    def __init__(self, chat_id, message_id, delete_at):
        self.chat_id = chat_id
        self.message_id = message_id
        self.delete_at = delete_at

def start() -> scoped_session:
    connect_args = {}
    if DB_URL and DB_URL.startswith(("postgres://", "postgresql://")):
        connect_args["sslmode"] = "require"
    engine = create_engine(
        DB_URL,
        connect_args=connect_args,
        poolclass=QueuePool,
        pool_size=10,
        max_overflow=20,
        pool_pre_ping=True,
        pool_recycle=1800  # Recycle connections every 1800 seconds (30 minutes)
    )
    BASE.metadata.bind = engine
    BASE.metadata.create_all(engine)
    SQLITE_BASE.metadata.bind = sqlite_engine
    SQLITE_BASE.metadata.create_all(sqlite_engine)
    


    # Migrate and drop legacy columns
    from sqlalchemy import text
    try:
        from sqlalchemy import inspect
        inspector = inspect(engine)
        columns = [col["name"] for col in inspector.get_columns("admin_settings")]
        
        # Only migrate and drop if obsolete columns actually exist in the table
        has_obsolete = any(c in columns for c in ["shortener_site", "shortener_api", "how_to_download"])
        
        if has_obsolete:
            
            # 1. Attempt to migrate old settings
            old_site = None
            old_api = None
            old_download = None
            shorteners_list_json = None
            
            col_selectors = []
            if "shortener_site" in columns:
                col_selectors.append("shortener_site")
            if "shortener_api" in columns:
                col_selectors.append("shortener_api")
            if "how_to_download" in columns:
                col_selectors.append("how_to_download")
                
            if len(col_selectors) > 0 and "shorteners_list" in columns:
                col_selectors.append("shorteners_list")
                try:
                    with engine.connect() as conn:
                        query_str = f"SELECT {', '.join(col_selectors)} FROM admin_settings LIMIT 1"
                        res = conn.execute(text(query_str)).fetchone()
                        if res:
                            val_map = dict(zip(col_selectors, res))
                            old_site = val_map.get("shortener_site")
                            old_api = val_map.get("shortener_api")
                            old_download = val_map.get("how_to_download")
                            shorteners_list_json = val_map.get("shorteners_list")
                            
                            if old_site and old_api:
                                import json
                                try:
                                    shorteners = json.loads(shorteners_list_json) if shorteners_list_json else []
                                except Exception:
                                    shorteners = []
                                exists = any(s.get("site") == old_site and s.get("api") == old_api for s in shorteners)
                                if not exists:
                                    shorteners.append({
                                        "site": old_site,
                                        "api": old_api,
                                        "how_to_download": old_download,
                                        "status": True
                                    })
                                    with engine.begin() as t_conn:
                                        t_conn.execute(
                                            text("UPDATE admin_settings SET shorteners_list = :sl"),
                                            {"sl": json.dumps(shorteners)}
                                        )
                                    LOGGER.info(f"Successfully migrated old shortener {old_site} to shorteners_list.")
                except Exception as e:
                    LOGGER.warning(f"Error migrating old settings: {e}")
                    
            # 2. Drop the obsolete columns since they exist (each column drop in its own transaction block)
            for col_to_drop in ["shortener_site", "shortener_api", "how_to_download"]:
                if col_to_drop in columns:
                    try:
                        with engine.begin() as t_conn:
                            t_conn.execute(text(f"ALTER TABLE admin_settings DROP COLUMN IF EXISTS {col_to_drop}"))
                        LOGGER.info(f"Obsolete column {col_to_drop} successfully dropped from admin_settings table.")
                    except Exception as e:
                        LOGGER.warning(f"Error dropping column {col_to_drop}: {e}")
    except Exception as e:
        LOGGER.warning(f"Error checking/dropping obsolete columns: {e}")

    from sqlalchemy import text
    try:
        with engine.connect() as conn:
            with conn.begin():
                # conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS shortener_site TEXT"))
                # conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS shortener_api TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS shortener_status BOOLEAN DEFAULT TRUE"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS admins_list TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS db_channels_list TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS token_shortener_enabled BOOLEAN DEFAULT FALSE"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS token_timeout NUMERIC DEFAULT 3600"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS log_channel NUMERIC"))
                # conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS how_to_download TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS log_channel_access_hash NUMERIC"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS fsub_channel_access_hash NUMERIC"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS fsub_channels_list TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS fsub_channels_access_hashes TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS shorteners_list TEXT"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS smart_rotator BOOLEAN DEFAULT FALSE"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS newsletter_days TEXT DEFAULT 'Friday'"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS newsletter_time TEXT DEFAULT '20:00'"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS newsletter_count INTEGER DEFAULT 10"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS newsletter_enabled BOOLEAN DEFAULT FALSE"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS newsletter_target TEXT DEFAULT 'subscribed'"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS announcement_enabled BOOLEAN DEFAULT FALSE"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS announcement_channel NUMERIC"))
                conn.execute(text("ALTER TABLE admin_settings ADD COLUMN IF NOT EXISTS announcement_channel_access_hash NUMERIC"))
                conn.execute(text("""
                    CREATE TABLE IF NOT EXISTS user_shortener_usage (
                        user_id BIGINT PRIMARY KEY,
                        last_index INTEGER DEFAULT 0,
                        last_date TEXT
                    )
                """))
                conn.execute(text("ALTER TABLE user_tokens ADD COLUMN IF NOT EXISTS req_msg_id NUMERIC"))
                conn.execute(text("""
                    CREATE TABLE IF NOT EXISTS pending_fsub_requests (
                        user_id NUMERIC PRIMARY KEY,
                        search_query TEXT,
                        target_file_id TEXT,
                        created_at NUMERIC
                    )
                """))
                try:
                    conn.execute(text("ALTER TABLE pending_fsub_requests ALTER COLUMN user_id TYPE NUMERIC"))
                except Exception:
                    pass
    except Exception as e:
        LOGGER.warning("Could not execute DDL migrations for admin_settings: %s", str(e))

    return scoped_session(sessionmaker(bind=engine, autoflush=False))

SESSION = start()

import asyncio
import time
import functools
from sqlalchemy.exc import OperationalError, PendingRollbackError

def db_retry(func):
    if asyncio.iscoroutinefunction(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(3):
                try:
                    return await func(*args, **kwargs)
                except (OperationalError, PendingRollbackError) as e:
                    LOGGER.warning("Database connection error in %s (attempt %s): %s. Reconnecting...", func.__name__, attempt + 1, str(e))
                    try:
                        SESSION.rollback()
                        SESSION.remove()
                    except Exception:
                        pass
                    if attempt == 2:
                        raise
                    await asyncio.sleep(0.5)
        return wrapper
    else:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(3):
                try:
                    return func(*args, **kwargs)
                except (OperationalError, PendingRollbackError) as e:
                    LOGGER.warning("Database connection error in %s (attempt %s): %s. Reconnecting...", func.__name__, attempt + 1, str(e))
                    try:
                        SESSION.rollback()
                        SESSION.remove()
                    except Exception:
                        pass
                    if attempt == 2:
                        raise
                    time.sleep(0.5)
        return wrapper

# ----------------- IN-MEMORY CACHE & MUTEX LOCK -----------------
class CachedAdminSettings:
    def __init__(self, admin_settings_db=None):
        self.setting_name = "default"
        self.auto_delete = 0
        self.custom_caption = None
        self.fsub_channel = None
        self.channel_link = None
        self.caption_uname = None
        self.repair_mode = False
        self.shortener_status = True
        self.admins_list = None
        self.db_channels_list = None
        self.token_shortener_enabled = False
        self.token_timeout = 3600
        self.log_channel = None
        self.log_channel_access_hash = None
        self.fsub_channel_access_hash = None
        self.fsub_channels_list = None
        self.fsub_channels_access_hashes = "{}"
        self.shorteners_list = "[]"
        self.smart_rotator = False
        self.newsletter_days = "Friday"
        self.newsletter_time = "20:00"
        self.newsletter_count = 10
        self.newsletter_enabled = False
        self.newsletter_target = "subscribed"
        self.announcement_enabled = False
        self.announcement_channel = None
        self.announcement_channel_access_hash = None
        self.shortener_site = None
        self.shortener_api = None
        self.how_to_download = None

        if admin_settings_db:
            def to_numeric(val):
                if val is None:
                    return None
                try:
                    val_str = str(val)
                    if "." not in val_str:
                        return int(val)
                    f_val = float(val)
                    if f_val.is_integer():
                        return int(f_val)
                    return f_val
                except (ValueError, TypeError):
                    return val

            self.setting_name = admin_settings_db.setting_name
            self.auto_delete = to_numeric(admin_settings_db.auto_delete) or 0
            self.custom_caption = admin_settings_db.custom_caption
            self.fsub_channel = to_numeric(admin_settings_db.fsub_channel)
            self.channel_link = admin_settings_db.channel_link
            self.caption_uname = admin_settings_db.caption_uname
            self.repair_mode = bool(admin_settings_db.repair_mode)
            self.shortener_status = bool(admin_settings_db.shortener_status) if admin_settings_db.shortener_status is not None else True
            self.admins_list = admin_settings_db.admins_list
            self.db_channels_list = admin_settings_db.db_channels_list
            self.token_shortener_enabled = bool(admin_settings_db.token_shortener_enabled) if admin_settings_db.token_shortener_enabled is not None else False
            self.token_timeout = to_numeric(admin_settings_db.token_timeout) or 3600
            self.log_channel = to_numeric(admin_settings_db.log_channel)
            self.log_channel_access_hash = to_numeric(admin_settings_db.log_channel_access_hash)
            self.fsub_channel_access_hash = to_numeric(admin_settings_db.fsub_channel_access_hash)
            self.fsub_channels_list = admin_settings_db.fsub_channels_list
            self.fsub_channels_access_hashes = admin_settings_db.fsub_channels_access_hashes or "{}"
            self.shorteners_list = admin_settings_db.shorteners_list or "[]"
            self.smart_rotator = bool(admin_settings_db.smart_rotator) if admin_settings_db.smart_rotator is not None else False
            self.newsletter_days = admin_settings_db.newsletter_days or "Friday"
            self.newsletter_time = admin_settings_db.newsletter_time or "20:00"
            self.newsletter_count = to_numeric(admin_settings_db.newsletter_count) or 10
            self.newsletter_enabled = bool(admin_settings_db.newsletter_enabled) if admin_settings_db.newsletter_enabled is not None else False
            self.newsletter_target = admin_settings_db.newsletter_target or "subscribed"
            self.announcement_enabled = bool(admin_settings_db.announcement_enabled) if admin_settings_db.announcement_enabled is not None else False
            self.announcement_channel = to_numeric(admin_settings_db.announcement_channel)
            self.announcement_channel_access_hash = to_numeric(admin_settings_db.announcement_channel_access_hash)

            # Dynamically resolve legacy columns from shorteners_list JSON
            import json
            try:
                sh_list = json.loads(self.shorteners_list)
            except Exception:
                sh_list = []
            if sh_list:
                first_sh = sh_list[0]
                self.shortener_site = first_sh.get("site")
                self.shortener_api = first_sh.get("api")
                self.how_to_download = first_sh.get("how_to_download")
            else:
                self.shortener_site = None
                self.shortener_api = None
                self.how_to_download = None

_cached_admin_settings = None
_cache_lock = threading.Lock()

def _update_cache_from_db(admin_setting):
    global _cached_admin_settings
    with _cache_lock:
        _cached_admin_settings = CachedAdminSettings(admin_setting)

@db_retry
def get_admin_settings_sync():
    global _cached_admin_settings
    with _cache_lock:
        if _cached_admin_settings is not None:
            return _cached_admin_settings
        
        for attempt in range(3):
            try:
                session = SESSION()
                admin_setting = session.query(AdminSettings).first()
                if not admin_setting:
                    admin_setting = AdminSettings(setting_name="default")
                    session.add(admin_setting)
                    session.commit()
                _cached_admin_settings = CachedAdminSettings(admin_setting)
                return _cached_admin_settings
            except (OperationalError, PendingRollbackError) as e:
                LOGGER.warning("Database connection error in get_admin_settings_sync (attempt %d): %s. Reconnecting...", attempt + 1, str(e))
                try:
                    SESSION.rollback()
                except Exception:
                    pass
                try:
                    SESSION.remove()
                except Exception:
                    pass
                if attempt == 2:
                    return CachedAdminSettings()
                time.sleep(0.5)
            except Exception as e:
                LOGGER.warning("Error in get_admin_settings_sync: %s", str(e))
                try:
                    SESSION.rollback()
                except Exception:
                    pass
                try:
                    SESSION.remove()
                except Exception:
                    pass
                return CachedAdminSettings()
            finally:
                SESSION.close()

@db_retry
def heal_admin_settings():
    import os
    try:
        session = SESSION()
        admin_setting = session.query(AdminSettings).first()
        
        env_admins = os.environ.get("ADMINS", "")
        owner_id = os.environ.get("OWNER_ID", "")
        admins_set = set()
        for x in env_admins.split():
            x = x.strip()
            if x:
                admins_set.add(x)
        if owner_id:
            admins_set.add(owner_id.strip())
        default_admins = ",".join(sorted(list(admins_set))) if admins_set else None

        env_channels = os.environ.get("DB_CHANNELS", "")
        channels_list = []
        for x in env_channels.split():
            x = x.strip()
            if x:
                channels_list.append(x)
        default_channels = ",".join(channels_list) if channels_list else None

        # Parse helpers
        def parse_bool(env_val, default=None):
            if env_val is None:
                return default
            val = env_val.strip().lower()
            if val in ("true", "yes", "on", "1"):
                return True
            if val in ("false", "no", "off", "0"):
                return False
            return default

        def parse_int(env_val, default=None):
            if env_val is None:
                return default
            try:
                return int(env_val.strip())
            except ValueError:
                return default

        def parse_time(env_val, default=None):
            if env_val is None:
                return default
            val = env_val.strip().lower()
            if val.endswith("h"):
                try:
                    return int(float(val[:-1]) * 3600)
                except ValueError:
                    pass
            elif val.endswith("m"):
                try:
                    return int(float(val[:-1]) * 60)
                except ValueError:
                    pass
            elif val.endswith("d"):
                try:
                    return int(float(val[:-1]) * 86400)
                except ValueError:
                    pass
            elif val.endswith("s"):
                try:
                    return int(float(val[:-1]))
                except ValueError:
                    pass
            else:
                try:
                    return int(val)
                except ValueError:
                    pass
            return default

        env_auto_delete = os.environ.get("AUTO_DELETE")
        env_custom_caption = os.environ.get("CUSTOM_CAPTION")
        env_fsub_channel = os.environ.get("FSUB_CHANNEL")
        env_channel_link = os.environ.get("CHANNEL_LINK")
        env_caption_uname = os.environ.get("CAPTION_UNAME")
        env_repair_mode = os.environ.get("REPAIR_MODE")
        env_shortener_site = os.environ.get("SHORTENER_SITE")
        env_shortener_api = os.environ.get("SHORTENER_API")
        env_shortener_status = os.environ.get("SHORTENER_STATUS")
        env_token_shortener_enabled = os.environ.get("TOKEN_SHORTENER_ENABLED")
        env_token_timeout = os.environ.get("TOKEN_TIMEOUT")
        env_log_channel = os.environ.get("LOG_CHANNEL")
        env_how_to_download = os.environ.get("HOW_TO_DOWNLOAD")

        if not admin_setting:
            admin_setting = AdminSettings(setting_name="default")
            admin_setting.admins_list = default_admins
            admin_setting.db_channels_list = default_channels
            
            if env_auto_delete is not None:
                admin_setting.auto_delete = parse_int(env_auto_delete, 0)
            if env_custom_caption is not None:
                cc = env_custom_caption.strip()
                admin_setting.custom_caption = None if cc.lower() in ("off", "disable", "disabled", "none") else cc
            if env_fsub_channel is not None:
                admin_setting.fsub_channel = parse_int(env_fsub_channel)
            if env_channel_link is not None:
                cl = env_channel_link.strip()
                admin_setting.channel_link = None if cl.lower() in ("off", "disable", "disabled", "none") else cl
            if env_caption_uname is not None:
                cu = env_caption_uname.strip()
                admin_setting.caption_uname = None if cu.lower() in ("off", "disable", "disabled", "none") else cu
            if env_repair_mode is not None:
                admin_setting.repair_mode = parse_bool(env_repair_mode, False)
            if env_shortener_site and env_shortener_api:
                new_list, _ = merge_env_shortener_into_list(
                    admin_setting.shorteners_list,
                    env_shortener_site,
                    env_shortener_api,
                    env_how_to_download
                )
                admin_setting.shorteners_list = new_list
            if env_shortener_status is not None:
                admin_setting.shortener_status = parse_bool(env_shortener_status, True)
            if env_token_shortener_enabled is not None:
                admin_setting.token_shortener_enabled = parse_bool(env_token_shortener_enabled, False)
            if env_token_timeout is not None:
                admin_setting.token_timeout = parse_time(env_token_timeout, 3600)
            if env_log_channel is not None:
                admin_setting.log_channel = parse_int(env_log_channel)

                
            # Enforce mutual exclusion
            if admin_setting.token_shortener_enabled and admin_setting.shortener_status:
                admin_setting.shortener_status = False
                
            session.add(admin_setting)
            session.commit()
            LOGGER.info("Created default AdminSettings with env values.")
        else:
            updated = False
            # Check if admins_list is None, empty, or has the test dummy values
            if (not admin_setting.admins_list or 
                admin_setting.admins_list == "12345,67890" or 
                (owner_id and owner_id.strip() not in admin_setting.admins_list.split(','))):
                admin_setting.admins_list = default_admins
                updated = True
                
            # Check if db_channels_list is None, empty, or has the test dummy values
            if (not admin_setting.db_channels_list or 
                admin_setting.db_channels_list == "-100123,-100456"):
                admin_setting.db_channels_list = default_channels
                updated = True

            # Heal other environment variables
            if env_auto_delete is not None:
                parsed_auto_delete = parse_int(env_auto_delete)
                if parsed_auto_delete is not None and admin_setting.auto_delete != parsed_auto_delete:
                    admin_setting.auto_delete = parsed_auto_delete
                    updated = True
                    
            if env_custom_caption is not None:
                parsed_custom_caption = env_custom_caption.strip()
                if parsed_custom_caption.lower() in ("off", "disable", "disabled", "none"):
                    parsed_custom_caption = None
                if admin_setting.custom_caption != parsed_custom_caption:
                    admin_setting.custom_caption = parsed_custom_caption
                    updated = True
                    
            if env_fsub_channel is not None:
                parsed_fsub_channel = parse_int(env_fsub_channel)
                if parsed_fsub_channel is not None and admin_setting.fsub_channel != parsed_fsub_channel:
                    admin_setting.fsub_channel = parsed_fsub_channel
                    updated = True
                    
            if env_channel_link is not None:
                parsed_channel_link = env_channel_link.strip()
                if parsed_channel_link.lower() in ("off", "disable", "disabled", "none"):
                    parsed_channel_link = None
                if admin_setting.channel_link != parsed_channel_link:
                    admin_setting.channel_link = parsed_channel_link
                    updated = True
                    
            if env_caption_uname is not None:
                parsed_caption_uname = env_caption_uname.strip()
                if parsed_caption_uname.lower() in ("off", "disable", "disabled", "none"):
                    parsed_caption_uname = None
                if admin_setting.caption_uname != parsed_caption_uname:
                    admin_setting.caption_uname = parsed_caption_uname
                    updated = True
                    
            if env_repair_mode is not None:
                parsed_repair_mode = parse_bool(env_repair_mode)
                if parsed_repair_mode is not None and admin_setting.repair_mode != parsed_repair_mode:
                    admin_setting.repair_mode = parsed_repair_mode
                    updated = True
                    
            if env_shortener_site and env_shortener_api:
                new_list, list_updated = merge_env_shortener_into_list(
                    admin_setting.shorteners_list,
                    env_shortener_site,
                    env_shortener_api,
                    env_how_to_download
                )
                if list_updated:
                    admin_setting.shorteners_list = new_list
                    updated = True
                    
            if env_shortener_status is not None:
                parsed_shortener_status = parse_bool(env_shortener_status)
                if parsed_shortener_status is not None and admin_setting.shortener_status != parsed_shortener_status:
                    admin_setting.shortener_status = parsed_shortener_status
                    updated = True
                    
            if env_token_shortener_enabled is not None:
                parsed_token_shortener_enabled = parse_bool(env_token_shortener_enabled)
                if parsed_token_shortener_enabled is not None and admin_setting.token_shortener_enabled != parsed_token_shortener_enabled:
                    admin_setting.token_shortener_enabled = parsed_token_shortener_enabled
                    updated = True
                    
            if env_token_timeout is not None:
                parsed_token_timeout = parse_time(env_token_timeout)
                if parsed_token_timeout is not None and admin_setting.token_timeout != parsed_token_timeout:
                    admin_setting.token_timeout = parsed_token_timeout
                    updated = True
                    
            if env_log_channel is not None:
                parsed_log_channel = parse_int(env_log_channel)
                if parsed_log_channel is not None and admin_setting.log_channel != parsed_log_channel:
                    admin_setting.log_channel = parsed_log_channel
                    updated = True
                    

                    
            # Enforce mutual exclusion
            if admin_setting.token_shortener_enabled and admin_setting.shortener_status:
                admin_setting.shortener_status = False
                updated = True
                
            if updated:
                session.commit()
                LOGGER.info("Healed/restored AdminSettings from env values.")

        # Prime the cache on boot
        _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Could not heal admin settings: %s", str(e))
    finally:
        SESSION.close()

heal_admin_settings()

INSERTION_LOCK = threading.RLock()

@db_retry
async def get_search_settings(user_id):
    for attempt in range(3):
        try:
            with INSERTION_LOCK:
                settings = SESSION.query(Settings).filter_by(user_id=user_id).first()
                return settings
        except (OperationalError, PendingRollbackError) as e:
            LOGGER.warning("Database connection error in get_search_settings (attempt %d): %s. Reconnecting...", attempt + 1, str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            if attempt == 2:
                return None
            await asyncio.sleep(0.5)
        except Exception as e:
            LOGGER.warning("Error getting search settings: %s", str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            return None
        finally:
            SESSION.close()

@db_retry
async def change_search_settings(user_id, precise_mode=None, button_mode=None, link_mode=None, list_mode=None):
    try:
        with INSERTION_LOCK:
            settings = SESSION.query(Settings).filter_by(user_id=user_id).first()
            if settings:
                if precise_mode is not None:
                    settings.precise_mode = precise_mode
                if button_mode is not None:
                    settings.button_mode = button_mode
                if link_mode is not None:
                    settings.link_mode = link_mode
                if list_mode is not None:
                    settings.list_mode = list_mode
            else:
                new_settings = Settings(
                    user_id=user_id,
                    precise_mode=precise_mode if precise_mode is not None else False,
                    button_mode=button_mode if button_mode is not None else False,
                    link_mode=link_mode if link_mode is not None else True,
                    list_mode=list_mode if list_mode is not None else False
                )
                SESSION.add(new_settings)
            SESSION.commit()
            return True
    except Exception as e:
        LOGGER.warning("Error changing search settings: %s", str(e))
        try:
            SESSION.rollback()
        except Exception:
            pass
        try:
            SESSION.remove()
        except Exception:
            pass
        return False
    finally:
        SESSION.close()

@db_retry
async def set_repair_mode(repair_mode):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.repair_mode = repair_mode
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting repair mode: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def set_auto_delete(dur):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.auto_delete = dur
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting auto delete: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def get_admin_settings():
    return get_admin_settings_sync()

@db_retry
async def set_custom_caption(caption):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.custom_caption = caption
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting custom caption: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def set_force_sub(channel, access_hash=None):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.fsub_channel = channel
            if access_hash is not None:
                admin_setting.fsub_channel_access_hash = access_hash
                
            # Keep fsub_channels_list in sync for backwards compatibility
            if channel:
                admin_setting.fsub_channels_list = str(channel)
                if access_hash is not None:
                    import json
                    admin_setting.fsub_channels_access_hashes = json.dumps({str(channel): access_hash})
            else:
                admin_setting.fsub_channels_list = None
                admin_setting.fsub_channels_access_hashes = "{}"

            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting Force Sub channel: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def set_force_sub_channels(channels_list, hashes_map):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.fsub_channels_list = channels_list
            import json
            admin_setting.fsub_channels_access_hashes = json.dumps(hashes_map)
            
            # Sync first channel to fsub_channel legacy columns for safety
            if channels_list:
                first_ch = channels_list.split(',')[0].strip()
                try:
                    admin_setting.fsub_channel = int(first_ch)
                except ValueError:
                    admin_setting.fsub_channel = None
                admin_setting.fsub_channel_access_hash = hashes_map.get(first_ch)
            else:
                admin_setting.fsub_channel = None
                admin_setting.fsub_channel_access_hash = None

            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting Force Sub channels list: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def save_pending_fsub_request(user_id, search_query=None, target_file_id=None):
    import time
    try:
        with INSERTION_LOCK:
            session = SESSION()
            req = session.query(PendingFsubRequests).filter(PendingFsubRequests.user_id == user_id).first()
            if req:
                req.search_query = search_query
                req.target_file_id = target_file_id
                req.created_at = time.time()
            else:
                req = PendingFsubRequests(user_id=user_id, search_query=search_query, target_file_id=target_file_id)
                req.created_at = time.time()
                session.add(req)
            session.commit()
    except Exception as e:
        LOGGER.warning("Error saving pending fsub request: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def get_pending_fsub_request(user_id):
    for attempt in range(3):
        try:
            session = SESSION()
            req = session.query(PendingFsubRequests).filter(PendingFsubRequests.user_id == user_id).first()
            return req
        except (OperationalError, PendingRollbackError) as e:
            LOGGER.warning("Database connection error in get_pending_fsub_request (attempt %d): %s. Reconnecting...", attempt + 1, str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            if attempt == 2:
                return None
            await asyncio.sleep(0.5)
        except Exception as e:
            LOGGER.warning("Error getting pending fsub request: %s", str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            return None
        finally:
            session.close()

@db_retry
async def delete_pending_fsub_request(user_id):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            session.query(PendingFsubRequests).filter(PendingFsubRequests.user_id == user_id).delete()
            session.commit()
    except Exception as e:
        LOGGER.warning("Error deleting pending fsub request: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def set_channel_link(link):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.channel_link = link
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error adding Force Sub channel link: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def get_channel():
    settings = get_admin_settings_sync()
    return settings.fsub_channel if settings.fsub_channel else False

@db_retry
async def get_link():
    settings = get_admin_settings_sync()
    return settings.channel_link if settings.channel_link else False

@db_retry
async def set_username(username):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.caption_uname = username
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error adding username: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def set_shortener_settings(site=None, api=None, status=None):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            import json
            try:
                shorteners = json.loads(admin_setting.shorteners_list) if admin_setting.shorteners_list else []
            except Exception:
                shorteners = []

            if site is not None or api is not None:
                if shorteners:
                    if site is not None:
                        shorteners[0]["site"] = site
                    if api is not None:
                        shorteners[0]["api"] = api
                else:
                    shorteners.append({
                        "site": site or "",
                        "api": api or "",
                        "how_to_download": None,
                        "status": True
                    })
                admin_setting.shorteners_list = json.dumps(shorteners)

            if status is not None:
                admin_setting.shortener_status = status
                if status:
                    admin_setting.token_shortener_enabled = False
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting shortener settings: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def delete_shortener_settings():
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if admin_setting:
                import json
                try:
                    shorteners = json.loads(admin_setting.shorteners_list) if admin_setting.shorteners_list else []
                except Exception:
                    shorteners = []
                if shorteners:
                    shorteners.pop(0)
                admin_setting.shorteners_list = json.dumps(shorteners)
                admin_setting.shortener_status = False
                session.commit()
                _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error deleting shortener settings: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def update_admins_list(admins_str):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.admins_list = admins_str
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error updating admins list: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def update_db_channels_list(channels_str):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.db_channels_list = channels_str
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error updating db channels list: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def get_user_token(user_id):
    for attempt in range(3):
        try:
            with INSERTION_LOCK:
                session = SESSION()
                return session.query(UserTokens).filter_by(user_id=user_id).first()
        except (OperationalError, PendingRollbackError) as e:
            LOGGER.warning("Database connection error in get_user_token (attempt %d): %s. Reconnecting...", attempt + 1, str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            if attempt == 2:
                return None
            await asyncio.sleep(0.5)
        except Exception as e:
            LOGGER.warning("Error getting user token: %s", str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            return None
        finally:
            SESSION.close()

@db_retry
async def create_user_token(user_id, token, search_query=None, target_file_id=None):
    import time
    try:
        await prune_all_expired_tokens()
        with INSERTION_LOCK:
            session = SESSION()
            tok_rec = session.query(UserTokens).filter_by(user_id=user_id).first()
            if tok_rec:
                tok_rec.token = token
                tok_rec.created_at = time.time()
                tok_rec.expires_at = 0
                tok_rec.status = "pending"
                tok_rec.search_query = search_query
                tok_rec.target_file_id = target_file_id
            else:
                tok_rec = UserTokens(
                    user_id=user_id,
                    token=token,
                    created_at=time.time(),
                    search_query=search_query,
                    target_file_id=target_file_id
                )
                session.add(tok_rec)
            session.commit()
            return tok_rec
    except Exception as e:
        LOGGER.warning("Error creating user token: %s", str(e))
        SESSION.rollback()
        return None
    finally:
        SESSION.close()

@db_retry
async def activate_user_token(user_id, expires_at):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            tok_rec = session.query(UserTokens).filter_by(user_id=user_id).first()
            if tok_rec:
                tok_rec.status = "active"
                tok_rec.expires_at = expires_at
                session.commit()
                return True
            return False
    except Exception as e:
        LOGGER.warning("Error activating user token: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def delete_user_token(user_id):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            tok_rec = session.query(UserTokens).filter_by(user_id=user_id).first()
            if tok_rec:
                session.delete(tok_rec)
                session.commit()
                return True
            return False
    except Exception as e:
        LOGGER.warning("Error deleting user token: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def prune_all_expired_tokens():
    import time
    try:
        with INSERTION_LOCK:
            session = SESSION()
            current_time = time.time()
            
            # Delete active tokens that have expired
            expired_active = session.query(UserTokens).filter(
                UserTokens.status == "active",
                UserTokens.expires_at < current_time
            )
            
            # Delete pending tokens that are older than 30 minutes (1800 seconds)
            stale_pending = session.query(UserTokens).filter(
                UserTokens.status == "pending",
                (current_time - UserTokens.created_at) > 1800
            )
            
            deleted_active = expired_active.delete(synchronize_session=False)
            deleted_pending = stale_pending.delete(synchronize_session=False)
            
            if deleted_active or deleted_pending:
                session.commit()
                LOGGER.info("Database row protection: Pruned %d expired active and %d stale pending tokens.", deleted_active, deleted_pending)
            return True
    except Exception as e:
        LOGGER.warning("Error pruning expired tokens from database: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def set_token_shortener_state(enabled):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
            admin_setting.token_shortener_enabled = enabled
            if enabled:
                admin_setting.shortener_status = False
            session.commit()
            _update_cache_from_db(admin_setting)
            return True
    except Exception as e:
        LOGGER.warning("Error setting token shortener state: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def set_token_timeout(seconds):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
            admin_setting.token_timeout = seconds
            session.commit()
            _update_cache_from_db(admin_setting)
            return True
    except Exception as e:
        LOGGER.warning("Error setting token timeout: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def set_log_channel(channel, access_hash=None):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
            admin_setting.log_channel = channel
            if access_hash is not None:
                admin_setting.log_channel_access_hash = access_hash
            session.commit()
            _update_cache_from_db(admin_setting)
            return True
    except Exception as e:
        LOGGER.warning("Error setting log channel: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def get_log_channel():
    settings = get_admin_settings_sync()
    return settings.log_channel if settings.log_channel else False

@db_retry
async def set_how_to_download(url):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
            admin_setting.how_to_download = url
            session.commit()
            _update_cache_from_db(admin_setting)
            return True
    except Exception as e:
        LOGGER.warning("Error setting how to download URL: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def update_user_token_msg_id(user_id, msg_id):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            tok_rec = session.query(UserTokens).filter_by(user_id=user_id).first()
            if tok_rec:
                tok_rec.req_msg_id = msg_id
                session.commit()
                return True
            return False
    except Exception as e:
        LOGGER.warning("Error updating user token message ID: %s", str(e))
        SESSION.rollback()
        return False
    finally:
        SESSION.close()

@db_retry
async def set_smart_rotator_state(state: bool):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.smart_rotator = state
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting smart rotator state: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def update_shorteners_list(shorteners_json_str):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            admin_setting.shorteners_list = shorteners_json_str
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error updating shorteners list: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def get_user_shortener_index(user_id: int, today_str: str) -> int:
    for attempt in range(3):
        try:
            session = SESSION()
            record = session.query(UserShortenerUsage).filter_by(user_id=user_id).first()
            if record and record.last_date == today_str:
                return record.last_index or 0
            return 0
        except (OperationalError, PendingRollbackError) as e:
            LOGGER.warning("Database connection error in get_user_shortener_index: %s. Reconnecting...", str(e))
            try:
                SESSION.rollback()
            except Exception:
                pass
            try:
                SESSION.remove()
            except Exception:
                pass
            if attempt == 2:
                return -1
            await asyncio.sleep(0.5)
        except Exception as e:
            LOGGER.warning("Error in get_user_shortener_index: %s", str(e))
            return -1
        finally:
            SESSION.close()

@db_retry
async def update_user_shortener_index(user_id: int, index: int, today_str: str):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            record = session.query(UserShortenerUsage).filter_by(user_id=user_id).first()
            if record:
                record.last_index = index
                record.last_date = today_str
            else:
                record = UserShortenerUsage(user_id=user_id, last_index=index, last_date=today_str)
                session.add(record)
            session.commit()
    except Exception as e:
        LOGGER.warning("Error in update_user_shortener_index: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def advance_user_shortener_index(user_id: int):
    try:
        import datetime
        local_now = datetime.datetime.utcnow() + datetime.timedelta(hours=5, minutes=30)
        today_str = local_now.strftime("%Y-%m-%d")
        with INSERTION_LOCK:
            session = SESSION()
            record = session.query(UserShortenerUsage).filter_by(user_id=user_id).first()
            if record:
                if record.last_date == today_str:
                    record.last_index = (record.last_index or 0) + 1
                else:
                    record.last_index = 1
                    record.last_date = today_str
            else:
                record = UserShortenerUsage(user_id=user_id, last_index=1, last_date=today_str)
                session.add(record)
            session.commit()
    except Exception as e:
        LOGGER.warning("Error in advance_user_shortener_index: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()

@db_retry
async def clear_all_user_shortener_usages():
    try:
        with INSERTION_LOCK:
            session = SESSION()
            session.query(UserShortenerUsage).delete()
            session.commit()
    except Exception as e:
        LOGGER.warning("Error clearing UserShortenerUsage: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()


@db_retry
async def set_newsletter_settings(enabled=None, days=None, time_str=None, count=None, target=None):
    try:
        with INSERTION_LOCK:
            session = SESSION()
            admin_setting = session.query(AdminSettings).first()
            if not admin_setting:
                admin_setting = AdminSettings(setting_name="default")
                session.add(admin_setting)
                session.commit()

            if enabled is not None:
                admin_setting.newsletter_enabled = enabled
            if days is not None:
                admin_setting.newsletter_days = days
            if time_str is not None:
                admin_setting.newsletter_time = time_str
            if count is not None:
                admin_setting.newsletter_count = count
            if target is not None:
                admin_setting.newsletter_target = target
            session.commit()
            _update_cache_from_db(admin_setting)
    except Exception as e:
        LOGGER.warning("Error setting newsletter settings: %s", str(e))
        SESSION.rollback()
    finally:
        SESSION.close()


def add_auto_delete_msg_sync(chat_id, message_id, delay):
    delete_at = time.time() + delay
    session = SQLITE_SESSION()
    try:
        msg = AutoDeleteMessage(chat_id=chat_id, message_id=message_id, delete_at=delete_at)
        session.add(msg)
        session.commit()
    except Exception as e:
        session.rollback()
        LOGGER.warning(f"Error in add_auto_delete_msg_sync: {e}")
    finally:
        SQLITE_SESSION.remove()

def remove_auto_delete_msg_sync(chat_id, message_id):
    session = SQLITE_SESSION()
    try:
        session.query(AutoDeleteMessage).filter_by(chat_id=chat_id, message_id=message_id).delete()
        session.commit()
    except Exception as e:
        session.rollback()
        LOGGER.warning(f"Error in remove_auto_delete_msg_sync: {e}")
    finally:
        SQLITE_SESSION.remove()

def get_all_auto_delete_msgs_sync():
    session = SQLITE_SESSION()
    try:
        all_msgs = session.query(AutoDeleteMessage).all()
        return [(msg.chat_id, msg.message_id, msg.delete_at) for msg in all_msgs]
    except Exception as e:
        LOGGER.warning(f"Error in get_all_auto_delete_msgs_sync: {e}")
        return []
    finally:
        SQLITE_SESSION.remove()

async def set_announcement_settings(enabled: bool = None, channel = None, access_hash = None) -> None:
    retries = 3
    while retries > 0:
        try:
            with INSERTION_LOCK:
                session = SESSION()
                settings = session.query(AdminSettings).first()
                if not settings:
                    settings = AdminSettings(setting_name="default")
                    session.add(settings)
                
                if enabled is not None:
                    settings.announcement_enabled = enabled
                if channel is not None:
                    if str(channel).lower() == "off":
                        settings.announcement_channel = None
                        settings.announcement_channel_access_hash = None
                    else:
                        settings.announcement_channel = int(channel)
                if access_hash is not None:
                    settings.announcement_channel_access_hash = int(access_hash)
                session.commit()
                _update_cache_from_db(settings)
                return True
        except OperationalError:
            reconnect_session()
            retries -= 1
        except Exception as e:
            LOGGER.warning(f"Error setting announcement settings: {e}")
            session.rollback()
            retries -= 1
        finally:
            try:
                session.close()
            except Exception:
                pass
    return False