Merge branch 'Talha' of https://git.utopiadeals.com/utopia-ai/HR-ATS-Portal into SQS_BROKER

backend/g_sheet/models.py

@@ -486,6 +486,7 @@ class FormData(SQLModel, table=True):
     async def list_by_emails(cls, session: AsyncSession, emails):
         """Sheet applicants for these addresses. Promoted rows are omitted —
         those already live on manual_upload_candidate."""
+        from g_sheet.plugins import form_applied_at_iso
         from job.job_post.models import JobPosts
 
         lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
@@ -513,10 +514,7 @@ class FormData(SQLModel, table=True):
                 "job_post_id": str(job_id) if job_id else None,
                 "job_title": title or rec.position_applied_for or None,
                 "status": rec.processing_state or None,
-                "applied_at": (
+                "applied_at": form_applied_at_iso(rec),
-                    rec.entry_date.isoformat() if rec.entry_date
-                    else (rec.created_at.isoformat() if rec.created_at else None)
-                ),
             })
         return rows
 

backend/g_sheet/plugins.py

@@ -444,6 +444,9 @@ def stringify_rows(rows):
 # -- FormData mapping ------------------------------------------------------
 
 _TIME_RE=re.compile(r"(\d{1,2}:\d{2}\s*(?:[AaPp][Mm])?)")
+_APPLIED_HMS_RE=re.compile(
+    r"(?P<h>\d{1,2}):(?P<m>\d{2})(?::(?P<s>\d{2}))?\s*(?P<ap>[AaPp][Mm])?",
+)
 _DAY_ORDINAL_RE=re.compile(r"\b(\d+)(st|nd|rd|th)\b",re.I)
 _DIGIT_RE=re.compile(r"\d")
 _NUMERIC_DATE_RE=re.compile(r"^(\d{1,2})([/\-.])(\d{1,2})\2(\d{2,4})$")
@@ -589,6 +592,58 @@ def parse_date_time(value):
     return parsed,time_str
 
 
+def _hms_from_text(text):
+    if not text:
+        return 0,0,0
+    match=_APPLIED_HMS_RE.search(str(text).strip())
+    if not match:
+        return 0,0,0
+    hours=int(match.group("h"))
+    minutes=int(match.group("m"))
+    seconds=int(match.group("s") or 0)
+    ap=(match.group("ap") or "").lower()
+    if ap=="pm" and hours<12:
+        hours+=12
+    if ap=="am" and hours==12:
+        hours=0
+    return min(hours,23),minutes,seconds
+
+
+def form_applied_at_iso(row):
+    """Wall-clock apply time for history. Not UTC midnight and not import time.
+
+    Google Form Timestamp is the source of truth (M/D/YYYY). entry_date is stored
+    as timestamptz at 00:00+00:00, so isoformat() would send `…T00:00:00+00:00`
+    and drop entry_time — the UI then paints 12:00am or shifts +5h.
+    """
+    raw=getattr(row,"raw_record",None)
+    ts=None
+    if isinstance(raw,dict):
+        for key,val in raw.items():
+            if str(key).strip().lower()=="timestamp" and val not in (None,""):
+                ts=str(val).strip()
+                break
+    if ts:
+        parsed,_=parse_date_time(ts)
+        if parsed is not None:
+            hours,minutes,seconds=_hms_from_text(ts)
+            return (
+                f"{parsed.year:04d}-{parsed.month:02d}-{parsed.day:02d}"
+                f"T{hours:02d}:{minutes:02d}:{seconds:02d}"
+            )
+    entry_date=getattr(row,"entry_date",None)
+    if entry_date is not None:
+        hours,minutes,seconds=_hms_from_text(getattr(row,"entry_time",None))
+        return (
+            f"{entry_date.year:04d}-{entry_date.month:02d}-{entry_date.day:02d}"
+            f"T{hours:02d}:{minutes:02d}:{seconds:02d}"
+        )
+    created=getattr(row,"created_at",None)
+    if created is None:
+        return None
+    return created.isoformat()
+
+
 def parse_age(value):
     """(int|None, raw|None) — first digit run if 0 < n < 100, always keep the raw."""
     if value is None:

