from __future__ import annotations

from datetime import date, datetime, timezone, timedelta
import secrets

from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session, selectinload

from app.modules.clients.models import Client, ClientAuditLog
from app.modules.client_groups.models import ClientGroup
from app.modules.consultants.models import ClientConsultantLink, ConsultantManagedClient, ConsultantProfile, ConsultantWorkspace, ConsultantServiceRequest
from app.core.security.passwords import hash_password
from app.modules.core.iam.invite_service import issue_invite_token
from app.modules.core.iam.models import User
from app.modules.core.tenancy.models import Branch
from app.modules.core.rbac.models import Role, UserRole
from app.modules.services.models import (
    ClientServiceSubscription,
    ClientServiceTaskInstance,
    ServiceCatalogue,
    ServiceTaskComment,
)

CONSULTANT_TYPES = [
    ("external_consultant", "External Consultant"),
    ("gst_consultant", "GST Consultant"),
    ("tax_consultant", "Tax Consultant"),
    ("roc_consultant", "ROC Consultant"),
    ("payroll_consultant", "Payroll Consultant"),
    ("franchise_partner", "Franchise Partner"),
    ("saas_customer", "SaaS Customer"),
]

CONSULTANT_RELATIONSHIP_TYPES = [
    ("accounts_consultant", "Accounts Consultant"),
    ("gst_consultant", "GST Consultant"),
    ("tax_consultant", "Tax Consultant"),
    ("roc_consultant", "ROC Consultant"),
    ("payroll_consultant", "Payroll Consultant"),
    ("audit_coordination", "Audit Coordination"),
    ("other", "Other"),
]

CONSULTANT_MANAGED_CLIENT_STATUSES = [
    ("active", "Active"),
    ("prospect", "Prospect"),
    ("on_hold", "On Hold"),
    ("closed", "Closed"),
]

CONSULTANT_MANAGED_CLIENT_STAGES = [
    ("managed", "Managed Client"),
    ("lead", "Lead / Prospect"),
    ("referred_to_firm", "Referred to Audit Firm"),
    ("linked_to_firm", "Linked to Firm Client"),
]

CONSULTANT_CLIENT_CONVERSION_STATUSES = [
    ("not_requested", "Not Requested"),
    ("requested", "Requested"),
    ("under_review", "Under Review"),
    ("approved", "Approved / Converted"),
    ("rejected", "Rejected"),
]

CONSULTANT_WORKSPACE_TYPES = [
    ("consultant_saas", "Consultant SaaS Workspace"),
    ("franchise_partner", "Franchise / Referral Partner"),
    ("platform_partner", "Platform Ecosystem Partner"),
]

CONSULTANT_WORKSPACE_PLANS = [
    ("starter", "Starter"),
    ("professional", "Professional"),
    ("business", "Business"),
    ("franchise", "Franchise"),
]

CONSULTANT_SUBSCRIPTION_STATUSES = [
    ("trial", "Trial"),
    ("active", "Active"),
    ("suspended", "Suspended"),
    ("cancelled", "Cancelled"),
]

CONSULTANT_BILLING_CYCLES = [
    ("manual", "Manual / Not Billed"),
    ("monthly", "Monthly"),
    ("quarterly", "Quarterly"),
    ("yearly", "Yearly"),
]

CONSULTANT_ONBOARDING_STATUSES = [
    ("draft", "Draft"),
    ("invited", "Invited"),
    ("active", "Active"),
    ("approved", "Approved"),
    ("suspended", "Suspended"),
    ("inactive", "Inactive"),
]


def normalise_text(value: str | None) -> str | None:
    text = (value or "").strip()
    return text or None



def _first_branch_id_for_tenant(db: Session, tenant_id: int) -> int | None:
    return db.execute(
        select(Branch.id).where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True)).order_by(Branch.id.asc())
    ).scalar_one_or_none()


def _consultant_role(db: Session) -> Role | None:
    return db.execute(select(Role).where(Role.name == "Consultant", Role.is_active.is_(True))).scalar_one_or_none()


def _user_has_role(db: Session, *, user_id: int, role_id: int) -> bool:
    return db.execute(
        select(UserRole.id).where(UserRole.user_id == user_id, UserRole.role_id == role_id)
    ).scalar_one_or_none() is not None


def ensure_consultant_login_user(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None,
    email: str,
    full_name: str,
    password: str | None,
    invite_user: bool,
) -> tuple[User, str | None]:
    """Create or update a portal login for a consultant and assign Consultant role.

    Returns (user, invite_url). Existing users are not overwritten except that the
    Consultant role is added if missing and basic login flags are enabled.
    """
    login_email = (email or "").strip().lower()
    if not login_email:
        raise ValueError("Login email is required to create consultant login.")

    resolved_branch_id = branch_id or _first_branch_id_for_tenant(db, tenant_id)
    if not resolved_branch_id:
        raise ValueError("A branch is required to create consultant login user.")

    user = db.execute(select(User).where(User.email == login_email)).scalar_one_or_none()
    if user and int(user.tenant_id) != int(tenant_id):
        raise ValueError("A user with this login email already exists in another tenant.")

    if user is None:
        temp_password = (password or "").strip() or secrets.token_urlsafe(12)
        user = User(
            email=login_email,
            full_name=(full_name or login_email).strip(),
            password_hash=hash_password(temp_password),
            tenant_id=tenant_id,
            branch_id=resolved_branch_id,
            is_active=True,
            allow_login=True,
            is_locked=False,
            deleted_at=None,
            must_change_password=True,
            password_changed_at_utc=None,
        )
        db.add(user)
        db.flush()
    else:
        user.full_name = (full_name or user.full_name or login_email).strip()
        user.branch_id = resolved_branch_id
        user.is_active = True
        user.allow_login = True
        user.is_locked = False
        if password and password.strip():
            user.password_hash = hash_password(password.strip())
            user.must_change_password = True

    role = _consultant_role(db)
    if not role:
        raise ValueError("Consultant role is not available. Run startup/permission seeding first.")
    if not _user_has_role(db, user_id=int(user.id), role_id=int(role.id)):
        db.add(UserRole(user_id=int(user.id), role_id=int(role.id)))

    invite_url = None
    if invite_user:
        db.flush()
        token = issue_invite_token(db, user)
        invite_url = f"/invite/accept?token={token}"
    return user, invite_url


def list_consultant_role_users(db: Session, *, tenant_id: int, branch_id: int | None = None) -> list[User]:
    query = (
        select(User)
        .join(UserRole, UserRole.user_id == User.id)
        .join(Role, Role.id == UserRole.role_id)
        .where(User.tenant_id == tenant_id, User.is_active.is_(True), Role.name == "Consultant")
    )
    if branch_id:
        query = query.where(or_(User.branch_id == branch_id, User.branch_id.is_(None)))
    return db.execute(query.order_by(User.full_name.asc(), User.email.asc())).scalars().all()


