from __future__ import annotations

import hashlib
import hmac
import os
import re
import secrets
import shutil
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from dataclasses import dataclass
from pathlib import Path
from typing import BinaryIO

from sqlalchemy import Select, func, or_, select
from sqlalchemy.orm import Session, joinedload

from app.modules.clients.models import Client
from app.modules.core.rbac.permission_guard import require_permission
from app.modules.core.tenancy.models import Branch, Tenant
from app.modules.documents.models import (
    BranchStorageNode,
    DocumentAccessLog,
    DocumentDownloadRequest,
    DocumentStorageJob,
    EngagementDocument,
    EngagementDocumentVersion,
    PermanentClientDocument,
    PermanentClientDocumentVersion,
    PermanentDocumentDownloadRequest,
    PermanentDocumentStorageJob,
)
from app.modules.services.models import ClientServiceSubscription, ServiceCatalogue

DOCUMENT_TYPES = [
    "GENERAL",
    "CLIENT_DOCUMENT",
    "WORKING_PAPER",
    "BANK_STATEMENT",
    "TRIAL_BALANCE",
    "LEDGER",
    "GST_RETURN",
    "INCOME_TAX",
    "ROC",
    "AUDIT_REPORT",
    "SIGNED_OUTPUT",
    "ACKNOWLEDGEMENT",
]

UDIN_STATUSES = ["not_required", "pending", "generated", "cancelled"]
FINAL_RELEASE_STATUSES = ["draft", "released"]
UDIN_REQUIRED_DOCUMENT_TYPES = {"AUDIT_REPORT", "SIGNED_OUTPUT", "ACKNOWLEDGEMENT"}

MAX_UPLOAD_BYTES = int(os.getenv("DOCUMENT_MAX_UPLOAD_MB", "50")) * 1024 * 1024

# A server copy is retained only as a temporary transfer buffer. It is deleted
# after the local storage agent confirms the exact expected SHA-256 hash.
# Emergency rollback: DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC=false
DELETE_VPS_AFTER_LOCAL_SYNC = (os.getenv("DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC", "true").strip().lower() not in {"0", "false", "no", "off"})


def _resolve_document_storage_root(env_name: str, default_leaf: str) -> Path:
    """Return a short absolute storage root.

    Windows has a practical path-length limit in many Python/OS operations.
    The project is often run from a very deep development folder, so using a
    relative default like ``documents/engagement_documents`` can exceed that
    limit after adding audit-firm/branch/year/client/engagement folders.

    If an environment variable is provided, it is respected. Otherwise, on
    Windows we use a short drive-root path such as
    ``D:/AuditFirmERPDocuments/engagement_documents``.
    """
    configured = (os.getenv(env_name) or "").strip()
    if configured:
        configured_path = Path(configured).expanduser()
        if configured_path.is_absolute():
            return configured_path
        return (Path.cwd() / configured_path).resolve()

    if os.name == "nt":
        anchor = Path.cwd().anchor or (os.environ.get("SystemDrive", "C:") + "\\")
        return Path(anchor) / "AuditFirmERPDocuments" / default_leaf

    return (Path.cwd() / "documents" / default_leaf).resolve()


DEFAULT_STORAGE_ROOT = _resolve_document_storage_root("DOCUMENT_STORAGE_ROOT", "engagement_documents")
DOWNLOAD_CACHE_ROOT = _resolve_document_storage_root("DOCUMENT_DOWNLOAD_CACHE_ROOT", "download_cache")

_SAFE_CHARS = re.compile(r"[^A-Za-z0-9._ -]+")


@dataclass
class DocumentScope:
    tenant_id: int | None
    branch_id: int | None
    role_names: set[str]
    is_system_admin: bool
    is_firm_admin: bool
    is_partner: bool
    is_branch_manager: bool
    is_staff: bool


def sanitize_segment(value: object, default: str = "NA") -> str:
    text = str(value or default).strip()
    text = _SAFE_CHARS.sub("_", text)
    text = text.strip("._-")
    return text[:80] or default


def _has_perm(db: Session, user, code: str) -> bool:
    try:
        require_permission(db, user, code)
        return True
    except Exception:
        return False


def _now_utc() -> datetime:
    return datetime.now(timezone.utc)


def _clean_udin_number(value: str | None) -> str | None:
    text = (value or "").strip().upper().replace(" ", "")
    return text or None


def _to_decimal(value: str | Decimal | int | float | None) -> Decimal | None:
    if value in (None, ""):
        return None
    try:
        return Decimal(str(value).replace(",", "")).quantize(Decimal("0.01"))
    except (InvalidOperation, ValueError):
        raise ValueError("Invalid UDIN amount.")


def _normalize_udin_required(document_type: str | None, explicit_value: bool | str | None = None) -> bool:
    if isinstance(explicit_value, str):
        lowered = explicit_value.strip().lower()
        if lowered in {"1", "true", "yes", "on", "required"}:
            return True
        if lowered in {"0", "false", "no", "off", "not_required"}:
            return False
    if explicit_value is not None:
        return bool(explicit_value)
    return (document_type or "").strip().upper() in UDIN_REQUIRED_DOCUMENT_TYPES


def user_can_manage_udin(db: Session, user, document: EngagementDocument, scope: DocumentScope) -> bool:
    if not _has_perm(db, user, "udin.manage"):
        return False
    engagement = getattr(document, "engagement", None) or db.get(ClientServiceSubscription, document.engagement_id)
    return bool(engagement and user_can_view_engagement(db, user, engagement, scope))


def user_can_view_udin_register(db: Session, user, scope: DocumentScope) -> bool:
    return _has_perm(db, user, "udin.view")


def build_document_scope(request, db: Session, user) -> DocumentScope:
    from app.modules.core.rbac.deps import get_user_roles

    roles = set(get_user_roles(db, user.id))
    active_tenant = request.session.get("active_tenant_id") or request.session.get("selected_tenant_id") or getattr(user, "tenant_id", None)
    active_branch = request.session.get("active_branch_id")
    if active_branch in (None, "", 0, "0"):
        active_branch = getattr(user, "branch_id", None)
    return DocumentScope(
        tenant_id=int(active_tenant) if active_tenant else None,
        branch_id=int(active_branch) if active_branch else None,
        role_names=roles,
        is_system_admin="System Admin" in roles,
        is_firm_admin="Firm Admin" in roles,
        is_partner="Partner" in roles,
        is_branch_manager="Branch Manager" in roles,
        is_staff="Staff" in roles,
    )


def user_can_view_engagement(db: Session, user, engagement: ClientServiceSubscription, scope: DocumentScope) -> bool:
    if not _has_perm(db, user, "documents.view"):
        return False

    # Professional document contents are not exposed to System Admin merely by SaaS support role.
    if scope.is_system_admin and not (scope.is_firm_admin or scope.is_partner or scope.is_branch_manager or scope.is_staff):
        return False

    if not scope.is_system_admin and scope.tenant_id and engagement.tenant_id != scope.tenant_id:
        return False

    if scope.is_firm_admin:
        return engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id)

    if scope.is_partner:
        client = getattr(engagement, "client", None)
        return (
            engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id)
            and (
                engagement.assigned_partner_user_id == user.id
                or getattr(client, "partner_id", None) == user.id
            )
        )

    if scope.is_branch_manager:
        return (
            engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id)
            and (engagement.branch_id is None or engagement.branch_id == getattr(user, "branch_id", None))
        )

    if scope.is_staff:
        return (
            engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id)
            and (engagement.branch_id is None or engagement.branch_id == getattr(user, "branch_id", None))
            and engagement.assigned_staff_user_id == user.id
        )

    return False


def user_can_upload_to_engagement(db: Session, user, engagement: ClientServiceSubscription, scope: DocumentScope) -> bool:
    return _has_perm(db, user, "documents.upload") and user_can_view_engagement(db, user, engagement, scope)


