"""Chunked upload service for tus.io resumable uploads (Issue #788).

Manages the lifecycle of chunked upload sessions:
- Session creation with semaphore-based concurrency limiting
- Chunk reception with offset validation and checksum verification
- Session state persistence via SQLAlchemy
- Background cleanup of expired sessions
- Per-session locking for concurrent chunk uploads
"""

import asyncio
import base64
import hashlib
import logging
import uuid
import zlib
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any, cast

from nexus.contracts.constants import ROOT_ZONE_ID
from nexus.contracts.exceptions import (
    UploadChecksumMismatchError,
    UploadExpiredError,
    UploadNotFoundError,
    UploadOffsetMismatchError,
    ValidationError,
)
from nexus.storage.models.upload_session import UploadSessionModel

from .upload_session import UploadSession, UploadStatus

if TYPE_CHECKING:
    from nexus.backends.engines.multipart import MultipartUpload
    from nexus.core.protocols.connector import ConnectorProtocol
    from nexus.storage.record_store import RecordStoreABC

logger = logging.getLogger(__name__)

# tus protocol constants
TUS_VERSION = "1.0.0"
TUS_EXTENSIONS = "creation,termination,checksum,expiration"
TUS_CHECKSUM_ALGORITHMS = "sha256,md5,crc32"

# Default config values
DEFAULT_MAX_CONCURRENT = 20
DEFAULT_SESSION_TTL_HOURS = 24
DEFAULT_CLEANUP_INTERVAL_SECONDS = 3600
DEFAULT_MIN_CHUNK_SIZE = 5 * 1024 * 1024  # 5 MB
DEFAULT_MAX_CHUNK_SIZE = 64 * 1024 * 1024  # 64 MB
DEFAULT_CHUNK_SIZE = 10 * 1024 * 1024  # 10 MB


class ChunkedUploadConfig:
    """Configuration for the chunked upload service."""

    def __init__(
        self,
        max_concurrent_uploads: int = DEFAULT_MAX_CONCURRENT,
        session_ttl_hours: int = DEFAULT_SESSION_TTL_HOURS,
        cleanup_interval_seconds: int = DEFAULT_CLEANUP_INTERVAL_SECONDS,
        min_chunk_size: int = DEFAULT_MIN_CHUNK_SIZE,
        max_chunk_size: int = DEFAULT_MAX_CHUNK_SIZE,
        default_chunk_size: int = DEFAULT_CHUNK_SIZE,
        max_upload_size: int = 10 * 1024 * 1024 * 1024,  # 10 GB
    ):
        self.max_concurrent_uploads = max_concurrent_uploads
        self.session_ttl_hours = session_ttl_hours
        self.cleanup_interval_seconds = cleanup_interval_seconds
        self.min_chunk_size = min_chunk_size
        self.max_chunk_size = max_chunk_size
        self.default_chunk_size = default_chunk_size
        self.max_upload_size = max_upload_size


