import asyncio
import logging
import uuid
import httpx,os
from fastapi import HTTPException
from inbox.enums import Candidate_application_Status
from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun,Inbox,SourceChannels,AtsResults
from inbox.file_decoder import extract_pdf_attachments
from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run, serialize_ats_result
from inbox.plugins import (
    EMAIL_API_TOKEN,
    attach_email_pdfs_to_s3,
    fetch_message_read_status,
    load_message_files,
    request_email_confirmation,
    send_mail,
)
from inbox.semaphore import GraphSemaphore

from inbox_classifier.decorators import is_manual_upload,triage_fields
from inbox_classifier.execute_agent import classify_email
from inbox_classifier.plugins import (
    TRIAGE_CONCURRENCY,
    TRIAGE_ENABLED,
    TRIAGE_STATUSES,
    sender_domain,
    should_ingest,
    triage_model_name,
)
from dotenv import load_dotenv
load_dotenv()
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime,timezone

logger=logging.getLogger("inbox.match")
triage_logger=logging.getLogger("inbox.triage")

# One statement, one round trip — but an unbounded id list is still a client-supplied
# IN () of arbitrary size, so the batch is capped and the route answers 413.
MAX_BULK_READ_IDS=100


class Email:
    def __init__(self,session:AsyncSession,token=None):
        self.session=session
        self.get_url=os.getenv("EMAIL_URL")
        self.token=token or EMAIL_API_TOKEN
        self.graph_slots=GraphSemaphore(concurrency=4,max_retries=5)
        self.pending_match_ids:list[str]=[]
        self.pending_confirmation_emails:list[str]=[]
        # Upstream ids the intake gate judged not to be job applications. They get a
        # verdict row and no inbox_messages row.
        self.skipped_message_ids:list[str]=[]
        self.triage_errors:list[str]=[]

    # async def get_all_applications(self,app_id=None):
    #     try:
    #         if app_id:
    #             application_lst=await Inbox_Messages.get_all_applications(self.session,message_id=app_id)
    #         else:
    #             application_lst=await Inbox_Messages.get_all_applications(self.session)
    #         return application_lst
    #     except Exception as e:
    #         raise HTTPException(status_code=500,detail=str(e))

    async def service_email(self,top,skip):
        async with httpx.AsyncClient() as client:
            try:
                response=await self.graph_slots.get(
                    client,f"{self.get_url}/emails",
                    params={"skip":skip,"top":top},
                    headers={"Authorization":f"Bearer {self.token}"},
                )
                if response.status_code==200:
                    return response.json()
                else:
                    raise HTTPException(status_code=response.status_code,detail=response.text)
            except HTTPException:
                raise
            except Exception as e:
                raise HTTPException(status_code=500,detail=str(e))

    async def fetch_message(self,message_id):
        """GET /emails/{id} on the upstream Email API -> the Graph payload."""
        async with httpx.AsyncClient() as client:
            response=await self.graph_slots.get(
                client,f"{self.get_url}/emails/{message_id}",
                headers={"Authorization":f"Bearer {self.token}"},
            )
            if response.status_code!=200:
                raise HTTPException(status_code=response.status_code,detail=response.text)
            return response.json()


    async def triage_round(self,message_ids):
        """Fetch and classify a whole /email/fetch page, bounded by a semaphore."""
        
        ids=[str(m) for m in message_ids or [] if m]
        decisions={}
        if not ids:
            return decisions
        known=await Inbox_Messages.existing_message_ids(self.session,ids)
        recorded=await Inbox_Message_Triage.verdicts_for_message_ids(self.session,ids)
        pending=[]
        for message_id in ids:
            if message_id in known:
                decisions[message_id]={"ingest":True,"status":"known","fresh":False}
            elif message_id in recorded:
                decisions[message_id]={"ingest":recorded[message_id],"status":"recorded","fresh":False}
            else:
                pending.append(message_id)
        if not pending:
            triage_logger.info("triage round: page=%s known=%s classified=0",len(ids),len(decisions))
            return decisions

        semaphore=asyncio.Semaphore(TRIAGE_CONCURRENCY)

        async def run(message_id):
            async with semaphore:
                data=await self.fetch_message(message_id)
                if not TRIAGE_ENABLED or is_manual_upload(data):
                    return message_id,{"data":data,"ingest":True,"status":"disabled","fresh":False}
                verdict,error=await classify_email(data)
                ingest,status,reason=should_ingest(verdict,error)
                return message_id,{"data":data,"verdict":verdict,"error":error,"ingest":ingest,
                                   "status":status,"reason":reason,"fresh":True}

        results=await asyncio.gather(*(run(m) for m in pending),return_exceptions=True)
        accepted=rejected=errors=0
        for result in results:
            if isinstance(result,BaseException):
                # First failure wins, preserving today's all-or-nothing behaviour for a
                # failing upstream message. Never catch BaseException itself: a
                # CancelledError must keep propagating.
                raise result
            message_id,decision=result
            decisions[message_id]=decision
            if decision.get("status")=="error":
                errors+=1
            if decision.get("ingest"):
                accepted+=1
            else:
                rejected+=1
        triage_logger.info(
            "triage round: page=%s known=%s classified=%s accepted=%s rejected=%s errors=%s",
            len(ids),len(known)+len(recorded),len(pending),accepted,rejected,errors,
        )
        return decisions

    async def record_triage(self,data,decision,ingested):
        """Persist one verdict. Never raises into the ingestion path.

        A failed audit write must not cost us a candidate: the worst case is that the
        next fetch re-classifies this message.
        """
        try:
            fields=triage_fields(
                data,
                decision.get("verdict"),
                decision.get("status") or "classified",
                decision.get("reason") or "",
                error=decision.get("error") or "",
                model_name=triage_model_name(),
                ingested=ingested,
            )
            await Inbox_Message_Triage.record_verdict(self.session,fields)
            # Allowlisted keys only: sender DOMAIN not address, attachment COUNT not
            # names, no subject or body text, no evidence text.
            verdict=decision.get("verdict")
            triage_logger.info(
                "triage %s: application=%s reason=%s confidence=%s domain=%s attachments=%s",
                fields["message_id"],
                fields["is_application"],
                fields["reason_code"],
                getattr(verdict,"confidence",None),
                sender_domain(fields["message_from"]),
                len(data.get("attachments") or []),
            )
        except Exception as e:
            triage_logger.warning("triage record failed: %s",type(e).__name__)

    async def get_email_by_id(self,message_id,test_on=True,decision=None):
        """Persist one upstream message, gated by the application classifier.

        `decision` is the pre-computed verdict from triage_round; without one this
        classifies inline, so a single-message call still works.

        Ordering is deliberate. The verdict comes BEFORE PDF extract / S3 upload: a
        rejected mail must not create a candidate Users row or queue confirmation mail.
        Flow: extract PDF bytes in memory → insert inbox_messages → link sender →
        upload Email/{id}/{user_id}/file.pdf → store permanent S3 URL on file_path.
        If S3 fails on a brand-new row, the table entry is deleted (atomicity).
        """
        try:
            if decision is None:
                decision=(await self.triage_round([message_id])).get(str(message_id)) or {}
            data=decision.get("data") or await self.fetch_message(message_id)

            if not decision.get("ingest"):
                if decision.get("fresh"):
                    await self.record_triage(data,decision,ingested=False)
                self.skipped_message_ids.append(str(message_id))
                if decision.get("status")=="error":
                    self.triage_errors.append(str(message_id))
                return {"message_id":str(message_id),"skipped":"not_application",
                        "reason":decision.get("reason") or "","status":decision.get("status") or ""}

            upstream_id=data.get("id")
            already=await Inbox_Messages.get_by_upstream_id(self.session,upstream_id) if upstream_id else None
            pdfs=extract_pdf_attachments(data.get("attachments"))
            # Insert first (no file_path yet) so S3 keys can use the table PK.
            row,new_user_email=await Inbox_Messages.insert_email(
                session=self.session,email_data=data,file_path=None,
            )
            await Reapplied(session=self.session).sync_for_email(row.message_from)
            if pdfs:
                row=await attach_email_pdfs_to_s3(
                    self.session,row,pdfs,created_new=(already is None),
                )
            if decision.get("fresh"):
                await self.record_triage(data,decision,ingested=True)
            if row.attachment and row.file_path and row.match_status is None:
                self.pending_match_ids.append(str(row.id))
            if test_on:
                return data
            if new_user_email:
                self.pending_confirmation_emails.append(new_user_email)
            return data
        except HTTPException:
            # Was missing: the bare `except Exception` below caught the upstream-status
            # HTTPException and re-raised every one of them as a 500.
            raise
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    async def get_inbox_messages(self,top,skip,search=None):
        messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search)
        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
        items=[]
        for m in messages:
            item=serialize_message(m,linkedin_url=urls.get(m.id))
            files=load_message_files(m)
            if files:
                item["files"]=files
            items.append(item)
        return items

    async def get_inbox_message_by_id(self,record_id):
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
        item=serialize_message(message,linkedin_url=urls.get(message.id))
        files=load_message_files(message)
        if files:
            item["files"]=files
        from job.candidate.views import CandidateView
        cv=CandidateView(session=self.session)
        suggested=[]
        for job_id in item.get("suggested_job_post_ids") or []:
            jp=await cv.get_job_post_by_id(record_id=job_id)
            if jp:
                if jp.get("is_deleted") or not jp.get("is_active"):
                    suggested.append({**jp,"unavailable":True})
                else:
                    suggested.append(jp)
            else:
                suggested.append({"id":str(job_id),"unavailable":True})
        item["suggested_job_posts"]=suggested
        assigned_id=item.get("assigned_job_post_id")
        if assigned_id:
            item["assigned_job_post"]=await cv.get_job_post_by_id(record_id=assigned_id)
        else:
            item["assigned_job_post"]=None
        items=await self._paint_inbox_ats([item])
        return await cv.attach_application_history(items[0])

    async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,city=None,source=None):
        if application_status in (Candidate_application_Status.PROCESS, Candidate_application_Status.REJECTED, Candidate_application_Status.SCREENING, Candidate_application_Status.ASSESSMENT, Candidate_application_Status.INTERVIEW, Candidate_application_Status.OFFER, Candidate_application_Status.HIRED) or processing_state:
            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
        elif isread==False:
            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
        else:
            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
        items=[serialize_application(m,linkedin_url=urls.get(m.id),light=True) for m in messages]
        items=await self._attach_job_posts(items)
        from job.candidate.views import CandidateView
        return await CandidateView(session=self.session).attach_application_history(items)

    async def _attach_job_posts(self,items):
        """List payload needs assigned + suggested job objects — export reads titles.

        Detail hydrates one row; the queue used to ship ids only. Assigned job and
        Suggested jobs in the Inbox .xlsx were then blank for email applicants.
        """
        ids=[]
        for item in items:
            aid=item.get("assigned_job_post_id")
            if aid:
                ids.append(aid)
            for sid in item.get("suggested_job_post_ids") or []:
                if sid:
                    ids.append(sid)
        by_id={}
        if ids:
            from job.job_post.models import JobPosts
            from job.job_post.serializers import serialize_job_post
            for post in await JobPosts.get_by_ids(self.session,ids,active_only=False):
                payload=serialize_job_post(post)
                if post.is_deleted or not post.is_active:
                    payload={**payload,"unavailable":True}
                by_id[str(post.id)]=payload
        for item in items:
            aid=item.get("assigned_job_post_id")
            item["assigned_job_post"]=by_id.get(str(aid)) if aid else None
            suggested=[]
            for sid in item.get("suggested_job_post_ids") or []:
                if not sid:
                    continue
                payload=by_id.get(str(sid))
                if payload is None:
                    suggested.append({"id":str(sid),"unavailable":True})
                else:
                    suggested.append(dict(payload))
            item["suggested_job_posts"]=suggested
        items=await self._paint_inbox_ats(items)
        return items

    async def _paint_inbox_ats(self,items):
        """Attach per-job ATS scores onto suggested/assigned posts and stamp max.

        Unassigned rows show the highest suggestion score; assigned rows show
        the score against assigned_job_post_id — same rule as Sheet Forms.
        """
        if not items:
            return items
        by_msg=await AtsResults.get_latest_by_job_for_messages(
            self.session,[item.get("id") for item in items],
        )
        for item in items:
            rows=by_msg.get(str(item.get("id") or "")) or []
            scores=[serialize_ats_result(r) for r in rows]
            item["ats_results"]=scores
            score_by_job={
                str(s["job_post_id"]):s for s in scores if s.get("job_post_id")
            }
            for post in item.get("suggested_job_posts") or []:
                hit=score_by_job.get(str(post.get("id")))
                if hit:
                    post["overall_score"]=hit.get("overall_score")
                    post["band"]=hit.get("band")
            assigned=item.get("assigned_job_post")
            if assigned:
                hit=score_by_job.get(str(assigned.get("id")))
                if hit:
                    assigned["overall_score"]=hit.get("overall_score")
                    assigned["band"]=hit.get("band")
            aid=item.get("assigned_job_post_id")
            assigned_score=score_by_job.get(str(aid)) if aid else None
            if assigned_score and assigned_score.get("overall_score") is not None:
                item["ats_score"]=round(float(assigned_score.get("overall_score")))
            else:
                nums=[s.get("overall_score") for s in scores if s.get("overall_score") is not None]
                if nums:
                    item["ats_score"]=round(float(max(nums)))
        return items

    async def get_application_by_id(self,record_id):
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Application not found")
        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
        item=serialize_application(message,linkedin_url=urls.get(message.id))
        from job.candidate.views import CandidateView
        return await CandidateView(session=self.session).attach_application_history(item)

    async def queue_rematch(self,record_id):
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        if not message.attachment or not message.file_path:
            raise HTTPException(status_code=400,detail="Message has no attachment to match")
        return str(message.id)

    async def enqueue_matching(self,inbox_ids,force=False):
        from inbox.tasks import match_inbox_message
        task_ids=[]
        for record_id in inbox_ids or []:
            created_at=datetime.now(timezone.utc).isoformat()
            task=await match_inbox_message.kicker().with_labels(
                created_at=created_at,
                correlation_id=str(record_id),
                queue="inbox",
            ).kiq(str(record_id),force=force)
            task_ids.append(task.task_id)
        return task_ids

    async def start_mailbox_sync(self,current_user=None,top=100,skip=0,test_on=True):
        """Enqueue Outlook pull on the mailbox_sync queue; return the run row.

        If a queued/running sync already exists, return it instead of stacking another.
        """
        active=await MailboxSyncRun.get_active(self.session)
        if active:
            return serialize_mailbox_sync_run(active)

        created_by=None
        if isinstance(current_user,dict) and current_user.get("id"):
            created_by=MailboxSyncRun._as_uuid(current_user.get("id"))

        row=await MailboxSyncRun.insert_run(self.session,{
            "status":"queued",
            "created_by":created_by,
            "top":int(top or 100),
            "skip":int(skip or 0),
            "test_on":bool(test_on) if test_on is not None else True,
        })

        from inbox.mailbox_sync_tasks import sync_mailbox
        task=await sync_mailbox.kicker().with_labels(
            created_at=datetime.now(timezone.utc).isoformat(),
            correlation_id=str(row.id),
            queue="mailbox_sync",
        ).kiq(str(row.id))
        row=await MailboxSyncRun.update_run(self.session,row.id,{"task_id":task.task_id})
        return serialize_mailbox_sync_run(row)

    async def get_mailbox_sync(self,run_id=None):
        if run_id:
            row=await MailboxSyncRun.get_by_id(self.session,run_id)
            if not row:
                raise HTTPException(status_code=404,detail="Sync run not found")
            return serialize_mailbox_sync_run(row)
        row=await MailboxSyncRun.get_active(self.session)
        if row:
            return serialize_mailbox_sync_run(row)
        # Latest finished run so the UI can still show the last result after refresh.
        row=await MailboxSyncRun.get_latest(self.session)
        if not row:
            raise HTTPException(status_code=404,detail="No sync runs yet")
        return serialize_mailbox_sync_run(row)

    async def run_mailbox_sync_page(self,top=100,skip=0,test_on=True,on_progress=None):
        """Pull one Outlook page, triage, ingest, enqueue matching. Returns summary.

        Shared by the legacy synchronous /email/fetch and the background Taskiq worker.
        `on_progress(processed, expected, entries)` is optional; the mailbox_sync
        worker uses it so the Sync button can poll a live percentage.
        """
        data=await self.service_email(top,skip)
        value=data.get("value") or []
        expected=len(value)
        if on_progress:
            await on_progress(0,expected,[])
        decisions=await self.triage_round([item.get("id") for item in value])
        entries=[]
        for item in value:
            message_id=item.get("id")
            decision=decisions.get(str(message_id)) or {}
            try:
                result=await self.get_email_by_id(message_id,test_on,decision=decision)
                if isinstance(result,dict) and result.get("skipped"):
                    entries.append({
                        "message_id":str(message_id),
                        "status":"skipped",
                        "reason":result.get("reason") or "",
                        "triage_status":result.get("status") or "",
                    })
                else:
                    # Prefer the decision status (known/recorded/…) when present.
                    status=decision.get("status") or "ingested"
                    if status in ("known","recorded","disabled"):
                        entry_status="known" if status=="known" else "ingested"
                    else:
                        entry_status="ingested"
                    entries.append({
                        "message_id":str(message_id),
                        "status":entry_status,
                        "reason":decision.get("reason") or "",
                        "triage_status":status,
                    })
            except Exception as e:
                entries.append({
                    "message_id":str(message_id),
                    "status":"error",
                    "reason":str(e),
                    "triage_status":"error",
                })
                self.triage_errors.append(str(message_id))
            if on_progress:
                await on_progress(len(entries),expected,entries)

        if self.pending_match_ids:
            await self.enqueue_matching(list(self.pending_match_ids),force=False)

        account_setup=[]
        if not test_on and self.pending_confirmation_emails:
            account_setup=await self.send_account_setup(list(self.pending_confirmation_emails))

        skipped=len(self.skipped_message_ids)
        ingested=sum(1 for e in entries if e.get("status") in ("ingested","known"))
        triage={
            "ingested":ingested,
            "skipped":skipped,
            "errors":len(self.triage_errors),
            "total":len(entries),
        }
        return {"entries":entries,"triage":triage,"account_setup":account_setup}

    async def send_account_setup(self,emails):
        results=[]
        for email in emails or []:
            try:
                status=await request_email_confirmation(email)
                results.append({"email":email,"sent":status==200})
            except Exception as e:
                logger.warning("confirmation request failed for %s: %s",email,e)
                results.append({"email":email,"sent":False})
        return results

    async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,city=None,source=None):
        if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED or processing_state:
            return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)
        elif isread==False:
            return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)
        else:
            return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)

    async def list_cities(self):
        """Distinct cities from inbox_messages and form_data, merged in Python."""
        from g_sheet.models import FormData
        inbox=await Inbox_Messages.distinct_cities(self.session)
        forms=await FormData.distinct_cities(self.session)
        return Reapplied(session=self.session).merge_cities(inbox,forms)

    async def list_sources(self):
        """Source / platform labels: seeded channels, Google Sheet, form sources."""
        from g_sheet.models import FormData
        channels=await SourceChannels.list_active(self.session)
        forms=await FormData.distinct_sources(self.session)
        return Reapplied(session=self.session).merge_cities(
            [c.label for c in channels],["Google Sheet"],forms,
        )

    async def assign_job_post(self,record_id,job_post_id):
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        if job_post_id is not None:
            from job.job_post.models import JobPosts
            post=await JobPosts.get_job_post_by_id(self.session,job_post_id)
            if not post or post.is_deleted or not post.is_active:
                raise HTTPException(status_code=422,detail="Job post is missing, deleted, or inactive")
        updated=await Inbox_Messages.set_assigned_job_post(self.session,record_id,job_post_id)
        if not updated:
            raise HTTPException(status_code=404,detail="Message not found")
        await Reapplied(session=self.session).sync_for_email(updated.message_from)
        if job_post_id is not None:
            # Assignment pairs this CV with a JD we already have — queue the ATS
            # score in the background so the recruiter is not held on an OpenAI
            # call. Idempotent server-side; broker-down just logs (the profile's
            # Score-with-ATS button remains the manual fallback).
            from inbox.tasks import score_inbox_message
            try:
                await score_inbox_message.kicker().with_labels(
                    created_at=datetime.now(timezone.utc).isoformat(),
                    correlation_id=str(record_id),
                    queue="inbox",
                ).kiq(str(record_id),str(job_post_id))
            except Exception as exc:
                logger.warning("could not queue ats score for %s: %s",record_id,exc)
        return await self.get_inbox_message_by_id(record_id)

    async def mark_read(self,record_id,read=True):
        message=await Inbox_Messages.mark_message_read(self.session,record_id,read)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        return serialize_message(message)

    async def set_read_bulk(self,record_ids,read):
        """Flip read state for a hand-picked selection.

        Returns counts, never rows: a 500-id selection would otherwise serialize 500
        full messages back at a client that only needs to know it worked.

        `updated` < `requested` means some ids no longer exist — a stale selection
        against a list that moved. That is reported, not raised, because the rows that
        DID exist were already committed.
        """
        ids=[str(r).strip() for r in (record_ids or []) if str(r or "").strip()]
        if not ids:
            raise HTTPException(status_code=422,detail="record_ids must contain at least one id")
        if len(ids)>MAX_BULK_READ_IDS:
            raise HTTPException(status_code=413,
                                detail=f"At most {MAX_BULK_READ_IDS} ids per request")
        updated=await Inbox_Messages.set_read_bulk(self.session,ids,read)
        logger.info("bulk read: requested=%s updated=%s read=%s",len(ids),updated,bool(read))
        return {"requested":len(ids),"updated":updated,"read":bool(read)}

    async def set_read_all(self,read,search=None,isread:bool=True,
                           application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,
                           assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,
                           city=None,source=None):
        """Mark every row the SAME filter set would have listed.

        The filter arguments are the caller's current view, not a free-form query: the
        button says "mark all read in this view" and the WHERE chain is literally the
        list's own (Inbox_Messages._apply_filters), so the two cannot drift.
        """
        updated=await Inbox_Messages.set_read_scope(
            self.session,read,search=search,isread=isread,
            application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,
            no_suggestions=no_suggestions,processing_state=processing_state,
            city=city,source=source,
        )
        logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s dup=%s",
                    updated,bool(read),isread,assigned,getattr(application_status,"value",application_status),is_duplicate)
        return {"updated":updated,"read":bool(read)}

    async def refresh_read_status(self,record_id):
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        if not message.message_id:
            raise HTTPException(status_code=400,detail="Message has no upstream id")
        try:
            status=await fetch_message_read_status(message.message_id,token=self.token)
        except httpx.HTTPStatusError as e:
            raise HTTPException(status_code=e.response.status_code,detail=e.response.text)
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))
        if status is None:
            raise HTTPException(status_code=404,detail="Message not found upstream")
        await Inbox_Messages.apply_read_status(self.session,[status])
        refreshed=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        return serialize_message(refreshed)

    async def get_counts(self):
        return await Inbox_Messages.count_processing(self.session)

    async def set_processing_state(self,record_id,processing_state,current_user=None):
        allowed=("unread","imported","processed","rejected")
        if processing_state not in allowed:
            raise HTTPException(status_code=422,detail=f"processing_state must be one of {', '.join(allowed)}")
        if processing_state=="processed":
            existing=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
            if not existing:
                raise HTTPException(status_code=404,detail="Message not found")
            if not existing.assigned_job_post_id:
                raise HTTPException(status_code=422,detail="Assign a job post before moving to shortlist")
            # Record CLOSED/PROCESS → PENDING so the board history matches the card.
            current=existing.application_status
            current_val=current.value if isinstance(current,Candidate_application_Status) else str(current or "")
            if current_val in ("","CLOSED","PROCESS"):
                link=await Inbox.get_inbox_by_message_id(self.session,existing.id)
                if link is not None:
                    from job.pipeline.views import Pipeline
                    try:
                        await Pipeline(self.session).change_stage(
                            Candidate_application_Status.PENDING.value,
                            current_user,
                            inbox_id=link.id,
                            change_reason="Moved to shortlist from inbox",
                        )
                    except HTTPException as exc:
                        if exc.status_code!=400:
                            raise
        message=await Inbox_Messages.set_processing_state(self.session,record_id,processing_state)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        return serialize_application(message)

    async def set_duplicate(self,record_id,is_duplicate):
        if not isinstance(is_duplicate,bool):
            raise HTTPException(status_code=422,detail="is_duplicate must be a boolean")
        message=await Inbox_Messages.set_duplicate(self.session,record_id,is_duplicate)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        return serialize_application(message)

    async def get_triage_messages(self,top,skip,search=None,is_application=None,status=None):
        """The intake gate's verdict log — mostly the mail that never became a row.

        A hard gate's only real risk is the silent false negative, so the rejections
        have to be reviewable.
        """
        if status is not None and status not in TRIAGE_STATUSES:
            raise HTTPException(status_code=422,detail=f"status must be one of {', '.join(TRIAGE_STATUSES)}")
        rows=await Inbox_Message_Triage.list_triage(self.session,top,skip,is_application,status,search)
        return [serialize_triage(row) for row in rows]

    async def count_triage(self,search=None,is_application=None,status=None):
        return await Inbox_Message_Triage.count_triage(self.session,is_application,status,search)

    async def override_triage(self,record_id,is_application,current_user=None):
        """Overturn a verdict a recruiter disagrees with.

        false -> true re-fetches the mail from upstream and runs the normal ingestion
        path, which is why the body was never stored.

        true -> false does NOT delete the inbox_messages row: inbox, ats_results,
        assessments, notifications and application_stage_transitions all reference it,
        so a purge would take candidate accounts and scores with it. It moves the row to
        processing_state 'rejected' instead, an already-allowlisted value.
        """
        if not isinstance(is_application,bool):
            raise HTTPException(status_code=422,detail="is_application must be a boolean")
        row=await Inbox_Message_Triage.get_triage_by_id(self.session,record_id)
        if not row:
            raise HTTPException(status_code=404,detail="Triage record not found")

        user_id=(current_user or {}).get("id")
        if is_application and not row.ingested:
            data=await self.fetch_message(row.message_id)
            already=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id)
            pdfs=extract_pdf_attachments(data.get("attachments"))
            message,new_user_email=await Inbox_Messages.insert_email(
                session=self.session,email_data=data,file_path=None,
            )
            await Reapplied(session=self.session).sync_for_email(message.message_from)
            if pdfs:
                message=await attach_email_pdfs_to_s3(
                    self.session,message,pdfs,created_new=(already is None),
                )
            await Inbox_Message_Triage.mark_ingested(self.session,row.message_id,True)
            if message.attachment and message.file_path and message.match_status is None:
                await self.enqueue_matching([str(message.id)],force=False)
            if new_user_email:
                await self.send_account_setup([new_user_email])
        elif not is_application and row.ingested:
            message=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id)
            if message:
                await Inbox_Messages.set_processing_state(self.session,message.id,"rejected")

        updated=await Inbox_Message_Triage.set_override(self.session,record_id,is_application,user_id)
        return serialize_triage(updated)

    async def send_email(self,payload,current_user):
        to_email=(payload.get("to") or "").strip()
        subject=(payload.get("subject") or "").strip()
        body=payload.get("body") or ""
        content_type=(payload.get("content_type") or "html").strip() or "html"
        if not to_email:
            raise HTTPException(status_code=422,detail="to is required")
        if not subject:
            raise HTTPException(status_code=422,detail="subject is required")
        if not body:
            raise HTTPException(status_code=422,detail="body is required")
        try:
            await send_mail(to_email,subject,body,content_type=content_type)
        except (httpx.HTTPError,RuntimeError) as e:
            raise HTTPException(status_code=502,detail="Failed to send email") from e
        inbox_id=payload.get("inbox_id")
        job_post_id=None
        try:
            from notifications.models import Notifications
            uid=None
            raw=current_user.get("id") if current_user else None
            if raw:
                uid=uuid.UUID(str(raw))
            if uid:
                await Notifications.insert_notification(self.session,{
                    "user_id":uid,
                    "kind":"message",
                    "title":"Email sent",
                    "body":f"Sent “{subject}” to {to_email}",
                    "link_path":"/inbox",
                    "inbox_id":int(inbox_id) if inbox_id is not None else None,
                    "job_post_id":job_post_id,
                })
        except Exception as exc:
            logger.warning("notification insert skipped: %s",exc)
        return {"accepted":True,"to":to_email,"subject":subject}

    async def reply_email(self,payload,current_user):
        record_id=payload.get("record_id")
        body=payload.get("body") or ""
        if not record_id:
            raise HTTPException(status_code=422,detail="record_id is required")
        if not str(body).strip():
            raise HTTPException(status_code=422,detail="body is required")
        message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
        if not message:
            raise HTTPException(status_code=404,detail="Message not found")
        to_email=(message.message_from or "").strip()
        if not to_email:
            raise HTTPException(status_code=422,detail="Message has no sender address")
        original=(message.message_subject or "").strip()
        subject=original if original.lower().startswith("re:") else f"Re: {original}" if original else "Re:"
        try:
            await send_mail(to_email,subject,body,content_type="html")
        except (httpx.HTTPError,RuntimeError) as e:
            raise HTTPException(status_code=502,detail="Failed to send email") from e
        try:
            from notifications.models import Notifications
            uid=None
            raw=current_user.get("id") if current_user else None
            if raw:
                uid=uuid.UUID(str(raw))
            if uid:
                await Notifications.insert_notification(self.session,{
                    "user_id":uid,
                    "kind":"message",
                    "title":"Email sent",
                    "body":f"Replied to {to_email}",
                    "link_path":"/inbox",
                })
        except Exception as exc:
            logger.warning("notification insert skipped: %s",exc)
        return {"accepted":True,"to":to_email,"subject":subject}