def list_consultants(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None = None,
    q: str = "",
    include_inactive: bool = False,
) -> list[ConsultantProfile]:
    query = (
        select(ConsultantProfile)
        .options(selectinload(ConsultantProfile.user))
        .where(ConsultantProfile.tenant_id == tenant_id)
    )
    if branch_id:
        query = query.where(or_(ConsultantProfile.branch_id == branch_id, ConsultantProfile.branch_id.is_(None)))
    if not include_inactive:
        query = query.where(ConsultantProfile.is_active.is_(True))
    if q.strip():
        term = f"%{q.strip()}%"
        query = query.where(
            or_(
                ConsultantProfile.contact_person.ilike(term),
                ConsultantProfile.firm_name.ilike(term),
                ConsultantProfile.email.ilike(term),
                ConsultantProfile.mobile.ilike(term),
                ConsultantProfile.specialisation.ilike(term),
            )
        )
    return db.execute(query.order_by(ConsultantProfile.contact_person.asc())).scalars().all()


def get_consultant(db: Session, *, tenant_id: int, consultant_id: int, branch_id: int | None = None) -> ConsultantProfile | None:
    query = (
        select(ConsultantProfile)
        .options(selectinload(ConsultantProfile.user), selectinload(ConsultantProfile.links))
        .where(ConsultantProfile.id == consultant_id, ConsultantProfile.tenant_id == tenant_id)
    )
    if branch_id:
        query = query.where(or_(ConsultantProfile.branch_id == branch_id, ConsultantProfile.branch_id.is_(None)))
    return db.execute(query).scalar_one_or_none()


def get_consultant_by_user(db: Session, *, tenant_id: int, user_id: int) -> ConsultantProfile | None:
    return db.execute(
        select(ConsultantProfile)
        .options(selectinload(ConsultantProfile.user))
        .where(
            ConsultantProfile.tenant_id == tenant_id,
            ConsultantProfile.user_id == user_id,
            ConsultantProfile.is_active.is_(True),
        )
    ).scalar_one_or_none()


def create_or_update_consultant(db: Session, *, payload: dict, user_id: int, consultant: ConsultantProfile | None = None) -> ConsultantProfile:
    if consultant is None:
        consultant = ConsultantProfile(
            tenant_id=payload["tenant_id"],
            branch_id=payload.get("branch_id"),
            user_id=payload.get("user_id"),
            contact_person=payload["contact_person"],
            created_by_user_id=user_id,
        )
        db.add(consultant)

    consultant.branch_id = payload.get("branch_id")
    consultant.user_id = payload.get("user_id")
    consultant.consultant_type = payload.get("consultant_type") or "external_consultant"
    consultant.firm_name = normalise_text(payload.get("firm_name"))
    consultant.contact_person = payload["contact_person"].strip()
    consultant.email = normalise_text(payload.get("email"))
    consultant.mobile = normalise_text(payload.get("mobile"))
    consultant.specialisation = normalise_text(payload.get("specialisation"))
    consultant.gstin = normalise_text(payload.get("gstin"))
    consultant.pan = normalise_text(payload.get("pan"))
    consultant.address = normalise_text(payload.get("address"))
    consultant.status = payload.get("status") or "active"
    consultant.onboarding_status = payload.get("onboarding_status") or "approved"
    consultant.is_platform_partner = bool(payload.get("is_platform_partner"))
    consultant.is_franchise_partner = bool(payload.get("is_franchise_partner"))
    consultant.is_saas_customer = bool(payload.get("is_saas_customer"))
    consultant.is_active = bool(payload.get("is_active", True))
    consultant.remarks = normalise_text(payload.get("remarks"))
    consultant.updated_by_user_id = user_id
    return consultant



def update_consultant_own_profile(db: Session, *, consultant: ConsultantProfile, payload: dict, user_id: int) -> ConsultantProfile:
    consultant.firm_name = normalise_text(payload.get("firm_name"))
    consultant.contact_person = (payload.get("contact_person") or consultant.contact_person or "").strip()
    consultant.email = normalise_text(payload.get("email"))
    consultant.mobile = normalise_text(payload.get("mobile"))
    consultant.specialisation = normalise_text(payload.get("specialisation"))
    consultant.gstin = normalise_text(payload.get("gstin"))
    consultant.pan = normalise_text(payload.get("pan"))
    consultant.address = normalise_text(payload.get("address"))
    consultant.updated_by_user_id = user_id
    return consultant


def list_clients_available_for_link(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None = None,
    partner_user_id: int | None = None,
) -> list[Client]:
    query = select(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False), Client.is_active.is_(True))
    if branch_id:
        query = query.where(Client.branch_id == branch_id)
    if partner_user_id:
        query = query.where(Client.partner_id == partner_user_id)
    return db.execute(query.order_by(Client.client_name.asc())).scalars().all()


def list_client_links_for_consultant(db: Session, *, consultant_id: int) -> list[ClientConsultantLink]:
    return db.execute(
        select(ClientConsultantLink)
        .options(selectinload(ClientConsultantLink.client), selectinload(ClientConsultantLink.service_catalogue))
        .where(ClientConsultantLink.consultant_id == consultant_id)
        .order_by(ClientConsultantLink.is_active.desc(), ClientConsultantLink.id.desc())
    ).scalars().all()


def consultant_linked_clients_page(
    db: Session,
    *,
    consultant: ConsultantProfile,
    q: str = "",
    client_group_id: int | None = None,
    status: str = "active",
    page: int = 1,
    per_page: int = 25,
    sort_by: str = "client_code",
    sort_order: str = "asc",
) -> dict:
    """Return every active consultant-linked firm client for instant browser filtering.

    The existing link permission fields and all downstream service, due-date,
    communication and document controls remain unchanged. Search, group, status
    and sorting are performed instantly in the rendered client table, so no
    pagination or form submission is required.
    """
    stmt = (
        select(ClientConsultantLink, Client, ClientGroup)
        .join(Client, Client.id == ClientConsultantLink.client_id)
        .outerjoin(ClientGroup, ClientGroup.id == Client.client_group_id)
        .where(
            ClientConsultantLink.tenant_id == consultant.tenant_id,
            ClientConsultantLink.consultant_id == consultant.id,
            ClientConsultantLink.is_active.is_(True),
            Client.tenant_id == consultant.tenant_id,
            Client.is_archived.is_(False),
        )
        .order_by(Client.client_code.asc(), Client.client_name.asc(), Client.id.asc())
    )

    rows = []
    for link, client, group in db.execute(stmt).all():
        rows.append({
            "link": link,
            "client": client,
            "client_group_id": getattr(group, "id", None),
            "client_group_code": getattr(group, "group_code", None),
            "client_group_name": getattr(group, "group_name", None),
        })

    groups_stmt = (
        select(ClientGroup.id, ClientGroup.group_code, ClientGroup.group_name)
        .join(Client, Client.client_group_id == ClientGroup.id)
        .join(ClientConsultantLink, ClientConsultantLink.client_id == Client.id)
        .where(
            ClientConsultantLink.tenant_id == consultant.tenant_id,
            ClientConsultantLink.consultant_id == consultant.id,
            ClientConsultantLink.is_active.is_(True),
            Client.tenant_id == consultant.tenant_id,
            Client.is_archived.is_(False),
            ClientGroup.is_active.is_(True),
        )
        .distinct()
        .order_by(ClientGroup.group_name.asc())
    )
    groups = [
        {"id": int(group_id), "code": code, "name": name}
        for group_id, code, name in db.execute(groups_stmt).all()
    ]

    total = len(rows)
    return {
        "rows": rows,
        "groups": groups,
        "meta": {"total": total, "page": 1, "per_page": total, "pages": 1},
        "filters": {
            "q": "",
            "client_group_id": None,
            "status": "active",
            "sort_by": "client_code",
            "sort_order": "asc",
        },
    }


