Add firm admin onboarding for partner manager and staff

app/modules/firm_admin_dashboard/service.py

@@ -1,16 +1,21 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-from datetime import date, datetime
+from datetime import date, datetime, timezone
 from typing import Any
 
 from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.core.security.passwords import hash_password
+from app.core.settings import get_settings
+from app.modules.core.iam.invite_service import issue_invite_token
+from app.modules.email_integration.services import send_user_invite_email
 from app.modules.core.iam.models import User
 from app.modules.core.rbac.models import Role, UserRole
 from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant
 from app.modules.clients.models import Client
+from app.modules.employees.models import Employee
 from app.modules.services.models import (
     FirmServiceSelection,
     FirmServiceTaskTemplate,
@@ -150,6 +155,9 @@ def _user_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
     ).scalars().all()
     rows: list[dict[str, Any]] = []
     for user in users:
+        employee = db.execute(
+            select(Employee).where(Employee.tenant_id == tenant_id, Employee.user_id == user.id)
+        ).scalar_one_or_none()
         rows.append(
             {
                 "id": user.id,
@@ -159,6 +167,9 @@ def _user_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
                 "mobile": user.mobile,
                 "branch_id": user.branch_id,
                 "roles": get_user_role_names(db, user.id),
+                "employee_id": employee.id if employee else None,
+                "employee_linked": bool(employee),
+                "employee_code": employee.employee_code if employee else None,
                 "is_active": user.is_active,
                 "allow_login": user.allow_login,
                 "is_locked": user.is_locked,
@@ -247,6 +258,268 @@ def _smtp_status(db: Session, tenant_id: int | None) -> dict[str, Any]:
     return {"available": False, "configured": False, "label": "Check settings", "message": "Open email settings to verify SMTP."}
 
 
+
+ONBOARDING_ROLE_CONFIG: dict[str, dict[str, str]] = {
+    "partner": {
+        "role_name": "Partner",
+        "label": "Partner",
+        "default_designation": "Partner",
+        "default_department": "Management",
+        "default_employment_type": "full_time",
+    },
+    "manager": {
+        "role_name": "Branch Manager",
+        "label": "Manager",
+        "default_designation": "Branch Manager",
+        "default_department": "Operations",
+        "default_employment_type": "full_time",
+    },
+    "staff": {
+        "role_name": "Staff",
+        "label": "Staff",
+        "default_designation": "Staff",
+        "default_department": "Operations",
+        "default_employment_type": "full_time",
+    },
+}
+
+
+def _public_invite_url(invite_token: str) -> str:
+    base = (get_settings().ERP_PUBLIC_BASE_URL or "").strip().rstrip("/") or "http://localhost:8000"
+    return f"{base}/invite/accept?token={invite_token}"
+
+
+def _parse_optional_date(value: str | None) -> date | None:
+    value = (value or "").strip()
+    if not value:
+        return None
+    return date.fromisoformat(value)
+
+
+def _next_employee_code(db: Session, tenant_id: int) -> str:
+    prefix = "EMP"
+    rows = db.execute(
+        select(Employee.employee_code)
+        .where(Employee.tenant_id == int(tenant_id), Employee.employee_code.ilike(f"{prefix}%"))
+        .order_by(Employee.employee_code.desc())
+    ).all()
+    max_no = 0
+    for (code,) in rows:
+        suffix = "".join(ch for ch in str(code or "") if ch.isdigit())
+        if suffix:
+            max_no = max(max_no, int(suffix))
+    return f"{prefix}{max_no + 1:05d}"
+
+
+def _ensure_employee_for_onboarded_user(
+    db: Session,
+    *,
+    actor: User,
+    user_obj: User,
+    employee_code: str | None,
+    mobile: str | None,
+    department: str | None,
+    designation: str | None,
+    date_of_joining: str | None,
+    employment_type: str | None,
+) -> dict[str, Any]:
+    if not user_obj.tenant_id or not user_obj.branch_id:
+        raise ValueError("Tenant and Branch are mandatory for internal firm users.")
+
+    existing_linked = db.execute(
+        select(Employee).where(Employee.tenant_id == int(user_obj.tenant_id), Employee.user_id == int(user_obj.id))
+    ).scalar_one_or_none()
+    if existing_linked:
+        return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_linked.id}
+
+    email = (user_obj.email or "").strip().lower()
+    existing_unlinked = None
+    if email:
+        existing_unlinked = db.execute(
+            select(Employee).where(
+                Employee.tenant_id == int(user_obj.tenant_id),
+                Employee.user_id.is_(None),
+                Employee.email == email,
+            )
+        ).scalar_one_or_none()
+
+    if existing_unlinked:
+        existing_unlinked.user_id = user_obj.id
+        existing_unlinked.branch_id = user_obj.branch_id
+        existing_unlinked.full_name = user_obj.full_name or existing_unlinked.full_name
+        existing_unlinked.mobile = (mobile or "").strip() or existing_unlinked.mobile
+        existing_unlinked.department = (department or "").strip() or existing_unlinked.department
+        existing_unlinked.designation = (designation or "").strip() or existing_unlinked.designation
+        existing_unlinked.updated_by_user_id = actor.id
+        existing_unlinked.updated_at_utc = datetime.now(timezone.utc)
+        return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_unlinked.id}
+
+    code = (employee_code or "").strip() or _next_employee_code(db, int(user_obj.tenant_id))
+    duplicate = db.execute(
+        select(Employee).where(Employee.tenant_id == int(user_obj.tenant_id), Employee.employee_code == code)
+    ).scalar_one_or_none()
+    if duplicate:
+        raise ValueError(f"Employee code '{code}' already exists in this firm.")
+
+    emp_type = (employment_type or "full_time").strip() or "full_time"
+    if emp_type not in {"full_time", "part_time", "article_assistant", "intern", "consultant", "contract"}:
+        emp_type = "full_time"
+
+    employee = Employee(
+        tenant_id=int(user_obj.tenant_id),
+        branch_id=int(user_obj.branch_id),
+        user_id=int(user_obj.id),
+        employee_code=code,
+        full_name=(user_obj.full_name or email or code).strip(),
+        email=email or None,
+        mobile=(mobile or "").strip() or None,
+        date_of_joining=_parse_optional_date(date_of_joining),
+        employment_type=emp_type,
+        status="active",
+        is_active=True,
+        department=(department or "").strip() or None,
+        designation=(designation or "").strip() or None,
+        created_by_user_id=actor.id,
+        updated_by_user_id=actor.id,
+    )
+    db.add(employee)
+    db.flush()
+    return {"required": True, "created": True, "linked_existing": False, "employee_id": employee.id}
+
+
+def get_onboarding_role_config(role_key: str) -> dict[str, str] | None:
+    return ONBOARDING_ROLE_CONFIG.get((role_key or "").strip().lower())
+
+
+def build_user_onboarding_payload(db: Session, request, current_user, role_key: str) -> dict[str, Any]:
+    roles = set(get_user_role_names(db, current_user.id))
+    tenant_id = _active_tenant_id(request, current_user, roles)
+    tenant = _tenant(db, tenant_id)
+    config = get_onboarding_role_config(role_key)
+    branches = _branch_rows(db, tenant_id)
+    return {
+        "tenant": tenant,
+        "tenant_id": tenant_id,
+        "role_key": (role_key or "").strip().lower(),
+        "role_config": config,
+        "branches": branches,
+        "employee_code_suggestion": _next_employee_code(db, int(tenant_id)) if tenant_id else "",
+    }
+
+
+def create_firm_internal_user(
+    db: Session,
+    *,
+    request,
+    actor: User,
+    role_key: str,
+    email: str,
+    full_name: str,
+    branch_id: int,
+    employee_code: str | None = None,
+    mobile: str | None = None,
+    department: str | None = None,
+    designation: str | None = None,
+    date_of_joining: str | None = None,
+    employment_type: str | None = None,
+) -> dict[str, Any]:
+    actor_roles = set(get_user_role_names(db, actor.id))
+    if not actor_roles.intersection(FIRM_ADMIN_ROLES):
+        raise PermissionError("Only Firm Admin or System Admin can add firm users.")
+
+    config = get_onboarding_role_config(role_key)
+    if not config:
+        raise ValueError("Invalid onboarding role.")
+
+    tenant_id = _active_tenant_id(request, actor, actor_roles)
+    if not tenant_id:
+        raise ValueError("Active firm context is required before adding users.")
+
+    branch = db.get(Branch, int(branch_id)) if branch_id else None
+    if not branch or int(branch.tenant_id) != int(tenant_id):
+        raise ValueError("Please select a valid branch for the active firm.")
+    if not branch.is_active:
+        raise ValueError("Selected branch is inactive.")
+
+    role = db.execute(
+        select(Role).where(Role.name == config["role_name"], Role.is_active.is_(True))
+    ).scalar_one_or_none()
+    if not role:
+        raise ValueError(f"Role '{config['role_name']}' is not available. Please seed startup roles first.")
+
+    email_clean = (email or "").strip().lower()
+    name_clean = (full_name or "").strip()
+    if not email_clean or "@" not in email_clean:
+        raise ValueError("A valid email is required.")
+    if not name_clean:
+        raise ValueError("Full name is required.")
+    if db.execute(select(User).where(User.email == email_clean)).scalar_one_or_none():
+        raise ValueError("Email already exists.")
+
+    temp_password = hash_password(__import__("secrets").token_urlsafe(18))
+    user = User(
+        email=email_clean,
+        full_name=name_clean,
+        password_hash=temp_password,
+        tenant_id=int(tenant_id),
+        branch_id=int(branch.id),
+        is_active=True,
+        allow_login=True,
+        is_locked=False,
+        deleted_at=None,
+        must_change_password=True,
+        password_changed_at_utc=None,
+        designation=(designation or config.get("default_designation") or "").strip() or None,
+        mobile=(mobile or "").strip() or None,
+    )
+    db.add(user)
+    db.flush()
+    db.add(UserRole(user_id=user.id, role_id=role.id))
+
+    employee_link_result = _ensure_employee_for_onboarded_user(
+        db,
+        actor=actor,
+        user_obj=user,
+        employee_code=employee_code,
+        mobile=mobile,
+        department=(department or config.get("default_department") or ""),
+        designation=(designation or config.get("default_designation") or ""),
+        date_of_joining=date_of_joining,
+        employment_type=(employment_type or config.get("default_employment_type") or "full_time"),
+    )
+
+    try:
+        db.commit()
+    except IntegrityError as exc:
+        db.rollback()
+        raise ValueError("Could not create user due to duplicate or invalid data.") from exc
+
+    db.refresh(user)
+    invite_token = issue_invite_token(db, user)
+    invite_url = _public_invite_url(invite_token)
+    email_status = "not_attempted"
+    email_error = None
+    try:
+        email_log = send_user_invite_email(db, user=user, invite_token=invite_token)
+        db.commit()
+        email_status = getattr(email_log, "status", None) or "attempted"
+        email_error = getattr(email_log, "error_message", None)
+    except Exception as exc:  # invite link fallback remains available
+        db.rollback()
+        email_status = "failed"
+        email_error = str(exc)
+
+    return {
+        "user": user,
+        "role_name": role.name,
+        "role_label": config["label"],
+        "branch": branch,
+        "employee_link_result": employee_link_result,
+        "invite_url": invite_url,
+        "email_status": email_status,
+        "email_error": email_error,
+    }
+
 def build_dashboard_payload(db: Session, request, current_user) -> dict[str, Any]:
     roles = set(get_user_role_names(db, current_user.id))
     tenant_id = _active_tenant_id(request, current_user, roles)
@@ -346,7 +619,7 @@ def _report_cards() -> list[dict[str, str]]:
 def _wizard_cards() -> list[dict[str, str]]:
     return [
         {"title": "Add / Manage Branches", "desc": "Create branch, office timing, login control and local storage path.", "href": "/system-settings/branches"},
-        {"title": "Invite Users", "desc": "Create Firm Admin, Partner, Manager, Staff and role mapping.", "href": "/system-settings/users"},
+        {"title": "Invite Users", "desc": "Create Partner, Manager and Staff directly from Firm Admin.", "href": "/firm-admin/dashboard?tab=users"},
         {"title": "Firm Profile & Branding", "desc": "Display name, logo, colours and contact details.", "href": "/system-settings/branding"},
         {"title": "Select Firm Services", "desc": "Enable services from system catalogue for this firm.", "href": "/services"},
         {"title": "Firm Task Templates", "desc": "Customize task list and document requirements for enabled services.", "href": "/services/templates"},