def user_can_delete_document(db: Session, user, document: EngagementDocument, scope: DocumentScope) -> bool:
    if not _has_perm(db, user, "documents.delete"):
        return False
    if scope.is_firm_admin:
        return document.tenant_id == getattr(user, "tenant_id", document.tenant_id)
    if scope.is_partner:
        engagement = db.get(ClientServiceSubscription, document.engagement_id)
        return bool(engagement and user_can_view_engagement(db, user, engagement, scope))
    return False


def _engagement_query_base() -> Select:
    return (
        select(ClientServiceSubscription)
        .options(
            joinedload(ClientServiceSubscription.client),
            joinedload(ClientServiceSubscription.catalogue),
            joinedload(ClientServiceSubscription.assigned_partner),
            joinedload(ClientServiceSubscription.assigned_staff),
        )
        .order_by(ClientServiceSubscription.financial_year.desc(), ClientServiceSubscription.id.desc())
    )


def list_visible_engagements(db: Session, user, scope: DocumentScope, q: str = "", financial_year: str | None = None, limit: int = 200):
    stmt = _engagement_query_base()
    if scope.is_firm_admin:
        stmt = stmt.where(ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None))
    elif scope.is_partner:
        stmt = stmt.outerjoin(Client, Client.id == ClientServiceSubscription.client_id).where(
            ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None),
            or_(ClientServiceSubscription.assigned_partner_user_id == user.id, Client.partner_id == user.id),
        )
    elif scope.is_branch_manager:
        stmt = stmt.where(
            ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None),
            or_(ClientServiceSubscription.branch_id == getattr(user, "branch_id", None), ClientServiceSubscription.branch_id.is_(None)),
        )
    elif scope.is_staff:
        stmt = stmt.where(
            ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None),
            or_(ClientServiceSubscription.branch_id == getattr(user, "branch_id", None), ClientServiceSubscription.branch_id.is_(None)),
            ClientServiceSubscription.assigned_staff_user_id == user.id,
        )
    else:
        return []

    if financial_year:
        stmt = stmt.where(ClientServiceSubscription.financial_year == financial_year.strip())
    if q:
        pattern = f"%{q.strip()}%"
        stmt = stmt.join(Client, Client.id == ClientServiceSubscription.client_id).join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceSubscription.service_catalogue_id).where(
            or_(Client.client_name.ilike(pattern), Client.client_code.ilike(pattern), ServiceCatalogue.service_name.ilike(pattern), ServiceCatalogue.service_code.ilike(pattern))
        )
    return db.execute(stmt.limit(limit)).unique().scalars().all()


def list_documents_for_engagement(db: Session, engagement_id: int):
    return db.execute(
        select(EngagementDocument)
        .options(joinedload(EngagementDocument.versions), joinedload(EngagementDocument.client), joinedload(EngagementDocument.engagement))
        .where(EngagementDocument.engagement_id == engagement_id, EngagementDocument.is_deleted.is_(False))
        .order_by(EngagementDocument.updated_at_utc.desc())
    ).unique().scalars().all()


def get_document(db: Session, document_id: int) -> EngagementDocument | None:
    return db.execute(
        select(EngagementDocument)
        .options(joinedload(EngagementDocument.versions), joinedload(EngagementDocument.client), joinedload(EngagementDocument.engagement))
        .where(EngagementDocument.id == document_id, EngagementDocument.is_deleted.is_(False))
    ).unique().scalar_one_or_none()


def get_latest_version(document: EngagementDocument) -> EngagementDocumentVersion | None:
    return document.versions[0] if document.versions else None


def get_version(db: Session, version_id: int) -> EngagementDocumentVersion | None:
    return db.execute(
        select(EngagementDocumentVersion)
        .options(joinedload(EngagementDocumentVersion.document).joinedload(EngagementDocument.engagement))
        .where(EngagementDocumentVersion.id == version_id)
    ).unique().scalar_one_or_none()


def client_folder_parts(client: Client | None, client_id: int | None) -> tuple[str, str]:
    """Return A-Z bucket and readable Client Name_Code folder for local storage."""
    client_name = sanitize_segment(getattr(client, "client_name", None) or getattr(client, "name", None) or f"Client {client_id}", "Client")
    client_code = sanitize_segment(getattr(client, "client_code", None) or client_id or "NA", "NA")
    first = (client_name[:1] or "#").upper()
    if not first.isalpha():
        first = "#"
    return first, f"{client_name}_{client_code}"


def _clean_original_name_parts(original_filename: str) -> tuple[str, str]:
    """Return Windows-safe file stem and extension for readable local storage names."""
    original = Path(original_filename or "document.bin").name
    suffix = Path(original).suffix
    stem = original[:-len(suffix)] if suffix else original
    safe_stem = sanitize_segment(stem, "document")
    safe_suffix = re.sub(r"[^A-Za-z0-9.]", "", suffix)[:20]
    return safe_stem, safe_suffix or ".bin"


def build_versioned_filename(original_filename: str, *, document_id: int, version_no: int, engagement_code: str | None = None) -> str:
    """Build readable traceable file names without extra DOC/Versions folders.

    Engagement document: OriginalFile_ENG-1-GST02_DOC000001_v001.xlsx
    Permanent document:  OriginalFile_DOC000001_v001.pdf
    """
    stem, suffix = _clean_original_name_parts(original_filename)
    doc_id = sanitize_segment(f"DOC{document_id:06d}", "DOC")
    version = sanitize_segment(f"v{version_no:03d}", "v001")
    if engagement_code:
        return f"{stem}_{sanitize_segment(engagement_code, 'ENGAGEMENT')}_{doc_id}_{version}{suffix}"
    return f"{stem}_{doc_id}_{version}{suffix}"


def _first_text_value(*values: object) -> str | None:
    for value in values:
        if value is None:
            continue
        text = str(value).strip()
        if text:
            return text
    return None


def infer_service_category(service: ServiceCatalogue | None) -> str:
    """Return a clean service category folder such as GST, Income Tax, ROC, Audit."""
    category_obj = getattr(service, "category", None) if service is not None else None
    category_text = _first_text_value(
        getattr(category_obj, "name", None),
        getattr(category_obj, "category_name", None),
        getattr(service, "category_name", None) if service is not None else None,
        getattr(service, "service_category", None) if service is not None else None,
        getattr(service, "category", None) if service is not None and not hasattr(getattr(service, "category", None), "name") else None,
    )
    if category_text:
        return sanitize_segment(category_text, "Other")

    name_text = (_first_text_value(
        getattr(service, "service_name", None) if service is not None else None,
        getattr(service, "name", None) if service is not None else None,
        getattr(service, "title", None) if service is not None else None,
        getattr(service, "service_code", None) if service is not None else None,
    ) or "").lower()
    if "gst" in name_text or "gstr" in name_text:
        return "GST"
    if "roc" in name_text or "mca" in name_text or "aoc" in name_text or "mgt" in name_text:
        return "ROC"
    if "income" in name_text or "itr" in name_text or "tax audit" in name_text or "3cd" in name_text:
        return "Income_Tax"
    if "tds" in name_text or "tcs" in name_text:
        return "TDS_TCS"
    if "audit" in name_text:
        return "Audit"
    if "pf" in name_text or "esi" in name_text:
        return "PF_ESI"
    if "account" in name_text or "book" in name_text:
        return "Accounts"
    return "Other"


def infer_engagement_period(engagement: ClientServiceSubscription, service: ServiceCatalogue | None) -> str:
    """Return period folder for recurring/annual services, with safe fallback."""
    candidate = _first_text_value(
        getattr(engagement, "period_label", None),
        getattr(engagement, "return_period", None),
        getattr(engagement, "compliance_period", None),
        getattr(engagement, "filing_period", None),
        getattr(engagement, "month_label", None),
        getattr(engagement, "quarter_label", None),
        getattr(engagement, "period", None),
    )
    if candidate:
        return sanitize_segment(candidate, "General")

    recurrence = (_first_text_value(
        getattr(engagement, "recurrence_type", None),
        getattr(service, "recurrence_type", None) if service is not None else None,
        getattr(service, "frequency", None) if service is not None else None,
    ) or "").lower()
    fy = sanitize_segment(f"FY{getattr(engagement, 'financial_year', '') or 'General'}", "FY")
    ay = sanitize_segment(f"AY{getattr(engagement, 'assessment_year', '')}", "AY") if getattr(engagement, "assessment_year", None) else None
    if "month" in recurrence:
        return "Monthly"
    if "quarter" in recurrence:
        return "Quarterly"
    if "annual" in recurrence or "year" in recurrence:
        return ay or fy
    return fy or "General"


