import functools
import traceback
import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401

""" IMPORTS """
# Std imports
from datetime import datetime, timezone
from base64 import b64decode

# 3-rd party imports
from typing import Any
from collections.abc import Iterator, Sequence
import urllib.parse
import urllib3
from akamai.edgegrid import EdgeGridAuth

# Local imports
from CommonServerUserPython import *
import asyncio
import aiohttp


"""GLOBALS/PARAMS

Attributes:
    INTEGRATION_NAME:
        Name of the integration as shown in the integration UI, for example: Microsoft Graph User.

    INTEGRATION_COMMAND_NAME:
        Command names should be written in all lower-case letters,
        and each word separated with a hyphen, for example: msgraph-user.

    INTEGRATION_CONTEXT_NAME:
        Context output names should be written in camel case, for example: MSGraphUser.
"""
INTEGRATION_NAME = "Akamai SIEM"
INTEGRATION_COMMAND_NAME = "akamai-siem"
INTEGRATION_CONTEXT_NAME = "Akamai"


VENDOR = "Akamai"
PRODUCT = "WAF"
DEFAULT_PAGE_SIZE = 20000  # Default events per request
TIME_TO_RUN_BUFFER = 30  # When calculating time left to run, will use this as a safe zone delta.
EXECUTION_START_TIME = datetime.now()
ALLOWED_PAGE_SIZE_DELTA_RATIO = 0.95  # uses this delta to overcome differences from Akamai When calculating latest request size.
MAX_ALLOWED_FETCH_LIMIT = 80000
MAX_INCIDENTS_FETCH_LIMIT = 2000  # Max total incidents per fetch (XSOAR)
DEFAULT_INCIDENTS_FETCH_LIMIT = 20  # Default total incidents per fetch
DEFAULT_EVENTS_FETCH_LIMIT = 60000  # Default total events per fetch
SEND_EVENTS_TO_XSIAM_CHUNK_SIZE = 9 * (10**6)  # 9 MB
AKAMAI_MAX_LOOKBACK_MINUTES = 715  # 11h55m: max recovery window (12h) minus a 5-minute safety buffer.

# Disable insecure warnings
urllib3.disable_warnings()


class Client(BaseClient):
    def get_events(
        self,
        config_ids: str,
        offset: str | None = "",
        limit: str | int | None = None,
        from_epoch: str | None = "",
        to_epoch: str | None = "",
    ) -> tuple[list[Any], Any]:
        """
            Get security events from Akamai WAF service by - https://developer.akamai.com/api/cloud_security/siem/v1.html,
            Pay attention response as text of multiple json objects
            Allowed query parameters combinations:
                1. offset - Since a prior request.
                2. offset, limit - Since a prior request, limited.
                3. from - Since a point in time.
                4. from, limit - Since a point in time, limited.
                5. from, to - Over a range of time.
                6. from, to, limit - Over a range of time, limited.
        Args:
            config_ids: Unique identifier for each security configuration. To report on more than one configuration, separate
                      integer identifiers with semicolons, e.g. 12892;29182;82912.
            offset: This token denotes the last message. If specified, this operation fetches only security events that have
                    occurred from offset. This is a required parameter for offset mode and you can't use it in time-based
                    requests.
            limit: Defines the approximate maximum number of security events each fetch returns, in both offset and
                   time-based modes. The default limit is 20000. Expect requests to return a slightly higher number of
                   security events than you set in the limit parameter, because data is stored in different buckets.
            from_epoch: The start of a specified time range, expressed in Unix epoch seconds.
                        This is a required parameter to get time-based results for a set period, and you can't use it in
                        offset mode.
            to_epoch: The end of a specified time range, expressed in Unix epoch seconds. You can't use this parameter in
                      offset mode and it's an optional parameter in time-based mode. If omitted, the value defaults to the
                      current time.

        Returns:
            Multiple json objects as list of dictionaries, offset for next pagination
        """
        params = {
            "offset": offset,
            "limit": limit,
            "to": to_epoch,
            "from": from_epoch,
        }
        raw_response: str = self._http_request(
            method="GET", url_suffix=f"/{config_ids}", params=assign_params(**params), resp_type="text"
        )
        events: list = []
        if '{ "total": 0' not in raw_response:
            events = [json.loads(event) for event in raw_response.split("\n")[:-2]]
            new_offset = str(max([int(event.get("httpMessage", {}).get("start")) for event in events]))
        else:
            new_offset = str(from_epoch)
        return events, new_offset

    def execute_get_events_request(self, params: dict[str, int | str], config_ids: str, prefix_msg: str = ""):
        demisto.debug(f"[Get Events] {prefix_msg}Init session and sending request to Akamai.")
        url_suffix = f"/{config_ids}"
        if "offset" in params:
            url_suffix = f"{url_suffix}?offset={params['offset']}"
            del params["offset"]
        raw_response: str = self._http_request(
            method="GET",
            url_suffix=url_suffix,
            params=params,
            resp_type="text",
        )
        demisto.debug(f"[Get Events] {prefix_msg}Finished executing request to Akamai, processing response.")
        return raw_response

    def prepare_params(self, limit, offset, from_epoch, prefix_msg: str = "") -> dict[str, int | str]:
        params: dict[str, int | str] = {"limit": limit}
        if offset:
            demisto.debug(f"[Get Events] {prefix_msg}Received {offset=}, running an offset based request.")
            params["offset"] = offset
        else:
            from_param = int(from_epoch)
            params["from"] = from_param
            demisto.debug(f"[Get Events] {prefix_msg} No offset received, running a time based request with {from_param=}.")
        return params

    def get_events_with_offset(
        self,
        config_ids: str,
        offset: str | None = "",
        limit: int = 20,
        from_epoch: str = "",
    ) -> tuple[list[str], str | None]:
        params = self.prepare_params(offset=offset, limit=limit, from_epoch=from_epoch)
        raw_response = self.execute_get_events_request(params, config_ids)
        events: list[str] = raw_response.split("\n")
        del raw_response  # free the full-page response string immediately; we only need the split lines
        demisto.debug(f"[Get Events] Split response into {len(events)} lines (events + offset context).")
        offset = None
        try:
            if events and events[-1] == "":
                events.pop()
            offset_context = events.pop()
            loaded_offset_context = json.loads(offset_context)
            offset = loaded_offset_context.get("offset")
        except Exception as e:
            demisto.error(f"[Get Events] Couldn't decode offset with {offset_context=}, reason {e}")
        return events, offset

    async def get_events_concurrently(
        self, config_ids: str, offset: str | None = "", limit: int = 200000, from_epoch: str = "", counter: int = 0
    ) -> tuple[list[str], str | None]:
        """Send request to get events from Akamai.

        Args:
            config_ids (str): security configuration ids to fetch, e.g. `51000;56080`.
            offset (str | None): The offset (hash) to use for offset based mechanism.
            limit (int, optional): The number of events to limit for every request.
            from_epoch (str): From when to fetch if first time.
            counter (int, optional): The execution number.

        Returns:
            tuple[list[str], str | None]: The events and offset obtained from last request.
        """
        params = self.prepare_params(
            offset=offset, limit=limit, from_epoch=from_epoch, prefix_msg=f"Running in interval = {counter}. "
        )
        loop = asyncio.get_event_loop()
        raw_response = await loop.run_in_executor(
            None,
            functools.partial(
                self.execute_get_events_request,
                config_ids=config_ids,
                params=params,
                prefix_msg=f"Running in interval = {counter}. ",
            ),
        )
        events: list[str] = raw_response.split("\n")
        new_offset = None
        try:
            if events and events[-1] == "":
                events.pop()
            offset_context = events.pop()
            loaded_offset_context = json.loads(offset_context)
            new_offset = loaded_offset_context.get("offset")
        except Exception as e:
            demisto.error(f"Running in interval = {counter}. Couldn't decode offset with {offset_context=}, reason {e}")
            new_offset = offset
        return events, new_offset


