# retoor <retoor@molodetz.nl>

import dataset
import logging
from pathlib import Path
from sqlalchemy import or_
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from devplacepy.cache import TTLCache
from devplacepy.config import DATABASE_URL, INTERNAL_GATEWAY_URL, ensure_data_dirs

logger = logging.getLogger(__name__)

ensure_data_dirs()
if DATABASE_URL.startswith("sqlite:///"):
    _db_file = DATABASE_URL[len("sqlite:///") :]
    if _db_file and _db_file != ":memory:":
        Path(_db_file).parent.mkdir(parents=True, exist_ok=True)

db = dataset.connect(
    DATABASE_URL,
    engine_kwargs={
        "connect_args": {
            "timeout": 30,
            "check_same_thread": False,
        },
    },
    on_connect_statements=[
        "PRAGMA journal_mode=WAL",
        "PRAGMA synchronous=NORMAL",
        "PRAGMA busy_timeout=30000",
        "PRAGMA cache_size=-8000",
        "PRAGMA temp_store=MEMORY",
        "PRAGMA mmap_size=268435456",
    ],
)


def refresh_snapshot() -> None:
    connection = db.executable
    if connection.in_transaction() and not db.in_transaction:
        connection.commit()


_local_cache_versions: dict = {}
_cache_state_ready = False


def _ensure_cache_state() -> None:
    global _cache_state_ready
    if _cache_state_ready:
        return
    db.query(
        "CREATE TABLE IF NOT EXISTS cache_state "
        "(name TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 0)"
    )
    _cache_state_ready = True


def get_cache_version(name: str) -> int:
    try:
        _ensure_cache_state()
        row = next(
            iter(
                db.query(
                    "SELECT version FROM cache_state WHERE name = :name", name=name
                )
            ),
            None,
        )
        return int(row["version"]) if row else 0
    except Exception as e:
        logger.warning(f"Could not read cache version {name}: {e}")
        return 0


def bump_cache_version(name: str) -> None:
    try:
        _ensure_cache_state()
        db.query(
            "INSERT OR IGNORE INTO cache_state (name, version) VALUES (:name, 0)",
            name=name,
        )
        db.query(
            "UPDATE cache_state SET version = version + 1 WHERE name = :name", name=name
        )
        connection = db.executable
        if connection.in_transaction():
            connection.commit()
    except Exception as e:
        logger.warning(f"Could not bump cache version {name}: {e}")


def sync_local_cache(name: str, cache) -> None:
    current = get_cache_version(name)
    if name not in _local_cache_versions:
        _local_cache_versions[name] = current
        return
    if _local_cache_versions[name] != current:
        cache.clear()
        _local_cache_versions[name] = current


