"""Phone verification workflow utilities."""

from __future__ import annotations

import uuid
import datetime
from typing import Any, Dict, Optional, Tuple, Union

from flask import Flask

from flaskr.common.cache_provider import cache as redis
from flaskr.dao import db
from sqlalchemy import text
from flaskr.service.common.dtos import UserToken
from flaskr.service.common.models import raise_error, raise_param_error
from flaskr.service.order.consts import LEARN_STATUS_RESET
from flaskr.service.shifu.models import PublishedShifu, DraftShifu
from flaskr.service.user.consts import (
    USER_STATE_REGISTERED,
    USER_STATE_UNREGISTERED,
    USER_STATE_TRAIL,
    USER_STATE_PAID,
)
from flaskr.service.user.models import UserInfo as UserEntity, UserVerifyCode
from flaskr.service.common.phone_numbers import normalize_phone_identifier
from flaskr.service.user.utils import (
    generate_token,
    ensure_admin_creator_and_demo_permissions,
)
from flaskr.service.user.repository import (
    build_user_info_from_aggregate,
    build_user_profile_snapshot_from_aggregate,
    ensure_user_for_identifier,
    get_user_entity_by_bid,
    load_user_aggregate,
    load_user_aggregate_by_identifier,
    mark_user_roles,
    update_user_entity_fields,
    upsert_credential,
    upsert_wechat_credentials,
    transactional_session,
)

FIX_CHECK_CODE = None
BOOTSTRAP_LOCK_NAME = "user_first_verified_bootstrap"


def configure_fix_check_code(value: Optional[str]) -> None:
    global FIX_CHECK_CODE
    FIX_CHECK_CODE = value


def _acquire_bootstrap_lock(app: Flask, timeout_seconds: int = 5) -> Optional[bool]:
    bind = db.session.get_bind()
    dialect_name = getattr(getattr(bind, "dialect", None), "name", "")
    if dialect_name != "mysql":
        return None

    lock_value = db.session.execute(
        text("SELECT GET_LOCK(:name, :timeout_seconds)"),
        {
            "name": BOOTSTRAP_LOCK_NAME,
            "timeout_seconds": timeout_seconds,
        },
    ).scalar()
    acquired = bool(lock_value)
    if not acquired:
        app.logger.warning(
            "init_first_course skip bootstrap: failed to acquire named lock %s",
            BOOTSTRAP_LOCK_NAME,
        )
    return acquired


def _release_bootstrap_lock() -> None:
    bind = db.session.get_bind()
    dialect_name = getattr(getattr(bind, "dialect", None), "name", "")
    if dialect_name != "mysql":
        return
    db.session.execute(
        text("SELECT RELEASE_LOCK(:name)"),
        {"name": BOOTSTRAP_LOCK_NAME},
    )


def _is_within_seconds(value: datetime.datetime, *, seconds: int) -> bool:
    if value is None:
        return False
    try:
        if value.tzinfo is not None:
            value = value.replace(tzinfo=None)
    except Exception:
        pass
    now = datetime.datetime.utcnow()
    return (now - value).total_seconds() <= seconds


def _consume_latest_sms_code_from_db(app: Flask, phone: str, code: str) -> str:
    """
    Consume the latest sent SMS verification code from the database.

    Returns:
      - "ok" when the code is valid and is marked as used.
      - "expired" when no valid code exists (missing/used/expired).
      - "invalid" when a code exists but does not match.
    """
    expire_seconds = int(app.config.get("PHONE_CODE_EXPIRE_TIME", 300))
    latest = (
        UserVerifyCode.query.filter(
            UserVerifyCode.phone == phone,
            UserVerifyCode.verify_code_type == 1,
            UserVerifyCode.verify_code_send == 1,
        )
        .order_by(UserVerifyCode.created.desc(), UserVerifyCode.id.desc())
        .first()
    )
    if not latest or int(getattr(latest, "verify_code_used", 0) or 0) == 1:
        return "expired"
    created_at = getattr(latest, "created", None)
    if not created_at or not _is_within_seconds(created_at, seconds=expire_seconds):
        return "expired"
    if (latest.verify_code or "") != (code or ""):
        return "invalid"
    latest.verify_code_used = 1
    db.session.flush()
    return "ok"