"""HELPER FUNCIONS"""


def date_format_converter(from_format: str, date_before: str, readable_format: str = "%Y-%m-%dT%H:%M:%SZ%Z") -> str:
    """
        Convert datatime object from epoch time to follow format %Y-%m-%dT%H:%M:%SZ
    Args:
        from_format: format to convert from.
        date_before: date before conversion epoch time or %Y-%m-%dT%H:%M:%SZ format
        readable_format: readable format by default %Y-%m-%dT%H:%M:%SZ
    Examples:
        >>> date_format_converter(from_format='epoch', date_before='1576570098')
        '2019-12-17T08:08:18Z'
        >>> date_format_converter(from_format='epoch', date_before='1576570098', readable_format='%Y-%m-%d %H:%M:%S')
        '2019-12-17 08:08:18'
        >>> date_format_converter(from_format='readable', date_before='2019-12-17T08:08:18Z')
        '1576570098'

    Returns:
        Converted date as Datetime object or string object
    """
    converted_date: str | int = ""
    if from_format == "epoch":
        converted_date = datetime.utcfromtimestamp(int(date_before)).strftime(readable_format)
    elif from_format == "readable":
        date_before += "UTC"
        converted_date = int(datetime.strptime(date_before, readable_format).replace(tzinfo=timezone.utc).timestamp())  # noqa: UP017

    return str(converted_date)


def decode_message(msg: str) -> Sequence[str | None]:
    """
        Follow these steps for data members that appear within the event's attackData section:
            1. If the member name is prefixed rule, URL-decode the value.
            2. The result is a series of base64-encoded chunks delimited with semicolons.
            3. Split the value at semicolon (;) characters.
            4. base64-decode each chunk of split data.
             The example above would yield a sequence of alert, alert, and deny.
    Args:
        msg: Messeage to decode

    Returns:
        Decoded message as array

    Examples:
        >>> decode_message(msg='ZGVueQ%3d%3d')
        ['deny']
        >>> decode_message(msg='Q3VzdG9tX1JlZ0VYX1J1bGU%3d%3bTm8gQWNjZXB0IEhlYWRlciBBTkQgTm8gVXNlciBBZ2VudCBIZWFkZXI%3d')
        ['Custom_RegEX_Rule', 'No Accept Header AND No User Agent Header']
    """
    if not msg:
        return []
    readable_msg = []
    translated_msg = urllib.parse.unquote(msg).split(";")
    for word in translated_msg:
        word = b64decode(word).decode("utf-8", errors="replace")
        if word:
            readable_msg.append(word)
    return readable_msg


def events_to_ec(raw_response: list) -> tuple[list, list, list]:
    """
        Convert raw response response to ec
    Args:
        raw_response: events as list from raw response

    Returns:
        events as defined entry context and events for human readable
    """
    events_ec: list[dict] = []
    ip_ec: list[dict] = []
    events_human_readable: list[dict] = []

    for event in raw_response:
        # Hoist the three nested sections once per event to avoid re-evaluating ``event.get(...)``
        # dozens of times below. Behavior is unchanged: missing sections still default to {}.
        attack_data = event.get("attackData", {})
        http_message = event.get("httpMessage", {})
        geo = event.get("geo", {})

        events_ec.append(
            {
                "AttackData": assign_params(
                    ConfigID=attack_data.get("configId"),
                    PolicyID=attack_data.get("policyId"),
                    ClientIP=attack_data.get("clientIP"),
                    Rules=decode_message(attack_data.get("rules")),
                    RuleMessages=decode_message(attack_data.get("ruleMessages")),
                    RuleTags=decode_message(attack_data.get("ruleTags")),
                    RuleData=decode_message(attack_data.get("ruleData")),
                    RuleSelectors=decode_message(attack_data.get("ruleSelectors")),
                    RuleActions=decode_message(attack_data.get("ruleActions")),
                ),
                "HttpMessage": assign_params(
                    RequestId=http_message.get("requestId"),
                    Start=http_message.get("start"),
                    Protocol=http_message.get("protocol"),
                    Method=http_message.get("method"),
                    Host=http_message.get("host"),
                    Port=http_message.get("port"),
                    Path=http_message.get("path"),
                    RequestHeaders=http_message.get("requestHeaders"),
                    Status=http_message.get("status"),
                    Bytes=http_message.get("bytes"),
                    ResponseHeaders=http_message.get("responseHeaders"),
                ),
                "Geo": assign_params(
                    Continent=geo.get("continent"),
                    Country=geo.get("country"),
                    City=geo.get("city"),
                    RegionCode=geo.get("regionCode"),
                    Asn=geo.get("asn"),
                ),
            }
        )

        ip_ec.append(
            assign_params(
                Address=attack_data.get("clientIP"),
                ASN=geo.get("asn"),
                Geo={"Country": geo.get("country")},
            )
        )

        events_human_readable.append(
            assign_params(
                **{
                    "Attacking IP": attack_data.get("clientIP"),
                    "Config ID": attack_data.get("configId"),
                    "Policy ID": attack_data.get("policyId"),
                    "Rules": decode_message(attack_data.get("rules")),
                    "Rule messages": decode_message(attack_data.get("ruleMessages")),
                    "Rule actions": decode_message(attack_data.get("ruleActions")),
                    "Date occured": date_format_converter(from_format="epoch", date_before=http_message.get("start")),
                    "Location": {"Country": geo.get("country"), "City": geo.get("city")},
                }
            )
        )

    return events_ec, ip_ec, events_human_readable


""" COMMANDS """


@logger
def test_module_command(client: Client, config_ids: str) -> tuple[None, None, str]:
    """Performs a basic GET request to check if the API is reachable and authentication is successful.

    Args:
        client: Client object with request
        config_ids: Validated config IDs from params

    Returns:
        'ok' if test successful.

    Raises:
        DemistoException: If test failed.
    """
    # Test on the following date Monday, 6 March 2017 16:07:22
    events, offset = client.get_events(config_ids=config_ids, from_epoch="1488816442", limit="1")
    if isinstance(events, list):
        return None, None, "ok"
    raise DemistoException(f"Test module failed, {events}")


