from __future__ import annotations

import base64
import calendar
import copy
import hashlib
import io
import json
import logging
import os
import re
import tarfile
import threading
import time
import warnings
import weakref
import zipfile
from collections import defaultdict
from collections.abc import Iterable
from datetime import datetime
from gzip import GzipFile
from sys import platform
from typing import Any, Optional, TypedDict, Union

import requests.exceptions

from moto import settings
from moto.awslambda.policy import Policy
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.common_models import BaseModel, CloudFormationModel
from moto.core.exceptions import RESTError
from moto.core.utils import iso_8601_datetime_with_nanoseconds, unix_time_millis, utcnow
from moto.dynamodb import dynamodb_backends
from moto.dynamodbstreams import dynamodbstreams_backends
from moto.ecr.exceptions import ImageNotFoundException
from moto.ecr.models import ecr_backends
from moto.iam.exceptions import NotFoundException as IAMNotFoundException
from moto.iam.models import iam_backends
from moto.kinesis.models import KinesisBackend, kinesis_backends
from moto.logs.models import logs_backends
from moto.moto_api._internal import mock_random as random
from moto.s3.exceptions import MissingBucket, MissingKey
from moto.s3.models import FakeKey, s3_backends
from moto.sqs.models import sqs_backends
from moto.utilities.docker_utilities import DockerModel
from moto.utilities.utils import (
    ARN_PARTITION_REGEX,
    get_partition,
    load_resource_as_bytes,
)

from .exceptions import (
    ConflictException,
    CrossAccountNotAllowed,
    GenericResourcNotFound,
    InvalidParameterValueException,
    InvalidRoleFormat,
    LambdaClientError,
    UnknownAliasException,
    UnknownEventConfig,
    UnknownFunctionException,
    UnknownLayerException,
    UnknownLayerVersionException,
    ValidationException,
)
from .utils import (
    make_event_source_mapping_arn,
    make_function_arn,
    make_function_ver_arn,
    make_layer_arn,
    make_layer_ver_arn,
    split_layer_arn,
)

logger = logging.getLogger(__name__)


class LayerDataType(TypedDict):
    Arn: str
    CodeSize: int


def zip2tar(zip_bytes: bytes) -> io.BytesIO:
    tarstream = io.BytesIO()
    timeshift = int((datetime.now() - utcnow()).total_seconds())
    tarf = tarfile.TarFile(fileobj=tarstream, mode="w")
    with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zipf:
        for zipinfo in zipf.infolist():
            if zipinfo.is_dir():
                continue

            tarinfo = tarfile.TarInfo(name=zipinfo.filename)
            tarinfo.size = zipinfo.file_size
            tarinfo.mtime = calendar.timegm(zipinfo.date_time) - timeshift
            infile = zipf.open(zipinfo.filename)
            tarf.addfile(tarinfo, infile)

    tarstream.seek(0)
    return tarstream


def file2tar(file_content: bytes, file_name: str) -> io.BytesIO:
    tarstream = io.BytesIO()
    tarf = tarfile.TarFile(fileobj=tarstream, mode="w")
    tarinfo = tarfile.TarInfo(name=file_name)
    tarinfo.size = len(file_content)
    tarf.addfile(tarinfo, io.BytesIO(file_content))

    tarstream.seek(0)
    return tarstream


class _VolumeRefCount:
    __slots__ = "refcount", "volume"

    def __init__(self, refcount: int, volume: Any):
        self.refcount = refcount
        self.volume = volume


class _DockerDataVolumeContext:
    # {sha256: _VolumeRefCount}
    _data_vol_map: dict[str, _VolumeRefCount] = defaultdict(
        lambda: _VolumeRefCount(0, None)
    )
    _lock = threading.Lock()

    def __init__(self, lambda_func: LambdaFunction):
        self._lambda_func = lambda_func
        self._vol_ref: Optional[_VolumeRefCount] = None

    @property
    def name(self) -> str:
        return self._vol_ref.volume.name  # type: ignore[union-attr]

    def __enter__(self) -> _DockerDataVolumeContext:
        # See if volume is already known
        with self.__class__._lock:
            self._vol_ref = self.__class__._data_vol_map[self._lambda_func.code_digest]
            self._vol_ref.refcount += 1
            if self._vol_ref.refcount > 1:
                return self

            # See if the volume already exists
            for vol in self._lambda_func.docker_client.volumes.list():
                if vol.name == self._lambda_func.code_digest:
                    self._vol_ref.volume = vol
                    return self

            # It doesn't exist so we need to create it
            self._vol_ref.volume = self._lambda_func.docker_client.volumes.create(
                self._lambda_func.code_digest
            )
            volumes = {self.name: {"bind": settings.LAMBDA_DATA_DIR, "mode": "rw"}}

            self._lambda_func.ensure_image_exists("busybox")
            container = self._lambda_func.docker_client.containers.run(
                "busybox", "sleep 100", volumes=volumes, detach=True
            )
            try:
                with zip2tar(self._lambda_func.code_bytes) as stream:
                    container.put_archive(settings.LAMBDA_DATA_DIR, stream)
                if settings.is_test_proxy_mode():
                    ca_cert = load_resource_as_bytes(__name__, "../moto_proxy/ca.crt")
                    with file2tar(ca_cert, "ca.crt") as cert_stream:
                        container.put_archive(settings.LAMBDA_DATA_DIR, cert_stream)
            finally:
                container.remove(force=True)

        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        with self.__class__._lock:
            self._vol_ref.refcount -= 1  # type: ignore[union-attr]
            if self._vol_ref.refcount == 0:  # type: ignore[union-attr]
                import docker.errors

                try:
                    self._vol_ref.volume.remove()  # type: ignore[union-attr]
                except docker.errors.APIError as e:
                    if e.status_code != 409:
                        raise

                    raise  # multiple processes trying to use same volume?


class _DockerDataVolumeLayerContext:
    _data_vol_map: dict[str, _VolumeRefCount] = defaultdict(
        lambda: _VolumeRefCount(0, None)
    )
    _lock = threading.Lock()

    def __init__(self, lambda_func: LambdaFunction):
        self._lambda_func = lambda_func
        self._layers: list[LayerDataType] = self._lambda_func.layers
        self._vol_ref: Optional[_VolumeRefCount] = None

    @property
    def name(self) -> str:
        return self._vol_ref.volume.name  # type: ignore[union-attr]

    @property
    def hash(self) -> str:
        return "-".join(
            [
                layer["Arn"].split("layer:")[-1].replace(":", "_")
                for layer in self._layers
            ]
        )

    def __enter__(self) -> _DockerDataVolumeLayerContext:
        # See if volume is already known
        with self.__class__._lock:
            self._vol_ref = self.__class__._data_vol_map[self.hash]
            self._vol_ref.refcount += 1
            if self._vol_ref.refcount > 1:
                return self

            # See if the volume already exists
            for vol in self._lambda_func.docker_client.volumes.list():
                if vol.name == self.hash:
                    self._vol_ref.volume = vol
                    return self

            # It doesn't exist so we need to create it
            self._vol_ref.volume = self._lambda_func.docker_client.volumes.create(
                self.hash
            )
            # If we don't have any layers to apply, just return at this point
            # When invoking the function, we will bind this empty volume
            if len(self._layers) == 0:
                return self
            volumes = {self.name: {"bind": "/opt", "mode": "rw"}}

            self._lambda_func.ensure_image_exists("busybox")
            container = self._lambda_func.docker_client.containers.run(
                "busybox", "sleep 100", volumes=volumes, detach=True
            )
            backend: LambdaBackend = lambda_backends[self._lambda_func.account_id][
                self._lambda_func.region
            ]
            try:
                for layer in self._layers:
                    try:
                        layer_zip = backend.layers_versions_by_arn(  # type: ignore[union-attr]
                            layer["Arn"]
                        ).code_bytes
                        layer_tar = zip2tar(layer_zip)
                        container.put_archive("/opt", layer_tar)
                    except zipfile.BadZipfile as e:
                        warnings.warn(
                            f"Error extracting layer to Lambda: {e}", stacklevel=2
                        )
            finally:
                container.remove(force=True)

        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        with self.__class__._lock:
            self._vol_ref.refcount -= 1  # type: ignore[union-attr]
            if self._vol_ref.refcount == 0:  # type: ignore[union-attr]
                import docker.errors

                try:
                    self._vol_ref.volume.remove()  # type: ignore[union-attr]
                except docker.errors.APIError as e:
                    if e.status_code != 409:
                        raise

                    raise  # multiple processes trying to use same volume?


def _zipfile_content(zipfile_content: Union[str, bytes]) -> tuple[bytes, int, str, str]:
    try:
        to_unzip_code = base64.b64decode(bytes(zipfile_content, "utf-8"))  # type: ignore[arg-type]
    except Exception:
        to_unzip_code = base64.b64decode(zipfile_content)

    sha_code = hashlib.sha256(to_unzip_code)
    base64ed_sha = base64.b64encode(sha_code.digest()).decode("utf-8")
    sha_hex_digest = sha_code.hexdigest()
    return to_unzip_code, len(to_unzip_code), base64ed_sha, sha_hex_digest


def _s3_content(key: Any) -> tuple[bytes, int, str, str]:
    sha_code = hashlib.sha256(key.value)
    base64ed_sha = base64.b64encode(sha_code.digest()).decode("utf-8")
    sha_hex_digest = sha_code.hexdigest()
    return key.value, key.size, base64ed_sha, sha_hex_digest


def _validate_s3_bucket_and_key(
    account_id: str, partition: str, data: dict[str, Any]
) -> Optional[FakeKey]:
    key = None
    try:
        # FIXME: does not validate bucket region
        key = s3_backends[account_id][partition].get_object(
            data["S3Bucket"], data["S3Key"]
        )
    except MissingBucket:
        if do_validate_s3():
            raise InvalidParameterValueException(
                "Error occurred while GetObject. S3 Error Code: NoSuchBucket. S3 Error Message: The specified bucket does not exist"
            )
    except MissingKey:
        if do_validate_s3():
            raise ValueError(
                "InvalidParameterValueException",
                "Error occurred while GetObject. S3 Error Code: NoSuchKey. S3 Error Message: The specified key does not exist.",
            )
    return key


