From f1758ac9bd2ef89e10ba3e8cf89fd953a720ef11 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Thu, 20 Aug 2026 17:55:17 +0500 Subject: [PATCH] Add mailbox sync functionality and related API endpoints - Introduced a new `MailboxSyncRun` model to manage mailbox synchronization jobs, allowing for tracking of sync status and results. - Implemented API endpoints for starting mailbox sync and fetching sync status, enabling asynchronous processing of email synchronization. - Updated the `Email` service to handle mailbox sync operations, including enqueueing tasks and managing sync results. - Enhanced frontend components to initiate mailbox sync and display sync status, improving user experience and feedback during the sync process. - Added necessary query keys and local storage management for tracking sync runs in the UI. This commit enhances the application's email synchronization capabilities, providing a more robust and user-friendly experience for managing mailbox data. --- backend/inbox/app.py | 94 ++++--- backend/inbox/models.py | 104 +++++++- backend/inbox/serializers.py | 18 ++ backend/inbox/views.py | 137 +++++++++- docker-compose.dev.yml | 5 + docker-compose.yml | 19 ++ frontend/src/api/inbox.js | 25 +- frontend/src/lib/queryKeys.js | 1 + frontend/src/screens/Inbox.jsx | 463 +++++++-------------------------- frontend/src/ui/DataTable.jsx | 2 +- 10 files changed, 450 insertions(+), 418 deletions(-) diff --git a/backend/inbox/app.py b/backend/inbox/app.py index df08030..28df178 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -47,6 +47,7 @@ class ReadAllBody(BaseModel): isread: bool = True application_status: Candidate_application_Status = Candidate_application_Status.CLOSED assigned: bool | None = None + is_duplicate: bool | None = None class TriageOverrideBody(BaseModel): @@ -73,36 +74,67 @@ async def fetch_email( token: str | None = Query(None), session: AsyncSession = Depends(get_session), ): + """Synchronous one-shot sync — kept for scripts/compat. UI uses POST /email/sync.""" try: service=Email(session=session,token=token) if not service.token: raise HTTPException(status_code=401,detail="Unauthorized") - data=await service.service_email(top,skip) - value=data.get("value") - items_lst=[] - # Classify the whole page first, bounded-parallel, then replay it in upstream - # order: the inserts stay serial on the one request session and pending_match_ids - # keeps the sequence it has today. - decisions=await service.triage_round([item.get("id") for item in value]) - for item in value: - message_id=item.get("id") - service_per_email=await service.get_email_by_id(message_id,test_on,decision=decisions.get(str(message_id))) - items_lst.append({"message_id":message_id,"email_contents":service_per_email}) - - if service.pending_match_ids: - await service.enqueue_matching(list(service.pending_match_ids),force=False) - - skipped=len(service.skipped_message_ids) - triage={"ingested":len(items_lst)-skipped,"skipped":skipped,"errors":len(service.triage_errors)} - - account_setup=[] + summary=await service.run_mailbox_sync_page(top=top,skip=skip,test_on=test_on) if test_on: - return JSONResponse(content={"data":items_lst,"triage":triage,"status_code":200}) - if service.pending_confirmation_emails: - account_setup=await service.send_account_setup(list(service.pending_confirmation_emails)) + return JSONResponse(content={ + "data":summary["entries"], + "triage":summary["triage"], + "status_code":200, + }) + return JSONResponse(content={ + "data":summary["entries"], + "account_setup":summary["account_setup"], + "triage":summary["triage"], + "status_code":200, + }) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) - return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"triage":triage,"status_code":200}) - + +@router.post("/email/sync") +async def start_email_sync( + top:int=Query(100), + skip:int=Query(0,ge=0), + test_on: bool = Query(True), + current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)), + session: AsyncSession = Depends(get_session), +): + """Enqueue mailbox sync on the dedicated mailbox_sync Taskiq queue. + + Returns immediately with a run id. Poll GET /email/sync/fetch until completed. + Closing the browser does not cancel the worker. + """ + try: + service=Email(session=session) + if not service.token: + raise HTTPException(status_code=401,detail="Unauthorized") + data=await service.start_mailbox_sync( + current_user=current_user,top=top,skip=skip,test_on=test_on, + ) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.get("/email/sync/fetch") +async def fetch_email_sync( + run_id: str | None = Query(None), + current_user: dict = Depends(require_permission(PermissionTag.INBOX_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=Email(session=session) + data=await service.get_mailbox_sync(run_id=run_id) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise except Exception as e: @@ -218,6 +250,7 @@ async def mark_all_inbox_read( isread=payload.isread, application_status=payload.application_status, assigned=payload.assigned, + is_duplicate=payload.is_duplicate, ) return JSONResponse(content={"data":data,"total":data["updated"],"status_code":200}) except HTTPException: @@ -248,6 +281,7 @@ async def get_all_applications( application_status: Candidate_application_Status = Query(default=Candidate_application_Status.CLOSED), isread: bool = Query(default=True), assigned: bool | None = Query(default=None), + is_duplicate: bool | None = Query(default=None), search: str | None = Query(None), top: int | None = Query(None), skip: int = Query(0, ge=0), @@ -258,19 +292,19 @@ async def get_all_applications( service=Email(session=session) 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): - items=await service.get_all_applications(top, skip, search, application_status=application_status, assigned=assigned) - total=await service.count_inbox_messages(search, application_status=application_status, assigned=assigned) + items=await service.get_all_applications(top, skip, search, application_status=application_status, assigned=assigned, is_duplicate=is_duplicate) + total=await service.count_inbox_messages(search, application_status=application_status, assigned=assigned, is_duplicate=is_duplicate) return JSONResponse(content={"data":items,"total":total,"status_code":200}) if isread==False: - items=await service.get_all_applications(top, skip, search, isread=False, assigned=assigned) - total=await service.count_inbox_messages(search, isread=False, assigned=assigned) + items=await service.get_all_applications(top, skip, search, isread=False, assigned=assigned, is_duplicate=is_duplicate) + total=await service.count_inbox_messages(search, isread=False, assigned=assigned, is_duplicate=is_duplicate) return JSONResponse(content={"data":items,"total":total,"status_code":200}) if record_id: item=await service.get_application_by_id(record_id) return JSONResponse(content={"data":item,"total":1,"status_code":200}) - items=await service.get_all_applications(top,skip,search,assigned=assigned) - total=await service.count_inbox_messages(search,assigned=assigned) + items=await service.get_all_applications(top,skip,search,assigned=assigned,is_duplicate=is_duplicate) + total=await service.count_inbox_messages(search,assigned=assigned,is_duplicate=is_duplicate) return JSONResponse(content={"data":items,"total":total,"status_code":200}) except HTTPException: raise diff --git a/backend/inbox/models.py b/backend/inbox/models.py index ea1d579..758d0c5 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -353,6 +353,13 @@ class Inbox_Messages(SQLModel, table=True): 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"}) + # 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 @@ -577,6 +584,7 @@ class Inbox_Messages(SQLModel, table=True): 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, ): """The one WHERE chain shared by the list, the count and the bulk read UPDATE. @@ -595,15 +603,21 @@ class Inbox_Messages(SQLModel, table=True): 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 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 + 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 ): + # Page size is the caller's `top` (Inbox sends 10); `skip` is (page-1)*top + # so page 1 -> 0..9, page 2 -> 10..19. Newest first via created_at. statement = cls._apply_filters( - select(cls).order_by(cls.message_received_time.desc()), - search, isread, application_status, assigned, + select(cls).order_by(cls.created_at.desc()), + search, isread, application_status, assigned, is_duplicate, ) if skip: statement = statement.offset(skip) @@ -654,10 +668,10 @@ class Inbox_Messages(SQLModel, table=True): return row @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): + 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): statement = cls._apply_filters( select(func.count()).select_from(cls), - search, isread, application_status, assigned, + search, isread, application_status, assigned, is_duplicate, ) result = await session.execute(statement) return result.scalar_one() @@ -760,6 +774,7 @@ class Inbox_Messages(SQLModel, table=True): 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, ) -> int: """Mark every row matching a list filter. Returns rows actually CHANGED. @@ -768,7 +783,7 @@ class Inbox_Messages(SQLModel, table=True): 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) + statement=cls._apply_filters(update(cls),search,isread,application_status,assigned,is_duplicate) statement=statement.where(cls.message_read!=bool(read)) result=await session.execute( statement.values(message_read=bool(read),read_overridden_at=_now()) @@ -782,8 +797,8 @@ class Inbox_Messages(SQLModel, table=True): 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.application_status == Candidate_application_Status.PROCESS, 1), else_=0)), 0).label("processed"), + func.coalesce(func.sum(case((cls.application_status == Candidate_application_Status.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.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"), @@ -1148,3 +1163,76 @@ class AtsResults(SQLModel, table=True): session.add(link) await session.commit() return row + + +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 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 diff --git a/backend/inbox/serializers.py b/backend/inbox/serializers.py index 6d368df..01c41a3 100644 --- a/backend/inbox/serializers.py +++ b/backend/inbox/serializers.py @@ -157,3 +157,21 @@ def serialize_triage(row: Inbox_Message_Triage) -> dict: "overridden_at": row.overridden_at.isoformat() if row.overridden_at else None, "classified_at": row.classified_at.isoformat() if row.classified_at else None, } + + +def serialize_mailbox_sync_run(row) -> dict: + return { + "id": str(row.id), + "status": row.status, + "task_id": row.task_id, + "created_by": str(row.created_by) if row.created_by else None, + "top": row.top, + "skip": row.skip, + "test_on": row.test_on, + "triage": row.triage, + "entries": row.entries or [], + "error": row.error, + "created_at": row.created_at.isoformat() if row.created_at else None, + "started_at": row.started_at.isoformat() if row.started_at else None, + "finished_at": row.finished_at.isoformat() if row.finished_at else None, + } diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 3a5bce8..43f83b5 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -4,9 +4,9 @@ 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 +from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun from inbox.file_decoder import decode_attachment -from inbox.serializers import serialize_application, serialize_message, serialize_triage +from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run from inbox.plugins import ( EMAIL_API_TOKEN, fetch_message_read_status, @@ -270,13 +270,13 @@ class Email: item["assigned_job_post"]=None return item - async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None): + 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): 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): - messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned) + messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate) elif isread==False: - messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned) + messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned,is_duplicate=is_duplicate) else: - messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned) + messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned,is_duplicate=is_duplicate) return [serialize_application(m) for m in messages] async def get_application_by_id(self,record_id): @@ -306,6 +306,115 @@ class Email: 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. + from sqlmodel import select + result=await self.session.execute( + select(MailboxSyncRun).order_by(MailboxSyncRun.created_at.desc()).limit(1) + ) + row=result.scalars().first() + 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): + """Pull one Outlook page, triage, ingest, enqueue matching. Returns summary. + + Shared by the legacy synchronous /email/fetch and the background Taskiq worker. + """ + data=await self.service_email(top,skip) + value=data.get("value") or [] + 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 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 []: @@ -317,13 +426,13 @@ class Email: 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): + 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): if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED: - return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned) + return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate) elif isread==False: - return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned) + return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned,is_duplicate=is_duplicate) else: - return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned) + return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned,is_duplicate=is_duplicate) async def assign_job_post(self,record_id,job_post_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) @@ -381,7 +490,7 @@ class Email: async def set_read_all(self,read,search=None,isread:bool=True, application_status:Candidate_application_Status=Candidate_application_Status.CLOSED, - assigned=None): + assigned=None,is_duplicate=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 @@ -390,10 +499,10 @@ class Email: """ updated=await Inbox_Messages.set_read_scope( self.session,read,search=search,isread=isread, - application_status=application_status,assigned=assigned, + application_status=application_status,assigned=assigned,is_duplicate=is_duplicate, ) - logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s", - updated,bool(read),isread,assigned,getattr(application_status,"value",application_status)) + 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): diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 9286ca9..64730d3 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -58,3 +58,8 @@ services: volumes: - ./backend:/app - ./app:/app/app + + taskiq-mailbox-sync-worker: + volumes: + - ./backend:/app + - ./app:/app/app diff --git a/docker-compose.yml b/docker-compose.yml index 346226b..745e7f2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -231,6 +231,25 @@ services: <<: *backend-env TASKIQ_CV_QUEUE_NAME: cv_upload + # Dedicated stream: Outlook pull/triage must not block match/ATS or CV uploads. + taskiq-mailbox-sync-worker: + <<: *backend-service + container_name: hrms-taskiq-mailbox-sync-worker + command: + [ + "taskiq", + "worker", + "taskiq_management.mailbox_sync_broker_setup:mailbox_sync_broker", + "inbox.mailbox_sync_tasks", + "--workers", + "1", + ] + environment: + <<: *backend-env + TASKIQ_MAILBOX_SYNC_QUEUE_NAME: mailbox_sync + TASKIQ_WORKER_NAME: mailbox-sync-worker-01 + volumes: *attachments + # --- Postgres: DEFINED HERE, NOT STARTED BY DEFAULT ------------------------------- # The profile is what keeps it out of `docker compose build` and `docker compose up`. # Nothing about the default stack changes by its presence in this file. diff --git a/frontend/src/api/inbox.js b/frontend/src/api/inbox.js index 7366f23..d8a3acc 100644 --- a/frontend/src/api/inbox.js +++ b/frontend/src/api/inbox.js @@ -20,13 +20,14 @@ export function listMessages() { * `assigned` is tri-valued: omit for no filter, true for rows with an * assigned_job_post_id, false for the Job Matching queue. */ -export function listApplications({ search, top, skip, recordId, isread, applicationStatus, assigned } = {}) { +export function listApplications({ search, top, skip, recordId, isread, applicationStatus, assigned, isDuplicate } = {}) { return request('/inbox/all-applications', { // `isread` is tri-valued on the wire: omit it for every tab (server defaults // to true = no filter), send false for the Unread tab only. buildUrl drops // undefined but keeps false, so `isread: undefined` sends no param at all. // Same for `application_status`: omit for every tab (server defaults to // CLOSED = no filter), send PROCESS / REJECTED for those tabs only. + // Same for `is_duplicate`: omit unless the Duplicates tab. params: { search, top, @@ -35,6 +36,7 @@ export function listApplications({ search, top, skip, recordId, isread, applicat isread, application_status: applicationStatus, assigned, + is_duplicate: isDuplicate, }, }) } @@ -51,7 +53,23 @@ export function getMessage(recordId) { return request('/inbox/fetch', { params: { record_id: recordId } }) } -/** Triggers the Graph proxy to pull new mail and persist it. */ +/** Enqueue Outlook pull on the mailbox_sync worker. Returns a run immediately. */ +export function startMailboxSync({ top, skip, testOn } = {}) { + return request('/email/sync', { + method: 'POST', + params: { top, skip, test_on: testOn }, + }) +} + +/** Poll one sync run (or the active/latest run when runId is omitted). */ +export function getMailboxSync(runId) { + return request('/email/sync/fetch', { params: { run_id: runId } }) +} + +/** + * Legacy synchronous sync — blocks until the page is ingested. Prefer + * startMailboxSync + getMailboxSync so closing the tab cannot kill the job. + */ export function syncMailbox({ token, top, skip } = {}) { return request('/email/fetch', { params: { token, top, skip } }) } @@ -96,7 +114,7 @@ export function bulkSetRead(recordIds, read) { * Resolves to `{updated, read}`, where `updated` counts rows that actually * CHANGED state, so it is safe to show in a toast. */ -export function setReadAll({ read, search, isread, applicationStatus, assigned } = {}) { +export function setReadAll({ read, search, isread, applicationStatus, assigned, isDuplicate } = {}) { return request('/inbox/read-all', { method: 'PATCH', body: { @@ -105,6 +123,7 @@ export function setReadAll({ read, search, isread, applicationStatus, assigned } isread, application_status: applicationStatus, assigned, + is_duplicate: isDuplicate, }, }) } diff --git a/frontend/src/lib/queryKeys.js b/frontend/src/lib/queryKeys.js index b1195d7..d23bba4 100644 --- a/frontend/src/lib/queryKeys.js +++ b/frontend/src/lib/queryKeys.js @@ -27,6 +27,7 @@ export const qk = { // qk.mailbox.all() already refreshes the intake gate's ledger, so an // override or a sync needs no extra invalidation. triage: (p = {}) => ['mailbox', 'triage', p], + sync: (id) => ['mailbox', 'sync', id], }, assessments: { all: () => ['assessments'], diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index a32133a..f634776 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -1,12 +1,6 @@ /* ============================================================ - Recruitment Inbox — application tabs plus the Email tab, which is the - app's oldest real network call (GET /inbox/fetch, previously the only fetch - in the entire prototype). - - The email body used to be interpolated raw into markup at js/inbox.js:292 — - the single widest XSS sink in the repository, and the one that mattered most - because inbound mail is attacker-supplied by definition. It renders as text - now, which is the structural fix. + Recruitment Inbox — application tabs over GET /inbox/all-applications. + Page size is 10, newest first (order by created_at on the server). ============================================================ */ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' @@ -16,6 +10,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import Modal from '../ui/Modal' import EmailBody, { looksLikeHtml } from '../ui/EmailBody' import { Tabs } from '../ui/Tabs' +import { Pagination, pageWindow } from '../ui/DataTable' import { Avatar, Badge, EmptyState, Icon, ScoreChip } from '../ui/primitives' import { JobCard, PickRoleModal } from '../ui/SuggestedRoles' import { useToast } from '../ui/Toast' @@ -29,33 +24,22 @@ import { inboxSources, sourceMeta, } from '../data/seed' -const TABS = ['All Applications', 'Unread', 'Processed', 'Rejected', 'Duplicates', 'Email'] +const TABS = ['All Applications', 'Unread', 'Processed', 'Rejected', 'Duplicates'] +const PAGE_SIZE = 10 /** - * Tabs that do not read /inbox/all-applications at all. Email reads - * /inbox/fetch; every other tab is a view over the applications list. - */ -const SPECIAL_TABS = new Set(['Email']) - -/** - * Server-side filters for tabs /inbox/all-applications can narrow. - * Processed / Rejected / Duplicates filter client-side on - * processing_state + is_duplicate (see GET /inbox/counts). + * Server-side filters for each tab. Processed / Rejected use + * Candidate_application_Status (PROCESS / REJECTED), not processing_state. */ const TAB_FILTERS = { Unread: { isread: false }, + Processed: { applicationStatus: 'PROCESS' }, + Rejected: { applicationStatus: 'REJECTED' }, + Duplicates: { isDuplicate: true }, } -/** - * Tabs whose visible set is EXACTLY what the server returns for TAB_FILTERS[tab]. - * - * Only these may use the scope endpoint for "mark all". Processed / Rejected / - * Duplicates narrow client-side over rows the server already handed back in full, - * so a scope call from one of those carries no such predicate and would mark the - * entire mailbox — rows the user never saw, with no undo. Those tabs send the - * visible ids instead. - */ -const SERVER_SCOPED_TABS = new Set(['All Applications', 'Unread']) +/** Every tab above is a true server-side scope — safe for "mark all". */ +const SERVER_SCOPED_TABS = new Set(TABS) /** * message_received_time / message_sent_time are plain string columns @@ -205,38 +189,43 @@ async function fetchMessageDetail(recordId) { * * `processing` prefers processing_state (imported/processed/rejected); otherwise * Read/Unread from message_read. Duplicate comes from is_duplicate. + * Returns `{ rows, total }` so the pager can show page 1 / 2 / 3… without + * loading the whole mailbox. */ async function fetchApplications(params) { const res = await inboxApi.listApplications(params) const rows = Array.isArray(res?.data) ? res.data : [] - return rows.map((row) => { - const name = row.name || row.email || 'Unknown' - return { - id: String(row.id), - name, - initials: initialsOf(name), - color: avatarColor(name), - email: row.email || '', - position: row.position || '(no subject)', - ...sourceFrom(row.source), - received: parseDate(row.received), - unread: Boolean(row.unread), - processing: row.processing || 'Unread', - processingState: row.processing_state || null, - applicationStatus: row.application_status || null, - resumeStatus: row.resume_status || 'Pending', - attachment: row.attachment, - hasAttachment: Boolean(row.has_attachment), - resumeText: row.resume_text || '', - atsScore: row.ats_score, - phone: row.phone, - experience: row.experience, - recruiter: row.recruiter, - duplicate: Boolean(row.duplicate), - suggestedIds: (row.suggested_job_post_ids || []).map(String), - assignedId: row.assigned_job_post_id ? String(row.assigned_job_post_id) : null, - } - }) + return { + rows: rows.map((row) => { + const name = row.name || row.email || 'Unknown' + return { + id: String(row.id), + name, + initials: initialsOf(name), + color: avatarColor(name), + email: row.email || '', + position: row.position || '(no subject)', + ...sourceFrom(row.source), + received: parseDate(row.received), + unread: Boolean(row.unread), + processing: row.processing || 'Unread', + processingState: row.processing_state || null, + applicationStatus: row.application_status || null, + resumeStatus: row.resume_status || 'Pending', + attachment: row.attachment, + hasAttachment: Boolean(row.has_attachment), + resumeText: row.resume_text || '', + atsScore: row.ats_score, + phone: row.phone, + experience: row.experience, + recruiter: row.recruiter, + duplicate: Boolean(row.duplicate), + suggestedIds: (row.suggested_job_post_ids || []).map(String), + assignedId: row.assigned_job_post_id ? String(row.assigned_job_post_id) : null, + } + }), + total: Number(res?.total ?? rows.length) || 0, + } } // Shared with the sidebar badge through qk.mailbox.counts — see inboxApi.fetchCounts. @@ -258,9 +247,10 @@ async function setReadChunked(ids, read) { } /** - * Rewrites ONE cache entry for a read-state flip: a row list, or the - * single-message detail object. Anything else (the counts object) passes through - * untouched — it has no `id`, and its delta is applied separately. + * Rewrites ONE cache entry for a read-state flip: a row list, a paginated + * `{rows, total}` page, or the single-message detail object. Anything else + * (the counts object) passes through untouched — it has no `id`, and its + * delta is applied separately. */ function patchReadState(data, idSet, read) { const patchRow = (r) => (idSet.has(r.id) @@ -275,6 +265,7 @@ function patchReadState(data, idSet, read) { } : r) if (Array.isArray(data)) return data.map(patchRow) + if (data && Array.isArray(data.rows)) return { ...data, rows: data.rows.map(patchRow) } if (data && typeof data === 'object' && data.id && idSet.has(data.id)) return patchRow(data) return data } @@ -304,8 +295,9 @@ async function optimisticRead(qc, ids, read) { // A Set, because the same id appears in several cached lists at once. const flipping = new Set() for (const [, data] of previous) { - if (!Array.isArray(data)) continue - for (const r of data) if (idSet.has(r.id) && Boolean(r.unread) === read) flipping.add(r.id) + const rows = Array.isArray(data) ? data : data?.rows + if (!Array.isArray(rows)) continue + for (const r of rows) if (idSet.has(r.id) && Boolean(r.unread) === read) flipping.add(r.id) } qc.setQueriesData({ queryKey: qk.mailbox.all() }, (data) => patchReadState(data, idSet, read)) @@ -559,60 +551,37 @@ export default function Inbox() { const updateInbox = useSeedMutation('inbox') const [tab, setTab] = useState('All Applications') + const [page, setPage] = useState(1) const [selectedId, setSelectedId] = useState(null) const [q, setQ] = useState('') const [previewing, setPreviewing] = useState(null) const [assigning, setAssigning] = useState(null) const [noting, setNoting] = useState(null) - // Tabs with a server-side filter pass their params; everything else (and - // countsQuery) passes `{}` so the backend defaults mean "no filter". const tabFilter = TAB_FILTERS[tab] ?? {} + const listParams = useMemo(() => ({ + ...tabFilter, + top: PAGE_SIZE, + skip: (page - 1) * PAGE_SIZE, + ...(q.trim() ? { search: q.trim() } : {}), + }), [tabFilter, page, q]) const applicationsQuery = useQuery({ - queryKey: qk.mailbox.applications(tabFilter), - queryFn: () => fetchApplications(tabFilter), - enabled: !SPECIAL_TABS.has(tab), + queryKey: qk.mailbox.applications(listParams), + queryFn: () => fetchApplications(listParams), }) const countsQuery = useQuery({ queryKey: qk.mailbox.counts(), queryFn: fetchInboxCounts, - enabled: tab !== 'Email', }) - const inbox = applicationsQuery.data ?? [] + const inbox = applicationsQuery.data?.rows ?? [] + const total = applicationsQuery.data?.total ?? 0 + const pages = Math.max(1, Math.ceil(total / PAGE_SIZE)) + const currentPage = Math.min(page, pages) const serverCounts = countsQuery.data ?? {} - const emailsQuery = useQuery({ - queryKey: qk.mailbox.messages(), - queryFn: async () => { - const res = await inboxApi.listMessages() - const rows = Array.isArray(res?.data) ? res.data : [] - return rows.map((row) => ({ - id: String(row.id), - from: row.sender_name || row.fromEmail || 'Unknown', - fromEmail: row.fromEmail || '', - subject: row.subject || '', - body: row.body || '', - when: parseDate(row.when) ?? parseDate(row.message_sent_time), - unread: Boolean(row.unread), - attachment: row.attachment_name || 'Resume.pdf', - attachmentSize: '—', - // The agent's verdict, straight off backend/inbox/serializers.py:44-48. - // suggested_job_post_ids is deliberately NOT carried: job posts stay - // dark to the inbox. - matchStatus: row.match_status || null, - matchSummary: row.match_summary || '', - matchReasoning: row.match_reasoning || '', - matchError: row.match_error || '', - matchedAt: parseDate(row.matched_at), - imported: false, - })) - }, - enabled: tab === 'Email', - }) - const counts = useMemo( () => ({ 'All Applications': serverCounts.all ?? 0, @@ -620,28 +589,16 @@ export default function Inbox() { Processed: serverCounts.processed ?? 0, Rejected: serverCounts.rejected ?? 0, Duplicates: serverCounts.duplicates ?? 0, - Email: (emailsQuery.data ?? []).filter((e) => e.unread).length, }), - [serverCounts, emailsQuery.data], + [serverCounts], ) - const list = useMemo(() => { - let l = inbox - if (tab === 'Unread') l = l.filter((i) => i.processing === 'Unread') - else if (tab === 'Processed') l = l.filter((i) => i.processing === 'Processed') - else if (tab === 'Rejected') l = l.filter((i) => i.processing === 'Rejected') - else if (tab === 'Duplicates') l = l.filter((i) => i.duplicate) - if (q) l = l.filter((i) => (i.name + i.position + i.source).toLowerCase().includes(q.toLowerCase())) - return l - }, [inbox, tab, q]) + const list = inbox - // Clicking a row fetches that one record from /inbox/fetch. The list row is - // kept as the base and the detail is overlaid, so the pane paints instantly - // from cached list data and fills in body/attachments when the fetch lands. const detailQuery = useQuery({ queryKey: qk.mailbox.message(selectedId), queryFn: () => fetchMessageDetail(selectedId), - enabled: !SPECIAL_TABS.has(tab) && Boolean(selectedId), + enabled: Boolean(selectedId), }) const selectedRow = inbox.find((i) => i.id === selectedId) @@ -677,15 +634,13 @@ export default function Inbox() { } function setReadEverything(read) { - // The scope endpoint may only be used where the server-side filter IS the - // view. The search box narrows client-side on name/position/source while the - // server's `search` matches subject/from/body, and three of the tabs narrow - // client-side entirely — handing either to a WHERE clause would mark rows - // that were never on screen. Those cases send the visible ids instead, which - // is exact and, since the list endpoint is unpaginated, complete. - if (SERVER_SCOPED_TABS.has(tab) && !q.trim()) { - // `ids` drives the optimistic patch only; the write itself is the filter. - setReadAll.mutate({ read, filter: tabFilter, ids: list.map((i) => i.id) }) + // Search goes on the wire now, so every tab is a true server scope. + if (SERVER_SCOPED_TABS.has(tab)) { + setReadAll.mutate({ + read, + filter: { ...tabFilter, ...(q.trim() ? { search: q.trim() } : {}) }, + ids: list.map((i) => i.id), + }) return } const ids = list.map((i) => i.id) @@ -791,15 +746,17 @@ export default function Inbox() {
{ setTab(t); setSelectedId(null); selection.clear() }} + onChange={(t) => { + setTab(t) + setPage(1) + setSelectedId(null) + selection.clear() + }} tabs={TABS.map((t) => ({ key: t, label: t, count: counts[t] }))} />
- {tab === 'Email' ? ( - - ) : ( -
+
{applicationsQuery.isSuccess && (
- setQ(e.target.value)} placeholder="Search applications…" /> + { setQ(e.target.value); setPage(1) }} + placeholder="Search applications…" + />
@@ -823,7 +784,7 @@ export default function Inbox() { Fetching applications from the server. )} {applicationsQuery.isError && ( - + {friendlyAuthError(applicationsQuery.error, 'Request failed')} )} @@ -861,9 +822,6 @@ export default function Inbox() {
{outlookListTime(i.received)}
- {/* No ATS score exists server-side — the agent returns a - verdict, not a number. The chip stays off rather than - rendering a placeholder that reads as a real score. */} {i.atsScore != null && (
)} @@ -872,6 +830,17 @@ export default function Inbox() { )) )}
+ {applicationsQuery.isSuccess && total > 0 && ( + { setPage(p); setSelectedId(null); selection.clear() }} + pageButtons={pageWindow(currentPage, pages)} + /> + )}
@@ -883,7 +852,7 @@ export default function Inbox() {
) : detailQuery.isError ? (
- + {friendlyAuthError(detailQuery.error, 'Request failed')}
@@ -904,7 +873,6 @@ export default function Inbox() { )}
- )} {previewing && ( @@ -1368,232 +1336,3 @@ function AssignRecruiter({ item, recruiters, onClose, onSave }) { ) } - -/** The live tab: real fetch, real loading state, real error state. */ -function EmailTab({ query, toast }) { - const qc = useQueryClient() - const { can } = useAuth() - const canEdit = can('inbox.edit') - const [selectedId, setSelectedId] = useState(null) - const [replying, setReplying] = useState(null) - const [replyBody, setReplyBody] = useState('') - const [importedIds, setImportedIds] = useState(() => new Set()) - - const emails = query.data ?? [] - const selected = emails.find((e) => e.id === selectedId) - const unread = emails.filter((e) => e.unread).length - - const selection = useRowSelection(emails) - const setRead = useSetRead(toast) - const setReadAll = useSetReadAll(toast) - - function setReadSelected(read) { - const ids = [...selection.selectedIds] - if (!ids.length) return - setRead.mutate({ ids, read }, { - onSuccess: () => { - toast(`${ids.length} ${ids.length === 1 ? 'message' : 'messages'} marked ${read ? 'read' : 'unread'}`, 'success') - selection.clear() - }, - }) - } - - /** - * This list is /inbox/fetch with no params — every persisted message, no - * filter, no pagination — so the empty scope really is the whole mailbox and - * the WHERE clause matches what is on screen exactly. - */ - function setReadEverything(read) { - setReadAll.mutate({ read, filter: {}, ids: emails.map((e) => e.id) }) - } - - const importMsg = useMutation({ - mutationFn: (id) => inboxApi.setProcessingState(id, 'imported'), - onSuccess: (_d, id) => { - setImportedIds((s) => new Set(s).add(id)) - toast('Marked as imported', 'success') - }, - onError: (err) => toast(friendlyAuthError(err, 'Could not import.'), 'error'), - onSettled: () => { - qc.invalidateQueries({ queryKey: qk.mailbox.all() }) - qc.invalidateQueries({ queryKey: qk.mailbox.counts() }) - }, - }) - - const reply = useMutation({ - mutationFn: ({ recordId, body }) => inboxApi.replyEmail({ recordId, body }), - onSuccess: () => { - toast('Reply sent', 'success') - setReplying(null) - setReplyBody('') - }, - onError: (err) => toast(friendlyAuthError(err, 'Could not send reply.'), 'error'), - }) - - function selectEmail(e) { - setSelectedId(e.id) - if (e.unread) setRead.mutate({ ids: [e.id], read: true }) - } - - const isImported = (e) => importedIds.has(e.id) - - return ( - <> -
- Outlook · Microsoft Graph API - - {query.isPending ? 'Loading…' : query.isError ? 'Failed to load' : `${emails.length} messages · ${unread} unread`} - -
- -
-
- {query.isSuccess && emails.length > 0 && ( - - )} - {query.isPending && Fetching mailbox from the server.} - {query.isError && ( - - {friendlyAuthError(query.error, 'Request failed')} - - )} - {query.isSuccess && emails.length === 0 && ( - No emails in the mailbox. - )} - {query.isSuccess && emails.map((e) => ( -
selectEmail(e)} - > - selection.toggle(e.id)} - label={`Select mail from ${e.from}`} - /> - -
-
{e.from}
-
{e.subject}
-
- - Outlook - - {isImported(e) && Imported} -
-
-
{outlookListTime(e.when)}
-
- ))} -
- -
- {!selected ? ( -
- - Preview email body and resume attachments here. - -
- ) : ( -
-
- -
-
{selected.from}
-
- {selected.fromEmail} · {selected.when ? fmtDate(selected.when) : 'Date unavailable'} -
-
-
- - {/* Same Subject-strip + framed-body template as /matching. It replaces - the old

subject rather than sitting under it — two subject - lines on one panel is worse than none. The Imported/New badge - moves into the strip, which is where a mail client puts status. */} -
-
- Subject: {selected.subject || '(no subject)'} - {isImported(selected) ? Imported : New} -
- {looksLikeHtml(selected.body) ? ( - - ) : ( - 
-                    {htmlToText(selected.body) || 'This email has no message body.'}
-
- )} -
- -
- -
-
{selected.attachment}
-
{selected.attachmentSize} · PDF
-
-
- -
- {isImported(selected) ? ( - - ) : ( - - )} - -
-

- )} -
-
- - {replying && ( - setReplying(null)} - footer={ - <> - - - - } - > -
- - -
-
- -