@logger
def fetch_incidents_command(
    client: Client, fetch_time: str, fetch_limit: str | int, config_ids: str, last_run: str | None = None
) -> tuple[list[dict[str, Any]], dict]:
    """Uses to fetch incidents into Demisto
    Documentation: https://github.com/demisto/content/tree/master/docs/fetching_incidents

    Args:
        client: Client object with request
        fetch_time: From when to fetch if first time, e.g. `3 days`
        fetch_limit: limit of incidents in a fetch
        config_ids: security configuration ids to fetch, e.g. `51000;56080`
        last_run: Last fetch object occurs.

    Returns:
        incidents, new last_run
    """
    raw_response: list | None = []
    if not last_run:
        last_run, _ = parse_date_range(date_range=fetch_time, date_format="%s")
    raw_response, offset = client.get_events(config_ids=config_ids, from_epoch=last_run, limit=fetch_limit)

    incidents = []
    if raw_response:
        for event in raw_response:
            attack_data = event.get("attackData", {})
            http_message = event.get("httpMessage", {})
            incidents.append(
                {
                    "name": f"{INTEGRATION_NAME}: {attack_data.get('configId')} - {http_message.get('requestId')}",
                    "occurred": date_format_converter(from_format="epoch", date_before=http_message.get("start")),
                    "rawJSON": json.dumps(event),
                }
            )

    return incidents, {"lastRun": offset}


def get_events_command(
    client: Client,
    config_ids: str,
    offset: str | None = None,
    limit: str | None = None,
    from_epoch: str | None = None,
    to_epoch: str | None = None,
    time_stamp: str | None = None,
) -> tuple[object, dict, list | dict]:
    """
        Get security events from Akamai WAF service
        Allowed query parameters combinations:
            1. offset - Since a prior request.
            2. offset, limit - Since a prior request, limited.
            3. from - Since a point in time.
            4. from, limit - Since a point in time, limited.
            5. from, to - Over a range of time.
            6. from, to, limit - Over a range of time, limited.
    Args:
        client: Client object
        config_ids: Unique identifier for each security configuration. To report on more than one configuration, separate
                  integer identifiers with semicolons, e.g. 12892;29182;82912.
        offset: This token denotes the last message. If specified, this operation fetches only security events that have
                occurred from offset. This is a required parameter for offset mode and you can't use it in time-based requests.
        limit: Defines the approximate maximum number of security events each fetch returns, in both offset and
               time-based modes. The default limit is 20000. Expect requests to return a slightly higher number of
               security events than you set in the limit parameter, because data is stored in different buckets.
        from_epoch: The start of a specified time range, expressed in Unix epoch seconds.
                    This is a required parameter to get time-based results for a set time_stamp, and you can't use it in
                    offset mode.
        to_epoch: The end of a specified time range, expressed in Unix epoch seconds. You can't use this parameter in
                  offset mode and it's an optional parameter in time-based mode. If omitted, the value defaults to the
                  current time.
        time_stamp: timestamp (<number> <time unit>, e.g., 12 hours, 7 days of events

    Returns:
        Human readable, entry context, raw response
    """
    if time_stamp:
        from_epoch, to_epoch = parse_date_range(date_range=time_stamp, date_format="%s")
    raw_response, offset = client.get_events(
        config_ids=config_ids, offset=offset, limit=limit, from_epoch=from_epoch, to_epoch=to_epoch
    )
    if raw_response:
        events_ec, ip_ec, events_human_readable = events_to_ec(raw_response)
        entry_context = {
            "Akamai.SIEM(val.HttpMessage.RequestId && val.HttpMessage.RequestId == obj.HttpMessage.RequestId)": events_ec,
            outputPaths.get("ip"): ip_ec,
        }
        title = f"{INTEGRATION_NAME} - Attacks data"

        human_readable = tableToMarkdown(name=title, t=events_human_readable, removeNull=True)

        return human_readable, entry_context, raw_response
    else:
        return f"{INTEGRATION_NAME} - Could not find any results for given query", {}, {}


def reset_offset_command(client: Client):  # pragma: no cover
    ctx = get_integration_context()
    if "offset" in ctx:
        del ctx["offset"]
    set_integration_context(ctx)
    return "Offset was reset successfully.", {}, {}


def is_offset_out_of_range_error(error: Exception) -> bool:
    """Whether the given Akamai error is the '416 - offset expired / out of range' error.

    Akamai keeps offsets valid for only 12 hours; requesting an offset older than that returns
    HTTP 416 "Requested Range Not Satisfiable".

    Args:
        error: The exception raised by the Akamai client.

    Returns:
        bool: True if this is the offset-out-of-range (expired offset) error.
    """
    return "Requested Range Not Satisfiable" in str(error)


def handle_offset_out_of_range(error: Exception, reset_context_offset: bool) -> str:
    """Recover from an expired/out-of-range Akamai offset (HTTP 416).

    Akamai only keeps offsets valid for 12 hours. When the integration is paused or unavailable
    for longer than that, the stored offset becomes stale and every request using it fails with 416.
    To recover in the same run without losing the whole cycle we:
        1. Drop the stale offset (so callers stop reusing it), optionally clearing it from the
           integration context (used by the offset-in-context ``fetch-events`` flow).
        2. Restart fetching from the latest window Akamai still accepts
           (``AKAMAI_MAX_LOOKBACK_MINUTES``, i.e. just under the 12-hour limit), keeping the data
           gap to the small safety buffer instead of skipping back only a few minutes.

    Args:
        error: The 416 exception raised by the Akamai client.
        reset_context_offset: Whether to also clear the offset persisted in the integration context.
            True for the ``fetch-events`` flow (offset stored in context); False for the
            long-running flow (offset held in a local variable).

    Returns:
        str: The recovery ``from_epoch`` (Unix epoch seconds) to restart the time-based fetch from.
    """
    demisto.error(f"[Offset Recovery] Offset expired / out of range (older than 12h). Recovering. Original error: {error}")
    if reset_context_offset:
        reset_offset_command(client=None)  # type: ignore[arg-type]  # client is unused by reset_offset_command
    recovery_from = f"{AKAMAI_MAX_LOOKBACK_MINUTES} minutes"
    from_epoch, _ = parse_date_range(date_range=recovery_from, date_format="%s")
    from_epoch = str(from_epoch)
    demisto.info(
        f"[Offset Recovery] Stale offset dropped. Restarting fetch from {recovery_from} ago (from_epoch={from_epoch}) "
        f"to stay within Akamai's 12-hour limit. Some events during the outage may be missed."
    )
    return from_epoch


def is_last_request_smaller_than_page_size(num_events_from_previous_request: int, page_size: int) -> bool:
    """Checks wether the number of events from the last API call was lower by a certain delta than the request page size.

    Args:
        num_events_from_previous_request (int): The length of the list of events from previous response
        page_size (int): the request limit for the last request.

    Returns:
        bool: True if the number of events from last API call was lower by a certain delta for the requested page size.
              Otherwise, return False
    """
    demisto.debug(
        f"[Breaking Conditions] Checking whether execution should break with "
        f"{num_events_from_previous_request=} and {page_size=}"
    )
    return num_events_from_previous_request < page_size * ALLOWED_PAGE_SIZE_DELTA_RATIO