class EventInvokeConfig:
    def __init__(self, arn: str, last_modified: str, config: dict[str, Any]) -> None:
        self.config = config
        self.validate_max()
        self.validate()
        self.arn = arn
        self.last_modified = last_modified

    def validate_max(self) -> None:
        if "MaximumRetryAttempts" in self.config:
            mra = self.config["MaximumRetryAttempts"]
            if mra > 2:
                raise ValidationException(
                    mra,
                    "maximumRetryAttempts",
                    "Member must have value less than or equal to 2",
                )

            # < 0 validation done by botocore
        if "MaximumEventAgeInSeconds" in self.config:
            mra = self.config["MaximumEventAgeInSeconds"]
            if mra > 21600:
                raise ValidationException(
                    mra,
                    "maximumEventAgeInSeconds",
                    "Member must have value less than or equal to 21600",
                )

            # < 60 validation done by botocore

    def validate(self) -> None:
        # https://docs.aws.amazon.com/lambda/latest/dg/API_OnSuccess.html
        regex = r"^$|arn:(aws[a-zA-Z0-9-]*):([a-zA-Z0-9\-])+:([a-z]{2}(-gov)?-[a-z]+-\d{1})?:(\d{12})?:(.*)"
        pattern = re.compile(regex)

        if self.config["DestinationConfig"]:
            destination_config = self.config["DestinationConfig"]
            if (
                "OnSuccess" in destination_config
                and "Destination" in destination_config["OnSuccess"]
            ):
                contents = destination_config["OnSuccess"]["Destination"]
                if not pattern.match(contents):
                    raise ValidationException(
                        contents,
                        "destinationConfig.onSuccess.destination",
                        f"Member must satisfy regular expression pattern: {regex}",
                    )
            if (
                "OnFailure" in destination_config
                and "Destination" in destination_config["OnFailure"]
            ):
                contents = destination_config["OnFailure"]["Destination"]
                if not pattern.match(contents):
                    raise ValidationException(
                        contents,
                        "destinationConfig.onFailure.destination",
                        f"Member must satisfy regular expression pattern: {regex}",
                    )

    def response(self) -> dict[str, Any]:
        response = {"FunctionArn": self.arn, "LastModified": self.last_modified}
        response.update(self.config)
        return response


class ImageConfig:
    def __init__(self, config: dict[str, Any]) -> None:
        self.cmd = config.get("Command", [])
        self.entry_point = config.get("EntryPoint", [])
        self.working_directory = config.get("WorkingDirectory", None)

    def response(self) -> dict[str, Any]:
        content = {
            "Command": self.cmd,
            "EntryPoint": self.entry_point,
        }
        if self.working_directory is not None:
            content["WorkingDirectory"] = self.working_directory
        return dict(content)


class Permission(CloudFormationModel):
    def __init__(self, region: str):
        self.region = region

    @staticmethod
    def cloudformation_name_type() -> str:
        return "Permission"

    @staticmethod
    def cloudformation_type() -> str:
        return "AWS::Lambda::Permission"

    @classmethod
    def create_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
        **kwargs: Any,
    ) -> Permission:
        properties = cloudformation_json["Properties"]
        backend = lambda_backends[account_id][region_name]
        fn = backend.get_function(properties["FunctionName"])
        fn.policy.add_statement(raw=json.dumps(properties))
        return Permission(region=region_name)


class LayerVersion(CloudFormationModel):
    def __init__(self, spec: dict[str, Any], account_id: str, region: str):
        # required
        self.account_id = account_id
        self.region = region
        self.name = spec["LayerName"]
        self.content = spec["Content"]

        # optional
        self.description = spec.get("Description", "")
        self.compatible_architectures = spec.get("CompatibleArchitectures", [])
        self.compatible_runtimes = spec.get("CompatibleRuntimes", [])
        self.license_info = spec.get("LicenseInfo", "")
        self.policy = Policy(self)  # type: ignore[no-untyped-call]

        # For `list_layer_versions`, AWS sends back a hardcoded string over the wire, with timezone info automatically set to UTC (+0000)
        # This despite the fact that the layer exists in the US (us-east-1), and my local timezone was set to UTC+2
        # So it seems like the date is automatically set to UTC on creation, that's why we can just return this hardcoded
        # Also add microseconds to be compatible with AWS (milliseconds '%f' adds six, we only want three chars)
        self.created_date = utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+0000"
        self.version: int = 1
        self._attached = False
        self._layer: Optional[Layer] = None

        if "ZipFile" in self.content:
            (
                self.code_bytes,
                self.code_size,
                self.code_sha_256,
                self.code_digest,
            ) = _zipfile_content(self.content["ZipFile"])
        else:
            partition = get_partition(region)
            key = _validate_s3_bucket_and_key(
                account_id, partition=partition, data=self.content
            )
            if key:
                (
                    self.code_bytes,
                    self.code_size,
                    self.code_sha_256,
                    self.code_digest,
                ) = _s3_content(key)
            else:
                self.code_bytes = b""
                self.code_size = 0
                self.code_sha_256 = ""
                self.code_digest = ""

    @property
    def arn(self) -> str:
        if self.version:
            return make_layer_ver_arn(
                self.region, self.account_id, self.name, self.version
            )
        raise ValueError("Layer version is not set")

    def attach(self, layer: Layer, version: int) -> None:
        self._attached = True
        self._layer = layer
        self.version = version

    def get_layer_version(self) -> dict[str, Any]:
        version = {
            "Content": {
                "Location": "s3://",
                "CodeSha256": self.code_sha_256,
                "CodeSize": self.code_size,
            },
            "Version": self.version,
            "LayerArn": self._layer.layer_arn,  # type: ignore[union-attr]
            "LayerVersionArn": self.arn,
            "CreatedDate": self.created_date,
            "LicenseInfo": self.license_info,
        }
        if self.description:
            version["Description"] = self.description
        if self.compatible_architectures:
            version["CompatibleArchitectures"] = self.compatible_architectures
        if self.compatible_runtimes:
            version["CompatibleRuntimes"] = self.compatible_runtimes
        return version

    @staticmethod
    def cloudformation_name_type() -> str:
        return "LayerVersion"

    @staticmethod
    def cloudformation_type() -> str:
        return "AWS::Lambda::LayerVersion"

    @classmethod
    def create_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
        **kwargs: Any,
    ) -> LayerVersion:
        properties = cloudformation_json["Properties"]
        optional_properties = ("Description", "CompatibleRuntimes", "LicenseInfo")

        # required
        spec = {
            "Content": properties["Content"],
            "LayerName": resource_name,
        }
        for prop in optional_properties:
            if prop in properties:
                spec[prop] = properties[prop]

        backend = lambda_backends[account_id][region_name]
        layer_version = backend.publish_layer_version(spec)
        return layer_version


class LambdaAlias(BaseModel):
    def __init__(
        self,
        account_id: str,
        region: str,
        name: str,
        function_name: str,
        function_version: str,
        description: str,
        routing_config: str,
    ):
        self.arn = f"arn:{get_partition(region)}:lambda:{region}:{account_id}:function:{function_name}:{name}"
        self.name = name
        self.function_version = function_version
        self.description = description
        self.routing_config = routing_config
        self.revision_id = str(random.uuid4())

    def update(
        self,
        description: Optional[str],
        function_version: Optional[str],
        routing_config: Optional[str],
    ) -> None:
        if description is not None:
            self.description = description
        if function_version is not None:
            self.function_version = function_version
        if routing_config is not None:
            self.routing_config = routing_config

    def to_json(self) -> dict[str, Any]:
        return {
            "AliasArn": self.arn,
            "Description": self.description,
            "FunctionVersion": self.function_version,
            "Name": self.name,
            "RevisionId": self.revision_id,
            "RoutingConfig": self.routing_config or None,
        }


class Layer:
    def __init__(self, layer_version: LayerVersion):
        self.region = layer_version.region
        self.name = layer_version.name

        self.layer_arn = make_layer_arn(
            self.region, layer_version.account_id, self.name
        )
        self._latest_version = 0
        self.layer_versions: dict[str, LayerVersion] = {}

    def attach_version(self, layer_version: LayerVersion) -> None:
        self._latest_version += 1
        layer_version.attach(self, self._latest_version)
        self.layer_versions[str(self._latest_version)] = layer_version

    def delete_version(self, layer_version: str) -> None:
        self.layer_versions.pop(str(layer_version), None)

    def to_dict(self) -> dict[str, Any]:
        if not self.layer_versions:
            return {}

        last_key = sorted(self.layer_versions.keys(), key=lambda version: int(version))[
            -1
        ]
        return {
            "LayerName": self.name,
            "LayerArn": self.layer_arn,
            "LatestMatchingVersion": self.layer_versions[last_key].get_layer_version(),
        }