def build_year_wise_relative_path(db: Session, engagement: ClientServiceSubscription, document: EngagementDocument, version_no: int, original_filename: str) -> Path:
    client = getattr(engagement, "client", None) or db.get(Client, engagement.client_id)
    service = getattr(engagement, "catalogue", None) or db.get(ServiceCatalogue, engagement.service_catalogue_id)

    fy = sanitize_segment(f"FY{engagement.financial_year}", "FY")
    letter, client_folder = client_folder_parts(client, engagement.client_id)
    service_code = sanitize_segment(getattr(service, "service_code", None) or engagement.service_catalogue_id, "ENG")
    engagement_code = sanitize_segment(f"ENG-{engagement.id}-{service_code}", "ENGAGEMENT")
    service_category = infer_service_category(service)
    period = infer_engagement_period(engagement, service)
    doc_type = sanitize_segment(document.document_type, "GENERAL")
    stored_filename = build_versioned_filename(original_filename, document_id=document.id, version_no=version_no, engagement_code=engagement_code)

    return Path(fy) / "Clients" / letter / client_folder / service_category / period / doc_type / stored_filename


def build_permanent_relative_path(db: Session, client: Client, document: PermanentClientDocument, version_no: int, original_filename: str) -> Path:
    letter, client_folder = client_folder_parts(client, document.client_id)
    category = sanitize_segment(document.category, "Other Permanent Documents")
    stored_filename = build_versioned_filename(original_filename, document_id=document.id, version_no=version_no)
    return Path("Permanent") / "Clients" / letter / client_folder / category / stored_filename


def create_document_code(db: Session, tenant_id: int, engagement_id: int, current_document_id: int | None = None) -> str:
    filters = [
        EngagementDocument.tenant_id == tenant_id,
        EngagementDocument.engagement_id == engagement_id,
    ]
    if current_document_id:
        filters.append(EngagementDocument.id != current_document_id)
    count = db.execute(select(func.count(EngagementDocument.id)).where(*filters)).scalar_one()
    return f"ENG{engagement_id}-DOC{int(count) + 1:04d}"


def log_document_access(db: Session, *, action: str, result: str, user, request=None, document=None, version=None, message: str | None = None):
    doc = document or getattr(version, "document", None)
    db.add(DocumentAccessLog(
        tenant_id=getattr(doc, "tenant_id", None),
        branch_id=getattr(doc, "branch_id", None),
        client_id=getattr(doc, "client_id", None),
        engagement_id=getattr(doc, "engagement_id", None),
        document_id=getattr(doc, "id", None),
        version_id=getattr(version, "id", None),
        action=action,
        result=result,
        user_id=getattr(user, "id", None),
        ip_address=(request.client.host if request and request.client else None),
        user_agent=(request.headers.get("user-agent")[:500] if request else None),
        message=message,
    ))




def update_document_udin(
    db: Session,
    *,
    document: EngagementDocument,
    udin_required: bool | str | None,
    udin_number: str | None,
    udin_date: date | None,
    udin_document_date: date | None,
    udin_document_type: str | None,
    udin_financial_year: str | None,
    udin_amount: str | Decimal | int | float | None,
    udin_notes: str | None,
    user,
) -> EngagementDocument:
    required = _normalize_udin_required(document.document_type, udin_required)
    number = _clean_udin_number(udin_number)

    document.udin_required = required
    document.udin_number = number
    document.udin_date = udin_date
    document.udin_document_date = udin_document_date
    document.udin_document_type = (udin_document_type or "").strip()[:120] or None
    document.udin_financial_year = (udin_financial_year or document.financial_year or "").strip()[:9] or None
    document.udin_amount = _to_decimal(udin_amount)
    document.udin_notes = (udin_notes or "").strip() or None
    document.updated_by_user_id = getattr(user, "id", None)

    if required:
        if number:
            document.udin_status = "generated"
            document.udin_generated_by_user_id = getattr(user, "id", None)
            document.udin_verified_at_utc = _now_utc()
        else:
            document.udin_status = "pending"
            document.udin_generated_by_user_id = None
            document.udin_verified_at_utc = None
    else:
        document.udin_status = "not_required"
        if not number:
            document.udin_generated_by_user_id = None
            document.udin_verified_at_utc = None

    db.flush()
    return document


def release_document_final(
    db: Session,
    *,
    document: EngagementDocument,
    release_notes: str | None,
    user,
) -> EngagementDocument:
    if not get_latest_version(document):
        raise ValueError("Cannot release a document without an uploaded version.")
    if document.udin_required and not _clean_udin_number(document.udin_number):
        raise ValueError("UDIN is required before final release for this document.")
    if document.udin_required and document.udin_status != "generated":
        raise ValueError("UDIN status must be Generated before final release.")

    # Phase 4 AQMM final release gate. Assurance engagements can release final
    # documents only after the AQMM-tagged task evidence/review trail is complete.
    engagement = document.engagement or db.get(ClientServiceSubscription, document.engagement_id)
    if engagement is not None:
        from app.modules.services.client_services import quality_required_for_engagement
        from app.modules.services.execution import assert_aqmm_quality_tasks_complete

        if quality_required_for_engagement(getattr(engagement, "engagement_type", None)):
            assert_aqmm_quality_tasks_complete(db, subscription_id=int(engagement.id))

    document.final_release_status = "released"
    document.status = "final"
    document.released_by_user_id = getattr(user, "id", None)
    document.released_at_utc = _now_utc()
    document.release_notes = (release_notes or "").strip() or None
    document.updated_by_user_id = getattr(user, "id", None)
    db.flush()
    return document


def reopen_final_document(
    db: Session,
    *,
    document: EngagementDocument,
    release_notes: str | None,
    user,
) -> EngagementDocument:
    document.final_release_status = "draft"
    document.status = "active"
    document.release_notes = (release_notes or "").strip() or document.release_notes
    document.updated_by_user_id = getattr(user, "id", None)
    db.flush()
    return document


def list_udin_records(
    db: Session,
    user,
    scope: DocumentScope,
    *,
    financial_year: str | None = None,
    status: str | None = None,
    partner_user_id: int | None = None,
    q: str | None = None,
    limit: int = 500,
) -> list[EngagementDocument]:
    stmt = (
        select(EngagementDocument)
        .options(
            joinedload(EngagementDocument.client),
            joinedload(EngagementDocument.engagement).joinedload(ClientServiceSubscription.assigned_partner),
            joinedload(EngagementDocument.versions),
        )
        .where(EngagementDocument.is_deleted.is_(False))
        .order_by(EngagementDocument.udin_date.desc().nullslast(), EngagementDocument.updated_at_utc.desc())
    )

    if scope.is_firm_admin:
        stmt = stmt.where(EngagementDocument.tenant_id == getattr(user, "tenant_id", None))
    elif scope.is_partner:
        stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).outerjoin(Client, Client.id == EngagementDocument.client_id).where(
            EngagementDocument.tenant_id == getattr(user, "tenant_id", None),
            or_(ClientServiceSubscription.assigned_partner_user_id == user.id, Client.partner_id == user.id, EngagementDocument.udin_generated_by_user_id == user.id),
        )
    elif scope.is_branch_manager:
        stmt = stmt.where(
            EngagementDocument.tenant_id == getattr(user, "tenant_id", None),
            or_(EngagementDocument.branch_id == getattr(user, "branch_id", None), EngagementDocument.branch_id.is_(None)),
        )
    elif scope.is_staff:
        stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).where(
            EngagementDocument.tenant_id == getattr(user, "tenant_id", None),
            ClientServiceSubscription.assigned_staff_user_id == user.id,
        )
    else:
        return []

    if financial_year:
        stmt = stmt.where(EngagementDocument.financial_year == financial_year.strip())
    if status:
        stmt = stmt.where(EngagementDocument.udin_status == status.strip())
    if partner_user_id:
        stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).where(
            or_(ClientServiceSubscription.assigned_partner_user_id == int(partner_user_id), EngagementDocument.udin_generated_by_user_id == int(partner_user_id))
        )
    if q:
        pattern = f"%{q.strip()}%"
        stmt = stmt.where(or_(EngagementDocument.title.ilike(pattern), EngagementDocument.document_code.ilike(pattern), EngagementDocument.udin_number.ilike(pattern)))
    return db.execute(stmt.limit(limit)).unique().scalars().all()


