#!/usr/bin/env python3

import asyncio
import re
import sys
from datetime import datetime, timedelta
from pathlib import Path

import pymysql

from telethon import TelegramClient, functions, types
from telethon.errors import (
    FloodWaitError,
    UserAlreadyParticipantError,
    ChannelPrivateError,
    UsernameInvalidError,
    UsernameNotOccupiedError,
)

BASE_DIR = Path(__file__).resolve().parent.parent
DB_CONFIG = BASE_DIR / "config" / "database.php"
SESSION_DIR = BASE_DIR / "engine" / "sessions"

PUBLISHER_ID = 1


def fail(message: str, code: int = 1):
    print("")
    print("❌ ERROR")
    print(message)
    sys.exit(code)


def read_php_database_config():
    if not DB_CONFIG.exists():
        fail(f"Database config not found: {DB_CONFIG}")

    content = DB_CONFIG.read_text(
        encoding="utf-8",
        errors="ignore"
    )

    patterns = {
        "host": r"\$dbHost\s*=\s*['\"]([^'\"]+)['\"]",
        "name": r"\$dbName\s*=\s*['\"]([^'\"]+)['\"]",
        "user": r"\$dbUser\s*=\s*['\"]([^'\"]+)['\"]",
        "password": r"\$dbPass\s*=\s*['\"]([^'\"]*)['\"]",
    }

    result = {}

    for key, pattern in patterns.items():
        match = re.search(pattern, content)

        if not match:
            fail(
                f"Could not read '{key}' "
                "from config/database.php"
            )

        result[key] = match.group(1)

    return result


def get_database():
    cfg = read_php_database_config()

    return pymysql.connect(
        host=cfg["host"],
        user=cfg["user"],
        password=cfg["password"],
        database=cfg["name"],
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
    )


def get_publisher(db):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT *
            FROM publishers
            WHERE id = %s
              AND is_active = 1
            LIMIT 1
            """,
            (PUBLISHER_ID,)
        )

        return cursor.fetchone()


def build_client(publisher):
    api_id_raw = str(
        publisher.get("telegram_api_id") or ""
    ).strip()

    api_hash = str(
        publisher.get("telegram_api_hash") or ""
    ).strip()

    session_name = str(
        publisher.get("session_name") or ""
    ).strip()

    if not api_id_raw:
        fail("Telegram API ID missing.")

    try:
        api_id = int(api_id_raw)
    except ValueError:
        fail("Telegram API ID must be numeric.")

    if not api_hash:
        fail("Telegram API Hash missing.")

    if not session_name:
        session_name = f"publisher_{PUBLISHER_ID}"

    session_name = re.sub(
        r"[^a-zA-Z0-9_-]",
        "_",
        session_name
    )

    session_path = SESSION_DIR / session_name

    return TelegramClient(
        str(session_path),
        api_id,
        api_hash
    )


def get_next_candidate(db):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT
                c.*,
                g.group_key,
                g.name AS group_name
            FROM source_discovery_candidates c
            INNER JOIN content_groups g
                ON g.id = c.group_id
            WHERE c.review_status = 'approved'
              AND c.join_status = 'queued'
              AND c.username IS NOT NULL
              AND c.username <> ''
              AND (
                    c.next_join_attempt_at IS NULL
                    OR c.next_join_attempt_at <= NOW()
              )
            ORDER BY
                c.next_join_attempt_at ASC,
                c.id ASC
            LIMIT 1
            """
        )

        return cursor.fetchone()


def mark_joining(db, candidate_id):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE source_discovery_candidates
            SET
                join_status = 'joining',
                join_attempts = join_attempts + 1,
                last_join_error = NULL
            WHERE id = %s
            LIMIT 1
            """,
            (candidate_id,)
        )


def mark_joined(
    db,
    candidate_id,
    source_id,
    entity
):
    telegram_chat_id = str(
        getattr(entity, "id", "")
        or ""
    )

    title = str(
        getattr(entity, "title", "")
        or ""
    ).strip()

    username = str(
        getattr(entity, "username", "")
        or ""
    ).strip()

    with db.cursor() as cursor:

        cursor.execute(
            """
            UPDATE source_discovery_candidates
            SET
                join_status = 'joined',
                telegram_chat_id = %s,
                name = COALESCE(NULLIF(%s, ''), name),
                joined_at = NOW(),
                next_join_attempt_at = NULL,
                last_join_error = NULL
            WHERE id = %s
            LIMIT 1
            """,
            (
                telegram_chat_id or None,
                title,
                candidate_id
            )
        )

        if source_id:

            cursor.execute(
                """
                UPDATE sources
                SET
                    telegram_chat_id = %s,
                    name =
                        CASE
                            WHEN %s <> ''
                            THEN %s
                            ELSE name
                        END,
                    username =
                        CASE
                            WHEN %s <> ''
                            THEN %s
                            ELSE username
                        END,
                    health_status = 'healthy',
                    last_success_at = NOW(),
                    last_error = NULL,
                    last_error_at = NULL
                WHERE id = %s
                LIMIT 1
                """,
                (
                    telegram_chat_id or None,
                    title,
                    title,
                    username,
                    username,
                    source_id
                )
            )


def mark_retry(
    db,
    candidate_id,
    seconds,
    error
):
    retry_at = (
        datetime.now()
        + timedelta(seconds=seconds)
    ).strftime("%Y-%m-%d %H:%M:%S")

    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE source_discovery_candidates
            SET
                join_status = 'queued',
                next_join_attempt_at = %s,
                last_join_error = %s
            WHERE id = %s
            LIMIT 1
            """,
            (
                retry_at,
                error[:2000],
                candidate_id
            )
        )