class LambdaFunction(CloudFormationModel, DockerModel):
    def __init__(
        self,
        account_id: str,
        spec: dict[str, Any],
        region: str,
        version: Union[str, int] = 1,
    ):
        DockerModel.__init__(self)
        # required
        self.account_id = account_id
        self.region = region
        self.partition = get_partition(region)
        self.code = spec["Code"]
        self.function_name = spec["FunctionName"]
        self.handler = spec.get("Handler")
        self.role = spec["Role"]
        self.run_time = spec.get("Runtime")
        self.logs_backend = logs_backends[account_id][self.region]
        self.environment_vars = spec.get("Environment", {}).get("Variables", {})
        self.policy = Policy(self)  # type: ignore[no-untyped-call]
        self.url_config: Optional[FunctionUrlConfig] = None
        self.state = "Active"
        self.reserved_concurrency = spec.get("ReservedConcurrentExecutions", None)

        # optional
        self.ephemeral_storage: str
        self.code_digest: str
        self.code_bytes: bytes
        self.event_invoke_config: list[EventInvokeConfig] = []

        self.description = spec.get("Description", "")
        self.memory_size = spec.get("MemorySize", 128)
        self.package_type = spec.get("PackageType", "Zip")
        self.publish = spec.get("Publish", False)  # this is ignored currently
        self.timeout = spec.get("Timeout", 3)
        self.layers: list[LayerDataType] = self._get_layers_data(spec.get("Layers", []))
        self.signing_profile_version_arn = spec.get("SigningProfileVersionArn")
        self.signing_job_arn = spec.get("SigningJobArn")
        self.code_signing_config_arn = spec.get("CodeSigningConfigArn")
        self.tracing_config = spec.get("TracingConfig") or {"Mode": "PassThrough"}
        self.architectures: list[str] = spec.get("Architectures", ["x86_64"])
        self.image_config: ImageConfig = ImageConfig(spec.get("ImageConfig", {}))
        _es = spec.get("EphemeralStorage")
        if _es:
            self.ephemeral_storage = _es["Size"]
        else:
            self.ephemeral_storage = 512

        self.logs_group_name = f"/aws/lambda/{self.function_name}"

        # VpcConfig has no default, if not supplied
        self._vpc_config = spec.get("VpcConfig")

        # auto-generated
        self.version = version
        self.last_modified = iso_8601_datetime_with_nanoseconds()

        self._set_function_code(self.code)
        self.code_or_config_updated = False

        self.function_arn: str = make_function_arn(
            self.region, self.account_id, self.function_name
        )

        self.tags = spec.get("Tags") or {}

    def __getstate__(self) -> dict[str, Any]:
        return {
            k: v
            for (k, v) in self.__dict__.items()
            if k != "_DockerModel__docker_client"
        }

    def set_version(self, version: int) -> None:
        self.function_arn = make_function_ver_arn(
            self.region, self.account_id, self.function_name, version
        )
        self.version = version
        self.last_modified = iso_8601_datetime_with_nanoseconds()

    @property
    def architectures(self) -> list[str]:
        return self._architectures

    @architectures.setter
    def architectures(self, architectures: list[str]) -> None:
        if (
            len(architectures) > 1
            or not architectures
            or architectures[0] not in ("x86_64", "arm64")
        ):
            raise ValidationException(
                str(architectures),
                "architectures",
                "Member must satisfy constraint: "
                "[Member must satisfy enum value set: [x86_64, arm64], Member must not be null]",
            )
        self._architectures = architectures

    @property
    def ephemeral_storage(self) -> int:
        return self._ephemeral_storage

    @ephemeral_storage.setter
    def ephemeral_storage(self, ephemeral_storage: int) -> None:
        if ephemeral_storage > 10240:
            raise ValidationException(
                str(ephemeral_storage),
                "ephemeralStorage.size",
                "Member must have value less than or equal to 10240",
            )

        # ephemeral_storage < 512 is handled by botocore 1.30.0
        self._ephemeral_storage = ephemeral_storage

    @property
    def vpc_config(self) -> Optional[dict[str, Any]]:  # type: ignore[misc]
        if not self._vpc_config:
            return None
        config = self._vpc_config.copy()
        if config["SecurityGroupIds"]:
            config.update({"VpcId": "vpc-123abc"})
        return config

    @property
    def physical_resource_id(self) -> str:
        return self.function_name

    def __repr__(self) -> str:
        return json.dumps(self.get_configuration())

    def _get_layers_data(self, layers_versions_arns: list[str]) -> list[LayerDataType]:
        backend = lambda_backends[self.account_id][self.region]
        layer_versions = [
            backend.layers_versions_by_arn(layer_version)
            for layer_version in layers_versions_arns
        ]
        if not all(layer_versions):
            raise UnknownLayerVersionException(layers_versions_arns)
        # The `if lv` part is not necessary - we know there are no None's, because of the `all()`-check earlier
        # But MyPy does not seem to understand this
        return [
            {"Arn": lv.arn, "CodeSize": lv.code_size} for lv in layer_versions if lv
        ]

    def get_function_code_signing_config(self) -> dict[str, Any]:
        return {
            "CodeSigningConfigArn": self.code_signing_config_arn,
            "FunctionName": self.function_name,
        }

    def get_configuration(self, on_create: bool = False) -> dict[str, Any]:
        config = {
            "CodeSha256": self.code_sha_256,
            "CodeSize": self.code_size,
            "Description": self.description,
            "FunctionArn": self.function_arn,
            "FunctionName": self.function_name,
            "Handler": self.handler,
            "LastModified": self.last_modified,
            "MemorySize": self.memory_size,
            "Role": self.role,
            "Runtime": self.run_time,
            "State": self.state,
            "PackageType": self.package_type,
            "Timeout": self.timeout,
            "Version": str(self.version),
            "VpcConfig": self.vpc_config,
            "Layers": self.layers,
            "SigningProfileVersionArn": self.signing_profile_version_arn,
            "SigningJobArn": self.signing_job_arn,
            "TracingConfig": self.tracing_config,
            "Architectures": self.architectures,
            "EphemeralStorage": {
                "Size": self.ephemeral_storage,
            },
            "SnapStart": {"ApplyOn": "None", "OptimizationStatus": "Off"},
        }
        if self.package_type == "Image":
            config["ImageConfigResponse"] = {
                "ImageConfig": self.image_config.response(),
            }
        if not on_create:
            # Only return this variable after the first creation
            config["LastUpdateStatus"] = "Successful"
        if self.environment_vars:
            config["Environment"] = {"Variables": self.environment_vars}

        return config

    def get_code(self) -> dict[str, Any]:
        resp = {"Configuration": self.get_configuration()}
        if "S3Key" in self.code:
            resp["Code"] = {
                "Location": f"https://awslambda-{self.region}-tasks.s3.{self.region}.amazonaws.com/{self.code['S3Key']}",
                "RepositoryType": "S3",
            }
        elif "ImageUri" in self.code:
            resp["Code"] = {
                "RepositoryType": "ECR",
                "ImageUri": self.code.get("ImageUri"),
                "ResolvedImageUri": self.code.get("ImageUri").split(":")[0]
                + "@sha256:"
                + self.code_sha_256,
            }
        if self.tags:
            resp["Tags"] = self.tags
        if self.reserved_concurrency:
            resp.update(
                {
                    "Concurrency": {
                        "ReservedConcurrentExecutions": self.reserved_concurrency
                    }
                }
            )
        return resp

    def update_configuration(self, config_updates: dict[str, Any]) -> dict[str, Any]:
        for key, value in config_updates.items():
            if key == "Description":
                self.description = value
            elif key == "Handler":
                self.handler = value
            elif key == "MemorySize":
                self.memory_size = value
            elif key == "Role":
                self.role = value
            elif key == "Runtime":
                self.run_time = value
            elif key == "Timeout":
                self.timeout = value
            elif key == "VpcConfig":
                self._vpc_config = value
            elif key == "Environment":
                self.environment_vars = value["Variables"]
            elif key == "Layers":
                self.layers = self._get_layers_data(value)

        self.code_or_config_updated = True

        return self.get_configuration()

    def _set_function_code(self, updated_spec: dict[str, Any]) -> None:
        from_update = updated_spec is not self.code

        # "DryRun" is only used for UpdateFunctionCode
        if from_update and "DryRun" in updated_spec and updated_spec["DryRun"]:
            return

        if "ZipFile" in updated_spec:
            if from_update:
                self.code["ZipFile"] = updated_spec["ZipFile"]

            (
                self.code_bytes,
                self.code_size,
                self.code_sha_256,
                self.code_digest,
            ) = _zipfile_content(updated_spec["ZipFile"])

            # TODO: we should be putting this in a lambda bucket
            self.code["UUID"] = str(random.uuid4())
            self.code["S3Key"] = f"{self.function_name}-{self.code['UUID']}"
        elif "S3Bucket" in updated_spec and "S3Key" in updated_spec:
            key = None
            try:
                if from_update:
                    # FIXME: does not validate bucket region
                    key = s3_backends[self.account_id][self.partition].get_object(
                        updated_spec["S3Bucket"], updated_spec["S3Key"]
                    )
                else:
                    key = _validate_s3_bucket_and_key(
                        self.account_id, partition=self.partition, data=self.code
                    )
            except MissingBucket:
                if do_validate_s3():
                    raise ValueError(
                        "InvalidParameterValueException",
                        "Error occurred while GetObject. S3 Error Code: NoSuchBucket. S3 Error Message: The specified bucket does not exist",
                    )
            except MissingKey:
                if do_validate_s3():
                    raise ValueError(
                        "InvalidParameterValueException",
                        "Error occurred while GetObject. S3 Error Code: NoSuchKey. S3 Error Message: The specified key does not exist.",
                    )
            if key:
                (
                    self.code_bytes,
                    self.code_size,
                    self.code_sha_256,
                    self.code_digest,
                ) = _s3_content(key)
            else:
                self.code_bytes = b""
                self.code_size = 0
                self.code_sha_256 = ""
            if from_update:
                self.code["S3Bucket"] = updated_spec["S3Bucket"]
                self.code["S3Key"] = updated_spec["S3Key"]
        elif "ImageUri" in updated_spec:
            if settings.lambda_stub_ecr():
                self.code_sha_256 = hashlib.sha256(
                    updated_spec["ImageUri"].encode("utf-8")
                ).hexdigest()
                self.code_size = 0
            else:
                if "@" in updated_spec["ImageUri"]:
                    # deploying via digest
                    uri, digest = updated_spec["ImageUri"].split("@")
                    image_id = {"imageDigest": digest}
                else:
                    # deploying via tag
                    uri, tag = updated_spec["ImageUri"].split(":")
                    image_id = {"imageTag": tag}

                repo_name = uri.split("/")[-1]
                ecr_backend = ecr_backends[self.account_id][self.region]
                registry_id = ecr_backend.describe_registry()["registryId"]
                images = ecr_backend.batch_get_image(
                    repository_name=repo_name, image_ids=[image_id]
                )["images"]

                if len(images) == 0:
                    exc = ImageNotFoundException(image_id, repo_name, registry_id)  # type: ignore
                    raise LambdaClientError(exc.code, exc.message)
                else:
                    manifest = json.loads(images[0].image_manifest)
                    self.code_sha_256 = (
                        images[0].image_id["imageDigest"].replace("sha256:", "")
                    )
                    self.code_size = manifest["config"]["size"]
            if from_update:
                self.code["ImageUri"] = updated_spec["ImageUri"]

    def update_function_code(self, updated_spec: dict[str, Any]) -> dict[str, Any]:
        self._set_function_code(updated_spec)
        self.code_or_config_updated = True
        return self.get_configuration()

    @staticmethod
    def convert(s: Any) -> str:  # type: ignore[misc]
        try:
            return str(s, encoding="utf-8")
        except Exception:
            return s

    def _invoke_lambda(self, event: Optional[str] = None) -> tuple[str, bool, str]:
        import docker
        import docker.errors

        # Create the LogGroup if necessary, to write the result to
        self.logs_backend.ensure_log_group(self.logs_group_name)
        # TODO: context not yet implemented
        if event is None:
            event = {}  # type: ignore[assignment]
        output = None

        try:
            # TODO: I believe we can keep the container running and feed events as needed
            #       also need to hook it up to the other services so it can make kws/s3 etc calls
            #  Should get invoke_id /RequestId from invocation
            env_vars = {
                "_HANDLER": self.handler,
                "AWS_EXECUTION_ENV": f"AWS_Lambda_{self.run_time}",
                "AWS_LAMBDA_FUNCTION_TIMEOUT": self.timeout,
                "AWS_LAMBDA_FUNCTION_NAME": self.function_name,
                "AWS_LAMBDA_FUNCTION_MEMORY_SIZE": self.memory_size,
                "AWS_LAMBDA_FUNCTION_VERSION": self.version,
                "AWS_REGION": self.region,
                "AWS_ACCESS_KEY_ID": "role-account-id",
                "AWS_SECRET_ACCESS_KEY": "role-secret-key",
                "AWS_SESSION_TOKEN": "session-token",
            }

            env_vars.update(self.environment_vars)
            env_vars["MOTO_HOST"] = settings.moto_server_host()
            moto_port = settings.moto_server_port()
            env_vars["MOTO_PORT"] = moto_port
            env_vars["MOTO_HTTP_ENDPOINT"] = f"{env_vars['MOTO_HOST']}:{moto_port}"

            if settings.is_test_proxy_mode():
                env_vars["HTTPS_PROXY"] = env_vars["MOTO_HTTP_ENDPOINT"]
                env_vars["AWS_CA_BUNDLE"] = "/var/task/ca.crt"

            container = exit_code = None
            log_config = docker.types.LogConfig(type=docker.types.LogConfig.types.JSON)

            with (
                _DockerDataVolumeContext(self) as data_vol,
                _DockerDataVolumeLayerContext(self) as layer_context,
            ):
                try:
                    run_kwargs: dict[str, Any] = {}
                    network_name = settings.moto_network_name()
                    network_mode = settings.moto_network_mode()
                    if network_name:
                        run_kwargs["network"] = network_name
                    elif network_mode:
                        run_kwargs["network_mode"] = network_mode
                    elif settings.TEST_SERVER_MODE:
                        # AWSLambda can make HTTP requests to a Docker container called 'motoserver'
                        # Only works if our Docker-container is named 'motoserver'
                        # TODO: should remove this and rely on 'network_mode' instead, as this is too tightly coupled with our own test setup
                        run_kwargs["links"] = {"motoserver": "motoserver"}

                    # add host.docker.internal host on linux to emulate Mac + Windows behavior
                    #   for communication with other mock AWS services running on localhost
                    if platform == "linux" or platform == "linux2":
                        run_kwargs["extra_hosts"] = {
                            "host.docker.internal": "host-gateway"
                        }

                    # Change entry point if requested
                    if self.image_config.entry_point:
                        run_kwargs["entrypoint"] = self.image_config.entry_point

                    # The requested image can be found in one of a few repos:
                    # - User-provided repo
                    # - mlupin/docker-lambda (the repo with up-to-date AWSLambda images
                    # - lambci/lambda (the repo with older/outdated AWSLambda images
                    #
                    # We'll cycle through all of them - when we find the repo that contains our image, we use it
                    image_repos = {  # dict maintains insertion order
                        settings.moto_lambda_image(): None,
                        "mlupin/docker-lambda": None,
                        "lambci/lambda": None,
                    }
                    for image_repo in image_repos:
                        image_ref = (
                            image_repo
                            if ":" in image_repo
                            else f"{image_repo}:{self.run_time}"
                        )
                        try:
                            self.ensure_image_exists(image_ref)
                            break
                        except docker.errors.NotFound:
                            pass
                    volumes = {
                        data_vol.name: {"bind": "/var/task", "mode": "rw"},
                        layer_context.name: {"bind": "/opt", "mode": "rw"},
                    }
                    container = self.docker_client.containers.run(
                        image_ref,
                        [self.handler, json.dumps(event)],
                        remove=False,
                        mem_limit=f"{self.memory_size}m",
                        volumes=volumes,
                        environment=env_vars,
                        detach=True,
                        log_config=log_config,
                        **run_kwargs,
                    )
                finally:
                    if container:
                        try:
                            exit_code = container.wait(timeout=300)["StatusCode"]
                        except requests.exceptions.ReadTimeout:
                            exit_code = -1
                            container.stop()
                            container.kill()

                        output = container.logs(stdout=False, stderr=True)
                        output += container.logs(stdout=True, stderr=False)
                        container.remove()

            output = output.decode("utf-8")  # type: ignore[union-attr]

            self.save_logs(output)

            # We only care about the response from the lambda
            # Which is the last line of the output, according to https://github.com/lambci/docker-lambda/issues/25
            resp = output.splitlines()[-1]
            logs = os.linesep.join(list(self.convert(output).splitlines()[:-1]))
            invocation_error = exit_code != 0
            return resp, invocation_error, logs
        except docker.errors.DockerException as e:
            # Docker itself is probably not running - there will be no Lambda-logs to handle
            msg = f"error running docker: {e}"
            logger.error(msg)
            self.save_logs(msg)
            return msg, True, ""

    def save_logs(self, output: str) -> None:
        # Send output to "logs" backend
        invoke_id = random.uuid4().hex
        date = utcnow()
        log_stream_name = (
            f"{date.year}/{date.month:02d}/{date.day:02d}/[{self.version}]{invoke_id}"
        )
        self.logs_backend.create_log_stream(self.logs_group_name, log_stream_name)
        log_events = [
            {"timestamp": unix_time_millis(), "message": line}
            for line in output.splitlines()
        ]
        self.logs_backend.put_log_events(
            self.logs_group_name, log_stream_name, log_events
        )

    def invoke(
        self, body: str, request_headers: Any, response_headers: Any
    ) -> Union[str, bytes]:
        if body:
            body = json.loads(body)
        else:
            body = "{}"

        # Get the invocation type:
        res, errored, logs = self._invoke_lambda(event=body)
        if errored:
            response_headers["x-amz-function-error"] = "Handled"

        inv_type = request_headers.get("x-amz-invocation-type", "RequestResponse")
        if inv_type == "RequestResponse":
            encoded = base64.b64encode(logs.encode("utf-8"))
            response_headers["x-amz-log-result"] = encoded.decode("utf-8")
            return res.encode("utf-8")
        else:
            return res

    @staticmethod
    def cloudformation_name_type() -> str:
        return "FunctionName"

    @staticmethod
    def cloudformation_type() -> str:
        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-function.html
        return "AWS::Lambda::Function"

    @classmethod
    def create_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
        **kwargs: Any,
    ) -> LambdaFunction:
        properties = cloudformation_json["Properties"]
        optional_properties = (
            "Description",
            "MemorySize",
            "Publish",
            "Timeout",
            "VpcConfig",
            "Environment",
            "ReservedConcurrentExecutions",
        )

        # required
        spec = {
            "Code": properties["Code"],
            "FunctionName": resource_name,
            "Handler": properties.get("Handler"),
            "Role": properties["Role"],
            "Runtime": properties.get("Runtime"),
        }

        # NOTE: Not doing `properties.get(k, DEFAULT)` to avoid duplicating the
        # default logic
        for prop in optional_properties:
            if prop in properties:
                spec[prop] = properties[prop]

        # when ZipFile is present in CloudFormation, per the official docs,
        # the code it's a plaintext code snippet up to 4096 bytes.
        # this snippet converts this plaintext code to a proper base64-encoded ZIP file.
        if "ZipFile" in properties["Code"]:
            spec["Code"]["ZipFile"] = base64.b64encode(
                cls._create_zipfile_from_plaintext_code(spec["Code"]["ZipFile"])
            )

        backend = lambda_backends[account_id][region_name]
        fn = backend.create_function(spec)
        return fn

    @classmethod
    def has_cfn_attr(cls, attr: str) -> bool:
        return attr in ["Arn"]

    def get_cfn_attribute(self, attribute_name: str) -> str:
        from moto.cloudformation.exceptions import UnformattedGetAttTemplateException

        if attribute_name == "Arn":
            return make_function_arn(self.region, self.account_id, self.function_name)
        raise UnformattedGetAttTemplateException()

    @classmethod
    def update_from_cloudformation_json(  # type: ignore[misc]
        cls,
        original_resource: LambdaFunction,
        new_resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
    ) -> LambdaFunction:
        updated_props = cloudformation_json["Properties"]
        original_resource.update_configuration(updated_props)
        original_resource.update_function_code(updated_props["Code"])
        return original_resource

    @staticmethod
    def _create_zipfile_from_plaintext_code(code: str) -> bytes:
        zip_output = io.BytesIO()
        zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
        zip_file.writestr("index.py", code)
        # This should really be part of the 'lambci' docker image
        from moto.packages.cfnresponse import cfnresponse

        with open(cfnresponse.__file__) as cfn:
            zip_file.writestr("cfnresponse.py", cfn.read())
        zip_file.close()
        zip_output.seek(0)
        return zip_output.read()

    def delete(self, account_id: str, region: str) -> None:
        lambda_backends[account_id][region].delete_function(self.function_name)

    def create_url_config(self, config: dict[str, Any]) -> FunctionUrlConfig:
        self.url_config = FunctionUrlConfig(function=self, config=config)
        return self.url_config

    def delete_url_config(self) -> None:
        self.url_config = None

    def get_url_config(self) -> FunctionUrlConfig:
        if not self.url_config:
            raise GenericResourcNotFound()
        return self.url_config

    def update_url_config(self, config: dict[str, Any]) -> FunctionUrlConfig:
        self.url_config.update(config)  # type: ignore[union-attr]
        return self.url_config  # type: ignore[return-value]


