Background saving terminated

Sync_read.md

@@ -0,0 +1,228 @@
+# Read-status sync (`/sync/*`)
+
+Tracks **which messages got read or unread** — and which were deleted — without
+re-downloading the mailbox. It sits on Microsoft Graph's **delta query**: Graph
+hands you a cursor, and every later call with that cursor returns *only* what
+changed since it was issued.
+
+Five of the six endpoints share one piece of state: a delta cursor per
+**(signed-in user + folder)**, persisted to disk so a restart doesn't re-backfill
+the whole folder. The sixth — the per-message lookup — is deliberately outside
+that machinery: it reads one id live and touches no cursor.
+
+| Method | Path | Purpose |
+| ------ | ---- | ------- |
+| GET | `/sync/read-status` | Run one sync round **now** (synchronous) |
+| GET | `/sync/read-status/changes` | Replay the last round's **full** result |
+| GET | `/sync/read-status/message/{id}` | One message's status, by id — cursor-free |
+| GET | `/sync/read-status/status` | Watcher health + cursor state |
+| POST | `/sync/read-status/watch` | Start the background poller |
+| DELETE | `/sync/read-status/watch` | Stop the background poller |
+
+All require the API bearer token, and act on the **signed-in user's** mailbox —
+they answer `401` until device-code sign-in completes.
+
+---
+
+## `GET /sync/read-status`
+
+The workhorse. Asks Graph "what changed in this folder since my cursor?", emits
+the changes, and advances the cursor.
+
+The **first** call has no cursor, so it backfills the entire folder — an Inbox
+with 4,700 messages is 47 pages of 100. Every call after that is incremental and
+usually near-empty.
+
+| Param | Default | Meaning |
+| ----- | ------- | ------- |
+| `folder` | `inbox` | Well-known name (`inbox`, `sentitems`, …) or folder id. Graph delta is **folder-scoped** — there is no all-mail delta |
+| `since` | – | ISO8601 lower bound, **initial sync only** (`receivedDateTime ge …`). The way to keep a first backfill small |
+| `reset` | `false` | Discard the saved cursor and start a fresh baseline |
+| `max_pages` | `10` | Cap on Graph pages (100 msgs each) fetched **per call** |
+| `limit` | `10` | Cap on messages returned **in this response** |
+
+`max_pages` and `limit` are independent and easy to confuse:
+
+- **`max_pages` bounds the work.** Hit the cap and the call returns
+  `complete: false`, having saved its position; the next call resumes exactly
+  where it stopped. No changes are skipped, and no cursor is written until the
+  backfill genuinely finishes.
+- **`limit` only trims the JSON.** It has no effect on how much is fetched.
+  `count` stays the true total, and the untruncated set is on
+  `/sync/read-status/changes`.
+
+```jsonc
+{
+  "synced_at": "2026-08-07T10:15:00Z",
+  "folder": "inbox",
+  "count": 1000,            // changed messages this call actually fetched
+  "removed_count": 0,       // deleted / moved out of the folder
+  "initial_sync": true,     // this round is part of the first backfill
+  "complete": false,        // hit max_pages — call again to continue
+  "pages": 10,              // Graph pages fetched by this call
+  "truncated": true,        // limit cut the lists below
+  "value":   [ { "id": "AAMk…", "isRead": true,
+                 "lastModifiedDateTime": "2026-08-07T10:14:52Z",
+                 "subject": "Invoice #421" } ],
+  "removed": [ { "id": "AAMk…", "reason": "deleted" } ]
+}
+```
+
+`value` is sorted newest-modified first before `limit` is applied, so a
+truncated response shows the most recent changes rather than an arbitrary slice.
+Only the four `$select` fields above come back — this endpoint is about *status*,
+not content; use `GET /emails/{id}` for bodies.
+
+## `GET /sync/read-status/changes`
+
+Read-only replay of whatever the **last** round produced. No Graph call, cursor
+untouched, safe to hit repeatedly.
+
+Two reasons it exists:
+
+1. It holds the **untruncated** lists — this is how you get the other 990 items
+   when `limit` trimmed the response.
+2. It's the only way to collect what the **background watcher** found, since the
+   watcher has no caller to return to.
+
+`404` until some sync has run. One buffer, last-writer-wins: the next round
+overwrites it, so with the watcher running you must read it faster than
+`interval` or you will miss rounds.
+
+## `GET /sync/read-status/message/{message_id}`
+
+One message, one record — a point lookup rather than a batch:
+
+```jsonc
+{ "id": "AAMk…", "isRead": true,
+  "lastModifiedDateTime": "2026-08-07T10:14:52Z", "subject": "Invoice #421" }
+```
+
+Identical shape to an entry in a sync `value` list, so both parse with the same
+code. What makes it different from the endpoints above:
+
+- **Cursor-free.** Touches no delta cursor, no cached state, and advances
+  nothing. Call it as often as you like without affecting a sync in progress.
+- **Live.** Reports the mailbox *now*, straight from Graph — not what the last
+  round happened to capture. That makes it the right tool for re-checking one
+  message ("has this been read yet?") and for confirming a status after the fact.
+- **Any id.** Works whether or not the message appeared in a sync, and whatever
+  folder it lives in.
+
+It costs one Graph call per message, so it's a lookup, not a substitute for
+delta — walking a mailbox with it would be far slower than a single sync round.
+
+```bash
+curl -H "$A" "$B/sync/read-status/message/AAMkAGI2TG93..."
+```
+
+URL-encode the id. Ids containing `/`, `+`, or `=` are handled (the route uses a
+`:path` converter), so an already-encoded `%2F` works too. Unknown or deleted
+ids surface Graph's own `404 ErrorItemNotFound`.
+
+## `GET /sync/read-status/status`
+
+Health check for the whole subsystem.
+
+| Field | Meaning |
+| ----- | ------- |
+| `watching` / `interval` | Is the poller thread alive, and at what period |
+| `folder` | Folder the cursor belongs to |
+| `last_sync_at` | Timestamp of the most recent round |
+| `last_change_count` / `last_removed_count` | Size of that round |
+| `has_delta_link` | A real cursor exists ⇒ running incrementally |
+| `backfill_in_progress` | Paused mid-backfill at the page cap ⇒ more rounds to go |
+| `last_error` | Last Graph failure from the background thread, else `null` |
+
+`has_delta_link: false` + `backfill_in_progress: true` is the normal state
+*during* a long first sync.
+
+## `POST /sync/read-status/watch`
+
+Starts a daemon thread that runs the same sync every `interval` seconds and
+writes each change to stdout.
+
+```jsonc
+{ "interval": 60, "folder": "inbox" }   // interval min 10, both optional
+```
+
+- Idempotent — a second POST while running just answers
+  `{"message": "Already watching read-status changes"}`.
+- While a backfill is still incomplete the loop continues immediately instead of
+  sleeping out the interval, so a big first sync finishes in consecutive chunks.
+- Delivery is `_emit_read_status_changes()`, which prints. **That's the hook
+  point** — replace it to push to Slack, a webhook, or a queue.
+
+## `DELETE /sync/read-status/watch`
+
+Signals the thread to stop; `404` if nothing is running. The cursor survives, so
+restarting the watcher resumes from where it left off rather than re-backfilling.
+
+---
+
+## Typical first run
+
+```bash
+export EMAIL_API_TOKEN=...
+A="Authorization: Bearer $EMAIL_API_TOKEN"
+B=http://localhost:5000
+
+curl -X POST -H "$A" $B/auth/start      # sign in once (see README)
+
+# Baseline. Keep calling while "complete": false.
+curl -H "$A" "$B/sync/read-status?since=2026-08-01T00:00:00Z"
+
+# From here on, each call returns only what changed.
+curl -H "$A" "$B/sync/read-status"
+
+# Or hand it to the background poller and read results out of /changes.
+curl -X POST -H "$A" -H "Content-Type: application/json" \
+     -d '{"interval":60,"folder":"inbox"}' $B/sync/read-status/watch
+curl -H "$A" $B/sync/read-status/status
+curl -H "$A" $B/sync/read-status/changes
+
+# Re-check one message any time — no cursor involved.
+curl -H "$A" "$B/sync/read-status/message/AAMkAGI2TG93..."
+```
+
+## Which endpoint do I want?
+
+| You want | Use |
+| -------- | --- |
+| Everything that changed since last time | `GET /sync/read-status` |
+| The full list a round produced (or the watcher's) | `GET /sync/read-status/changes` |
+| The status of **one** message you already have an id for | `GET /sync/read-status/message/{id}` |
+| Continuous tracking without calling in a loop | `POST /sync/read-status/watch` |
+| Whether any of the above is healthy | `GET /sync/read-status/status` |
+
+Rule of thumb: **delta for "what changed", point lookup for "what about this
+one".** Using the lookup in a loop over a mailbox works but costs one Graph call
+per message — a single sync round does the same job in pages of 100.
+
+## How the cursor works
+
+- A finished round returns Graph's **deltaLink**, saved to
+  `.delta_cache.json` (override with `EMAIL_API_DELTA_CACHE`; in Docker it lives
+  on the `/data` volume beside the token cache). Keyed by user + folder — change
+  either and the cache is ignored rather than misapplied.
+- A round stopped by `max_pages` has no deltaLink yet, so it saves Graph's
+  **nextLink** instead. That resume position takes priority over any older
+  deltaLink on the following call, which is what makes a capped backfill safe:
+  the cursor never advances past data you haven't received.
+- Cursors expire. Graph answers `410 Gone`, and the sync automatically falls
+  back to a fresh baseline for that folder.
+- `reset=true` throws the cursor away deliberately — expect a full backfill, and
+  pass `since` with it unless you want the whole history again.
+
+## Limits worth knowing
+
+- **Folder-scoped only.** `/me/messages/delta` is not supported by Graph. Watch
+  another folder by passing `folder=`, but each folder is its own cursor and the
+  disk cache holds one at a time — switching folders forces a re-backfill.
+- **Polling, not push.** Latency floor is the poll `interval`. True push needs a
+  Graph change-notification subscription (public HTTPS endpoint, validation
+  handshake, ~3-day renewals) — and you'd keep delta anyway as the catch-up path
+  for dropped notifications.
+- **Single worker.** Cursor, watcher thread, and the `last_changes` buffer are
+  in-memory per process, so this only behaves with one uvicorn worker (which is
+  what the Docker service runs, for the same reason auth needs it).

Sync_write_request.md

@@ -0,0 +1,86 @@
+# Email service — write read-status (`PATCH /sync/read-status/...`)
+
+Copy everything below the line into any LLM session (or hand it to whoever owns the
+email microservice) before implementing the write endpoint.
+
+---
+
+You are extending the **email microservice** that already exposes the read-status
+delta and point-lookup APIs documented in `Sync_read.md`. Implement a **write**
+path that marks a message read (or unread) in the signed-in user's Outlook mailbox
+via Microsoft Graph. Mirror the existing `/sync/read-status/*` style exactly —
+same bearer auth, same `:path` id handling, same response shape.
+
+## Why we need this
+
+The HR-ATS inbox app learns that a user opened a message before Outlook does.
+Today that signal dies in our database: we have no Graph write permission and the
+email service exposes no write endpoint. Without this PATCH, local mark-read and
+Outlook drift permanently (and a later delta can even revert our flag).
+
+## Requested contract
+
+Mirror the existing read endpoints so both parse with one code path:
+
+```
+PATCH /sync/read-status/message/{id}
+Authorization: Bearer <api token>
+Content-Type: application/json
+
+{ "isRead": true }
+```
+
+**200 response** — identical shape to `GET /sync/read-status/message/{id}`:
+
+```jsonc
+{
+  "id": "AAMk…",
+  "isRead": true,
+  "lastModifiedDateTime": "2026-08-07T10:14:52Z",
+  "subject": "Invoice #421"
+}
+```
+
+Same `:path` converter for Graph ids that contain `/`, `+`, or `=`. Same bearer
+auth as every other `/sync/*` route. Answer `401` until device-code sign-in
+completes.
+
+## Required behaviour
+
+- **Idempotent.** Re-PATCHing `isRead: true` when already true is a no-op `200`
+  with the current record.
+- **Must not advance or disturb the delta cursor.** This is a point write, not a
+  sync round. Cursor, watcher, and `/changes` buffer stay untouched.
+- **404 `ErrorItemNotFound`** for unknown or deleted ids (same as the GET).
+- **403 surfaced distinctly** if the Graph scope is missing, so callers can tell
+  "not permitted" from "not found".
+
+## Graph scope prerequisite
+
+Needs `Mail.ReadWrite`. The service currently signs in read-only. Treat upgrading
+the consent / device-code scopes as an explicit product decision before shipping
+the route — not an implementation footnote.
+
+## Optional batch form
+
+For bulk reconcile without N round-trips:
+
+```
+PATCH /sync/read-status/messages
+{ "ids": ["AAMk…", "AAMk…"], "isRead": true }
+```
+
+Return a list of the same per-message records (or per-id errors). Nice-to-have;
+the single-id PATCH is the hard requirement.
+
+## What the caller will do with it
+
+HR-ATS will enqueue one Taskiq task per human mark-read, retried via existing
+smart-retry middleware. Expected volume is low (opens, not sweeps). After this
+lands we will stop treating local-only mark-read as a known divergence.
+
+## Out of scope for this request
+
+- Changing the delta `/sync/read-status` contract
+- Push / Graph change-notification subscriptions
+- Writing any field other than `isRead`

backend/.env.example

@@ -4,6 +4,10 @@ DB_HOST=
 DB_PORT=
 DB_NAME=
 EMAIL_URL=
+EMAIL_API_TOKEN=
+EMAIL_SYNC_FOLDER=inbox
+EMAIL_SYNC_SINCE=
+EMAIL_SYNC_CRON=* * * * *
 
 JWT_SECRET_KEY=
 JWT_ALGORITHM=HS256

backend/inbox/app.py

@@ -14,13 +14,13 @@ router = APIRouter()
 async def fetch_email(
     top:int=Query(100),
     skip:int=Query(0,ge=0),
-    token=Query(...),
+    token: str | None = Query(None),
     session: AsyncSession = Depends(get_session),
 ):
     try:
-        if not token:
-            raise HTTPException(status_code=401,detail="Unauthorized")
         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=[]
@@ -78,3 +78,36 @@ async def rematch_inbox(
         raise
     except Exception as e:
         raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.post("/inbox/{record_id}/read")
+async def mark_inbox_read(
+    record_id: str,
+    current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Email(session=session)
+        data=await service.mark_read(record_id)
+        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("/inbox/{record_id}/read-status")
+async def get_inbox_read_status(
+    record_id: str,
+    token: str | None = Query(None),
+    current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Email(session=session,token=token)
+        data=await service.refresh_read_status(record_id)
+        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))

backend/inbox/models.py

@@ -2,7 +2,7 @@ import uuid
 from datetime import datetime, timezone
 from typing import Any, Optional
 
-from sqlalchemy import Column, DateTime, func, or_
+from sqlalchemy import Column, DateTime, func, or_, update
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import Field, Relationship, SQLModel, select
@@ -211,3 +211,35 @@ class Inbox_Messages(SQLModel, table=True):
             statement = statement.where(cls._search_filter(search))
         result = await session.execute(statement)
         return result.scalar_one()
+
+    @classmethod
+    async def apply_read_status(cls, session: AsyncSession, changes) -> int:
+        """[{id, isRead, ...}] -> bulk UPDATE message_read. Returns rows touched."""
+        if not changes:
+            return 0
+        read_ids=[c.get("id") for c in changes if c.get("id") and c.get("isRead")]
+        unread_ids=[c.get("id") for c in changes if c.get("id") and not c.get("isRead")]
+        touched=0
+        if read_ids:
+            result=await session.execute(
+                update(cls).where(cls.message_id.in_(read_ids)).values(message_read=True)
+            )
+            touched+=result.rowcount or 0
+        if unread_ids:
+            result=await session.execute(
+                update(cls).where(cls.message_id.in_(unread_ids)).values(message_read=False)
+            )
+            touched+=result.rowcount or 0
+        await session.commit()
+        return touched
+
+    @classmethod
+    async def mark_message_read(cls, session: AsyncSession, record_id):
+        row=await cls.get_inbox_message_by_id(session,record_id)
+        if not row:
+            return None
+        row.message_read=True
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row

backend/inbox/plugins.py

@@ -1,16 +1,75 @@
-"""Inbox helpers — attachment loading and resume text extraction."""
+"""Inbox helpers — attachment loading, resume text extraction, read-status sync."""
 
 from __future__ import annotations
 
 import base64
+import os
 from pathlib import Path
+from urllib.parse import quote
+
+import httpx
+from dotenv import load_dotenv
 
 from inbox.models import Inbox_Messages
 from job.candidate.views import FileRead
 
+load_dotenv()
+
+EMAIL_URL=os.getenv("EMAIL_URL")
+EMAIL_API_TOKEN=os.getenv("EMAIL_API_TOKEN")
+
 _ATTACHMENTS_DIR=Path(__file__).resolve().parent/"decoded_attachments"
 
 
+async def fetch_read_status_delta(folder, since=None, limit=1000, max_pages=10, token=None):
+    """GET /sync/read-status -> the raw round dict."""
+    if not EMAIL_URL:
+        raise RuntimeError("EMAIL_URL must be set")
+    auth_token=token or EMAIL_API_TOKEN
+    if not auth_token:
+        raise RuntimeError("EMAIL_API_TOKEN must be set")
+    params={"folder":folder,"limit":limit,"max_pages":max_pages}
+    if since:
+        params["since"]=since
+    async with httpx.AsyncClient(timeout=15.0) as client:
+        response=await client.get(
+            f"{EMAIL_URL.rstrip('/')}/sync/read-status",
+            params=params,
+            headers={"Authorization":f"Bearer {auth_token}"},
+        )
+    if response.status_code>=400:
+        raise httpx.HTTPStatusError(
+            response.text,
+            request=response.request,
+            response=response,
+        )
+    return response.json()
+
+
+async def fetch_message_read_status(message_id, token=None):
+    """GET /sync/read-status/message/{id} -> record dict, or None on 404."""
+    if not EMAIL_URL:
+        raise RuntimeError("EMAIL_URL must be set")
+    auth_token=token or EMAIL_API_TOKEN
+    if not auth_token:
+        raise RuntimeError("EMAIL_API_TOKEN must be set")
+    encoded_id=quote(str(message_id),safe="")
+    async with httpx.AsyncClient(timeout=15.0) as client:
+        response=await client.get(
+            f"{EMAIL_URL.rstrip('/')}/sync/read-status/message/{encoded_id}",
+            headers={"Authorization":f"Bearer {auth_token}"},
+        )
+    if response.status_code==404:
+        return None
+    if response.status_code>=400:
+        raise httpx.HTTPStatusError(
+            response.text,
+            request=response.request,
+            response=response,
+        )
+    return response.json()
+
+
 def resolve_attachment_path(path_str:str) -> Path:
     """Prefer stored path; fall back to basename under decoded_attachments."""
     path=Path(path_str.strip())

backend/inbox/sync_tasks.py

@@ -0,0 +1,82 @@
+"""Inbox Taskiq tasks — Outlook read-status delta sweep."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+import httpx
+import redis.asyncio as redis
+from dotenv import load_dotenv
+
+from db_setup import session_scope
+from inbox.models import Inbox_Messages
+from inbox.plugins import fetch_read_status_delta
+from taskiq_management.broker_setup import broker
+
+load_dotenv()
+
+logger=logging.getLogger("inbox.sync")
+
+EMAIL_SYNC_FOLDER=os.getenv("EMAIL_SYNC_FOLDER","inbox")
+EMAIL_SYNC_SINCE=os.getenv("EMAIL_SYNC_SINCE") or None
+EMAIL_SYNC_CRON=os.getenv("EMAIL_SYNC_CRON","* * * * *")
+REDIS_URL=os.getenv("REDIS_URL","redis://localhost:6379/0")
+
+_LOCK_KEY="inbox:sync_read_status:lock"
+_LOCK_TTL=300
+_MAX_ROUNDS=10
+
+
+@broker.task(task_name="inbox.sync_read_status",schedule=[{"cron":EMAIL_SYNC_CRON}])
+async def sync_read_status() -> dict:
+    client=redis.from_url(REDIS_URL,decode_responses=True)
+    try:
+        acquired=await client.set(_LOCK_KEY,"1",nx=True,ex=_LOCK_TTL)
+        if not acquired:
+            logger.info("sync_read_status skipped — lock held")
+            return {"skipped":"locked"}
+
+        try:
+            rounds=0
+            applied_total=0
+            removed_total=0
+            since=EMAIL_SYNC_SINCE
+
+            while rounds<_MAX_ROUNDS:
+                rounds+=1
+                try:
+                    round_data=await fetch_read_status_delta(
+                        EMAIL_SYNC_FOLDER,
+                        since=since if rounds==1 else None,
+                        limit=1000,
+                        max_pages=10,
+                    )
+                except httpx.HTTPStatusError as e:
+                    if e.response.status_code==401:
+                        logger.warning("sync_read_status 401 — device-code sign-in required")
+                        return {"error":"unauthorized","status_code":401}
+                    raise
+
+                changes=round_data.get("value") or []
+                removed=round_data.get("removed") or []
+                removed_total+=len(removed)
+                if removed:
+                    logger.info("sync_read_status removed=%s",len(removed))
+
+                async with session_scope() as session:
+                    applied=await Inbox_Messages.apply_read_status(session,changes)
+                applied_total+=applied
+
+                if round_data.get("complete",True):
+                    break
+
+            return {
+                "rounds":rounds,
+                "applied":applied_total,
+                "removed":removed_total,
+            }
+        finally:
+            await client.delete(_LOCK_KEY)
+    finally:
+        await client.aclose()

backend/inbox/views.py

@@ -4,7 +4,11 @@ from fastapi import HTTPException
 from inbox.models import Inbox_Messages
 from inbox.file_decoder import decode_attachment
 from inbox.serializers import serialize_message
-from inbox.plugins import load_message_files
+from inbox.plugins import (
+    EMAIL_API_TOKEN,
+    fetch_message_read_status,
+    load_message_files,
+)
 from dotenv import load_dotenv
 load_dotenv()
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -17,7 +21,7 @@ class Email:
     def __init__(self,session:AsyncSession,token=None):
         self.session=session
         self.get_url=os.getenv("EMAIL_URL")
-        self.token=token
+        self.token=token or EMAIL_API_TOKEN
         self.pending_match_ids:list[str]=[]
 
     async def service_email(self,top,skip):
@@ -100,3 +104,27 @@ class Email:
 
     async def count_inbox_messages(self,search=None):
         return await Inbox_Messages.count_inbox_messages(self.session,search)
+
+    async def mark_read(self,record_id):
+        message=await Inbox_Messages.mark_message_read(self.session,record_id)
+        if not message:
+            raise HTTPException(status_code=404,detail="Message not found")
+        return serialize_message(message)
+
+    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)

backend/taskiq_management/broker_setup.py

@@ -1,6 +1,6 @@
 """Taskiq broker — Redis Streams + smart retry + DLQ.
 
-Worker:    taskiq worker taskiq_management.broker_setup:broker inbox.tasks taskiq_management.tasks
+Worker:    taskiq worker taskiq_management.broker_setup:broker inbox.tasks inbox.sync_tasks taskiq_management.tasks
 Scheduler: taskiq scheduler taskiq_management.broker_setup:scheduler
 """
 
@@ -11,6 +11,7 @@ import os
 from dotenv import load_dotenv
 from taskiq import TaskiqScheduler
 from taskiq.middlewares import SmartRetryMiddleware
+from taskiq.schedule_sources import LabelScheduleSource
 from taskiq_redis import (
     ListRedisScheduleSource,
     RedisAsyncResultBackend,
@@ -52,4 +53,7 @@ broker=(
     )
 )
 
-scheduler=TaskiqScheduler(broker=broker,sources=[schedule_source])
+scheduler=TaskiqScheduler(
+    broker=broker,
+    sources=[schedule_source,LabelScheduleSource(broker)],
+)

docker-compose.yml

@@ -24,6 +24,7 @@ services:
         "worker",
         "taskiq_management.broker_setup:broker",
         "inbox.tasks",
+        "inbox.sync_tasks",
         "taskiq_management.tasks",
         "--workers",
         "1",
@@ -48,7 +49,7 @@ services:
     build:
       context: ./backend
     container_name: hrms-taskiq-scheduler
-    command: ["taskiq", "scheduler", "taskiq_management.broker_setup:scheduler"]
+    command: ["taskiq", "scheduler", "taskiq_management.broker_setup:scheduler", "inbox.sync_tasks"]
     env_file:
       - ./backend/.env
     environment:

frontend/src/api/inbox.js

@@ -15,3 +15,8 @@ export function listMessages() {
 export function syncMailbox({ token, top, skip } = {}) {
   return request('/email/fetch', { params: { token, top, skip } })
 }
+
+/** Marks one persisted inbox row read (local DB only). */
+export function markRead(recordId) {
+  return request(`/inbox/${recordId}/read`, { method: 'POST' })
+}

frontend/src/screens/Inbox.jsx

@@ -11,7 +11,7 @@
 
 import { useMemo, useState } from 'react'
 import { useNavigate } from 'react-router-dom'
-import { useQuery, useQueryClient } from '@tanstack/react-query'
+import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
 
 import Modal from '../ui/Modal'
 import { Tabs } from '../ui/Tabs'
@@ -436,6 +436,14 @@ function EmailTab({ query, jobs, updateCandidates, toast }) {
   const selected = emails.find((e) => e.id === selectedId)
   const unread = emails.filter((e) => e.unread).length
 
+  const markRead = useMutation({
+    mutationFn: (recordId) => inboxApi.markRead(recordId),
+    onSuccess: () => {
+      qc.invalidateQueries({ queryKey: qk.mailbox.all() })
+    },
+    onError: (err) => toast(friendlyAuthError(err, 'Could not mark message read.'), 'error'),
+  })
+
   async function sync() {
     toast('Fetching from Outlook…', 'info')
     const res = await qc.refetchQueries({ queryKey: qk.mailbox.messages() })
@@ -472,6 +480,11 @@ function EmailTab({ query, jobs, updateCandidates, toast }) {
 
   const isImported = (e) => imported.has(e.id)
 
+  function selectEmail(e) {
+    setSelectedId(e.id)
+    if (e.unread) markRead.mutate(e.id)
+  }
+
   return (
     <>
       <div style={{ padding: '12px 18px', borderBottom: '1px solid var(--border)', display: 'flex', alignItems: 'center', gap: 12 }}>
@@ -498,8 +511,8 @@ function EmailTab({ query, jobs, updateCandidates, toast }) {
           {query.isSuccess && emails.map((e) => (
             <div
               key={e.id}
-              className={`inbox-item${e.unread && selectedId !== e.id ? ' unread' : ''}${selectedId === e.id ? ' active' : ''}`}
+              className={`inbox-item${e.unread ? ' unread' : ''}${selectedId === e.id ? ' active' : ''}`}
-              onClick={() => setSelectedId(e.id)}
+              onClick={() => selectEmail(e)}
             >
               <Avatar name={e.from} />
               <div className="ii-main">