def link_client_to_consultant(
    db: Session,
    *,
    tenant_id: int,
    consultant_id: int,
    client: Client,
    relationship_type: str,
    is_primary: bool,
    can_view_client: bool,
    can_view_services: bool,
    can_view_due_dates: bool,
    can_view_communications: bool,
    remarks: str | None,
    user_id: int,
) -> ClientConsultantLink:
    link = db.execute(
        select(ClientConsultantLink).where(
            ClientConsultantLink.tenant_id == tenant_id,
            ClientConsultantLink.client_id == client.id,
            ClientConsultantLink.consultant_id == consultant_id,
        )
    ).scalar_one_or_none()
    if link is None:
        link = ClientConsultantLink(
            tenant_id=tenant_id,
            branch_id=client.branch_id,
            client_id=client.id,
            consultant_id=consultant_id,
            created_by_user_id=user_id,
        )
        db.add(link)
    link.relationship_type = relationship_type or "accounts_consultant"
    link.is_primary = bool(is_primary)
    link.can_view_client = bool(can_view_client)
    link.can_view_services = bool(can_view_services)
    link.can_view_due_dates = bool(can_view_due_dates)
    link.can_view_communications = bool(can_view_communications)
    link.is_active = True
    link.remarks = normalise_text(remarks)
    link.updated_by_user_id = user_id
    return link


def set_link_active(db: Session, *, tenant_id: int, link_id: int, active: bool, user_id: int) -> ClientConsultantLink | None:
    link = db.execute(
        select(ClientConsultantLink).where(ClientConsultantLink.id == link_id, ClientConsultantLink.tenant_id == tenant_id)
    ).scalar_one_or_none()
    if link:
        link.is_active = active
        link.updated_by_user_id = user_id
    return link


def linked_client_ids_for_consultant(db: Session, *, consultant_id: int, require_communications: bool = False) -> list[int]:
    query = select(ClientConsultantLink.client_id).where(
        ClientConsultantLink.consultant_id == consultant_id,
        ClientConsultantLink.is_active.is_(True),
    )
    if require_communications:
        query = query.where(ClientConsultantLink.can_view_communications.is_(True))
    return [int(x) for x in db.execute(query).scalars().all()]


def consultant_dashboard_payload(db: Session, *, consultant: ConsultantProfile) -> dict:
    """Build the consultant portal dashboard payload.

    This function is intentionally read-only. It does not create or update any
    consultant records. It only aggregates already available data from consultant
    links, task communications, service requests, conversion requests and the
    consultant workspace.
    """
    links = list_client_links_for_consultant(db, consultant_id=consultant.id)
    active_links = [link for link in links if link.is_active]
    communication_client_ids = [int(link.client_id) for link in active_links if link.can_view_communications]
    due_date_client_ids = [int(link.client_id) for link in active_links if link.can_view_due_dates]

    comments: list[tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue]] = []
    if communication_client_ids:
        comments = db.execute(
            select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue)
            .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id)
            .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id)
            .join(Client, Client.id == ClientServiceTaskInstance.client_id)
            .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id)
            .where(
                ServiceTaskComment.tenant_id == consultant.tenant_id,
                ServiceTaskComment.visibility == "consultant",
                ServiceTaskComment.is_deleted.is_(False),
                ClientServiceTaskInstance.client_id.in_(communication_client_ids),
                ClientServiceTaskInstance.tenant_id == consultant.tenant_id,
                ClientServiceTaskInstance.is_active.is_(True),
            )
            .order_by(ServiceTaskComment.created_at_utc.desc(), ServiceTaskComment.id.desc())
            .limit(50)
        ).all()

    today = date.today()
    next_30_days = today + timedelta(days=30)
    upcoming_due = 0
    overdue = 0
    due_items: list[tuple[ClientServiceSubscription, Client, ServiceCatalogue]] = []
    overdue_items: list[tuple[ClientServiceSubscription, Client, ServiceCatalogue]] = []

    if due_date_client_ids:
        due_query = (
            select(ClientServiceSubscription, Client, ServiceCatalogue)
            .join(Client, Client.id == ClientServiceSubscription.client_id)
            .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceSubscription.service_catalogue_id)
            .where(
                ClientServiceSubscription.tenant_id == consultant.tenant_id,
                ClientServiceSubscription.client_id.in_(due_date_client_ids),
                ClientServiceSubscription.is_active.is_(True),
                ClientServiceSubscription.current_due_date.is_not(None),
            )
        )
        due_items = db.execute(
            due_query.where(
                ClientServiceSubscription.current_due_date >= today,
                ClientServiceSubscription.current_due_date <= next_30_days,
            )
            .order_by(ClientServiceSubscription.current_due_date.asc(), Client.client_name.asc())
            .limit(8)
        ).all()
        overdue_items = db.execute(
            due_query.where(
                ClientServiceSubscription.current_due_date < today,
                ClientServiceSubscription.status != "completed",
            )
            .order_by(ClientServiceSubscription.current_due_date.asc(), Client.client_name.asc())
            .limit(8)
        ).all()
        upcoming_due = len(due_items)
        overdue = len(overdue_items)

    managed_clients = list_consultant_managed_clients(
        db,
        tenant_id=consultant.tenant_id,
        consultant_id=consultant.id,
        include_inactive=False,
    )
    managed_stats = consultant_managed_clients_stats(db, consultant=consultant)
    workspace_summary = consultant_workspace_summary(db, consultant=consultant)
    service_requests = list_consultant_service_requests(db, consultant=consultant)
    open_service_requests = [r for r in service_requests if r.status in {"submitted", "under_review", "accepted"}]
    pending_service_requests = [r for r in service_requests if r.status in {"submitted", "under_review"}]

    conversion_requests = [
        r
        for r in managed_clients
        if getattr(r, "conversion_status", "not_requested") in {"requested", "under_review", "approved", "rejected"}
    ]
    pending_conversions = [r for r in conversion_requests if r.conversion_status in {"requested", "under_review"}]
    converted_clients = [r for r in managed_clients if getattr(r, "linked_firm_client_id", None)]

    consultant_user_id = int(getattr(consultant, "user_id", 0) or 0)
    firm_messages = [row for row in comments if int(getattr(row[0], "created_by_user_id", 0) or 0) != consultant_user_id]
    consultant_replies = [row for row in comments if int(getattr(row[0], "created_by_user_id", 0) or 0) == consultant_user_id]

    workspace = workspace_summary.get("workspace")
    workspace_alerts: list[str] = []
    remaining = workspace_summary.get("managed_clients_remaining")
    limit = workspace_summary.get("managed_clients_limit")
    usage_percent = int(workspace_summary.get("usage_percent") or 0)
    if workspace is None:
        workspace_alerts.append("Workspace setup is pending.")
    elif not getattr(workspace, "is_active", True):
        workspace_alerts.append("Workspace is inactive.")
    elif getattr(workspace, "subscription_status", "") in {"suspended", "cancelled"}:
        workspace_alerts.append(f"Subscription status is {workspace.subscription_status.replace('_', ' ').title()}.")
    if limit and remaining is not None and remaining <= 0:
        workspace_alerts.append("Managed client limit has been reached.")
    elif limit and usage_percent >= 80:
        workspace_alerts.append("Managed client usage is above 80% of the workspace limit.")

    return {
        "links": links,
        "active_links": active_links,
        "comments": comments[:8],
        "recent_firm_messages": firm_messages[:6],
        "recent_consultant_replies": consultant_replies[:6],
        "managed_clients": managed_clients[:8],
        "workspace_summary": workspace_summary,
        "workspace": workspace,
        "workspace_alerts": workspace_alerts,
        "service_requests": service_requests[:6],
        "open_service_requests": open_service_requests[:6],
        "pending_service_requests": pending_service_requests[:6],
        "conversion_requests": conversion_requests[:6],
        "pending_conversions": pending_conversions[:6],
        "due_items": due_items,
        "overdue_items": overdue_items,
        "stats": {
            "linked_clients": len(active_links),
            "communication_enabled_clients": len(communication_client_ids),
            "due_date_enabled_clients": len(due_date_client_ids),
            "pending_clarifications": len(firm_messages),
            "consultant_replies": len(consultant_replies),
            "upcoming_due": int(upcoming_due or 0),
            "overdue": int(overdue or 0),
            "open_service_requests": len(open_service_requests),
            "pending_service_requests": len(pending_service_requests),
            "pending_conversions": len(pending_conversions),
            "converted_clients": len(converted_clients),
            **managed_stats,
        },
    }


