Claim: {claim_number}
"
+ f"Client: {client_number} — {company_name}
"
+ f"Debtor: {debtor_name}
"
+ f"Amount: ${dollars:,.2f}
Hello,
" + f"Click the button below to log in to your DRE client portal. " + f"This link expires in 15 minutes and can only be used once.
" + f"" + f"If the button doesn't work, copy this link: {link}
" + f"If you did not request this link, you can ignore this email.
" + ) + return send_email(to_addr, "Your DRE Portal Login Link", body, html) + + +def notify_team_message(claim_number: str, subject: str, content: str, + author: str) -> bool: + team = _env("DRE_TEAM_NOTIFY", "dre@debtrecoveryexperts.com") + body = ( + f"New client message on claim {claim_number}.\n\n" + f"From: {author}\n" + f"Subject: {subject}\n\n" + f"{content}\n" + ) + return send_email(team, f"Client message on {claim_number}: {subject}", body) diff --git a/backend/intake.py b/backend/intake.py new file mode 100644 index 0000000..8e3f805 --- /dev/null +++ b/backend/intake.py @@ -0,0 +1,147 @@ +"""Intake endpoint — POST /api/intake. Creates client (+reuse by email) + debtor + claim. +Writes audit rows, SYSTEM note, emails the DRE team (best-effort). +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Request, status +from pydantic import ValidationError + +from . import auth as authmod +from . import db +from . import dreemail +from .db import get_conn, new_uuid, next_sequence_number, utcnow_iso +from .models import IntakeRequest + +logger = logging.getLogger("dre.intake") +router = APIRouter() + + +def _money_display(cents: int) -> str: + return f"${cents / 100.0:,.2f}" + + +@router.post("/api/intake") +async def intake(request: Request): + # Rate limit + ip = authmod.get_client_ip(request) + if not authmod.rate_limit_intake(ip): + return _json_error("rate_limited", "Too many requests. Please try again later.", + status.HTTP_429_TOO_MANY_REQUESTS) + # Parse JSON body + try: + body = await request.json() + except Exception: + return _json_error("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY) + # Validate + try: + req = IntakeRequest.model_validate(body) + except ValidationError as exc: + return _json_error("validation_error", _format_validation_error(exc), + status.HTTP_422_UNPROCESSABLE_ENTITY) + + client_in = req.client + debtor_in = req.debtor + claim_in = req.claim + email_lc = client_in.email.lower() + + with get_conn() as conn: + now = utcnow_iso() + # Reuse client by email or create new + existing = conn.execute("SELECT id, client_number FROM clients WHERE email = ?", (email_lc,)).fetchone() + if existing is not None: + client_id = existing["id"] + client_number = existing["client_number"] + conn.execute( + "UPDATE clients SET company_name = ?, contact_name = ?, phone = ?, tos_accepted_at = COALESCE(tos_accepted_at, ?), updated_at = ? WHERE id = ?", + (client_in.company_name, client_in.contact_name, client_in.phone, now, now, client_id), + ) + # audit update + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (new_uuid(), "client", client_id, "update", "contact_info", None, None, "intake", "reused existing client", now), + ) + else: + client_id = new_uuid() + client_number = next_sequence_number(conn, "CLT") + conn.execute( + "INSERT INTO clients (id, client_number, company_name, contact_name, email, phone, tos_accepted_at, twentycrm_id, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)", + (client_id, client_number, client_in.company_name, client_in.contact_name, + email_lc, client_in.phone, now, now, now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (new_uuid(), "client", client_id, "create", "intake", now), + ) + + # Debtor (always new per spec — denormalized avoided, but new claim = new debtor row) + debtor_id = new_uuid() + conn.execute( + "INSERT INTO debtors (id, name, business_type, contact_email, contact_phone, physical_address, twentycrm_id, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?)", + (debtor_id, debtor_in.name, debtor_in.business_type, debtor_in.contact_email, + debtor_in.contact_phone, debtor_in.physical_address, now, now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (new_uuid(), "debtor", debtor_id, "create", "intake", now), + ) + + # Claim + claim_id = new_uuid() + claim_number = next_sequence_number(conn, "DRE") + conn.execute( + "INSERT INTO claims (id, claim_number, client_id, debtor_id, amount_cents, currency, status, tier, description, client_reference, invoice_date, date_assigned, date_resolved, twentycrm_id, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, 'USD', 'NEW', 'TIER_1', ?, ?, ?, NULL, NULL, NULL, ?, ?)", + (claim_id, claim_number, client_id, debtor_id, claim_in.amount_cents, + claim_in.description, claim_in.client_reference, claim_in.invoice_date, now, now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, old_value, new_value, actor, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (new_uuid(), "claim", claim_id, "create", None, claim_number, "intake", now), + ) + + # SYSTEM shared note + note_id = new_uuid() + conn.execute( + "INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) " + "VALUES (?, ?, 'SYSTEM', 'System', NULL, 'Claim received.', 'SHARED', NULL, ?)", + (note_id, claim_id, now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (new_uuid(), "note", note_id, "note_add", "system", now), + ) + conn.commit() + + # Best-effort team notification + try: + dreemail.notify_team_intake(claim_number, client_number, client_in.company_name, + claim_in.amount_cents, debtor_in.name) + except Exception as exc: # noqa: BLE001 + logger.error("intake team notify failed: %s", exc) + + return { + "claim_number": claim_number, + "client_number": client_number, + "status": "NEW", + "message": "Claim received. Our team will review and contact you shortly.", + } + + +def _format_validation_error(exc: ValidationError) -> str: + parts = [] + for err in exc.errors(): + loc = ".".join(str(x) for x in err["loc"]) + parts.append(f"{loc}: {err['msg']}") + return "; ".join(parts) + + +def _json_error(code: str, message: str, status_code: int): + from fastapi.responses import JSONResponse + return JSONResponse( + status_code=status_code, + content={"error": {"code": code, "message": message}}, + ) diff --git a/backend/main.py b/backend/main.py new file mode 100644 index 0000000..6a4561f --- /dev/null +++ b/backend/main.py @@ -0,0 +1,195 @@ +"""DRE Customer Portal API — FastAPI app + all 18 routers. +Port 127.0.0.1:8093. systemd: dre-portal.service (User=root). +""" +from __future__ import annotations + +import logging + +from fastapi import Depends, FastAPI, HTTPException, Request, status +from fastapi.exceptions import RequestValidationError +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from pydantic import ValidationError + +from . import auth as authmod +from . import db +from . import dreemail +from .db import get_conn, utcnow_iso + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s") +logger = logging.getLogger("dre.main") + +app = FastAPI(title="DRE Customer Portal API", version="1.0.0", docs_url="/docs", redoc_url=None) + +app.add_middleware( + CORSMiddleware, + allow_origins=["https://portal.debtrecoveryexperts.com", "http://127.0.0.1:8093"], + allow_methods=["GET", "POST", "PATCH"], + allow_headers=["*"], +) + +# Include routers (intake, claims, staff) +from .intake import router as intake_router # noqa: E402 +from .claims import router as claims_router # noqa: E402 +from .staff import router as staff_router # noqa: E402 + +app.include_router(intake_router) +app.include_router(claims_router) +app.include_router(staff_router) + + +def _err(code: str, message: str, status_code: int): + return JSONResponse(status_code=status_code, + content={"error": {"code": code, "message": message}}) + + +@app.on_event("startup") +async def _startup(): + db.init_db() + logger.info("DRE portal started; db=%s", db.get_db_path()) + + +# --------------------------------------------------------------- +# Error envelopes +# --------------------------------------------------------------- +@app.exception_handler(RequestValidationError) +async def _validation_handler(request: Request, exc: RequestValidationError): + errors = exc.errors() + parts = [] + for e in errors: + loc = ".".join(str(x) for x in e.get("loc", [])) + parts.append(f"{loc}: {e.get('msg', 'invalid')}") + msg = "; ".join(parts) if parts else "Validation error" + return _err("validation_error", msg, status.HTTP_422_UNPROCESSABLE_ENTITY) + + +@app.exception_handler(ValidationError) +async def _pyd_validation_handler(request: Request, exc: ValidationError): + parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()] + return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY) + + +@app.exception_handler(HTTPException) +async def _http_handler(request: Request, exc: HTTPException): + """Map auth/forbidden HTTPExceptions onto the unified {'error':{...}} envelope.""" + detail = exc.detail + if isinstance(detail, dict) and "code" in detail and "message" in detail: + return _err(detail["code"], detail["message"], exc.status_code) + return _err("internal_error", "An internal error occurred.", exc.status_code) + + +@app.exception_handler(Exception) +async def _internal_handler(request: Request, exc: Exception): + logger.exception("internal error: %s", exc) + return _err("internal_error", "An internal error occurred.", status.HTTP_500_INTERNAL_SERVER_ERROR) + + +# --------------------------------------------------------------- +# 1. GET /api/health +# --------------------------------------------------------------- +@app.get("/api/health") +async def health(): + return {"status": "ok", "time": utcnow_iso()} + + +# --------------------------------------------------------------- +# 2-5. Auth endpoints (magic-link) +# --------------------------------------------------------------- +@app.post("/api/auth/request") +async def auth_request(request: Request): + try: + body = await request.json() + except Exception: + return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY) + email_in = body.get("email") if isinstance(body, dict) else None + if not email_in: + return _err("validation_error", "email is required.", status.HTTP_422_UNPROCESSABLE_ENTITY) + ip = authmod.get_client_ip(request) + if not authmod.rate_limit_auth_request(email_in.lower(), ip): + return _err("rate_limited", "Too many requests. Please try again later.", + status.HTTP_429_TOO_MANY_REQUESTS) + email_lc = email_in.lower() + with get_conn() as conn: + row = conn.execute("SELECT id, contact_name, company_name FROM clients WHERE email = ?", (email_lc,)).fetchone() + if row is not None: + raw_token = authmod.create_magic_token(conn, row["id"], ip) + conn.commit() + # Best-effort email + try: + dreemail.send_magic_link(email_lc, raw_token, "") + except Exception as exc: # noqa: BLE001 + logger.error("magic link email failed: %s", exc) + # Anti-enumeration: always same response + return {"message": "If an account exists, a login link has been sent."} + + +@app.post("/api/auth/verify") +async def auth_verify(request: Request): + try: + body = await request.json() + except Exception: + return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY) + token = body.get("token") if isinstance(body, dict) else None + if not token: + return _err("validation_error", "token is required.", status.HTTP_422_UNPROCESSABLE_ENTITY) + ip = authmod.get_client_ip(request) + if not authmod.rate_limit_auth_verify(ip): + return _err("rate_limited", "Too many attempts. Please try again later.", + status.HTTP_429_TOO_MANY_REQUESTS) + with get_conn() as conn: + client_id = authmod.verify_magic_token(conn, token) + if client_id is None: + return _err("unauthorized", "Invalid or expired token.", status.HTTP_401_UNAUTHORIZED) + session_token, expires_at = authmod.create_session(conn, client_id) + client = conn.execute( + "SELECT client_number, company_name, contact_name FROM clients WHERE id = ?", (client_id,) + ).fetchone() + conn.commit() + return { + "session_token": session_token, + "expires_at": expires_at, + "client": { + "client_number": client["client_number"], + "company_name": client["company_name"], + "contact_name": client["contact_name"], + }, + } + + +@app.post("/api/auth/logout") +async def auth_logout(request: Request, session: dict = Depends(authmod.require_client)): + """Revoke the current session by hash.""" + auth = request.headers.get("authorization", "") + raw_token = auth.split(" ", 1)[1].strip() if auth.lower().startswith("bearer ") else "" + import hashlib + session_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest() + with get_conn() as conn: + conn.execute( + "UPDATE sessions SET revoked_at = ? WHERE session_hash = ?", + (utcnow_iso(), session_hash), + ) + conn.commit() + return {"message": "Logged out."} + + +@app.get("/api/auth/me") +async def auth_me(session: dict = Depends(authmod.require_client)): + import hashlib + client_id = session["client_id"] + with get_conn() as conn: + c = conn.execute( + "SELECT client_number, company_name, contact_name, email, phone, created_at FROM clients WHERE id = ?", + (client_id,), + ).fetchone() + claim_count = conn.execute("SELECT COUNT(*) AS n FROM claims WHERE client_id = ?", (client_id,)).fetchone()["n"] + return { + "client": { + "client_number": c["client_number"], + "company_name": c["company_name"], + "contact_name": c["contact_name"], + "email": c["email"], + "phone": c["phone"], + "member_since": c["created_at"], + }, + "claim_count": claim_count, + } diff --git a/backend/models.py b/backend/models.py new file mode 100644 index 0000000..dded1c4 --- /dev/null +++ b/backend/models.py @@ -0,0 +1,190 @@ +"""Pydantic v2 request/response models. extra='forbid' on every request body. +Includes PII rejection (SSN/PAN regex) on all free-text fields. +""" +from __future__ import annotations + +import re +from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator + +# --------------------------------------------------------------- +# PII rejection — compliance-critical +# --------------------------------------------------------------- +_SSN_RE = re.compile(r"\b\d{3}-?\d{2}-?\d{4}\b") +_PAN_RE = re.compile(r"\b(?:\d[ -]?){13,19}\b") + + +def _scan_pii(value: str) -> str: + """Raise ValueError if value matches SSN or PAN regex.""" + if value is None: + return value + if _SSN_RE.search(value): + raise ValueError("Do not include Social Security or bank/card numbers.") + if _PAN_RE.search(value): + raise ValueError("Do not include Social Security or bank/card numbers.") + return value + + +def _pii_validator(field_name: str): + return field_validator(field_name)(lambda v: _scan_pii(v)) + + +# --------------------------------------------------------------- +# Intake request +# --------------------------------------------------------------- +BUSINESS_TYPES = ( + "INDIVIDUAL", "SOLE_PROPRIETORSHIP", "LLC", "CORPORATION", "PARTNERSHIP", "OTHER" +) +MESSAGE_SUBJECTS = ( + "Question about my claim", + "New information about the debtor", + "Payment received / want to stop recovery", + "Update my contact info", + "Complaint or concern", + "Other", +) +CLAIM_STATUSES = ( + "NEW", "UNDER_REVIEW", "ACTIVE", "NEGOTIATION", "LEGAL", + "SETTLED", "CLOSED", "WRITE_OFF", "REJECTED", +) +TIERS = ("TIER_1", "TIER_2", "TIER_2_5", "TIER_3", "TIER_4") + + +class IntakeClient(BaseModel): + model_config = ConfigDict(extra="forbid") + company_name: str = Field(..., min_length=1, max_length=200) + contact_name: str = Field(..., min_length=1, max_length=200) + email: EmailStr + phone: str | None = Field(None, max_length=50) + + @field_validator("company_name", "contact_name", "phone") + @classmethod + def _v(cls, v): + return _scan_pii(v) + + +class IntakeDebtor(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str = Field(..., min_length=1, max_length=200) + business_type: str = Field("OTHER") + contact_email: str | None = Field(None, max_length=254) + contact_phone: str | None = Field(None, max_length=50) + physical_address: str | None = Field(None, max_length=500) + + @field_validator("business_type") + @classmethod + def _bt(cls, v): + v = v.upper() + if v not in BUSINESS_TYPES: + raise ValueError(f"business_type must be one of {BUSINESS_TYPES}") + return v + + @field_validator("name", "contact_email", "contact_phone", "physical_address") + @classmethod + def _v(cls, v): + return _scan_pii(v) + + +class IntakeClaim(BaseModel): + model_config = ConfigDict(extra="forbid") + amount_cents: int = Field(..., gt=0, le=100_000_000) + description: str | None = Field(None, max_length=5000) + client_reference: str | None = Field(None, max_length=200) + invoice_date: str | None = Field(None, max_length=20) + + @field_validator("description", "client_reference", "invoice_date") + @classmethod + def _v(cls, v): + return _scan_pii(v) + + +class IntakeRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + client: IntakeClient + debtor: IntakeDebtor + claim: IntakeClaim + tos_accepted: bool = True + turnstile_token: str | None = None + + @field_validator("tos_accepted") + @classmethod + def _tos(cls, v): + if v is not True: + raise ValueError("tos_accepted must be true") + return v + + +# --------------------------------------------------------------- +# Auth +# --------------------------------------------------------------- +class AuthRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + email: EmailStr + + +class AuthVerify(BaseModel): + model_config = ConfigDict(extra="forbid") + token: str = Field(..., min_length=10, max_length=200) + + +# --------------------------------------------------------------- +# Messages +# --------------------------------------------------------------- +class MessageCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + subject: str + content: str = Field(..., min_length=1, max_length=10000) + + @field_validator("subject") + @classmethod + def _subj(cls, v): + if v not in MESSAGE_SUBJECTS: + raise ValueError(f"subject must be one of {MESSAGE_SUBJECTS}") + return v + + @field_validator("content") + @classmethod + def _cont(cls, v): + return _scan_pii(v) + + +# --------------------------------------------------------------- +# Staff +# --------------------------------------------------------------- +class StaffClaimPatch(BaseModel): + model_config = ConfigDict(extra="forbid") + status: str | None = None + tier: str | None = None + reason: str | None = Field(None, max_length=500) + + @field_validator("status") + @classmethod + def _st(cls, v): + if v is not None and v not in CLAIM_STATUSES: + raise ValueError(f"status must be one of {CLAIM_STATUSES}") + return v + + @field_validator("tier") + @classmethod + def _tr(cls, v): + if v is not None and v not in TIERS: + raise ValueError(f"tier must be one of {TIERS}") + return v + + +class StaffNoteCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + content: str = Field(..., min_length=1, max_length=10000) + visibility: str = "INTERNAL" + author_name: str = Field(..., min_length=1, max_length=200) + + @field_validator("visibility") + @classmethod + def _vis(cls, v): + if v not in ("SHARED", "INTERNAL"): + raise ValueError("visibility must be SHARED or INTERNAL") + return v + + @field_validator("content", "author_name") + @classmethod + def _v(cls, v): + return _scan_pii(v) diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..7c3d73b --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.110 +uvicorn[standard]>=0.29 +python-multipart>=0.0.9 +email-validator>=2.0 +pydantic>=2.6 diff --git a/backend/schema.sql b/backend/schema.sql new file mode 100644 index 0000000..6849384 --- /dev/null +++ b/backend/schema.sql @@ -0,0 +1,139 @@ +-- ============================================================ +-- DRE Customer Portal — SQLite Schema (spec §1) +-- DB file: /opt/dre-portal/data/dre.db +-- Pragmas (set on every connection): foreign_keys=ON, journal_mode=WAL, busy_timeout=5000 +-- All timestamps ISO-8601 UTC TEXT. Money as INTEGER cents. PKs TEXT UUID4. +-- ============================================================ + +CREATE TABLE IF NOT EXISTS clients ( + id TEXT PRIMARY KEY, -- uuid4 + client_number TEXT UNIQUE NOT NULL, -- CLT-YYYY-NNNN + company_name TEXT NOT NULL, + contact_name TEXT NOT NULL, + email TEXT UNIQUE NOT NULL, -- lowercased; magic-link identity + phone TEXT, + tos_accepted_at TEXT, -- set when ToS accepted at intake + twentycrm_id TEXT, -- nullable; set by future CRM sync + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_clients_email ON clients(email); + +CREATE TABLE IF NOT EXISTS debtors ( + id TEXT PRIMARY KEY, -- uuid4 + name TEXT NOT NULL, -- business or individual name + business_type TEXT NOT NULL DEFAULT 'OTHER' + CHECK (business_type IN + ('INDIVIDUAL','SOLE_PROPRIETORSHIP','LLC','CORPORATION','PARTNERSHIP','OTHER')), + contact_email TEXT, + contact_phone TEXT, + physical_address TEXT, -- free-text single line + twentycrm_id TEXT, -- nullable + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS claims ( + id TEXT PRIMARY KEY, -- uuid4 + claim_number TEXT UNIQUE NOT NULL, -- DRE-YYYY-NNNN + client_id TEXT NOT NULL REFERENCES clients(id) ON DELETE RESTRICT, + debtor_id TEXT NOT NULL REFERENCES debtors(id) ON DELETE RESTRICT, + amount_cents INTEGER NOT NULL CHECK (amount_cents > 0), + currency TEXT NOT NULL DEFAULT 'USD', + status TEXT NOT NULL DEFAULT 'NEW' + CHECK (status IN + ('NEW','UNDER_REVIEW','ACTIVE','NEGOTIATION','LEGAL','SETTLED','CLOSED','WRITE_OFF','REJECTED')), + tier TEXT NOT NULL DEFAULT 'TIER_1' + CHECK (tier IN ('TIER_1','TIER_2','TIER_2_5','TIER_3','TIER_4')), + description TEXT, + client_reference TEXT, + invoice_date TEXT, -- ISO date + date_assigned TEXT, -- set when moved out of NEW + date_resolved TEXT, -- set on SETTLED/CLOSED/WRITE_OFF + twentycrm_id TEXT, -- nullable + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_claims_client ON claims(client_id); +CREATE INDEX IF NOT EXISTS idx_claims_status ON claims(status); +CREATE INDEX IF NOT EXISTS idx_claims_debtor ON claims(debtor_id); + +CREATE TABLE IF NOT EXISTS documents ( + id TEXT PRIMARY KEY, -- uuid4 + claim_id TEXT NOT NULL REFERENCES claims(id) ON DELETE CASCADE, + original_name TEXT NOT NULL, -- sanitized display name + stored_path TEXT NOT NULL, -- absolute path on disk (uuid-named) + mime_type TEXT NOT NULL, + size_bytes INTEGER NOT NULL, + sha256 TEXT NOT NULL, -- integrity + dedupe + uploaded_by TEXT NOT NULL DEFAULT 'CLIENT' -- CLIENT | STAFF + CHECK (uploaded_by IN ('CLIENT','STAFF')), + twentycrm_id TEXT, -- nullable + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_documents_claim ON documents(claim_id); + +CREATE TABLE IF NOT EXISTS case_notes ( + id TEXT PRIMARY KEY, -- uuid4 + claim_id TEXT NOT NULL REFERENCES claims(id) ON DELETE CASCADE, + author_type TEXT NOT NULL + CHECK (author_type IN ('CLIENT','STAFF','SYSTEM')), + author_name TEXT NOT NULL, + subject TEXT, -- for client->team structured messages + content TEXT NOT NULL, -- plaintext; rendered escaped + visibility TEXT NOT NULL DEFAULT 'SHARED' + CHECK (visibility IN ('SHARED','INTERNAL')), + twentycrm_id TEXT, -- nullable + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_notes_claim ON case_notes(claim_id); + +CREATE TABLE IF NOT EXISTS auth_tokens ( + id TEXT PRIMARY KEY, -- uuid4 + client_id TEXT NOT NULL REFERENCES clients(id) ON DELETE CASCADE, + token_hash TEXT UNIQUE NOT NULL, -- sha256 of raw token + expires_at TEXT NOT NULL, -- created_at + 15 min + consumed_at TEXT, -- NULL = unused + requested_ip TEXT, + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_tokens_hash ON auth_tokens(token_hash); +CREATE INDEX IF NOT EXISTS idx_tokens_client ON auth_tokens(client_id); + +CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, -- uuid4 + client_id TEXT NOT NULL REFERENCES clients(id) ON DELETE CASCADE, + session_hash TEXT UNIQUE NOT NULL, -- sha256 of raw session token + expires_at TEXT NOT NULL, -- + 7 days + revoked_at TEXT, + created_at TEXT NOT NULL, + last_seen_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_sessions_hash ON sessions(session_hash); + +CREATE TABLE IF NOT EXISTS audit_log ( + id TEXT PRIMARY KEY, -- uuid4 + entity_type TEXT NOT NULL, -- 'claim'|'client'|'document'|'note' + entity_id TEXT NOT NULL, + action TEXT NOT NULL, -- 'create'|'status_change'|'update'|'upload'|'note_add' + field TEXT, + old_value TEXT, + new_value TEXT, + actor TEXT NOT NULL, + reason TEXT, + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_audit_entity ON audit_log(entity_type, entity_id); + +CREATE TABLE IF NOT EXISTS number_sequences ( + prefix TEXT NOT NULL, -- 'DRE'|'CLT' + year INTEGER NOT NULL, + last_value INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (prefix, year) +); + +-- Migration tracking table +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TEXT NOT NULL +); diff --git a/backend/staff.py b/backend/staff.py new file mode 100644 index 0000000..6b69ce0 --- /dev/null +++ b/backend/staff.py @@ -0,0 +1,460 @@ +"""Internal staff endpoints (staff-key auth): claims list/detail/patch, notes, documents, stats, audit.""" +from __future__ import annotations + +import hashlib +import html +import logging +import os +import uuid + +from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status +from fastapi.responses import JSONResponse +from pydantic import ValidationError + +from . import auth as authmod +from . import db +from .db import get_conn, get_upload_dir, new_uuid, utcnow_iso +from .models import StaffClaimPatch, StaffNoteCreate +from .claims import ( # noqa: E402 + MAX_FILE_BYTES, ALLOWED_EXT, MAGIC_BYTES, EXT_TO_KIND, STATUS_LABELS, + TIER_STEPS, _money, _err, +) + +logger = logging.getLogger("dre.staff") +router = APIRouter() + + +# --------------------------------------------------------------- +# GET /api/staff/claims +# --------------------------------------------------------------- +@router.get("/api/staff/claims") +async def staff_list_claims(request: Request, _staff=Depends(authmod.require_staff)): + status_filter = request.query_params.get("status") + tier_filter = request.query_params.get("tier") + q = request.query_params.get("q", "").strip() + limit = min(int(request.query_params.get("limit", "50")), 200) + offset = max(int(request.query_params.get("offset", "0")), 0) + where = [] + params: list = [] + if status_filter: + where.append("c.status = ?") + params.append(status_filter) + if tier_filter: + where.append("c.tier = ?") + params.append(tier_filter) + if q: + where.append("(c.claim_number LIKE ? OR cl.company_name LIKE ? OR d.name LIKE ?)") + like = f"%{q}%" + params.extend([like, like, like]) + where_sql = ("WHERE " + " AND ".join(where)) if where else "" + sql = ( + "SELECT c.claim_number, c.status, c.tier, c.amount_cents, c.created_at, c.date_resolved, " + "c.date_assigned, cl.company_name, cl.client_number, d.name AS debtor_name, d.business_type " + "FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id " + f"{where_sql} ORDER BY c.created_at DESC LIMIT ? OFFSET ?" + ) + params.extend([limit, offset]) + with get_conn() as conn: + rows = conn.execute(sql, tuple(params)).fetchall() + total = conn.execute(f"SELECT COUNT(*) AS n FROM claims c JOIN clients cl ON cl.id=c.client_id JOIN debtors d ON d.id=c.debtor_id {where_sql}", tuple(params[:-2])).fetchone()["n"] + return { + "claims": [ + { + "claim_number": r["claim_number"], + "status": r["status"], + "status_label": STATUS_LABELS.get(r["status"], r["status"]), + "tier": r["tier"], + "amount_cents": r["amount_cents"], + "amount_display": _money(r["amount_cents"]), + "company_name": r["company_name"], + "client_number": r["client_number"], + "debtor_name": r["debtor_name"], + "business_type": r["business_type"], + "created_at": r["created_at"], + "date_assigned": r["date_assigned"], + "date_resolved": r["date_resolved"], + } + for r in rows + ], + "total": total, + "limit": limit, + "offset": offset, + } + + +# --------------------------------------------------------------- +# GET /api/staff/claims/{claim_number} +# --------------------------------------------------------------- +@router.get("/api/staff/claims/{claim_number}") +async def staff_get_claim(claim_number: str, _staff=Depends(authmod.require_staff)): + with get_conn() as conn: + row = conn.execute( + "SELECT c.*, cl.company_name, cl.contact_name, cl.email AS client_email, cl.phone AS client_phone, " + "cl.client_number, d.name AS debtor_name, d.business_type, d.contact_email AS debtor_email, " + "d.contact_phone AS debtor_phone, d.physical_address AS debtor_address " + "FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id " + "WHERE c.claim_number = ?", + (claim_number,), + ).fetchone() + if row is None: + return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND) + docs = conn.execute( + "SELECT id, original_name, size_bytes, mime_type, uploaded_by, created_at FROM documents WHERE claim_id = ?", + (row["id"],), + ).fetchall() + notes = conn.execute( + "SELECT author_type, author_name, subject, content, visibility, created_at FROM case_notes " + "WHERE claim_id = ? ORDER BY created_at ASC", + (row["id"],), + ).fetchall() + audit = conn.execute( + "SELECT entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at " + "FROM audit_log WHERE entity_type IN ('claim','document','note') AND entity_id = ? " + "ORDER BY created_at ASC", + (row["id"],), + ).fetchall() + return { + "claim_number": row["claim_number"], + "status": row["status"], + "status_label": STATUS_LABELS.get(row["status"], row["status"]), + "tier": row["tier"], + "tier_step": TIER_STEPS.get(row["tier"], 1), + "amount_cents": row["amount_cents"], + "amount_display": _money(row["amount_cents"]), + "description": row["description"], + "client_reference": row["client_reference"], + "invoice_date": row["invoice_date"], + "date_assigned": row["date_assigned"], + "date_resolved": row["date_resolved"], + "created_at": row["created_at"], + "client": { + "client_number": row["client_number"], + "company_name": row["company_name"], + "contact_name": row["contact_name"], + "email": row["client_email"], + "phone": row["client_phone"], + }, + "debtor": { + "name": row["debtor_name"], + "business_type": row["business_type"], + "contact_email": row["debtor_email"], + "contact_phone": row["debtor_phone"], + "physical_address": row["debtor_address"], + }, + "documents": [ + { + "id": d["id"], "original_name": d["original_name"], "size_bytes": d["size_bytes"], + "mime_type": d["mime_type"], "uploaded_by": d["uploaded_by"], "created_at": d["created_at"], + } for d in docs + ], + "notes": [ + { + "author_type": n["author_type"], "author_name": n["author_name"], "subject": n["subject"], + "content": html.escape(n["content"]), "visibility": n["visibility"], "created_at": n["created_at"], + } for n in notes + ], + "audit": [ + { + "entity_type": a["entity_type"], "entity_id": a["entity_id"], "action": a["action"], + "field": a["field"], "old_value": a["old_value"], "new_value": a["new_value"], + "actor": a["actor"], "reason": a["reason"], "created_at": a["created_at"], + } for a in audit + ], + } + + +# --------------------------------------------------------------- +# PATCH /api/staff/claims/{claim_number} +# --------------------------------------------------------------- +@router.patch("/api/staff/claims/{claim_number}") +async def staff_patch_claim(claim_number: str, request: Request, _staff=Depends(authmod.require_staff)): + try: + body = await request.json() + except Exception: + return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY) + try: + patch = StaffClaimPatch.model_validate(body) + except ValidationError as exc: + parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()] + return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY) + with get_conn() as conn: + row = conn.execute("SELECT * FROM claims WHERE claim_number = ?", (claim_number,)).fetchone() + if row is None: + return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND) + claim_id = row["id"] + now = utcnow_iso() + updates: list[tuple] = [] + new_status = row["status"] + new_tier = row["tier"] + date_assigned = row["date_assigned"] + date_resolved = row["date_resolved"] + # Status change + if patch.status is not None and patch.status != row["status"]: + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (new_uuid(), "claim", claim_id, "status_change", "status", row["status"], patch.status, "staff", patch.reason, now), + ) + new_status = patch.status + # set date_assigned on first move out of NEW + if row["status"] == "NEW" and patch.status != "NEW" and not date_assigned: + date_assigned = now + # set date_resolved on resolved states + if patch.status in ("SETTLED", "CLOSED", "WRITE_OFF", "REJECTED") and not date_resolved: + date_resolved = now + # auto-add SYSTEM/SHARED note so client sees status change + conn.execute( + "INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) " + "VALUES (?, ?, 'SYSTEM', 'System', NULL, ?, 'SHARED', NULL, ?)", + (new_uuid(), claim_id, f"Status updated to: {STATUS_LABELS.get(patch.status, patch.status)}", now), + ) + # Tier change + if patch.tier is not None and patch.tier != row["tier"]: + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (new_uuid(), "claim", claim_id, "status_change", "tier", row["tier"], patch.tier, "staff", patch.reason, now), + ) + new_tier = patch.tier + conn.execute( + "UPDATE claims SET status = ?, tier = ?, date_assigned = ?, date_resolved = ?, updated_at = ? WHERE id = ?", + (new_status, new_tier, date_assigned, date_resolved, now, claim_id), + ) + conn.commit() + # Return updated detail + return await staff_get_claim_inner(claim_number) + + +async def staff_get_claim_inner(claim_number: str): + """Reuse detail query without re-checking staff auth.""" + with get_conn() as conn: + row = conn.execute( + "SELECT c.*, cl.company_name, cl.contact_name, cl.email AS client_email, cl.phone AS client_phone, " + "cl.client_number, d.name AS debtor_name, d.business_type " + "FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id " + "WHERE c.claim_number = ?", + (claim_number,), + ).fetchone() + if row is None: + return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, + content={"error": {"code": "not_found", "message": "Claim not found."}}) + return { + "claim_number": row["claim_number"], + "status": row["status"], + "status_label": STATUS_LABELS.get(row["status"], row["status"]), + "tier": row["tier"], + "tier_step": TIER_STEPS.get(row["tier"], 1), + "amount_cents": row["amount_cents"], + "amount_display": _money(row["amount_cents"]), + "date_assigned": row["date_assigned"], + "date_resolved": row["date_resolved"], + "updated_at": row["updated_at"], + } + + +# --------------------------------------------------------------- +# POST /api/staff/claims/{claim_number}/notes +# --------------------------------------------------------------- +@router.post("/api/staff/claims/{claim_number}/notes") +async def staff_add_note(claim_number: str, request: Request, _staff=Depends(authmod.require_staff)): + try: + body = await request.json() + except Exception: + return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY) + try: + note = StaffNoteCreate.model_validate(body) + except ValidationError as exc: + parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()] + return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY) + with get_conn() as conn: + row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone() + if row is None: + return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND) + claim_id = row["id"] + now = utcnow_iso() + note_id = new_uuid() + conn.execute( + "INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) " + "VALUES (?, ?, 'STAFF', ?, NULL, ?, ?, NULL, ?)", + (note_id, claim_id, note.author_name, note.content, note.visibility, now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (new_uuid(), "note", note_id, "note_add", f"staff:{note.author_name}", now), + ) + conn.commit() + return { + "id": note_id, + "author_type": "STAFF", + "author_name": note.author_name, + "content": html.escape(note.content), + "visibility": note.visibility, + "created_at": now, + } + + +# --------------------------------------------------------------- +# POST /api/staff/claims/{claim_number}/documents +# --------------------------------------------------------------- +@router.post("/api/staff/claims/{claim_number}/documents") +async def staff_upload_document(claim_number: str, request: Request, file: UploadFile = File(...), + _staff=Depends(authmod.require_staff)): + cl = request.headers.get("content-length") + if cl and int(cl) > MAX_FILE_BYTES + 4096: + return _err("payload_too_large", "File exceeds 20 MB limit.", status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) + ext = os.path.splitext(file.filename or "")[1].lower() + if ext not in ALLOWED_EXT: + return _err("unsupported_media_type", "File type not allowed.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) + kind = EXT_TO_KIND.get(ext) + if not kind: + return _err("unsupported_media_type", "File type not allowed.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) + with get_conn() as conn: + row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone() + if row is None: + return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND) + claim_id = row["id"] + count = conn.execute("SELECT COUNT(*) AS n FROM documents WHERE claim_id = ?", (claim_id,)).fetchone()["n"] + if count >= 50: + return _err("conflict", "Document limit reached for this claim.", status.HTTP_409_CONFLICT) + upload_dir = os.path.join(get_upload_dir(), claim_id) + os.makedirs(upload_dir, exist_ok=True) + stored_uuid = str(uuid.uuid4()) + stored_name = f"{stored_uuid}{ext}" + stored_path = os.path.join(upload_dir, stored_name) + sha = hashlib.sha256() + total = 0 + magic_seen = False + with open(stored_path, "wb") as f: + while True: + chunk = await file.read(64 * 1024) + if not chunk: + break + total += len(chunk) + if total > MAX_FILE_BYTES: + f.close() + os.remove(stored_path) + return _err("payload_too_large", "File exceeds 20 MB limit.", status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) + if not magic_seen: + sigs = MAGIC_BYTES.get(kind, ()) + if sigs and any(chunk.startswith(s) for s in sigs): + magic_seen = True + elif sigs: + f.close() + os.remove(stored_path) + return _err("unsupported_media_type", "File content does not match extension.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) + sha.update(chunk) + f.write(chunk) + os.chmod(stored_path, 0o640) + now = utcnow_iso() + doc_id = new_uuid() + mime = file.content_type or "application/octet-stream" + conn.execute( + "INSERT INTO documents (id, claim_id, original_name, stored_path, mime_type, size_bytes, sha256, uploaded_by, twentycrm_id, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, 'STAFF', NULL, ?)", + (doc_id, claim_id, os.path.basename(file.filename or "file"), stored_path, mime, total, sha.hexdigest(), now), + ) + conn.execute( + "INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (new_uuid(), "document", doc_id, "upload", "staff", now), + ) + conn.commit() + return { + "id": doc_id, + "original_name": os.path.basename(file.filename or "file"), + "size_bytes": total, + "mime_type": mime, + "uploaded_by": "STAFF", + "created_at": now, + } + + +# --------------------------------------------------------------- +# GET /api/staff/stats +# --------------------------------------------------------------- +@router.get("/api/staff/stats") +async def staff_stats(_staff=Depends(authmod.require_staff)): + with get_conn() as conn: + total = conn.execute("SELECT COUNT(*) AS n FROM claims").fetchone()["n"] + by_status = {} + for r in conn.execute("SELECT status, COUNT(*) AS n FROM claims GROUP BY status").fetchall(): + by_status[r["status"]] = r["n"] + by_tier = {} + for r in conn.execute("SELECT tier, COUNT(*) AS n FROM claims GROUP BY tier").fetchall(): + by_tier[r["tier"]] = r["n"] + total_amount = conn.execute("SELECT COALESCE(SUM(amount_cents),0) AS s FROM claims").fetchone()["s"] + # recovered = claims in SETTLED/CLOSED + recovered = conn.execute( + "SELECT COALESCE(SUM(amount_cents),0) AS s FROM claims WHERE status IN ('SETTLED','CLOSED')" + ).fetchone()["s"] + open_amount = total_amount - recovered + # aging: claims not resolved, by age + now_iso = utcnow_iso() + over_30 = conn.execute( + "SELECT COUNT(*) AS n FROM claims WHERE date_resolved IS NULL AND created_at < ?", + (over_30_cutoff(),), + ).fetchone()["n"] + over_60 = conn.execute( + "SELECT COUNT(*) AS n FROM claims WHERE date_resolved IS NULL AND created_at < ?", + (over_60_cutoff(),), + ).fetchone()["n"] + return { + "total_claims": total, + "by_status": by_status, + "by_tier": by_tier, + "total_amount_cents": total_amount, + "total_amount_display": _money(total_amount), + "recovered_amount_cents": recovered, + "recovered_amount_display": _money(recovered), + "open_amount_cents": open_amount, + "open_amount_display": _money(open_amount), + "aging": {"over_30_days": over_30, "over_60_days": over_60}, + } + + +def over_30_cutoff() -> str: + from datetime import datetime, timedelta, timezone + return (datetime.now(timezone.utc) - timedelta(days=30)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def over_60_cutoff() -> str: + from datetime import datetime, timedelta, timezone + return (datetime.now(timezone.utc) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------- +# GET /api/staff/audit +# --------------------------------------------------------------- +@router.get("/api/staff/audit") +async def staff_audit(request: Request, _staff=Depends(authmod.require_staff)): + entity_type = request.query_params.get("entity_type") + entity_id = request.query_params.get("entity_id") + limit = min(int(request.query_params.get("limit", "100")), 500) + offset = max(int(request.query_params.get("offset", "0")), 0) + where = [] + params: list = [] + if entity_type: + where.append("entity_type = ?") + params.append(entity_type) + if entity_id: + where.append("entity_id = ?") + params.append(entity_id) + where_sql = ("WHERE " + " AND ".join(where)) if where else "" + with get_conn() as conn: + rows = conn.execute( + f"SELECT * FROM audit_log {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?", + tuple(params + [limit, offset]), + ).fetchall() + total = conn.execute(f"SELECT COUNT(*) AS n FROM audit_log {where_sql}", tuple(params)).fetchone()["n"] + return { + "audit": [ + { + "id": r["id"], "entity_type": r["entity_type"], "entity_id": r["entity_id"], + "action": r["action"], "field": r["field"], "old_value": r["old_value"], + "new_value": r["new_value"], "actor": r["actor"], "reason": r["reason"], + "created_at": r["created_at"], + } for r in rows + ], + "total": total, + "limit": limit, + "offset": offset, + } diff --git a/operations/architecture-2026-08-21.md b/operations/architecture-2026-08-21.md new file mode 100644 index 0000000..b059626 --- /dev/null +++ b/operations/architecture-2026-08-21.md @@ -0,0 +1,721 @@ +# DRE Customer Portal — Backend Architecture Specification + +**Author:** Claude Opus 4-8 (System Architect) +**Date:** 2026-08-21 +**Status:** BUILD-READY — hand off to GLM-5.2 (backend) + Sonnet 5 (frontend) +**Stack:** FastAPI + SQLite (single source of truth), magic-link auth, systemd + uvicorn behind Caddy + +--- + +## 0. Scope & Principles + +This spec defines the FIRST production backend for DRE. It replaces the 12 static mockups' dead +`