def is_interval_doesnt_have_enough_time_to_run(min_allowed_delta: int, max_time_took: float) -> tuple[bool, float]:
    """
    Checking whether there's enough time for another fetch request to the Akamai API before docker timeout.
    The function calculates the time of the first request (including the send_events_to_xsiam_part).
    And checks wether the remaining running time (plus a little delta) is less or equal the expected running time.
    The remaining running time is docker timeout limit in seconds - the run time so far (now time - docker execution start time).

    Args:
        min_allowed_delta (int): The minimum allowed delta that should remain before going on another fetch interval.
        max_time_took (float): The worst case execution (the first execution) to compare the rest of the executions to.
    Returns:
        bool: Return True if there's not enough time. Otherwise, return False.
    """
    timeout_time_nano_seconds = demisto.callingContext.get("context", {}).get("TimeoutDuration")
    demisto.debug(f"[Breaking Conditions] Got {timeout_time_nano_seconds} nano seconds timeout budget for the execution.")
    timeout_time_seconds = timeout_time_nano_seconds / 1_000_000_000
    now = datetime.now()
    time_since_interval_beginning = (now - EXECUTION_START_TIME).total_seconds()
    if not max_time_took:
        max_time_took = time_since_interval_beginning
    demisto.debug(
        f"[Breaking Conditions] Checking if execution should break with {time_since_interval_beginning=}, {max_time_took=}."
    )
    return (timeout_time_seconds - time_since_interval_beginning - min_allowed_delta) <= max_time_took, max_time_took


@logger
def fetch_events_command(
    client: Client, fetch_time: str, fetch_limit: int, config_ids: str, ctx: dict, page_size: int, should_skip_decode_events: bool
) -> Iterator[Any]:
    """Iteratively gathers events from Akamai SIEM. Stores the offset in integration context.

    Args:
        client: Client object with request
        fetch_time: From when to fetch if first time, e.g. `3 days`
        fetch_limit: limit of events in a fetch
        config_ids: security configuration ids to fetch, e.g. `51000;56080`
        ctx: The integration context
        page_size: The number of events to limit for every request.
        should_skip_decode_events: Whether to skip events decoding or not.

    Yields:
        (list, str | None, int, bool): The page of events, the new offset, the total number of events
        fetched so far, and whether to set nextTrigger=0 for the next execution.
    """
    total_events_count = 0
    offset = ctx.get("offset")
    from_epoch, _ = parse_date_range(fetch_time, date_format="%s")
    auto_trigger_next_run = False
    worst_case_time: float = 0
    execution_counter = 0
    base_page_size = page_size
    log_prefix = "[Fetch Events]"
    offset_recovery_attempted = False
    while total_events_count < fetch_limit:
        if execution_counter > 0:
            demisto.debug(f"{log_prefix} Execution number {execution_counter}: checking for breaking conditions.")
            if is_last_request_smaller_than_page_size(last_page_size, base_page_size):  # type: ignore[has-type]  # pylint: disable=E0601
                demisto.debug(f"{log_prefix} Last request was smaller than the page size, breaking.")
                break
            should_break, worst_case_time = is_interval_doesnt_have_enough_time_to_run(TIME_TO_RUN_BUFFER, worst_case_time)
            if should_break:
                demisto.debug(f"{log_prefix} Not enough time for another execution, breaking and triggering next run.")
                auto_trigger_next_run = True
                break
        if (remaining_events_to_fetch := fetch_limit - total_events_count) < page_size:
            demisto.debug(
                f"{log_prefix} Remaining events to fetch ({remaining_events_to_fetch}) < page_size ({page_size}), "
                f"lowering page_size to {remaining_events_to_fetch}."
            )
            page_size = remaining_events_to_fetch
        demisto.debug(f"{log_prefix} Requesting events with {offset=}, {page_size=}, {fetch_limit=}.")
        try:
            events, offset = client.get_events_with_offset(config_ids, offset, page_size, from_epoch)
        except DemistoException as e:
            if is_offset_out_of_range_error(e):
                if offset_recovery_attempted:
                    demisto.error(
                        f"{log_prefix} Offset still out of range after recovery - aborting to avoid an infinite retry loop."
                    )
                    raise DemistoException(e)
                # The stored offset expired (older than 12h). Recover in-run: drop the stale offset and
                # restart from the latest window Akamai still accepts, then retry immediately.
                offset_recovery_attempted = True
                from_epoch = handle_offset_out_of_range(e, reset_context_offset=True)
                offset = None
                continue
            demisto.error(f"{log_prefix} Failed requesting new events from Akamai: {e}")
            raise DemistoException(e)
        offset_recovery_attempted = False

        if not events:
            demisto.debug(f"{log_prefix} Received no events, breaking.")
            break
        last_page_size = len(events)
        demisto.debug(f"{log_prefix} Received {last_page_size} events (Akamai API), processing.")
        if should_skip_decode_events:
            demisto.debug(f"{log_prefix} should_skip_decode_events is set, skipping events decode.")
        else:
            demisto.debug(f"{log_prefix} Loading and decoding {last_page_size} events.")
            for index, event in enumerate(events):
                events[index] = decode_event(event)  # type: ignore[assignment]
        total_events_count += last_page_size
        execution_counter += 1
        demisto.debug(f"{log_prefix} Processed page of {last_page_size} events ({total_events_count=}).")
        yield events, offset, total_events_count, auto_trigger_next_run
    yield [], offset, total_events_count, auto_trigger_next_run


def decode_url(headers: str) -> dict:
    """Decoding the httpMessage headers parts of the response.

    Args:
        headers (str): The headers to decode

    Returns:
        dict: The decoded and parsed headers as a dictionary.
    """
    if not headers:
        return {}
    decoded_lines = urllib.parse.unquote(headers).replace("\r", "").split("\n")
    decoded_dict = {}
    for line in decoded_lines:
        parts = line.split(": ", 1)
        if len(parts) == 2:
            key, value = parts
            decoded_dict[key.replace("-", "_")] = value.replace('"', "")
    return decoded_dict


# Keys inside an event's attackData section whose values are base64/URL-encoded and must be decoded.
ATTACK_DATA_KEYS_TO_DECODE = [
    "rules",
    "ruleMessages",
    "ruleTags",
    "ruleData",
    "ruleSelectors",
    "ruleActions",
    "ruleVersions",
]


def decode_event(event: str) -> dict | str:
    """Deserialize a single raw JSON event string and decode its encoded fields in place.

    The event's ``attackData`` rule fields are base64/URL decoded (via ``decode_message``) and the
    ``httpMessage`` request/response headers are decoded (via ``decode_url``). This is the single
    source of truth for the per-event decoding previously duplicated in ``fetch_events_command`` and
    ``process_and_send_events_to_xsiam``.

    Args:
        event (str): A single event as a raw JSON string.

    Returns:
        dict | str: The decoded event as a dict. If the input can't be decoded (e.g. malformed JSON),
        the original raw string is returned unchanged, so a bad event never breaks a whole page.
    """
    try:
        parsed: dict = json.loads(event)
    except Exception as e:
        demisto.debug(f"Couldn't decode {event=}, reason: {e}\n{traceback.format_exc()}")
        return event
    if "attackData" in parsed:
        for attack_data_key in ATTACK_DATA_KEYS_TO_DECODE:
            parsed["attackData"][attack_data_key] = decode_message(parsed["attackData"].get(attack_data_key, ""))
    if "httpMessage" in parsed:
        parsed["httpMessage"]["requestHeaders"] = decode_url(parsed["httpMessage"].get("requestHeaders", ""))
        parsed["httpMessage"]["responseHeaders"] = decode_url(parsed["httpMessage"].get("responseHeaders", ""))
    return parsed


def post_latest_event_time(latest_event, base_msg):
    try:
        if isinstance(latest_event, str):
            latest_event = json.loads(latest_event)
        latest_event_time = date_format_converter(
            from_format="epoch", date_before=latest_event.get("httpMessage", {}).get("start", "0")
        )
        demisto.debug(f"[Fetch] {base_msg} latest event time is: {latest_event_time}")
    except Exception as e:
        demisto.debug(f"[Fetch] Caught an exception when attempting to compute latest event time: {e}")
        demisto.debug(f"[Fetch] {base_msg}")