def _next_workspace_code(db: Session, *, tenant_id: int) -> str:
    count = db.execute(
        select(func.count(ConsultantWorkspace.id)).where(ConsultantWorkspace.tenant_id == tenant_id)
    ).scalar_one()
    return f"CW-{int(count or 0) + 1:04d}"


def get_workspace_by_consultant(
    db: Session,
    *,
    tenant_id: int,
    consultant_id: int,
) -> ConsultantWorkspace | None:
    return db.execute(
        select(ConsultantWorkspace).where(
            ConsultantWorkspace.tenant_id == tenant_id,
            ConsultantWorkspace.consultant_id == consultant_id,
        )
    ).scalar_one_or_none()


def ensure_consultant_workspace(
    db: Session,
    *,
    consultant: ConsultantProfile,
    user_id: int | None = None,
) -> ConsultantWorkspace:
    workspace = get_workspace_by_consultant(db, tenant_id=consultant.tenant_id, consultant_id=consultant.id)
    if workspace:
        return workspace

    workspace = ConsultantWorkspace(
        tenant_id=consultant.tenant_id,
        branch_id=consultant.branch_id,
        consultant_id=consultant.id,
        workspace_code=_next_workspace_code(db, tenant_id=consultant.tenant_id),
        workspace_name=consultant.firm_name or consultant.contact_person,
        workspace_type="franchise_partner" if consultant.is_franchise_partner else "consultant_saas",
        plan_code="franchise" if consultant.is_franchise_partner else "starter",
        subscription_status="active" if consultant.is_saas_customer or consultant.is_franchise_partner else "trial",
        billing_cycle="manual",
        max_managed_clients=100 if consultant.is_franchise_partner else 25,
        max_user_accounts=5 if consultant.is_franchise_partner else 1,
        allow_client_portal=False,
        allow_firm_referrals=True,
        allow_service_marketplace=bool(consultant.is_franchise_partner or consultant.is_platform_partner),
        is_active=True,
        created_by_user_id=user_id,
        updated_by_user_id=user_id,
    )
    db.add(workspace)
    return workspace


def parse_optional_date(value: str | None) -> date | None:
    if not value:
        return None
    text = value.strip()
    if not text:
        return None
    for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(text, fmt).date()
        except ValueError:
            continue
    return None


def update_consultant_workspace(
    db: Session,
    *,
    workspace: ConsultantWorkspace,
    payload: dict,
    user_id: int,
    admin_mode: bool = False,
) -> ConsultantWorkspace:
    """Update consultant workspace.

    Normal consultant portal users may update only safe self-service fields
    (workspace name and remarks). Commercial controls such as plan, billing,
    subscription status, client/user limits and feature switches remain firm/admin
    controlled unless admin_mode=True is explicitly used by a future internal route.
    """
    workspace.workspace_name = normalise_text(payload.get("workspace_name")) or workspace.workspace_name
    workspace.remarks = normalise_text(payload.get("remarks"))

    if admin_mode:
        workspace.workspace_type = payload.get("workspace_type") or workspace.workspace_type
        workspace.plan_code = payload.get("plan_code") or workspace.plan_code
        workspace.billing_cycle = payload.get("billing_cycle") or workspace.billing_cycle
        workspace.subscription_status = payload.get("subscription_status") or workspace.subscription_status
        workspace.subscription_start_date = payload.get("subscription_start_date")
        workspace.subscription_end_date = payload.get("subscription_end_date")
        workspace.max_managed_clients = int(payload.get("max_managed_clients") or workspace.max_managed_clients or 25)
        workspace.max_user_accounts = int(payload.get("max_user_accounts") or workspace.max_user_accounts or 1)
        workspace.allow_client_portal = bool(payload.get("allow_client_portal"))
        workspace.allow_firm_referrals = bool(payload.get("allow_firm_referrals"))
        workspace.allow_service_marketplace = bool(payload.get("allow_service_marketplace"))
        workspace.is_active = bool(payload.get("is_active", True))

    workspace.updated_by_user_id = user_id
    return workspace

def consultant_workspace_summary(db: Session, *, consultant: ConsultantProfile) -> dict:
    workspace = get_workspace_by_consultant(db, tenant_id=consultant.tenant_id, consultant_id=consultant.id)
    managed_total = db.execute(
        select(func.count(ConsultantManagedClient.id)).where(
            ConsultantManagedClient.tenant_id == consultant.tenant_id,
            ConsultantManagedClient.consultant_id == consultant.id,
            ConsultantManagedClient.is_active.is_(True),
        )
    ).scalar_one()
    if not workspace:
        return {
            "workspace": None,
            "managed_clients_used": int(managed_total or 0),
            "managed_clients_limit": None,
            "managed_clients_remaining": None,
            "usage_percent": 0,
        }
    limit = max(int(workspace.max_managed_clients or 0), 0)
    used = int(managed_total or 0)
    remaining = max(limit - used, 0) if limit else None
    usage_percent = int(round((used / limit) * 100)) if limit else 0
    return {
        "workspace": workspace,
        "managed_clients_used": used,
        "managed_clients_limit": limit,
        "managed_clients_remaining": remaining,
        "usage_percent": usage_percent,
    }


def consultant_can_add_managed_client(db: Session, *, consultant: ConsultantProfile) -> tuple[bool, dict]:
    """Return whether the consultant can add one more managed client under workspace limits."""
    summary = consultant_workspace_summary(db, consultant=consultant)
    limit = summary.get("managed_clients_limit")
    used = int(summary.get("managed_clients_used") or 0)
    if limit is None or int(limit or 0) <= 0:
        return True, summary
    return used < int(limit), summary