backend/inbox/mailbox_sync_tasks.py

@@ -8,12 +8,13 @@ from datetime import datetime,timezone
 
 import redis.asyncio as redis
 from dotenv import load_dotenv
+from taskiq import TaskiqEvents
 
 from db_setup import session_scope
 from inbox.models import MailboxSyncRun
 from inbox.views import Email
 from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY
-from taskiq_management.mailbox_sync_broker_setup import mailbox_sync_broker
+from taskiq_management.mailbox_sync_broker_setup import MAILBOX_SYNC_QUEUE_NAME,mailbox_sync_broker
 from taskiq_management.middleware import PermanentTaskError
 
 load_dotenv()
@@ -22,6 +23,21 @@ logger=logging.getLogger("inbox.mailbox_sync")
 REDIS_URL=os.getenv("REDIS_URL","redis://localhost:6379/0")
 _LOCK_KEY="inbox:mailbox_sync:lock"
 _LOCK_TTL=900
+_CONSUMER_GROUP=os.getenv("TASKIQ_CONSUMER_GROUP","taskiq")
+# taskiq-redis listen() skips XAUTOCLAIM while this key exists. redis-py Lock has
+# no TTL by default, so SIGKILL (compose rebuild) leaves it forever and the Sync
+# button stays on a running row nobody will finish.
+_AUTOCLAIM_KEY=f"autoclaim:{_CONSUMER_GROUP}:{MAILBOX_SYNC_QUEUE_NAME}"
+
+
+@mailbox_sync_broker.on_event(TaskiqEvents.WORKER_STARTUP)
+async def _drop_stale_autoclaim(_state) -> None:
+    client=redis.from_url(REDIS_URL,decode_responses=True)
+    try:
+        if await client.delete(_AUTOCLAIM_KEY):
+            logger.warning("dropped stale autoclaim lock %s",_AUTOCLAIM_KEY)
+    finally:
+        await client.aclose()
 
 
 async def _fail(run_id:str,error:str) -> dict:
@@ -49,13 +65,25 @@ async def sync_mailbox(run_id:str) -> dict:
     try:
         acquired=await client.set(_LOCK_KEY,run_id,nx=True,ex=_LOCK_TTL)
         if not acquired:
-            return await _fail(run_id,"another mailbox sync is already running")
+            holder=await client.get(_LOCK_KEY)
+            # Crash/restart redelivers the same run_id while the TTL lock is
+            # still set. Failing that as "another sync" strands the lock until
+            # expiry and every later click also bounces.
+            if holder==run_id:
+                await client.expire(_LOCK_KEY,_LOCK_TTL)
+                logger.warning("mailbox sync %s reclaimed its own stale lock",run_id)
+            else:
+                logger.warning("mailbox sync %s skipped: lock held by %s",run_id,holder)
+                return await _fail(run_id,"another mailbox sync is already running")
 
         try:
             async with session_scope() as session:
                 row=await MailboxSyncRun.get_by_id(session,run_id)
                 if not row:
                     raise PermanentTaskError(f"sync run {run_id} not found")