def hash_storage_secret(secret: str) -> str:
    return hashlib.sha256((secret or "").encode("utf-8")).hexdigest()


def generate_storage_secret() -> str:
    return secrets.token_urlsafe(32)


def _same_branch_node_filter(stmt, tenant_id: int, branch_id: int | None):
    stmt = stmt.where(BranchStorageNode.tenant_id == int(tenant_id))
    if branch_id is None:
        stmt = stmt.where(BranchStorageNode.branch_id.is_(None))
    else:
        stmt = stmt.where(BranchStorageNode.branch_id == int(branch_id))
    return stmt


def get_canonical_storage_node(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None:
    """Return the permanent storage node for an audit-firm/branch pair.

    The oldest row is canonical so repeated package generation never changes the
    node code already written in the branch storage identity file.
    """
    stmt = select(BranchStorageNode).where(BranchStorageNode.tenant_id == int(tenant_id))
    if branch_id is None:
        stmt = stmt.where(BranchStorageNode.branch_id.is_(None))
    else:
        stmt = stmt.where(BranchStorageNode.branch_id == int(branch_id))
    return db.execute(stmt.order_by(BranchStorageNode.id.asc()).limit(1)).scalar_one_or_none()


def deactivate_duplicate_storage_nodes(db: Session, *, keep_node: BranchStorageNode) -> int:
    """Disable duplicate nodes for the same audit-firm/branch.

    Business rule: one branch must have only one active Local Storage Agent.
    Callers should normally pass the canonical/oldest node as keep_node.
    """
    stmt = select(BranchStorageNode).where(
        BranchStorageNode.tenant_id == int(keep_node.tenant_id),
        BranchStorageNode.id != int(keep_node.id),
        BranchStorageNode.is_active.is_(True),
    )
    if keep_node.branch_id is None:
        stmt = stmt.where(BranchStorageNode.branch_id.is_(None))
    else:
        stmt = stmt.where(BranchStorageNode.branch_id == int(keep_node.branch_id))
    duplicates = db.execute(stmt).scalars().all()
    for dup in duplicates:
        dup.is_active = False
        dup.status = "disabled_duplicate"
    return len(duplicates)


def create_branch_storage_node(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None,
    node_code: str,
    node_name: str,
    connector_url: str | None,
    storage_root_path: str | None,
    quota_limit_gb: int | None,
    user,
) -> tuple[BranchStorageNode, str]:
    raw_secret = generate_storage_secret()

    existing = get_canonical_storage_node(db, int(tenant_id), branch_id)
    if existing:
        # Reuse the original node code forever. This prevents new -002/-003
        # packages for the same branch and keeps the local storage identity
        # file valid. Package regeneration rotates only the secret.
        existing.node_name = existing.node_name or (node_name or existing.node_code or "Branch Storage Node")[:200]
        existing.connector_url = (connector_url or "").strip() or existing.connector_url
        existing.storage_root_path = existing.storage_root_path or ((storage_root_path or "").strip() or None)
        existing.quota_limit_bytes = (int(quota_limit_gb) * 1024 * 1024 * 1024 if quota_limit_gb else existing.quota_limit_bytes)
        existing.secret_key_hash = hash_storage_secret(raw_secret)
        existing.is_active = True
        existing.status = "active"
        deactivate_duplicate_storage_nodes(db, keep_node=existing)
        db.flush()
        return existing, raw_secret

    node = BranchStorageNode(
        tenant_id=tenant_id,
        branch_id=branch_id,
        node_code=sanitize_segment(node_code or node_name, "NODE"),
        node_name=(node_name or node_code or "Branch Storage Node")[:200],
        connector_url=(connector_url or "").strip() or None,
        storage_root_path=(storage_root_path or "").strip() or None,
        quota_limit_bytes=(int(quota_limit_gb) * 1024 * 1024 * 1024 if quota_limit_gb else None),
        secret_key_hash=hash_storage_secret(raw_secret),
        created_by_user_id=getattr(user, "id", None),
        is_active=True,
        status="active",
    )
    db.add(node)
    db.flush()
    deactivate_duplicate_storage_nodes(db, keep_node=node)
    return node, raw_secret


def list_storage_nodes(db: Session, tenant_id: int | None = None, branch_id: int | None = None):
    stmt = select(BranchStorageNode).order_by(BranchStorageNode.tenant_id, BranchStorageNode.branch_id, BranchStorageNode.node_name)
    if tenant_id:
        stmt = stmt.where(BranchStorageNode.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(BranchStorageNode.branch_id == branch_id)
    return db.execute(stmt).scalars().all()


def get_active_storage_node_for_branch(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None:
    stmt = (
        select(BranchStorageNode)
        .where(
            BranchStorageNode.tenant_id == tenant_id,
            BranchStorageNode.is_active.is_(True),
            BranchStorageNode.status == "active",
        )
        .order_by(BranchStorageNode.branch_id.desc(), BranchStorageNode.id.asc())
    )
    if branch_id:
        stmt = stmt.where(or_(BranchStorageNode.branch_id == branch_id, BranchStorageNode.branch_id.is_(None)))
    else:
        stmt = stmt.where(BranchStorageNode.branch_id.is_(None))
    return db.execute(stmt.limit(1)).scalar_one_or_none()


def create_storage_job_for_version(db: Session, *, version: EngagementDocumentVersion, user) -> DocumentStorageJob | None:
    node = get_active_storage_node_for_branch(db, version.tenant_id, version.branch_id)
    if not node:
        return None
    job = DocumentStorageJob(
        tenant_id=version.tenant_id,
        branch_id=version.branch_id,
        storage_node_id=node.id,
        document_id=version.document_id,
        version_id=version.id,
        staging_relative_path=version.local_relative_path,
        target_relative_path=version.local_relative_path,
        file_size_bytes=version.file_size_bytes,
        expected_hash_sha256=version.file_hash_sha256,
        created_by_user_id=getattr(user, "id", None),
    )
    version.storage_backend = "BRANCH_STORAGE_NODE"
    version.storage_status = "pending_local_sync"
    db.add(job)
    return job


def authenticate_storage_node(db: Session, node_code: str | None, secret: str | None, request=None) -> BranchStorageNode | None:
    if not node_code or not secret:
        return None
    node = db.execute(
        select(BranchStorageNode).where(BranchStorageNode.node_code == node_code, BranchStorageNode.is_active.is_(True))
    ).scalar_one_or_none()
    if not node or not hmac.compare_digest(node.secret_key_hash, hash_storage_secret(secret)):
        return None
    node.last_seen_at_utc = __import__("datetime").datetime.now(__import__("datetime").timezone.utc)
    node.last_seen_ip = request.client.host if request and request.client else None
    return node


def list_pending_storage_jobs(db: Session, node: BranchStorageNode, limit: int = 20):
    return db.execute(
        select(DocumentStorageJob)
        .where(DocumentStorageJob.storage_node_id == node.id, DocumentStorageJob.status.in_(["pending", "retry"]))
        .order_by(DocumentStorageJob.priority.asc(), DocumentStorageJob.created_at_utc.asc())
        .limit(limit)
    ).scalars().all()


def get_storage_job_for_node(db: Session, node: BranchStorageNode, job_id: int) -> DocumentStorageJob | None:
    return db.execute(
        select(DocumentStorageJob).where(DocumentStorageJob.id == job_id, DocumentStorageJob.storage_node_id == node.id)
    ).scalar_one_or_none()


def acknowledge_storage_job(db: Session, *, node: BranchStorageNode, job: DocumentStorageJob, acknowledged_hash: str, local_final_path: str | None, success: bool, error: str | None = None) -> bool:
    from datetime import datetime, timezone
    job.attempts = int(job.attempts or 0) + 1
    if success and acknowledged_hash and acknowledged_hash.lower() == job.expected_hash_sha256.lower():
        job.status = "completed"
        job.completed_at_utc = datetime.now(timezone.utc)
        job.acknowledged_hash_sha256 = acknowledged_hash.lower()
        job.local_final_path = local_final_path
        version = db.get(EngagementDocumentVersion, job.version_id)
        if version:
            version.storage_status = "stored_on_branch_node"
        node.used_storage_bytes = int(node.used_storage_bytes or 0) + int(job.file_size_bytes or 0)
        return True
    job.status = "failed" if job.attempts >= 3 else "retry"
    job.last_error = error or "Hash mismatch or local storage acknowledgement failed."
    return False


def _prune_empty_stage_directories(file_path: Path, root: Path) -> None:
    """Remove empty parent folders without crossing the configured root."""
    root = root.resolve()
    parent = file_path.parent
    while parent != root:
        try:
            parent.rmdir()
        except OSError:
            break
        parent = parent.parent


def _verified_stage_path(version) -> Path | None:
    """Resolve the staged path and reject traversal outside storage root."""
    if not version or not getattr(version, "local_relative_path", None):
        return None
    root = DEFAULT_STORAGE_ROOT.resolve()
    candidate = (DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)).resolve()
    try:
        candidate.relative_to(root)
    except ValueError:
        return None
    return candidate


def _cleanup_verified_vps_stage(*, job, version) -> dict[str, object]:
    """Delete one VPS stage only after verified branch-local storage."""
    result: dict[str, object] = {"enabled": DELETE_VPS_AFTER_LOCAL_SYNC, "deleted": False}
    if not DELETE_VPS_AFTER_LOCAL_SYNC:
        result["reason"] = "cleanup_disabled"
        return result
    if not job or job.status != "completed":
        result["reason"] = "job_not_completed"
        return result
    if getattr(job, "vps_cleanup_status", "pending") == "deleted":
        result.update({"deleted": True, "already_completed": True, "reason": "already_deleted"})
        return result

    expected = (job.expected_hash_sha256 or "").strip().lower()
    acknowledged = (job.acknowledged_hash_sha256 or "").strip().lower()
    if not expected or acknowledged != expected:
        result["reason"] = "local_hash_not_verified"
        return result
    if not (job.local_final_path or "").strip():
        result["reason"] = "local_final_path_missing"
        return result

    job.vps_cleanup_attempts = int(job.vps_cleanup_attempts or 0) + 1
    job.vps_cleanup_status = "deleting"
    job.vps_cleanup_error = None
    stage_path = _verified_stage_path(version)
    if stage_path is None:
        job.vps_cleanup_status = "failed"
        job.vps_cleanup_error = "Unsafe or missing staged path."
        result["reason"] = "unsafe_stage_path"
        return result

    try:
        deleted_bytes = int(stage_path.stat().st_size) if stage_path.exists() and stage_path.is_file() else 0
        if stage_path.exists():
            if not stage_path.is_file():
                raise OSError("Staged path is not a regular file.")
            stage_path.unlink()
            _prune_empty_stage_directories(stage_path, DEFAULT_STORAGE_ROOT)
        job.vps_cleanup_status = "deleted"
        job.vps_cleanup_error = None
        job.vps_deleted_at_utc = _now_utc()
        job.vps_deleted_bytes = deleted_bytes
        result.update({
            "deleted": True,
            "already_missing": deleted_bytes == 0,
            "deleted_bytes": deleted_bytes,
            "reason": "deleted_after_verified_local_sync",
        })
        return result
    except OSError as exc:
        job.vps_cleanup_status = "retry" if int(job.vps_cleanup_attempts or 0) < 20 else "failed"
        job.vps_cleanup_error = f"{type(exc).__name__}: {str(exc)[:900]}"
        result.update({"reason": "delete_failed", "error": job.vps_cleanup_error})
        return result


def cleanup_completed_storage_job_vps_stage(db: Session, job: DocumentStorageJob) -> dict[str, object]:
    version = db.get(EngagementDocumentVersion, job.version_id) if job else None
    return _cleanup_verified_vps_stage(job=job, version=version)


def cleanup_completed_permanent_storage_job_vps_stage(db: Session, job: PermanentDocumentStorageJob) -> dict[str, object]:
    version = db.get(PermanentClientDocumentVersion, job.version_id) if job else None
    return _cleanup_verified_vps_stage(job=job, version=version)


def retry_verified_vps_cleanup_for_node(db: Session, node: BranchStorageNode, limit: int = 50) -> dict[str, int]:
    """Retry interrupted/failed deletion during heartbeat or tunnel sync."""
    summary = {"checked": 0, "deleted": 0, "failed": 0}
    if not DELETE_VPS_AFTER_LOCAL_SYNC or not node:
        return summary
    retry_states = ["pending", "retry", "failed", "deleting"]
    engagement_jobs = db.execute(
        select(DocumentStorageJob)
        .where(
            DocumentStorageJob.storage_node_id == node.id,
            DocumentStorageJob.status == "completed",
            DocumentStorageJob.vps_cleanup_status.in_(retry_states),
        )
        .order_by(DocumentStorageJob.completed_at_utc.asc(), DocumentStorageJob.id.asc())
        .limit(limit)
    ).scalars().all()
    remaining = max(0, limit - len(engagement_jobs))
    permanent_jobs = []
    if remaining:
        permanent_jobs = db.execute(
            select(PermanentDocumentStorageJob)
            .where(
                PermanentDocumentStorageJob.storage_node_id == node.id,
                PermanentDocumentStorageJob.status == "completed",
                PermanentDocumentStorageJob.vps_cleanup_status.in_(retry_states),
            )
            .order_by(PermanentDocumentStorageJob.completed_at_utc.asc(), PermanentDocumentStorageJob.id.asc())
            .limit(remaining)
        ).scalars().all()

    for job in engagement_jobs:
        summary["checked"] += 1
        result = cleanup_completed_storage_job_vps_stage(db, job)
        summary["deleted" if result.get("deleted") else "failed"] += 1
    for job in permanent_jobs:
        summary["checked"] += 1
        result = cleanup_completed_permanent_storage_job_vps_stage(db, job)
        summary["deleted" if result.get("deleted") else "failed"] += 1
    return summary


def list_storage_jobs(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200):
    stmt = select(DocumentStorageJob).order_by(DocumentStorageJob.created_at_utc.desc())
    if tenant_id:
        stmt = stmt.where(DocumentStorageJob.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(DocumentStorageJob.branch_id == branch_id)
    if status:
        stmt = stmt.where(DocumentStorageJob.status == status)
    return db.execute(stmt.limit(limit)).scalars().all()



def _now_utc():
    from datetime import datetime, timezone
    return datetime.now(timezone.utc)


def get_completed_storage_job_for_version(db: Session, version_id: int) -> DocumentStorageJob | None:
    """Return the latest completed local-storage job for a document version."""
    return db.execute(
        select(DocumentStorageJob)
        .where(DocumentStorageJob.version_id == version_id, DocumentStorageJob.status == "completed")
        .order_by(DocumentStorageJob.completed_at_utc.desc(), DocumentStorageJob.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def create_download_request_for_version(db: Session, *, version: EngagementDocumentVersion, user, request=None) -> DocumentDownloadRequest | None:
    """Queue a secure local-to-cloud download request for the branch storage agent.

    If the staged ERP file is already available, callers should stream that file
    directly and avoid creating a request. This request is used when the version
    is stored on the branch node and the ERP copy is unavailable.
    """
    completed_job = get_completed_storage_job_for_version(db, version.id)
    if not completed_job or not completed_job.storage_node_id:
        return None

    # Reuse a recent pending/ready request for the same user/version to avoid duplicate queues.
    existing = db.execute(
        select(DocumentDownloadRequest)
        .where(
            DocumentDownloadRequest.version_id == version.id,
            DocumentDownloadRequest.requested_by_user_id == getattr(user, "id", None),
            DocumentDownloadRequest.request_status.in_(["pending", "picked", "ready"]),
        )
        .order_by(DocumentDownloadRequest.created_at_utc.desc())
        .limit(1)
    ).scalar_one_or_none()
    if existing:
        return existing

    download_request = DocumentDownloadRequest(
        tenant_id=version.tenant_id,
        branch_id=version.branch_id,
        storage_node_id=completed_job.storage_node_id,
        document_id=version.document_id,
        version_id=version.id,
        local_relative_path=completed_job.local_final_path or completed_job.target_relative_path,
        expected_hash_sha256=version.file_hash_sha256,
        file_size_bytes=version.file_size_bytes,
        requested_by_user_id=getattr(user, "id", None),
        requested_ip=(request.client.host if request and request.client else None),
        requested_user_agent=(request.headers.get("user-agent")[:500] if request else None),
    )
    db.add(download_request)
    return download_request


def list_pending_download_requests(db: Session, node: BranchStorageNode, limit: int = 20):
    return db.execute(
        select(DocumentDownloadRequest)
        .where(
            DocumentDownloadRequest.storage_node_id == node.id,
            DocumentDownloadRequest.request_status.in_(["pending", "retry"]),
        )
        .order_by(DocumentDownloadRequest.created_at_utc.asc())
        .limit(limit)
    ).scalars().all()


def get_download_request_for_node(db: Session, node: BranchStorageNode, request_id: int) -> DocumentDownloadRequest | None:
    return db.execute(
        select(DocumentDownloadRequest).where(
            DocumentDownloadRequest.id == request_id,
            DocumentDownloadRequest.storage_node_id == node.id,
        )
    ).scalar_one_or_none()


def get_download_request(db: Session, request_id: int) -> DocumentDownloadRequest | None:
    return db.execute(
        select(DocumentDownloadRequest)
        .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version))
        .where(DocumentDownloadRequest.id == request_id)
    ).unique().scalar_one_or_none()


def list_download_requests(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200):
    stmt = (
        select(DocumentDownloadRequest)
        .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version))
        .order_by(DocumentDownloadRequest.created_at_utc.desc())
    )
    if tenant_id:
        stmt = stmt.where(DocumentDownloadRequest.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(DocumentDownloadRequest.branch_id == branch_id)
    if status:
        stmt = stmt.where(DocumentDownloadRequest.request_status == status)
    return db.execute(stmt.limit(limit)).scalars().all()


def list_recent_download_requests_for_engagement(db: Session, engagement_id: int, user_id: int | None = None, limit: int = 10):
    stmt = (
        select(DocumentDownloadRequest)
        .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version))
        .join(EngagementDocumentVersion, EngagementDocumentVersion.id == DocumentDownloadRequest.version_id)
        .where(EngagementDocumentVersion.engagement_id == engagement_id)
        .order_by(DocumentDownloadRequest.created_at_utc.desc())
        .limit(limit)
    )
    if user_id:
        stmt = stmt.where(DocumentDownloadRequest.requested_by_user_id == user_id)
    return db.execute(stmt).scalars().all()