class ChunkedUploadService:
    """Core service for tus.io resumable chunked uploads.

    Manages upload session lifecycle, chunk handling, and backend integration.
    Uses asyncio.Semaphore for concurrency limiting and per-session locks
    for safe concurrent access.

    Architecture:
        - Session state is persisted in SQLAlchemy (survives server restarts)
        - Chunk data flows through backend's MultipartUpload interface if available
        - Falls back to temp directory assembly for non-multipart backends
        - Checksum verification per-chunk (sha256, md5, crc32)
        - TTL-based expiration with background sweep + lazy cleanup
    """

    def __init__(
        self,
        record_store: "RecordStoreABC",
        backend: "ConnectorProtocol",
        metadata_store: Any = None,
        config: ChunkedUploadConfig | None = None,
        nexus_fs: Any = None,
    ):
        self._session_factory = record_store.session_factory
        self._backend = backend
        self._metadata_store = metadata_store
        self._nexus_fs = nexus_fs
        # Pull the kernel out of the proxy so the metadata put inside
        # ``finalize`` lands on ``kernel.metastore_*`` (survives W3).
        self._kernel = metadata_store
        self._config = config or ChunkedUploadConfig()

        self._semaphore = asyncio.Semaphore(self._config.max_concurrent_uploads)
        self._session_locks: dict[str, asyncio.Lock] = {}
        self._parts_registry: dict[str, list[dict[str, Any]]] = {}
        self._last_cleanup = datetime.now(UTC)
        self._cleanup_task: asyncio.Task[None] | None = None

    # --- Public API ---

    def attach_filesystem(self, nexus_fs: Any) -> None:
        """Attach NexusFS so completed uploads use the normal write pipeline."""
        self._nexus_fs = nexus_fs

    async def create_upload(
        self,
        target_path: str,
        upload_length: int,
        *,
        metadata: dict[str, str] | None = None,
        zone_id: str = ROOT_ZONE_ID,
        user_id: str = "anonymous",
        checksum_algorithm: str | None = None,
        context: Any | None = None,
    ) -> UploadSession:
        """Create a new upload session.

        Args:
            target_path: Virtual path where the final file will be written.
            upload_length: Total declared file size in bytes.
            metadata: Optional tus Upload-Metadata key-value pairs.
            zone_id: Zone ID for permission scoping.
            user_id: User who initiated the upload.
            checksum_algorithm: Preferred checksum algorithm.

        Returns:
            New UploadSession with upload_id and expiration.

        Raises:
            ValidationError: If upload_length is invalid.
            RuntimeError: If concurrency limit reached (429).
        """
        # Validate
        if upload_length < 0:
            raise ValidationError(f"upload_length must be non-negative, got {upload_length}")
        if upload_length > self._config.max_upload_size:
            raise ValidationError(
                f"upload_length {upload_length} exceeds maximum {self._config.max_upload_size}"
            )

        # Lazy cleanup on create
        await self._lazy_cleanup()

        await self._enforce_write_permission(target_path, context)

        # Acquire semaphore (non-blocking — return 429 if full)
        if not self._semaphore.locked():
            await self._semaphore.acquire()
        else:
            # Semaphore value is 0 — all permits taken
            raise RuntimeError("Too many concurrent uploads")

        try:
            upload_id = str(uuid.uuid4())
            now = datetime.now(UTC)
            expires_at = now + timedelta(hours=self._config.session_ttl_hours)

            # Initialize backend multipart only for standalone services. When
            # NexusFS is attached, completed uploads must flow through
            # NexusFS.write so permission, metadata, and audit hooks match
            # ordinary writes.
            backend_upload_id: str | None = None
            backend_name: str | None = None
            if self._nexus_fs is None and self._supports_multipart():
                mixin = cast("MultipartUpload", self._backend)
                backend_upload_id = await asyncio.to_thread(
                    mixin.init_multipart,
                    target_path,
                    metadata.get("content_type", "application/octet-stream")
                    if metadata
                    else "application/octet-stream",
                    metadata,
                )
                backend_name = getattr(self._backend, "name", "unknown")

            session = UploadSession(
                upload_id=upload_id,
                target_path=target_path,
                upload_length=upload_length,
                upload_offset=0,
                status=UploadStatus.CREATED,
                zone_id=zone_id,
                user_id=user_id,
                metadata=metadata or {},
                checksum_algorithm=checksum_algorithm,
                created_at=now,
                expires_at=expires_at,
                backend_upload_id=backend_upload_id,
                backend_name=backend_name,
                parts=(),
            )

            await self._persist_session(session)
            self._parts_registry[upload_id] = []

            logger.info(
                "Upload session created: %s -> %s (%d bytes, expires %s)",
                upload_id,
                target_path,
                upload_length,
                expires_at.isoformat(),
            )
            return session

        except Exception:
            self._semaphore.release()
            raise

    async def receive_chunk(
        self,
        upload_id: str,
        offset: int,
        chunk_data: bytes,
        checksum_header: str | None = None,
        context: Any | None = None,
    ) -> UploadSession:
        """Receive and store a chunk of data.

        Args:
            upload_id: Upload session ID.
            offset: Expected byte offset for this chunk.
            chunk_data: Raw chunk bytes.
            checksum_header: Optional tus Upload-Checksum header value.

        Returns:
            Updated UploadSession with new offset.

        Raises:
            UploadNotFoundError: If session doesn't exist.
            UploadExpiredError: If session has expired.
            UploadOffsetMismatchError: If offset doesn't match current position.
            UploadChecksumMismatchError: If checksum verification fails.
            ValidationError: If chunk size is invalid.
        """
        lock = self._get_or_create_session_lock(upload_id)
        async with lock:
            session = await self._load_session(upload_id)

            # Validate session state
            if session.status in (UploadStatus.COMPLETED, UploadStatus.TERMINATED):
                raise UploadNotFoundError(upload_id, f"Upload is {session.status.value}")

            if session.is_expired:
                await self._expire_session(session)
                raise UploadExpiredError(upload_id)

            await self._enforce_write_permission(session.target_path, context)

            # Validate offset
            if offset != session.upload_offset:
                raise UploadOffsetMismatchError(upload_id, session.upload_offset, offset)

            # Validate chunk size
            chunk_size = len(chunk_data)
            remaining = session.remaining_bytes
            is_last_chunk = chunk_size == remaining

            if chunk_size > remaining:
                raise ValidationError(
                    f"Chunk would exceed upload_length: "
                    f"offset={offset} + chunk={chunk_size} > total={session.upload_length}"
                )

            if not is_last_chunk and chunk_size < self._config.min_chunk_size:
                raise ValidationError(
                    f"Chunk size {chunk_size} below minimum {self._config.min_chunk_size} "
                    f"(except for last chunk)"
                )

            if chunk_size > self._config.max_chunk_size:
                raise ValidationError(
                    f"Chunk size {chunk_size} exceeds maximum {self._config.max_chunk_size}"
                )

            # Verify checksum if provided
            if checksum_header:
                self._verify_checksum(upload_id, chunk_data, checksum_header)

            # Store chunk
            part_number = session.parts_received + 1
            part_info = await self._store_chunk(session, part_number, chunk_data)

            parts = list(session.parts)
            parts.append(part_info)
            self._parts_registry[upload_id] = parts

            # Update session
            new_offset = session.upload_offset + chunk_size
            new_status = UploadStatus.IN_PROGRESS

            updated = UploadSession(
                upload_id=session.upload_id,
                target_path=session.target_path,
                upload_length=session.upload_length,
                upload_offset=new_offset,
                status=new_status,
                zone_id=session.zone_id,
                user_id=session.user_id,
                metadata=session.metadata,
                checksum_algorithm=session.checksum_algorithm,
                created_at=session.created_at,
                expires_at=session.expires_at,
                backend_upload_id=session.backend_upload_id,
                backend_name=session.backend_name,
                parts_received=part_number,
                parts=tuple(parts),
                content_id=session.content_id,
            )

            # Persist partial progress only after we know the upload still needs
            # more chunks. For the final chunk, defer persistence until the
            # assembled write succeeds so a transient finalize failure stays
            # retryable at the previous offset.
            if updated.is_complete:
                updated = await self._assemble_and_write(updated, parts, context=context)
            else:
                await self._update_session(updated)

            return updated

    async def get_upload_status(self, upload_id: str) -> UploadSession:
        """Get the current status of an upload session.

        Args:
            upload_id: Upload session ID.

        Returns:
            Current UploadSession state.

        Raises:
            UploadNotFoundError: If session doesn't exist or was terminated.
            UploadExpiredError: If session has expired.
        """
        session = await self._load_session(upload_id)

        # Terminated uploads are no longer accessible (tus termination extension)
        if session.status == UploadStatus.TERMINATED:
            raise UploadNotFoundError(upload_id)

        if session.is_expired and session.status not in (
            UploadStatus.COMPLETED,
            UploadStatus.EXPIRED,
        ):
            await self._expire_session(session)
            raise UploadExpiredError(upload_id)

        return session

    async def terminate_upload(self, upload_id: str) -> None:
        """Terminate an upload and clean up resources.

        Args:
            upload_id: Upload session ID.

        Raises:
            UploadNotFoundError: If session doesn't exist.
        """
        session = await self._load_session(upload_id)

        # Only release semaphore for sessions that actually hold a permit
        should_release = session.status in (UploadStatus.CREATED, UploadStatus.IN_PROGRESS)

        # Abort backend multipart if active
        if session.backend_upload_id and self._supports_multipart():
            try:
                mixin = cast("MultipartUpload", self._backend)
                await asyncio.to_thread(
                    mixin.abort_multipart,
                    session.target_path,
                    session.backend_upload_id,
                )
            except Exception as e:
                logger.warning("Failed to abort backend multipart for %s: %s", upload_id, e)

        # Mark as terminated
        terminated = UploadSession(
            upload_id=session.upload_id,
            target_path=session.target_path,
            upload_length=session.upload_length,
            upload_offset=session.upload_offset,
            status=UploadStatus.TERMINATED,
            zone_id=session.zone_id,
            user_id=session.user_id,
            metadata=session.metadata,
            checksum_algorithm=session.checksum_algorithm,
            created_at=session.created_at,
            expires_at=session.expires_at,
            backend_upload_id=session.backend_upload_id,
            backend_name=session.backend_name,
            parts_received=session.parts_received,
            parts=session.parts,
            content_id=session.content_id,
        )
        await self._update_session(terminated)

        # Clean up
        self._parts_registry.pop(upload_id, None)
        self._session_locks.pop(upload_id, None)
        if should_release:
            self._semaphore.release()

        logger.info("Upload session terminated: %s", upload_id)

    def get_server_capabilities(self) -> dict[str, str]:
        """Return tus server capability headers for OPTIONS response."""
        return {
            "Tus-Resumable": TUS_VERSION,
            "Tus-Version": TUS_VERSION,
            "Tus-Extension": TUS_EXTENSIONS,
            "Tus-Max-Size": str(self._config.max_upload_size),
            "Tus-Checksum-Algorithm": TUS_CHECKSUM_ALGORITHMS,
        }

    async def cleanup_expired(self) -> int:
        """Sweep and clean up expired sessions.

        Returns:
            Number of sessions cleaned up.
        """
        now = datetime.now(UTC)
        cleaned = 0

        def _query_expired() -> list[dict[str, Any]]:
            db = self._session_factory()
            try:
                from sqlalchemy import select

                rows = (
                    db.execute(
                        select(UploadSessionModel).where(
                            UploadSessionModel.expires_at < now,
                            UploadSessionModel.status.in_(["created", "in_progress"]),
                        )
                    )
                    .scalars()
                    .all()
                )
                return [row.to_session_dict() for row in rows]
            finally:
                db.close()

        expired_dicts = await asyncio.to_thread(_query_expired)

        for session_dict in expired_dicts:
            try:
                session = UploadSession.from_dict(session_dict)
                await self._expire_session(session)
                cleaned += 1
            except Exception as e:
                logger.warning(
                    "Failed to clean up expired session %s: %s",
                    session_dict.get("upload_id"),
                    e,
                )

        if cleaned > 0:
            logger.info("Cleaned up %d expired upload sessions", cleaned)

        self._last_cleanup = now
        return cleaned

    async def start_cleanup_loop(self) -> None:
        """Start the background cleanup loop."""
        while True:
            try:
                await asyncio.sleep(self._config.cleanup_interval_seconds)
                await self.cleanup_expired()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error("Error in upload cleanup loop: %s", e)

    # --- Internal helpers ---

    def _supports_multipart(self) -> bool:
        """Check if backend supports native multipart uploads."""
        from nexus.backends.engines.multipart import MultipartUpload

        return isinstance(self._backend, MultipartUpload)

    def _get_or_create_session_lock(self, upload_id: str) -> asyncio.Lock:
        """Get or create a per-session asyncio.Lock."""
        if upload_id not in self._session_locks:
            self._session_locks[upload_id] = asyncio.Lock()
        return self._session_locks[upload_id]

    async def _persist_session(self, session: UploadSession) -> None:
        """Persist a new session to the database."""

        def _write(s: UploadSession) -> None:
            db = self._session_factory()
            try:
                model = UploadSessionModel.from_upload_session(s)
                db.add(model)
                db.commit()
            except Exception:
                db.rollback()
                raise
            finally:
                db.close()

        await asyncio.to_thread(_write, session)

    async def _load_session(self, upload_id: str) -> UploadSession:
        """Load a session from the database.

        Raises:
            UploadNotFoundError: If session doesn't exist.
        """

        def _read(uid: str) -> dict[str, Any] | None:
            db = self._session_factory()
            try:
                from sqlalchemy import select

                model = (
                    db.execute(select(UploadSessionModel).filter_by(upload_id=uid))
                    .scalars()
                    .first()
                )
                if model is None:
                    return None
                return model.to_session_dict()
            finally:
                db.close()

        result = await asyncio.to_thread(_read, upload_id)
        if result is None:
            raise UploadNotFoundError(upload_id)
        return UploadSession.from_dict(result)

    async def _update_session(self, session: UploadSession) -> None:
        """Update an existing session in the database."""

        def _write(s: UploadSession) -> None:
            db = self._session_factory()
            try:
                from sqlalchemy import select

                model = (
                    db.execute(select(UploadSessionModel).filter_by(upload_id=s.upload_id))
                    .scalars()
                    .first()
                )
                if model is None:
                    raise UploadNotFoundError(s.upload_id)
                model.upload_offset = s.upload_offset
                model.status = s.status.value
                model.parts_received = s.parts_received
                model.content_id = s.content_id
                model.backend_upload_id = s.backend_upload_id
                model.metadata_json = UploadSessionModel.encode_metadata_state(
                    s.metadata,
                    s.parts,
                )
                db.commit()
            except Exception:
                db.rollback()
                raise
            finally:
                db.close()

        await asyncio.to_thread(_write, session)

    async def _delete_session(self, upload_id: str) -> None:
        """Delete a session from the database."""

        def _delete(uid: str) -> None:
            db = self._session_factory()
            try:
                from sqlalchemy import delete

                db.execute(delete(UploadSessionModel).filter_by(upload_id=uid))
                db.commit()
            except Exception:
                db.rollback()
                raise
            finally:
                db.close()

        await asyncio.to_thread(_delete, upload_id)

    async def _store_chunk(
        self,
        session: UploadSession,
        part_number: int,
        data: bytes,
    ) -> dict[str, Any]:
        """Store a chunk via the backend.

        Returns:
            Part info dict with at least "etag" and "part_number".
        """
        if self._supports_multipart() and session.backend_upload_id:
            mixin = cast("MultipartUpload", self._backend)
            return await asyncio.to_thread(
                mixin.upload_part,
                session.target_path,
                session.backend_upload_id,
                part_number,
                data,
            )

        # Fallback: write chunk directly to CAS and track
        chunk_hash = hashlib.sha256(data).hexdigest()
        result = await asyncio.to_thread(self._backend.write_content, data)
        return {
            "etag": chunk_hash,
            "part_number": part_number,
            "content_id": result.content_id,
        }

    async def _assemble_and_write(
        self,
        session: UploadSession,
        parts: list[dict[str, Any]],
        *,
        context: Any | None = None,
    ) -> UploadSession:
        """Assemble chunks and write the final file.

        For multipart backends, delegates to complete_multipart().
        For others, reads parts and concatenates.

        Returns:
            Updated session with COMPLETED status and content_id.
        """
        content_id: str

        if self._supports_multipart() and session.backend_upload_id:
            await self._enforce_write_permission(session.target_path, context)
            mixin = cast("MultipartUpload", self._backend)
            content_id = await asyncio.to_thread(
                mixin.complete_multipart,
                session.target_path,
                session.backend_upload_id,
                parts,
            )
        else:
            # Read all chunks and concatenate
            sorted_parts = sorted(parts, key=lambda p: p["part_number"])
            assembled = bytearray()
            for part_info in sorted_parts:
                part_hash = part_info.get("content_id")
                if part_hash:
                    content = await asyncio.to_thread(self._backend.read_content, part_hash)
                    assembled.extend(content)

            # Write assembled content to CAS
            content = bytes(assembled)
            if self._nexus_fs is not None:
                result = await asyncio.to_thread(
                    self._nexus_fs.write,
                    session.target_path,
                    content,
                    context=context,
                )
                content_id = result["content_id"]
            else:
                _write = self._backend.write_content
                result = await asyncio.to_thread(_write, content)
                content_id = result.content_id

            # Link content to target_path via metadata store when NexusFS is not
            # attached. NexusFS.write handles metadata, hooks, and observers.
            if self._nexus_fs is None and self._kernel is not None:
                from nexus.contracts.metadata import FileMetadata

                metadata = FileMetadata(
                    path=session.target_path,
                    size=len(content),
                    content_id=content_id,
                )
                write_metadata = getattr(self._kernel, "metastore_put", None)
                if write_metadata is None:
                    write_metadata = self._kernel.put
                await asyncio.to_thread(write_metadata, metadata)

        # Update session to completed
        completed = UploadSession(
            upload_id=session.upload_id,
            target_path=session.target_path,
            upload_length=session.upload_length,
            upload_offset=session.upload_length,
            status=UploadStatus.COMPLETED,
            zone_id=session.zone_id,
            user_id=session.user_id,
            metadata=session.metadata,
            checksum_algorithm=session.checksum_algorithm,
            created_at=session.created_at,
            expires_at=session.expires_at,
            backend_upload_id=session.backend_upload_id,
            backend_name=session.backend_name,
            parts_received=session.parts_received,
            parts=tuple(parts),
            content_id=content_id,
        )
        await self._update_session(completed)

        # Clean up tracking
        self._parts_registry.pop(session.upload_id, None)
        self._session_locks.pop(session.upload_id, None)
        self._semaphore.release()

        logger.info(
            "Upload completed: %s -> %s (hash=%s)",
            session.upload_id,
            session.target_path,
            content_id,
        )
        return completed

    async def _expire_session(self, session: UploadSession) -> None:
        """Mark a session as expired and clean up."""
        # Abort backend multipart if active
        if session.backend_upload_id and self._supports_multipart():
            try:
                mixin = cast("MultipartUpload", self._backend)
                await asyncio.to_thread(
                    mixin.abort_multipart,
                    session.target_path,
                    session.backend_upload_id,
                )
            except Exception as e:
                logger.warning(
                    "Failed to abort backend multipart for expired %s: %s",
                    session.upload_id,
                    e,
                )

        expired = UploadSession(
            upload_id=session.upload_id,
            target_path=session.target_path,
            upload_length=session.upload_length,
            upload_offset=session.upload_offset,
            status=UploadStatus.EXPIRED,
            zone_id=session.zone_id,
            user_id=session.user_id,
            metadata=session.metadata,
            checksum_algorithm=session.checksum_algorithm,
            created_at=session.created_at,
            expires_at=session.expires_at,
            backend_upload_id=session.backend_upload_id,
            backend_name=session.backend_name,
            parts_received=session.parts_received,
            parts=session.parts,
            content_id=session.content_id,
        )

        try:
            await self._update_session(expired)
        except Exception as e:
            logger.debug("Failed to update expired session: %s", e)

        self._parts_registry.pop(session.upload_id, None)
        self._session_locks.pop(session.upload_id, None)

        # Release semaphore if session was active
        if session.status in (UploadStatus.CREATED, UploadStatus.IN_PROGRESS):
            self._semaphore.release()

    async def _lazy_cleanup(self) -> None:
        """Trigger cleanup if enough time has passed since last sweep."""
        now = datetime.now(UTC)
        elapsed = (now - self._last_cleanup).total_seconds()
        if elapsed >= self._config.cleanup_interval_seconds:
            try:
                await self.cleanup_expired()
            except Exception as e:
                logger.warning("Lazy cleanup failed: %s", e)

    async def _enforce_write_permission(self, path: str, context: Any | None) -> None:
        if context is None or self._nexus_fs is None:
            return
        enforce = getattr(self._nexus_fs, "_enforce_lock_write_permission", None)
        if callable(enforce):
            await asyncio.to_thread(enforce, path, context)

    @staticmethod
    def _verify_checksum(
        upload_id: str,
        data: bytes,
        checksum_header: str,
    ) -> None:
        """Verify chunk checksum against tus Upload-Checksum header.

        Format: "<algorithm> <base64_digest>"
        Supported: sha256, md5, crc32

        Raises:
            UploadChecksumMismatchError: If checksum doesn't match.
            ValidationError: If algorithm is unsupported.
        """
        parts = checksum_header.strip().split(" ", 1)
        if len(parts) != 2:
            raise ValidationError(f"Invalid Upload-Checksum header format: {checksum_header}")

        algorithm, expected_b64 = parts[0].lower(), parts[1]

        if algorithm == "sha256":
            actual = base64.b64encode(hashlib.sha256(data).digest()).decode()
        elif algorithm == "md5":
            actual = base64.b64encode(hashlib.md5(data).digest()).decode()  # noqa: S324
        elif algorithm == "crc32":
            crc = zlib.crc32(data) & 0xFFFFFFFF
            actual = base64.b64encode(crc.to_bytes(4, "big")).decode()
        else:
            raise ValidationError(
                f"Unsupported checksum algorithm: {algorithm}. Supported: {TUS_CHECKSUM_ALGORITHMS}"
            )

        if actual != expected_b64:
            raise UploadChecksumMismatchError(upload_id, algorithm)