def _next_consultant_client_code(db: Session, *, tenant_id: int, consultant_id: int) -> str:
    count = db.execute(
        select(func.count(ConsultantManagedClient.id)).where(
            ConsultantManagedClient.tenant_id == tenant_id,
            ConsultantManagedClient.consultant_id == consultant_id,
        )
    ).scalar_one()
    return f"CMC-{consultant_id}-{int(count or 0) + 1:04d}"


def list_consultant_managed_clients(
    db: Session,
    *,
    tenant_id: int,
    consultant_id: int,
    q: str = "",
    status: str = "",
    include_inactive: bool = False,
) -> list[ConsultantManagedClient]:
    query = select(ConsultantManagedClient).where(
        ConsultantManagedClient.tenant_id == tenant_id,
        ConsultantManagedClient.consultant_id == consultant_id,
    )
    if not include_inactive:
        query = query.where(ConsultantManagedClient.is_active.is_(True))
    if status.strip():
        query = query.where(ConsultantManagedClient.status == status.strip())
    if q.strip():
        term = f"%{q.strip()}%"
        query = query.where(
            or_(
                ConsultantManagedClient.client_name.ilike(term),
                ConsultantManagedClient.trade_name.ilike(term),
                ConsultantManagedClient.client_code.ilike(term),
                ConsultantManagedClient.pan.ilike(term),
                ConsultantManagedClient.gstin.ilike(term),
                ConsultantManagedClient.email.ilike(term),
                ConsultantManagedClient.mobile.ilike(term),
            )
        )
    return db.execute(query.order_by(ConsultantManagedClient.client_name.asc())).scalars().all()


def get_consultant_managed_client(
    db: Session,
    *,
    tenant_id: int,
    consultant_id: int,
    managed_client_id: int,
) -> ConsultantManagedClient | None:
    return db.execute(
        select(ConsultantManagedClient).where(
            ConsultantManagedClient.id == managed_client_id,
            ConsultantManagedClient.tenant_id == tenant_id,
            ConsultantManagedClient.consultant_id == consultant_id,
        )
    ).scalar_one_or_none()


def create_or_update_managed_client(
    db: Session,
    *,
    consultant: ConsultantProfile,
    payload: dict,
    user_id: int,
    managed_client: ConsultantManagedClient | None = None,
) -> ConsultantManagedClient:
    if managed_client is None:
        can_add, summary = consultant_can_add_managed_client(db, consultant=consultant)
        if not can_add:
            raise ValueError(
                "Managed client limit reached for this consultant workspace "
                f"({summary.get('managed_clients_used')}/{summary.get('managed_clients_limit')})."
            )
        managed_client = ConsultantManagedClient(
            tenant_id=consultant.tenant_id,
            branch_id=consultant.branch_id,
            consultant_id=consultant.id,
            created_by_user_id=user_id,
        )
        db.add(managed_client)

    managed_client.client_code = normalise_text(payload.get("client_code")) or managed_client.client_code or _next_consultant_client_code(
        db, tenant_id=consultant.tenant_id, consultant_id=consultant.id
    )
    managed_client.client_name = (payload.get("client_name") or "").strip()
    managed_client.trade_name = normalise_text(payload.get("trade_name"))
    managed_client.client_type = payload.get("client_type") or "Other"
    managed_client.pan = normalise_text(payload.get("pan"))
    managed_client.gstin = normalise_text(payload.get("gstin"))
    managed_client.tan = normalise_text(payload.get("tan"))
    managed_client.contact_person_name = normalise_text(payload.get("contact_person_name"))
    managed_client.mobile = normalise_text(payload.get("mobile"))
    managed_client.email = normalise_text(payload.get("email"))
    managed_client.address_line_1 = normalise_text(payload.get("address_line_1"))
    managed_client.address_line_2 = normalise_text(payload.get("address_line_2"))
    managed_client.city = normalise_text(payload.get("city"))
    managed_client.state = normalise_text(payload.get("state"))
    managed_client.pincode = normalise_text(payload.get("pincode"))
    managed_client.country = normalise_text(payload.get("country")) or "India"
    managed_client.service_interest = normalise_text(payload.get("service_interest"))
    managed_client.relationship_stage = payload.get("relationship_stage") or "managed"
    managed_client.status = payload.get("status") or "active"
    managed_client.is_active = bool(payload.get("is_active", True))
    managed_client.notes = normalise_text(payload.get("notes"))
    managed_client.updated_by_user_id = user_id
    return managed_client


def consultant_managed_clients_stats(db: Session, *, consultant: ConsultantProfile) -> dict:
    rows = list_consultant_managed_clients(
        db,
        tenant_id=consultant.tenant_id,
        consultant_id=consultant.id,
        include_inactive=True,
    )
    return {
        "managed_clients": len([r for r in rows if r.is_active]),
        "prospects": len([r for r in rows if r.relationship_stage == "lead" and r.is_active]),
        "referred_to_firm": len([r for r in rows if r.relationship_stage == "referred_to_firm" and r.is_active]),
    }


CONSULTANT_SERVICE_REQUEST_STATUSES = [
    ("submitted", "Submitted"),
    ("under_review", "Under Review"),
    ("accepted", "Accepted"),
    ("rejected", "Rejected"),
    ("converted", "Converted to Engagement"),
    ("closed", "Closed"),
]

CONSULTANT_SERVICE_REQUEST_PRIORITIES = [
    ("low", "Low"),
    ("normal", "Normal"),
    ("high", "High"),
    ("urgent", "Urgent"),
]


def _allowed_consultant_client_ids(db: Session, *, consultant: ConsultantProfile, require_communications: bool = True) -> list[int]:
    query = select(ClientConsultantLink.client_id).where(
        ClientConsultantLink.tenant_id == consultant.tenant_id,
        ClientConsultantLink.consultant_id == consultant.id,
        ClientConsultantLink.is_active.is_(True),
    )
    if require_communications:
        query = query.where(ClientConsultantLink.can_view_communications.is_(True))
    return [int(x) for x in db.execute(query).scalars().all()]


def list_consultant_visible_communications(
    db: Session,
    *,
    consultant: ConsultantProfile,
    q: str = "",
    limit: int = 100,
) -> list[tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue]]:
    client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True)
    if not client_ids:
        return []
    safe_limit = max(1, min(int(limit or 100), 200))
    query = (
        select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue)
        .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id)
        .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id)
        .join(Client, Client.id == ClientServiceTaskInstance.client_id)
        .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id)
        .where(
            ServiceTaskComment.tenant_id == consultant.tenant_id,
            ServiceTaskComment.visibility == "consultant",
            ServiceTaskComment.is_deleted.is_(False),
            ClientServiceTaskInstance.client_id.in_(client_ids),
            ClientServiceTaskInstance.tenant_id == consultant.tenant_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    if q.strip():
        term = f"%{q.strip()}%"
        query = query.where(
            or_(
                Client.client_name.ilike(term),
                Client.client_code.ilike(term),
                ServiceCatalogue.service_name.ilike(term),
                ServiceCatalogue.service_code.ilike(term),
                ClientServiceTaskInstance.task_name.ilike(term),
                ServiceTaskComment.message.ilike(term),
            )
        )
    return db.execute(query.order_by(ServiceTaskComment.created_at_utc.desc(), ServiceTaskComment.id.desc()).limit(safe_limit)).all()


def get_consultant_visible_communication(
    db: Session,
    *,
    consultant: ConsultantProfile,
    comment_id: int,
) -> tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue] | None:
    client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True)
    if not client_ids:
        return None
    return db.execute(
        select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue)
        .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id)
        .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id)
        .join(Client, Client.id == ClientServiceTaskInstance.client_id)
        .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id)
        .where(
            ServiceTaskComment.id == comment_id,
            ServiceTaskComment.tenant_id == consultant.tenant_id,
            ServiceTaskComment.visibility == "consultant",
            ServiceTaskComment.is_deleted.is_(False),
            ClientServiceTaskInstance.client_id.in_(client_ids),
            ClientServiceTaskInstance.tenant_id == consultant.tenant_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    ).first()