def download_request_cache_path(download_request: DocumentDownloadRequest) -> Path | None:
    if not download_request.cached_relative_path:
        return None
    return DOWNLOAD_CACHE_ROOT / Path(download_request.cached_relative_path)


def fulfill_download_request_from_upload(db: Session, *, node: BranchStorageNode, download_request: DocumentDownloadRequest, upload_file) -> bool:
    """Store the file uploaded by local agent into ERP download cache and verify hash."""
    if download_request.storage_node_id != node.id:
        return False
    version = db.get(EngagementDocumentVersion, download_request.version_id)
    if not version:
        download_request.request_status = "failed"
        download_request.failed_at_utc = _now_utc()
        download_request.last_error = "Document version not found."
        return False

    safe_name = sanitize_segment(version.original_filename or f"version_{version.id}.bin", "document.bin")
    cache_rel = Path(f"request_{download_request.id}") / f"v{version.version_no:03d}_{safe_name}"
    cache_abs = DOWNLOAD_CACHE_ROOT / cache_rel
    cache_abs.parent.mkdir(parents=True, exist_ok=True)

    hasher = hashlib.sha256()
    total = 0
    with cache_abs.open("wb") as out:
        while True:
            chunk = upload_file.file.read(1024 * 1024)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_UPLOAD_BYTES:
                out.close()
                cache_abs.unlink(missing_ok=True)
                download_request.request_status = "failed"
                download_request.failed_at_utc = _now_utc()
                download_request.last_error = f"Uploaded stream exceeds maximum limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB."
                return False
            hasher.update(chunk)
            out.write(chunk)

    actual_hash = hasher.hexdigest()
    download_request.attempts = int(download_request.attempts or 0) + 1
    if actual_hash.lower() != (download_request.expected_hash_sha256 or "").lower():
        cache_abs.unlink(missing_ok=True)
        download_request.request_status = "failed" if download_request.attempts >= 3 else "retry"
        download_request.last_error = "Uploaded file hash does not match the original document version."
        if download_request.request_status == "failed":
            download_request.failed_at_utc = _now_utc()
        return False

    download_request.request_status = "ready"
    download_request.cached_relative_path = str(cache_rel).replace("\\", "/")
    download_request.cached_hash_sha256 = actual_hash.lower()
    download_request.file_size_bytes = total
    download_request.fulfilled_at_utc = _now_utc()
    download_request.last_error = None
    return True