def migrate_user_study_record(
    app: Flask, from_user_id: str, to_user_id: str, course_id: Optional[str] = None
) -> None:
    from flaskr.service.learn.models import LearnProgressRecord

    normalized_course_id = str(course_id or "").strip()
    if not normalized_course_id:
        app.logger.warning(
            "migrate_user_study_record skipped: missing course_id, from_user_id=%s, to_user_id=%s",
            from_user_id,
            to_user_id,
        )
        return

    app.logger.info(
        "migrate_user_study_record from_user_id:%s to_user_id:%s course_id:%s",
        from_user_id,
        to_user_id,
        normalized_course_id,
    )
    from_attends = LearnProgressRecord.query.filter(
        LearnProgressRecord.user_bid == from_user_id,
        LearnProgressRecord.status != LEARN_STATUS_RESET,
        LearnProgressRecord.shifu_bid == normalized_course_id,
    ).all()
    to_attends = LearnProgressRecord.query.filter(
        LearnProgressRecord.user_bid == to_user_id,
        LearnProgressRecord.status != LEARN_STATUS_RESET,
        LearnProgressRecord.shifu_bid == normalized_course_id,
    ).all()
    migrate_attends = []
    for from_attend in from_attends:
        to_attend = [
            attend
            for attend in to_attends
            if attend.outline_item_bid == from_attend.outline_item_bid
        ]
        if to_attend:
            continue
        migrate_attends.append(from_attend)

    if not migrate_attends:
        app.logger.info(
            "migrate_user_study_record no-op: from_records=%s to_records=%s course_id=%s",
            len(from_attends),
            len(to_attends),
            normalized_course_id,
        )
        return

    db.session.execute(
        text(
            "update learn_progress_records set user_bid = '%s' where id in (%s)"
            % (to_user_id, ",".join(str(attend.id) for attend in migrate_attends))
        )
    )
    db.session.execute(
        text(
            "update learn_generated_blocks set user_bid = '%s' where progress_record_bid in (%s)"
            % (
                to_user_id,
                ",".join(
                    "'" + str(attend.progress_record_bid) + "'"
                    for attend in migrate_attends
                ),
            )
        )
    )
    db.session.flush()
    app.logger.info(
        "migrate_user_study_record done: migrated_records=%s course_id=%s",
        len(migrate_attends),
        normalized_course_id,
    )


def init_first_course(app: Flask, user_id: str) -> bool:
    # Ensure pending state changes are visible to subsequent queries
    db.session.flush()

    # Count only verified users for the bootstrap check.
    # Support both legacy verified states (1..3) and canonical verified states
    # (1102..1104), while intentionally excluding unregistered states.
    verified_states = [
        1,
        2,
        3,
        USER_STATE_REGISTERED,
        USER_STATE_TRAIL,
        USER_STATE_PAID,
    ]
    lock_acquired = _acquire_bootstrap_lock(app)
    if lock_acquired is False:
        return False
    creator_granted_now = False
    try:
        verified_users = (
            UserEntity.query.filter(UserEntity.deleted == 0)
            .filter(UserEntity.state.in_(verified_states))
            .order_by(UserEntity.created_at.asc(), UserEntity.id.asc())
            .limit(2)
            .all()
        )
        if len(verified_users) != 1 or verified_users[0].user_bid != user_id:
            db.session.flush()
            return False

        # Bootstrap the first verified account so self-hosted deployments are
        # manageable without extra manual role assignment.
        creator_granted_now = not bool(verified_users[0].is_creator)
        mark_user_roles(user_id, is_creator=True, is_operator=True)

        ShifuModel: Union[PublishedShifu, DraftShifu] = PublishedShifu
        # Assign demo shifu only when there is exactly one published course
        course_count = PublishedShifu.query.filter(PublishedShifu.deleted == 0).count()
        if course_count == 0:
            course_count = DraftShifu.query.filter(DraftShifu.deleted == 0).count()
            ShifuModel = DraftShifu
        if course_count != 1:
            db.session.flush()
            return creator_granted_now

        course = (
            ShifuModel.query.filter(ShifuModel.deleted == 0)
            .order_by(ShifuModel.id.asc())
            .first()
        )
        if course:
            # Persist creator on the published record
            course.created_user_bid = user_id
            # Also persist creator on the corresponding draft (used by permission checks)
            draft = DraftShifu.query.filter(
                DraftShifu.deleted == 0,
                DraftShifu.shifu_bid == course.shifu_bid,
            ).first()
            if draft:
                draft.created_user_bid = user_id
        db.session.flush()
        return creator_granted_now
    finally:
        if lock_acquired:
            _release_bootstrap_lock()