############################################## Beginning of beta part ##############################################
BETA_FETCH_EVENTS_MAX_PAGE_SIZE = 600000  # Allowed events limit per request.
MAX_ALLOWED_CONCURRENT_TASKS = 10000


""" COMMANDS """


async def wait_until_tasks_load_decrease(counter: int, max_concurrent_tasks: int):
    """Testing if there's a tasks load. If so, will go to sleep for 30 seconds so execution can catch up.

    Args:
        counter (int): The current execution number.
        max_concurrent_tasks (int): The maximum number of tasks allowed to run concurrently.
    """
    while (num_of_tasks := len(asyncio.all_tasks())) > max_concurrent_tasks:
        demisto.debug(
            f"Running in interval = {counter}. current tasks total size = {num_of_tasks} is larger than the max allowed"
            f" number of tasks {max_concurrent_tasks}. sleeping for 30 seconds to let other tasks finish."
        )
        await asyncio.sleep(30)


@logger
async def fetch_events_long_running_command(
    client: Client,
    from_time: str,
    page_size: int,
    config_ids: str,
    ctx: dict,
    should_skip_decode_events: bool,
    max_concurrent_tasks: int,
):
    """Asynchronously gathers events from Akamai SIEM. Decode them, and send them to xsiam.

    Args:
        client: Client object with request.
        from_time (str): From when to fetch if first time, e.g. `3 days`.
        page_size (int): The number of events to limit for every request.
        config_ids (str): security configuration ids to fetch, e.g. `51000;56080`.
        ctx (dict): The integration context.
        should_skip_decode_events (bool): Whether we should skip serializing and decoding events or not.
        max_concurrent_tasks (int): The maximum number of tasks allowed to run concurrently.

    """
    offset = ctx.get("offset")
    async for events, counter, last_offset in get_events_from_akamai(
        client, config_ids, from_time, page_size, offset, max_concurrent_tasks
    ):
        asyncio.create_task(  # noqa: RUF006
            process_and_send_events_to_xsiam(
                events,
                should_skip_decode_events,  # noqa: RUF006
                last_offset,
                counter,
            )
        )
        offset = last_offset


async def process_and_send_events_to_xsiam(events: list[str], should_skip_decode_events: bool, offset: str | None, counter: int):
    """Process the event if should_skip_decode_events=False (json serialize and decode).
    Send the events to CSP's send_events_to_xsiam.
    save the offset after the execution is done and update the module health.

    Args:
        events (list[str]): The list of json serialized events.
        should_skip_decode_events (bool): Whether we should skip serializing and decoding events or not.
        offset (str | None): The offset hash.
        counter (int): The current execution number.
    """
    demisto.debug(f"Running in interval = {counter}. got {len(events)} events, moving to processing events data.")
    processed_events: list = []
    if should_skip_decode_events:
        demisto.debug(f"Running in interval = {counter}. Skipping decode events.")
        processed_events = events
    else:
        demisto.debug(f"Running in interval = {counter}. decoding events.")
        processed_events = [decode_event(event) for event in events]
    post_latest_event_time(
        latest_event=processed_events[-1],
        base_msg=f"Running in interval = {counter}. Sending {len(processed_events)} events to xsiam.",
    )
    tasks = send_events_to_xsiam_akamai(
        processed_events,
        VENDOR,
        PRODUCT,
        should_update_health_module=False,
        chunk_size=SEND_EVENTS_TO_XSIAM_CHUNK_SIZE,
        send_events_asynchronously=True,
        url_key="host",
        data_format="json",
        data_size_expected_to_split_evenly=True,
        counter=counter,
    )
    demisto.debug(f"Running in interval = {counter}. Finished executing send_events_to_xsiam, waiting for tasks to end.")
    await asyncio.gather(*tasks)
    demisto.debug(f"Running in interval = {counter}. Finished gathering all tasks.")
    demisto.debug(f"Running in interval = {counter}. Updating module health.")
    set_integration_context({"offset": offset})
    demisto.updateModuleHealth({"eventsPulled": len(processed_events)})
    demisto.debug(f"Running in interval = {counter}. Finished updating module health.")


async def get_events_from_akamai(
    client: Client, config_ids: str, from_time: str, page_size: int, offset: str | None, max_concurrent_tasks: int
):
    """Iteratively checks for condition where the loop should wait due to multiple reasons
    and then gathers events from Akamai SIEM.

    Args:
        client: Client object with request.
        config_ids (str): security configuration ids to fetch, e.g. `51000;56080`.
        from_time (str): From when to fetch if first time, e.g. `3 days`.
        page_size: The number of events to limit for every request.
        offset (str | None): The offset (hash) to use for offset based mechanism.
        max_concurrent_tasks (int): The maximum number of tasks allowed to run concurrently.

    Yields:
        (list[str], int, str): events (not serialized), counter (the current execution number), and the new offset.
    """
    counter = 0
    while True:
        if counter > 100_000:
            demisto.debug("counter reach 100k, bringing it back to 0.")
            counter = 0
        counter += 1
        from_epoch, _ = parse_date_range(date_range=from_time, date_format="%s")
        demisto.debug(f"Running in interval = {counter}. Preparing to get events with {offset=}, and {page_size=}.")
        try:
            demisto.debug(f"Running in interval = {counter}. Testing for possible tasks qt overflow.")
            await wait_until_tasks_load_decrease(counter, max_concurrent_tasks)
            demisto.debug(f"Running in interval = {counter}. Finished testing for possible tasks qt overflow.")
            get_events_task = client.get_events_concurrently(config_ids, offset, page_size, from_epoch, counter=counter)
            events, offset = None, None
            events, offset = await get_events_task
            demisto.debug(f"Running in interval = {counter}. got {len(events)} events and {offset=}.")
        except DemistoException as e:
            if is_offset_out_of_range_error(e):
                # The offset expired (older than 12h). This is a self-healing situation:
                # drop the offset so the next iteration falls back to a time-based request, and shorten the
                # fetch window to the latest range Akamai still accepts (restarting from the original from_time
                # could again exceed 12h and loop on the same 416). The long-running flow keeps the offset in a
                # local variable, so no context reset is needed. Since we keep running, this is NOT reported as
                # an error in module health.
                handle_offset_out_of_range(e, reset_context_offset=False)
                offset = None
                from_time = f"{AKAMAI_MAX_LOOKBACK_MINUTES} minutes"
                demisto.error(
                    f"Running in interval = {counter}. Offset expired (older than 12h); restarting fetch from "
                    f"{from_time} ago and continuing. Some events during the outage may be missed."
                )
            else:
                demisto.error(f"{e.message}")
                demisto.updateModuleHealth(str(e), is_error=True)
            demisto.debug(f"Running in interval = {counter}. Going to sleep for 60 seconds.")
            await asyncio.sleep(60)
            demisto.debug(f"Running in interval = {counter}. Done sleeping 60 seconds.")
        if events:
            yield events, counter, offset
        if not events or is_last_request_smaller_than_page_size(len(events), page_size):
            if not events:
                demisto.debug(
                    f"Running in interval = {counter}. No events were received from Akamai,going to sleep for 60 seconds."
                )
            else:
                demisto.debug(
                    f"Running in interval = {counter}. got {len(events)} events which is less"
                    f" than {ALLOWED_PAGE_SIZE_DELTA_RATIO} % of the {page_size=}, going to sleep for 60 seconds."
                )
            await asyncio.sleep(60)
            demisto.debug(f"Running in interval = {counter}. Finished sleeping for 60 seconds.")