def save_uploaded_revision(
    db: Session,
    *,
    engagement: ClientServiceSubscription,
    upload_file,
    title: str,
    document_type: str,
    description: str | None,
    remarks: str | None,
    user,
    existing_document_id: int | None = None,
    udin_required: bool | str | None = None,
) -> EngagementDocument:
    original_filename = Path(upload_file.filename or "document.bin").name
    document_type = document_type if document_type in DOCUMENT_TYPES else "GENERAL"
    resolved_udin_required = _normalize_udin_required(document_type, udin_required)

    if existing_document_id:
        document = db.get(EngagementDocument, existing_document_id)
        if not document or document.is_deleted or document.engagement_id != engagement.id:
            raise ValueError("Invalid document selected for new revision.")
        if title:
            document.title = title.strip()[:255]
        if description is not None:
            document.description = description.strip() or None
        document.document_type = document_type
        document.udin_required = resolved_udin_required
        if resolved_udin_required and (document.udin_status or "not_required") == "not_required":
            document.udin_status = "pending"
        if not resolved_udin_required and not document.udin_number:
            document.udin_status = "not_required"
    else:
        document = EngagementDocument(
            tenant_id=engagement.tenant_id,
            branch_id=engagement.branch_id,
            client_id=engagement.client_id,
            engagement_id=engagement.id,
            financial_year=engagement.financial_year,
            assessment_year=engagement.assessment_year,
            document_code="PENDING",
            document_type=document_type,
            udin_required=resolved_udin_required,
            udin_status="pending" if resolved_udin_required else "not_required",
            final_release_status="draft",
            title=(title.strip()[:255] if title else original_filename[:255]),
            description=description.strip() if description else None,
            created_by_user_id=user.id,
            updated_by_user_id=user.id,
        )
        db.add(document)
        db.flush()
        document.document_code = create_document_code(db, engagement.tenant_id, engagement.id, document.id)

    next_version_no = int(document.current_version_no or 0) + 1
    rel_path = build_year_wise_relative_path(db, engagement, document, next_version_no, original_filename)
    abs_path = DEFAULT_STORAGE_ROOT / rel_path
    try:
        abs_path.parent.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise RuntimeError(
            "Unable to create ERP document staging folder. "
            f"storage_root='{DEFAULT_STORAGE_ROOT}', relative_path='{rel_path}'. "
            "On Windows, set DOCUMENT_STORAGE_ROOT to a short absolute path like "
            "D:\\AuditFirmERPDocuments\\engagement_documents."
        ) from exc

    hasher = hashlib.sha256()
    total = 0
    with abs_path.open("wb") as out:
        while True:
            chunk = upload_file.file.read(1024 * 1024)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_UPLOAD_BYTES:
                out.close()
                abs_path.unlink(missing_ok=True)
                raise ValueError(f"File exceeds maximum upload limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB.")
            hasher.update(chunk)
            out.write(chunk)

    version = EngagementDocumentVersion(
        document_id=document.id,
        tenant_id=document.tenant_id,
        branch_id=document.branch_id,
        client_id=document.client_id,
        engagement_id=document.engagement_id,
        version_no=next_version_no,
        original_filename=original_filename,
        stored_filename=abs_path.name,
        content_type=getattr(upload_file, "content_type", None),
        file_size_bytes=total,
        file_hash_sha256=hasher.hexdigest(),
        local_relative_path=str(rel_path).replace("\\", "/"),
        remarks=remarks.strip() if remarks else None,
        uploaded_by_user_id=user.id,
    )
    document.current_version_no = next_version_no
    document.updated_by_user_id = user.id
    db.add(version)
    db.flush()
    create_storage_job_for_version(db, version=version, user=user)
    return document



