"""Sales router — opportunities, proposals, and contracts.""" from datetime import datetime, timezone from uuid import UUID from typing import Optional import arq from arq.connections import RedisSettings from fastapi import APIRouter, Body, Depends, File, Query, Request, UploadFile from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from src.core.config import settings from src.core.dependencies import require_min_role, require_tenant from src.core.constants import UserRole from src.core.schemas.response import success, paginated from src.core.exceptions import ValidationError from src.database.session import get_db from src.apps.auth.models.user import User from src.apps.sales.models.opportunity import Opportunity from src.apps.sales.schemas.requests import ( OpportunityCreateRequest, OpportunityUpdateRequest, OpportunityLostRequest, ContractWizardCreateRequest, ContractSignRequest, ProposalCreateRequest, ProposalSendRequest, ProposalStatusUpdateRequest, ) from src.apps.sales.schemas.responses import ( OpportunityResponse, ContractResponse, ContractDetailResponse, ProposalResponse, ) from src.apps.sales.services.opportunity_service import OpportunityService from src.apps.sales.services.proposal_service import ProposalService from src.apps.sales.services.contract_service import ContractService from src.apps.sales.services.contract_issue_service import ContractIssueService from src.apps.sales.services.contract_document_service import ( ContractDocumentService, MAX_FILE_SIZE_BYTES as MAX_SIGNED_DOCUMENT_UPLOAD_BYTES, ) from src.apps.sales.services.inquiry_service import InquiryService from src.apps.sales.services.stage_service import StageService from src.apps.sales.schemas.inquiry import ContactInquiryOut from src.apps.sales.schemas.log_call import LogCallRequest from src.apps.sales.schemas.log_email import LogEmailRequest from src.apps.sales.schemas.stage_transition import StageTransitionRequest, MarkLostRequest router = APIRouter(prefix="/sales", tags=["sales"]) # --------------------------------------------------------------------------- # Opportunities # --------------------------------------------------------------------------- @router.get("/opportunities/kanban") async def get_opportunities_kanban( search: Optional[str] = Query(None), assigned_to: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): kanban = await OpportunityService.get_kanban(db, tenant_id, search=search, assigned_to=assigned_to) return success(kanban) @router.get("/opportunities") async def list_opportunities( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), stage: Optional[str] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): filters = [ Opportunity.tenant_id == tenant_id, Opportunity.deleted_at.is_(None), ] if stage: filters.append(Opportunity.stage == stage) total = ( await db.execute(select(func.count()).select_from(Opportunity).where(*filters)) ).scalar_one() offset = (page - 1) * page_size result = await db.execute( select(Opportunity) .where(*filters) .order_by(Opportunity.created_at.desc()) .offset(offset) .limit(page_size) ) items = [] now = datetime.now(timezone.utc) for opp in result.scalars().all(): data = OpportunityResponse.model_validate(opp).model_dump() if opp.stage_entered_at: entered = opp.stage_entered_at if entered.tzinfo is None: entered = entered.replace(tzinfo=timezone.utc) data["days_in_stage"] = (now - entered).days items.append(data) return paginated(items, total, page, page_size) @router.get("/opportunities/{opportunity_id}") async def get_opportunity( opportunity_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): opp = await OpportunityService.get_by_id(db, tenant_id, opportunity_id) data = OpportunityResponse.model_validate(opp).model_dump() if opp.stage_entered_at: now = datetime.now(timezone.utc) entered = opp.stage_entered_at if entered.tzinfo is None: entered = entered.replace(tzinfo=timezone.utc) data["days_in_stage"] = (now - entered).days return success(data) @router.get("/opportunities/{opportunity_id}/activities") async def list_opportunity_activities( opportunity_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): activities = await OpportunityService.get_activities(db, tenant_id, opportunity_id) return success(activities) @router.post("/opportunities", status_code=201) async def create_opportunity( body: OpportunityCreateRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): opp = await OpportunityService.create(db, tenant_id, body.model_dump(), current_user) data = OpportunityResponse.model_validate(opp).model_dump() data["days_in_stage"] = 0 return success(data, "Opportunity created") @router.patch("/opportunities/{opportunity_id}") async def update_opportunity( opportunity_id: UUID, body: OpportunityUpdateRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): opp = await OpportunityService.update( db, tenant_id, opportunity_id, body.model_dump(exclude_unset=True), current_user ) data = OpportunityResponse.model_validate(opp).model_dump() if opp.stage_entered_at: now = datetime.now(timezone.utc) entered = opp.stage_entered_at if entered.tzinfo is None: entered = entered.replace(tzinfo=timezone.utc) data["days_in_stage"] = (now - entered).days return success(data) @router.delete("/opportunities/{opportunity_id}") async def delete_opportunity( opportunity_id: UUID, body: OpportunityLostRequest = Body(default_factory=OpportunityLostRequest), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): await OpportunityService.soft_delete( db, tenant_id, opportunity_id, body.lost_reason, current_user ) return success(None, "Opportunity marked as lost") # --------------------------------------------------------------------------- # Proposals # --------------------------------------------------------------------------- @router.post("/proposals", status_code=201) async def create_proposal( body: ProposalCreateRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): data = body.model_dump() # Convert line item objects to plain dicts for the service data["line_items"] = [li.model_dump() for li in body.line_items] proposal = await ProposalService.create(db, tenant_id, data, current_user) return success(ProposalResponse.model_validate(proposal), "Proposal created") @router.get("/proposals/{proposal_id}") async def get_proposal( proposal_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): proposal = await ProposalService.get_by_id(db, tenant_id, proposal_id) data = ProposalResponse.model_validate(proposal) # Compute is_expired: status=sent AND expires_at < now if proposal.status == "sent" and proposal.expires_at is not None: from datetime import timezone as _tz now = datetime.now(_tz.utc) exp = proposal.expires_at if exp.tzinfo is None: exp = exp.replace(tzinfo=_tz.utc) data.is_expired = exp < now return success(data) @router.post("/proposals/{proposal_id}/send") async def send_proposal( proposal_id: UUID, body: ProposalSendRequest = Body(default_factory=ProposalSendRequest), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): proposal = await ProposalService.send(db, tenant_id, proposal_id, current_user) return success(ProposalResponse.model_validate(proposal), "Proposal sent") @router.post("/proposals/{proposal_id}/accept") async def accept_proposal( proposal_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): proposal = await ProposalService.mark_accepted(db, tenant_id, proposal_id, current_user) return success(ProposalResponse.model_validate(proposal), "Proposal accepted") @router.patch("/proposals/{proposal_id}/status") async def decline_proposal( proposal_id: UUID, body: ProposalStatusUpdateRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """PATCH /proposals/{id}/status — decline a sent proposal.""" proposal = await ProposalService.decline(db, tenant_id, proposal_id, current_user) return success(ProposalResponse.model_validate(proposal), "Proposal declined") @router.post("/proposals/{proposal_id}/resend") async def resend_proposal( proposal_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """POST /proposals/{id}/resend — re-send the proposal email.""" proposal = await ProposalService.resend(db, tenant_id, proposal_id, current_user) return success( ProposalResponse.model_validate(proposal), f"Proposal resent to {proposal.to_email}", ) # --------------------------------------------------------------------------- # Contracts # --------------------------------------------------------------------------- @router.get("/contracts") async def list_contracts( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): items, total = await ContractService.list( db, tenant_id, page=page, page_size=page_size, status=status, search=search ) # Use list-safe variant: strips purchaser_signature_b64 and raw pdf_s3_key data = [ContractResponse.from_orm_list(c) for c in items] return paginated(data, total, page, page_size) @router.get("/contracts/{contract_id}") async def get_contract( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): contract = await ContractService.get_by_id(db, tenant_id, contract_id) # ContractDetailResponse includes purchaser_signature_b64 (single-record view only) detail = ContractDetailResponse.model_validate(contract) detail.pdf_available = bool(getattr(contract, "pdf_s3_key", None)) detail.signed_document_available = bool(getattr(contract, "signed_document_s3_key", None)) return success(detail) @router.post("/contracts", status_code=201) async def create_contract( body: ContractWizardCreateRequest, current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): data = body.model_dump() data["line_items"] = [li.model_dump() for li in body.line_items] contract = await ContractService.create(db, tenant_id, data, current_user) resp = ContractResponse.from_orm_list(contract) return success(resp, "Contract created") @router.post("/contracts/{contract_id}/sign") async def sign_contract( contract_id: UUID, body: ContractSignRequest, current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): arq_redis = None try: arq_redis = await arq.create_pool(RedisSettings.from_dsn(settings.REDIS_URL)) except Exception: pass try: contract = await ContractService.sign( db, tenant_id, contract_id, signature_b64=body.purchaser_signature_b64, witness_name=body.witness_name, current_user=current_user, arq_redis=arq_redis, ) finally: if arq_redis is not None: await arq_redis.aclose() detail = ContractDetailResponse.model_validate(contract) detail.pdf_available = bool(getattr(contract, "pdf_s3_key", None)) detail.signed_document_available = bool(getattr(contract, "signed_document_s3_key", None)) return success(detail, "Contract signed") @router.patch("/contracts/{contract_id}/issue") async def issue_contract( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Issue a signed contract: generate payment link, create invoices, email purchaser.""" arq_redis = None try: arq_redis = await arq.create_pool(RedisSettings.from_dsn(settings.REDIS_URL)) except Exception: pass try: result = await ContractIssueService.issue( db=db, tenant_id=tenant_id, contract_id=contract_id, current_user=current_user, arq_redis=arq_redis, ) finally: if arq_redis is not None: await arq_redis.aclose() return success(result, "Contract issued and emailed successfully.") @router.post("/contracts/{contract_id}/resend") async def resend_contract( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Re-enqueue the contract email for an already-issued contract.""" arq_redis = None try: arq_redis = await arq.create_pool(RedisSettings.from_dsn(settings.REDIS_URL)) except Exception: pass try: result = await ContractIssueService.resend( db=db, tenant_id=tenant_id, contract_id=contract_id, current_user=current_user, arq_redis=arq_redis, ) finally: if arq_redis is not None: await arq_redis.aclose() return success(result, "Contract email resent successfully.") @router.post("/contracts/{contract_id}/signed-document") async def upload_signed_contract_document( contract_id: UUID, request: Request, file: UploadFile = File(...), current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Proxy-upload the staff-supplied wet-signed/scanned contract PDF to S3.""" # Cheap early rejection via the declared Content-Length, ahead of the # service-layer byte-count check (which only runs after `file.read()` # below). Starlette has already parsed the multipart body into a spooled # temp file by the time `file` is bound, so this does not prevent that # parse — it's defense-in-depth against re-reading/uploading an oversized # payload, not a true pre-buffer guard. A spoofed/absent header still # falls through to the in-service size check (API4). content_length = request.headers.get("content-length") if content_length is not None: try: declared_size = int(content_length) except ValueError: declared_size = None if declared_size is not None and declared_size > MAX_SIGNED_DOCUMENT_UPLOAD_BYTES: raise ValidationError(message="File size exceeds the 20 MB limit.") mime_type = file.content_type or "application/octet-stream" content = await file.read() await ContractDocumentService.upload_signed_document( db=db, tenant_id=tenant_id, contract_id=contract_id, current_user=current_user, file_content=content, filename=file.filename or "signed-contract.pdf", mime_type=mime_type, ) # Re-fetch with line_items eager-loaded (ContractService.get_by_id) — the # contract instance returned by the service above only has the row-lock # query's columns loaded, and lazy-loading line_items during # model_validate() outside an awaited context raises MissingGreenlet. contract = await ContractService.get_by_id(db, tenant_id, contract_id) detail = ContractDetailResponse.model_validate(contract) detail.pdf_available = bool(getattr(contract, "pdf_s3_key", None)) detail.signed_document_available = bool(getattr(contract, "signed_document_s3_key", None)) return success(detail, "Signed document uploaded successfully.") @router.get("/contracts/{contract_id}/signed-document") async def get_signed_contract_document_url( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Return a short-lived presigned S3 GET URL for the uploaded signed document.""" presigned_url = await ContractDocumentService.get_signed_document_url( db=db, tenant_id=tenant_id, contract_id=contract_id, current_user=current_user, ) return success({"url": presigned_url, "expires_in": 600}) @router.delete("/contracts/{contract_id}") async def delete_contract( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.ADMINISTRATOR)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): await ContractService.soft_delete(db, tenant_id, contract_id, current_user) return success(None, "Contract deleted") @router.get("/contracts/{contract_id}/pdf") async def get_contract_pdf_url( contract_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Return a short-lived presigned S3 URL, or stream the PDF on-demand when S3 is not configured.""" import asyncio from fastapi import HTTPException from fastapi.responses import StreamingResponse from sqlalchemy.orm import selectinload from src.apps.sales.models.contract import Contract as _Contract from src.apps.sales.services.contract_pdf_service import build_contract_pdf_bytes # Re-fetch with line_items eagerly loaded (needed by PDF builder) result = await db.execute( select(_Contract) .options(selectinload(_Contract.line_items)) .where( _Contract.id == contract_id, _Contract.tenant_id == tenant_id, _Contract.deleted_at.is_(None), ) ) contract = result.scalar_one_or_none() if not contract: raise HTTPException(status_code=404, detail="Contract not found") # ── S3 path: return presigned URL ──────────────────────────────────────── if contract.pdf_s3_key and settings.AWS_ACCESS_KEY_ID and settings.AWS_SECRET_ACCESS_KEY: import boto3 from botocore.exceptions import NoCredentialsError, ClientError s3_key = contract.pdf_s3_key def _make_presigned() -> str: s3 = boto3.client( "s3", region_name=settings.AWS_REGION, aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, ) return s3.generate_presigned_url( "get_object", Params={"Bucket": settings.S3_BUCKET, "Key": s3_key}, ExpiresIn=600, ) try: presigned_url = await asyncio.to_thread(_make_presigned) return success({"url": presigned_url, "expires_in": 600}) except (NoCredentialsError, ClientError): pass # Fall through to on-demand generation # ── Fallback: generate on-demand and stream directly ───────────────────── pdf_bytes = await asyncio.to_thread(build_contract_pdf_bytes, contract) filename = f"{contract.contract_number}.pdf" return StreamingResponse( iter([pdf_bytes]), media_type="application/pdf", headers={"Content-Disposition": f'inline; filename="{filename}"'}, ) # --------------------------------------------------------------------------- # Fee Items (for wizard step 4 — fee catalog) # --------------------------------------------------------------------------- @router.get("/fee-items") async def list_fee_items( current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): from src.apps.settings.models.fee_item import FeeItem as FeeItemModel result = await db.execute( select(FeeItemModel).where( FeeItemModel.tenant_id == tenant_id, FeeItemModel.is_active.is_(True), ).order_by(FeeItemModel.category, FeeItemModel.name) ) items = result.scalars().all() return success([{ "id": str(fi.id), "name": fi.name, "category": fi.category, "description": getattr(fi, "description", None), "unit_price": str(fi.unit_price), "is_taxable": getattr(fi, "is_taxable", True), "is_active": fi.is_active, } for fi in items]) # --------------------------------------------------------------------------- # Inquiries (INDL-25) # --------------------------------------------------------------------------- @router.get("/inquiries") async def list_inquiries( status: Optional[str] = Query(None), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): items, total = await InquiryService.list_inquiries(db, tenant_id, status=status, page=page, page_size=page_size) data = [ContactInquiryOut.model_validate(i).model_dump() for i in items] return paginated(data, total, page, page_size) @router.get("/inquiries/count") async def count_new_inquiries( current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Unread ('new') inquiry count for the Sales > Inbox tab badge. Redis-cached for 15s.""" cache_key = f"count:inquiries-new:{tenant_id}" try: import redis.asyncio as aioredis r = aioredis.from_url(settings.REDIS_URL, decode_responses=True) cached = await r.get(cache_key) if cached is not None: await r.aclose() return success({"new": int(cached)}) count = await InquiryService.count_new(db, tenant_id) await r.set(cache_key, count, ex=15) await r.aclose() return success({"new": count}) except Exception: count = await InquiryService.count_new(db, tenant_id) return success({"new": count}) @router.post("/inquiries/log-call", status_code=201) async def log_call( body: LogCallRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): inq = await InquiryService.log_call(db, tenant_id, body, current_user) return success(ContactInquiryOut.model_validate(inq).model_dump(), "Call log saved.") @router.post("/inquiries/log-email", status_code=201) async def log_email( body: LogEmailRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): inq = await InquiryService.log_email(db, tenant_id, body, current_user) return success(ContactInquiryOut.model_validate(inq).model_dump(), "Email log saved.") @router.get("/inquiries/{inquiry_id}") async def get_inquiry( inquiry_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): inq = await InquiryService.get_by_id(db, tenant_id, inquiry_id) return success(ContactInquiryOut.model_validate(inq).model_dump()) @router.patch("/inquiries/{inquiry_id}") async def update_inquiry( inquiry_id: UUID, body: dict = Body(default_factory=dict), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): inq = await InquiryService.update_inquiry(db, tenant_id, inquiry_id, body) return success(ContactInquiryOut.model_validate(inq).model_dump()) @router.post("/inquiries/{inquiry_id}/convert", status_code=201) async def convert_inquiry( inquiry_id: UUID, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): inq, opp = await InquiryService.convert_to_opportunity(db, tenant_id, inquiry_id, current_user) return success({ "inquiry": ContactInquiryOut.model_validate(inq).model_dump(), "opportunity_id": str(opp.id), }, "Inquiry converted to opportunity.") # --------------------------------------------------------------------------- # Stage Transitions (INDL-26) # --------------------------------------------------------------------------- @router.patch("/opportunities/{opportunity_id}/stage") async def update_opportunity_stage( opportunity_id: UUID, body: StageTransitionRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): opp = await StageService.transition_stage(db, tenant_id, opportunity_id, body.stage) data = OpportunityResponse.model_validate(opp).model_dump() return success(data, f"Opportunity moved to {opp.stage.replace('_', ' ').title()}.") @router.post("/opportunities/{opportunity_id}/mark-lost") async def mark_opportunity_lost( opportunity_id: UUID, body: MarkLostRequest, current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): opp = await StageService.mark_lost(db, tenant_id, opportunity_id, body.loss_reason, body.loss_notes) data = OpportunityResponse.model_validate(opp).model_dump() return success(data, "Opportunity marked as lost.")