############################################## Beginning of CSP copy-paste part ##############################################


def akamai_send_data_to_xsiam(
    data,
    vendor,
    product,
    data_format=None,
    url_key="url",
    num_of_attempts=3,
    chunk_size=XSIAM_EVENT_CHUNK_SIZE,
    data_type=EVENTS,
    should_update_health_module=True,
    add_proxy_to_request=False,
    snapshot_id="",
    items_count=None,
    send_events_asynchronously=False,
    data_size_expected_to_split_evenly=False,
    counter=0,
):  # pragma: no cover
    """
    Send the supported fetched data types into the XDR data-collector private api.

    :type data: ``Union[str, list]``
    :param data: The data to send to XSIAM server. Should be of the following:
        1. List of strings or dicts where each string or dict represents an event or asset.
        2. String containing raw events separated by a new line.

    :type vendor: ``str``
    :param vendor: The vendor corresponding to the integration that originated the data.

    :type product: ``str``
    :param product: The product corresponding to the integration that originated the data.

    :type data_format: ``str``
    :param data_format: Should only be filled in case the 'events' parameter contains a string of raw
        events in the format of 'leef' or 'cef'. In other cases the data_format will be set automatically.

    :type url_key: ``str``
    :param url_key: The param dict key where the integration url is located at. the default is 'url'.

    :type num_of_attempts: ``int``
    :param num_of_attempts: The num of attempts to do in case there is an api limit (429 error codes)

    :type chunk_size: ``int``
    :param chunk_size: Advanced - The maximal size of each chunk size we send to API. Limit of 9 MB will be inforced.

    :type data_type: ``str``
    :param data_type: Type of data to send to Xsiam, events or assets.

    :type should_update_health_module: ``bool``
    :param should_update_health_module: whether to trigger the health module showing how many events were sent to xsiam
        This can be useful when using send_data_to_xsiam in batches for the same fetch.

    :type add_proxy_to_request: ``bool``
    :param add_proxy_to_request: whether to add proxy to the send evnets request.

    :type snapshot_id: ``str``
    :param snapshot_id: the snapshot id.

    :type items_count: ``str``
    :param items_count: the asset snapshot items count.

    :type send_events_asynchronously: ``bool``
    :param send_events_asynchronously: whether to use asyncio to send the events to xsiam asynchronously or not.
    Note that when set to True, the updateModuleHealth should be done from the integration itself.

    :type data_size_expected_to_split_evenly: ``bool``
    :param data_size_expected_to_split_evenly: whether the events should be about the same size or not.
    Use this to split data to chunks faster.

    :return: Either None if running regularly or a list of asyncio task objects if running asynchronously:.
    In case of running asynchronously:, the list of tasks will hold the number of events sent and can be accessed by:
    await asyncio.gather(*tasks)
    :rtype: ``List[Task]`` or ``None``
    """
    data_size = 0
    params = demisto.params()
    url = params.get(url_key)
    calling_context = demisto.callingContext.get("context", {})
    instance_name = calling_context.get("IntegrationInstance", "")
    collector_name = calling_context.get("IntegrationBrand", "")
    if not items_count:
        items_count = len(data) if isinstance(data, list) else 1
    if data_type not in DATA_TYPES:
        demisto.debug(f"data type must be one of these values: {DATA_TYPES}")
        return None

    if not data:
        demisto.debug(f"send_data_to_xsiam function received no {data_type}, skipping the API call to send {data_type} to XSIAM")
        demisto.updateModuleHealth({f"{data_type}Pulled": data_size})
        return None

    # only in case we have data to send to XSIAM we continue with this flow.
    # Correspond to case 1: List of strings or dicts where each string or dict represents an one event or asset or snapshot.
    if isinstance(data, list):
        # In case we have list of dicts we set the data_format to json and parse each dict to a stringify each dict.
        demisto.debug(f"Sending {len(data)} {data_type} to XSIAM")
        if isinstance(data[0], dict):
            data = [json.dumps(item) for item in data]
            data_format = "json"
        # Separating each event with a new line
        data = "\n".join(data)
    elif not isinstance(data, str):
        raise DemistoException(f"Unsupported type: {type(data)} for the {data_type} parameter. Should be a string or list.")
    if not data_format:
        data_format = "text"

    xsiam_api_token = demisto.getLicenseCustomField("Http_Connector.token")
    xsiam_domain = demisto.getLicenseCustomField("Http_Connector.url")
    xsiam_url = f"https://api-{xsiam_domain}"
    headers = remove_empty_elements(
        {
            "authorization": xsiam_api_token,
            "format": data_format,
            "product": product,
            "vendor": vendor,
            "content-encoding": "gzip",
            "collector-name": collector_name,
            "instance-name": instance_name,
            "final-reporting-device": url,
            "collector-type": ASSETS if data_type == ASSETS else EVENTS,
        }
    )
    if data_type == ASSETS:
        if not snapshot_id:
            snapshot_id = str(round(time.time() * 1000))

        # We are setting a time stamp ahead of the instance name since snapshot-ids must be configured in ascending
        # alphabetical order such that first_snapshot < second_snapshot etc.
        headers["snapshot-id"] = snapshot_id + instance_name
        headers["total-items-count"] = str(items_count)

    header_msg = f"Error sending new {data_type} into XSIAM.\n"

    def data_error_handler(res):
        """
        Internal function to parse the XSIAM API errors
        """
        try:
            response = res.json()
            error = res.reason
            if response.get("error").lower() == "false":
                xsiam_server_err_msg = response.get("error")
                error += ": " + xsiam_server_err_msg

        except ValueError:
            if res.text:
                error = f"\n{res.text}"
            else:
                error = "Received empty response from the server"

        api_call_info = (
            "Parameters used:\n"
            f"\tURL: {xsiam_url}\n"
            f"\tHeaders: {json.dumps(headers, indent=8)}\n\n"
            f"Response status code: {res.status_code}\n"
            f"Error received:\n\t{error}"
        )

        demisto.error(header_msg + api_call_info)
        raise DemistoException(header_msg + error, DemistoException)

    client = BaseClient(base_url=xsiam_url, proxy=add_proxy_to_request)
    if data_size_expected_to_split_evenly:
        data_chunks = split_data_by_slices(data, chunk_size)
    else:
        data_chunks = split_data_to_chunks(data, chunk_size)

    def send_events(data_chunk):
        chunk_size = len(data_chunk)
        data_chunk = "\n".join(data_chunk)
        zipped_data = gzip.compress(data_chunk.encode("utf-8"))  # type: ignore[AttributeError,attr-defined]
        xsiam_api_call_with_retries(
            client=client,
            events_error_handler=data_error_handler,
            error_msg=header_msg,
            headers=headers,
            num_of_attempts=num_of_attempts,
            xsiam_url=xsiam_url,
            zipped_data=zipped_data,
            is_json_response=True,
            data_type=data_type,
        )
        return chunk_size

    async def send_events_async(data_chunk):
        chunk_size = len(data_chunk)
        data_chunk = "\n".join(data_chunk)
        zipped_data = gzip.compress(data_chunk.encode("utf-8"))  # type: ignore[AttributeError,attr-defined]
        _ = await xsiam_api_call_async_with_retries(
            headers=headers, num_of_attempts=num_of_attempts, xsiam_url=xsiam_url, zipped_data=zipped_data, data_type=data_type
        )
        return chunk_size

    if send_events_asynchronously:
        demisto.debug(f"Running in interval = {counter}. Sending events to xsiam asynchronously.")
        all_chunks = list(data_chunks)
        demisto.debug(f"Running in interval = {counter}. Finished appending all data_chunks to a list.")
        tasks = [asyncio.create_task(send_events_async(chunk)) for chunk in all_chunks]

        demisto.debug(f"Finished submiting {len(tasks)} tasks for the {counter} time")
        return tasks
    else:
        demisto.debug("Sending events to xsiam synchronously.")
        for chunk in data_chunks:
            data_size += send_events(chunk)

        if should_update_health_module:
            demisto.updateModuleHealth({f"{data_type}Pulled": data_size})
    return None