+                if row.status in ("completed","failed"):
+                    logger.info("mailbox sync %s already %s, skipping",run_id,row.status)
+                    return {"status":row.status,"error":row.error}
                 await MailboxSyncRun.update_run(session,run_id,{
                     "status":"running",
                     "started_at":datetime.now(timezone.utc),

backend/inbox/models.py

@@ -3,7 +3,7 @@ import os
 from shlex import join
 import uuid
 from datetime import datetime, timezone
-from typing import Any, List, Optional
+from typing import Any, ClassVar, List, Optional
 
 from dotenv import load_dotenv
 from fastapi import HTTPException
@@ -2246,6 +2246,11 @@ class MailboxSyncRun(SQLModel, table=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))
 
+    # Worker should pick a queued run in seconds. Running past this means the
+    # consumer died (compose rebuild) and Redis never reclaimed the message.
+    QUEUED_STALE_SECONDS: ClassVar[int] = 180
+    RUNNING_STALE_SECONDS: ClassVar[int] = 2700
+
     @staticmethod
     def _as_uuid(record_id) -> uuid.UUID | None:
         if record_id in (None, ""):
@@ -2263,6 +2268,59 @@ class MailboxSyncRun(SQLModel, table=True):
         result = await session.execute(select(cls).where(cls.id == uid))
         return result.scalars().first()
 
+    @staticmethod
+    def _age_seconds(stamp, now):
+        if stamp is None:
+            return None
+        if getattr(stamp, "tzinfo", None) is None:
+            stamp = stamp.replace(tzinfo=timezone.utc)
+        if now.tzinfo is None:
+            now = now.replace(tzinfo=timezone.utc)
+        return (now - stamp).total_seconds()
+
+    @classmethod
+    def is_stale(cls, row, now=None) -> bool:
+        """True when a queued/running row has outlived a live worker."""
+        now = now or datetime.now(timezone.utc)
+        status = getattr(row, "status", None)
+        if status == "queued":
+            age = cls._age_seconds(getattr(row, "created_at", None), now)
+            return age is None or age >= cls.QUEUED_STALE_SECONDS
+        if status == "running":
+            age = cls._age_seconds(
+                getattr(row, "started_at", None) or getattr(row, "created_at", None),
+                now,
+            )
+            return age is None or age >= cls.RUNNING_STALE_SECONDS
+        return False
+
+    @classmethod
+    async def fail_stale(cls, session: AsyncSession, *, now=None, commit: bool = True):
+        """Mark stranded queued/running rows failed so Sync is clickable again."""
+        now = now or datetime.now(timezone.utc)
+        result = await session.execute(
+            select(cls).where(cls.status.in_(("queued", "running")))
+        )
+        failed = []
+        for row in result.scalars().all():
+            if not cls.is_stale(row, now):
+                continue
+            previous = row.status
+            row.status = "failed"
+            row.error = (
+                "Mailbox sync did not start. Click Sync to try again."
+                if previous == "queued"
+                else "Mailbox sync was interrupted. Click Sync to try again."
+            )
+            row.finished_at = now
+            session.add(row)
+            failed.append(row)
+        if failed and commit:
+            await session.commit()
+            for row in failed:
+                await session.refresh(row)
+        return failed
+
     @classmethod
     async def get_active(cls, session: AsyncSession):
         result = await session.execute(

backend/inbox/views.py

@@ -385,7 +385,9 @@ class Email:
         """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.
+        A stranded row from a killed worker is failed first so Sync is clickable again.
         """
+        await MailboxSyncRun.fail_stale(self.session)
         active=await MailboxSyncRun.get_active(self.session)
         if active:
             return serialize_mailbox_sync_run(active)
@@ -412,6 +414,7 @@ class Email:
         return serialize_mailbox_sync_run(row)
 
     async def get_mailbox_sync(self,run_id=None):
+        await MailboxSyncRun.fail_stale(self.session)
         if run_id:
             row=await MailboxSyncRun.get_by_id(self.session,run_id)
             if not row:

backend/notifications/models.py

@@ -148,7 +148,9 @@ class Notifications(SQLModel, table=True):
         skip: int = 0,
     ):
         statement = select(cls).where(
-            cls.user_id == user_id, cls.is_deleted == False  # noqa: E712
+            cls.user_id == user_id,
+            cls.is_deleted == False,  # noqa: E712
+            cls.title != "ATS score ready",
         )
         if unread_only:
             statement = statement.where(cls.is_read == False)  # noqa: E712
@@ -158,6 +160,7 @@ class Notifications(SQLModel, table=True):
             cls.user_id == user_id,
             cls.is_deleted == False,  # noqa: E712
             cls.is_read == False,  # noqa: E712
+            cls.title != "ATS score ready",
         )
         unread = (await session.execute(unread_statement)).scalar_one()
         statement = statement.order_by(cls.created_at.desc())

