From 8b4ebb1bb83d9faac5bd7c9e99fdb245fd66ac87 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Fri, 7 Aug 2026 18:26:37 +0500 Subject: [PATCH] Background saving terminated --- Sync_read.md | 228 ++++++++++++++++++++++ Sync_write_request.md | 86 ++++++++ backend/.env.example | 4 + backend/inbox/app.py | 39 +++- backend/inbox/models.py | 34 +++- backend/inbox/plugins.py | 61 +++++- backend/inbox/sync_tasks.py | 82 ++++++++ backend/inbox/views.py | 32 ++- backend/taskiq_management/broker_setup.py | 8 +- docker-compose.yml | 3 +- frontend/src/api/inbox.js | 5 + frontend/src/screens/Inbox.jsx | 19 +- 12 files changed, 588 insertions(+), 13 deletions(-) create mode 100644 Sync_read.md create mode 100644 Sync_write_request.md create mode 100644 backend/inbox/sync_tasks.py diff --git a/Sync_read.md b/Sync_read.md new file mode 100644 index 0000000..6f24af0 --- /dev/null +++ b/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). diff --git a/Sync_write_request.md b/Sync_write_request.md new file mode 100644 index 0000000..c906988 --- /dev/null +++ b/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 +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` diff --git a/backend/.env.example b/backend/.env.example index 54f01dc..32f5440 100644 --- a/backend/.env.example +++ b/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 diff --git a/backend/inbox/app.py b/backend/inbox/app.py index ae0906f..a4d8d89 100644 --- a/backend/inbox/app.py +++ b/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)) diff --git a/backend/inbox/models.py b/backend/inbox/models.py index a2e4a64..a23c4e0 100644 --- a/backend/inbox/models.py +++ b/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 diff --git a/backend/inbox/plugins.py b/backend/inbox/plugins.py index 3c31eb4..5983537 100644 --- a/backend/inbox/plugins.py +++ b/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()) diff --git a/backend/inbox/sync_tasks.py b/backend/inbox/sync_tasks.py new file mode 100644 index 0000000..42dc6ac --- /dev/null +++ b/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() diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 9ae3313..e786a70 100644 --- a/backend/inbox/views.py +++ b/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) diff --git a/backend/taskiq_management/broker_setup.py b/backend/taskiq_management/broker_setup.py index 6313ab8..931634c 100644 --- a/backend/taskiq_management/broker_setup.py +++ b/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)], +) diff --git a/docker-compose.yml b/docker-compose.yml index 40c623b..be849d0 100644 --- a/docker-compose.yml +++ b/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: diff --git a/frontend/src/api/inbox.js b/frontend/src/api/inbox.js index 8ad8e13..df813d3 100644 --- a/frontend/src/api/inbox.js +++ b/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' }) +} diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index fd29bde..dcf2003 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/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 ( <>
@@ -498,8 +511,8 @@ function EmailTab({ query, jobs, updateCandidates, toast }) { {query.isSuccess && emails.map((e) => (
setSelectedId(e.id)} + className={`inbox-item${e.unread ? ' unread' : ''}${selectedId === e.id ? ' active' : ''}`} + onClick={() => selectEmail(e)} >