def _index(db, table, name, columns):
    try:
        if table in db.tables:
            cols = ", ".join(columns)
            db.query(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({cols})")
    except Exception as e:
        logger.warning(f"Could not create index {name} on {table}: {e}")


SOFT_DELETE_TABLES = [
    "posts",
    "comments",
    "gists",
    "projects",
    "news",
    "news_images",
    "project_files",
    "attachments",
    "votes",
    "reactions",
    "bookmarks",
    "follows",
    "poll_votes",
    "polls",
    "poll_options",
    "sessions",
    "instances",
    "instance_schedules",
    "devii_conversations",
    "devii_tasks",
    "devii_lessons",
    "devii_virtual_tools",
    "user_customizations",
    "project_forks",
    "bug_tickets",
    "bug_comment_authors",
    "notification_preferences",
    "deepsearch_sessions",
    "deepsearch_messages",
]


def ensure_soft_delete_columns(table, *, db_handle=None):
    handle = db_handle or db
    if table not in handle.tables:
        return
    target = handle[table]
    if not target.has_column("deleted_at"):
        target.create_column_by_example("deleted_at", "")
    if not target.has_column("deleted_by"):
        target.create_column_by_example("deleted_by", "")
    _index(handle, table, f"idx_{table}_deleted", ["deleted_at"])


def init_db():
    tables = db.tables
    _index(db, "users", "idx_users_username", ["username"])
    _index(db, "users", "idx_users_email", ["email"])
    _index(db, "users", "idx_users_api_key", ["api_key"])
    _index(db, "posts", "idx_posts_user_uid", ["user_uid"])
    _index(db, "posts", "idx_posts_created_at", ["created_at"])
    _index(db, "posts", "idx_posts_topic", ["topic"])
    _index(db, "comments", "idx_comments_post_uid", ["post_uid"])
    _index(db, "comments", "idx_comments_target", ["target_type", "target_uid"])
    _index(db, "comments", "idx_comments_user_uid", ["user_uid"])
    _index(db, "comments", "idx_comments_created_at", ["created_at"])
    _index(db, "votes", "idx_votes_target", ["target_uid", "target_type"])
    _index(db, "messages", "idx_messages_sender", ["sender_uid"])
    _index(db, "messages", "idx_messages_receiver", ["receiver_uid"])
    _index(db, "notifications", "idx_notifications_user", ["user_uid"])
    _index(db, "notifications", "idx_notifications_user_read", ["user_uid", "read"])
    _index(db, "push_registration", "idx_push_registration_user", ["user_uid"])
    _index(db, "sessions", "idx_sessions_token", ["session_token"])
    _index(db, "projects", "idx_projects_user", ["user_uid"])
    _index(db, "projects", "idx_projects_private", ["is_private"])
    _index(db, "project_files", "idx_project_files_path", ["project_uid", "path"])
    _index(
        db, "project_files", "idx_project_files_parent", ["project_uid", "parent_path"]
    )
    _index(db, "badges", "idx_badges_user", ["user_uid"])
    _index(db, "follows", "idx_follows_follower", ["follower_uid"])
    _index(db, "follows", "idx_follows_following", ["following_uid"])
    _index(db, "password_resets", "idx_password_resets_token", ["token"])
    _index(db, "gists", "idx_gists_user_uid", ["user_uid"])
    _index(db, "gists", "idx_gists_language", ["language"])
    attachments = get_table("attachments")
    for column, example in (
        ("uid", ""),
        ("target_type", ""),
        ("target_uid", ""),
        ("resource_type", ""),
        ("resource_uid", ""),
        ("user_uid", ""),
        ("original_filename", ""),
        ("stored_name", ""),
        ("directory", ""),
        ("file_size", 0),
        ("mime_type", ""),
        ("image_width", 0),
        ("image_height", 0),
        ("has_thumbnail", 0),
        ("thumbnail_name", ""),
        ("created_at", ""),
        ("deleted_at", ""),
    ):
        if not attachments.has_column(column):
            attachments.create_column_by_example(column, example)

    _index(
        db, "attachments", "idx_attachments_resource", ["resource_type", "resource_uid"]
    )
    _index(db, "attachments", "idx_attachments_target", ["target_type", "target_uid"])
    _index(db, "attachments", "idx_attachments_user_created", ["user_uid", "created_at"])
    _index(db, "reactions", "idx_reactions_target", ["target_type", "target_uid"])
    _index(
        db,
        "reactions",
        "idx_reactions_user_target",
        ["user_uid", "target_type", "target_uid"],
    )
    _index(db, "bookmarks", "idx_bookmarks_user", ["user_uid"])
    _index(db, "bookmarks", "idx_bookmarks_target", ["target_type", "target_uid"])
    _index(db, "polls", "idx_polls_post", ["post_uid"])
    _index(db, "poll_options", "idx_poll_options_poll", ["poll_uid"])
    _index(db, "poll_votes", "idx_poll_votes_poll", ["poll_uid"])
    _index(db, "poll_votes", "idx_poll_votes_user", ["poll_uid", "user_uid"])

    if "site_settings" in tables:
        defaults = {
            "site_name": "DevPlace",
            "site_description": "The Developer Social Network",
            "site_tagline": "Track industry shifts. Discover bold releases. Share what you are building in an open, uncensored environment.",
        }
        for key, value in defaults.items():
            existing = db["site_settings"].find_one(key=key)
            if not existing:
                db["site_settings"].insert(
                    {"uid": f"default_{key}", "key": key, "value": value}
                )

    news = get_table("news")
    for column, example in (
        ("uid", ""),
        ("slug", ""),
        ("title", ""),
        ("external_id", ""),
        ("status", ""),
        ("source_name", ""),
        ("url", ""),
        ("description", ""),
        ("content", ""),
        ("synced_at", ""),
        ("grade", 0),
        ("show_on_landing", 0),
    ):
        if not news.has_column(column):
            news.create_column_by_example(column, example)

    _index(db, "news", "idx_news_external_id", ["external_id"])
    _index(db, "news", "idx_news_synced_at", ["synced_at"])
    _index(db, "news", "idx_news_status", ["status"])

    news_images = get_table("news_images")
    if not news_images.has_column("news_uid"):
        news_images.create_column_by_example("news_uid", "")
    if not news_images.has_column("url"):
        news_images.create_column_by_example("url", "")

    news_sync = get_table("news_sync")
    if not news_sync.has_column("external_id"):
        news_sync.create_column_by_example("external_id", "")

    _index(db, "news_images", "idx_news_images_news_uid", ["news_uid"])
    _index(db, "news_sync", "idx_news_sync_external_id", ["external_id"])

    bug_tickets = get_table("bug_tickets")
    for column, example in (
        ("uid", ""),
        ("gitea_number", 0),
        ("author_uid", ""),
        ("original_title", ""),
        ("original_description", ""),
        ("enhanced_title", ""),
        ("html_url", ""),
        ("last_status", ""),
        ("last_comment_count", 0),
        ("created_at", ""),
        ("updated_at", ""),
        ("deleted_at", ""),
        ("deleted_by", ""),
    ):
        if not bug_tickets.has_column(column):
            bug_tickets.create_column_by_example(column, example)

    bug_comment_authors = get_table("bug_comment_authors")
    for column, example in (
        ("uid", ""),
        ("gitea_comment_id", 0),
        ("gitea_number", 0),
        ("author_uid", ""),
        ("created_at", ""),
        ("deleted_at", ""),
        ("deleted_by", ""),
    ):
        if not bug_comment_authors.has_column(column):
            bug_comment_authors.create_column_by_example(column, example)

    _index(db, "bug_tickets", "idx_bug_tickets_number", ["gitea_number"])
    _index(db, "bug_tickets", "idx_bug_tickets_author", ["author_uid"])
    _index(
        db,
        "bug_comment_authors",
        "idx_bug_comment_authors_comment",
        ["gitea_comment_id"],
    )
    _index(
        db, "bug_comment_authors", "idx_bug_comment_authors_number", ["gitea_number"]
    )
    _index(db, "service_state", "idx_service_state_name", ["name"])
    if "devii_conversations" in db.tables:
        conversations = get_table("devii_conversations")
        if not conversations.has_column("channel"):
            conversations.create_column_by_example("channel", "main")
        try:
            db.query(
                "UPDATE devii_conversations SET channel='main' WHERE channel IS NULL"
            )
        except Exception as e:  # noqa: BLE001
            logger.warning(f"Could not backfill devii_conversations.channel: {e}")
    _index(
        db,
        "devii_conversations",
        "idx_devii_conv_owner",
        ["owner_kind", "owner_id", "channel"],
    )
    _index(
        db,
        "devii_usage_ledger",
        "idx_devii_ledger_owner_time",
        ["owner_kind", "owner_id", "created_at"],
    )
    _index(
        db,
        "devii_turns",
        "idx_devii_turns_owner_time",
        ["owner_kind", "owner_id", "started_at"],
    )
    _index(db, "devii_tasks", "idx_devii_tasks_owner", ["owner_kind", "owner_id"])
    _index(
        db, "devii_tasks", "idx_devii_tasks_due", ["enabled", "status", "next_run_at"]
    )
    _index(db, "devii_lessons", "idx_devii_lessons_owner", ["owner_kind", "owner_id"])
    _index(
        db, "devii_virtual_tools", "idx_devii_vtools_owner", ["owner_kind", "owner_id"]
    )
    _index(
        db,
        "devii_virtual_tools",
        "idx_devii_vtools_name",
        ["owner_kind", "owner_id", "name"],
    )
    _index(db, "devii_behavior", "idx_devii_behavior_owner", ["owner_kind", "owner_id"])
    _index(db, "user_customizations", "idx_user_cust_owner", ["owner_kind", "owner_id"])
    _index(
        db,
        "user_customizations",
        "idx_user_cust_scope",
        ["owner_kind", "owner_id", "scope", "lang"],
    )
    _index(db, "gateway_usage_ledger", "idx_gw_usage_time", ["created_at"])
    _index(
        db,
        "gateway_usage_ledger",
        "idx_gw_usage_backend_time",
        ["backend", "created_at"],
    )
    _index(
        db, "gateway_usage_ledger", "idx_gw_usage_model_time", ["model", "created_at"]
    )
    _index(
        db,
        "gateway_usage_ledger",
        "idx_gw_usage_owner_time",
        ["owner_kind", "owner_id", "created_at"],
    )
    _index(
        db,
        "gateway_usage_ledger",
        "idx_gw_usage_endpoint_time",
        ["endpoint", "created_at"],
    )
    _index(db, "gateway_concurrency_samples", "idx_gw_conc_time", ["created_at"])
    _index(db, "jobs", "idx_jobs_kind_status", ["kind", "status"])
    _index(db, "jobs", "idx_jobs_owner", ["owner_kind", "owner_id"])
    _index(db, "jobs", "idx_jobs_expires", ["expires_at"])
    _index(db, "project_forks", "idx_project_forks_source", ["source_project_uid"])
    _index(db, "project_forks", "idx_project_forks_forked", ["forked_project_uid"])
    _index(db, "instances", "idx_instances_project", ["project_uid"])
    _index(db, "instances", "idx_instances_state", ["desired_state", "status"])
    _index(db, "instances", "idx_instances_container", ["container_id"])
    _index(db, "instances", "idx_instances_ingress", ["ingress_slug"])
    _index(
        db,
        "instance_events",
        "idx_instance_events_instance",
        ["instance_uid", "created_at"],
    )
    _index(
        db,
        "instance_metrics",
        "idx_instance_metrics_instance_ts",
        ["instance_uid", "ts"],
    )
    _index(
        db,
        "instance_schedules",
        "idx_instance_schedules_due",
        ["enabled", "next_run_at"],
    )

    from devplacepy.services.audit import store as audit_store

    audit_store.ensure_tables()
    _index(db, "audit_log", "idx_audit_created_at", ["created_at"])
    _index(db, "audit_log", "idx_audit_event_key", ["event_key"])
    _index(db, "audit_log", "idx_audit_category", ["category"])
    _index(db, "audit_log", "idx_audit_actor_uid", ["actor_uid"])
    _index(db, "audit_log", "idx_audit_actor_role", ["actor_role"])
    _index(db, "audit_log", "idx_audit_origin", ["origin"])
    _index(db, "audit_log", "idx_audit_target", ["target_type", "target_uid"])
    _index(db, "audit_log", "idx_audit_result", ["result"])
    _index(db, "audit_log_links", "idx_audit_links_audit", ["audit_uid"])
    _index(db, "audit_log_links", "idx_audit_links_object", ["object_type", "object_uid"])
    _index(db, "audit_log_links", "idx_audit_links_relation", ["relation"])

    notification_preferences = get_table("notification_preferences")
    for column, example in (
        ("uid", ""),
        ("user_uid", ""),
        ("notification_type", ""),
        ("in_app_enabled", 1),
        ("push_enabled", 1),
        ("created_at", ""),
        ("updated_at", ""),
        ("deleted_at", ""),
        ("deleted_by", ""),
    ):
        if not notification_preferences.has_column(column):
            notification_preferences.create_column_by_example(column, example)
    _index(db, "notification_preferences", "idx_notif_prefs_user", ["user_uid"])
    _index(
        db,
        "notification_preferences",
        "idx_notif_prefs_user_type",
        ["user_uid", "notification_type"],
    )

    deepsearch_sessions = get_table("deepsearch_sessions")
    for column, example in (
        ("uid", ""),
        ("owner_kind", ""),
        ("owner_id", ""),
        ("query", ""),
        ("status", ""),
        ("depth", 0),
        ("max_pages", 0),
        ("score", 0),
        ("confidence", 0.0),
        ("source_diversity", 0.0),
        ("page_count", 0),
        ("chunk_count", 0),
        ("collection", ""),
        ("summary", ""),
        ("created_at", ""),
        ("completed_at", ""),
        ("deleted_at", ""),
        ("deleted_by", ""),
    ):
        if not deepsearch_sessions.has_column(column):
            deepsearch_sessions.create_column_by_example(column, example)
    _index(db, "deepsearch_sessions", "idx_deepsearch_sessions_owner", ["owner_kind", "owner_id"])
    _index(db, "deepsearch_sessions", "idx_deepsearch_sessions_status", ["status"])
    _index(db, "deepsearch_sessions", "idx_deepsearch_sessions_created", ["created_at"])

    deepsearch_messages = get_table("deepsearch_messages")
    for column, example in (
        ("uid", ""),
        ("session_uid", ""),
        ("role", ""),
        ("content", ""),
        ("citations", ""),
        ("created_at", ""),
        ("deleted_at", ""),
        ("deleted_by", ""),
    ):
        if not deepsearch_messages.has_column(column):
            deepsearch_messages.create_column_by_example(column, example)
    _index(db, "deepsearch_messages", "idx_deepsearch_messages_session", ["session_uid"])

    deepsearch_url_cache = get_table("deepsearch_url_cache")
    for column, example in (
        ("uid", ""),
        ("url_hash", ""),
        ("url", ""),
        ("title", ""),
        ("content_hash", ""),
        ("status", 0),
        ("byte_size", 0),
        ("fetched_at", ""),
    ):
        if not deepsearch_url_cache.has_column(column):
            deepsearch_url_cache.create_column_by_example(column, example)
    _index(db, "deepsearch_url_cache", "idx_deepsearch_url_cache_hash", ["url_hash"])

    for table in SOFT_DELETE_TABLES:
        ensure_soft_delete_columns(table)

    if "news" in tables:
        for article in db["news"].find(status=None):
            was_featured = article.get("featured", 0)
            db["news"].update(
                {
                    "uid": article["uid"],
                    "status": "published" if was_featured else "draft",
                },
                ["uid"],
            )
        for article in db["news"].find(show_on_landing=None):
            db["news"].update(
                {
                    "uid": article["uid"],
                    "show_on_landing": 0,
                },
                ["uid"],
            )
        for article in db["news"].find(slug=None):
            from devplacepy.utils import make_combined_slug

            slug = make_combined_slug(
                article.get("title", "") or "news", article["uid"]
            )
            db["news"].update(
                {
                    "uid": article["uid"],
                    "slug": slug,
                },
                ["uid"],
            )

    if "news_sync" in tables:
        for entry in db["news_sync"].find():
            current = entry.get("status", "")
            if current in ("below_threshold", ""):
                db["news_sync"].update(
                    {
                        "id": entry["id"],
                        "status": "graded",
                    },
                    ["id"],
                )

    if "site_settings" in tables:
        news_defaults = {
            "news_grade_threshold": "7",
            "news_api_url": "https://news.app.molodetz.nl/api",
            "news_ai_url": INTERNAL_GATEWAY_URL,
            "news_ai_model": "molodetz",
        }
        for key, value in news_defaults.items():
            existing = db["site_settings"].find_one(key=key)
            if not existing:
                db["site_settings"].insert(
                    {"uid": f"default_{key}", "key": key, "value": value}
                )

        upload_defaults = {
            "max_upload_size_mb": "10",
            "allowed_file_types": "",
            "max_attachments_per_resource": "10",
        }
        for key, value in upload_defaults.items():
            existing = db["site_settings"].find_one(key=key)
            if not existing:
                db["site_settings"].insert(
                    {"uid": f"default_{key}", "key": key, "value": value}
                )

        operational_defaults = {
            "rate_limit_per_minute": "60",
            "rate_limit_window_seconds": "60",
            "news_service_interval": "3600",
            "session_max_age_days": "7",
            "session_remember_days": "30",
            "registration_open": "1",
            "maintenance_mode": "0",
            "maintenance_message": "DevPlace is undergoing scheduled maintenance. Please check back shortly.",
            "customization_enabled": "1",
            "customization_js_enabled": "1",
            "audit_log_retention_days": "90",
            "docs_search_mode": "agent",
        }
        for key, value in operational_defaults.items():
            existing = db["site_settings"].find_one(key=key)
            if not existing:
                db["site_settings"].insert(
                    {"uid": f"default_{key}", "key": key, "value": value}
                )

    _backfill_gamification()
    backfill_api_keys()
    migrate_ai_gateway_settings()

    logger.info("Database initialized")


def internal_gateway_key() -> str:
    return get_setting("gateway_internal_key", "")


OLD_GATEWAY_URL = "https://openai.app.molodetz.nl/v1/chat/completions"


def migrate_ai_gateway_settings() -> None:
    import os
    import uuid

    if not get_setting("gateway_internal_key", ""):
        set_setting("gateway_internal_key", str(uuid.uuid4()))
        logger.info("Generated internal gateway key")
    deepseek = os.environ.get("DEEPSEEK_API_KEY", "")
    openrouter = os.environ.get("OPENROUTER_API_KEY", "")
    if not get_setting("gateway_api_key", "") and (deepseek or openrouter):
        set_setting("gateway_api_key", deepseek or openrouter)
        logger.info("Migrated gateway upstream key from environment")
    if not get_setting("gateway_vision_key", "") and openrouter:
        set_setting("gateway_vision_key", openrouter)
        logger.info("Migrated gateway vision key from environment")
    for key in ("news_ai_url", "bot_api_url", "devii_ai_url"):
        if get_setting(key, "") == OLD_GATEWAY_URL:
            set_setting(key, INTERNAL_GATEWAY_URL)
            logger.info(f"Migrated {key} to the internal gateway")
    if get_setting("bot_model", "") == "deepseek-chat":
        set_setting("bot_model", "molodetz")
        logger.info("Migrated bot_model to molodetz")


def backfill_api_keys() -> int:
    if "users" not in db.tables:
        return 0
    users = db["users"]
    if not users.has_column("api_key"):
        users.create_column_by_example("api_key", "")
    if not users.has_column("cust_disable_global"):
        users.create_column_by_example("cust_disable_global", 0)
    if not users.has_column("cust_disable_pagetype"):
        users.create_column_by_example("cust_disable_pagetype", 0)
    import uuid_utils

    updated = 0
    for user in users.find():
        if not user.get("api_key"):
            users.update(
                {"uid": user["uid"], "api_key": str(uuid_utils.uuid7())}, ["uid"]
            )
            updated += 1
    return updated


def _backfill_gamification():
    if "users" not in db.tables:
        return
    from devplacepy.utils import (
        level_for_xp,
        check_milestone_badges,
        XP_POST,
        XP_COMMENT,
        XP_PROJECT,
        XP_GIST,
        XP_UPVOTE,
        XP_FOLLOW,
    )

    pending = list(db["users"].find(xp=0))
    if not pending:
        return

    xp_by_user = defaultdict(int)

    def add_counts(table, column, points):
        if table not in db.tables:
            return
        for row in db.query(
            f"SELECT {column} AS uid, COUNT(*) AS c FROM {table} GROUP BY {column}"
        ):
            if row["uid"]:
                xp_by_user[row["uid"]] += row["c"] * points

    add_counts("posts", "user_uid", XP_POST)
    add_counts("comments", "user_uid", XP_COMMENT)
    add_counts("projects", "user_uid", XP_PROJECT)
    add_counts("gists", "user_uid", XP_GIST)
    add_counts("follows", "following_uid", XP_FOLLOW)

    if "votes" in db.tables:
        for content_table, target_type in (
            ("posts", "post"),
            ("projects", "project"),
            ("gists", "gist"),
            ("comments", "comment"),
        ):
            if content_table not in db.tables:
                continue
            rows = db.query(
                f"SELECT c.user_uid AS uid, COUNT(*) AS c "
                f"FROM votes v JOIN {content_table} c ON v.target_uid = c.uid "
                f"WHERE v.target_type = :t AND v.value = 1 GROUP BY c.user_uid",
                t=target_type,
            )
            for row in rows:
                if row["uid"]:
                    xp_by_user[row["uid"]] += row["c"] * XP_UPVOTE

    for user in pending:
        xp = xp_by_user.get(user["uid"], 0)
        if xp <= 0:
            continue
        db["users"].update(
            {"uid": user["uid"], "xp": xp, "level": level_for_xp(xp)}, ["uid"]
        )

    _authors_cache.clear()
    for user in pending:
        check_milestone_badges(user["uid"])
    logger.info(f"Gamification backfill processed {len(pending)} users")


def get_table(name):
    return db[name]


def _in_clause(uids, prefix="p"):
    placeholders = ", ".join(f":{prefix}{i}" for i in range(len(uids)))
    params = {f"{prefix}{i}": uid for i, uid in enumerate(uids)}
    return placeholders, params


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def soft_delete(table_name, deleted_by, *, stamp=None, **criteria):
    if table_name not in db.tables:
        return 0
    table = db[table_name]
    if not table.has_column("deleted_at"):
        return 0
    rows = list(table.find(deleted_at=None, **criteria))
    if not rows:
        return 0
    stamp = stamp or _now_iso()
    for row in rows:
        table.update(
            {"id": row["id"], "deleted_at": stamp, "deleted_by": deleted_by}, ["id"]
        )
    return len(rows)


def soft_delete_in(table_name, column, uids, deleted_by, *, stamp=None, **extra):
    uids = [uid for uid in (uids or []) if uid]
    if not uids or table_name not in db.tables:
        return 0
    if not db[table_name].has_column("deleted_at"):
        return 0
    placeholders, params = _in_clause(uids)
    params["dat"] = stamp or _now_iso()
    params["dby"] = deleted_by
    extra_sql = ""
    for index, (key, value) in enumerate(extra.items()):
        params[f"x{index}"] = value
        extra_sql += f" AND {key} = :x{index}"
    db.query(
        f"UPDATE {table_name} SET deleted_at = :dat, deleted_by = :dby "
        f"WHERE {column} IN ({placeholders}) AND deleted_at IS NULL{extra_sql}",
        **params,
    )
    return len(uids)


def restore(table_name, **criteria):
    if table_name not in db.tables:
        return 0
    table = db[table_name]
    if not table.has_column("deleted_at"):
        return 0
    rows = [row for row in table.find(**criteria) if row.get("deleted_at")]
    for row in rows:
        table.update({"id": row["id"], "deleted_at": None, "deleted_by": None}, ["id"])
    return len(rows)


def purge(table_name, **criteria):
    if table_name not in db.tables:
        return 0
    table = db[table_name]
    count = table.count(**criteria)
    table.delete(**criteria)
    return int(count)


def list_deleted(table_name, page=1, per_page=25):
    if table_name not in db.tables or not db[table_name].has_column("deleted_at"):
        return [], build_pagination(page, 0, per_page)
    table = db[table_name]
    column = table.table.columns.deleted_at
    total = table.count(column.isnot(None))
    pagination = build_pagination(page, total, per_page)
    offset = (pagination["page"] - 1) * pagination["per_page"]
    rows = list(
        table.find(
            column.isnot(None),
            order_by=["-deleted_at"],
            _limit=pagination["per_page"],
            _offset=offset,
        )
    )
    return rows, pagination


def count_deleted(table_name):
    if table_name not in db.tables or not db[table_name].has_column("deleted_at"):
        return 0
    table = db[table_name]
    return int(table.count(table.table.columns.deleted_at.isnot(None)))


def restore_event(stamp):
    if not stamp:
        return 0
    restored = 0
    for table_name in SOFT_DELETE_TABLES:
        if table_name in db.tables and db[table_name].has_column("deleted_at"):
            restored += int(
                db.query(
                    f"SELECT COUNT(*) AS n FROM {table_name} WHERE deleted_at = :s",
                    s=stamp,
                ).__next__()["n"]
            )
            db.query(
                f"UPDATE {table_name} SET deleted_at = NULL, deleted_by = NULL "
                f"WHERE deleted_at = :s",
                s=stamp,
            )
    return restored


def purge_event(stamp):
    if not stamp:
        return []
    purged = []
    for table_name in SOFT_DELETE_TABLES:
        if table_name in db.tables and db[table_name].has_column("deleted_at"):
            rows = list(
                db.query(
                    f"SELECT * FROM {table_name} WHERE deleted_at = :s", s=stamp
                )
            )
            if rows:
                purged.append((table_name, rows))
                db.query(
                    f"DELETE FROM {table_name} WHERE deleted_at = :s", s=stamp
                )
    return purged


def get_users_by_uids(uids):
    if not uids or "users" not in db.tables:
        return {}
    users = db["users"]
    if "uid" not in users.columns:
        return {}
    seen = set()
    unique = [u for u in uids if u not in seen and not seen.add(u)]
    return {u["uid"]: u for u in users.find(users.table.columns.uid.in_(unique))}


def get_comment_counts_by_post_uids(post_uids):
    if not post_uids or "comments" not in db.tables:
        return {}
    placeholders, params = _in_clause(post_uids)
    rows = db.query(
        f"SELECT target_uid, COUNT(*) as c FROM comments WHERE target_type='post' AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid",
        **params,
    )
    return {r["target_uid"]: r["c"] for r in rows}


def get_post_counts_by_user_uids(user_uids):
    if not user_uids or "posts" not in db.tables:
        return {}
    placeholders, params = _in_clause(user_uids)
    rows = db.query(
        f"SELECT user_uid, COUNT(*) as c FROM posts WHERE user_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY user_uid",
        **params,
    )
    return {r["user_uid"]: r["c"] for r in rows}


def get_vote_counts(target_uids):
    if not target_uids or "votes" not in db.tables:
        return {}, {}
    placeholders, params = _in_clause(target_uids)
    rows = db.query(
        f"SELECT target_uid, value, COUNT(*) as c FROM votes WHERE target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, value",
        **params,
    )
    ups = {}
    downs = {}
    for r in rows:
        if r["value"] == 1:
            ups[r["target_uid"]] = r["c"]
        else:
            downs[r["target_uid"]] = r["c"]
    return ups, downs


def get_user_votes(user_uid, target_uids):
    if not user_uid or not target_uids or "votes" not in db.tables:
        return {}
    placeholders, params = _in_clause(target_uids)
    params["uid"] = user_uid
    rows = db.query(
        f"SELECT target_uid, value FROM votes WHERE user_uid = :uid AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
        **params,
    )
    return {r["target_uid"]: r["value"] for r in rows}


def get_reactions_by_targets(target_type, target_uids, user=None):
    if not target_uids or "reactions" not in db.tables:
        return {}
    placeholders, params = _in_clause(target_uids)
    params["tt"] = target_type
    rows = db.query(
        f"SELECT target_uid, emoji, COUNT(*) as c FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, emoji",
        **params,
    )
    counts = defaultdict(dict)
    for row in rows:
        counts[row["target_uid"]][row["emoji"]] = row["c"]
    mine = defaultdict(list)
    if user:
        placeholders, params = _in_clause(target_uids, prefix="m")
        params["tt"] = target_type
        params["u"] = user["uid"]
        for row in db.query(
            f"SELECT target_uid, emoji FROM reactions WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
            **params,
        ):
            mine[row["target_uid"]].append(row["emoji"])
    result = {}
    for uid in target_uids:
        result[uid] = {
            "counts": dict(counts.get(uid, {})),
            "mine": list(mine.get(uid, [])),
        }
    return result


def get_user_bookmarks(user_uid, target_type, target_uids):
    if not user_uid or not target_uids or "bookmarks" not in db.tables:
        return set()
    placeholders, params = _in_clause(target_uids)
    params["u"] = user_uid
    params["tt"] = target_type
    rows = db.query(
        f"SELECT target_uid FROM bookmarks WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
        **params,
    )
    return {row["target_uid"] for row in rows}


def get_polls_by_post_uids(post_uids, user=None):
    if not post_uids or "polls" not in db.tables:
        return {}
    placeholders, params = _in_clause(post_uids)
    polls = list(
        db.query(
            f"SELECT * FROM polls WHERE post_uid IN ({placeholders}) AND deleted_at IS NULL",
            **params,
        )
    )
    if not polls:
        return {}
    poll_uids = [poll["uid"] for poll in polls]
    option_placeholders, option_params = _in_clause(poll_uids, prefix="o")
    options = list(
        db.query(
            f"SELECT * FROM poll_options WHERE poll_uid IN ({option_placeholders}) AND deleted_at IS NULL ORDER BY position",
            **option_params,
        )
    )
    counts = defaultdict(dict)
    totals = defaultdict(int)
    if "poll_votes" in db.tables:
        vote_placeholders, vote_params = _in_clause(poll_uids, prefix="v")
        for row in db.query(
            f"SELECT poll_uid, option_uid, COUNT(*) as c FROM poll_votes WHERE poll_uid IN ({vote_placeholders}) AND deleted_at IS NULL GROUP BY poll_uid, option_uid",
            **vote_params,
        ):
            counts[row["poll_uid"]][row["option_uid"]] = row["c"]
            totals[row["poll_uid"]] += row["c"]
    user_choice = {}
    if user and "poll_votes" in db.tables:
        placeholders, params = _in_clause(poll_uids, prefix="m")
        params["u"] = user["uid"]
        for row in db.query(
            f"SELECT poll_uid, option_uid FROM poll_votes WHERE user_uid=:u AND poll_uid IN ({placeholders}) AND deleted_at IS NULL",
            **params,
        ):
            user_choice[row["poll_uid"]] = row["option_uid"]
    options_by_poll = defaultdict(list)
    for option in options:
        options_by_poll[option["poll_uid"]].append(option)
    result = {}
    for poll in polls:
        poll_uid = poll["uid"]
        total = totals.get(poll_uid, 0)
        rendered = []
        for option in options_by_poll.get(poll_uid, []):
            count = counts.get(poll_uid, {}).get(option["uid"], 0)
            rendered.append(
                {
                    "uid": option["uid"],
                    "label": option["label"],
                    "count": count,
                    "pct": round(count * 100 / total) if total else 0,
                }
            )
        result[poll["post_uid"]] = {
            "uid": poll_uid,
            "question": poll["question"],
            "options": rendered,
            "total": total,
            "my_choice": user_choice.get(poll_uid),
        }
    return result


def get_poll_for_post(post_uid, user=None):
    return get_polls_by_post_uids([post_uid], user).get(post_uid)


def _build_comment_items(raw, user=None):
    uids = [c["user_uid"] for c in raw]
    cids = [c["uid"] for c in raw]
    users = get_users_by_uids(uids)
    ups, downs = get_vote_counts(cids)
    user_votes = get_user_votes(user["uid"], cids) if user else {}
    reactions = get_reactions_by_targets("comment", cids, user)
    from devplacepy.utils import time_ago
    from devplacepy.attachments import get_attachments_batch as _gab

    atts_map = _gab("comment", cids) if "attachments" in db.tables else {}
    items = {}
    for c in raw:
        items[c["uid"]] = {
            "comment": c,
            "author": users.get(c["user_uid"]),
            "time_ago": time_ago(c["created_at"]),
            "votes": {"up": ups.get(c["uid"], 0), "down": downs.get(c["uid"], 0)},
            "my_vote": user_votes.get(c["uid"], 0),
            "children": [],
            "attachments": atts_map.get(c["uid"], []),
            "reactions": reactions.get(c["uid"], {"counts": {}, "mine": []}),
        }
    return items


def load_comments(target_type, target_uid, user=None):
    if "comments" not in db.tables:
        return []
    comments_table = db["comments"]
    raw = list(
        comments_table.find(
            target_type=target_type,
            target_uid=target_uid,
            deleted_at=None,
            order_by=["created_at"],
        )
    )
    if not raw and target_type == "post":
        raw = list(
            comments_table.find(
                post_uid=target_uid, deleted_at=None, order_by=["created_at"]
            )
        )
    if not raw:
        return []
    cmap = _build_comment_items(raw, user)
    top = []
    for item in cmap.values():
        parent = item["comment"].get("parent_uid")
        if parent and parent in cmap:
            cmap[parent]["children"].append(item)
        else:
            top.append(item)
    return top

def get_recent_comments_by_post_uids(post_uids, limit=3, user=None):
    if not post_uids or "comments" not in db.tables:
        return {}
    placeholders, params = _in_clause(post_uids)
    params["lim"] = limit
    raw = list(
        db.query(
            f"SELECT * FROM ("
            f"  SELECT *, ROW_NUMBER() OVER ("
            f"    PARTITION BY target_uid ORDER BY created_at DESC, id DESC"
            f"  ) AS rn FROM comments"
            f"  WHERE target_type='post' AND target_uid IN ({placeholders}) AND deleted_at IS NULL"
            f") WHERE rn <= :lim ORDER BY target_uid, created_at ASC",
            **params,
        )
    )
    if not raw:
        return {}
    items = _build_comment_items(raw, user)
    result = defaultdict(list)
    for c in raw:
        result[c["target_uid"]].append(items[c["uid"]])
    return dict(result)


def load_comments_by_target_uids(target_type, target_uids, user=None):
    if not target_uids or "comments" not in db.tables:
        return {}
    placeholders, params = _in_clause(target_uids)
    params["tt"] = target_type
    raw = list(
        db.query(
            f"SELECT * FROM comments WHERE target_type=:tt AND target_uid IN ({placeholders}) ORDER BY created_at",
            **params,
        )
    )
    if not raw:
        return {}
    from collections import defaultdict
    by_uid = defaultdict(list)
    for c in raw:
        by_uid[c["target_uid"]].append(c)
    result = {}
    for uid in target_uids:
        group = by_uid.get(uid, [])
        if not group:
            result[uid] = []
            continue
        cmap = _build_comment_items(group, user)
        tree = []
        for item in cmap.values():
            parent = item["comment"].get("parent_uid")
            if parent and parent in cmap:
                cmap[parent]["children"].append(item)
            else:
                tree.append(item)
        result[uid] = tree
    return result


def get_attachments(resource_type: str, resource_uid: str) -> list:
    if "attachments" not in db.tables:
        return []
    return list(
        db["attachments"].find(
            resource_type=resource_type,
            resource_uid=resource_uid,
            deleted_at=None,
            order_by=["created_at"],
        )
    )


def get_attachments_by_type(resource_type: str, resource_uids: list) -> dict:
    if not resource_uids or "attachments" not in db.tables:
        return {}
    rows = list(
        db["attachments"].find(
            db["attachments"].table.columns.resource_uid.in_(resource_uids),
            db["attachments"].table.columns.deleted_at.is_(None),
            resource_type=resource_type,
        )
    )
    result = {}
    for a in rows:
        key = a["resource_uid"]
        if key not in result:
            result[key] = []
        result[key].append(a)
    return result


def get_news_images_by_uids(news_uids: list) -> dict:
    if not news_uids or "news_images" not in db.tables:
        return {}
    images_table = db["news_images"]
    if not images_table.has_column("news_uid"):
        return {}
    rows = images_table.find(
        images_table.table.columns.news_uid.in_(news_uids),
        images_table.table.columns.deleted_at.is_(None),
        order_by=["uid"],
    )
    result = {}
    for r in rows:
        result.setdefault(r["news_uid"], r["url"])
    return result


def delete_attachment_record(uid: str) -> None:
    if "attachments" not in db.tables:
        return
    att = db["attachments"].find_one(uid=uid)
    if att:
        _delete_attachment_file(att.get("storage_path", ""))
        db["attachments"].delete(id=att["id"])


def delete_attachments(resource_type: str, resource_uid: str) -> None:
    if "attachments" not in db.tables:
        return
    for a in db["attachments"].find(
        resource_type=resource_type, resource_uid=resource_uid
    ):
        _delete_attachment_file(a.get("storage_path", ""))
    db["attachments"].delete(resource_type=resource_type, resource_uid=resource_uid)


def _delete_attachment_file(storage_path: str) -> None:
    if not storage_path:
        return
    from devplacepy.config import UPLOADS_DIR

    file_path = UPLOADS_DIR / storage_path
    try:
        file_path.unlink(missing_ok=True)
        parent = file_path.parent
        if parent.exists() and not any(parent.iterdir()):
            parent.rmdir()
        grandparent = parent.parent
        if grandparent.exists() and not any(grandparent.iterdir()):
            grandparent.rmdir()
    except Exception as e:
        logger.warning(f"Failed to delete attachment file {storage_path}: {e}")


_settings_cache = TTLCache(ttl=60, max_size=100)


def get_setting(key: str, default: str = "") -> str:
    sync_local_cache("settings", _settings_cache)
    cached = _settings_cache.get(key)
    if cached is not None:
        return cached
    if "site_settings" not in db.tables:
        return default
    entry = db["site_settings"].find_one(key=key)
    if entry is None:
        return default
    _settings_cache.set(key, entry["value"])
    return entry["value"]


def get_int_setting(key: str, default: int) -> int:
    raw = get_setting(key, str(default))
    try:
        return int(raw)
    except (TypeError, ValueError):
        return default


def set_setting(key: str, value: str) -> None:
    settings = get_table("site_settings")
    existing = settings.find_one(key=key)
    if existing:
        settings.update({"id": existing["id"], "key": key, "value": value}, ["id"])
    else:
        settings.insert({"uid": f"setting_{key}", "key": key, "value": value})
    _settings_cache.set(key, value)
    bump_cache_version("settings")


def clear_settings_cache() -> None:
    _settings_cache.clear()
    bump_cache_version("settings")


CUSTOMIZATION_GLOBAL_SCOPE = "global"
CUSTOMIZATION_LANGS = ("css", "js")

_customizations_cache = TTLCache(ttl=300, max_size=100)


def _customization_key(owner_kind: str, owner_id: str, page_type: str) -> str:
    return f"{owner_kind}\x1f{owner_id}\x1f{page_type}"


CUSTOMIZATION_PREF_COLUMNS = {
    "global": "cust_disable_global",
    "pagetype": "cust_disable_pagetype",
}


def get_customization_prefs(owner_kind: str, owner_id: str) -> dict:
    if owner_kind != "user" or "users" not in db.tables:
        return {"disable_global": False, "disable_pagetype": False}
    user = db["users"].find_one(uid=owner_id)
    if user is None:
        return {"disable_global": False, "disable_pagetype": False}
    return {
        "disable_global": bool(user.get("cust_disable_global", 0)),
        "disable_pagetype": bool(user.get("cust_disable_pagetype", 0)),
    }


def set_customization_pref(owner_id: str, category: str, disabled: bool) -> None:
    column = CUSTOMIZATION_PREF_COLUMNS.get(category)
    if column is None:
        raise ValueError(f"Unknown customization category: {category}")
    from devplacepy.utils import clear_user_cache

    get_table("users").update(
        {"uid": owner_id, column: 1 if disabled else 0}, ["uid"]
    )
    clear_user_cache(owner_id)
    bump_cache_version("customizations")


def get_custom_overrides(owner_kind: str, owner_id: str, page_type: str) -> dict:
    sync_local_cache("customizations", _customizations_cache)
    key = _customization_key(owner_kind, owner_id, page_type)
    cached = _customizations_cache.get(key)
    if cached is not None:
        return cached
    result = {"css": "", "js": ""}
    if "user_customizations" in db.tables:
        prefs = get_customization_prefs(owner_kind, owner_id)
        scopes = (CUSTOMIZATION_GLOBAL_SCOPE, page_type)
        rows = db["user_customizations"].find(
            owner_kind=owner_kind,
            owner_id=owner_id,
            enabled=1,
            deleted_at=None,
        )
        pieces: dict[str, dict[str, str]] = {lang: {} for lang in CUSTOMIZATION_LANGS}
        for row in rows:
            lang = row.get("lang")
            scope = row.get("scope")
            if lang not in pieces or scope not in scopes:
                continue
            if scope == CUSTOMIZATION_GLOBAL_SCOPE and prefs["disable_global"]:
                continue
            if scope != CUSTOMIZATION_GLOBAL_SCOPE and prefs["disable_pagetype"]:
                continue
            pieces[lang][scope] = row.get("code") or ""
        for lang in CUSTOMIZATION_LANGS:
            ordered = [pieces[lang][scope] for scope in scopes if scope in pieces[lang]]
            result[lang] = "\n".join(part for part in ordered if part.strip())
    _customizations_cache.set(key, result)
    return result


def get_custom_override(
    owner_kind: str, owner_id: str, scope: str, lang: str
) -> dict | None:
    if "user_customizations" not in db.tables:
        return None
    return db["user_customizations"].find_one(
        owner_kind=owner_kind,
        owner_id=owner_id,
        scope=scope,
        lang=lang,
        deleted_at=None,
    )


def list_custom_overrides(owner_kind: str, owner_id: str) -> list:
    if "user_customizations" not in db.tables:
        return []
    return list(
        db["user_customizations"].find(
            owner_kind=owner_kind, owner_id=owner_id, deleted_at=None
        )
    )


def set_custom_override(
    owner_kind: str, owner_id: str, scope: str, lang: str, code: str
) -> dict:
    from devplacepy.utils import generate_uid

    table = get_table("user_customizations")
    now = datetime.now(timezone.utc).isoformat()
    existing = table.find_one(
        owner_kind=owner_kind, owner_id=owner_id, scope=scope, lang=lang
    )
    if existing:
        record = {
            "id": existing["id"],
            "code": code,
            "enabled": 1,
            "updated_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
        table.update(record, ["id"])
        result = {**existing, **record}
    else:
        result = {
            "uid": generate_uid(),
            "owner_kind": owner_kind,
            "owner_id": owner_id,
            "scope": scope,
            "lang": lang,
            "code": code,
            "enabled": 1,
            "created_at": now,
            "updated_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
        table.insert(result)
    bump_cache_version("customizations")
    return result


def delete_custom_override(
    owner_kind: str,
    owner_id: str,
    scope: str | None = None,
    lang: str | None = None,
    deleted_by: str | None = None,
) -> int:
    if "user_customizations" not in db.tables:
        return 0
    criteria: dict = {"owner_kind": owner_kind, "owner_id": owner_id}
    if scope is not None:
        criteria["scope"] = scope
    if lang is not None:
        criteria["lang"] = lang
    count = soft_delete(
        "user_customizations", deleted_by or f"{owner_kind}:{owner_id}", **criteria
    )
    bump_cache_version("customizations")
    return int(count)


NOTIFICATION_TYPES = [
    {"key": "comment", "label": "Comments", "description": "Someone comments on your post"},
    {"key": "reply", "label": "Replies", "description": "Someone replies to your comment"},
    {"key": "mention", "label": "Mentions", "description": "Someone mentions you with @username"},
    {"key": "vote", "label": "Upvotes", "description": "Someone ++'d your content"},
    {"key": "follow", "label": "Followers", "description": "Someone starts following you"},
    {"key": "message", "label": "Direct messages", "description": "Someone sends you a message"},
    {"key": "badge", "label": "Badges", "description": "You earn a badge"},
    {"key": "level", "label": "Level-ups", "description": "You reach a new level"},
    {"key": "bug", "label": "Bug tracker", "description": "Updates on bug reports you filed"},
]

NOTIFICATION_CHANNELS = ("in_app", "push")
_NOTIFICATION_CHANNEL_COLUMNS = {"in_app": "in_app_enabled", "push": "push_enabled"}
_NOTIFICATION_TYPE_KEYS = {entry["key"] for entry in NOTIFICATION_TYPES}

_notification_prefs_cache = TTLCache(ttl=300, max_size=500)


def _notification_default(notification_type: str, channel: str) -> bool:
    return get_int_setting(f"notif_default_{notification_type}_{channel}", 1) != 0


def get_notification_default(notification_type: str, channel: str) -> bool:
    return _notification_default(notification_type, channel)


def set_notification_default(
    notification_type: str, channel: str, enabled: bool
) -> None:
    if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
        raise ValueError(f"Unknown notification channel: {channel}")
    set_setting(f"notif_default_{notification_type}_{channel}", "1" if enabled else "0")


def _notification_overrides(user_uid: str) -> dict:
    sync_local_cache("notif_prefs", _notification_prefs_cache)
    cached = _notification_prefs_cache.get(user_uid)
    if cached is not None:
        return cached
    overrides: dict = {}
    if "notification_preferences" in db.tables:
        for row in db["notification_preferences"].find(
            user_uid=user_uid, deleted_at=None
        ):
            overrides[row["notification_type"]] = {
                "in_app": bool(row.get("in_app_enabled", 1)),
                "push": bool(row.get("push_enabled", 1)),
            }
    _notification_prefs_cache.set(user_uid, overrides)
    return overrides


def notification_enabled(user_uid: str, notification_type: str, channel: str) -> bool:
    if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
        return True
    override = _notification_overrides(user_uid).get(notification_type)
    if override is not None:
        return bool(override[channel])
    return _notification_default(notification_type, channel)


def get_notification_prefs(user_uid: str) -> list:
    overrides = _notification_overrides(user_uid)
    result = []
    for entry in NOTIFICATION_TYPES:
        key = entry["key"]
        override = overrides.get(key)
        in_app = (
            bool(override["in_app"]) if override else _notification_default(key, "in_app")
        )
        push = bool(override["push"]) if override else _notification_default(key, "push")
        result.append(
            {
                "key": key,
                "label": entry["label"],
                "description": entry["description"],
                "in_app": in_app,
                "push": push,
                "customized": override is not None,
            }
        )
    return result


def set_notification_pref(
    user_uid: str, notification_type: str, channel: str, enabled: bool
) -> dict:
    if notification_type not in _NOTIFICATION_TYPE_KEYS:
        raise ValueError(f"Unknown notification type: {notification_type}")
    if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
        raise ValueError(f"Unknown notification channel: {channel}")
    from devplacepy.utils import generate_uid

    table = get_table("notification_preferences")
    now = datetime.now(timezone.utc).isoformat()
    existing = table.find_one(user_uid=user_uid, notification_type=notification_type)
    if existing:
        values = {
            "in_app": bool(existing.get("in_app_enabled", 1)),
            "push": bool(existing.get("push_enabled", 1)),
        }
    else:
        values = {
            "in_app": _notification_default(notification_type, "in_app"),
            "push": _notification_default(notification_type, "push"),
        }
    values[channel] = enabled
    if existing:
        record = {
            "id": existing["id"],
            "in_app_enabled": 1 if values["in_app"] else 0,
            "push_enabled": 1 if values["push"] else 0,
            "updated_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
        table.update(record, ["id"])
        result = {**existing, **record}
    else:
        result = {
            "uid": generate_uid(),
            "user_uid": user_uid,
            "notification_type": notification_type,
            "in_app_enabled": 1 if values["in_app"] else 0,
            "push_enabled": 1 if values["push"] else 0,
            "created_at": now,
            "updated_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
        table.insert(result)
    bump_cache_version("notif_prefs")
    return result


def reset_notification_prefs(user_uid: str, deleted_by: str | None = None) -> int:
    if "notification_preferences" not in db.tables:
        return 0
    count = soft_delete(
        "notification_preferences", deleted_by or f"user:{user_uid}", user_uid=user_uid
    )
    bump_cache_version("notif_prefs")
    return int(count)


_stats_cache = TTLCache(ttl=30, max_size=50)


def get_site_stats() -> dict:
    cached = _stats_cache.get("site")
    if cached is not None:
        return cached
    today_start = (
        datetime.now(timezone.utc)
        .replace(hour=0, minute=0, second=0, microsecond=0)
        .isoformat()
    )
    stats = {
        "total_members": db["users"].count() if "users" in db.tables else 0,
        "posts_today": db["posts"].count(
            created_at={">=": today_start}, deleted_at=None
        )
        if "posts" in db.tables
        else 0,
        "total_projects": db["projects"].count(deleted_at=None)
        if "projects" in db.tables
        else 0,
        "total_gists": db["gists"].count(deleted_at=None)
        if "gists" in db.tables
        else 0,
    }
    _stats_cache.set("site", stats)
    return stats


_analytics_cache = TTLCache(ttl=300, max_size=50)


def get_platform_analytics(top_n: int = 10) -> dict:
    top_n = max(1, min(int(top_n or 10), 50))
    cache_key = f"analytics:{top_n}"
    cached = _analytics_cache.get(cache_key)
    if cached is not None:
        return cached

    now = datetime.now(timezone.utc)
    d1 = (now - timedelta(days=1)).isoformat()
    d7 = (now - timedelta(days=7)).isoformat()
    d30 = (now - timedelta(days=30)).isoformat()

    active_24h = active_7d = active_30d = 0
    sources = [
        table
        for table in ("posts", "comments", "gists", "projects")
        if table in db.tables
    ]
    if sources:
        union = " UNION ALL ".join(
            f"SELECT user_uid, created_at FROM {table} WHERE deleted_at IS NULL"
            for table in sources
        )
        rows = db.query(
            "SELECT "
            "COUNT(DISTINCT CASE WHEN created_at >= :d1 THEN user_uid END) AS a1, "
            "COUNT(DISTINCT CASE WHEN created_at >= :d7 THEN user_uid END) AS a7, "
            "COUNT(DISTINCT CASE WHEN created_at >= :d30 THEN user_uid END) AS a30 "
            f"FROM ({union})",
            d1=d1,
            d7=d7,
            d30=d30,
        )
        for row in rows:
            active_24h = row["a1"] or 0
            active_7d = row["a7"] or 0
            active_30d = row["a30"] or 0

    signed_in_now = 0
    if "sessions" in db.tables:
        for row in db.query(
            "SELECT COUNT(DISTINCT user_uid) AS n FROM sessions WHERE expires_at > :now AND deleted_at IS NULL",
            now=now.isoformat(),
        ):
            signed_in_now = row["n"] or 0

    def _users_created_since(cutoff: str) -> int:
        return (
            db["users"].count(created_at={">=": cutoff}) if "users" in db.tables else 0
        )

    site = get_site_stats()
    totals = {
        "posts": db["posts"].count(deleted_at=None) if "posts" in db.tables else 0,
        "comments": db["comments"].count(deleted_at=None)
        if "comments" in db.tables
        else 0,
        "gists": site["total_gists"],
        "projects": site["total_projects"],
        "news": db["news"].count(deleted_at=None) if "news" in db.tables else 0,
    }
    top_authors = [
        {"username": author.get("username", ""), "stars": author.get("stars", 0)}
        for author in get_top_authors(top_n)
    ]

    result = {
        "total_members": site["total_members"],
        "active_24h": active_24h,
        "active_7d": active_7d,
        "active_30d": active_30d,
        "signed_in_now": signed_in_now,
        "new_24h": _users_created_since(d1),
        "new_7d": _users_created_since(d7),
        "new_30d": _users_created_since(d30),
        "posts_today": site["posts_today"],
        "totals": totals,
        "top_authors": top_authors,
        "active_definition": (
            "Active = created a post, comment, gist, or project within the window. "
            "Automated/bot accounts are not separately flagged."
        ),
        "signed_in_definition": (
            "signed_in_now = distinct members holding an unexpired session cookie, i.e. who "
            "logged in within the session lifetime (default 7 days, up to 30 with remember-me) "
            "and have not logged out. It is NOT a real-time presence/online count - the platform "
            "does not track per-request last-seen activity - so it is normally a large fraction "
            "of recently active members. Do not describe it as 'currently online' or 'logged in "
            "right now'."
        ),
    }
    _analytics_cache.set(cache_key, result)
    return result


_activity_cache = TTLCache(ttl=300, max_size=1000)

_ACTIVITY_TABLES = ("posts", "comments", "gists", "projects")


def get_activity_calendar(user_uid: str) -> dict:
    cached = _activity_cache.get(user_uid)
    if cached is not None:
        return cached
    sources = [table for table in _ACTIVITY_TABLES if table in db.tables]
    calendar: dict[str, int] = {}
    if sources:
        cutoff = (datetime.now(timezone.utc) - timedelta(days=364)).date().isoformat()
        union = " UNION ALL ".join(
            f"SELECT created_at FROM {table} WHERE user_uid = :u AND deleted_at IS NULL"
            for table in sources
        )
        rows = db.query(
            f"SELECT date(created_at) AS day, COUNT(*) AS c FROM ({union}) WHERE date(created_at) >= :cutoff GROUP BY day",
            u=user_uid,
            cutoff=cutoff,
        )
        for row in rows:
            if row["day"]:
                calendar[row["day"]] = row["c"]
    _activity_cache.set(user_uid, calendar)
    return calendar


def _activity_level(count: int) -> int:
    if count <= 0:
        return 0
    if count == 1:
        return 1
    if count <= 3:
        return 2
    if count <= 6:
        return 3
    return 4


def get_first_activity_date(user_uid: str):
    sources = [table for table in _ACTIVITY_TABLES if table in db.tables]
    if not sources:
        return None
    union = " UNION ALL ".join(
        f"SELECT MIN(created_at) AS m FROM {table} WHERE user_uid = :u AND deleted_at IS NULL"
        for table in sources
    )
    for row in db.query(f"SELECT MIN(m) AS first FROM ({union})", u=user_uid):
        if row["first"]:
            return datetime.fromisoformat(row["first"]).date()
    return None


HEATMAP_WEEKS = 53


def get_activity_heatmap(user_uid: str) -> list:
    calendar = get_activity_calendar(user_uid)
    today = datetime.now(timezone.utc).date()
    week_start = today - timedelta(days=today.weekday())
    start = week_start - timedelta(weeks=HEATMAP_WEEKS - 1)
    first = get_first_activity_date(user_uid)
    if first:
        first_week = first - timedelta(days=first.weekday())
        if first_week > start:
            start = first_week
    weeks = []
    for w in range(HEATMAP_WEEKS):
        week = []
        for d in range(7):
            day = start + timedelta(days=w * 7 + d)
            iso = day.isoformat()
            count = calendar.get(iso, 0)
            week.append({"date": iso, "count": count, "level": _activity_level(count)})
        weeks.append(week)
    return weeks


def get_activity_months(weeks: list) -> list:
    if not weeks:
        return []
    last = len(weeks) - 1
    labels = []
    for i in range(6):
        column = round(i * last / 5)
        iso = weeks[column][0]["date"]
        labels.append(datetime.fromisoformat(iso).strftime("%b"))
    return labels


def get_streaks(user_uid: str) -> dict:
    calendar = get_activity_calendar(user_uid)
    if not calendar:
        return {"current": 0, "longest": 0}
    dates = sorted(datetime.fromisoformat(day).date() for day in calendar)
    date_set = set(dates)
    longest = 1
    run = 1
    for index in range(1, len(dates)):
        if (dates[index] - dates[index - 1]).days == 1:
            run += 1
        else:
            run = 1
        longest = max(longest, run)
    today = datetime.now(timezone.utc).date()
    cursor = today
    if today not in date_set and (today - timedelta(days=1)) in date_set:
        cursor = today - timedelta(days=1)
    current = 0
    while cursor in date_set:
        current += 1
        cursor = cursor - timedelta(days=1)
    return {"current": current, "longest": longest}


_gist_languages_cache = TTLCache(ttl=60, max_size=200)


def get_gist_languages() -> set[str]:
    cached = _gist_languages_cache.get("codes")
    if cached is not None:
        return cached
    codes: set[str] = set()
    if "gists" in db.tables:
        for row in db.query(
            "SELECT DISTINCT language FROM gists WHERE deleted_at IS NULL"
        ):
            language = row.get("language")
            if language:
                codes.add(language)
    _gist_languages_cache.set("codes", codes)
    return codes


VOTABLE_TARGETS: dict[str, str] = {
    "post": "posts",
    "project": "projects",
    "gist": "gists",
    "comment": "comments",
}

STAR_TARGETS: set[str] = {"post", "project", "gist"}

_authors_cache = TTLCache(ttl=60, max_size=200)


def _ranked_authors() -> list:
    cached = _authors_cache.get("ranked")
    if cached is not None:
        return cached
    sources = [
        (target_type, table_name)
        for target_type, table_name in VOTABLE_TARGETS.items()
        if table_name in db.tables
    ]
    if "votes" not in db.tables or not sources:
        _authors_cache.set("ranked", [])
        return []
    target_union = " UNION ALL ".join(
        f"SELECT uid, user_uid, '{target_type}' AS target_type FROM {table_name} WHERE deleted_at IS NULL"
        for target_type, table_name in sources
    )
    rows = db.query(
        f"SELECT t.user_uid, SUM(v.value) AS total "
        f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
        f"WHERE v.deleted_at IS NULL "
        f"GROUP BY t.user_uid HAVING SUM(v.value) > 0 ORDER BY total DESC"
    )
    ranked = [(row["user_uid"], row["total"]) for row in rows]
    users_map = get_users_by_uids([uid for uid, _ in ranked])
    authors = []
    for uid, total in ranked:
        user = users_map.get(uid)
        if user:
            author = dict(user)
            author["stars"] = total
            authors.append(author)
    _authors_cache.set("ranked", authors)
    return authors


def get_top_authors(limit: int = 5) -> list:
    return _ranked_authors()[:limit]


def get_leaderboard(limit: int = 50, offset: int = 0) -> list:
    sliced = _ranked_authors()[offset : offset + limit]
    leaderboard = []
    for position, author in enumerate(sliced, start=offset + 1):
        entry = dict(author)
        entry["rank"] = position
        leaderboard.append(entry)
    return leaderboard


def get_user_rank(user_uid: str):
    for position, author in enumerate(_ranked_authors(), start=1):
        if author["uid"] == user_uid:
            return position
    return None


def get_user_stars(user_uid: str) -> int:
    if "votes" not in db.tables:
        return 0
    target_union = " UNION ALL ".join(
        f"SELECT uid, '{target_type}' AS target_type FROM {table_name} WHERE user_uid = :u AND deleted_at IS NULL"
        for target_type, table_name in VOTABLE_TARGETS.items()
        if table_name in db.tables
    )
    if not target_union:
        return 0
    for row in db.query(
        f"SELECT COALESCE(SUM(v.value), 0) AS s "
        f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
        f"WHERE v.deleted_at IS NULL",
        u=user_uid,
    ):
        return row["s"] or 0
    return 0


def resolve_by_slug(table, slug, include_deleted=False):
    has_soft_delete = table.has_column("deleted_at")
    flt = {} if include_deleted or not has_soft_delete else {"deleted_at": None}
    entry = table.find_one(slug=slug, **flt)
    if not entry:
        entry = table.find_one(uid=slug, **flt)
    return entry


def record_fork(
    source_project_uid: str, forked_project_uid: str, forked_by_uid: str
) -> None:
    from devplacepy.utils import generate_uid

    get_table("project_forks").insert(
        {
            "uid": generate_uid(),
            "source_project_uid": source_project_uid,
            "forked_project_uid": forked_project_uid,
            "forked_by_uid": forked_by_uid,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "deleted_at": None,
            "deleted_by": None,
        }
    )


def get_fork_parent(forked_project_uid: str) -> dict | None:
    if "project_forks" not in db.tables:
        return None
    relation = get_table("project_forks").find_one(
        forked_project_uid=forked_project_uid, deleted_at=None
    )
    if not relation:
        return None
    return get_table("projects").find_one(
        uid=relation["source_project_uid"], deleted_at=None
    )


def count_forks(source_project_uid: str) -> int:
    if "project_forks" not in db.tables:
        return 0
    return get_table("project_forks").count(
        source_project_uid=source_project_uid, deleted_at=None
    )


def soft_delete_fork_relations(project_uid: str, deleted_by: str) -> None:
    if "project_forks" not in db.tables:
        return
    stamp = _now_iso()
    soft_delete("project_forks", deleted_by, stamp=stamp, forked_project_uid=project_uid)
    soft_delete("project_forks", deleted_by, stamp=stamp, source_project_uid=project_uid)


def delete_fork_relations(project_uid: str) -> None:
    if "project_forks" not in db.tables:
        return
    forks = get_table("project_forks")
    forks.delete(forked_project_uid=project_uid)
    forks.delete(source_project_uid=project_uid)


def resolve_object_url(target_type: str, target_uid: str) -> str:
    if target_type == "post":
        post = resolve_by_slug(get_table("posts"), target_uid)
        return f"/posts/{post['slug'] or post['uid']}" if post else "/feed"
    if target_type == "project":
        project = resolve_by_slug(get_table("projects"), target_uid)
        return (
            f"/projects/{project['slug'] or project['uid']}" if project else "/projects"
        )
    if target_type == "news":
        article = resolve_by_slug(get_table("news"), target_uid)
        if article:
            return f"/news/{article.get('slug', '') or article['uid']}"
        return "/news"
    if target_type == "bug":
        return f"/bugs?highlight={target_uid}"
    if target_type == "gist":
        gist = resolve_by_slug(get_table("gists"), target_uid)
        return f"/gists/{gist['slug'] or gist['uid']}" if gist else "/gists"
    if target_type == "comment":
        comment = get_table("comments").find_one(uid=target_uid, deleted_at=None)
        if not comment:
            return "/feed"
        parent_url = resolve_object_url(
            comment.get("target_type", "post"),
            comment.get("target_uid") or comment.get("post_uid", ""),
        )
        return f"{parent_url}#comment-{target_uid}"
    return "/feed"


def update_target_stars(target_type: str, target_uid: str, net_stars: int) -> None:
    table_name = VOTABLE_TARGETS.get(target_type)
    if not table_name:
        return
    if target_type in STAR_TARGETS:
        get_table(table_name).update({"uid": target_uid, "stars": net_stars}, ["uid"])
    _authors_cache.clear()


def soft_delete_engagement(target_type: str, target_uids: list, deleted_by: str) -> None:
    uids = [uid for uid in (target_uids or []) if uid]
    if not uids:
        return
    stamp = _now_iso()
    soft_delete_in(
        "reactions", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
    )
    soft_delete_in(
        "bookmarks", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
    )
    if target_type == "post" and "polls" in db.tables:
        for uid in uids:
            for poll in db["polls"].find(post_uid=uid, deleted_at=None):
                soft_delete("poll_votes", deleted_by, stamp=stamp, poll_uid=poll["uid"])
                soft_delete("poll_options", deleted_by, stamp=stamp, poll_uid=poll["uid"])
            soft_delete("polls", deleted_by, stamp=stamp, post_uid=uid)


def delete_engagement(target_type: str, target_uids: list) -> None:
    uids = [uid for uid in (target_uids or []) if uid]
    if not uids:
        return
    if "reactions" in db.tables:
        placeholders, params = _in_clause(uids)
        params["tt"] = target_type
        db.query(
            f"DELETE FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders})",
            **params,
        )
    if "bookmarks" in db.tables:
        placeholders, params = _in_clause(uids)
        params["tt"] = target_type
        db.query(
            f"DELETE FROM bookmarks WHERE target_type=:tt AND target_uid IN ({placeholders})",
            **params,
        )
    if target_type == "post" and "polls" in db.tables:
        for uid in uids:
            for poll in db["polls"].find(post_uid=uid):
                if "poll_votes" in db.tables:
                    db["poll_votes"].delete(poll_uid=poll["uid"])
                if "poll_options" in db.tables:
                    db["poll_options"].delete(poll_uid=poll["uid"])
            db["polls"].delete(post_uid=uid)


def get_target_owner_uid(target_type: str, target_uid: str) -> str | None:
    table_name = VOTABLE_TARGETS.get(target_type)
    if not table_name:
        return None
    row = get_table(table_name).find_one(uid=target_uid, deleted_at=None)
    return row["user_uid"] if row else None


PAGE_SIZE = 25


def text_search_clause(table, search, fields=("title", "description")):
    if not search or not search.strip() or not table.exists:
        return None
    columns = table.table.columns
    like = f"%{search.strip()}%"
    matches = [columns[field].ilike(like) for field in fields if field in columns]
    return or_(*matches) if matches else None


def paginate(
    table, *clauses, before=None, order=None, cursor_field="created_at", **filters
):
    order = order or ["-" + cursor_field]
    clauses = list(clauses)
    if table.has_column("deleted_at") and "deleted_at" not in filters:
        clauses.append(table.table.columns.deleted_at.is_(None))
    if before:
        clauses.append(table.table.columns[cursor_field] < before)
    rows = list(table.find(*clauses, **filters, order_by=order, _limit=PAGE_SIZE + 1))
    has_more = len(rows) > PAGE_SIZE
    rows = rows[:PAGE_SIZE]
    next_cursor = rows[-1][cursor_field] if has_more and rows else None
    return rows, next_cursor


def diversify_by_author(rows, max_per_author=2, limit=None, uid_key="user_uid"):
    counts: dict = {}
    kept = []
    for row in rows:
        owner = row.get(uid_key)
        if counts.get(owner, 0) >= max_per_author:
            continue
        counts[owner] = counts.get(owner, 0) + 1
        kept.append(row)
        if limit is not None and len(kept) >= limit:
            break
    return kept


def paginate_diverse(
    table,
    *clauses,
    before=None,
    order=None,
    cursor_field="created_at",
    max_per_author=2,
    uid_key="user_uid",
    pool=200,
    **filters,
):
    order = order or ["-" + cursor_field]
    clauses = list(clauses)
    if table.has_column("deleted_at") and "deleted_at" not in filters:
        clauses.append(table.table.columns.deleted_at.is_(None))
    if before:
        clauses.append(table.table.columns[cursor_field] < before)
    rows = list(table.find(*clauses, **filters, order_by=order, _limit=pool + 1))
    exhausted = len(rows) <= pool
    rows = rows[:pool]
    kept = diversify_by_author(
        rows, max_per_author=max_per_author, limit=PAGE_SIZE, uid_key=uid_key
    )
    if not kept:
        return [], None
    last_cursor = kept[-1][cursor_field]
    older_exists = (not exhausted) or any(r[cursor_field] < last_cursor for r in rows)
    next_cursor = last_cursor if older_exists else None
    return kept, next_cursor


def get_user_post_count(user_uid: str) -> int:
    if "posts" not in db.tables:
        return 0
    return get_table("posts").count(user_uid=user_uid, deleted_at=None)


def build_pagination(page, total, per_page=25):
    total_pages = max(1, __import__("math").ceil(total / per_page))
    page = max(1, min(page, total_pages))
    return {
        "page": page,
        "per_page": per_page,
        "total": total,
        "total_pages": total_pages,
        "has_prev": page > 1,
        "has_next": page < total_pages,
        "prev_page": page - 1,
        "next_page": page + 1,
    }


def get_user_media(user_uid: str, page: int = 1, per_page: int = 24) -> tuple:
    if "attachments" not in db.tables:
        return [], build_pagination(page, 0, per_page)
    from devplacepy.attachments import _row_to_attachment

    total = list(
        db.query(
            "SELECT COUNT(*) AS n FROM attachments "
            "WHERE user_uid=:u AND target_type != '' AND deleted_at IS NULL",
            u=user_uid,
        )
    )[0]["n"]
    pagination = build_pagination(page, total, per_page)
    offset = (pagination["page"] - 1) * pagination["per_page"]
    rows = db.query(
        "SELECT * FROM attachments "
        "WHERE user_uid=:u AND target_type != '' AND deleted_at IS NULL "
        "ORDER BY created_at DESC LIMIT :limit OFFSET :offset",
        u=user_uid,
        limit=pagination["per_page"],
        offset=offset,
    )
    items = []
    for row in rows:
        item = _row_to_attachment(row)
        item["target_url"] = resolve_object_url(item["target_type"], item["target_uid"])
        items.append(item)
    return items, pagination


def get_deleted_media(page: int = 1, per_page: int = 24) -> tuple:
    if "attachments" not in db.tables:
        return [], build_pagination(page, 0, per_page)
    from devplacepy.attachments import _row_to_attachment

    total = list(
        db.query("SELECT COUNT(*) AS n FROM attachments WHERE deleted_at IS NOT NULL")
    )[0]["n"]
    pagination = build_pagination(page, total, per_page)
    offset = (pagination["page"] - 1) * pagination["per_page"]
    rows = db.query(
        "SELECT * FROM attachments WHERE deleted_at IS NOT NULL "
        "ORDER BY deleted_at DESC LIMIT :limit OFFSET :offset",
        limit=pagination["per_page"],
        offset=offset,
    )
    usernames = {u["uid"]: u["username"] for u in get_table("users").find()}
    items = []
    for row in rows:
        item = _row_to_attachment(row)
        item["target_url"] = resolve_object_url(item["target_type"], item["target_uid"])
        item["deleted_at"] = row.get("deleted_at", "")
        item["uploader"] = usernames.get(row.get("user_uid"), "unknown")
        items.append(item)
    return items, pagination


def get_follow_counts(user_uid: str) -> dict:
    if "follows" not in db.tables:
        return {"followers": 0, "following": 0}
    follows = get_table("follows")
    return {
        "followers": follows.count(following_uid=user_uid, deleted_at=None),
        "following": follows.count(follower_uid=user_uid, deleted_at=None),
    }


def get_follow_list(
    user_uid: str, mode: str, page: int = 1, per_page: int = 25
) -> tuple:
    if "follows" not in db.tables:
        return [], build_pagination(page, 0, per_page)
    follows = get_table("follows")
    key = "following_uid" if mode == "followers" else "follower_uid"
    other = "follower_uid" if mode == "followers" else "following_uid"
    total = follows.count(deleted_at=None, **{key: user_uid})
    pagination = build_pagination(page, total, per_page)
    offset = (pagination["page"] - 1) * pagination["per_page"]
    rows = list(
        follows.find(
            order_by=["-created_at"],
            _limit=pagination["per_page"],
            _offset=offset,
            deleted_at=None,
            **{key: user_uid},
        )
    )
    users_map = get_users_by_uids([row[other] for row in rows])
    people = []
    for row in rows:
        person = users_map.get(row[other])
        if person:
            people.append(
                {
                    "uid": person["uid"],
                    "username": person["username"],
                    "bio": (person.get("bio") or "")[:140],
                    "followed_at": row.get("created_at"),
                }
            )
    return people, pagination


def get_following_among(follower_uid: str, target_uids: list) -> set:
    if not follower_uid or not target_uids or "follows" not in db.tables:
        return set()
    placeholders, params = _in_clause(target_uids)
    params["f"] = follower_uid
    rows = db.query(
        f"SELECT following_uid FROM follows WHERE follower_uid = :f AND following_uid IN ({placeholders}) AND deleted_at IS NULL",
        **params,
    )
    return {row["following_uid"] for row in rows}


def get_daily_topic():
    if "news" in db.tables:
        article = db["news"].find_one(
            status="published", deleted_at=None, order_by=["-synced_at"]
        )
        if article:
            desc = (article.get("description") or "")[:200] or (
                article.get("content") or ""
            )[:200]
            return {
                "title": article.get("title", ""),
                "summary": desc,
                "slug": article.get("slug", ""),
                "url": article.get("url", ""),
                "image_url": article.get("image_url", ""),
            }
    return {
        "title": "Welcome to DevPlace",
        "summary": "Stay tuned for the latest dev news.",
    }


def get_featured_news(limit=5):
    if "news" not in db.tables:
        return []
    from devplacepy.utils import time_ago

    rows = list(
        db["news"].find(
            show_on_landing=1, deleted_at=None, order_by=["-synced_at"], _limit=limit
        )
    )
    articles = []
    for article in rows:
        summary = (article.get("description") or "")[:120] or (
            article.get("content") or ""
        )[:120]
        articles.append(
            {
                "title": article.get("title", ""),
                "summary": summary,
                "slug": article.get("slug", ""),
                "url": article.get("url", ""),
                "source_name": article.get("source_name", ""),
                "time_ago": time_ago(article["synced_at"])
                if article.get("synced_at")
                else "",
            }
        )
    return articles


def _ds_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def create_deepsearch_session(
    uid: str,
    owner_kind: str,
    owner_id: str,
    query: str,
    depth: int,
    max_pages: int,
    collection: str,
) -> None:
    get_table("deepsearch_sessions").insert(
        {
            "uid": uid,
            "owner_kind": owner_kind,
            "owner_id": owner_id,
            "query": query,
            "status": "pending",
            "depth": depth,
            "max_pages": max_pages,
            "score": 0,
            "confidence": 0.0,
            "source_diversity": 0.0,
            "page_count": 0,
            "chunk_count": 0,
            "collection": collection,
            "summary": "",
            "created_at": _ds_now(),
            "completed_at": "",
            "deleted_at": None,
            "deleted_by": None,
        }
    )


def update_deepsearch_session(uid: str, fields: dict) -> None:
    if "deepsearch_sessions" not in db.tables:
        return
    payload = dict(fields)
    payload["uid"] = uid
    get_table("deepsearch_sessions").update(payload, ["uid"])


def get_deepsearch_session(uid: str) -> dict | None:
    if "deepsearch_sessions" not in db.tables:
        return None
    return get_table("deepsearch_sessions").find_one(uid=uid, deleted_at=None)


def add_deepsearch_message(
    uid: str, session_uid: str, role: str, content: str, citations: str = ""
) -> None:
    get_table("deepsearch_messages").insert(
        {
            "uid": uid,
            "session_uid": session_uid,
            "role": role,
            "content": content,
            "citations": citations,
            "created_at": _ds_now(),
            "deleted_at": None,
            "deleted_by": None,
        }
    )


def get_deepsearch_messages(session_uid: str, limit: int = 50) -> list[dict]:
    if "deepsearch_messages" not in db.tables:
        return []
    return list(
        get_table("deepsearch_messages").find(
            session_uid=session_uid,
            deleted_at=None,
            order_by=["created_at"],
            _limit=limit,
        )
    )


def get_cached_deepsearch_url(url_hash: str) -> dict | None:
    if "deepsearch_url_cache" not in db.tables:
        return None
    return get_table("deepsearch_url_cache").find_one(url_hash=url_hash)


def upsert_deepsearch_url_cache(
    url_hash: str,
    url: str,
    title: str,
    content_hash: str,
    status: int,
    byte_size: int,
) -> None:
    table = get_table("deepsearch_url_cache")
    existing = table.find_one(url_hash=url_hash)
    row = {
        "url_hash": url_hash,
        "url": url,
        "title": title,
        "content_hash": content_hash,
        "status": status,
        "byte_size": byte_size,
        "fetched_at": _ds_now(),
    }
    if existing:
        row["uid"] = existing["uid"]
        table.update(row, ["uid"])
    else:
        from devplacepy.utils import generate_uid

        row["uid"] = generate_uid()
        table.insert(row)