diff --git a/backend/interview/app.py b/backend/interview/app.py index 82dbbed..ca4283d 100644 --- a/backend/interview/app.py +++ b/backend/interview/app.py @@ -38,6 +38,7 @@ async def create_calendar_event( data=await service.create_for_interview( interview_id, duration_minutes=body.duration_minutes, + current_user=current_user, ) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: @@ -59,6 +60,7 @@ async def reschedule_calendar_event( interview_id, instant=payload.instant, duration_minutes=payload.duration_minutes, + current_user=current_user, ) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: @@ -77,7 +79,9 @@ async def cancel_calendar_event( try: body=payload or CalendarCancelBody() service=Calendar(session=session) - data=await service.cancel_for_interview(interview_id,comment=body.comment) + data=await service.cancel_for_interview( + interview_id,comment=body.comment,current_user=current_user, + ) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise diff --git a/backend/interview/views.py b/backend/interview/views.py index 5965920..8d2d025 100644 --- a/backend/interview/views.py +++ b/backend/interview/views.py @@ -15,6 +15,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from interview.plugins import CALENDAR_API_TOKEN,cancel_event,create_event,reschedule_event from interview.serializers import serialize_event from job.candidate.models import Interviews +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder from job.interviews.serializers import serialize_interview from job.job_post.models import JobPosts @@ -72,7 +74,7 @@ class Calendar: async def _serialize(self,row): return serialize_interview(row,job_title=await self._job_title(row)) - async def create_for_interview(self,interview_id,duration_minutes=None): + async def create_for_interview(self,interview_id,duration_minutes=None,current_user=None): row=await self._load_interview(interview_id) if row.graph_event_id: return await self._serialize(row) @@ -121,13 +123,21 @@ class Calendar: row=await Interviews.set_calendar_event( self.session,interview_id,event["id"],event.get("web_link"), ) + await HistoryRecorder(self.session).record( + HistoryEvent.CALENDAR_CREATED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="interview",entity_id=row.id, + to_value=row.graph_event_id, + description=f"Calendar invite sent to {email}",commit=True, + ) return await self._serialize(row) - async def reschedule_for_interview(self,interview_id,instant,duration_minutes=None): + async def reschedule_for_interview(self,interview_id,instant,duration_minutes=None,current_user=None): row=await self._load_interview(interview_id) if not row.graph_event_id: raise HTTPException(status_code=404,detail="No calendar event for this interview") + old_instant=row.interview_date or row.interview_time start_dt=_as_datetime(instant) if start_dt is None: raise HTTPException(status_code=422,detail="instant is required") @@ -157,13 +167,21 @@ class Calendar: if event.get("web_link"): fields["web_link"]=event["web_link"] row=await Interviews.update_interview(self.session,interview_id,fields) + await HistoryRecorder(self.session).record( + HistoryEvent.CALENDAR_RESCHEDULED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="interview",entity_id=row.id, + from_value=old_instant.isoformat() if old_instant else None, + to_value=start_dt.isoformat(),commit=True, + ) return await self._serialize(row) - async def cancel_for_interview(self,interview_id,comment=None): + async def cancel_for_interview(self,interview_id,comment=None,current_user=None): row=await self._load_interview(interview_id) if not row.graph_event_id: return await self._serialize(row) + old_event_id=row.graph_event_id try: await cancel_event(row.graph_event_id,comment=comment,token=self.token) except httpx.HTTPStatusError as e: @@ -172,4 +190,11 @@ class Calendar: raise HTTPException(status_code=500,detail=str(e)) row=await Interviews.set_calendar_event(self.session,interview_id,None,None) + await HistoryRecorder(self.session).record( + HistoryEvent.CALENDAR_CANCELLED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="interview",entity_id=row.id, + from_value=old_event_id,to_value=None, + description=comment,commit=True, + ) return await self._serialize(row) diff --git a/backend/job/app.py b/backend/job/app.py index 5134fd1..4e2b225 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -8,6 +8,7 @@ from job.notes.views import Note from job.activity.views import ActivityLog from job.feedback.views import FeedbackView from job.pipeline.views import Pipeline +from job.history.views import HistoryRecorder from job.assignment.views import Assignment from job.cost.views import HiringCost from sqlalchemy.ext.asyncio import AsyncSession @@ -245,6 +246,7 @@ async def cv_upload( service=FileRead(session=session,filename=file.filename,file=file_content) data=await service.ingest_upload( candidate_email=candidate_email,candidate_name=candidate_name, + current_user=current_user, ) return JSONResponse(content={"data":data,"status_code":200}) except HTTPException: @@ -261,7 +263,7 @@ async def candidate_inbox_match( ): try: service=FileRead(session=session) - data=await service.match_inbox_cv(inbox_message_id) + data=await service.match_inbox_cv(inbox_message_id,current_user=current_user) return JSONResponse(content={"data":data,"status_code":200}) except HTTPException: raise @@ -512,7 +514,7 @@ async def update_candidate( ): try: service=CandidateView(session=session) - data=await service.update_candidate(user_id,payload.model_dump(exclude_unset=True)) + data=await service.update_candidate(user_id,payload.model_dump(exclude_unset=True),current_user) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise @@ -520,6 +522,24 @@ async def update_candidate( raise HTTPException(status_code=500,detail=str(e)) +@router.get("/candidate/history/fetch") +async def fetch_candidate_history( + user_id:str=Query(...), + limit:int=Query(200,ge=1,le=500), + offset:int=Query(0,ge=0), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=HistoryRecorder(session=session) + data,total=await service.list_for_user(user_id,limit=limit,offset=offset) + return JSONResponse(content={"data":data,"total":total,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + @router.get("/interview/fetch") async def fetch_interview( interview_id:str=Query(None), @@ -562,7 +582,7 @@ async def create_interview( ): try: service=Interview(session=session) - data=await service.create_interview(payload.model_dump(exclude_unset=True)) + data=await service.create_interview(payload.model_dump(exclude_unset=True),current_user) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise @@ -579,7 +599,7 @@ async def update_interview( ): try: service=Interview(session=session) - data=await service.update_interview(interview_id,payload.model_dump(exclude_unset=True)) + data=await service.update_interview(interview_id,payload.model_dump(exclude_unset=True),current_user) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise @@ -630,7 +650,7 @@ async def update_note( ): try: service=Note(session=session) - data=await service.update_note(note_id,payload.model_dump(exclude_unset=True)) + data=await service.update_note(note_id,payload.model_dump(exclude_unset=True),current_user) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise @@ -723,7 +743,7 @@ async def update_feedback( ): try: service=FeedbackView(session=session) - data=await service.update_feedback(feedback_id,payload.model_dump(exclude_unset=True)) + data=await service.update_feedback(feedback_id,payload.model_dump(exclude_unset=True),current_user) return JSONResponse(content={"data":data,"total":1,"status_code":200}) except HTTPException: raise diff --git a/backend/job/candidate/models.py b/backend/job/candidate/models.py index 9df020d..6ec9676 100644 --- a/backend/job/candidate/models.py +++ b/backend/job/candidate/models.py @@ -3,7 +3,7 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING, List, Optional from fastapi import HTTPException -from sqlalchemy import JSON, DateTime, func, UniqueConstraint +from sqlalchemy import JSON, DateTime, Index, func, UniqueConstraint from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -797,4 +797,71 @@ class ApplicationStageTransitions(SQLModel, table=True): return result.scalar_one() +class CandidateHistory(SQLModel, table=True): + """Append-only audit log for one candidate (users.id), scoped to an application. + + user_id is the anchor: the profile modal is keyed by users.id, so every read is + one indexed scan. inbox_id / manual_upload_candidate_id record WHICH application + the event happened on and are BOTH nullable -- unlike application_stage_transitions, + several events (notes, rating changes, an import before the inbox link exists) have + neither. That is why there is no XOR check constraint here. + """ + + __tablename__ = "candidate_history" + __table_args__ = ( + Index("ix_candidate_history_user_created", "user_id", "created_at"), + ) + + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + user_id: uuid.UUID = Field(foreign_key="users.id", nullable=False) + inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id") + manual_upload_candidate_id: uuid.UUID | None = Field( + default=None, index=True, foreign_key="manual_upload_candidate.id" + ) + event_type: str = Field(index=True) + entity_type: str | None = Field(default=None) + entity_id: str | None = Field(default=None) + from_value: str | None = Field(default=None) + to_value: str | None = Field(default=None) + description: str | None = Field(default=None) + actor_id: uuid.UUID | None = Field(default=None, foreign_key="users.id") + actor_kind: str = Field(default="user") + meta: dict | None = Field(default=None, sa_type=JSON) + created_at: datetime = Field(default_factory=_now, 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 fetch_by_user(cls, session: AsyncSession, user_id, limit=200, offset=0): + uid = cls._as_uuid(user_id) + if uid is None: + return [], 0 + total_stmt = select(func.count()).select_from(cls).where(cls.user_id == uid) + total = (await session.execute(total_stmt)).scalar_one() + result = await session.execute( + select(cls) + .where(cls.user_id == uid) + .order_by(cls.created_at.desc(), cls.id.desc()) + .offset(int(offset)) + .limit(int(limit)) + ) + return list(result.scalars().all()), int(total) + + @classmethod + async def insert_event(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 + + import users.models as _users_models # noqa: E402, F401 diff --git a/backend/job/candidate/serializers.py b/backend/job/candidate/serializers.py index 7ba7351..b1b9448 100644 --- a/backend/job/candidate/serializers.py +++ b/backend/job/candidate/serializers.py @@ -72,6 +72,7 @@ def serialize_candidate_profile( message = link.messages payload = { "inbox_id": link.id, + "manual_upload_candidate_id": None, "user_id": str(link.user_id) if link.user_id else None, "candidate_id": None, "name": user.name if user else None, @@ -139,6 +140,7 @@ def serialize_manual_candidate_profile(row, user, job_post) -> Dict[str, Any]: position = (row.current_position or "").strip() or None return { "inbox_id": None, + "manual_upload_candidate_id": str(row.id), "user_id": str(user.id) if user else (str(row.user_id) if row.user_id else None), "candidate_id": None, "name": (user.name if user else None) or row.candidate_name or None, diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 491f4e0..3c0fa40 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -29,6 +29,8 @@ from job.candidate.serializers import serialize_candidate,serialize_candidate_pr from job.job_post.models import JobPosts from job.job_post.serializers import serialize_job_post from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder from job.notes.serializers import serialize_note from job.candidate.plugins import extract_candidate_email from users.models import Users @@ -123,7 +125,7 @@ class FileRead: except OSError as e: logger.warning("could not remove orphaned upload %s: %s",file_path,e) - async def ingest_upload(self,candidate_email=None,candidate_name=None): + async def ingest_upload(self,candidate_email=None,candidate_name=None,current_user=None): """Persist a recruiter-uploaded CV with full email-ingestion parity.""" from inbox.file_decoder import AttachmentDecodeError,decode_attachment from inbox.cv_tasks import match_uploaded_cv @@ -196,6 +198,23 @@ class FileRead: logger.warning("account setup mail failed for %s: %s",new_user_email,e) account_setup=[{"email":new_user_email,"sent":False}] + # Inbox link may not exist yet (match task creates it later); resolve + # by email because insert_email creates the Users row synchronously. + user=await Users.get_user_by_email(self.session,email) + if user: + await HistoryRecorder(self.session).record( + HistoryEvent.CANDIDATE_IMPORTED.value, + current_user=current_user,user_id=user.id, + entity_type="inbox_message",entity_id=row.id, + to_value=email,description=f"CV uploaded: {filename}",commit=True, + ) + await HistoryRecorder(self.session).record( + HistoryEvent.DOCUMENT_UPLOADED.value, + current_user=current_user,user_id=user.id, + entity_type="document",entity_id=row.id, + to_value=filename,commit=True, + ) + return { "queued":True, "inbox_message_id":str(row.id), @@ -208,7 +227,7 @@ class FileRead: "text":text, } - async def match_inbox_cv(self,inbox_message_id): + async def match_inbox_cv(self,inbox_message_id,current_user=None): from inbox.plugins import resolve_attachment_path from inbox.tasks import match_inbox_message @@ -235,6 +254,12 @@ class FileRead: ).kiq(str(row.id),force=True) file_name=(row.file_name or "").split(",")[0].strip() or found.name + await HistoryRecorder(self.session).record( + HistoryEvent.CANDIDATE_IMPORTED.value, + current_user=current_user,message_id=inbox_message_id, + entity_type="inbox_message",entity_id=row.id, + to_value=file_name,description=f"CV uploaded: {file_name}",commit=True, + ) return { "queued":True, "inbox_message_id":str(row.id), @@ -358,7 +383,7 @@ class CandidateScoring: rows=[] for slot in range(len(sources)): rows.append(await Candidates.upsert_candidate(self.session,{**fields_by_slot[slot],**common})) - await self._sync_ats_results(source_kind,job,rows,sources) + await self._sync_ats_results(source_kind,job,rows,sources,current_user) rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0)) return [serialize_candidate(row) for row in rows] @@ -401,7 +426,7 @@ class CandidateScoring: fields_by_slot[slot]=candidate_failed_fields(source,result.error_code,result.error_message) return fields_by_slot - async def _sync_ats_results(self,source_kind,job,rows,sources): + async def _sync_ats_results(self,source_kind,job,rows,sources,current_user=None): if source_kind=="inbox": best={} for source,row in zip(sources,rows): @@ -412,7 +437,7 @@ class CandidateScoring: best[mid]=row for message_id,row in best.items(): try: - await self._sync_inbox_ats(message_id,job,row) + await self._sync_inbox_ats(message_id,job,row,current_user=current_user) except Exception: await self.session.rollback() logger.exception("inbox ATS denorm failed for message %s",message_id) @@ -421,12 +446,12 @@ class CandidateScoring: if row.status!="completed": continue try: - await self._sync_upload_ats(job,row) + await self._sync_upload_ats(job,row,current_user=current_user) except Exception: await self.session.rollback() logger.exception("upload ATS history failed for candidate %s",row.id) - async def _sync_inbox_ats(self,message_id,job,row): + async def _sync_inbox_ats(self,message_id,job,row,current_user=None): """Land a completed score on inbox_messages / inbox / ats_results. message_id is the scoring call's known inbox_messages PK, not a column @@ -446,6 +471,8 @@ class CandidateScoring: await Inbox_Messages.set_ats_score(self.session,message_id,row.match_score,band) if link is None: return + old=await AtsResults.get_current_for_inbox(self.session,link.id) + old_score=old.overall_score if old else None identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id) await AtsResults.insert_result(self.session,{ "inbox_id":link.id, @@ -456,11 +483,24 @@ class CandidateScoring: "model_name":row.model, "is_current":True, }) + await HistoryRecorder(self.session).record( + HistoryEvent.ATS_SCORED.value, + current_user=current_user,user_id=link.user_id,inbox_id=link.id, + entity_type="ats_result",entity_id=row.id, + from_value=old_score,to_value=row.match_score, + description=f"{band} against {job.title or 'job'}",commit=True, + ) - async def _sync_upload_ats(self,job,row): + async def _sync_upload_ats(self,job,row,current_user=None): """History row for an upload-sourced score — inbox_id stays NULL.""" band=CandidateView._recommendation(row.match_score) or "" identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id) + old=None + if identity.get("user_id"): + old=await AtsResults.get_current_for_user(self.session,identity["user_id"],job.id) + elif identity.get("candidate_id"): + old=await AtsResults.get_current_for_candidate(self.session,identity["candidate_id"]) + old_score=old.overall_score if old else None await AtsResults.insert_result(self.session,{ "inbox_id":None, **identity, @@ -470,7 +510,13 @@ class CandidateScoring: "model_name":row.model, "is_current":True, }) - + await HistoryRecorder(self.session).record( + HistoryEvent.ATS_SCORED.value, + current_user=current_user,user_id=identity.get("user_id"), + entity_type="ats_result",entity_id=row.id, + from_value=old_score,to_value=row.match_score, + description=f"{band} against {job.title or 'job'}",commit=True, + ) class CandidateView: def __init__(self,session:AsyncSession): @@ -517,6 +563,22 @@ class CandidateView: } row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data) + await HistoryRecorder(self.session).record( + HistoryEvent.CANDIDATE_CREATED.value, + actor_id=current_user,user_id=row.user_id, + manual_upload_candidate_id=row.id, + entity_type="manual_upload_candidate",entity_id=row.id, + to_value=row.candidate_email, + description=(row.platform or "").strip() or "manual_upload",commit=True, + ) + if (row.file_name or "").strip(): + await HistoryRecorder(self.session).record( + HistoryEvent.DOCUMENT_UPLOADED.value, + actor_id=current_user,user_id=row.user_id, + manual_upload_candidate_id=row.id, + entity_type="document",entity_id=row.id, + to_value=row.file_name,commit=True, + ) return serialize_manual_upload_candidate(row) except HTTPException: raise @@ -569,7 +631,7 @@ class CandidateView: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - async def update_candidate(self,user_id,payload): + async def update_candidate(self,user_id,payload,current_user=None): try: if not user_id: raise HTTPException(status_code=400,detail="user_id is required") @@ -580,8 +642,25 @@ class CandidateView: records=links if isinstance(links,list) else ([links] if links else []) if not records: raise HTTPException(status_code=404,detail="Candidate not found") + first=records[0] + old_favorite=getattr(first,"favorite",None) + old_rating=getattr(first,"rating",None) for link in records: await Inbox.update_inbox(self.session,link.id,fields) + if "favorite" in fields and fields["favorite"]!=old_favorite: + await HistoryRecorder(self.session).record( + HistoryEvent.FAVORITE_CHANGED.value, + current_user=current_user,user_id=user_id,inbox_id=first.id, + entity_type="candidate",entity_id=user_id, + from_value=old_favorite,to_value=fields["favorite"],commit=True, + ) + if "rating" in fields and fields["rating"]!=old_rating: + await HistoryRecorder(self.session).record( + HistoryEvent.RATING_CHANGED.value, + current_user=current_user,user_id=user_id,inbox_id=first.id, + entity_type="candidate",entity_id=user_id, + from_value=old_rating,to_value=fields["rating"],commit=True, + ) refreshed=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0) return await self.attach_profile_detail(refreshed) except HTTPException: diff --git a/backend/job/feedback/views.py b/backend/job/feedback/views.py index 16e2e57..6af94e6 100644 --- a/backend/job/feedback/views.py +++ b/backend/job/feedback/views.py @@ -6,6 +6,8 @@ from sqlmodel import select from job.candidate.models import Feedback from job.feedback.models import FeedbackTemplates from job.feedback.serializers import serialize_feedback, serialize_feedback_template +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder class FeedbackView: @@ -49,17 +51,39 @@ class FeedbackView: ), } row=await Feedback.insert_feedback(self.session,fields) + desc=row.note or None + await HistoryRecorder(self.session).record( + HistoryEvent.FEEDBACK_CREATED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="feedback",entity_id=row.id, + to_value=row.review or None,description=desc,commit=True, + ) row=await self._load(row.id) return serialize_feedback(row) - async def update_feedback(self,feedback_id,payload): + async def update_feedback(self,feedback_id,payload,current_user=None): allowed=("review","financial_status","score","note","inbox_id","reviewed_by") fields={k:v for k,v in payload.items() if v is not None and k in allowed} if not fields: raise HTTPException(status_code=400,detail="No fields to update") + before=await self._load(feedback_id) + if not before: + raise HTTPException(status_code=404,detail="Feedback not found") + old_review=before.review or "" + old_score=before.score row=await Feedback.update_feedback(self.session,feedback_id,fields) if not row: raise HTTPException(status_code=404,detail="Feedback not found") + desc=row.note + if old_score!=row.score: + desc=f"score {old_score} → {row.score}" + await HistoryRecorder(self.session).record( + HistoryEvent.FEEDBACK_UPDATED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="feedback",entity_id=row.id, + from_value=old_review,to_value=row.review or None, + description=desc,commit=True, + ) row=await self._load(row.id) return serialize_feedback(row) diff --git a/backend/job/history/enums.py b/backend/job/history/enums.py new file mode 100644 index 0000000..5a3cdc1 --- /dev/null +++ b/backend/job/history/enums.py @@ -0,0 +1,20 @@ +from enum import Enum + + +class HistoryEvent(str, Enum): + STAGE_CHANGED = "stage.changed" + NOTE_CREATED = "note.created" + NOTE_UPDATED = "note.updated" + FEEDBACK_CREATED = "feedback.created" + FEEDBACK_UPDATED = "feedback.updated" + INTERVIEW_CREATED = "interview.created" + INTERVIEW_UPDATED = "interview.updated" + CALENDAR_CREATED = "calendar.created" + CALENDAR_RESCHEDULED = "calendar.rescheduled" + CALENDAR_CANCELLED = "calendar.cancelled" + FAVORITE_CHANGED = "favorite.changed" + RATING_CHANGED = "rating.changed" + CANDIDATE_CREATED = "candidate.created" + CANDIDATE_IMPORTED = "candidate.imported" + DOCUMENT_UPLOADED = "document.uploaded" + ATS_SCORED = "ats.scored" diff --git a/backend/job/history/serializers.py b/backend/job/history/serializers.py new file mode 100644 index 0000000..0f91e87 --- /dev/null +++ b/backend/job/history/serializers.py @@ -0,0 +1,20 @@ +def serialize_history(row, actor_name=None) -> dict: + return { + "id": str(row.id), + "user_id": str(row.user_id) if row.user_id else None, + "inbox_id": row.inbox_id, + "manual_upload_candidate_id": ( + str(row.manual_upload_candidate_id) if row.manual_upload_candidate_id else None + ), + "event_type": row.event_type, + "entity_type": row.entity_type, + "entity_id": row.entity_id, + "from_value": row.from_value, + "to_value": row.to_value, + "description": row.description, + "actor_id": str(row.actor_id) if row.actor_id else None, + "actor_name": actor_name, + "actor_kind": row.actor_kind, + "meta": row.meta, + "created_at": row.created_at.isoformat() if row.created_at else None, + } diff --git a/backend/job/history/views.py b/backend/job/history/views.py new file mode 100644 index 0000000..7d204be --- /dev/null +++ b/backend/job/history/views.py @@ -0,0 +1,126 @@ +import logging +from datetime import datetime + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlmodel import select + +from inbox.models import Inbox +from job.candidate.models import CandidateHistory, Manual_UPLOAD_CANDIDATE +from job.history.serializers import serialize_history +from users.models import Users + +logger = logging.getLogger(__name__) + + +def _text(value): + if value is None: + return None + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, datetime): + return value.isoformat() + return str(value) + + +class HistoryRecorder: + def __init__(self, session: AsyncSession): + self.session = session + + async def resolve_user_id( + self, + *, + user_id=None, + inbox_id=None, + manual_upload_candidate_id=None, + message_id=None, + ): + uid = CandidateHistory._as_uuid(user_id) + if uid is not None: + return uid + if inbox_id is not None: + row = await Inbox.get_inbox_by_id(self.session, inbox_id) + if row and row.user_id: + return row.user_id + if manual_upload_candidate_id is not None: + row = await Manual_UPLOAD_CANDIDATE.get_by_id(self.session, manual_upload_candidate_id) + if row and row.user_id: + return row.user_id + if message_id is not None: + row = await Inbox.get_inbox_by_message_id(self.session, message_id) + if row and row.user_id: + return row.user_id + return None + + def _actor_id(self, current_user=None, actor_id=None): + if actor_id is not None: + return CandidateHistory._as_uuid(actor_id) + if isinstance(current_user, dict) and current_user.get("id"): + return CandidateHistory._as_uuid(current_user.get("id")) + if current_user and not isinstance(current_user, dict): + return CandidateHistory._as_uuid(current_user) + return None + + async def record( + self, + event_type, + *, + current_user=None, + actor_id=None, + user_id=None, + inbox_id=None, + manual_upload_candidate_id=None, + message_id=None, + entity_type=None, + entity_id=None, + from_value=None, + to_value=None, + description=None, + meta=None, + actor_kind=None, + commit=False, + ): + try: + resolved = await self.resolve_user_id( + user_id=user_id, + inbox_id=inbox_id, + manual_upload_candidate_id=manual_upload_candidate_id, + message_id=message_id, + ) + if resolved is None: + return None + fields = { + "user_id": resolved, + "inbox_id": int(inbox_id) if inbox_id is not None else None, + "manual_upload_candidate_id": CandidateHistory._as_uuid(manual_upload_candidate_id), + "event_type": event_type, + "entity_type": entity_type, + "entity_id": str(entity_id) if entity_id is not None else None, + "from_value": _text(from_value), + "to_value": _text(to_value), + "description": description, + "actor_id": self._actor_id(current_user=current_user, actor_id=actor_id), + "actor_kind": actor_kind or "user", + "meta": meta, + } + return await CandidateHistory.insert_event(self.session, fields, commit=commit) + except Exception: + logger.exception("candidate history record failed for %s", event_type) + if commit: + try: + await self.session.rollback() + except Exception: + logger.exception("candidate history rollback failed") + return None + + async def list_for_user(self, user_id, *, limit=200, offset=0): + rows, total = await CandidateHistory.fetch_by_user( + self.session, user_id, limit=limit, offset=offset + ) + actor_ids = {r.actor_id for r in rows if r.actor_id} + names = {} + if actor_ids: + result = await self.session.execute( + select(Users.id, Users.name).where(Users.id.in_(actor_ids)) + ) + names = {uid: name for uid, name in result.all()} + return [serialize_history(r, actor_name=names.get(r.actor_id)) for r in rows], total diff --git a/backend/job/interviews/views.py b/backend/job/interviews/views.py index 5c75538..bc6d3eb 100644 --- a/backend/job/interviews/views.py +++ b/backend/job/interviews/views.py @@ -2,6 +2,8 @@ from fastapi import HTTPException from sqlalchemy.ext.asyncio import AsyncSession from job.candidate.models import Interviews +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder from job.interviews.serializers import serialize_interview from job.job_post.models import JobPosts @@ -49,7 +51,7 @@ class Interview: titles=await Interviews.job_titles_by_inbox(self.session,[r.inbox_id for r in rows]) return [serialize_interview(r,job_title=titles.get(r.inbox_id)) for r in rows],total - async def create_interview(self,payload): + async def create_interview(self,payload,current_user=None): fields={ "interview_date":payload.get("interview_date"), "interview_time":payload.get("interview_time"), @@ -59,13 +61,34 @@ class Interview: } fields={k:v for k,v in fields.items() if v is not None} row=await Interviews.insert_interview(self.session,fields) + when=row.interview_date or row.interview_time + desc=f"{row.interview_type or 'Interview'} on {when.isoformat() if when else '—'}" + await HistoryRecorder(self.session).record( + HistoryEvent.INTERVIEW_CREATED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="interview",entity_id=row.id, + to_value=row.interview_status or None,description=desc,commit=True, + ) return await self._serialize(row) - async def update_interview(self,interview_id,payload): + async def update_interview(self,interview_id,payload,current_user=None): fields={k:v for k,v in payload.items() if v is not None} if not fields: raise HTTPException(status_code=400,detail="No fields to update") + before=await Interviews.get_interview_by_id(self.session,interview_id) + if not before: + raise HTTPException(status_code=404,detail="Interview not found") + old_status=before.interview_status or None row=await Interviews.update_interview(self.session,interview_id,fields) if not row: raise HTTPException(status_code=404,detail="Interview not found") + when=row.interview_date or row.interview_time + desc=f"{row.interview_type or 'Interview'} on {when.isoformat() if when else '—'}" + await HistoryRecorder(self.session).record( + HistoryEvent.INTERVIEW_UPDATED.value, + current_user=current_user,inbox_id=row.inbox_id, + entity_type="interview",entity_id=row.id, + from_value=old_status,to_value=row.interview_status or None, + description=desc,commit=True, + ) return await self._serialize(row) diff --git a/backend/job/notes/views.py b/backend/job/notes/views.py index 433431c..bb2498b 100644 --- a/backend/job/notes/views.py +++ b/backend/job/notes/views.py @@ -4,6 +4,8 @@ from sqlalchemy.orm import selectinload from sqlmodel import select from job.candidate.models import Notes +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder from job.notes.serializers import serialize_note @@ -48,15 +50,32 @@ class Note: if not fields["user_id"]: raise HTTPException(status_code=400,detail="user_id is required") row=await Notes.insert_note(self.session,fields) + await HistoryRecorder(self.session).record( + HistoryEvent.NOTE_CREATED.value, + current_user=current_user,user_id=row.user_id, + entity_type="note",entity_id=row.id, + to_value=(row.note or "")[:120],commit=True, + ) row=await self._load(row.id) return serialize_note(row) - async def update_note(self,note_id,payload): + async def update_note(self,note_id,payload,current_user=None): fields={k:v for k,v in payload.items() if v is not None and k in ("note",)} if not fields: raise HTTPException(status_code=400,detail="No fields to update") + before=await self._load(note_id) + if not before: + raise HTTPException(status_code=404,detail="Note not found") + old_note=before.note or "" row=await Notes.update_note(self.session,note_id,fields) if not row: raise HTTPException(status_code=404,detail="Note not found") + new_note=row.note or "" + await HistoryRecorder(self.session).record( + HistoryEvent.NOTE_UPDATED.value, + current_user=current_user,user_id=row.user_id, + entity_type="note",entity_id=row.id, + from_value=old_note[:120],to_value=new_note[:120],commit=True, + ) row=await self._load(row.id) return serialize_note(row) diff --git a/backend/job/pipeline/views.py b/backend/job/pipeline/views.py index 1d34dec..48ef30d 100644 --- a/backend/job/pipeline/views.py +++ b/backend/job/pipeline/views.py @@ -4,6 +4,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from inbox.enums import Candidate_application_Status from inbox.models import Inbox from job.candidate.models import ApplicationStageTransitions, Manual_UPLOAD_CANDIDATE, _now +from job.history.enums import HistoryEvent +from job.history.views import HistoryRecorder from job.pipeline.serializers import serialize_pipeline_counts, serialize_stage_transition from inbox.plugins import get_ats_score_for_manual_user, get_ats_score_for_user @@ -63,10 +65,10 @@ class Pipeline: if isinstance(current_user,dict) and current_user.get("id"): changed_by=ApplicationStageTransitions._as_uuid(current_user.get("id")) if inbox_id is not None: - return await self._change_inbox_stage(inbox_id,stage,changed_by,change_reason) - return await self._change_manual_stage(manual_upload_id,stage,changed_by,change_reason) + return await self._change_inbox_stage(inbox_id,stage,changed_by,change_reason,current_user) + return await self._change_manual_stage(manual_upload_id,stage,changed_by,change_reason,current_user) - async def _change_inbox_stage(self,inbox_id,stage,changed_by,change_reason): + async def _change_inbox_stage(self,inbox_id,stage,changed_by,change_reason,current_user): inbox=await Inbox.get_inbox_with_message(self.session,inbox_id) if not inbox: raise HTTPException(status_code=404,detail="Inbox not found") @@ -91,6 +93,13 @@ class Pipeline: }, commit=False, ) + await HistoryRecorder(self.session).record( + HistoryEvent.STAGE_CHANGED.value, + current_user=current_user,inbox_id=inbox.id,user_id=inbox.user_id, + entity_type="application",entity_id=inbox.id, + from_value=from_stage,to_value=stage.value, + description=change_reason,commit=False, + ) message.application_status=stage self.session.add(message) await self.session.commit() @@ -101,7 +110,7 @@ class Pipeline: "transition":serialize_stage_transition(transition), } - async def _change_manual_stage(self,manual_upload_id,stage,changed_by,change_reason): + async def _change_manual_stage(self,manual_upload_id,stage,changed_by,change_reason,current_user): row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,manual_upload_id) if not row: raise HTTPException(status_code=404,detail="Manual upload candidate not found") @@ -124,6 +133,13 @@ class Pipeline: }, commit=False, ) + await HistoryRecorder(self.session).record( + HistoryEvent.STAGE_CHANGED.value, + current_user=current_user,manual_upload_candidate_id=row.id,user_id=row.user_id, + entity_type="application",entity_id=row.id, + from_value=from_stage,to_value=stage.value, + description=change_reason,commit=False, + ) row.status=stage.value row.updated_at=_now() self.session.add(row) diff --git a/docker-compose.yml b/docker-compose.yml index bcb6da6..346226b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -44,7 +44,9 @@ x-backend-env: &backend-env # compose network on 5432, not the published host port. DB_HOST: ${DB_HOST:-host.docker.internal} REDIS_URL: redis://redis:6379/0 - EMAIL_URL: http://host.docker.internal:5000 + # Prefer root/.env EMAIL_URL (e.g. http://3.140.173.13:5000). Fall back to the + # host-gateway alias when the mail service runs on this machine's :5000. + EMAIL_URL: ${EMAIL_URL:-http://host.docker.internal:5000} BACKEND_URL: http://backend-api:8000 # The one shared folder. Every process that decodes, scores or serves a CV reads and diff --git a/frontend/src/api/candidates.js b/frontend/src/api/candidates.js index 862eb84..72d4c1b 100644 --- a/frontend/src/api/candidates.js +++ b/frontend/src/api/candidates.js @@ -316,6 +316,16 @@ export function createActivity({ inboxId, type, status, description }) { }) } +/** + * Audit log for one candidate. UNLIKE the four child collections above, history is + * NOT bundled into the detail payload: it is append-only and unbounded, and the + * detail query refetches on every write in the modal. Fetched lazily when the + * History tab opens, paginated server-side. + */ +export function listHistory(userId, { limit = 200, offset = 0 } = {}) { + return request('/candidate/history/fetch', { params: { user_id: userId, limit, offset } }) +} + /** * Authenticated attachment download. Never send a filesystem path — the server * resolves by owning record + index. `inboxId` is the `inbox` table PK (int), diff --git a/frontend/src/lib/queryKeys.js b/frontend/src/lib/queryKeys.js index 299cea5..b1195d7 100644 --- a/frontend/src/lib/queryKeys.js +++ b/frontend/src/lib/queryKeys.js @@ -68,6 +68,7 @@ export const qk = { all: () => ['candidates'], list: (p = {}) => ['candidates', 'list', p], detail: (id) => ['candidates', 'detail', id], + history: (id, p = {}) => ['candidates', 'history', id, p], }, // Board rows come from the same endpoint as qk.candidates.list but are cached // MAPPED (kanban cards, not the raw envelope), so they need their own key — diff --git a/frontend/src/screens/CandidateProfile.jsx b/frontend/src/screens/CandidateProfile.jsx index d68fc32..7463765 100644 --- a/frontend/src/screens/CandidateProfile.jsx +++ b/frontend/src/screens/CandidateProfile.jsx @@ -1,4 +1,4 @@ -/* The 8-tab candidate profile modal, split out of Candidates.jsx — it was the +/* The 9-tab candidate profile modal, split out of Candidates.jsx — it was the single largest block in js/candidates.js and deserves its own file. TWO DATA MODES, selected by whether the caller passes a `userId`: @@ -10,7 +10,8 @@ match verdict, documents, and the four child collections (interviews, notes, activity, feedback). The write tabs POST to their own endpoints and invalidate this one query, so the whole modal repaints from a single - refetch. + refetch. History is fetched separately (GET /candidate/history/fetch) + when that tab opens — it is append-only and not part of the detail payload. Live collections are NEVER padded with the seed's demo rows. An empty tab gets an empty state, because inventing three scorecards for a real applicant is @@ -34,7 +35,7 @@ import { friendlyAuthError } from '../lib/errors' import * as candidatesApi from '../api/candidates' import { companies, fmtDate, moneyK, pick } from '../data/seed' -const TABS = ['Overview', 'Resume', 'Timeline', 'Interview', 'Notes', 'Activity', 'Documents', 'Feedback'] +const TABS = ['Overview', 'Resume', 'Timeline', 'History', 'Interview', 'Notes', 'Activity', 'Documents', 'Feedback'] const LABEL = { fontSize: 12, color: 'var(--text-3)', fontWeight: 600, textTransform: 'uppercase', marginBottom: 8 } const REVIEWS = ['Strong Hire', 'Hire', 'Lean Hire', 'No Hire'] const INTERVIEW_TYPES = ['Phone Screen', 'Technical', 'System Design', 'Culture Fit', 'Final Round'] @@ -96,6 +97,7 @@ function useProfileWrite({ userId, mutationFn, success, onDone }) { mutationFn, onSuccess: async (_data, vars) => { await qc.invalidateQueries({ queryKey: qk.candidates.detail(userId) }) + await qc.invalidateQueries({ queryKey: ['candidates', 'history', userId] }) toast(typeof success === 'function' ? success(vars) : success, 'success') onDone?.() }, @@ -385,6 +387,14 @@ export default function CandidateProfile({ )))} + {tab === 'History' && (guard || (live ? ( + + ) : ( + + Audit history is recorded for live candidates only. + + )))} + {tab === 'Interview' && (guard || (live ? ( ) : ( @@ -619,6 +629,139 @@ function TimelineTab({ live }) { ) } +const HISTORY_ICON = { + 'stage.changed': { icon: 'arrow-right', tone: 'i-indigo' }, + 'note.created': { icon: 'edit', tone: 'i-blue' }, + 'note.updated': { icon: 'edit', tone: 'i-blue' }, + 'feedback.created': { icon: 'award', tone: 'i-teal' }, + 'feedback.updated': { icon: 'award', tone: 'i-teal' }, + 'interview.created': { icon: 'calendar', tone: 'i-purple' }, + 'interview.updated': { icon: 'calendar', tone: 'i-purple' }, + 'calendar.created': { icon: 'send', tone: 'i-purple' }, + 'calendar.rescheduled': { icon: 'clock', tone: 'i-amber' }, + 'calendar.cancelled': { icon: 'x-circle', tone: 'i-red' }, + 'favorite.changed': { icon: 'star', tone: 'i-amber' }, + 'rating.changed': { icon: 'star', tone: 'i-amber' }, + 'candidate.created': { icon: 'user-plus', tone: 'i-green' }, + 'candidate.imported': { icon: 'upload', tone: 'i-green' }, + 'document.uploaded': { icon: 'paperclip', tone: 'i-red' }, + 'ats.scored': { icon: 'sparkles', tone: 'i-blue' }, +} + +const HISTORY_TITLE = { + 'stage.changed': 'Stage changed', + 'note.created': 'Note added', + 'note.updated': 'Note updated', + 'feedback.created': 'Feedback submitted', + 'feedback.updated': 'Feedback updated', + 'interview.created': 'Interview scheduled', + 'interview.updated': 'Interview updated', + 'calendar.created': 'Calendar invite sent', + 'calendar.rescheduled': 'Calendar rescheduled', + 'calendar.cancelled': 'Calendar cancelled', + 'favorite.changed': 'Favorite updated', + 'rating.changed': 'Rating updated', + 'candidate.created': 'Candidate created', + 'candidate.imported': 'Candidate imported', + 'document.uploaded': 'Document uploaded', + 'ats.scored': 'ATS scored', +} + +function historyDayLabel(value) { + const d = value instanceof Date ? value : new Date(value ?? NaN) + if (Number.isNaN(d.getTime())) return 'Unknown' + const today = new Date() + const yday = new Date() + yday.setDate(today.getDate() - 1) + const sameDay = (a, b) => a.getFullYear() === b.getFullYear() && a.getMonth() === b.getMonth() && a.getDate() === b.getDate() + if (sameDay(d, today)) return 'Today' + if (sameDay(d, yday)) return 'Yesterday' + return fmtDate(d) +} + +function HistoryTab({ userId }) { + const [limit, setLimit] = useState(200) + const q = useQuery({ + queryKey: qk.candidates.history(userId, { limit }), + queryFn: async () => { + const res = await candidatesApi.listHistory(userId, { limit }) + return { rows: res?.data ?? [], total: res?.total ?? 0 } + }, + enabled: Boolean(userId), + }) + + if (q.isPending) { + return Fetching the audit trail. + } + if (q.isError) { + return ( + + {friendlyAuthError(q.error, 'Please try again.')} + + ) + } + + const rows = q.data?.rows ?? [] + const total = q.data?.total ?? 0 + if (!rows.length) { + return Actions on this candidate will appear here. + } + + const groups = [] + for (const r of rows) { + const label = historyDayLabel(r.created_at) + const last = groups[groups.length - 1] + if (!last || last.label !== label) groups.push({ label, rows: [r] }) + else last.rows.push(r) + } + + return ( + <> + {groups.map((g) => ( +
+
{g.label}
+
+ {g.rows.map((r) => )} +
+
+ ))} + {total > rows.length && ( + + )} + + ) +} + +function HistoryRow({ row: r }) { + const look = HISTORY_ICON[r.event_type] || { icon: 'clock', tone: 'i-blue' } + const title = HISTORY_TITLE[r.event_type] || r.event_type + const change = [r.from_value, r.to_value].some((v) => v != null && v !== '') + ? `${r.from_value ?? '—'} → ${r.to_value ?? '—'}` + : null + const actor = r.actor_name || (r.actor_id ? 'Unknown' : 'System') + const when = [fmtWhen(r.created_at), fmtClock(r.created_at)].filter(Boolean).join(' · ') + return ( +
+ + + +
+
{title}
+ {change &&
{change}
} + {r.description &&
{r.description}
} +
{actor} · {when}
+
+
+ {r.actor_name || r.actor_id + ? + : System} +
+
+ ) +} + function InterviewTab({ userId, inboxId, rows }) { const { toast } = useToast() const [form, setForm] = useState({ type: INTERVIEW_TYPES[0], date: '', time: '', status: INTERVIEW_STATES[0] }) diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index 9ac144e..a32133a 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -744,6 +744,24 @@ export default function Inbox() { markDuplicate.mutate({ id: item.id, isDuplicate: !item.duplicate }) } + const sync = useMutation({ + mutationFn: () => inboxApi.syncMailbox(), + onSuccess: async (res) => { + await qc.invalidateQueries({ queryKey: qk.mailbox.all() }) + // /email/fetch reports what the intake gate did with the page. The key is + // additive and absent when the gate is disabled, so fall back to the old + // message rather than rendering "undefined filtered out". + const t = res?.triage + toast( + t + ? `Mailbox synced — ${t.ingested} imported, ${t.skipped} filtered out` + : 'Mailbox synced', + 'success', + ) + }, + onError: (err) => toast(friendlyAuthError(err, 'Sync failed'), 'error'), + }) + return (
@@ -755,12 +773,13 @@ export default function Inbox() { Microsoft Graph API · Connected