def split_data_by_slices(data, target_chunk_size):  # pragma: no cover
    """
    Splits a string/list of data into chunks of an approximately specified size.
    The actual size can be lower. the slicing is based on the assumption that all entries have the same size.

    :type data: ``list`` or a ``string``
    :param data: A list of data or a string delimited with \n  to split to chunks.
    :type target_chunk_size: ``int``
    :param target_chunk_size: The maximum size of each chunk. The maximal size allowed is 9MB.

    :return: An iterable of lists where each list contains events with approx size of chunk size.
    :rtype: ``collections.Iterable[list]``
    """
    target_chunk_size = min(target_chunk_size, XSIAM_EVENT_CHUNK_SIZE_LIMIT)
    if isinstance(data, str):
        data = data.split("\n")
    entry_size = sys.getsizeof(data[0])
    num_of_entries_per_chunk = target_chunk_size // entry_size
    for i in range(0, len(data), num_of_entries_per_chunk):
        chunk = data[i : i + num_of_entries_per_chunk]
        yield chunk


async def xsiam_api_call_async_with_retries(
    xsiam_url,
    zipped_data,
    headers,
    num_of_attempts,
    data_type=EVENTS,
):  # pragma: no cover
    """
    Send the fetched events or assets into the XDR data-collector private api.
    :type xsiam_url: ``str``
    :param xsiam_url: The URL of XSIAM to send the api request.
    :type zipped_data: ``bytes``
    :param zipped_data: encoded events
    :type headers: ``dict``
    :param headers: headers for the request
    :type num_of_attempts: ``int``
    :param num_of_attempts: The num of attempts to do in case there is an api limit (429 error codes).
    :type data_type: ``str``
    :param data_type: events or assets
    :return: Response object or DemistoException
    :rtype: ``requests.Response`` or ``DemistoException``
    """
    # retry mechanism in case there is a rate limit (429) from xsiam.
    status_code = None
    attempt_num = 1
    response = None
    while status_code != 200 and attempt_num < num_of_attempts + 1:
        demisto.debug(f"Sending {data_type} into xsiam, attempt number {attempt_num}")
        # in the last try we should raise an exception if any error occurred, including 429
        ok_codes = (200, 429) if attempt_num < num_of_attempts else None
        async with aiohttp.ClientSession() as session:  # noqa: SIM117
            async with session.post(urljoin(xsiam_url, "/logs/v1/xsiam"), data=zipped_data, headers=headers) as response:
                try:
                    response.raise_for_status()  # This raises an exception for non-2xx status codes
                    status_code = response.status
                except aiohttp.ClientResponseError as e:
                    if ok_codes and e.status in ok_codes:
                        continue
                    else:
                        header_msg = f"Error sending new {data_type} into XSIAM.\n"
                        api_call_info = (
                            "Parameters used:\n"
                            f"\tURL: {xsiam_url}\n"
                            f"\tHeaders: {json.dumps(e.headers, indent=8)}\n\n"
                            f"Response status code: {e.status}\n"
                            f"Error received:\n\t{e.message}\n"
                            f"additional request info: \n\t{e.request_info}"
                        )

                        demisto.error(header_msg + api_call_info)
                        demisto.updateModuleHealth(header_msg + e.message, is_error=True)

        demisto.debug(f"received status code: {status_code}")
        if status_code == 429:
            await asyncio.sleep(1)
        attempt_num += 1
    return response


def send_events_to_xsiam_akamai(
    events,
    vendor,
    product,
    data_format=None,
    url_key="url",
    num_of_attempts=3,
    chunk_size=XSIAM_EVENT_CHUNK_SIZE,
    should_update_health_module=True,
    add_proxy_to_request=False,
    send_events_asynchronously=False,
    data_size_expected_to_split_evenly=False,
    counter=0,
):  # pragma: no cover
    """
    Send the fetched events into the XDR data-collector private api.

    :type events: ``Union[str, list]``
    :param events: The events to send to XSIAM server. Should be of the following:
        1. List of strings or dicts where each string or dict represents an event.
        2. String containing raw events separated by a new line.

    :type vendor: ``str``
    :param vendor: The vendor corresponding to the integration that originated the events.

    :type product: ``str``
    :param product: The product corresponding to the integration that originated the events.

    :type data_format: ``str``
    :param data_format: Should only be filled in case the 'events' parameter contains a string of raw
        events in the format of 'leef' or 'cef'. In other cases the data_format will be set automatically.

    :type url_key: ``str``
    :param url_key: The param dict key where the integration url is located at. the default is 'url'.

    :type num_of_attempts: ``int``
    :param num_of_attempts: The num of attempts to do in case there is an api limit (429 error codes)

    :type chunk_size: ``int``
    :param chunk_size: Advanced - The maximal size of each chunk size we send to API. Limit of 9 MB will be inforced.

    :type should_update_health_module: ``bool``
    :param should_update_health_module: whether to trigger the health module showing how many events were sent to xsiam

    :type add_proxy_to_request :``bool``
    :param add_proxy_to_request: whether to add proxy to the send evnets request.

    :type send_events_asynchronously: ``bool``
    :param send_events_asynchronously: whether to use asyncio to send the events to xsiam asynchronously or not.
    Note that when set to True, the updateModuleHealth should be done from the integration itself.

    :type data_size_expected_to_split_evenly: ``bool``
    :param data_size_expected_to_split_evenly: whether the events should be about the same size or not.
    Use this to split data to chunks faster.

    :return: Either None if running regularly or a list of asyncio task objects if running asynchronously:.
    In case of running asynchronously:, the list of tasks will hold the number of events sent and can be accessed by:
    await asyncio.gather(*tasks)
    :rtype: ``List[Task]`` or ``None``
    """
    return akamai_send_data_to_xsiam(
        events,
        vendor,
        product,
        data_format,
        url_key,
        num_of_attempts,
        chunk_size,
        data_type="events",
        should_update_health_module=should_update_health_module,
        add_proxy_to_request=add_proxy_to_request,
        send_events_asynchronously=send_events_asynchronously,
        data_size_expected_to_split_evenly=data_size_expected_to_split_evenly,
        counter=counter,
    )


############################################## end of CSP copy-paste part ##############################################


############################################## end of beta part ##############################################


""" COMMANDS MANAGER / SWITCH PANEL """