class FunctionUrlConfig:
    def __init__(self, function: LambdaFunction, config: dict[str, Any]):
        self.function = function
        self.config = config
        self.url = f"https://{random.uuid4().hex}.lambda-url.{function.region}.on.aws"
        self.created = utcnow().strftime("%Y-%m-%dT%H:%M:%S.000+0000")
        self.last_modified = self.created

    def to_dict(self) -> dict[str, Any]:
        return {
            "FunctionUrl": self.url,
            "FunctionArn": self.function.function_arn,
            "AuthType": self.config.get("AuthType"),
            "Cors": self.config.get("Cors"),
            "CreationTime": self.created,
            "LastModifiedTime": self.last_modified,
            "InvokeMode": self.config.get("InvokeMode") or "Buffered",
        }

    def update(self, new_config: dict[str, Any]) -> None:
        if new_config.get("Cors"):
            self.config["Cors"] = new_config["Cors"]
        if new_config.get("AuthType"):
            self.config["AuthType"] = new_config["AuthType"]
        self.last_modified = utcnow().strftime("%Y-%m-%dT%H:%M:%S")


class EventSourceMapping(CloudFormationModel):
    def __init__(self, spec: dict[str, Any]):
        # required
        self.region = spec["RegionName"]
        self.account_id = spec["AccountId"]
        self.uuid = str(random.uuid4())
        self.function_name = spec["FunctionName"]
        self.event_source_arn = spec["EventSourceArn"]
        self.arn: str = make_event_source_mapping_arn(
            self.region, self.account_id, self.uuid
        )

        # optional
        self.batch_size = spec.get("BatchSize")  # type: ignore[assignment]
        self.starting_position = spec.get("StartingPosition")
        if (
            not self.starting_position
            and "sqs" not in self._get_service_source_from_arn()
        ):
            self.starting_position = "TRIM_HORIZON"
        self.enabled = spec.get("Enabled", True)
        self.starting_position_timestamp = spec.get("StartingPositionTimestamp")
        self.kafka_config = spec.get("AmazonManagedKafkaEventSourceConfig")
        self.bisect_on_error = spec.get("BisectBatchOnFunctionError")
        self.destination_config = spec.get("DestinationConfig")
        self.document_db_config = spec.get("DocumentDBEventSourceConfig")
        self.filter_criteria = spec.get("FilterCriteria")
        self.function_response_types = spec.get("FunctionResponseTypes")
        self.kms_key = spec.get("KMSKeyArn")
        self.max_batch_window = spec.get("MaximumBatchingWindowInSeconds")
        self.max_record_age = spec.get("MaximumRecordAgeInSeconds")
        self.max_retry_attempts = spec.get("MaximumRetryAttempts")
        self.metrics = spec.get("MetricsConfig")
        self.parallelization_factor = spec.get("ParallelizationFactor")
        self.poller_config = spec.get("ProvisionedPollerConfig")
        self.queues = spec.get("Queues")
        self.scaling_config = spec.get("ScalingConfig")
        self.self_managed_event_source = spec.get("SelfManagedEventSource")
        self.self_managed_kafka_event_source_config = spec.get(
            "SelfManagedKafkaEventSourceConfig"
        )
        self.source_access_config = spec.get("SourceAccessConfigurations")
        self.tags = spec.get("Tags") or {}
        self.topics = spec.get("Topics")
        self.tumbling_window = spec.get("TumblingWindowInSeconds")

        self.function_arn: str = spec["FunctionArn"]
        self.last_modified = time.mktime(utcnow().timetuple())

    def _get_service_source_from_arn(
        self, event_source_arn: Optional[str] = None
    ) -> str:
        arn = event_source_arn or self.event_source_arn
        service = arn.split(":")[2].lower()
        if service == "sqs" and arn.endswith(".fifo"):
            return "sqs-fifo"
        return service

    @property
    def event_source_arn(self) -> str:
        return self._event_source_arn

    @event_source_arn.setter
    def event_source_arn(self, event_source_arn: str) -> None:
        service = self._get_service_source_from_arn(event_source_arn)
        if service not in ("dynamodb", "kinesis", "sqs", "sqs-fifo"):
            raise ValueError(
                "InvalidParameterValueException", "Unsupported event source type"
            )
        self._event_source_arn = event_source_arn

    @property
    def batch_size(self) -> int:
        return self._batch_size

    @batch_size.setter
    def batch_size(self, new_batch_size: Optional[int]) -> None:
        batch_size_service_map = {
            "kinesis": (100, 10000),
            "dynamodb": (100, 10000),
            "sqs": (10, 10000),
            "sqs-fifo": (10, 10),
        }

        source_type = self._get_service_source_from_arn()
        batch_size_for_source = batch_size_service_map[source_type]

        if new_batch_size is None:
            self._batch_size = batch_size_for_source[0]
        elif new_batch_size > batch_size_for_source[1]:
            error_message = f"BatchSize {new_batch_size} exceeds the max of {batch_size_for_source[1]}"
            raise ValueError("InvalidParameterValueException", error_message)
        else:
            self._batch_size = int(new_batch_size)

    def get_configuration(self) -> dict[str, Any]:
        response_dict = {
            "UUID": self.uuid,
            "BatchSize": self.batch_size,
            "EventSourceArn": self.event_source_arn,
            "EventSourceMappingArn": self.arn,
            "FunctionArn": self.function_arn,
            "LastModified": self.last_modified,
            "LastProcessingResult": None,
            "State": "Enabled" if self.enabled else "Disabled",
            "StateTransitionReason": "User initiated",
            "StartingPosition": self.starting_position,
            "AmazonManagedKafkaEventSourceConfig": self.kafka_config,
            "BisectBatchOnFunctionError": self.bisect_on_error,
            "DestinationConfig": self.destination_config,
            "DocumentDBEventSourceConfig": self.document_db_config,
            "FilterCriteria": self.filter_criteria,
            "FunctionResponseTypes": self.function_response_types,
            "KMSKeyArn": self.kms_key,
            "MaximumBatchingWindowInSeconds": self.max_batch_window,
            "MaximumRecordAgeInSeconds": self.max_record_age,
            "MaximumRetryAttempts": self.max_retry_attempts,
            "MetricsConfig": self.metrics,
            "ParallelizationFactor": self.parallelization_factor,
            "ProvisionedPollerConfig": self.poller_config,
            "Queues": self.queues,
            "ScalingConfig": self.scaling_config,
            "SelfManagedEventSource": self.self_managed_event_source,
            "SelfManagedKafkaEventSourceConfig": self.self_managed_kafka_event_source_config,
            "SourceAccessConfigurations": self.source_access_config,
            "Tags": self.tags,
            "Topics": self.topics,
            "TumblingWindowInSeconds": self.tumbling_window,
        }
        # Only return fields with a value - we don't want to return None for a boolean/int field
        return {k: v for k, v in response_dict.items() if v is not None}

    def delete(self, account_id: str, region_name: str) -> None:
        lambda_backend = lambda_backends[account_id][region_name]
        lambda_backend.delete_event_source_mapping(self.uuid)

    @staticmethod
    def cloudformation_type() -> str:
        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html
        return "AWS::Lambda::EventSourceMapping"

    @classmethod
    def create_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
        **kwargs: Any,
    ) -> EventSourceMapping:
        properties = cloudformation_json["Properties"]
        lambda_backend = lambda_backends[account_id][region_name]
        return lambda_backend.create_event_source_mapping(properties)

    @classmethod
    def update_from_cloudformation_json(  # type: ignore[misc]
        cls,
        original_resource: Any,
        new_resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
    ) -> EventSourceMapping:
        properties = cloudformation_json["Properties"]
        event_source_uuid = original_resource.uuid
        lambda_backend = lambda_backends[account_id][region_name]
        return lambda_backend.update_event_source_mapping(event_source_uuid, properties)  # type: ignore[return-value]

    @classmethod
    def delete_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
    ) -> None:
        properties = cloudformation_json["Properties"]
        lambda_backend = lambda_backends[account_id][region_name]
        esms = lambda_backend.list_event_source_mappings(
            event_source_arn=properties["EventSourceArn"],
            function_name=properties["FunctionName"],
        )

        for esm in esms:
            if esm.uuid == resource_name:
                esm.delete(account_id, region_name)

    @property
    def physical_resource_id(self) -> str:
        return self.uuid