class Reapplied:
    def __init__(self,session:AsyncSession):
        self.session=session
        self.collected={}
        self.seen={}

    def _norm_email(self,email):
        return (email or "").strip().lower()

    def _as_job_id(self,value):
        if value in (None,""):
            return None
        text=str(value).strip()
        return text or None

    def _add(self,email,job_id):
        uid=self._as_job_id(job_id)
        if not uid or email not in self.seen or uid in self.seen[email]:
            return
        self.seen[email].add(uid)
        self.collected[email].append(uid)

    def merge_cities(self,*groups):
        """Case-insensitive unique cities, first spelling wins, sorted."""
        seen=set()
        out=[]
        for group in groups:
            for raw in group or []:
                text=(raw or "").strip()
                if not text:
                    continue
                key=text.lower()
                if key in seen:
                    continue
                seen.add(key)
                out.append(text)
        out.sort(key=str.lower)
        return out

    async def sync_for_email(self,email):
        stamped=await self.sync_for_emails([email])
        return stamped.get(self._norm_email(email),[])

    async def sync_for_emails(self,emails):
        """Collect linked job_post_ids for these emails and stamp reapplied on all 3 tables.

        Records with no job_post_id are ignored while collecting. Empty collections
        leave reapplied untouched.
        """
        from g_sheet.models import FormData
        from job.candidate.models import Manual_UPLOAD_CANDIDATE
        from users.models import Users

        lowers=sorted({self._norm_email(e) for e in (emails or []) if self._norm_email(e)})
        if not lowers:
            return {}
        self.collected={email:[] for email in lowers}
        self.seen={email:set() for email in lowers}

        for email,job_id in await Manual_UPLOAD_CANDIDATE.job_post_ids_by_emails(self.session,lowers):
            self._add(email,job_id)
        for email,job_id in await FormData.job_post_ids_by_emails(self.session,lowers):
            self._add(email,job_id)
        for email,job_id in await Inbox_Messages.assigned_job_post_ids_by_emails(self.session,lowers):
            self._add(email,job_id)

        stamped={email:ids for email,ids in self.collected.items() if ids}
        if not stamped:
            return {}
        await Users.set_reapplied_by_emails(self.session,stamped)
        await Manual_UPLOAD_CANDIDATE.set_reapplied_by_emails(self.session,stamped)
        await FormData.set_reapplied_by_emails(self.session,stamped)
        return stamped