# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Implementation of Amazon S3 file backend."""

import datetime as dt
import filecmp
import tempfile
from collections.abc import Generator
from contextlib import contextmanager
from pathlib import Path
from typing import Self, TYPE_CHECKING, override
from urllib.parse import urlsplit

import boto3

from debusine.assets.models import AWSProviderAccountData, CloudProvidersType
from debusine.db.models import File, FileStore
from debusine.server.file_backend.interface import (
    FileBackendEntryInterface,
    FileBackendInterface,
)
from debusine.server.file_backend.models import S3FileBackendConfiguration

if TYPE_CHECKING:
    from mypy_boto3_s3.client import S3Client

    S3Client  # fake usage for vulture


class S3FileBackendEntry(FileBackendEntryInterface["S3FileBackend", str]):
    """
    An entry in an Amazon S3 file backend.

    S3 identifiers are "hash-size", where "hash" is the SHA-256 hex digest
    of the file contents and "size" is the file size in bytes.
    """

    @classmethod
    def _create_from_file(cls, backend: "S3FileBackend", fileobj: File) -> Self:
        """Make an entry based on a `File`."""
        return cls(
            backend=backend, identifier=f"{fileobj.sha256.hex()}-{fileobj.size}"
        )

    @override
    def __init__(
        self,
        backend: "S3FileBackend",
        identifier: str,
        mtime: dt.datetime | None = None,
    ) -> None:
        super().__init__(backend, identifier)
        self._mtime = mtime

    @override
    def __str__(self) -> str:
        """Return a string describing this entry."""
        return self._identifier

    @override
    def get_local_path(self) -> None:
        """Return None: no local paths for S3 files."""
        return None

    @override
    @contextmanager
    def get_temporary_local_path(self) -> Generator[Path]:
        """Yield a temporary local path for the file."""
        with tempfile.NamedTemporaryFile(
            prefix="debusine-s3-", delete_on_close=False
        ) as temp_file:
            self.backend.client.download_fileobj(
                self.backend.configuration.bucket_name,
                self._identifier,
                temp_file,
            )
            temp_file.close()
            yield Path(temp_file.name)

    @override
    def get_url(self, content_type: str | None = None) -> str:
        """Return a presigned URL for the file."""
        params = {
            "Bucket": self.backend.configuration.bucket_name,
            "Key": self._identifier,
        }
        if content_type is not None:
            params["ResponseContentType"] = content_type
        url = self.backend.client.generate_presigned_url(
            "get_object",
            Params=params,
            # Hardcoded to one minute for now.  These URLs are currently
            # returned in redirects which are presumably followed promptly
            # by clients.
            ExpiresIn=60,
        )
        assert isinstance(url, str)
        return url

    @override
    def get_mtime(self) -> dt.datetime:
        if self._mtime is None:
            # We could possibly support this by making a separate query, but
            # at the moment the only thing that wants this is
            # vacuum_storage, and it gets the mtimes in bulk from
            # list_entries.
            raise AssertionError(
                "Cannot get modification time; S3 file backend entry was not"
                " returned from list_entries"
            )
        return self._mtime

    @override
    def add(self, local_path: Path) -> None:
        extra_args = {
            "ChecksumAlgorithm": "SHA256",
        }
        if self.backend.configuration.storage_class is not None:
            extra_args["StorageClass"] = (
                self.backend.configuration.storage_class
            )
        self.backend.client.upload_file(
            str(local_path),
            self.backend.configuration.bucket_name,
            self._identifier,
            ExtraArgs=extra_args,
        )

    @override
    def same_contents(self, local_path: Path) -> bool:
        with self.get_temporary_local_path() as temp_path:
            # This standard library function maintains a cache indexed by
            # the paths and stat signatures (type, size, mtime) of both
            # files.  Since `temp_path` was always created in this method,
            # the cache will never be hit.
            return filecmp.cmp(local_path, temp_path, shallow=False)

    @override
    def remove(self) -> None:
        self.backend.client.delete_object(
            Bucket=self.backend.configuration.bucket_name, Key=self._identifier
        )


class S3FileBackend(FileBackendInterface[S3FileBackendConfiguration]):
    """Amazon S3 file backend."""

    def __init__(self, file_store: FileStore) -> None:
        """Initialize `S3FileBackend`."""
        super().__init__()
        self.db_store = file_store
        if file_store.provider_account is None:
            raise RuntimeError(
                "S3 file backend requires a cloud provider account"
            )
        if (
            file_store.provider_account.data.get("provider_type")
            != CloudProvidersType.AWS
        ):
            raise RuntimeError(
                "S3 file backend requires an AWS cloud provider account"
            )
        provider_account_data = file_store.provider_account.data_model
        assert isinstance(provider_account_data, AWSProviderAccountData)
        s3_endpoint_url = provider_account_data.configuration.s3_endpoint_url
        self._is_hetzner = s3_endpoint_url is not None and urlsplit(
            s3_endpoint_url
        ).netloc.endswith(".your-objectstorage.com")
        self.client = self._make_client(provider_account_data)

    def _make_client(
        self, provider_account_data: AWSProviderAccountData
    ) -> "S3Client":
        """
        Make an S3 client for this provider account.

        This exists mainly so that it can be patched by tests.
        """
        return boto3.client(
            "s3",
            region_name=provider_account_data.configuration.region_name,
            endpoint_url=provider_account_data.configuration.s3_endpoint_url,
            aws_access_key_id=provider_account_data.credentials.access_key_id,
            aws_secret_access_key=(
                provider_account_data.credentials.secret_access_key
            ),
        )

    @override
    def get_entry(self, fileobj: File) -> S3FileBackendEntry:
        return S3FileBackendEntry._create_from_file(self, fileobj)

    @override
    def list_entries(self) -> Generator[S3FileBackendEntry]:
        paginator = self.client.get_paginator("list_objects_v2")
        iterator = paginator.paginate(Bucket=self.configuration.bucket_name)
        for page in iterator:
            for item in page.get("Contents", []):
                yield S3FileBackendEntry(
                    backend=self,
                    identifier=item["Key"],
                    mtime=item["LastModified"],
                )