class LambdaVersion(CloudFormationModel):
    def __init__(self, spec: dict[str, Any]):
        self.version = spec["Version"]

    def __repr__(self) -> str:
        return str(self.logical_resource_id)  # type: ignore[attr-defined]

    @staticmethod
    def cloudformation_type() -> str:
        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-version.html
        return "AWS::Lambda::Version"

    @classmethod
    def create_from_cloudformation_json(  # type: ignore[misc]
        cls,
        resource_name: str,
        cloudformation_json: dict[str, Any],
        account_id: str,
        region_name: str,
        **kwargs: Any,
    ) -> LambdaVersion:
        properties = cloudformation_json["Properties"]
        function_name = properties["FunctionName"]
        func = lambda_backends[account_id][region_name].publish_version(function_name)
        spec = {"Version": func.version}  # type: ignore[union-attr]
        return LambdaVersion(spec)


class LambdaStorage:
    def __init__(self, region_name: str, account_id: str):
        # Format 'func_name' {'versions': []}
        self._functions: dict[str, Any] = {}
        self._arns: weakref.WeakValueDictionary[str, LambdaFunction] = (
            weakref.WeakValueDictionary()
        )
        self.region_name = region_name
        self.account_id = account_id
        self.partition = get_partition(region_name)

        # function-arn -> alias -> LambdaAlias
        self._aliases: dict[str, dict[str, LambdaAlias]] = defaultdict(lambda: {})

    def _get_latest(self, name: str) -> LambdaFunction:
        return self._functions[name]["latest"]

    def _get_version(self, name: str, version: str) -> Optional[LambdaFunction]:
        for config in self._functions[name]["versions"]:
            if str(config.version) == version:
                return config

        return None

    def _get_function_aliases(self, function_name: str) -> dict[str, LambdaAlias]:
        fn = self.get_function_by_name_or_arn_with_qualifier(function_name)
        # Split ARN to retrieve an ARN without a qualifier present
        [arn, _, _] = self.split_function_arn(fn.function_arn)
        return self._aliases[arn]

    def delete_alias(self, name: str, function_name: str) -> None:
        aliases = self._get_function_aliases(function_name)
        aliases.pop(name, None)

    def get_alias(self, name: str, function_name: str) -> LambdaAlias:
        aliases = self._get_function_aliases(function_name)
        if name in aliases:
            return aliases[name]

        arn = f"arn:{get_partition(self.region_name)}:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
        raise UnknownAliasException(arn)

    def put_alias(
        self,
        name: str,
        function_name: str,
        function_version: str,
        description: str,
        routing_config: str,
    ) -> LambdaAlias:
        fn = self.get_function_by_name_or_arn_with_qualifier(
            function_name, function_version
        )
        aliases = self._get_function_aliases(function_name)
        if name in aliases:
            arn = f"arn:{get_partition(self.region_name)}:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
            raise ConflictException(f"Alias already exists: {arn}")

        alias = LambdaAlias(
            account_id=self.account_id,
            region=self.region_name,
            name=name,
            function_name=fn.function_name,
            function_version=function_version,
            description=description,
            routing_config=routing_config,
        )
        aliases[name] = alias
        return alias

    def update_alias(
        self,
        name: str,
        function_name: str,
        function_version: str,
        description: str,
        routing_config: str,
    ) -> LambdaAlias:
        alias = self.get_alias(name, function_name)

        # errors if new function version doesn't exist
        self.get_function_by_name_or_arn_with_qualifier(function_name, function_version)

        alias.update(description, function_version, routing_config)
        return alias

    def get_function_by_name_forbid_qualifier(self, name: str) -> LambdaFunction:
        """
        Get function by name forbidding a qualifier
        :raises: UnknownFunctionException if function not found
        :raises: InvalidParameterValue if qualifier is provided
        """

        if name.count(":") == 1:
            raise InvalidParameterValueException("Cannot provide qualifier")

        if name not in self._functions:
            raise self.construct_unknown_function_exception(name)

        return self._get_latest(name)

    def get_function_by_name_with_qualifier(
        self, name: str, qualifier: Optional[str] = None
    ) -> LambdaFunction:
        """
        Get function by name with an optional qualifier
        :raises: UnknownFunctionException if function not found
        """

        # Function name may contain an alias
        # <fn_name>:<alias_name>
        if ":" in name:
            # Prefer qualifier in name over qualifier arg
            [name, qualifier] = name.split(":")

        # Find without qualifier
        if qualifier is None:
            return self.get_function_by_name_forbid_qualifier(name)

        if name not in self._functions:
            raise self.construct_unknown_function_exception(name, qualifier)

        # Find by latest
        if qualifier.lower() == "$latest":
            return self._functions[name]["latest"]

        # Find by version
        found_version = self._get_version(name, qualifier)
        if found_version:
            return found_version

        # Find by alias
        aliases = self._get_function_aliases(name)
        if qualifier in aliases:
            alias_version = aliases[qualifier].function_version

            # Find by alias pointing to latest
            if alias_version.lower() == "$latest":
                return self._functions[name]["latest"]

            # Find by alias pointing to version
            found_alias = self._get_version(name, alias_version)
            if found_alias:
                return found_alias

        raise self.construct_unknown_function_exception(name, qualifier)

    def list_versions_by_function(self, name: str) -> Iterable[LambdaFunction]:
        if name not in self._functions:
            return []

        latest = copy.copy(self._functions[name]["latest"])
        latest.function_arn += ":$LATEST"
        return [latest] + self._functions[name]["versions"]

    def list_aliases(self, function_name: str) -> Iterable[LambdaAlias]:
        aliases = self._get_function_aliases(function_name)
        return sorted(aliases.values(), key=lambda alias: alias.name)

    def get_arn(self, arn: str) -> Optional[LambdaFunction]:
        [arn_without_qualifier, _, _] = self.split_function_arn(arn)
        return self._arns.get(arn_without_qualifier, None)

    def split_function_arn(self, arn: str) -> tuple[str, str, Optional[str]]:
        """
        Handy utility to parse an ARN into:
        - ARN without qualifier
        - Function name
        - Optional qualifier
        """
        qualifier = None
        # Function ARN may contain an alias
        # arn:aws:lambda:region:account_id:function:<fn_name>:<alias_name>
        if ":" in arn.split(":function:")[-1]:
            qualifier = arn.split(":")[-1]
            # arn = arn:aws:lambda:region:account_id:function:<fn_name>
            arn = ":".join(arn.split(":")[0:-1])
        name = arn.split(":")[-1]
        return arn, name, qualifier

    def get_function_by_name_or_arn_forbid_qualifier(
        self, name_or_arn: str
    ) -> LambdaFunction:
        """
        Get function by name or arn forbidding a qualifier
        :raises: UnknownFunctionException if function not found
        :raises: InvalidParameterValue if qualifier is provided
        """

        if re.match(ARN_PARTITION_REGEX, name_or_arn):
            [_, name, qualifier] = self.split_function_arn(name_or_arn)

            if qualifier is not None:
                raise InvalidParameterValueException("Cannot provide qualifier")

            return self.get_function_by_name_forbid_qualifier(name)
        else:
            # name_or_arn is not an arn
            return self.get_function_by_name_forbid_qualifier(name_or_arn)

    def get_function_by_name_or_arn_with_qualifier(
        self, name_or_arn: str, qualifier: Optional[str] = None
    ) -> LambdaFunction:
        """
        Get function by name or arn with an optional qualifier
        :raises: UnknownFunctionException if function not found
        """

        if re.match(ARN_PARTITION_REGEX, name_or_arn):
            [_, name, qualifier_in_arn] = self.split_function_arn(name_or_arn)
            return self.get_function_by_name_with_qualifier(
                name, qualifier_in_arn or qualifier
            )
        else:
            return self.get_function_by_name_with_qualifier(name_or_arn, qualifier)

    def construct_unknown_function_exception(
        self, name_or_arn: str, qualifier: Optional[str] = None
    ) -> UnknownFunctionException:
        if re.match(ARN_PARTITION_REGEX, name_or_arn):
            arn = name_or_arn
        else:
            # name_or_arn is a function name with optional qualifier <func_name>[:<qualifier>]
            arn = make_function_arn(self.region_name, self.account_id, name_or_arn)
            # Append explicit qualifier to arn only if the name doesn't already have it
            if qualifier and ":" not in name_or_arn:
                arn = f"{arn}:{qualifier}"
        return UnknownFunctionException(arn)

    def put_function(self, fn: LambdaFunction) -> None:
        valid_role = re.match(InvalidRoleFormat.pattern, fn.role)
        if valid_role:
            account = valid_role.group(2)
            if account != self.account_id:
                raise CrossAccountNotAllowed()
            try:
                iam_backend = iam_backends[self.account_id][self.partition]
                iam_backend.get_role_by_arn(fn.role)
            except IAMNotFoundException:
                raise InvalidParameterValueException(
                    "The role defined for the function cannot be assumed by Lambda."
                )
        else:
            raise InvalidRoleFormat(fn.role)
        if fn.function_name in self._functions:
            self._functions[fn.function_name]["latest"] = fn
        else:
            self._functions[fn.function_name] = {"latest": fn, "versions": []}
        self._arns[fn.function_arn] = fn

    def publish_version(
        self, name_or_arn: str, description: str = ""
    ) -> Optional[LambdaFunction]:
        function = self.get_function_by_name_or_arn_forbid_qualifier(name_or_arn)
        name = function.function_name
        if name not in self._functions:
            return None
        if not self._functions[name]["latest"]:
            return None

        all_versions = self._functions[name]["versions"]
        if all_versions:
            latest_published = all_versions[-1]
            if not function.code_or_config_updated:
                # Nothing has changed, don't publish
                return latest_published
            else:
                function.code_or_config_updated = False

        new_version = len(all_versions) + 1
        fn = copy.copy(self._functions[name]["latest"])
        fn.set_version(new_version)
        if description:
            fn.description = description

        self._functions[name]["versions"].append(fn)
        self._arns[fn.function_arn] = fn
        return fn

    def del_function(self, name_or_arn: str, qualifier: Optional[str] = None) -> None:
        # Qualifier may be explicitly passed or part of function name or ARN, extract it here
        if re.match(ARN_PARTITION_REGEX, name_or_arn):
            # Extract from ARN
            if ":" in name_or_arn.split(":function:")[-1]:
                qualifier = name_or_arn.split(":")[-1]
        else:
            # Extract from function name
            if ":" in name_or_arn:
                qualifier = name_or_arn.split(":")[1]

        function = self.get_function_by_name_or_arn_with_qualifier(
            name_or_arn, qualifier
        )
        name = function.function_name
        if not qualifier:
            # Something is still reffing this so delete all arns
            latest = self._functions[name]["latest"].function_arn
            del self._arns[latest]

            for fn in self._functions[name]["versions"]:
                del self._arns[fn.function_arn]

            del self._functions[name]

        else:
            if qualifier == "$LATEST":
                self._functions[name]["latest"] = None
            else:
                self._functions[name]["versions"].remove(function)

            # If theres no functions left
            if (
                not self._functions[name]["versions"]
                and not self._functions[name]["latest"]
            ):
                del self._functions[name]

        self._aliases[function.function_arn] = {}

    def all(self) -> Iterable[LambdaFunction]:
        result = []

        for function_group in list(self._functions.values()):
            latest = copy.deepcopy(function_group["latest"])
            latest.function_arn = f"{latest.function_arn}:$LATEST"
            result.append(latest)

            result.extend(function_group["versions"])

        return result

    def latest(self) -> Iterable[LambdaFunction]:
        """
        Return the list of functions with version @LATEST
        :return:
        """
        result = []
        for function_group in self._functions.values():
            if function_group["latest"] is not None:
                result.append(function_group["latest"])

        return result