def main():
    params = demisto.params()

    # Validate that configIds is not empty
    config_ids = params.get("configIds")
    if not config_ids:
        raise DemistoException(
            "Config IDs parameter is required and cannot be empty. Please provide your Akamai security configuration ID(s). "
        )

    client = Client(
        base_url=urljoin(params.get("host"), "/siem/v1/configs"),
        verify=not params.get("insecure", False),
        proxy=params.get("proxy", False),
        auth=EdgeGridAuth(
            client_token=params.get("clienttoken_creds", {}).get("password") or params.get("clientToken"),
            access_token=params.get("accesstoken_creds", {}).get("password") or params.get("accessToken"),
            client_secret=params.get("clientsecret_creds", {}).get("password") or params.get("clientSecret"),
        ),
    )
    commands = {
        "test-module": test_module_command,
        f"{INTEGRATION_COMMAND_NAME}-get-events": get_events_command,
        f"{INTEGRATION_COMMAND_NAME}-reset-offset": reset_offset_command,
    }
    command = demisto.command()
    demisto.debug(f"Command being called is {command}")

    try:
        if params.get("isFetch") and not (
            0
            < (arg_to_number(params.get("fetchLimit", DEFAULT_INCIDENTS_FETCH_LIMIT)) or DEFAULT_INCIDENTS_FETCH_LIMIT)
            <= MAX_INCIDENTS_FETCH_LIMIT
        ):
            raise DemistoException(f"Fetch limit must be an integer between 1 and {MAX_INCIDENTS_FETCH_LIMIT}")

        if command == "fetch-incidents":
            incidents, new_last_run = fetch_incidents_command(
                client,
                fetch_time=params.get("fetchTime", "1 hours"),
                fetch_limit=params.get("fetchLimit", DEFAULT_INCIDENTS_FETCH_LIMIT),
                config_ids=config_ids,
                last_run=demisto.getLastRun().get("lastRun"),
            )
            demisto.incidents(incidents)
            demisto.setLastRun(new_last_run)
        elif command == "fetch-events":
            if params.get("longRunning", False):
                raise DemistoException(
                    "Cannot run both fetch events and long-running command simultaneously.\n"
                    "Please make sure to set either isFetchEvents or longRunning to false in"
                    " the integration configuration."
                )
            page_size = DEFAULT_PAGE_SIZE
            # Event collection uses its own dedicated "eventsFetchLimit" param (defaulting to
            # DEFAULT_EVENTS_FETCH_LIMIT). The "fetchLimit" param controls incident fetching only.
            events_fetch_limit = params.get("eventsFetchLimit") or DEFAULT_EVENTS_FETCH_LIMIT
            limit = int(events_fetch_limit)
            if limit > MAX_ALLOWED_FETCH_LIMIT:
                demisto.debug(
                    f"[Fetch Events] Got {limit=} larger than {MAX_ALLOWED_FETCH_LIMIT=}, "
                    f"setting limit to {MAX_ALLOWED_FETCH_LIMIT}."
                )
                limit = MAX_ALLOWED_FETCH_LIMIT
            if limit < page_size:
                demisto.debug(f"[Fetch Events] Got {limit=} lower than {page_size=}, lowering page_size to {limit}.")
                page_size = limit
            demisto.info(f"[Fetch Events] Fetching with fetch_limit={limit}, page_size={page_size}.")
            should_skip_decode_events = params.get("should_skip_decode_events", False)
            should_fail = False
            page_counter = 0
            total_events_count = 0
            offset = None
            auto_trigger_next_run = False
            for events, offset, total_events_count, auto_trigger_next_run in (  # noqa: B007
                fetch_events_command(
                    client,
                    params.get("fetchTime", "5 minutes"),
                    fetch_limit=limit,
                    config_ids=config_ids,
                    ctx=get_integration_context() or {},
                    page_size=page_size,
                    should_skip_decode_events=should_skip_decode_events,
                )
            ):
                if events:
                    page_counter += 1
                    post_latest_event_time(
                        latest_event=events[-1], base_msg=f"Sending {len(events)} events to xsiam using streaming send."
                    )
                    data_size = len(events)
                    should_fail = False
                    try:
                        send_events_to_xsiam(
                            events,
                            VENDOR,
                            PRODUCT,
                            should_update_health_module=False,
                            chunk_size=SEND_EVENTS_TO_XSIAM_CHUNK_SIZE,
                            use_streaming_send=True,
                            data_format="json",
                        )
                    except Exception as e:
                        demisto.error(f"Got an error when executing send_events_to_xsiam: {e}")
                        should_fail = True
                    if should_fail:
                        raise DemistoException(
                            "Encountered an error while sending events to xsiam, will attempt to send all events again."
                        )
                    demisto.debug("Finished executing streaming send_events_to_xsiam.")
                    demisto.debug(
                        f"[Send Events] Done sending {data_size} events to xsiam. "
                        f"Sent {total_events_count} events to xsiam in total during this interval."
                    )
                    events.clear()
            demisto.info(
                f"[Fetch Events] Cycle complete: sent {total_events_count} total events to XSIAM across "
                f"{page_counter} page(s)."
            )
            if not should_fail:
                set_integration_context({"offset": offset})
            demisto.updateModuleHealth({"eventsPulled": (total_events_count or 0)})
            next_run = {}
            if auto_trigger_next_run or total_events_count >= limit:
                demisto.info(
                    f"[Fetch Events] Got {auto_trigger_next_run=} or at least {limit} events this interval - "
                    f"setting nextTrigger=0."
                )
                next_run["nextTrigger"] = "0"
            else:
                demisto.debug(
                    f"[Fetch Events] Got less than {limit} events this interval - will not trigger next run automatically."
                )
            demisto.setLastRun(next_run)
        elif command == "long-running-execution":
            if params.get("isFetchEvents", False):
                raise DemistoException(
                    "Cannot run both fetch events and long-running command simultaneously.\n"
                    "Please make sure to set either isFetchEvents or longRunning to false in"
                    " the integration configuration."
                )
            page_size = min(int(params.get("beta_page_size", BETA_FETCH_EVENTS_MAX_PAGE_SIZE)), BETA_FETCH_EVENTS_MAX_PAGE_SIZE)
            should_skip_decode_events = params.get("should_skip_decode_events", False)
            max_concurrent_tasks = min(int(params.get("max_concurrent_tasks", 100)), MAX_ALLOWED_CONCURRENT_TASKS)
            demisto.debug("Starting long-running execution.")

            asyncio.run(
                fetch_events_long_running_command(
                    client,
                    from_time=params.get("fetchTime", "5 minutes"),
                    page_size=page_size,
                    config_ids=config_ids,
                    ctx=get_integration_context() or {},
                    should_skip_decode_events=should_skip_decode_events,
                    max_concurrent_tasks=max_concurrent_tasks,
                )
            )

        else:
            if command == "test-module":
                human_readable, entry_context, raw_response = commands[command](client, config_ids)
            else:
                human_readable, entry_context, raw_response = commands[command](client, **demisto.args())
            return_outputs(human_readable, entry_context, raw_response)

    except Exception as e:
        err_msg = f"Error in {INTEGRATION_NAME} Integration [{e}]"
        return_error(err_msg, error=e)


if __name__ in ["__builtin__", "builtins", "__main__"]:  # pragma: no cover
    main()