def list_consultant_task_timeline(
    db: Session,
    *,
    consultant: ConsultantProfile,
    task_id: int,
) -> list[ServiceTaskComment]:
    client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True)
    if not client_ids:
        return []
    task = db.execute(
        select(ClientServiceTaskInstance).where(
            ClientServiceTaskInstance.id == task_id,
            ClientServiceTaskInstance.tenant_id == consultant.tenant_id,
            ClientServiceTaskInstance.client_id.in_(client_ids),
            ClientServiceTaskInstance.is_active.is_(True),
        )
    ).scalar_one_or_none()
    if not task:
        return []
    return db.execute(
        select(ServiceTaskComment)
        .options(selectinload(ServiceTaskComment.created_by))
        .where(
            ServiceTaskComment.tenant_id == consultant.tenant_id,
            ServiceTaskComment.task_instance_id == task_id,
            ServiceTaskComment.visibility == "consultant",
            ServiceTaskComment.is_deleted.is_(False),
        )
        .order_by(ServiceTaskComment.created_at_utc.asc(), ServiceTaskComment.id.asc())
    ).scalars().all()


def add_consultant_task_reply(
    db: Session,
    *,
    consultant: ConsultantProfile,
    task: ClientServiceTaskInstance,
    message: str,
    user_id: int,
) -> ServiceTaskComment | None:
    client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True)
    if int(task.client_id) not in client_ids:
        return None
    if getattr(task, "is_locked", False) or getattr(getattr(task, "subscription", None), "is_locked", False):
        return None
    clean_message = (message or "").strip()
    if not clean_message:
        return None
    row = ServiceTaskComment(
        tenant_id=task.tenant_id,
        branch_id=task.branch_id,
        subscription_id=task.subscription_id,
        task_instance_id=task.id,
        comment_type="consultant_clarification",
        visibility="consultant",
        message=clean_message,
        created_by_user_id=user_id,
    )
    db.add(row)
    task.updated_by_user_id = user_id
    return row


def _next_service_request_no(db: Session, *, tenant_id: int) -> str:
    count = db.execute(
        select(func.count(ConsultantServiceRequest.id)).where(ConsultantServiceRequest.tenant_id == tenant_id)
    ).scalar_one()
    return f"CSR-{int(count or 0) + 1:05d}"


def list_consultant_requestable_services(db: Session) -> list[ServiceCatalogue]:
    flagged = db.execute(
        select(ServiceCatalogue)
        .where(ServiceCatalogue.is_active.is_(True), ServiceCatalogue.is_consultant_requestable.is_(True))
        .order_by(ServiceCatalogue.service_name.asc())
    ).scalars().all()
    if flagged:
        return flagged
    return db.execute(
        select(ServiceCatalogue)
        .where(ServiceCatalogue.is_active.is_(True))
        .order_by(ServiceCatalogue.service_name.asc())
    ).scalars().all()


def _valid_managed_client_for_consultant(db: Session, *, consultant: ConsultantProfile, managed_client_id: int | None) -> ConsultantManagedClient | None:
    if not managed_client_id:
        return None
    return get_consultant_managed_client(
        db,
        tenant_id=consultant.tenant_id,
        consultant_id=consultant.id,
        managed_client_id=int(managed_client_id),
    )


def _valid_firm_client_for_consultant(db: Session, *, consultant: ConsultantProfile, firm_client_id: int | None) -> Client | None:
    if not firm_client_id:
        return None
    allowed_client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=False)
    if int(firm_client_id) not in allowed_client_ids:
        return None
    return db.execute(
        select(Client).where(
            Client.id == int(firm_client_id),
            Client.tenant_id == consultant.tenant_id,
            Client.is_archived.is_(False),
        )
    ).scalar_one_or_none()


def create_consultant_service_request(
    db: Session,
    *,
    consultant: ConsultantProfile,
    payload: dict,
    user_id: int,
) -> ConsultantServiceRequest:
    service_id = payload.get("service_catalogue_id")
    service = None
    if service_id:
        service = db.execute(
            select(ServiceCatalogue).where(ServiceCatalogue.id == int(service_id), ServiceCatalogue.is_active.is_(True))
        ).scalar_one_or_none()
    managed_client = _valid_managed_client_for_consultant(
        db, consultant=consultant, managed_client_id=payload.get("managed_client_id")
    )
    firm_client = _valid_firm_client_for_consultant(
        db, consultant=consultant, firm_client_id=payload.get("firm_client_id")
    )
    if not managed_client and not firm_client:
        raise ValueError("Select either one managed client or one linked firm client for the service request.")
    requested_service_name = normalise_text(payload.get("requested_service_name")) or (service.service_name if service else None)
    if not requested_service_name:
        raise ValueError("Service name is required.")
    subject = normalise_text(payload.get("subject")) or requested_service_name
    row = ConsultantServiceRequest(
        tenant_id=consultant.tenant_id,
        branch_id=consultant.branch_id,
        consultant_id=consultant.id,
        managed_client_id=managed_client.id if managed_client else None,
        firm_client_id=firm_client.id if firm_client else None,
        service_catalogue_id=service.id if service else None,
        request_no=_next_service_request_no(db, tenant_id=consultant.tenant_id),
        request_type=payload.get("request_type") or "service_request",
        status="submitted",
        priority=payload.get("priority") or "normal",
        requested_service_name=requested_service_name,
        requested_due_date=payload.get("requested_due_date"),
        subject=subject,
        description=normalise_text(payload.get("description")),
        consultant_notes=normalise_text(payload.get("consultant_notes")),
        created_by_user_id=user_id,
        updated_by_user_id=user_id,
        is_active=True,
    )
    db.add(row)
    return row


def list_consultant_service_requests(
    db: Session,
    *,
    consultant: ConsultantProfile,
    status: str = "",
) -> list[ConsultantServiceRequest]:
    query = (
        select(ConsultantServiceRequest)
        .options(
            selectinload(ConsultantServiceRequest.managed_client),
            selectinload(ConsultantServiceRequest.firm_client),
            selectinload(ConsultantServiceRequest.service_catalogue),
            selectinload(ConsultantServiceRequest.reviewed_by),
        )
        .where(
            ConsultantServiceRequest.tenant_id == consultant.tenant_id,
            ConsultantServiceRequest.consultant_id == consultant.id,
            ConsultantServiceRequest.is_active.is_(True),
        )
    )
    if status.strip():
        query = query.where(ConsultantServiceRequest.status == status.strip())
    return db.execute(query.order_by(ConsultantServiceRequest.created_at_utc.desc(), ConsultantServiceRequest.id.desc())).scalars().all()