class LayerStorage:
    def __init__(self) -> None:
        self._layers: dict[str, Layer] = {}
        self._arns: weakref.WeakValueDictionary[str, LambdaFunction] = (
            weakref.WeakValueDictionary()
        )

    def _find_layer_by_name_or_arn(self, name_or_arn: str) -> Layer:
        if name_or_arn in self._layers:
            return self._layers[name_or_arn]
        for layer in self._layers.values():
            if layer.layer_arn == name_or_arn:
                return layer
        raise UnknownLayerException()

    def put_layer_version(self, layer_version: LayerVersion) -> None:
        """
        :param layer_version: LayerVersion
        """
        if layer_version.name not in self._layers:
            self._layers[layer_version.name] = Layer(layer_version)
        self._layers[layer_version.name].attach_version(layer_version)

    def list_layers(self) -> Iterable[dict[str, Any]]:
        return [
            layer.to_dict() for layer in self._layers.values() if layer.layer_versions
        ]

    def delete_layer_version(self, layer_name: str, layer_version: str) -> None:
        layer = self._find_layer_by_name_or_arn(layer_name)
        layer.delete_version(layer_version)

    def get_layer_version(self, layer_name: str, layer_version: str) -> LayerVersion:
        layer = self._find_layer_by_name_or_arn(layer_name)
        for lv in layer.layer_versions.values():
            if lv.version == int(layer_version):
                return lv
        raise UnknownLayerException()

    def get_layer_versions(self, layer_name: str) -> list[LayerVersion]:
        if layer_name in self._layers:
            return list(iter(self._layers[layer_name].layer_versions.values()))
        return []

    def get_layer_version_by_arn(
        self, layer_version_arn: str
    ) -> Optional[LayerVersion]:
        split_arn = split_layer_arn(layer_version_arn)
        if split_arn.layer_name in self._layers:
            return self._layers[split_arn.layer_name].layer_versions.get(
                split_arn.version, None
            )
        return None