backend/notifications/views.py

@@ -29,7 +29,7 @@ from users.models import Users
 
 logger = logging.getLogger(__name__)
 
-# Candidate-history event_type → in-app kind, title, and profile tab.
+_SILENT_HISTORY_EVENTS = frozenset({"ats.scored"})
 _HISTORY_KIND = {
     "stage.changed": "application",
     "candidate.created": "application",
@@ -432,6 +432,8 @@ async def notify_candidate_history(
     """One in-app row per job stakeholder for a candidate_history write."""
     if row is None:
         return []
+    if row.event_type in _SILENT_HISTORY_EVENTS:
+        return []
     try:
         from job.job_post.models import JobPosts
 

frontend/format.test.mjs

@@ -36,11 +36,14 @@ eq(instant.getTime(), Date.parse(GRAPH), 'toInstant is the UTC instant')
 eq(toInstant(GRAPH_NAKED).getTime(), Date.parse(GRAPH), 'Graph without Z is still UTC')
 eq(toInstant(GRAPH_OFFSET).getTime(), Date.parse(GRAPH), 'Graph +00:00 is UTC')
 
-eq(fmtTime(wall), '7:52am', 'fmtTime(toDate) prints stored digits')
+eq(fmtTime(instant), '12:52pm', '7:52 UTC Date paints 12:52pm PKT')
-eq(fmtTime(instant), fmtTime(new Date(GRAPH)), 'fmtTime(toInstant) matches local clock')
+eq(fmtTime(GRAPH), '12:52pm', '7:52 UTC string paints 12:52pm PKT')
+eq(fmtTime(GRAPH_OFFSET), '12:52pm', '+00:00 paints PKT')
+eq(fmtTime('2026-09-09T11:50:24Z'), '4:50pm', 'live Graph sample is 4:50pm PKT')
+eq(fmtTime('2026-08-27T13:23:32'), '1:23pm', 'naive sheet ISO is PKT digits')
 
-if (new Date().getTimezoneOffset() !== 0) {
+if (new Date().getTimezoneOffset() === -300) {
-  eq(fmtTime(instant) === fmtTime(wall), false, 'non-UTC zone: instant display differs from wall-clock')
+  eq(fmtTime(wall), '7:52am', 'on a PKT machine wall-clock Date stays 7:52am')
 }
 
 if (failed) {

frontend/src/lib/format.js

@@ -15,12 +15,9 @@ export function formatRole(name) {
    time  → 7:30pm  (12-hour, lowercase, no space)
    both  → 3rd Sept 2026 - 7:30pm
 
-   API timestamps are year-month-day, then optional time, e.g.
+   Painted in Pakistan Standard Time (Asia/Karachi, UTC+5, no DST).
-     2026-08-28 03:56:19.685066-07
+   Naive stamps (no Z / offset) are already PKT digits.
-     2026-08-28T03:56:19.685066-07:00
+   UTC instants (…Z or +00:00) convert +5 to PKT. */
-     2026-08-28
-   Display uses those numbers as written: 2026 = year, 08 = month (Aug),
-   28 = day, 03:56 = 3:56am. The timezone suffix is not applied. */
 const MONTHS = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sept', 'Oct', 'Nov', 'Dec']
 const WEEKDAYS = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday']
 const WEEKDAYS_SHORT = ['Sun', 'Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat']
@@ -53,14 +50,32 @@ export function toDate(value) {
 }
 
 const HAS_OFFSET = /[zZ]|[+-]\d{2}:?\d{2}$/
+/** Recruiter clock. PKT has no DST — UTC+5 year-round. */
+const DISPLAY_TZ = 'Asia/Karachi'
+
+function pad2(n) {
+  return String(n).padStart(2, '0')
+}
+
+function pktParts(d) {
+  const out = {}
+  for (const p of new Intl.DateTimeFormat('en-GB', {
+    timeZone: DISPLAY_TZ,
+    year: 'numeric',
+    month: 'numeric',
+    day: 'numeric',
+    hour: 'numeric',
+    minute: 'numeric',
+    hourCycle: 'h23',
+  }).formatToParts(d)) {
+    if (p.type !== 'literal') out[p.type] = Number(p.value)
+  }
+  return out
+}
 
 /**
  * Parse a UTC instant (Graph receivedDateTime, timestamptz) into a Date
- * whose local clock matches the browser — the same conversion Outlook does.
+ * Display uses Asia/Karachi (PKT, UTC+5), not the browser zone.
- *
- * toDate() is wall-clock: it prints the stored digits and ignores Z/+00:00.
- * Graph always stores UTC, so that path shows 7:52 when Outlook shows 12:52
- * in Pakistan (UTC+5). Only call this for fields that are actually UTC.
  */
 export function toInstant(value) {
   if (value == null || value === '') return null
@@ -93,11 +108,34 @@ function ordinal(n) {
   }
 }
 
-/** 3rd Sept 2026 */
+/** Date to paint. Naive ISO digits are PKT; Z / +00:00 are UTC → PKT. */
+export function toDisplayDate(value) {
+  if (value instanceof Date) {
+    return Number.isNaN(value.getTime()) ? null : value
+  }
+  if (value == null || value === '') return null
+  const s = String(value).trim()
+  if (!s) return null
+  if (HAS_OFFSET.test(s)) return toInstant(s) || toDate(s)
+  const m = s.match(API_TS)
+  if (m) {
+    const hour = m[4] != null ? Number(m[4]) : 0
+    const minute = m[5] != null ? Number(m[5]) : 0
+    const second = m[6] != null ? Number(m[6]) : 0
+    const iso = `${m[1]}-${m[2]}-${m[3]}T${pad2(hour)}:${pad2(minute)}:${pad2(second)}+05:00`
+    const d = new Date(iso)
+    if (!Number.isNaN(d.getTime())) return d
+  }
+  return toDate(s) || toInstant(s)
+}
+
+/** 3rd Sept 2026 (PKT calendar) */
 export function fmtDate(value) {
-  const d = toDate(value)
+  const d = toDisplayDate(value)
   if (!d) return ''
-  return `${ordinal(d.getDate())} ${MONTHS[d.getMonth()]} ${d.getFullYear()}`
+  const p = pktParts(d)
+  if (p.day == null) return ''
+  return `${ordinal(p.day)} ${MONTHS[p.month - 1]} ${p.year}`
 }
 
 /** Same as fmtDate — the platform uses one date spelling. */
@@ -107,19 +145,21 @@ export function fmtShort(value) {
 
 /** 7:30pm */
 export function fmtTime(value) {
-  const d = toDate(value)
+  const d = toDisplayDate(value)
   if (!d) return ''
-  let hours = d.getHours()
+  const p = pktParts(d)
-  const minutes = d.getMinutes()
+  if (p.hour == null) return ''
+  let hours = p.hour
+  const minutes = p.minute
   const suffix = hours >= 12 ? 'pm' : 'am'
   hours = hours % 12
   if (hours === 0) hours = 12
-  return `${hours}:${String(minutes).padStart(2, '0')}${suffix}`
+  return `${hours}:${pad2(minutes)}${suffix}`
 }
 
 /** 3rd Sept 2026 - 7:30pm */
 export function fmtDateTime(value, sep = ' - ') {
-  const d = toDate(value)
+  const d = toDisplayDate(value)
   if (!d) return ''
   return `${fmtDate(d)}${sep}${fmtTime(d)}`
 }