def verify_phone_code(
    app: Flask,
    user_id: Optional[str],
    phone: str,
    code: str,
    course_id: Optional[str] = None,
    language: Optional[str] = None,
    login_context: Optional[str] = None,
) -> Tuple[UserToken, bool, Dict[str, Optional[str]]]:
    # Local import avoids circular dependency during module initialization.
    from flaskr.service.profile.funcs import (
        get_user_profile_labels,
        update_user_profile_with_lable,
    )

    if FIX_CHECK_CODE is None:
        configure_fix_check_code(app.config.get("UNIVERSAL_VERIFICATION_CODE"))

    raw_phone = (phone or "").strip()
    normalized_phone = normalize_phone_identifier(raw_phone)
    if not normalized_phone:
        raise_param_error("mobile")
    lookup_phones = [normalized_phone]
    if raw_phone and raw_phone not in lookup_phones:
        lookup_phones.append(raw_phone)
    code_keys = [
        app.config["REDIS_KEY_PREFIX_PHONE_CODE"] + lookup_phone
        for lookup_phone in lookup_phones
    ]
    if code != FIX_CHECK_CODE:
        cached = None
        cached_phone = normalized_phone
        for code_key, lookup_phone in zip(code_keys, lookup_phones):
            cached = redis.get(code_key)
            if cached is not None:
                cached_phone = lookup_phone
                break
        if cached is not None:
            cached_str = (
                cached.decode("utf-8") if isinstance(cached, bytes) else str(cached)
            )
            if code != cached_str:
                raise_error("server.user.smsCheckError")
            status = _consume_latest_sms_code_from_db(app, cached_phone, code)
            if status != "ok" and cached_phone != normalized_phone:
                _consume_latest_sms_code_from_db(app, normalized_phone, code)
        else:
            status = "expired"
            for lookup_phone in lookup_phones:
                status = _consume_latest_sms_code_from_db(app, lookup_phone, code)
                if status == "ok":
                    break
                if status == "invalid":
                    break
            if status == "invalid":
                raise_error("server.user.smsCheckError")
            if status != "ok":
                raise_error("server.user.smsSendExpired")

    redis.delete(*code_keys)

    created_new_user = False
    creator_granted_now = False
    normalized_course_id = str(course_id or "").strip() or None

    with transactional_session():
        target_aggregate = load_user_aggregate_by_identifier(
            normalized_phone, providers=["phone"]
        )
        origin_aggregate = load_user_aggregate(user_id) if user_id else None

        if not target_aggregate and origin_aggregate:
            target_aggregate = origin_aggregate

        if target_aggregate and user_id and target_aggregate.user_bid != user_id:
            app.logger.info(
                "verify_phone_code merge_candidate origin_user_id=%s target_user_id=%s course_id=%s",
                user_id,
                target_aggregate.user_bid,
                normalized_course_id,
            )
            if normalized_course_id is None:
                app.logger.warning(
                    "verify_phone_code skip_study_migration missing_course_id origin_user_id=%s target_user_id=%s",
                    user_id,
                    target_aggregate.user_bid,
                )
            else:
                new_profiles = get_user_profile_labels(
                    app, user_id, normalized_course_id
                )
                update_user_profile_with_lable(
                    app,
                    target_aggregate.user_bid,
                    new_profiles,
                    False,
                    normalized_course_id,
                )
                migrate_user_study_record(
                    app,
                    origin_aggregate.user_bid if origin_aggregate else user_id,
                    target_aggregate.user_bid,
                    normalized_course_id,
                )
            if origin_aggregate:
                missing_open_id = (
                    origin_aggregate.wechat_open_id
                    and not target_aggregate.wechat_open_id
                )
                missing_union_id = (
                    origin_aggregate.wechat_union_id
                    and not target_aggregate.wechat_union_id
                )
                if missing_open_id or missing_union_id:
                    upsert_wechat_credentials(
                        app,
                        user_bid=target_aggregate.user_bid,
                        open_id=(
                            origin_aggregate.wechat_open_id if missing_open_id else None
                        ),
                        union_id=(
                            origin_aggregate.wechat_union_id
                            if missing_union_id
                            else None
                        ),
                        verified=True,
                    )

        if target_aggregate is None:
            defaults = {
                "user_bid": user_id or uuid.uuid4().hex,
                "nickname": "",
                "language": language,
                "state": USER_STATE_REGISTERED,
            }
            target_aggregate, created_new_user = ensure_user_for_identifier(
                app,
                provider="phone",
                identifier=normalized_phone,
                defaults=defaults,
            )
            creator_granted_now = (
                init_first_course(app, target_aggregate.user_bid) or creator_granted_now
            )
        else:
            entity = get_user_entity_by_bid(
                target_aggregate.user_bid, include_deleted=True
            )
            if entity:
                updates: Dict[str, Any] = {"identify": normalized_phone}
                promote_state = target_aggregate.state in (
                    USER_STATE_UNREGISTERED,
                    0,
                )
                if promote_state:
                    updates["state"] = USER_STATE_REGISTERED
                if language:
                    updates["language"] = language
                entity = update_user_entity_fields(entity, **updates)
                if promote_state:
                    creator_granted_now = (
                        init_first_course(app, entity.user_bid) or creator_granted_now
                    )

        upsert_credential(
            app,
            user_bid=target_aggregate.user_bid,
            provider_name="phone",
            subject_id=normalized_phone,
            subject_format="phone",
            identifier=normalized_phone,
            metadata={"course_id": normalized_course_id, "language": language},
            verified=True,
        )

        # If configured, automatically grant creator and demo-course permissions
        creator_granted_now = (
            ensure_admin_creator_and_demo_permissions(
                app,
                target_aggregate.user_bid,
                target_aggregate.language,
                login_context,
            )
            or creator_granted_now
        )

        refreshed = load_user_aggregate(target_aggregate.user_bid)
        if not refreshed:
            raise_error("USER.USER_NOT_FOUND")
        token = generate_token(app, user_id=refreshed.user_bid)
        user_dto = build_user_info_from_aggregate(refreshed)
        snapshot = build_user_profile_snapshot_from_aggregate(refreshed)

    return (
        UserToken(userInfo=user_dto, token=token),
        created_new_user,
        {
            "course_id": normalized_course_id,
            "creator_granted_now": creator_granted_now,
            "language": language,
            "snapshot": snapshot.to_dict(),
        },
    )