def mark_failed(
    db,
    candidate_id,
    error
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE source_discovery_candidates
            SET
                join_status = 'failed',
                last_join_error = %s
            WHERE id = %s
            LIMIT 1
            """,
            (
                error[:2000],
                candidate_id
            )
        )


async def main():
    db = get_database()
    client = None

    try:

        publisher = get_publisher(db)

        if not publisher:
            fail(
                f"Active Publisher "
                f"#{PUBLISHER_ID} not found."
            )

        candidate = get_next_candidate(db)

        print("")
        print("SOURCE JOIN WORKER V1.0")
        print(
            "=============================================="
        )
        print("Max Join/Run:   1")
        print("Auto Archive:   OFF")
        print(
            "=============================================="
        )

        if not candidate:

            print("Queue:           EMPTY")
            print("")
            print("Nothing to join.")
            return

        candidate_id = int(candidate["id"])

        source_id = (
            int(candidate["approved_source_id"])
            if candidate.get("approved_source_id")
            else 0
        )

        username = str(
            candidate.get("username") or ""
        ).strip().lstrip("@")

        print(
            f"Candidate:      #{candidate_id}"
        )
        print(
            f"Group:          "
            f"{candidate.get('group_key')}"
        )
        print(
            f"Username:       @{username}"
        )
        print(
            f"Attempts:       "
            f"{candidate.get('join_attempts')}"
        )

        mark_joining(
            db,
            candidate_id
        )

        client = build_client(
            publisher
        )

        await client.connect()

        if not await client.is_user_authorized():
            mark_failed(
                db,
                candidate_id,
                "Telegram session is not authorized."
            )

            fail(
                "Telegram session is not authorized."
            )

        entity = await client.get_entity(
            username
        )

        try:

            await client(
                functions.channels.JoinChannelRequest(
                    channel=entity
                )
            )

            print("Telegram:       JOINED")

        except UserAlreadyParticipantError:

            print(
                "Telegram:       ALREADY JOINED"
            )

        mark_joined(
            db,
            candidate_id,
            source_id,
            entity
        )

        print("DB Status:      JOINED")

        # Mute joined source forever.
        # This applies both to newly joined channels and channels
        # where the reader account was already a participant.
        try:
            await client(
                functions.account.UpdateNotifySettingsRequest(
                    peer=entity,
                    settings=types.InputPeerNotifySettings(
                        mute_until=2147483647
                    )
                )
            )

            print("Mute:           FOREVER ✅")

        except Exception as mute_exc:

            # Mute failure must NOT break source joining,
            # collecting or archiving.
            print(
                "Mute:           FAILED - "
                + str(mute_exc)
            )

        # Move joined source to Telegram Archived Chats.
        try:
            await client.edit_folder(
                entity,
                folder=1
            )

            with db.cursor() as cursor:
                cursor.execute(
                    """
                    UPDATE source_discovery_candidates
                    SET
                        join_status = 'archived',
                        archived_at = NOW(),
                        last_join_error = NULL
                    WHERE id = %s
                    LIMIT 1
                    """,
                    (candidate_id,)
                )

            print("Archive:        DONE")
            print("DB Status:      ARCHIVED")

        except Exception as archive_exc:

            with db.cursor() as cursor:
                cursor.execute(
                    """
                    UPDATE source_discovery_candidates
                    SET last_join_error = %s
                    WHERE id = %s
                    LIMIT 1
                    """,
                    (
                        (
                            "Archive failed: "
                            + str(archive_exc)
                        )[:2000],
                        candidate_id
                    )
                )

            print(
                "Archive:        FAILED - "
                + str(archive_exc)
            )

        print("")
        print(
            "SOURCE JOIN SUCCESS ✅"
        )

    except FloodWaitError as exc:

        seconds = int(exc.seconds)

        if "candidate_id" in locals():

            mark_retry(
                db,
                candidate_id,
                seconds,
                f"FloodWait: {seconds} seconds"
            )

        print("")
        print(
            f"⚠️ Telegram FloodWait: "
            f"{seconds} seconds"
        )
        print(
            "Candidate returned to queue safely."
        )

    except (
        UsernameInvalidError,
        UsernameNotOccupiedError,
        ChannelPrivateError,
    ) as exc:

        if "candidate_id" in locals():

            mark_failed(
                db,
                candidate_id,
                str(exc)
            )

        print("")
        print("❌ JOIN FAILED")
        print(str(exc))

    except Exception as exc:

        error_text = str(exc)
        error_lower = error_text.lower()

        # Telethon uses a SQLite session file.
        # A temporary SQLite lock must never permanently fail
        # an approved source. Return it safely to the queue.
        if (
            "database is locked" in error_lower
            or "database table is locked" in error_lower
        ):

            if "candidate_id" in locals():

                mark_retry(
                    db,
                    candidate_id,
                    300,
                    "Telegram session database is locked; retry in 300 seconds"
                )

            print("")
            print("⚠️ TELEGRAM SESSION DATABASE LOCKED")
            print("Candidate returned to queue.")
            print("Retry: 300 seconds")

        else:

            if "candidate_id" in locals():

                mark_failed(
                    db,
                    candidate_id,
                    error_text
                )

            print("")
            print("❌ JOIN FAILED")
            print(error_text)

    finally:

        if client is not None:
            try:
                if client.is_connected():
                    await client.disconnect()
            except Exception:
                pass

        db.close()


if __name__ == "__main__":
    asyncio.run(main())