def get_consultant_service_request(
    db: Session,
    *,
    tenant_id: int,
    request_id: int,
    consultant_id: int | None = None,
) -> ConsultantServiceRequest | None:
    query = (
        select(ConsultantServiceRequest)
        .options(
            selectinload(ConsultantServiceRequest.consultant),
            selectinload(ConsultantServiceRequest.managed_client),
            selectinload(ConsultantServiceRequest.firm_client),
            selectinload(ConsultantServiceRequest.service_catalogue),
            selectinload(ConsultantServiceRequest.created_by),
            selectinload(ConsultantServiceRequest.reviewed_by),
        )
        .where(ConsultantServiceRequest.id == request_id, ConsultantServiceRequest.tenant_id == tenant_id)
    )
    if consultant_id:
        query = query.where(ConsultantServiceRequest.consultant_id == consultant_id)
    return db.execute(query).scalar_one_or_none()


def list_all_consultant_service_requests(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None = None,
    status: str = "",
) -> list[ConsultantServiceRequest]:
    query = (
        select(ConsultantServiceRequest)
        .options(
            selectinload(ConsultantServiceRequest.consultant),
            selectinload(ConsultantServiceRequest.managed_client),
            selectinload(ConsultantServiceRequest.firm_client),
            selectinload(ConsultantServiceRequest.service_catalogue),
        )
        .where(ConsultantServiceRequest.tenant_id == tenant_id, ConsultantServiceRequest.is_active.is_(True))
    )
    if branch_id:
        query = query.where(ConsultantServiceRequest.branch_id == branch_id)
    if status.strip():
        query = query.where(ConsultantServiceRequest.status == status.strip())
    return db.execute(query.order_by(ConsultantServiceRequest.created_at_utc.desc(), ConsultantServiceRequest.id.desc())).scalars().all()


def update_consultant_service_request_status(
    db: Session,
    *,
    request: ConsultantServiceRequest,
    status: str,
    firm_response: str | None,
    user_id: int,
) -> ConsultantServiceRequest:
    allowed = {code for code, _label in CONSULTANT_SERVICE_REQUEST_STATUSES}
    clean_status = (status or "under_review").strip()
    if clean_status not in allowed:
        clean_status = "under_review"
    request.status = clean_status
    request.firm_response = normalise_text(firm_response)
    request.reviewed_by_user_id = user_id
    request.reviewed_at_utc = datetime.now(timezone.utc)
    request.updated_by_user_id = user_id
    return request


# ---------------------------------------------------------------------------
# Phase 5A.7 / 5A.8 helpers
# Consultant-managed client conversion + workspace limit enforcement
# ---------------------------------------------------------------------------

def request_managed_client_conversion(
    db: Session,
    *,
    consultant: ConsultantProfile,
    managed_client: ConsultantManagedClient,
    notes: str | None,
    user_id: int,
) -> ConsultantManagedClient:
    if managed_client.tenant_id != consultant.tenant_id or managed_client.consultant_id != consultant.id:
        raise ValueError("Managed client is not available for this consultant.")
    if managed_client.linked_firm_client_id:
        raise ValueError("This managed client is already linked to a firm client.")
    if managed_client.conversion_status in {"requested", "under_review"}:
        raise ValueError("Conversion request is already pending review.")
    managed_client.conversion_status = "requested"
    managed_client.relationship_stage = "conversion_requested"
    managed_client.conversion_requested_at_utc = datetime.now(timezone.utc)
    managed_client.conversion_requested_by_user_id = user_id
    managed_client.conversion_notes = normalise_text(notes)
    managed_client.updated_by_user_id = user_id
    return managed_client


def list_consultant_conversion_requests(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None = None,
    status: str = "",
) -> list[ConsultantManagedClient]:
    query = (
        select(ConsultantManagedClient)
        .options(
            selectinload(ConsultantManagedClient.consultant),
            selectinload(ConsultantManagedClient.linked_firm_client),
        )
        .where(
            ConsultantManagedClient.tenant_id == tenant_id,
            ConsultantManagedClient.conversion_status != "not_requested",
        )
    )
    if branch_id:
        query = query.where(ConsultantManagedClient.branch_id == branch_id)
    if status.strip():
        query = query.where(ConsultantManagedClient.conversion_status == status.strip())
    return db.execute(
        query.order_by(
            ConsultantManagedClient.conversion_requested_at_utc.desc().nullslast(),
            ConsultantManagedClient.id.desc(),
        )
    ).scalars().all()


def get_client_conversion_request(
    db: Session,
    *,
    tenant_id: int,
    managed_client_id: int,
    branch_id: int | None = None,
) -> ConsultantManagedClient | None:
    query = (
        select(ConsultantManagedClient)
        .options(
            selectinload(ConsultantManagedClient.consultant),
            selectinload(ConsultantManagedClient.linked_firm_client),
        )
        .where(
            ConsultantManagedClient.id == managed_client_id,
            ConsultantManagedClient.tenant_id == tenant_id,
            ConsultantManagedClient.conversion_status != "not_requested",
        )
    )
    if branch_id:
        query = query.where(ConsultantManagedClient.branch_id == branch_id)
    return db.execute(query).scalar_one_or_none()


def _next_converted_client_code(db: Session, *, tenant_id: int, source_code: str | None) -> str:
    base = (normalise_text(source_code) or "CONSULTANT").replace(" ", "-").upper()[:35]
    candidate = f"FC-{base}"
    existing = db.execute(select(Client.id).where(Client.tenant_id == tenant_id, Client.client_code == candidate)).first()
    if not existing:
        return candidate
    count = db.execute(select(func.count(Client.id)).where(Client.tenant_id == tenant_id)).scalar_one()
    return f"FC-{base}-{int(count or 0) + 1:04d}"[:50]