PERMANENT_DOCUMENT_CATEGORIES = [
    "Company Registration",
    "GST Registration",
    "Income Tax",
    "Bank",
    "KYC",
    "Agreements",
    "Licenses",
    "DSC",
    "ROC Master Data",
    "Other Permanent Documents",
]


def user_can_view_client_documents(db: Session, user, client: Client, scope: DocumentScope) -> bool:
    if not _has_perm(db, user, "documents.view"):
        return False
    if scope.is_system_admin and not (scope.is_firm_admin or scope.is_partner or scope.is_branch_manager or scope.is_staff):
        return False
    if scope.is_firm_admin:
        return client.tenant_id == getattr(user, "tenant_id", client.tenant_id)
    if scope.is_partner:
        return client.tenant_id == getattr(user, "tenant_id", client.tenant_id) and getattr(client, "partner_id", None) == user.id
    if scope.is_branch_manager or scope.is_staff:
        return client.tenant_id == getattr(user, "tenant_id", client.tenant_id) and (getattr(client, "branch_id", None) in (None, getattr(user, "branch_id", None)))
    return False


def user_can_upload_client_documents(db: Session, user, client: Client, scope: DocumentScope) -> bool:
    return _has_perm(db, user, "documents.upload") and user_can_view_client_documents(db, user, client, scope)


def list_visible_clients_for_permanent_documents(db: Session, user, scope: DocumentScope, q: str = "", limit: int = 200):
    stmt = select(Client).order_by(Client.client_name.asc()).limit(limit)
    if scope.is_firm_admin:
        stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None))
    elif scope.is_partner:
        stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None), Client.partner_id == user.id)
    elif scope.is_branch_manager or scope.is_staff:
        stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None), or_(Client.branch_id == getattr(user, "branch_id", None), Client.branch_id.is_(None)))
    else:
        return []
    if q:
        pattern = f"%{q.strip()}%"
        stmt = stmt.where(or_(Client.client_name.ilike(pattern), Client.client_code.ilike(pattern)))
    return db.execute(stmt).scalars().all()


def list_permanent_documents_for_client(db: Session, client_id: int):
    return db.execute(
        select(PermanentClientDocument)
        .options(joinedload(PermanentClientDocument.versions), joinedload(PermanentClientDocument.client))
        .where(PermanentClientDocument.client_id == client_id, PermanentClientDocument.is_deleted.is_(False))
        .order_by(PermanentClientDocument.category.asc(), PermanentClientDocument.updated_at_utc.desc())
    ).unique().scalars().all()


def get_permanent_document(db: Session, document_id: int) -> PermanentClientDocument | None:
    return db.execute(
        select(PermanentClientDocument)
        .options(joinedload(PermanentClientDocument.versions), joinedload(PermanentClientDocument.client))
        .where(PermanentClientDocument.id == document_id, PermanentClientDocument.is_deleted.is_(False))
    ).unique().scalar_one_or_none()


def get_permanent_version(db: Session, version_id: int) -> PermanentClientDocumentVersion | None:
    return db.execute(
        select(PermanentClientDocumentVersion)
        .options(joinedload(PermanentClientDocumentVersion.document).joinedload(PermanentClientDocument.client))
        .where(PermanentClientDocumentVersion.id == version_id)
    ).unique().scalar_one_or_none()


def get_latest_permanent_version(document: PermanentClientDocument) -> PermanentClientDocumentVersion | None:
    return document.versions[0] if document.versions else None


def create_permanent_document_code(db: Session, tenant_id: int, client_id: int, current_document_id: int | None = None) -> str:
    filters = [PermanentClientDocument.tenant_id == tenant_id, PermanentClientDocument.client_id == client_id]
    if current_document_id:
        filters.append(PermanentClientDocument.id != current_document_id)
    count = db.execute(select(func.count(PermanentClientDocument.id)).where(*filters)).scalar_one()
    return f"PERM{client_id}-DOC{int(count) + 1:04d}"


