HR-ATS-Portal/backend/inbox/models.py

import logging
import os
from shlex import join
import uuid
from datetime import datetime, timezone
from typing import Any, List, Optional

from dotenv import load_dotenv
from fastapi import HTTPException
from inbox.enums import Candidate_application_Status
from role.models import EnumRoles, Roles
from sqlalchemy import Column, DateTime, case, false, func, or_, update
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import defer, selectinload
from sqlmodel import Field, Relationship, SQLModel, select, true

from job.candidate.models import Activity, Feedback, Interviews
from linkedin_utils import slug_from_url, NO_SLUG
from users.models import Users
from users.plugins import hash_password

load_dotenv()
logger = logging.getLogger("inbox.models")

# Placeholder only. The account lands inactive and the candidate is mailed a
# confirmation link; the real password comes from the reset flow afterwards.
DEFAULT_CANDIDATE_PASSWORD = os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")
CANDIDATE_ROLE_ID = 8                   # seeded candidate role (id 4 is hiring_manager, the signup default)
SKIP_SENDER_PREFIXES = ("noreply", "no-reply", "donotreply", "do-not-reply",
                        "mailer-daemon", "postmaster", "bounce")


def _now() -> datetime:
    return datetime.now(timezone.utc)


class Inbox(SQLModel, table=True):
    __tablename__ = "inbox"

    id: int | None = Field(default=None, primary_key=True)
    user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")

    alert_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_alerts.id")
    alerts: Optional["Inbox_Alerts"] = Relationship(back_populates="inbox")

    message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id")
    messages: Optional["Inbox_Messages"] = Relationship(back_populates="inbox")

    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    favorite: Optional[bool] = Field(default=False)
    rating: Optional[float] = Field(default=0.0)

    ats_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")

    interviews: List["Interviews"] = Relationship(
        back_populates="inbox",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    activity: List["Activity"] = Relationship(
        back_populates="inbox",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    feedback: List["Feedback"] = Relationship(
        back_populates="inbox",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    user: Optional["Users"] = Relationship(
        back_populates="inbox",
        sa_relationship_kwargs={"lazy": "joined"},
    )

    @classmethod
    async def get_all(cls,session:AsyncSession,job_post_id=None,job_post_ids=None,limit=None,offset=0):
        try:
            from job.job_post.models import JobPosts
            qry=(
                select(
                    cls.id.label("inbox_id"),
                    cls.user_id,
                    Users.name,
                    Users.email,
                    Users.linkedin_url,
                    Inbox_Messages.candidate_phone_number.label("phone"),
                    Inbox_Messages.assigned_job_post_id,
                    Inbox_Messages.application_status,
                    Inbox_Messages.is_duplicate,
                    Inbox_Messages.current_employment,
                    Inbox_Messages.current_title,
                    Inbox_Messages.experience,
                    cls.created_at,
                    JobPosts.title,
                    AtsResults.id.label("ats_result_id"),
                    AtsResults.overall_score,
                    AtsResults.band,
                    AtsResults.job_post_id.label("ats_job_post_id"),
                    AtsResults.computed_at,
                    AtsResults.candidate_id,
                    AtsResults.user_id.label("ats_user_id"),
                )
                .join(Users,cls.user_id==Users.id)
                .join(Roles,Users.role_id==Roles.id)
                .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
                .join(JobPosts,Inbox_Messages.assigned_job_post_id==JobPosts.id)
                .outerjoin(AtsResults,cls.ats_id==AtsResults.id)
                .where(Inbox_Messages.assigned_job_post_id.is_not(None))
                .where(Inbox_Messages.attachment==True)  # noqa: E712
                .where(Roles.role_name==EnumRoles.CANDIDATE.value)
                # Newest-first is the list contract; score is only a tiebreak
                # within the same instant. id keeps paging stable.
                .order_by(
                    cls.created_at.desc(),
                    AtsResults.overall_score.desc().nulls_last(),
                    cls.id.desc(),
                )
            )
            if job_post_ids is not None:
                ids=list(job_post_ids)
                if not ids:
                    return []
                qry=qry.where(Inbox_Messages.assigned_job_post_id.in_(ids))
            elif job_post_id:
                qry=qry.where(Inbox_Messages.assigned_job_post_id==job_post_id)
            if limit is not None:
                qry=qry.limit(limit).offset(offset)
            result=await session.execute(qry)
            rows=[]
            for row in result.mappings().all():
                ats=None
                if row["ats_result_id"] is not None:
                    ats={
                        "id":str(row["ats_result_id"]),
                        "overall_score": float(row["overall_score"]) if row["overall_score"] is not None else None,
                        "band":row["band"] or None,
                        "job_post_id":str(row["ats_job_post_id"]) if row["ats_job_post_id"] else None,
                        "computed_at":row["computed_at"].isoformat() if row["computed_at"] else None,
                        "candidate_id":str(row["candidate_id"]) if row["candidate_id"] else None,
                        "user_id":str(row["ats_user_id"]) if row["ats_user_id"] else None,
                    }
                status=row["application_status"]
                rows.append({
                    "inbox_id":row["inbox_id"],
                    "user_id":str(row["user_id"]) if row["user_id"] else None,
                    "name":row["name"],
                    "email":row["email"],
                    "linkedin_url":row["linkedin_url"] or None,
                    "application_status":status.value if status else None,
                    "is_duplicate": bool(row["is_duplicate"]),
                    "phone":row["phone"],
                    "assigned_job_post_id":str(row["assigned_job_post_id"]) if row["assigned_job_post_id"] else None,
                    "title":row["title"] or None,
                    "current_employment":row["current_employment"] or None,
                    "current_title":row["current_title"] or None,
                    "experience":row["experience"] or None,
                    "created_at":row["created_at"].isoformat() if row["created_at"] else None,
                    "ats_result":ats,
                })
            return rows
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    async def list_applications_by_emails(cls, session: AsyncSession, emails):
        """Every inbox_messages row from these senders — linked or not.

        The applications list hides body-only mail (`attachment == True`). History
        still needs those rows: a later proper CV is a new attempt, and the first
        one must remain visible if the system dropped it for format.
        """
        from job.job_post.models import JobPosts

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result = await session.execute(
            select(
                Inbox.id.label("inbox_id"),
                Inbox.user_id.label("user_id"),
                Inbox_Messages.id.label("message_pk"),
                Inbox_Messages.message_id.label("upstream_id"),
                Inbox_Messages.message_from,
                Users.email.label("user_email"),
                Inbox_Messages.assigned_job_post_id,
                Inbox_Messages.application_status,
                Inbox_Messages.message_received_time,
                Inbox_Messages.created_at,
                Inbox_Messages.attachment,
                Inbox_Messages.match_status,
                JobPosts.title,
            )
            .select_from(Inbox_Messages)
            .outerjoin(Inbox, Inbox.message_id == Inbox_Messages.id)
            .outerjoin(Users, Inbox.user_id == Users.id)
            .outerjoin(JobPosts, Inbox_Messages.assigned_job_post_id == JobPosts.id)
            .where(or_(
                func.lower(Inbox_Messages.message_from).in_(lowers),
                func.lower(Users.email).in_(lowers),
            ))
            .order_by(Inbox_Messages.created_at.desc())
        )
        rows = []
        seen = set()
        for row in result.mappings().all():
            mid = row["message_pk"]
            if mid in seen:
                continue
            seen.add(mid)
            sender = (row["message_from"] or "").strip().lower() or None
            user_email = (row["user_email"] or "").strip().lower() or None
            email = sender if sender in lowers else (user_email if user_email in lowers else sender)
            status = row["application_status"]
            applied = row["message_received_time"] or row["created_at"]
            rows.append({
                "source": "inbox",
                "email": email,
                "inbox_id": row["inbox_id"],
                "message_id": str(mid) if mid else None,
                "upstream_id": str(row["upstream_id"]) if row["upstream_id"] else None,
                "manual_upload_candidate_id": None,
                "form_data_id": None,
                "candidate_id": None,
                "user_id": str(row["user_id"]) if row["user_id"] else None,
                "job_post_id": str(row["assigned_job_post_id"]) if row["assigned_job_post_id"] else None,
                "job_title": row["title"] or None,
                "status": status.value if status else None,
                "applied_at": applied.isoformat() if hasattr(applied, "isoformat") else (applied or None),
                "attachment": bool(row["attachment"]),
                "match_status": row["match_status"] or None,
            })
        return rows

    @classmethod
    async def list_silver_medalists(cls, session: AsyncSession, *, min_score=60, limit=500):
        """Rejected applicants who scored well — the CV Bank's second population.

        Read live rather than copied into manual_upload_candidate: these rows
        already exist, and a copy would immediately start drifting from the
        application it was taken from.

        Only REJECTED counts. CLOSED is the ingest DEFAULT for any unprocessed
        email (see Inbox_Messages.application_status), so treating it as a
        rejection would tip the entire unread inbox into the bank.

        Requiring a score is what makes these "silver" rather than merely
        "not hired": an unscored rejection carries no evidence worth keeping.
        """
        try:
            from job.job_post.models import JobPosts
            from job.candidate.models import Candidates
            qry=(
                select(
                    cls.id.label("inbox_id"),
                    cls.user_id,
                    Users.name,
                    Users.email,
                    Users.linkedin_url,
                    Inbox_Messages.candidate_phone_number.label("phone"),
                    Inbox_Messages.current_employment.label("current_company"),
                    Inbox_Messages.current_title,
                    Inbox_Messages.candidate_education.label("education"),
                    Inbox_Messages.file_name,
                    Inbox_Messages.file_path,
                    Inbox_Messages.ats_score,
                    Inbox_Messages.ats_band,
                    cls.created_at,
                    JobPosts.title.label("last_job_title"),
                    Candidates.matched_keywords,
                    Candidates.years_experience,
                )
                .join(Users,cls.user_id==Users.id)
                .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
                .outerjoin(JobPosts,Inbox_Messages.assigned_job_post_id==JobPosts.id)
                .outerjoin(AtsResults,cls.ats_id==AtsResults.id)
                # candidate_id is NULL whenever the CV email matched a user, so
                # this join only sometimes lands — hence the keyword list being
                # optional rather than the filter.
                .outerjoin(Candidates,AtsResults.candidate_id==Candidates.id)
                .where(Inbox_Messages.application_status==Candidate_application_Status.REJECTED)
                .where(Inbox_Messages.ats_score.is_not(None))
                .where(Inbox_Messages.ats_score>=float(min_score))
                .where(Inbox_Messages.is_duplicate==False)  # noqa: E712
                .order_by(Inbox_Messages.ats_score.desc(),cls.created_at.desc(),cls.id.desc())
                .limit(limit)
            )
            result=await session.execute(qry)
            rows=[]
            for row in result.mappings().all():
                rows.append({
                    "inbox_id":row["inbox_id"],
                    "user_id":str(row["user_id"]) if row["user_id"] else None,
                    "name":row["name"],
                    "email":row["email"],
                    "phone":row["phone"] or None,
                    "linkedin_url":row["linkedin_url"] or None,
                    "current_company":row["current_company"] or None,
                    "current_title":row["current_title"] or None,
                    "education":row["education"] or None,
                    "file_name":row["file_name"] or None,
                    "file_path":row["file_path"] or None,
                    "ai_score":int(row["ats_score"]) if row["ats_score"] is not None else None,
                    "recommendation":row["ats_band"] or None,
                    "last_job_title":row["last_job_title"] or None,
                    "matched_keywords":list(row["matched_keywords"] or []),
                    "years_experience":row["years_experience"],
                    "bank_expires_at":None,
                    "created_at":row["created_at"],
                })
            return rows
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    async def linkedin_urls_by_message_ids(cls, session: AsyncSession, message_ids) -> dict:
        """users.linkedin_url keyed by inbox_messages.id for one list page."""
        ids = [mid for mid in (message_ids or []) if mid is not None]
        if not ids:
            return {}
        result = await session.execute(
            select(cls.message_id, Users.linkedin_url)
            .join(Users, Users.id == cls.user_id)
            .where(cls.message_id.in_(ids))
            .where(Users.linkedin_url.is_not(None))
            .where(Users.linkedin_url != "")
        )
        out = {}
        for mid, url in result.all():
            if mid not in out and url:
                out[mid] = url
        return out

    @classmethod
    async def count_by_status(cls,session:AsyncSession,job_post_id=None):
        try:
            from job.job_post.models import JobPosts
            qry=(
                select(Inbox_Messages.application_status,func.count())
                .select_from(cls)
                .join(Users,cls.user_id==Users.id)
                .join(Roles,Users.role_id==Roles.id)
                .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
                .join(JobPosts,Inbox_Messages.assigned_job_post_id==JobPosts.id)
                .where(Inbox_Messages.assigned_job_post_id.is_not(None))
                .where(Inbox_Messages.attachment==True)  # noqa: E712
                .where(Roles.role_name==EnumRoles.CANDIDATE.value)
                .group_by(Inbox_Messages.application_status)
            )
            if job_post_id:
                qry=qry.where(Inbox_Messages.assigned_job_post_id==job_post_id)
            result=await session.execute(qry)
            counts={}
            for status,n in result.all():
                key=status.value if hasattr(status,"value") else (str(status) if status else None)
                if not key:
                    continue
                counts[key]=int(n or 0)
            return counts
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    def _candidate_search_filter(cls, search: str):
        pattern = f"%{search}%"
        return or_(Users.name.ilike(pattern), Users.email.ilike(pattern))

    @classmethod
    async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None,limit:int=10,offset:int=0,search:str|None=None,job_post_ids=None):
        try:
            if job_post_ids is not None and not list(job_post_ids):
                return []
            options=[selectinload(cls.messages)]
            if user_id:
                options.extend([
                    selectinload(cls.interviews),
                    selectinload(cls.activity),
                    selectinload(cls.feedback),
                ])
            qry = (
                select(cls)
                .options(*options)
                .join(Users, cls.user_id == Users.id)
                .join(Roles, Users.role_id == Roles.id)
                .where(Roles.role_name == EnumRoles.CANDIDATE.value)
            )
            if user_id:
                qry = qry.where(cls.user_id == user_id)
            if search:
                qry = qry.where(cls._candidate_search_filter(search))
            if job_post_ids is not None:
                qry = (
                    qry.join(Inbox_Messages, cls.message_id == Inbox_Messages.id)
                    .where(Inbox_Messages.assigned_job_post_id.in_(list(job_post_ids)))
                )
            # Most-recent-first is the list contract; id breaks ties so a page
            # boundary can't drop or repeat a row when created_at collides.
            qry = qry.order_by(cls.created_at.desc(), cls.id.desc())
            qry = qry.limit(limit).offset(offset)
            result = await session.execute(qry)
            rows = result.scalars().all()
            if user_id and len(rows) == 1:
                return rows[0]
            return rows
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    async def count_candidate_profiles(cls,session:AsyncSession,user_id:uuid.UUID|None=None,search:str|None=None,job_post_ids=None):
        """Result-set size for the same predicate get_candidate_profile pages over."""
        try:
            if job_post_ids is not None and not list(job_post_ids):
                return 0
            qry = (
                select(func.count())
                .select_from(cls)
                .join(Users, cls.user_id == Users.id)
                .join(Roles, Users.role_id == Roles.id)
                .where(Roles.role_name == EnumRoles.CANDIDATE.value)
            )
            if user_id:
                qry = qry.where(cls.user_id == user_id)
            if search:
                qry = qry.where(cls._candidate_search_filter(search))
            if job_post_ids is not None:
                qry = (
                    qry.join(Inbox_Messages, cls.message_id == Inbox_Messages.id)
                    .where(Inbox_Messages.assigned_job_post_id.in_(list(job_post_ids)))
                )
            result = await session.execute(qry)
            return result.scalar_one()
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    async def get_users_by_job_post_id(cls,session:AsyncSession,job_post_id):
        """Distinct users.id on inbox rows assigned to this job post.

        Join is inbox → inbox_messages.assigned_job_post_id only — suggestions
        are not a link. Invalid ids yield an empty set, not a 500.
        """
        try:
            try:
                jid=uuid.UUID(str(job_post_id))
            except (TypeError,ValueError,AttributeError):
                return set()
            qry=(
                select(cls.user_id)
                .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
                .where(Inbox_Messages.assigned_job_post_id==jid)
                .where(cls.user_id.is_not(None))
                .distinct()
            )
            result=await session.execute(qry)
            return {uid for uid in result.scalars().all() if uid is not None}
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    def scoped_to_job(cls, statement, department=None, recruiter_id=None):
        """Optional department / recruiter via inbox_messages → job_posts."""
        if not department and not recruiter_id:
            return statement
        from job.job_post.models import JobPosts
        statement = (
            statement
            .outerjoin(Inbox_Messages, cls.message_id == Inbox_Messages.id)
            .outerjoin(JobPosts, Inbox_Messages.assigned_job_post_id == JobPosts.id)
        )
        if department:
            statement = statement.where(JobPosts.department == department)
        try:
            rid = uuid.UUID(str(recruiter_id)) if recruiter_id not in (None, "") else None
        except (TypeError, ValueError):
            rid = None
        if rid is not None:
            statement = statement.where(
                or_(Inbox_Messages.recruiter_id == rid, JobPosts.current_recruiter_id == rid)
            )
        return statement

    @classmethod
    async def count_in_window(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        statement = (
            select(func.count())
            .select_from(cls)
            .join(Users, cls.user_id == Users.id)
            .join(Roles, Users.role_id == Roles.id)
            .where(Roles.role_name == EnumRoles.CANDIDATE.value)
        )
        if from_date is not None:
            statement = statement.where(cls.created_at >= from_date)
        if to_date is not None:
            statement = statement.where(cls.created_at < to_date)
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    async def counts_by_month(
        cls, session: AsyncSession, start, department=None, recruiter_id=None,
    ):
        month_bucket = func.date_trunc("month", cls.created_at)
        statement = (
            select(month_bucket.label("month"), func.count().label("count"))
            .select_from(cls)
            .where(cls.created_at >= start)
            .group_by(month_bucket)
            .order_by(month_bucket)
        )
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        result = await session.execute(statement)
        return [(month, int(count or 0)) for month, count in result.all()]

    @classmethod
    async def get_inbox_by_id(cls,session:AsyncSession,record_id:int|str|None):
        if record_id is None:
            return None
        try:
            iid=int(record_id)
        except (TypeError,ValueError):
            return None
        result=await session.execute(select(cls).where(cls.id==iid))
        return result.scalars().first()

    @classmethod
    async def get_by_ids(cls,session:AsyncSession,ids):
        keys=[]
        for raw in ids or []:
            try:
                keys.append(int(raw))
            except (TypeError,ValueError):
                continue
        if not keys:
            return []
        result=await session.execute(
            select(cls)
            .options(selectinload(cls.messages),selectinload(cls.user))
            .where(cls.id.in_(keys))
        )
        return list(result.scalars().all())

    @classmethod
    async def get_inbox_with_message(cls,session:AsyncSession,record_id:int|str|None):
        """Inbox row with `messages` selectin-loaded for stage / application writers."""
        if record_id is None:
            return None
        try:
            iid=int(record_id)
        except (TypeError,ValueError):
            return None
        result=await session.execute(
            select(cls).options(selectinload(cls.messages)).where(cls.id==iid)
        )
        return result.scalars().first()

    @classmethod
    async def get_inbox_by_message_id(cls,session:AsyncSession,message_id):
        try:
            mid=uuid.UUID(str(message_id))
        except ValueError:
            return None
        result=await session.execute(
            select(cls).where(cls.message_id==mid).order_by(cls.created_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_inbox_by_user_id(cls,session:AsyncSession,user_id):
        try:
            uid=uuid.UUID(str(user_id))
        except ValueError:
            return None
        result=await session.execute(
            select(cls).where(cls.user_id==uid).order_by(cls.created_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def newest_cv_by_user_ids(cls,session:AsyncSession,user_ids):
        """Newest inbox.id + first file_path per user — search Open resume."""
        ids=[]
        for raw in (user_ids or []):
            try:
                ids.append(uuid.UUID(str(raw)))
            except (TypeError,ValueError):
                continue
        if not ids:
            return {}
        result=await session.execute(
            select(cls.user_id,cls.id,Inbox_Messages.file_path)
            .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
            .where(cls.user_id.in_(ids))
            .order_by(cls.created_at.desc())
        )
        out={}
        for user_id,inbox_id,file_path in result.all():
            key=str(user_id)
            if key in out:
                continue
            first=(file_path or "").split(",")[0].strip() or None
            out[key]={"inbox_id":inbox_id,"file_path":first}
        return out

    @classmethod
    async def update_inbox(cls,session:AsyncSession,record_id,fields:dict):
        row=await cls.get_inbox_by_id(session,record_id)
        if not row:
            return None
        for key,value in fields.items():
            setattr(row,key,value)
        row.updated_at=_now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row


class Inbox_Alerts(SQLModel, table=True):
    __tablename__ = "inbox_alerts"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    alert_sender_name: str
    alert_sender_email: str
    is_read: bool = Field(default=False)
    recieve_time: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    inbox: list[Inbox] = Relationship(back_populates="alerts")


class Inbox_Messages(SQLModel, table=True):
    __tablename__ = "inbox_messages"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    message_id: str | None = Field(default=None, index=True, unique=True)
    full_email_response: dict[str, Any] | None = Field(
        default=None, sa_column=Column(JSONB)
    )
    application_status: Candidate_application_Status = Field(default=Candidate_application_Status.CLOSED)
    message_subject: str
    message_body: str
    message_sent_time: str
    message_received_time: str
    message_from: str
    message_to: str
    message_cc: str | None = Field(default=None)
    message_bcc: str | None = Field(default=None)
    message_read: bool = Field(default=False)
    # Stamped whenever a human flips read state from the app (single row, bulk, or
    # whole view). apply_read_status skips these rows: nothing pushes local state
    # back to Outlook, so without the stamp the every-minute sync_read_status sweep
    # would silently re-read a mail the recruiter deliberately marked unread.
    read_overridden_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    attachment: bool = Field(default=False)
    message_reply: str | None = Field(default=None)
    file_name: str | None = Field(default=None)
    file_path: str | None = Field(default=None)
    resume_text: str | None = Field(default=None)
    # Lowercase /in/<slug> extracted from resume_text ("" = scanned, none
    # found; NULL = not yet scanned — see linkedin_utils). Lets Find Talent
    # flag sourced profiles that already applied.
    linkedin_slug: str | None = Field(default=None, index=True)
    experience: str | None = Field(default=None)
    suggested_job_post_ids: list[str] | None = Field(default=None, sa_column=Column(JSONB))
    assigned_job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id", index=True)

    match_summary: str | None = Field(default=None)
    match_reasoning: str | None = Field(default=None)
    match_status: str | None = Field(default=None)
    match_error: str | None = Field(default=None)
    matched_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    candidate_phone_number: str | None = Field(default="xxx-xxx-xxxx")
    candidate_education: str | None = Field(default=None)
    current_employment: str | None = Field(default=None)
    current_title: str | None = Field(default=None)
    # Denormalised dashboard / list-screen fields. server_default is load-bearing
    # for every NOT NULL column — these arrive as ALTERs on a populated table.
    ats_score: float | None = Field(default=None)
    ats_band: str = Field(default="", sa_column_kwargs={"server_default": ""})
    recruiter_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    is_duplicate: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
    source_channel_id: int | None = Field(default=None, foreign_key="source_channels.id")
    processing_state: str = Field(default="unread", sa_column_kwargs={"server_default": "unread"})
    city: str | None = Field(default=None)
    # Job-agnostic tech-stack + department, overwritten on every completed ATS run.
    professional_summary: str | None = Field(default=None)
    # Ingestion time — list screens order by this (newest first). server_default
    # backfills existing rows on the ALTER so NOT NULL is safe on a populated table.
    created_at: datetime = Field(
        default_factory=_now,
        sa_type=DateTime(timezone=True),
        sa_column_kwargs={"server_default": "now()"},
    )
    inbox: list[Inbox] = Relationship(back_populates="messages")

    @staticmethod
    def _body_text(email_data: dict) -> str:
        body = email_data.get("body")
        if isinstance(body, dict):
            return body.get("content") or ""
        if isinstance(body, str):
            return body
        return email_data.get("bodyPreview") or ""


    @classmethod
    async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None):
        try:
            qry=select(cls.message_id,cls.full_email_response,cls.message_subject,cls.message_from,cls.message_to,cls.message_sent_time,cls.message_read,cls.attachment)
            if message_id:
                qry=qry.where(cls.message_id==message_id)
            result=await session.execute(qry)
            return result.scalars().all()

        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))
    @classmethod
    async def set_match_result(
        cls,
        session: AsyncSession,
        record_id,
        *,
        resume_text=None,
        experience=None,
        candidate_education=None,
        candidate_phone_number=None,
        current_employment=None,
        current_title=None,
        linkedin_url=None,
        city=None,
        suggested_job_post_ids=None,
        summary="",
        reasoning="",
        status="",
        error="",
    ):
        """Persist agent output onto one inbox row; returns the row or None."""
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        if resume_text is not None:
            row.resume_text = resume_text
        url = (linkedin_url or "").strip() or None
        user_id = None
        if url:
            row.linkedin_slug = slug_from_url(url) or NO_SLUG
            user_id = await cls.get_linked_user_id(session, row.id)
            if user_id:
                await Users.set_linkedin_url_if_empty(
                    session, user_id=user_id, url=url,
                )
        elif resume_text is not None:
            # Agent ran and found no profile — mark scanned so talent backfill
            # does not regex-scan this CV again.
            row.linkedin_slug = NO_SLUG
        city_value = (city or "").strip() or None
        if city_value:
            row.city = city_value
            if user_id is None:
                user_id = await cls.get_linked_user_id(session, row.id)
            if user_id:
                await Users.set_city_if_empty(
                    session, user_id=user_id, city=city_value,
                )
        if candidate_phone_number is not None:
            row.candidate_phone_number = candidate_phone_number
        if candidate_education is not None:
            row.candidate_education = candidate_education
        if current_employment is not None:
            row.current_employment = current_employment
        if current_title is not None:
            row.current_title = current_title
        row.suggested_job_post_ids = suggested_job_post_ids
        row.match_summary = summary or None
        row.match_reasoning = reasoning or None
        row.match_status = status or None
        row.match_error = error or None
        row.experience = experience or None
        row.matched_at = datetime.now(timezone.utc)
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict:
        return {
            "message_subject": email_data.get("subject") or "",
            "message_body": cls._body_text(email_data),
            "message_sent_time": email_data.get("sentDateTime") or "",
            "message_read": bool(email_data.get("isRead")),
            "message_received_time": email_data.get("receivedDateTime") or "",
            "message_from": email_data.get("from", {})
            .get("emailAddress", {})
            .get("address", ""),
            "message_to": ",".join(
                [r["emailAddress"]["address"] for r in email_data.get("toRecipients", [])]
            ),
            "message_cc": ",".join(
                [r["emailAddress"]["address"] for r in email_data.get("ccRecipients", [])]
            ),
            "message_bcc": ",".join(
                [r["emailAddress"]["address"] for r in email_data.get("bccRecipients", [])]
            ),
            "attachment": bool(email_data.get("hasAttachments")),
            "message_reply": ",".join(
                [r["emailAddress"]["address"] for r in email_data.get("replyTo", [])]
            ),
            "message_id": email_data.get("id"),
            "file_name": ",".join(
                [r.get("name") for r in email_data.get("attachments", [])]
            ),
            "file_path": ",".join(file_path) if file_path else None,
            "full_email_response": email_data,
        }

    @classmethod
    def _sender_address(cls, email_data: dict) -> str:
        return (
            email_data.get("from", {})
            .get("emailAddress", {})
            .get("address", "")
            or ""
        ).strip().lower()

    @classmethod
    def _sender_display_name(cls, email_data: dict, address: str) -> str:
        name = (
            email_data.get("from", {})
            .get("emailAddress", {})
            .get("name")
            or ""
        ).strip()
        if name:
            return name
        return address.split("@", 1)[0] if address else "candidate"

    @classmethod
    def _is_linkable_sender(cls, address: str) -> bool:
        if not address or "@" not in address:
            return False
        local = address.split("@", 1)[0]
        return not local.startswith(SKIP_SENDER_PREFIXES)

    @classmethod
    async def _link_sender(cls,session:AsyncSession,email_data:dict,email):
        address=cls._sender_address(email_data)
        if not cls._is_linkable_sender(address):
            return None
        try:
            # id-only: avoid Users.job_posts selectin / role lazy loads under asyncio
            user_id=(await session.execute(
                select(Users.id).where(func.lower(Users.email)==address)
            )).scalar_one_or_none()

            if user_id is None:
                user=Users(
                    name=cls._sender_display_name(email_data,address),
                    email=address,
                    role_id=CANDIDATE_ROLE_ID,
                    password=hash_password(DEFAULT_CANDIDATE_PASSWORD),
                    is_approved=True,
                )
                session.add(user)
                # autoflush=False: flush so users.id exists before inbox FK insert
                # (Relationship helps ordering, but flush keeps this path explicit).
                await session.flush()
                session.add(Inbox(user_id=user.id,message_id=email.id))
                await session.commit()
                return address

            link=(await session.execute(
                select(Inbox.id).where(Inbox.message_id==email.id,Inbox.user_id==user_id)
            )).scalar_one_or_none()
            if link is None:
                session.add(Inbox(user_id=user_id,message_id=email.id))
                await session.commit()
            return None
        except IntegrityError:
            await session.rollback()
            return None
        except Exception as e:
            await session.rollback()
            logger.warning("sender link failed for %s: %s",address,e)
            return None

    @classmethod
    async def insert_email(
        cls,
        session: AsyncSession,
        email_data: dict,
        file_path: list[str] | None = None,
    ):
        """Returns (row, new_user_email). new_user_email is set only when this call
        created the sender's Users row."""
        fields = cls._fields_from_email(email_data, file_path)
        external_id = fields.get("message_id")
        link_user=None
        if external_id:
            existing = (
                await session.execute(
                    select(cls).where(cls.message_id == external_id)
                )
            ).scalars().first()
            if existing:
                for key, value in fields.items():
                    # Keep prior S3 URLs until attach_email_pdfs_to_s3 replaces them.
                    if key in ("file_path","file_name") and not value:
                        continue
                    setattr(existing, key, value)
                session.add(existing)
                await session.commit()
                await session.refresh(existing)

                if fields.get("attachment") or existing.attachment:
                    link_user=await cls._link_sender(session, email_data, existing)
                    # _link_sender may rollback (IntegrityError); that expires this row
                    await session.refresh(existing)
                return existing, link_user

        email = cls(**fields)
        session.add(email)
        await session.commit()
        await session.refresh(email)

        if fields.get("attachment"):
            link_user=await cls._link_sender(session, email_data, email)
            await session.refresh(email)
        return email, link_user

    @classmethod
    async def get_linked_user_id(cls,session:AsyncSession,message_id):
        try:
            mid=uuid.UUID(str(message_id))
        except (ValueError,TypeError):
            return None
        return (
            await session.execute(select(Inbox.user_id).where(Inbox.message_id==mid))
        ).scalar_one_or_none()

    @classmethod
    async def assigned_job_post_ids_by_emails(cls,session:AsyncSession,emails):
        """(email, assigned_job_post_id) for senders or linked users. Unlinked skipped."""
        from users.models import Users
        lowers=sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result=await session.execute(
            select(cls.message_from,cls.assigned_job_post_id,Users.email)
            .select_from(cls)
            .outerjoin(Inbox,Inbox.message_id==cls.id)
            .outerjoin(Users,Inbox.user_id==Users.id)
            .where(cls.assigned_job_post_id.is_not(None))
            .where(or_(
                func.lower(cls.message_from).in_(lowers),
                func.lower(Users.email).in_(lowers),
            ))
        )
        rows=[]
        for sender,job_id,user_email in result.all():
            if job_id is None:
                continue
            jid=str(job_id)
            sender_key=(sender or "").strip().lower()
            user_key=(user_email or "").strip().lower()
            if sender_key in lowers:
                rows.append((sender_key,jid))
            if user_key in lowers and user_key!=sender_key:
                rows.append((user_key,jid))
        return rows

    @classmethod
    async def set_file_paths(cls,session:AsyncSession,record_id,file_paths,file_names=None):
        row=await cls.get_inbox_message_by_id(session,record_id)
        if not row:
            return None
        paths=file_paths if isinstance(file_paths,list) else ([file_paths] if file_paths else [])
        cleaned=[str(p).strip() for p in paths if p and str(p).strip()]
        row.file_path=",".join(cleaned) if cleaned else None
        row.attachment=bool(cleaned)
        if file_names is not None:
            names=file_names if isinstance(file_names,list) else [file_names]
            row.file_name=",".join(str(n).strip() for n in names if n and str(n).strip()) or row.file_name
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def delete_by_id(cls,session:AsyncSession,record_id):
        """Hard-delete message + inbox links — roll back when S3 upload fails after insert."""
        row=await cls.get_inbox_message_by_id(session,record_id)
        if not row:
            return False
        links=(await session.execute(select(Inbox).where(Inbox.message_id==row.id))).scalars().all()
        for link in links:
            session.delete(link)
        session.delete(row)
        await session.commit()
        return True

    @classmethod
    def _search_filter(cls, search: str):
        pattern = f"%{search}%"
        return or_(
            cls.message_subject.ilike(pattern),
            cls.message_from.ilike(pattern),
            cls.message_body.ilike(pattern),
        )

    @classmethod
    def _no_suggested_jobs(cls):
        """True when suggested_job_post_ids is missing, not an array, or [].

        jsonb_array_length() raises on scalar JSONB (a lone uuid string, an
        object, json null). CASE evaluates WHEN arms in order, so length is
        only read after jsonb_typeof confirms an array.
        """
        typeof = func.jsonb_typeof(cls.suggested_job_post_ids)
        return case(
            (cls.suggested_job_post_ids.is_(None), True),
            (typeof != "array", True),
            (func.jsonb_array_length(cls.suggested_job_post_ids) == 0, True),
            else_=False,
        )

    @staticmethod
    def _cities_match(column, cities):
        """Agent city name vs stored raw text: Karachi matches Karachi(Malir)."""
        clauses = []
        for city in cities or []:
            text = (city or "").strip()
            if not text:
                continue
            safe = text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
            clauses.append(column.ilike(f"%{safe}%", escape="\\"))
        return or_(*clauses) if clauses else None

    @classmethod
    def _apply_filters(
        cls, statement, search: str | None=None, isread: bool=True,
        application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
        assigned: bool | None=None,
        is_duplicate: bool | None=None,
        no_suggestions: bool | None=None,
        processing_state: str | None=None,
        city=None,
        source: str | None=None,
        inbox_filter: str | None=None,
    ):
        """The one WHERE chain shared by the list, the count and the bulk read UPDATE.

        Works on a Select or an Update — both expose .where() — which is the whole
        point: "mark all read in this view" must narrow on exactly the predicates the
        list narrowed on. A scope filter that drifts from the list filter silently
        touches rows the user never saw, and there is no undo for that.
        """
        if search:
            statement = statement.where(cls._search_filter(search))
        if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED:
            statement = statement.where(cls.application_status==application_status)
        if assigned is True:
            statement = statement.where(cls.assigned_job_post_id.is_not(None))
        elif assigned is False:
            statement = statement.where(cls.assigned_job_post_id.is_(None))
        if isread==False:
            statement = statement.where(cls.message_read==False)
        if is_duplicate is True:
            statement = statement.where(cls.is_duplicate==True)  # noqa: E712
        elif is_duplicate is False:
            statement = statement.where(cls.is_duplicate==False)  # noqa: E712
        if no_suggestions is True:
            statement = statement.where(cls._no_suggested_jobs())
        if processing_state:
            statement = statement.where(cls.processing_state == processing_state)
        cities = [c.strip() for c in (city or []) if (c or "").strip()]
        if cities:
            clause = cls._cities_match(cls.city, cities)
            if clause is not None:
                statement = statement.where(clause)
        if source:
            text = source.strip()
            lowered = text.lower()
            if lowered in ("google sheet", "google_sheet", "sheet"):
                statement = statement.where(false())
            else:
                key = lowered.replace(" ", "_")
                channel_ids = select(SourceChannels.id).where(
                    or_(
                        func.lower(SourceChannels.label) == lowered,
                        SourceChannels.key == key,
                    )
                )
                statement = statement.where(or_(
                    cls.source_channel_id.in_(channel_ids),
                    cls.message_to.ilike(f"%{text}%"),
                ))
        if inbox_filter == "matched":
            statement = statement.where(cls.match_status == "matched")
        elif inbox_filter == "unassigned":
            statement = statement.where(cls.assigned_job_post_id.is_(None))
        elif inbox_filter == "rejected":
            statement = statement.where(cls.processing_state == "rejected")
        elif inbox_filter == "duplicate":
            statement = statement.where(cls.is_duplicate == True)  # noqa: E712
        # Inbox / Job Matching only list applications that arrived with a file.
        # Graph hasAttachments lands on this column; body-only mail stays out.
        statement = statement.where(cls.attachment == True)  # noqa: E712
        return statement

    @classmethod
    async def get_inbox_messages(
        cls, session: AsyncSession, top: int | None, skip: int, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None, no_suggestions: bool | None=None, processing_state: str | None=None, city=None, source=None, light: bool=False
    ):
        statement = cls._apply_filters(
            select(cls).order_by(cls.created_at.desc(), cls.id.desc()),
            search, isread, application_status, assigned, is_duplicate,
            no_suggestions, processing_state, city, source,
        )
        if skip:
            statement = statement.offset(skip)

        if top is not None:
            statement = statement.limit(top)

        # List screens never render these. Loading them for every row is what
        # makes GET /inbox/all-applications hang on a full mailbox: each value
        # is TOASTed (Graph payload, extracted CV, HTML body). Do not read
        # them after this — a deferred access lazy-loads per row.
        if light:
            statement = statement.options(
                defer(cls.full_email_response),
                defer(cls.resume_text),
                defer(cls.message_body),
                defer(cls.message_reply),
                defer(cls.match_reasoning),
            )

        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def list_on_hold_scan_rows(cls, session: AsyncSession):
        """On-Hold email applications: id + sender + CV path. No TOAST columns."""
        statement = cls._apply_filters(
            select(cls.id, cls.message_from, cls.file_path, cls.professional_summary),
            no_suggestions=True,
        )
        result = await session.execute(statement)
        rows = []
        for record_id, email, file_path, summary in result.all():
            rows.append({
                "id": record_id,
                "email": (email or "").strip().lower() or None,
                "file_path": (file_path or "").strip() or None,
                "professional_summary": (summary or "").strip() or None,
            })
        return rows

    @classmethod
    async def get_inbox_message_by_id(cls, session: AsyncSession, record_id: str):
        try:
            uid = uuid.UUID(str(record_id))
        except ValueError:
            return None
        result = await session.execute(select(cls).where(cls.id == uid))
        return result.scalars().first()

    @classmethod
    async def set_ats_score(cls, session: AsyncSession, record_id, score, band):
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        row.ats_score = float(score)
        row.ats_band = band or ""
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_professional_summary(cls, session: AsyncSession, record_id, summary):
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        text=(summary or "").strip() or None
        row.professional_summary = text
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_assigned_job_post(cls, session: AsyncSession, record_id, job_post_id):
        """Set or clear assigned_job_post_id; returns the row or None if missing."""
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        if job_post_id is None:
            row.assigned_job_post_id = None
        else:
            try:
                row.assigned_job_post_id = uuid.UUID(str(job_post_id))
            except ValueError:
                return None
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def counts_by_job_post_ids(cls, session: AsyncSession, job_post_ids) -> dict[str, int]:
        """Resolve {job_post_id: applicant_count} for a page of rows in a single query.

        Counts Inbox_Messages rows, not Inbox rows: one message fans out to several
        Inbox rows (one per recipient), so counting Inbox would over-count.
        """
        uids = {u for u in (job_post_ids or []) if u}
        if not uids:
            return {}
        result = await session.execute(
            select(cls.assigned_job_post_id, func.count().label("applicants"))
            .where(cls.assigned_job_post_id.in_(uids))
            .group_by(cls.assigned_job_post_id)
        )
        return {str(job_id): int(n) for job_id, n in result.all()}

    @classmethod
    async def count_inbox_messages(cls, session: AsyncSession, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None, no_suggestions: bool | None=None, processing_state: str | None=None, city=None, source=None):
        statement = cls._apply_filters(
            select(func.count()).select_from(cls),
            search, isread, application_status, assigned, is_duplicate,
            no_suggestions, processing_state, city, source,
        )
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def distinct_cities(cls, session: AsyncSession):
        """Non-blank city values on inbox applications. Distinct only within this table."""
        result = await session.execute(
            select(cls.city)
            .where(cls.city.is_not(None), cls.city != "")
            .where(cls.attachment == True)  # noqa: E712
            .distinct()
        )
        return [text.strip() for text in result.scalars().all() if (text or "").strip()]

    @classmethod
    async def apply_read_status(cls, session: AsyncSession, changes) -> int:
        """[{id, isRead, ...}] -> bulk UPDATE message_read. Returns rows touched.

        read is a ONE-WAY LATCH: only false -> true is applied, never the reverse.
        mark_message_read writes the local column only — nothing pushes the state
        back to Outlook — so upstream keeps reporting isRead=false and the
        every-minute sync_read_status sweep would otherwise revert a mail the user
        just opened. Cost of the latch: un-reading a mail in Outlook no longer
        propagates here.

        Rows with read_overridden_at set are excluded outright. The latch alone is not
        enough once the UI can mark UNREAD: a mail that is read in Outlook keeps being
        reported isRead=true, so the next sweep would undo the recruiter's click within
        the minute. A human decision on this row wins permanently; the only rows
        excluded are ones somebody already decided about.
        """
        if not changes:
            return 0
        read_ids=[c.get("id") for c in changes if c.get("id") and c.get("isRead")]
        if not read_ids:
            return 0
        result=await session.execute(
            update(cls)
            .where(cls.message_id.in_(read_ids),cls.read_overridden_at.is_(None))
            .values(message_read=True)
        )
        await session.commit()
        return result.rowcount or 0

    @classmethod
    async def get_by_upstream_id(cls, session: AsyncSession, message_id):
        """Lookup by the UPSTREAM Graph id, not the local PK.

        get_inbox_message_by_id above takes the uuid primary key; the triage ledger is
        keyed on the upstream id, so overturning a verdict needs this direction.
        """
        result=await session.execute(select(cls).where(cls.message_id == str(message_id)))
        return result.scalars().first()

    @classmethod
    async def existing_message_ids(cls, session: AsyncSession, message_ids) -> set:
        """The subset of upstream ids already persisted — the free half of the gate.

        A message already in this table was judged an application once, so the intake
        classifier must never be paid for a second time; insert_email's upsert still
        refreshes the row. One query per fetch round, columns only.
        """
        ids=[str(m) for m in message_ids or [] if m]
        if not ids:
            return set()
        result=await session.execute(
            select(cls.message_id).where(cls.message_id.in_(ids))
        )
        return {row for (row,) in result.all() if row}

    @classmethod
    async def mark_message_read(cls, session: AsyncSession, record_id, read: bool=True):
        row=await cls.get_inbox_message_by_id(session,record_id)
        if not row:
            return None
        row.message_read=bool(read)
        row.read_overridden_at=_now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_read_bulk(cls, session: AsyncSession, record_ids, read: bool) -> int:
        """Flip read state for an explicit id list in ONE statement. Returns rows matched.

        Unparseable ids are dropped rather than raising: a stale row id in a selection
        must not sink the other 49 the recruiter ticked. The caller compares `updated`
        against `requested` to notice.

        No `message_read != read` predicate here — the caller wants to know how many of
        its ids actually EXIST, which is what rowcount reports without it.
        """
        uids=[]
        for raw in record_ids or []:
            try:
                uids.append(uuid.UUID(str(raw)))
            except (AttributeError, TypeError, ValueError):
                continue
        if not uids:
            return 0
        result=await session.execute(
            update(cls).where(cls.id.in_(uids)).values(message_read=bool(read),read_overridden_at=_now())
        )
        await session.commit()
        return result.rowcount or 0

    @classmethod
    async def set_read_scope(
        cls, session: AsyncSession, read: bool, search: str | None=None, isread: bool=True,
        application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
        assigned: bool | None=None,
        is_duplicate: bool | None=None,
        processing_state: str | None=None,
        no_suggestions: bool | None=None,
        city=None,
        source: str | None=None,
    ) -> int:
        """Mark every row matching a list filter. Returns rows actually CHANGED.

        The extra `message_read != read` predicate is what makes the count honest: the
        recruiter is told "12 marked read", not "1,240 rows touched" on a mailbox that
        was already read. It also keeps read_overridden_at off rows nobody decided
        anything about, so the Outlook sweep keeps its reach over untouched mail.
        """
        statement=cls._apply_filters(
            update(cls),search,isread,application_status,assigned,is_duplicate,
            no_suggestions,processing_state,city,source,
        )
        statement=statement.where(cls.message_read!=bool(read))
        result=await session.execute(
            statement.values(message_read=bool(read),read_overridden_at=_now())
        )
        await session.commit()
        return result.rowcount or 0

    @classmethod
    async def count_processing(cls, session: AsyncSession):
        statement = select(
            func.count().label("all_count"),
            func.coalesce(func.sum(case((cls.message_read == False, 1), else_=0)), 0).label("unread"),  # noqa: E712
            func.coalesce(func.sum(case((cls.processing_state == "imported", 1), else_=0)), 0).label("imported"),
            func.coalesce(func.sum(case((cls.processing_state == "processed", 1), else_=0)), 0).label("processed"),
            func.coalesce(func.sum(case((cls.processing_state == "rejected", 1), else_=0)), 0).label("rejected"),
            func.coalesce(func.sum(case((cls.is_duplicate == True, 1), else_=0)), 0).label("duplicates"),  # noqa: E712
            func.coalesce(func.sum(case((cls._no_suggested_jobs(), 1), else_=0)), 0).label("on_hold"),
            func.coalesce(func.sum(case((cls.assigned_job_post_id.is_not(None), 1), else_=0)), 0).label("assigned"),
            func.coalesce(func.sum(case((cls.assigned_job_post_id.is_(None), 1), else_=0)), 0).label("unassigned"),
        ).where(cls.attachment == True)  # noqa: E712
        row = (await session.execute(statement)).one()
        return {
            "all": int(row.all_count or 0),
            "unread": int(row.unread or 0),
            "imported": int(row.imported or 0),
            "processed": int(row.processed or 0),
            "rejected": int(row.rejected or 0),
            "duplicates": int(row.duplicates or 0),
            "on_hold": int(row.on_hold or 0),
            "assigned": int(row.assigned or 0),
            "unassigned": int(row.unassigned or 0),
        }

    @classmethod
    async def set_processing_state(cls, session: AsyncSession, record_id, processing_state: str):
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        row.processing_state = processing_state
        # Pipeline Shortlist reads application_status=PENDING. CLOSED is the
        # inbox default and maps to Rejected on the board — leaving it unchanged
        # here is why "Move to Shortlist" never landed in Shortlist.
        current = row.application_status
        current_val = current.value if isinstance(current, Candidate_application_Status) else str(current or "")
        if processing_state == "processed" and current_val in ("", "CLOSED", "PROCESS"):
            row.application_status = Candidate_application_Status.PENDING
        elif processing_state == "rejected":
            row.application_status = Candidate_application_Status.REJECTED
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_duplicate(cls, session: AsyncSession, record_id, is_duplicate: bool):
        row = await cls.get_inbox_message_by_id(session, record_id)
        if not row:
            return None
        row.is_duplicate = bool(is_duplicate)
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def count_hires_by_recruiter(
        cls, session: AsyncSession, recruiter_id, *, from_date=None, to_date=None, department=None,
    ):
        """Messages currently HIRED and assigned to this recruiter_id."""
        try:
            uid = uuid.UUID(str(recruiter_id))
        except (TypeError, ValueError):
            return 0
        from job.job_post.models import JobPosts

        statement = select(func.count()).select_from(cls).where(
            cls.recruiter_id == uid,
            cls.application_status == Candidate_application_Status.HIRED,
        )
        if department:
            statement = statement.outerjoin(JobPosts, cls.assigned_job_post_id == JobPosts.id).where(
                JobPosts.department == department
            )
        if from_date is not None or to_date is not None:
            statement = statement.join(Inbox, Inbox.message_id == cls.id)
            if from_date is not None:
                statement = statement.where(Inbox.created_at >= from_date)
            if to_date is not None:
                statement = statement.where(Inbox.created_at < to_date)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    def scoped_to_job(cls, statement, department=None, recruiter_id=None):
        """Optional department / recruiter via assigned job post."""
        if not department and not recruiter_id:
            return statement
        from job.job_post.models import JobPosts
        statement = statement.outerjoin(JobPosts, cls.assigned_job_post_id == JobPosts.id)
        if department:
            statement = statement.where(JobPosts.department == department)
        try:
            rid = uuid.UUID(str(recruiter_id)) if recruiter_id not in (None, "") else None
        except (TypeError, ValueError):
            rid = None
        if rid is not None:
            statement = statement.where(
                or_(cls.recruiter_id == rid, JobPosts.current_recruiter_id == rid)
            )
        return statement

    @classmethod
    def _window_by_inbox(cls, statement, from_date=None, to_date=None):
        if from_date is None and to_date is None:
            return statement
        statement = statement.join(Inbox, Inbox.message_id == cls.id)
        if from_date is not None:
            statement = statement.where(Inbox.created_at >= from_date)
        if to_date is not None:
            statement = statement.where(Inbox.created_at < to_date)
        return statement

    @classmethod
    async def count_hired(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Messages currently HIRED, windowed on inbox.created_at."""
        statement = (
            select(func.count())
            .select_from(cls)
            .join(Inbox, Inbox.message_id == cls.id)
            .where(cls.application_status == Candidate_application_Status.HIRED)
        )
        if from_date is not None:
            statement = statement.where(Inbox.created_at >= from_date)
        if to_date is not None:
            statement = statement.where(Inbox.created_at < to_date)
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    async def counts_by_application_status(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Current-stage counts for the same inbox population the pipeline board uses.

        Assigned-to-a-job, candidate-role only — Inbox.count_by_status without a
        job filter. Optional department / recruiter / created_at window sit on
        top of that; with none of those this matches the board's inbox column.
        """
        from job.job_post.models import JobPosts
        statement = (
            select(cls.application_status, func.count().label("count"))
            .select_from(Inbox)
            .join(Users, Inbox.user_id == Users.id)
            .join(Roles, Users.role_id == Roles.id)
            .join(cls, Inbox.message_id == cls.id)
            .join(JobPosts, cls.assigned_job_post_id == JobPosts.id)
            .where(cls.assigned_job_post_id.is_not(None))
            .where(Roles.role_name == EnumRoles.CANDIDATE.value)
        )
        if from_date is not None:
            statement = statement.where(Inbox.created_at >= from_date)
        if to_date is not None:
            statement = statement.where(Inbox.created_at < to_date)
        if department:
            statement = statement.where(JobPosts.department == department)
        try:
            rid = uuid.UUID(str(recruiter_id)) if recruiter_id not in (None, "") else None
        except (TypeError, ValueError):
            rid = None
        if rid is not None:
            statement = statement.where(
                or_(cls.recruiter_id == rid, JobPosts.current_recruiter_id == rid)
            )
        statement = statement.group_by(cls.application_status)
        result = await session.execute(statement)
        counts = {}
        for status, n in result.all():
            key = str(status.value if hasattr(status, "value") else status)
            counts[key] = int(n or 0)
        return counts

    @classmethod
    async def counts_by_job_post(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Application counts per assigned job post, over the exact population
        counts_by_application_status counts — so a per-job breakdown sums back
        to the funnel's inbox half under the same filters.

        Distinct from counts_by_job_post_ids (which feeds /jobs/fetch): that one
        counts raw inbox_messages with no role gate and no filters.
        """
        from job.job_post.models import JobPosts
        statement = (
            select(cls.assigned_job_post_id, func.count().label("count"))
            .select_from(Inbox)
            .join(Users, Inbox.user_id == Users.id)
            .join(Roles, Users.role_id == Roles.id)
            .join(cls, Inbox.message_id == cls.id)
            .join(JobPosts, cls.assigned_job_post_id == JobPosts.id)
            .where(cls.assigned_job_post_id.is_not(None))
            .where(Roles.role_name == EnumRoles.CANDIDATE.value)
        )
        if from_date is not None:
            statement = statement.where(Inbox.created_at >= from_date)
        if to_date is not None:
            statement = statement.where(Inbox.created_at < to_date)
        if department:
            statement = statement.where(JobPosts.department == department)
        try:
            rid = uuid.UUID(str(recruiter_id)) if recruiter_id not in (None, "") else None
        except (TypeError, ValueError):
            rid = None
        if rid is not None:
            statement = statement.where(
                or_(cls.recruiter_id == rid, JobPosts.current_recruiter_id == rid)
            )
        statement = statement.group_by(cls.assigned_job_post_id)
        result = await session.execute(statement)
        return {str(job_id): int(n or 0) for job_id, n in result.all()}

    @classmethod
    async def counts_by_source(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        statement = (
            select(
                SourceChannels.id.label("source_id"),
                func.coalesce(SourceChannels.label, "Unknown").label("source"),
                func.count().label("count"),
            )
            .select_from(cls)
            .outerjoin(SourceChannels, cls.source_channel_id == SourceChannels.id)
        )
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        statement = cls._window_by_inbox(statement, from_date, to_date)
        statement = statement.group_by(SourceChannels.id, SourceChannels.label).order_by(func.count().desc())
        result = await session.execute(statement)
        return [(source_id, source, int(count or 0)) for source_id, source, count in result.all()]


class Inbox_Message_Triage(SQLModel, table=True):


    __tablename__ = "inbox_message_triage"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    # Upstream Graph id — the same key insert_email upserts on.
    message_id: str = Field(index=True, unique=True)
    is_application: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
    reason_code: str = Field(default="", sa_column_kwargs={"server_default": ""})
    confidence: float | None = Field(default=None)
    evidence: str = Field(default="", sa_column_kwargs={"server_default": ""})
    # Triage_Status: classified | low_confidence | error. Plain text, not a PG enum —
    # alembic autogenerate cannot see new enum labels, and the enum in
    # inbox_classifier/enums.py already gates what code writes here.
    status: str = Field(default="classified", sa_column_kwargs={"server_default": "classified"})
    error: str | None = Field(default=None)
    model_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
    message_subject: str = Field(default="", sa_column_kwargs={"server_default": ""})
    message_from: str = Field(default="", sa_column_kwargs={"server_default": ""})
    message_received_time: str = Field(default="", sa_column_kwargs={"server_default": ""})
    file_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
    attachment: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
    ingested: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
    overridden_by_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    overridden_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    # _now(), never datetime.now(): a naive local value bound to a timestamptz column
    # is read back as UTC and silently backdates the row (see AtsResults below).
    classified_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        if record_id in (None, ""):
            return None
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_by_message_id(cls, session: AsyncSession, message_id):
        result=await session.execute(select(cls).where(cls.message_id == str(message_id)))
        return result.scalars().first()

    @classmethod
    async def get_triage_by_id(cls, session: AsyncSession, record_id):
        rid=cls._as_uuid(record_id)
        if rid is None:
            return None
        result=await session.execute(select(cls).where(cls.id == rid))
        return result.scalars().first()

    @classmethod
    async def verdicts_for_message_ids(cls, session: AsyncSession, message_ids) -> dict:
        """{upstream message_id: is_application} for a whole fetch round, one query."""
        ids=[str(m) for m in message_ids or [] if m]
        if not ids:
            return {}
        result=await session.execute(
            select(cls.message_id, cls.is_application).where(cls.message_id.in_(ids))
        )
        return {message_id: bool(is_application) for message_id, is_application in result.all()}

    @classmethod
    async def record_verdict(cls, session: AsyncSession, fields: dict):
        """Upsert one verdict on message_id.

        Two fetch rounds can race the unique index, so IntegrityError rolls back and
        re-reads rather than failing the round — same shape as _link_sender above.
        """
        message_id=str(fields.get("message_id") or "")
        if not message_id:
            return None
        existing=await cls.get_by_message_id(session, message_id)
        if existing:
            for key, value in fields.items():
                setattr(existing, key, value)
            existing.classified_at=_now()
            session.add(existing)
            await session.commit()
            await session.refresh(existing)
            return existing
        row=cls(**fields)
        session.add(row)
        try:
            await session.commit()
        except IntegrityError:
            await session.rollback()
            return await cls.get_by_message_id(session, message_id)
        await session.refresh(row)
        return row

    @classmethod
    async def list_filtered_by_emails(cls, session: AsyncSession, emails):
        """Non-ingested classifier rows for these senders.

        The inbox never listed these — either the CV could not be read or the
        gate kept the mail out. History still needs the attempt, labelled as
        rejected for format rather than dropped.
        """
        from inbox_classifier.enums import Triage_Reason_Code

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result = await session.execute(
            select(cls)
            .where(func.lower(cls.message_from).in_(lowers))
            .where(cls.ingested == False)  # noqa: E712
            .where(or_(
                cls.attachment == True,  # noqa: E712
                cls.reason_code == Triage_Reason_Code.JOB_APPLICATION.value,
            ))
            .order_by(cls.classified_at.desc())
        )
        rows = []
        for rec in result.scalars().all():
            applied = rec.message_received_time or rec.classified_at
            rows.append({
                "source": "filtered",
                "email": (rec.message_from or "").strip().lower() or None,
                "inbox_id": None,
                "message_id": rec.message_id or None,
                "upstream_id": rec.message_id or None,
                "manual_upload_candidate_id": None,
                "form_data_id": None,
                "candidate_id": None,
                "job_post_id": None,
                "job_title": None,
                "status": "WRONG_FORMAT",
                "applied_at": applied.isoformat() if hasattr(applied, "isoformat") else (applied or None),
                "attachment": bool(rec.attachment),
                "match_status": None,
                "rejection_reason": "wrong_format",
            })
        return rows

    @classmethod
    def _triage_filter(cls, statement, is_application, status, search):
        if is_application is not None:
            statement=statement.where(cls.is_application == bool(is_application))
        if status:
            statement=statement.where(cls.status == str(status))
        if search:
            pattern=f"%{search}%"
            statement=statement.where(
                or_(cls.message_subject.ilike(pattern), cls.message_from.ilike(pattern))
            )
        return statement

    @classmethod
    async def list_triage(cls, session: AsyncSession, top, skip, is_application=None,
                          status=None, search=None):
        statement=cls._triage_filter(select(cls), is_application, status, search)
        statement=statement.order_by(cls.classified_at.desc()).offset(skip).limit(top)
        result=await session.execute(statement)
        return list(result.scalars().all())

    @classmethod
    async def count_triage(cls, session: AsyncSession, is_application=None, status=None,
                           search=None) -> int:
        statement=cls._triage_filter(select(func.count(cls.id)), is_application, status, search)
        result=await session.execute(statement)
        return int(result.scalar() or 0)

    @classmethod
    async def set_override(cls, session: AsyncSession, record_id, is_application, user_id=None):
        row=await cls.get_triage_by_id(session, record_id)
        if not row:
            return None
        row.is_application=bool(is_application)
        row.overridden_by_id=cls._as_uuid(user_id)
        row.overridden_at=_now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def mark_ingested(cls, session: AsyncSession, message_id, ingested: bool = True):
        row=await cls.get_by_message_id(session, message_id)
        if not row:
            return None
        row.ingested=bool(ingested)
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row


class SourceChannels(SQLModel, table=True):
    __tablename__ = "source_channels"

    id: int | None = Field(default=None, primary_key=True)
    key: str = Field(max_length=40, unique=True, index=True)
    label: str
    is_active: bool = Field(default=True)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    @classmethod
    async def get_by_id(cls, session: AsyncSession, record_id: int):
        result = await session.execute(select(cls).where(cls.id == record_id))
        return result.scalars().first()

    @classmethod
    async def get_by_key(cls, session: AsyncSession, key: str):
        result = await session.execute(select(cls).where(cls.key == key))
        return result.scalars().first()

    @classmethod
    async def list_active(cls, session: AsyncSession):
        result = await session.execute(
            select(cls).where(cls.is_active == True).order_by(cls.id.asc())  # noqa: E712
        )
        return list(result.scalars().all())

    @classmethod
    async def labels_by_ids(cls, session: AsyncSession, ids):
        keys = [cid for cid in (ids or []) if cid is not None]
        if not keys:
            return []
        result = await session.execute(select(cls.id, cls.label).where(cls.id.in_(keys)))
        return [(cid, label) for cid, label in result.all()]


class AtsResults(SQLModel, table=True):
    __tablename__ = "ats_results"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    # Inbox applications link here; NULL for upload-sourced scores, which have no
    # inbox row. Identity is XOR: matching users.email -> user_id (candidate_id
    # NULL); otherwise candidate_id (user_id NULL). Blank email is the latter.
    inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id")
    candidate_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="candidates.id")
    user_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="users.id")
    job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
    # Sheet Forms scores: no inbox/user. Identity is (form_data_id, job_post_id).
    form_data_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="form_data.id")
    # On-Hold ReScan run that produced this row. NULL for assign-job / upload scores.
    rescan_run_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="inbox_rescan_runs.id")
    professional_summary: str | None = Field(default=None)
    overall_score: float = Field(default=0.0)
    band: str = Field(default="")
    is_current: bool = Field(default=True)
    superseded_by_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")
    model_name: str | None = Field(default=None)
    # default_factory was datetime.now: a naive LOCAL value bound to a timestamptz
    # column, which asyncpg reads as UTC. That silently backdated every row by the
    # host's offset (+5 h here) instead of raising, unlike the naive-column case.
    computed_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        if record_id in (None, ""):
            return None
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_current_for_inbox(cls, session: AsyncSession, inbox_id: int):
        result = await session.execute(
            select(cls)
            .where(cls.inbox_id == int(inbox_id), cls.is_current == True)  # noqa: E712
            .order_by(cls.computed_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_by_ids(cls, session: AsyncSession, ids):
        keys = []
        for raw in ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return []
        result = await session.execute(select(cls).where(cls.id.in_(keys)))
        return list(result.scalars().all())

    @classmethod
    async def get_for_inbox_job(cls, session: AsyncSession, inbox_id, job_post_id):
        """Any score for this application against this job — current or superseded.

        Inbox-side idempotency is (inbox_id, job_post_id) only. candidate_id may
        be NULL when the CV email matched a user; do not key off it here.
        """
        jid = cls._as_uuid(job_post_id)
        if jid is None:
            return None
        result = await session.execute(
            select(cls)
            .where(cls.inbox_id == int(inbox_id), cls.job_post_id == jid)
            .order_by(cls.computed_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_latest_by_job_for_inbox(cls, session: AsyncSession, inbox_id):
        """Newest score per job_post_id for one inbox row (current or superseded)."""
        if inbox_id is None:
            return []
        result = await session.execute(
            select(cls)
            .where(cls.inbox_id == int(inbox_id))
            .order_by(cls.computed_at.desc())
        )
        by_job = {}
        for row in result.scalars():
            jid = str(row.job_post_id) if row.job_post_id else None
            if jid and jid not in by_job:
                by_job[jid] = row
        return list(by_job.values())

    @classmethod
    async def get_latest_by_job_for_messages(cls, session: AsyncSession, message_ids) -> dict:
        """{str(message_id): [newest score per job]} for a page of inbox messages."""
        keys = []
        for raw in message_ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return {}
        result = await session.execute(
            select(cls, Inbox.message_id)
            .join(Inbox, cls.inbox_id == Inbox.id)
            .where(Inbox.message_id.in_(keys))
            .order_by(cls.computed_at.desc())
        )
        grouped = {}
        for row, mid in result.all():
            if mid is None:
                continue
            bucket = grouped.setdefault(mid, {})
            jid = str(row.job_post_id) if row.job_post_id else None
            if jid and jid not in bucket:
                bucket[jid] = row
        return {str(mid): list(jobs.values()) for mid, jobs in grouped.items()}

    @classmethod
    async def get_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id):
        """Any score for this form_data row against this job — current or superseded."""
        fid = cls._as_uuid(form_data_id)
        jid = cls._as_uuid(job_post_id)
        if fid is None or jid is None:
            return None
        result = await session.execute(
            select(cls)
            .where(cls.form_data_id == fid, cls.job_post_id == jid)
            .order_by(cls.computed_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_current_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id=None):
        """Current Sheet Forms score, optionally pinned to one job post."""
        fid = cls._as_uuid(form_data_id)
        if fid is None:
            return None
        qry = select(cls).where(cls.form_data_id == fid, cls.is_current == True)  # noqa: E712
        jid = cls._as_uuid(job_post_id) if job_post_id is not None else None
        if jid is not None:
            qry = qry.where(cls.job_post_id == jid)
        result = await session.execute(qry.order_by(cls.computed_at.desc()))
        return result.scalars().first()

    @classmethod
    async def get_current_for_forms(cls, session: AsyncSession, form_data_ids) -> dict:
        """Current Sheet Forms scores grouped by form_data_id (newest first)."""
        keys = []
        for raw in form_data_ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return {}
        result = await session.execute(
            select(cls)
            .where(cls.form_data_id.in_(keys), cls.is_current == True)  # noqa: E712
            .order_by(cls.computed_at.desc())
        )
        grouped: dict = {}
        for row in result.scalars():
            grouped.setdefault(row.form_data_id, []).append(row)
        return grouped

    @classmethod
    async def job_ids_by_emails(cls, session: AsyncSession, emails) -> dict:
        """{email: {job_post_id, ...}} for any prior ATS score of these people."""
        from g_sheet.models import FormData

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return {}
        grouped: dict[str, set[str]] = {}

        def _add(email, job_id):
            key = (email or "").strip().lower()
            if not key or job_id is None:
                return
            grouped.setdefault(key, set()).add(str(job_id))

        inbox_rows = await session.execute(
            select(func.lower(Inbox_Messages.message_from), cls.job_post_id)
            .join(Inbox, cls.inbox_id == Inbox.id)
            .join(Inbox_Messages, Inbox.message_id == Inbox_Messages.id)
            .where(func.lower(Inbox_Messages.message_from).in_(lowers))
            .where(cls.job_post_id.is_not(None))
        )
        for email, job_id in inbox_rows.all():
            _add(email, job_id)

        form_rows = await session.execute(
            select(func.lower(FormData.candidate_email), cls.job_post_id)
            .join(FormData, cls.form_data_id == FormData.id)
            .where(func.lower(FormData.candidate_email).in_(lowers))
            .where(cls.job_post_id.is_not(None))
        )
        for email, job_id in form_rows.all():
            _add(email, job_id)

        user_rows = await session.execute(
            select(func.lower(Users.email), cls.job_post_id)
            .join(Users, cls.user_id == Users.id)
            .where(func.lower(Users.email).in_(lowers))
            .where(cls.job_post_id.is_not(None))
        )
        for email, job_id in user_rows.all():
            _add(email, job_id)
        return grouped

    @classmethod
    async def job_ids_for_messages(cls, session: AsyncSession, message_ids) -> dict:
        """{message_id: {job_post_id, ...}} scored for these inbox_messages rows."""
        keys = []
        for raw in message_ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return {}
        result = await session.execute(
            select(Inbox.message_id, cls.job_post_id)
            .join(Inbox, cls.inbox_id == Inbox.id)
            .where(Inbox.message_id.in_(keys))
            .where(cls.job_post_id.is_not(None))
        )
        grouped: dict[str, set[str]] = {}
        for mid, job_id in result.all():
            if mid is None or job_id is None:
                continue
            grouped.setdefault(str(mid), set()).add(str(job_id))
        return grouped

    @classmethod
    async def job_ids_for_forms(cls, session: AsyncSession, form_ids) -> dict:
        """{form_data_id: {job_post_id, ...}} scored for these Sheet Forms rows."""
        keys = []
        for raw in form_ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return {}
        result = await session.execute(
            select(cls.form_data_id, cls.job_post_id)
            .where(cls.form_data_id.in_(keys))
            .where(cls.job_post_id.is_not(None))
        )
        grouped: dict[str, set[str]] = {}
        for fid, job_id in result.all():
            if fid is None or job_id is None:
                continue
            grouped.setdefault(str(fid), set()).add(str(job_id))
        return grouped

    @classmethod
    async def resolve_identity(cls, session: AsyncSession, email, candidate_id):
        """XOR identity for a score row from the scored candidate's email.

        Matching users.email (case-insensitive) -> user_id, candidate_id NULL.
        Missing/blank email or no user -> candidate_id, user_id NULL.
        """
        normalized = (email or "").strip().lower()
        if normalized:
            user_id = (
                await session.execute(
                    select(Users.id).where(func.lower(Users.email) == normalized)
                )
            ).scalar_one_or_none()
            if user_id is not None:
                return {"candidate_id": None, "user_id": user_id}
        return {"candidate_id": candidate_id, "user_id": None}

    @classmethod
    async def get_current_for_candidate(cls, session: AsyncSession, candidate_id):
        """Current row for an upload-sourced score, chained per candidates row —
        stable across re-scores because upsert_candidate keeps the same id for
        the same (job, content_sha256)."""
        result = await session.execute(
            select(cls)
            .where(cls.candidate_id == candidate_id, cls.is_current == True)  # noqa: E712
            .order_by(cls.computed_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_current_for_user(cls, session: AsyncSession, user_id, job_post_id=None):
        """Current upload-sourced score for a matched user, scoped per job."""
        uid = cls._as_uuid(user_id)
        if uid is None:
            return None
        qry = select(cls).where(cls.user_id == uid, cls.is_current == True)  # noqa: E712
        jid = cls._as_uuid(job_post_id) if job_post_id is not None else None
        if jid is not None:
            qry = qry.where(cls.job_post_id == jid)
        result = await session.execute(qry.order_by(cls.computed_at.desc()))
        return result.scalars().first()

    @classmethod
    async def insert_result(cls, session: AsyncSession, fields: dict):
        """Insert a score row and supersede the previous current one.

        Inbox scores chain on inbox_id and repoint inbox.ats_id (candidate_id
        may be NULL). Upload scores chain on candidate_id, or on (user_id,
        job_post_id) when identity resolved to a user. Sheet Forms scores chain
        on (form_data_id, job_post_id) with inbox_id and user_id left NULL.
        Flush the INSERT first: with no relationship() edge the unit of work
        emits the UPDATEs first, and the FKs reject a pointer to a row not yet
        inserted."""
        inbox_id = fields.get("inbox_id")
        candidate_id = fields.get("candidate_id")
        user_id = fields.get("user_id")
        job_post_id = fields.get("job_post_id")
        form_data_id = fields.get("form_data_id")
        if inbox_id is not None:
            prev = await cls.get_current_for_inbox(session, inbox_id)
        elif form_data_id is not None:
            prev = await cls.get_current_for_form_job(session, form_data_id, job_post_id)
        elif candidate_id is not None:
            prev = await cls.get_current_for_candidate(session, candidate_id)
        elif user_id is not None:
            prev = await cls.get_current_for_user(session, user_id, job_post_id)
        else:
            prev = None
        row = cls(**fields)
        session.add(row)
        await session.flush()
        if prev is not None:
            prev.is_current = False
            prev.superseded_by_id = row.id
            session.add(prev)
        if inbox_id is not None:
            link = await Inbox.get_inbox_by_id(session, inbox_id)
            if link is not None:
                link.ats_id = row.id
                link.updated_at = _now()
                session.add(link)
        await session.commit()
        return row

    @classmethod
    async def list_by_rescan_run(cls, session: AsyncSession, rescan_run_id):
        uid = cls._as_uuid(rescan_run_id)
        if uid is None:
            return []
        result = await session.execute(
            select(cls).where(cls.rescan_run_id == uid).order_by(cls.computed_at.desc())
        )
        return list(result.scalars().all())


class MailboxSyncRun(SQLModel, table=True):
    """One Outlook mailbox sync job — survives tab close because work runs in Taskiq.

    The Sync button enqueues a run and returns immediately. The UI polls this row for
    status / per-message entries / triage totals. Closing the browser does not cancel
    the worker.
    """

    __tablename__ = "mailbox_sync_runs"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    status: str = Field(default="queued", index=True)  # queued|running|completed|failed
    task_id: str | None = Field(default=None)
    created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    top: int = Field(default=100)
    skip: int = Field(default=0)
    test_on: bool = Field(default=True)
    triage: dict | None = Field(default=None, sa_column=Column(JSONB))
    entries: list | None = Field(default=None, sa_column=Column(JSONB))
    error: str | None = Field(default=None)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        if record_id in (None, ""):
            return None
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_by_id(cls, session: AsyncSession, record_id):
        uid = cls._as_uuid(record_id)
        if uid is None:
            return None
        result = await session.execute(select(cls).where(cls.id == uid))
        return result.scalars().first()

    @classmethod
    async def get_active(cls, session: AsyncSession):
        result = await session.execute(
            select(cls)
            .where(cls.status.in_(("queued", "running")))
            .order_by(cls.created_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_latest(cls, session: AsyncSession):
        result = await session.execute(
            select(cls).order_by(cls.created_at.desc()).limit(1)
        )
        return result.scalars().first()

    @classmethod
    async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True):
        row = cls(**fields)
        session.add(row)
        if commit:
            await session.commit()
            await session.refresh(row)
        return row

    @classmethod
    async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True):
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        session.add(row)
        if commit:
            await session.commit()
            await session.refresh(row)
        return row


class InboxRescanRun(SQLModel, table=True):
    """On-Hold catalogue ATS rescan — one row the Inbox ReScan button polls."""

    __tablename__ = "inbox_rescan_runs"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    status: str = Field(default="queued", index=True)  # queued|running|scoring|completed|failed
    channel: str = Field(default="all")
    sheet: str | None = Field(default=None)
    task_id: str | None = Field(default=None)
    created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    job_count: int = Field(default=0)
    candidate_count: int = Field(default=0)
    skipped_candidates: int = Field(default=0)
    skipped_pairs: int = Field(default=0)
    pair_count: int = Field(default=0)
    done_count: int = Field(default=0)
    entries: list | None = Field(default=None, sa_column=Column(JSONB))
    summaries: list = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"})
    error: str | None = Field(default=None)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        if record_id in (None, ""):
            return None
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_by_id(cls, session: AsyncSession, record_id):
        uid = cls._as_uuid(record_id)
        if uid is None:
            return None
        result = await session.execute(select(cls).where(cls.id == uid))
        return result.scalars().first()

    @classmethod
    async def get_active(cls, session: AsyncSession):
        result = await session.execute(
            select(cls)
            .where(cls.status.in_(("queued", "running", "scoring")))
            .order_by(cls.created_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def get_latest(cls, session: AsyncSession):
        result = await session.execute(
            select(cls).order_by(cls.created_at.desc()).limit(1)
        )
        return result.scalars().first()

    @classmethod
    async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True):
        row = cls(**fields)
        session.add(row)
        if commit:
            await session.commit()
            await session.refresh(row)
        return row

    @classmethod
    async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True):
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        session.add(row)
        if commit:
            await session.commit()
            await session.refresh(row)
        return row

    @classmethod
    async def append_summary(cls, session: AsyncSession, record_id, entry):
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        items = list(row.summaries or [])
        items.append(entry)
        row.summaries = items
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row