def approve_managed_client_conversion(
    db: Session,
    *,
    managed_client: ConsultantManagedClient,
    user_id: int,
    partner_user_id: int | None = None,
    client_code: str | None = None,
    firm_notes: str | None = None,
) -> Client:
    if managed_client.linked_firm_client_id:
        existing_client = db.get(Client, int(managed_client.linked_firm_client_id))
        if existing_client:
            return existing_client
    if managed_client.conversion_status not in {"requested", "under_review", "rejected"}:
        raise ValueError("Only requested/under-review conversion records can be approved.")
    if not managed_client.branch_id:
        raise ValueError("Managed client has no branch. Set branch before approving conversion.")

    pan = normalise_text(managed_client.pan)
    gstin = normalise_text(managed_client.gstin)
    if pan:
        duplicate_pan = db.execute(
            select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.pan == pan, Client.is_archived.is_(False))
        ).scalar_one_or_none()
        if duplicate_pan:
            raise ValueError(f"A firm client with PAN {pan} already exists: {duplicate_pan.client_name}.")
    if gstin:
        duplicate_gstin = db.execute(
            select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.gstin == gstin, Client.is_archived.is_(False))
        ).scalar_one_or_none()
        if duplicate_gstin:
            raise ValueError(f"A firm client with GSTIN {gstin} already exists: {duplicate_gstin.client_name}.")

    clean_code = normalise_text(client_code) or _next_converted_client_code(
        db, tenant_id=managed_client.tenant_id, source_code=managed_client.client_code
    )
    duplicate_code = db.execute(
        select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.client_code == clean_code)
    ).scalar_one_or_none()
    if duplicate_code:
        raise ValueError(f"Client code {clean_code} already exists.")

    client = Client(
        tenant_id=managed_client.tenant_id,
        branch_id=int(managed_client.branch_id),
        partner_id=partner_user_id,
        engagement_mode="hybrid",
        client_code=clean_code,
        client_name=managed_client.client_name,
        trade_name=managed_client.trade_name,
        client_type=managed_client.client_type or "Other",
        pan=pan,
        gstin=gstin,
        tan=normalise_text(managed_client.tan),
        contact_person_name=managed_client.contact_person_name,
        mobile=managed_client.mobile,
        email=managed_client.email,
        address_line_1=managed_client.address_line_1,
        address_line_2=managed_client.address_line_2,
        city=managed_client.city,
        state=managed_client.state,
        pincode=managed_client.pincode,
        country=managed_client.country or "India",
        status="active",
        client_category="Consultant Referral",
        notes=(managed_client.notes or "") + ("\n\nConverted from consultant-managed client."),
        is_active=True,
        is_archived=False,
        created_by_user_id=user_id,
        updated_by_user_id=user_id,
    )
    db.add(client)
    db.flush()

    managed_client.linked_firm_client_id = client.id
    managed_client.conversion_status = "approved"
    managed_client.relationship_stage = "converted_to_firm_client"
    managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc)
    managed_client.conversion_reviewed_by_user_id = user_id
    managed_client.conversion_firm_notes = normalise_text(firm_notes)
    managed_client.updated_by_user_id = user_id

    link_client_to_consultant(
        db,
        tenant_id=managed_client.tenant_id,
        consultant_id=managed_client.consultant_id,
        client=client,
        relationship_type="audit_coordination",
        is_primary=True,
        can_view_client=True,
        can_view_services=True,
        can_view_due_dates=True,
        can_view_communications=True,
        remarks="Auto-linked after consultant-managed client conversion.",
        user_id=user_id,
    )
    db.add(
        ClientAuditLog(
            client_id=client.id,
            tenant_id=client.tenant_id,
            branch_id=client.branch_id,
            actor_user_id=user_id,
            action="converted_from_consultant_client",
            summary="Client created from consultant-managed client conversion request.",
            payload_json={
                "consultant_id": managed_client.consultant_id,
                "managed_client_id": managed_client.id,
                "partner_id": partner_user_id,
            },
        )
    )
    return client


def mark_conversion_under_review(
    db: Session,
    *,
    managed_client: ConsultantManagedClient,
    user_id: int,
    firm_notes: str | None = None,
) -> ConsultantManagedClient:
    if managed_client.conversion_status != "requested":
        raise ValueError("Only requested conversions can be marked under review.")
    managed_client.conversion_status = "under_review"
    managed_client.conversion_reviewed_by_user_id = user_id
    managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc)
    managed_client.conversion_firm_notes = normalise_text(firm_notes)
    managed_client.updated_by_user_id = user_id
    return managed_client


def reject_managed_client_conversion(
    db: Session,
    *,
    managed_client: ConsultantManagedClient,
    user_id: int,
    firm_notes: str | None,
) -> ConsultantManagedClient:
    if managed_client.conversion_status not in {"requested", "under_review"}:
        raise ValueError("Only pending conversion requests can be rejected.")
    managed_client.conversion_status = "rejected"
    managed_client.relationship_stage = "managed"
    managed_client.conversion_reviewed_by_user_id = user_id
    managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc)
    managed_client.conversion_firm_notes = normalise_text(firm_notes)
    managed_client.updated_by_user_id = user_id
    return managed_client


def get_primary_client_consultant_link(db: Session, *, tenant_id: int, client_id: int) -> ClientConsultantLink | None:
    return db.execute(
        select(ClientConsultantLink)
        .where(
            ClientConsultantLink.tenant_id == tenant_id,
            ClientConsultantLink.client_id == client_id,
            ClientConsultantLink.is_primary.is_(True),
            ClientConsultantLink.is_active.is_(True),
        )
        .order_by(ClientConsultantLink.id.asc())
    ).scalars().first()


def sync_primary_client_consultant_link(
    db: Session, *, client: Client, consultant_id: int | None, actor_user_id: int
) -> ClientConsultantLink | None:
    """Synchronise the client's unrestricted primary consultant link without deleting history."""
    links = db.execute(
        select(ClientConsultantLink).where(
            ClientConsultantLink.tenant_id == client.tenant_id,
            ClientConsultantLink.client_id == client.id,
        )
    ).scalars().all()
    selected = None
    for link in links:
        if consultant_id and int(link.consultant_id) == int(consultant_id) and link.service_catalogue_id is None:
            selected = link
        elif link.is_primary:
            link.is_primary = False
            link.updated_by_user_id = actor_user_id
    if consultant_id is None:
        db.flush()
        return None
    consultant = get_consultant(db, tenant_id=int(client.tenant_id), consultant_id=int(consultant_id))
    if not consultant or not consultant.is_active:
        raise ValueError("Selected primary consultant is not active in this firm.")
    if selected is None:
        selected = ClientConsultantLink(
            tenant_id=client.tenant_id, branch_id=client.branch_id, client_id=client.id,
            consultant_id=consultant.id, service_catalogue_id=None,
            relationship_type="accounts_consultant", is_primary=True, is_active=True,
            can_view_client=True, can_view_services=True, can_view_due_dates=True,
            can_view_communications=True, can_view_engagements=True, can_view_task_status=True,
            can_view_assignee=True, can_view_document_requests=True, can_upload_documents=True,
            can_reply_to_clarifications=True, can_view_filing_details=True,
            can_view_final_documents=True, can_view_permanent_documents=False,
            can_receive_notifications=True, can_act_for_client=True,
            effective_from=getattr(client, "referral_date", None) or date.today(),
            created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
        )
        db.add(selected)
    else:
        selected.branch_id = client.branch_id
        selected.is_primary = True
        selected.is_active = True
        selected.updated_by_user_id = actor_user_id
    db.flush()
    return selected


def get_client_consultant_summary(db: Session, *, tenant_id: int, client_id: int) -> dict:
    primary = get_primary_client_consultant_link(db, tenant_id=tenant_id, client_id=client_id)
    referred_id = db.execute(select(Client.referred_by_consultant_id).where(Client.id == client_id)).scalar_one_or_none()
    ids = {int(x) for x in (getattr(primary, "consultant_id", None), referred_id) if x}
    profiles = db.execute(select(ConsultantProfile).where(ConsultantProfile.id.in_(ids))).scalars().all() if ids else []
    by_id = {int(x.id): x for x in profiles}
    return {
        "primary": by_id.get(int(primary.consultant_id)) if primary else None,
        "referred_by": by_id.get(int(referred_id)) if referred_id else None,
    }