class LambdaBackend(BaseBackend):
    """
    Implementation of the AWS Lambda endpoint.
    Invoking functions is supported - they will run inside a Docker container, emulating the real AWS behaviour as closely as possible.

    .. warning:: When invoking a function using the decorators, the created Docker container cannot reach Moto (or it's in-memory state). Any AWS SDK-requests within your Lambda will try to connect to AWS instead.

    It is possible to connect from AWS Lambdas to other services, as long as you are running MotoProxy or the MotoServer in a Docker container.

    When running the MotoProxy, calls to other AWS services are automatically proxied.

    When running MotoServer, the Lambda has access to environment variables `MOTO_HOST` and `MOTO_PORT`, which can be used to build the url that MotoServer runs on:

    .. sourcecode:: python

        def lambda_handler(event, context):
            host = os.environ.get("MOTO_HOST")
            port = os.environ.get("MOTO_PORT")
            url = host + ":" + port
            ec2 = boto3.client('ec2', region_name='us-west-2', endpoint_url=url)

            # Or even simpler:
            full_url = os.environ.get("MOTO_HTTP_ENDPOINT")
            ec2 = boto3.client("ec2", region_name="eu-west-1", endpoint_url=full_url)

            ec2.do_whatever_inside_the_existing_moto_server()

    Moto will run on port 5000 by default. This can be overwritten by setting an environment variable when starting Moto:

    .. sourcecode:: bash

        # This env var will be propagated to the Docker container running the Lambda functions
        MOTO_PORT=5000 moto_server

    The Docker container uses the default network mode, `bridge`.
    The following environment variables are available for fine-grained control over the Docker connection options:

    .. sourcecode:: bash

        # Provide the name of a custom network to connect to
        MOTO_DOCKER_NETWORK_NAME=mycustomnetwork moto_server

        # Override the network mode
        # For example, network_mode=host would use the network of the host machine
        # Note that this option will be ignored if MOTO_DOCKER_NETWORK_NAME is also set
        MOTO_DOCKER_NETWORK_MODE=host moto_server

    _-_-_-_

    The Docker images used by Moto are taken from the following repositories:

    - `mlupin/docker-lambda` (for recent versions)
    - `lambci/lambda` (for older/outdated versions)

    Use the following environment variable to configure Moto to look for images in an additional repository:

    .. sourcecode:: bash

        MOTO_DOCKER_LAMBDA_IMAGE=mlupin/docker-lambda

    _-_-_-_

    Use the following environment variable if you want to configure the data directory used by the Docker containers:

    .. sourcecode:: bash

        MOTO_LAMBDA_DATA_DIR=/tmp/data

    _-_-_-_

    If you want to mock the Lambda-containers invocation without starting a Docker-container, use the simple decorator:

    .. sourcecode:: python

        @mock_aws(config={"lambda": {"use_docker": False}})

    """

    def __init__(self, region_name: str, account_id: str):
        super().__init__(region_name, account_id)
        self._lambdas = LambdaStorage(region_name=region_name, account_id=account_id)
        self._event_source_mappings: dict[str, EventSourceMapping] = {}
        self._layers = LayerStorage()

    def create_alias(
        self,
        name: str,
        function_name: str,
        function_version: str,
        description: str,
        routing_config: str,
    ) -> LambdaAlias:
        return self._lambdas.put_alias(
            name, function_name, function_version, description, routing_config
        )

    def delete_alias(self, name: str, function_name: str) -> None:
        return self._lambdas.delete_alias(name, function_name)

    def get_alias(self, name: str, function_name: str) -> LambdaAlias:
        return self._lambdas.get_alias(name, function_name)

    def update_alias(
        self,
        name: str,
        function_name: str,
        function_version: str,
        description: str,
        routing_config: str,
    ) -> LambdaAlias:
        """
        The RevisionId parameter is not yet implemented
        """
        return self._lambdas.update_alias(
            name, function_name, function_version, description, routing_config
        )

    def create_function(self, spec: dict[str, Any]) -> LambdaFunction:
        """
        The Code.ImageUri is not validated by default. Set environment variable MOTO_LAMBDA_STUB_ECR=false if you want to validate the image exists in our mocked ECR.
        """
        function_name = spec.get("FunctionName", None)
        if function_name is None:
            raise RESTError("InvalidParameterValueException", "Missing FunctionName")

        fn = LambdaFunction(
            account_id=self.account_id,
            spec=spec,
            region=self.region_name,
            version="$LATEST",
        )

        self._lambdas.put_function(fn)

        if spec.get("Publish"):
            ver = self.publish_version(function_name)
            fn = copy.deepcopy(
                fn
            )  # We don't want to change the actual version - just the return value
            fn.version = ver.version  # type: ignore[union-attr]
        return fn

    def create_function_url_config(
        self, name_or_arn: str, config: dict[str, Any]
    ) -> FunctionUrlConfig:
        """
        The Qualifier-parameter is not yet implemented.
        Function URLs are not yet mocked, so invoking them will fail
        """
        function = self._lambdas.get_function_by_name_or_arn_forbid_qualifier(
            name_or_arn
        )
        return function.create_url_config(config)

    def delete_function_url_config(self, name_or_arn: str) -> None:
        """
        The Qualifier-parameter is not yet implemented
        """
        function = self._lambdas.get_function_by_name_or_arn_forbid_qualifier(
            name_or_arn
        )
        function.delete_url_config()

    def get_function_url_config(self, name_or_arn: str) -> FunctionUrlConfig:
        """
        The Qualifier-parameter is not yet implemented
        """
        function = self._lambdas.get_function_by_name_or_arn_forbid_qualifier(
            name_or_arn
        )
        if not function:
            raise UnknownFunctionException(arn=name_or_arn)
        return function.get_url_config()

    def update_function_url_config(
        self, name_or_arn: str, config: dict[str, Any]
    ) -> FunctionUrlConfig:
        """
        The Qualifier-parameter is not yet implemented
        """
        function = self._lambdas.get_function_by_name_or_arn_forbid_qualifier(
            name_or_arn
        )
        return function.update_url_config(config)

    def create_event_source_mapping(self, spec: dict[str, Any]) -> EventSourceMapping:
        required = ["EventSourceArn", "FunctionName"]
        for param in required:
            if not spec.get(param):
                raise RESTError("InvalidParameterValueException", f"Missing {param}")
        spec["RegionName"] = self.region_name
        spec["AccountId"] = self.account_id

        # Validate function name
        func = self._lambdas.get_function_by_name_or_arn_with_qualifier(
            spec.get("FunctionName", "")
        )

        # Validate queue
        sqs_backend = sqs_backends[self.account_id][self.region_name]
        for queue in sqs_backend.queues.values():
            if queue.queue_arn == spec["EventSourceArn"]:
                if queue.lambda_event_source_mappings.get("func.function_arn"):
                    # TODO: Correct exception?
                    raise RESTError(
                        "ResourceConflictException", "The resource already exists."
                    )
                spec.update({"FunctionArn": func.function_arn})
                esm = EventSourceMapping(spec)
                self._event_source_mappings[esm.uuid] = esm

                # Set backend function on queue
                queue.lambda_event_source_mappings[esm.function_arn] = esm

                return esm

        ddbstream_backend = dynamodbstreams_backends[self.account_id][self.region_name]
        ddb_backend = dynamodb_backends[self.account_id][self.region_name]
        for ddbstream in ddbstream_backend.list_streams():
            if ddbstream["StreamArn"] == spec["EventSourceArn"]:
                spec.update({"FunctionArn": func.function_arn})
                esm = EventSourceMapping(spec)
                self._event_source_mappings[esm.uuid] = esm
                table_name = ddbstream["TableName"]
                table = ddb_backend.get_table(table_name)
                table.lambda_event_source_mappings[esm.function_arn] = esm
                return esm

        kinesis_backend: KinesisBackend = kinesis_backends[self.account_id][
            self.region_name
        ]
        for stream in kinesis_backend.streams.values():
            if stream.arn == spec["EventSourceArn"]:
                spec.update({"FunctionArn": func.function_arn})
                esm = EventSourceMapping(spec)
                self._event_source_mappings[esm.uuid] = esm
                stream.lambda_event_source_mappings[esm.event_source_arn] = esm
                return esm

        raise RESTError("ResourceNotFoundException", "Invalid EventSourceArn")

    def publish_layer_version(self, spec: dict[str, Any]) -> LayerVersion:
        required = ["LayerName", "Content"]
        for param in required:
            if not spec.get(param):
                raise InvalidParameterValueException(f"Missing {param}")
        layer_version = LayerVersion(
            spec, account_id=self.account_id, region=self.region_name
        )
        self._layers.put_layer_version(layer_version)
        return layer_version

    def list_layers(self) -> Iterable[dict[str, Any]]:
        return self._layers.list_layers()

    def delete_layer_version(self, layer_name: str, layer_version: str) -> None:
        return self._layers.delete_layer_version(layer_name, layer_version)

    def get_layer_version(self, layer_name: str, layer_version: str) -> LayerVersion:
        return self._layers.get_layer_version(layer_name, layer_version)

    def list_layer_versions(self, layer_name: str) -> Iterable[LayerVersion]:
        return self._layers.get_layer_versions(layer_name)

    def layers_versions_by_arn(self, layer_version_arn: str) -> Optional[LayerVersion]:
        return self._layers.get_layer_version_by_arn(layer_version_arn)

    def publish_version(
        self, function_name: str, description: str = ""
    ) -> Optional[LambdaFunction]:
        return self._lambdas.publish_version(function_name, description)

    def get_function(
        self, function_name_or_arn: str, qualifier: Optional[str] = None
    ) -> LambdaFunction:
        return self._lambdas.get_function_by_name_or_arn_with_qualifier(
            function_name_or_arn, qualifier
        )

    def list_versions_by_function(self, function_name: str) -> Iterable[LambdaFunction]:
        return self._lambdas.list_versions_by_function(function_name)

    def list_aliases(self, function_name: str) -> Iterable[LambdaAlias]:
        return self._lambdas.list_aliases(function_name)

    def get_event_source_mapping(self, uuid: str) -> Optional[EventSourceMapping]:
        return self._event_source_mappings.get(uuid)

    def delete_event_source_mapping(self, uuid: str) -> Optional[EventSourceMapping]:
        return self._event_source_mappings.pop(uuid, None)

    def update_event_source_mapping(
        self, uuid: str, spec: dict[str, Any]
    ) -> Optional[EventSourceMapping]:
        esm = self.get_event_source_mapping(uuid)
        if not esm:
            return None

        for key in spec.keys():
            if key == "FunctionName":
                func = self._lambdas.get_function_by_name_or_arn_with_qualifier(
                    spec[key]
                )
                esm.function_arn = func.function_arn
            elif key == "BatchSize":
                esm.batch_size = spec[key]
            elif key == "Enabled":
                esm.enabled = spec[key]
            elif key == "FilterCriteria":
                esm.filter_criteria = spec[key]
            elif key == "MaximumBatchingWindowInSeconds":
                esm.max_batch_window = spec[key]
            elif key == "ParallelizationFactor":
                esm.parallelization_factor = spec[key]
            elif key == "DestinationConfig":
                esm.destination_config = spec[key]
            elif key == "MaximumRecordAgeInSeconds":
                esm.max_record_age = spec[key]
            elif key == "BisectBatchOnFunctionError":
                esm.bisect_on_error = spec[key]
            elif key == "MaximumRetryAttempts":
                esm.max_retry_attempts = spec[key]
            elif key == "TumblingWindowInSeconds":
                esm.tumbling_window = spec[key]
            elif key == "SourceAccessConfigurations":
                esm.source_access_config = spec[key]
            elif key == "FunctionResponseTypes":
                esm.function_response_types = spec[key]
            elif key == "ScalingConfig":
                esm.scaling_config = spec[key]
            elif key == "DocumentDBEventSourceConfig":
                esm.document_db_config = spec[key]
            elif key == "KMSKeyArn":
                esm.kms_key = spec[key]
            elif key == "MetricsConfig":
                esm.metrics = spec[key]
            elif key == "ProvisionedPollerConfig":
                esm.poller_config = spec[key]

        esm.last_modified = time.mktime(utcnow().timetuple())
        return esm

    def list_event_source_mappings(
        self, event_source_arn: str, function_name: str
    ) -> Iterable[EventSourceMapping]:
        esms = list(self._event_source_mappings.values())
        if event_source_arn:
            esms = list(filter(lambda x: x.event_source_arn == event_source_arn, esms))
        if function_name:
            esms = list(filter(lambda x: x.function_name == function_name, esms))
        return esms

    def get_function_by_arn(self, function_arn: str) -> Optional[LambdaFunction]:
        return self._lambdas.get_arn(function_arn)

    def delete_function(
        self, function_name: str, qualifier: Optional[str] = None
    ) -> None:
        self._lambdas.del_function(function_name, qualifier)

    def list_functions(
        self, func_version: Optional[str] = None
    ) -> Iterable[LambdaFunction]:
        if func_version == "ALL":
            return self._lambdas.all()
        return self._lambdas.latest()

    def send_sqs_batch(self, function_arn: str, messages: Any, queue_arn: str) -> bool:
        success = True
        for message in messages:
            result = self._send_sqs_message(function_arn, message, queue_arn)
            if not result:
                success = False
        return success

    def _send_sqs_message(
        self, function_arn: str, message: Any, queue_arn: str
    ) -> bool:
        event = {
            "Records": [
                {
                    "messageId": message.id,
                    "receiptHandle": message.receipt_handle,
                    "body": message.body,
                    "attributes": {
                        "ApproximateReceiveCount": "1",
                        "SentTimestamp": "1545082649183",
                        "SenderId": "AIDAIENQZJOLO23YVJ4VO",
                        "ApproximateFirstReceiveTimestamp": "1545082649185",
                    },
                    "messageAttributes": {},
                    "md5OfBody": "098f6bcd4621d373cade4e832627b4f6",
                    "eventSource": "aws:sqs",
                    "eventSourceARN": queue_arn,
                    "awsRegion": self.region_name,
                }
            ]
        }
        if queue_arn.endswith(".fifo"):
            # Messages from FIFO queue have additional attributes
            event["Records"][0]["attributes"].update(
                {
                    "MessageGroupId": message.group_id,
                    "MessageDeduplicationId": message.deduplication_id,
                }
            )

        request_headers: dict[str, Any] = {}
        response_headers: dict[str, Any] = {}
        self.invoke(
            function_name=function_arn,
            qualifier=None,
            body=json.dumps(event),
            headers=request_headers,
            response_headers=response_headers,
        )
        return "x-amz-function-error" not in response_headers

    def send_kinesis_message(
        self,
        function_name: str,
        kinesis_stream: str,
        kinesis_partition_key: str,
        kinesis_sequence_number: str,
        kinesis_data: str,
        kinesis_shard_id: str,
    ) -> None:
        func = self._lambdas.get_function_by_name_or_arn_with_qualifier(
            function_name, qualifier=None
        )
        event = {
            "Records": [
                {
                    "kinesis": {
                        "kinesisSchemaVersion": "1.0",
                        "partitionKey": kinesis_partition_key,
                        "sequenceNumber": kinesis_sequence_number,
                        "data": kinesis_data,
                        "approximateArrivalTimestamp": round(time.time(), 3),
                    },
                    "eventSource": "aws:kinesis",
                    "eventVersion": "1.0",
                    "eventID": f"{kinesis_shard_id}:{kinesis_sequence_number}",
                    "eventName": "aws:kinesis:record",
                    "invokeIdentityArn": func.role,
                    "awsRegion": self.region_name,
                    "eventSourceARN": kinesis_stream,
                }
            ]
        }
        func.invoke(json.dumps(event), {}, {})

    def send_sns_message(
        self,
        function_name: str,
        message: str,
        subject: Optional[str] = None,
        qualifier: Optional[str] = None,
    ) -> None:
        event = {
            "Records": [
                {
                    "EventVersion": "1.0",
                    "EventSubscriptionArn": "arn:aws:sns:EXAMPLE",
                    "EventSource": "aws:sns",
                    "Sns": {
                        "SignatureVersion": "1",
                        "Timestamp": "1970-01-01T00:00:00.000Z",
                        "Signature": "EXAMPLE",
                        "SigningCertUrl": "EXAMPLE",
                        "MessageId": "95df01b4-ee98-5cb9-9903-4c221d41eb5e",
                        "Message": message,
                        "MessageAttributes": {
                            "Test": {"Type": "String", "Value": "TestString"},
                            "TestBinary": {"Type": "Binary", "Value": "TestBinary"},
                        },
                        "Type": "Notification",
                        "UnsubscribeUrl": "EXAMPLE",
                        "TopicArn": "arn:aws:sns:EXAMPLE",
                        "Subject": subject or "TestInvoke",
                    },
                }
            ]
        }
        func = self._lambdas.get_function_by_name_or_arn_with_qualifier(
            function_name, qualifier
        )
        func.invoke(json.dumps(event), {}, {})

    def send_dynamodb_items(
        self, function_arn: str, items: list[Any], source: str
    ) -> Union[str, bytes]:
        event = {
            "Records": [
                {
                    "eventID": item.to_json()["eventID"],
                    "eventName": item.to_json()["eventName"],
                    "eventVersion": item.to_json()["eventVersion"],
                    "eventSource": item.to_json()["eventSource"],
                    "awsRegion": self.region_name,
                    "dynamodb": item.to_json()["dynamodb"],
                    "eventSourceARN": source,
                }
                for item in items
            ]
        }
        func = self._lambdas.get_arn(function_arn)
        return func.invoke(json.dumps(event), {}, {})  # type: ignore[union-attr]

    def send_log_event(
        self,
        function_arn: str,
        filter_name: str,
        log_group_name: str,
        log_stream_name: str,
        log_events: Any,
    ) -> None:
        data = {
            "messageType": "DATA_MESSAGE",
            "owner": self.account_id,
            "logGroup": log_group_name,
            "logStream": log_stream_name,
            "subscriptionFilters": [filter_name],
            "logEvents": log_events,
        }

        output = io.BytesIO()
        with GzipFile(fileobj=output, mode="w") as f:
            f.write(json.dumps(data, separators=(",", ":")).encode("utf-8"))
        payload_gz_encoded = base64.b64encode(output.getvalue()).decode("utf-8")

        event = {"awslogs": {"data": payload_gz_encoded}}

        func = self._lambdas.get_arn(function_arn)
        func.invoke(json.dumps(event), {}, {})  # type: ignore[union-attr]

    def _get_resource_by_arn(self, arn: str) -> EventSourceMapping | LambdaFunction:
        arn_breakdown = arn.split(":")
        resource_type = arn_breakdown[len(arn_breakdown) - 2]
        resource_name = arn_breakdown[len(arn_breakdown) - 1]
        if resource_type == "event-source-mapping":
            esm = self._event_source_mappings.get(resource_name)
            if not esm:
                raise RESTError(
                    "ResourceNotFoundException",
                    f"Event source mapping {resource_name} not found.",
                )
            return esm
        return self._lambdas.get_function_by_name_or_arn_with_qualifier(arn)

    def list_tags(self, resource_arn: str) -> dict[str, str]:
        resource = self._get_resource_by_arn(resource_arn)
        return resource.tags

    def tag_resource(self, resource_arn: str, tags: dict[str, str]) -> None:
        resource = self._get_resource_by_arn(resource_arn)
        resource.tags.update(tags)

    def untag_resource(self, resource_arn: str, tagKeys: list[str]) -> None:
        resource = self._get_resource_by_arn(resource_arn)
        for key in tagKeys:
            resource.tags.pop(key, None)

    def add_permission(
        self, function_name: str, qualifier: str, raw: str
    ) -> dict[str, Any]:
        fn = self.get_function(function_name, qualifier)
        statement, revision = fn.policy.add_statement(raw, qualifier)
        return statement

    def remove_permission(
        self, function_name: str, sid: str, revision: str = ""
    ) -> None:
        fn = self.get_function(function_name)
        fn.policy.del_statement(sid, revision)

    def get_function_code_signing_config(self, function_name: str) -> dict[str, Any]:
        fn = self.get_function(function_name)
        return fn.get_function_code_signing_config()

    def get_policy(self, function_name: str, qualifier: Optional[str] = None) -> str:
        fn = self._lambdas.get_function_by_name_or_arn_with_qualifier(
            function_name, qualifier
        )
        return fn.policy.wire_format()

    def update_function_code(
        self, function_name: str, qualifier: str, body: dict[str, Any]
    ) -> Optional[dict[str, Any]]:
        fn: LambdaFunction = self.get_function(function_name, qualifier)
        fn.update_function_code(body)

        if body.get("Publish", False):
            fn = self.publish_version(function_name)  # type: ignore[assignment]

        return fn.update_function_code(body)

    def update_function_configuration(
        self, function_name: str, qualifier: str, body: dict[str, Any]
    ) -> Optional[dict[str, Any]]:
        fn = self.get_function(function_name, qualifier)

        return fn.update_configuration(body)

    def invoke(
        self,
        function_name: str,
        qualifier: Optional[str],
        body: Any,
        headers: Any,
        response_headers: Any,
    ) -> Optional[Union[str, bytes]]:
        """
        Invoking a Function with PackageType=Image is not yet supported.

        Invoking a Function against Lambda without docker now supports customised responses, the default being `Simple Lambda happy path OK`.
        You can use a dedicated API to override this, by configuring a queue of expected results.

        A request to `invoke` will take the first result from that queue.

        Configure this queue by making an HTTP request to `/moto-api/static/lambda-simple/response`. An example invocation looks like this:

        .. sourcecode:: python

            expected_results = {"results": ["test", "test 2"], "region": "us-east-1"}
            resp = requests.post(
                "http://motoapi.amazonaws.com/moto-api/static/lambda-simple/response",
                json=expected_results
            )
            assert resp.status_code == 201

            client = boto3.client("lambda", region_name="us-east-1")
            resp = client.invoke(...) # resp["Payload"].read().decode() == "test"
            resp = client.invoke(...) # resp["Payload"].read().decode() == "test2"
        """
        fn = self.get_function(function_name, qualifier)
        payload = fn.invoke(body, headers, response_headers)
        response_headers["Content-Length"] = str(len(payload))
        return payload

    def put_function_concurrency(
        self, function_name: str, reserved_concurrency: str
    ) -> str:
        """Establish concurrency limit/reservations for a function

        Actual lambda restricts concurrency to 1000 (default) per region/account
        across all functions; we approximate that behavior by summing across all
        functions (hopefully all in the same account and region) and allowing the
        caller to simulate an increased quota.

        By default, no quota is enforced in order to preserve compatibility with
        existing code that assumes it can do as many things as it likes. To model
        actual AWS behavior, define the MOTO_LAMBDA_CONCURRENCY_QUOTA environment
        variable prior to testing.
        """

        quota: Optional[str] = os.environ.get("MOTO_LAMBDA_CONCURRENCY_QUOTA")
        if quota is not None:
            # Enforce concurrency limits as described above
            available = int(quota) - int(reserved_concurrency)
            for fnx in self.list_functions():
                if fnx.reserved_concurrency and fnx.function_name != function_name:
                    available -= int(fnx.reserved_concurrency)
            if available < 100:
                raise InvalidParameterValueException(
                    "Specified ReservedConcurrentExecutions for function decreases account's UnreservedConcurrentExecution below its minimum value of [100]."
                )

        fn = self.get_function(function_name)
        fn.reserved_concurrency = reserved_concurrency
        return fn.reserved_concurrency

    def delete_function_concurrency(self, function_name: str) -> Optional[str]:
        fn = self.get_function(function_name)
        fn.reserved_concurrency = None
        return fn.reserved_concurrency

    def get_function_concurrency(self, function_name: str) -> Optional[str]:
        fn = self.get_function(function_name)
        return fn.reserved_concurrency

    def put_function_event_invoke_config(
        self, function_name: str, config: dict[str, Any]
    ) -> dict[str, Any]:
        fn = self.get_function(function_name)
        event_config = EventInvokeConfig(fn.function_arn, fn.last_modified, config)
        fn.event_invoke_config.append(event_config)
        return event_config.response()

    def update_function_event_invoke_config(
        self, function_name: str, config: dict[str, Any]
    ) -> dict[str, Any]:
        # partial functionality, the update function just does a put
        # instead of partial update
        return self.put_function_event_invoke_config(function_name, config)

    def get_function_event_invoke_config(self, function_name: str) -> dict[str, Any]:
        fn = self.get_function(function_name)
        if fn.event_invoke_config:
            response = fn.event_invoke_config[0]
            return response.response()
        else:
            raise UnknownEventConfig(fn.function_arn)

    def delete_function_event_invoke_config(self, function_name: str) -> None:
        if self.get_function_event_invoke_config(function_name):
            fn = self.get_function(function_name)
            fn.event_invoke_config = []

    def list_function_event_invoke_configs(self, function_name: str) -> dict[str, Any]:
        response: dict[str, list[dict[str, Any]]] = {"FunctionEventInvokeConfigs": []}
        try:
            response["FunctionEventInvokeConfigs"] = [
                self.get_function_event_invoke_config(function_name)
            ]
            return response
        except UnknownEventConfig:
            return response

    def add_layer_version_permission(
        self, layer_name: str, version_number: int, statement: str
    ) -> tuple[str, str]:
        layer_version = self.get_layer_version(layer_name, str(version_number))
        return layer_version.policy.add_statement(statement)

    def get_layer_version_policy(self, layer_name: str, version_number: int) -> str:
        layer_version = self.get_layer_version(layer_name, str(version_number))
        return layer_version.policy.wire_format()

    def remove_layer_version_permission(
        self, layer_name: str, version_number: str, sid: str, revision: str = ""
    ) -> None:
        layer_version = self.get_layer_version(layer_name, str(version_number))
        layer_version.policy.del_statement(sid, revision)


def do_validate_s3() -> bool:
    return os.environ.get("VALIDATE_LAMBDA_S3", "") in ["", "1", "true"]


lambda_backends = BackendDict(LambdaBackend, "lambda")