def get_active_storage_node_for_permanent_client(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None:
    stmt = select(BranchStorageNode).where(
        BranchStorageNode.tenant_id == tenant_id,
        BranchStorageNode.is_active.is_(True),
        BranchStorageNode.status == "active",
    )
    if branch_id is None:
        stmt = stmt.where(BranchStorageNode.branch_id.is_(None))
    else:
        stmt = stmt.where(or_(BranchStorageNode.branch_id == branch_id, BranchStorageNode.branch_id.is_(None)))
    return db.execute(stmt.order_by(BranchStorageNode.branch_id.desc(), BranchStorageNode.id.asc()).limit(1)).scalar_one_or_none()


def create_permanent_storage_job_for_version(db: Session, *, version: PermanentClientDocumentVersion, user) -> PermanentDocumentStorageJob | None:
    node = get_active_storage_node_for_permanent_client(db, version.tenant_id, version.branch_id)
    if not node:
        return None
    job = PermanentDocumentStorageJob(
        tenant_id=version.tenant_id,
        branch_id=version.branch_id,
        storage_node_id=node.id,
        document_id=version.document_id,
        version_id=version.id,
        staging_relative_path=version.local_relative_path,
        target_relative_path=version.local_relative_path,
        file_size_bytes=version.file_size_bytes,
        expected_hash_sha256=version.file_hash_sha256,
        created_by_user_id=getattr(user, "id", None),
    )
    db.add(job)
    return job


def list_pending_permanent_storage_jobs(db: Session, node: BranchStorageNode, limit: int = 20):
    return db.execute(
        select(PermanentDocumentStorageJob)
        .where(PermanentDocumentStorageJob.storage_node_id == node.id, PermanentDocumentStorageJob.status.in_(["pending", "retry"]))
        .order_by(PermanentDocumentStorageJob.priority.asc(), PermanentDocumentStorageJob.created_at_utc.asc())
        .limit(limit)
    ).scalars().all()


def get_permanent_storage_job_for_node(db: Session, node: BranchStorageNode, job_id: int) -> PermanentDocumentStorageJob | None:
    return db.execute(select(PermanentDocumentStorageJob).where(PermanentDocumentStorageJob.id == job_id, PermanentDocumentStorageJob.storage_node_id == node.id)).scalar_one_or_none()


def acknowledge_permanent_storage_job(db: Session, *, node: BranchStorageNode, job: PermanentDocumentStorageJob, acknowledged_hash: str, local_final_path: str | None, success: bool, error: str | None = None) -> bool:
    if job.storage_node_id != node.id:
        return False
    job.attempts = int(job.attempts or 0) + 1
    if not success:
        job.status = "failed" if job.attempts >= 3 else "retry"
        job.last_error = error or "Local storage agent reported failure."
        return False
    if acknowledged_hash.lower() != (job.expected_hash_sha256 or "").lower():
        job.status = "failed" if job.attempts >= 3 else "retry"
        job.last_error = "Hash mismatch after local storage write."
        return False
    job.status = "completed"
    job.completed_at_utc = _now_utc()
    job.acknowledged_hash_sha256 = acknowledged_hash.lower()
    job.local_final_path = local_final_path or job.target_relative_path
    job.last_error = None
    version = db.get(PermanentClientDocumentVersion, job.version_id)
    if version:
        version.storage_status = "synced_local"
    return True


def get_completed_permanent_storage_job_for_version(db: Session, version_id: int) -> PermanentDocumentStorageJob | None:
    return db.execute(
        select(PermanentDocumentStorageJob)
        .where(PermanentDocumentStorageJob.version_id == version_id, PermanentDocumentStorageJob.status == "completed")
        .order_by(PermanentDocumentStorageJob.completed_at_utc.desc(), PermanentDocumentStorageJob.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def create_permanent_download_request_for_version(db: Session, *, version: PermanentClientDocumentVersion, user, request=None) -> PermanentDocumentDownloadRequest | None:
    completed_job = get_completed_permanent_storage_job_for_version(db, version.id)
    if not completed_job:
        return None
    existing = db.execute(
        select(PermanentDocumentDownloadRequest)
        .where(
            PermanentDocumentDownloadRequest.version_id == version.id,
            PermanentDocumentDownloadRequest.requested_by_user_id == getattr(user, "id", None),
            PermanentDocumentDownloadRequest.request_status.in_(["pending", "picked", "ready"]),
        )
        .order_by(PermanentDocumentDownloadRequest.created_at_utc.desc())
        .limit(1)
    ).scalar_one_or_none()
    if existing:
        return existing
    item = PermanentDocumentDownloadRequest(
        tenant_id=version.tenant_id,
        branch_id=version.branch_id,
        storage_node_id=completed_job.storage_node_id,
        document_id=version.document_id,
        version_id=version.id,
        local_relative_path=completed_job.local_final_path or completed_job.target_relative_path,
        expected_hash_sha256=version.file_hash_sha256,
        file_size_bytes=version.file_size_bytes,
        requested_by_user_id=getattr(user, "id", None),
        requested_ip=(request.client.host if request and request.client else None),
        requested_user_agent=(request.headers.get("user-agent")[:500] if request else None),
    )
    db.add(item)
    return item


def list_pending_permanent_download_requests(db: Session, node: BranchStorageNode, limit: int = 20):
    return db.execute(
        select(PermanentDocumentDownloadRequest)
        .where(PermanentDocumentDownloadRequest.storage_node_id == node.id, PermanentDocumentDownloadRequest.request_status.in_(["pending", "retry"]))
        .order_by(PermanentDocumentDownloadRequest.created_at_utc.asc())
        .limit(limit)
    ).scalars().all()


def get_permanent_download_request_for_node(db: Session, node: BranchStorageNode, request_id: int) -> PermanentDocumentDownloadRequest | None:
    return db.execute(select(PermanentDocumentDownloadRequest).where(PermanentDocumentDownloadRequest.id == request_id, PermanentDocumentDownloadRequest.storage_node_id == node.id)).scalar_one_or_none()


def permanent_download_request_cache_path(download_request: PermanentDocumentDownloadRequest) -> Path | None:
    if not download_request.cached_relative_path:
        return None
    return DOWNLOAD_CACHE_ROOT / Path(download_request.cached_relative_path)


def fulfill_permanent_download_request_from_upload(db: Session, *, node: BranchStorageNode, download_request: PermanentDocumentDownloadRequest, upload_file) -> bool:
    if download_request.storage_node_id != node.id:
        return False
    version = db.get(PermanentClientDocumentVersion, download_request.version_id)
    if not version:
        download_request.request_status = "failed"
        download_request.failed_at_utc = _now_utc()
        download_request.last_error = "Permanent document version not found."
        return False
    safe_name = sanitize_segment(version.original_filename or f"version_{version.id}.bin", "document.bin")
    cache_rel = Path(f"permanent_request_{download_request.id}") / f"v{version.version_no:03d}_{safe_name}"
    cache_abs = DOWNLOAD_CACHE_ROOT / cache_rel
    cache_abs.parent.mkdir(parents=True, exist_ok=True)
    hasher = hashlib.sha256()
    total = 0
    with cache_abs.open("wb") as out:
        while True:
            chunk = upload_file.file.read(1024 * 1024)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_UPLOAD_BYTES:
                out.close(); cache_abs.unlink(missing_ok=True)
                download_request.request_status = "failed"; download_request.failed_at_utc = _now_utc(); download_request.last_error = "Uploaded stream exceeds maximum limit."
                return False
            hasher.update(chunk); out.write(chunk)
    actual_hash = hasher.hexdigest()
    download_request.attempts = int(download_request.attempts or 0) + 1
    if actual_hash.lower() != (download_request.expected_hash_sha256 or "").lower():
        cache_abs.unlink(missing_ok=True)
        download_request.request_status = "failed" if download_request.attempts >= 3 else "retry"
        download_request.last_error = "Uploaded file hash does not match the permanent document version."
        if download_request.request_status == "failed":
            download_request.failed_at_utc = _now_utc()
        return False
    download_request.request_status = "ready"
    download_request.cached_relative_path = str(cache_rel).replace("\\", "/")
    download_request.cached_hash_sha256 = actual_hash.lower()
    download_request.file_size_bytes = total
    download_request.fulfilled_at_utc = _now_utc()
    download_request.last_error = None
    return True


def save_uploaded_permanent_revision(db: Session, *, client: Client, upload_file, title: str, category: str, description: str | None, remarks: str | None, user, existing_document_id: int | None = None) -> PermanentClientDocument:
    original_filename = Path(upload_file.filename or "document.bin").name
    category = category if category in PERMANENT_DOCUMENT_CATEGORIES else "Other Permanent Documents"
    if existing_document_id:
        document = db.get(PermanentClientDocument, existing_document_id)
        if not document or document.is_deleted or document.client_id != client.id:
            raise ValueError("Invalid permanent document selected for new revision.")
        if title:
            document.title = title.strip()[:255]
        if description is not None:
            document.description = description.strip() or None
        document.category = category
    else:
        document = PermanentClientDocument(
            tenant_id=client.tenant_id,
            branch_id=getattr(client, "branch_id", None),
            client_id=client.id,
            document_code="PENDING",
            category=category,
            title=(title.strip()[:255] if title else original_filename[:255]),
            description=description.strip() if description else None,
            created_by_user_id=getattr(user, "id", None),
            updated_by_user_id=getattr(user, "id", None),
        )
        db.add(document); db.flush()
        document.document_code = create_permanent_document_code(db, client.tenant_id, client.id, document.id)
    next_version_no = int(document.current_version_no or 0) + 1
    rel_path = build_permanent_relative_path(db, client, document, next_version_no, original_filename)
    abs_path = DEFAULT_STORAGE_ROOT / rel_path
    abs_path.parent.mkdir(parents=True, exist_ok=True)
    hasher = hashlib.sha256(); total = 0
    with abs_path.open("wb") as out:
        while True:
            chunk = upload_file.file.read(1024 * 1024)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_UPLOAD_BYTES:
                out.close(); abs_path.unlink(missing_ok=True)
                raise ValueError(f"File exceeds maximum upload limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB.")
            hasher.update(chunk); out.write(chunk)
    version = PermanentClientDocumentVersion(
        document_id=document.id,
        tenant_id=document.tenant_id,
        branch_id=document.branch_id,
        client_id=document.client_id,
        version_no=next_version_no,
        original_filename=original_filename,
        stored_filename=abs_path.name,
        content_type=getattr(upload_file, "content_type", None),
        file_size_bytes=total,
        file_hash_sha256=hasher.hexdigest(),
        local_relative_path=str(rel_path).replace("\\", "/"),
        remarks=remarks.strip() if remarks else None,
        uploaded_by_user_id=getattr(user, "id", None),
    )
    document.current_version_no = next_version_no
    document.updated_by_user_id = getattr(user, "id", None)
    db.add(version); db.flush()
    create_permanent_storage_job_for_version(db, version=version, user=user)
    return document


def permanent_version_absolute_path(version: PermanentClientDocumentVersion) -> Path:
    return DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)


def version_absolute_path(version: EngagementDocumentVersion) -> Path:
    return DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)
