#!/usr/bin/env python3
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2021 ScyllaDB
import platform
from textwrap import dedent

from sdcm.utils.mp_start import ensure_start_method

ensure_start_method()

from collections import defaultdict
from datetime import datetime, timezone, timedelta, UTC
import json
import os
import re
import sys
import unittest
import logging
import time
import subprocess
import traceback
import pprint
import xml.etree.ElementTree as ET
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from functools import partial, reduce
from typing import List, Literal
from uuid import UUID

import pytest
import click
import yaml
from rich.table import Table
from argus.client.sct.types import EventsInfo, LogLink
from argus.client.base import ArgusClientError
from argus.common.enums import TestStatus
from argus.common.sct_types import RawEventPayload

import sct_sizing
import sct_ssh
import sct_scan_issues
from sdcm.cloud_api_client import ScyllaCloudAPIClient
from sdcm.cluster_cloud import extract_short_test_id_from_name
from sdcm.keystore import KeyStore
from sdcm.localhost import LocalHost
from sdcm.provision import AzureProvisioner
from sdcm.provision.provisioner import VmInstance, VmArch
from sdcm.remote import LOCALRUNNER
from sdcm.nemesis.monkey.runners import SisyphusMonkey
from sdcm.sct_config import AWS_SUPPORTED_REGIONS, SCTConfiguration, init_and_verify_sct_config, available_backends
from sdcm.sct_provision.common.layout import SCTProvisionLayout
from sdcm.sct_provision.instances_provider import provision_sct_resources
from sdcm.sct_runner import (
    AwsSctRunner,
    GceSctRunner,
    AzureSctRunner,
    OciSctRunner,
    get_sct_runner,
    clean_sct_runners,
    update_sct_runner_tags,
    list_sct_runners,
)

from sdcm.utils.ci_tools import get_job_name, get_job_url
from sdcm.utils.decorators import retrying
from sdcm.utils.git import get_git_commit_id, get_git_status_info, clone_repo
from sdcm.utils.trigger_matrix import (
    VERSION_RESOLUTION_STRATEGIES,
    resolve_image_architecture,
    resolve_scylla_version_from_image,
    resolve_to_full_version,
    trigger_matrix as run_trigger_matrix,
)
from sdcm.utils.argus import (
    ReplayOnlyArgusSCTClient,
    argus_offline_collect_events,
    create_proxy_argus_s3_url,
    get_argus_client,
    get_argus_use_tunnel_from_env,
)
from sdcm.utils.aws_kms import AwsKms
from sdcm.utils.azure_region import AzureRegion
from sdcm.utils.cloud_monitor import cloud_report, cloud_qa_report
from sdcm.utils.cloud_monitor.cloud_monitor import cloud_non_qa_report
from sdcm.utils.lint.env_builder import build_env
from sdcm.utils.lint.jenkins_parser import parse_jenkinsfile, discover_pipeline_files
from sdcm.utils.lint.docs_linter import lint_test_metadata
from sdcm.utils.oci_utils import get_scylla_images_by_branch, get_scylla_images_by_version, list_instances_oci
from sdcm.utils.common import (
    S3Storage,
    aws_tags_to_dict,
    create_pretty_table,
    rich_table_to_string,
    get_ami_images,
    get_ami_images_versioned,
    get_gce_images,
    get_gce_images_versioned,
    gce_meta_to_dict,
    get_builder_by_test_id,
    list_clusters_eks,
    list_clusters_gke,
    list_elastic_ips_aws,
    list_test_security_groups,
    list_load_balancers_aws,
    list_cloudformation_stacks_aws,
    list_instances_aws,
    list_placement_groups_aws,
    list_instances_gce,
    list_logs_by_test_id,
    list_resources_docker,
    search_test_id_in_latest,
    get_latest_scylla_release,
    images_dict_in_json_format,
    get_hdr_tags,
    download_and_unpack_logs,
    find_equivalent_ami,
)
from sdcm.nemesis.generator import generate_nemesis_yaml, NemesisJobGenerator
from sdcm.utils.open_with_diff import OpenWithDiff, ErrorCarrier
from sdcm.utils.resources_cleanup import (
    clean_cloud_resources,
    clean_resources_according_post_behavior,
    init_argus_client,
)
from sdcm.utils.net import get_sct_runner_ip
from sdcm.utils.jepsen import JepsenResults
from sdcm.utils.docker_utils import docker_hub_login, running_in_podman
from sdcm.monitorstack.restore import (
    restore_monitoring_stack,
    get_monitoring_stack_services,
    kill_running_monitoring_stack_services,
)
from sdcm.utils.log import setup_stdout_logger, disable_loggers_during_startup
from sdcm.utils.aws_region import AwsRegion
from sdcm.utils.aws_builder import AwsCiBuilder, AwsBuilder, AwsFipsCiBuilder
from sdcm.utils.gce_region import GceRegion
from sdcm.utils.gce_builder import GceBuilder
from sdcm.utils.oci_region import OciRegion
from sdcm.utils.oci_builder import OciBuilder
from sdcm.utils.aws_peering import AwsVpcPeering
from sdcm.utils.oci_peering import OciVcnPeering
from sdcm.utils.get_username import get_username
from sdcm.utils.sct_cmd_helpers import add_file_logger, CloudRegion, get_test_config, get_all_regions
from sdcm.utils.aws_utils import AwsArchType, get_arch_from_instance_type
from sdcm.utils.aws_okta import try_auth_with_okta
from sdcm.utils.gce_utils import SUPPORTED_PROJECTS, gce_public_addresses
from sdcm.utils.context_managers import environment
from sdcm.cluster_k8s import mini_k8s
from sdcm.utils.version_utils import get_s3_scylla_repos_mapping, parse_scylla_version_tag
import sdcm.provision.azure.utils as azure_utils
from utils.build_system.create_test_release_jobs import JenkinsPipelines
from utils.build_system.throttle_categories import ThrottleCategoryManager, ThrottleCategory
from utils.get_supported_scylla_base_versions import UpgradeBaseVersion, fetch_official_supported_versions
from sdcm.utils.docker_utils import get_ip_address_of_container
from sdcm.utils.hdrhistogram import make_hdrhistogram_summary_by_interval
from unit_tests.unit.nemesis.fake_cluster import FakeTester
from sdcm.logcollector import Collector

SUPPORTED_CLOUDS = (
    "aws",
    "gce",
    "azure",
    "oci",
)
DEFAULT_CLOUD = SUPPORTED_CLOUDS[0]

SCT_RUNNER_HOST = get_sct_runner_ip()

LOGGER = setup_stdout_logger()


def install_callback(ctx, _, value):
    if not value or ctx.resilient_parsing:
        return value
    LOGGER.info("install-bash-completion current path: %s", os.getcwd())
    shell, path = "bash", Path.home() / ".bash_completion"
    path.write_text((Path(__file__).parent / "utils" / ".bash_completion").read_text())
    click.echo("%s completion installed in %s" % (shell, path))
    return sys.exit(0)


def install_package_from_dir(ctx, _, directories):
    if directories or not ctx.resilient_parsing:
        for directory in directories:
            subprocess.check_call(["sudo", sys.executable, "-m", "pip", "install", directory])
    return directories


def cloud_provider_option(
    function=None, default: str | None = DEFAULT_CLOUD, required: bool = True, help: str = "Cloud provider"
):
    def actual_decorator(func):
        return click.option(
            "-c",
            "--cloud-provider",
            required=required,
            type=click.Choice(choices=SUPPORTED_CLOUDS, case_sensitive=False),
            default=default,
            is_eager=True,
            help=help,
        )(func)

    if function:
        return actual_decorator(function)
    return actual_decorator


class SctLoader(unittest.TestLoader):
    def getTestCaseNames(self, testCaseClass):
        test_cases = super().getTestCaseNames(testCaseClass)
        num_of_cases = len(test_cases)
        assert num_of_cases < 2, (
            f"SCT expect only one test case to be selected, found {num_of_cases}:\n{pprint.pformat(test_cases)}"
        )
        return test_cases


@click.group()
@click.option(
    "--install-bash-completion",
    is_flag=True,
    callback=install_callback,
    expose_value=False,
    help="Install completion for the current shell. Make sure to have psutil installed.",
)
@click.option(
    "--install-package-from-directory",
    callback=install_package_from_dir,
    multiple=True,
    envvar="PACKAGES_PATHS",
    type=click.Path(),
    expose_value=False,
    help="Install paths for extra python packages to install, scylla-cluster-plugins for example",
)
@click.pass_context
def cli(ctx):
    disable_loggers_during_startup()
    # Ugly way of filtering the few command that do not require OKTA verification
    if ctx.invoked_subcommand not in (
        "update-conf-docs",
        "conf-docs",
        "nemesis-list",
        "create-nemesis-pipelines",
        "create-nemesis-yaml",
        "pre-commit",
        "unit-tests",
        "lint-pipelines",
        "trigger-matrix",
    ):
        try_auth_with_okta()

        key_store = KeyStore()
        key_store.sync(
            keys=["scylla_test_id_ed25519"],
            local_path=Path("~/.ssh/").expanduser(),
            permissions=0o0600,
        )

        docker_hub_login(remoter=LOCALRUNNER)


def _report_provision_error_to_argus(backend: str, exc: Exception):
    """Report a provision-stage failure to Argus with TEST_ERROR status."""
    test_config = get_test_config()
    try:
        params = SCTConfiguration()
        test_id = params.get("test_id")
        if test_id and test_id != "None":
            test_config.set_test_id_only(test_id)
        test_config.init_argus_client(params)
    except Exception:  # noqa: BLE001
        LOGGER.warning("Failed to initialize Argus client for error reporting", exc_info=True)
        return

    error_message = (
        f"(TestFrameworkEvent Severity.CRITICAL) Failed to provision {backend} resources: {type(exc).__name__}: {exc}"
    )
    event_payload: RawEventPayload = {
        "run_id": str(test_config.test_id()),
        "severity": "CRITICAL",
        "ts": time.time(),
        "message": error_message,
        "event_type": "TestFrameworkEvent",
        "received_timestamp": None,
        "nemesis_name": None,
        "duration": None,
        "node": None,
        "target_node": None,
        "known_issue": None,
        "nemesis_status": None,
    }

    try:
        test_config.argus_client().submit_event(event_payload)
        LOGGER.info("Error event submitted to Argus successfully")
    except Exception as argus_exc:  # noqa: BLE001
        LOGGER.warning("Failed to submit error event to Argus: %s", argus_exc)

    try:
        test_config.argus_client().set_sct_run_status(TestStatus.TEST_ERROR)
    except Exception:  # noqa: BLE001
        LOGGER.warning("Failed to set TEST_ERROR status in Argus", exc_info=True)


@cli.command("provision-resources", help="Provision resources for the test")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option("-t", "--test-name", type=str, help="Test name")
@click.option(
    "-c",
    "--config",
    multiple=True,
    type=click.Path(exists=True),
    help="Test config .yaml to use, can have multiple of those",
)
def provision_resources(backend, test_name: str, config: str):
    if config:
        os.environ["SCT_CONFIG_FILES"] = str(list(config))
    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    add_file_logger()

    params = None
    test_config = None

    try:
        params = init_and_verify_sct_config()
        test_config = get_test_config()
        test_config.set_test_name(test_name)
        test_id = params.get("test_id")
        if not test_id or test_id == "None":
            raise ValueError("No test_id was provided. Aborting provisioning.")
        test_config.set_test_id_only(test_id)
        localhost = LocalHost(user_prefix=params.get("user_prefix"), test_id=test_config.test_id())

        if params.get("logs_transport") == "syslog-ng":
            click.echo("Provision syslog-ng logging service")
            test_config.configure_syslogng(localhost)
        elif params.get("logs_transport") == "vector":
            click.echo("Provision vector logging service")
            test_config.configure_vector(localhost)
        else:
            click.echo("No need provision logging service")

        agent_config = params.get("agent")
        if agent_config.get("enabled") and agent_config.get("binary_url"):
            click.echo("Generate SCT agent API key")
            test_config.generate_and_save_agent_api_key()

        original_region = " ".join(params.region_names) if params.region_names else None
        handoff_test_id = params.get("reuse_cluster") or test_id

        click.echo(f"Provision {backend} cloud resources")
        if backend == "aws":
            SCTProvisionLayout(params=params).provision()
            test_config.persist_resolved_placement_if_changed(
                handoff_test_id,
                original_region=original_region,
                region_name=" ".join(params.region_names) if params.region_names else None,
                availability_zone=params.get("availability_zone"),
                amis={key: params.get(key) for key in params.ami_id_params if params.get(key)},
            )
        elif backend in ("azure", "gce", "oci"):
            # GCE region fallback + placement handoff live inside provision_sct_resources now, so all
            # three backends share the same entry point (Azure/OCI simply have no fallback yet).
            provision_sct_resources(params=params, test_config=test_config)
        elif backend == "xcloud":
            cloud_provider = params.get("xcloud_provider").lower()
            original_backend = params.get("cluster_backend")
            # as 'xcloud' backend requires provisioning on a cloud provider, we need temporarily set the
            # 'cluster_backend' SCT config parameter to match the provider selected for cloud cluster.
            # This is only needed when provisioning resources is executed as a separate step of a test run
            if cloud_provider == "aws":
                params.update({"cluster_backend": "aws"})
                try:
                    SCTProvisionLayout(params=params).provision()
                    test_config.persist_resolved_placement_if_changed(
                        handoff_test_id,
                        original_region=original_region,
                        region_name=" ".join(params.region_names) if params.region_names else None,
                        availability_zone=params.get("availability_zone"),
                        amis={key: params.get(key) for key in params.ami_id_params if params.get(key)},
                    )
                finally:
                    params.update({"cluster_backend": original_backend})
        else:
            raise ValueError(f"backend {backend} is not supported")
    except Exception as exc:
        LOGGER.error("Unable to provision resources - aborting the test...", exc_info=True)
        _report_provision_error_to_argus(exc, params=params, test_config=test_config, backend=backend)
        sys.exit(1)


def _report_provision_error_to_argus(
    exc: Exception,
    params: "SCTConfiguration | None",
    test_config: object | None,
    backend: str,
) -> None:
    """Report a provisioning error to Argus.

    Handles scenarios where params or test_config may not be available yet
    (e.g., when init_and_verify_sct_config() itself fails).
    """
    if test_config is None:
        test_config = get_test_config()

    # Determine test_id: prefer params, fall back to environment variable
    test_id = None
    if params:
        test_id = params.get("test_id")
    if not test_id or test_id == "None":
        test_id = os.environ.get("SCT_TEST_ID")

    if not test_id or test_id == "None":
        LOGGER.warning("No test_id available - cannot report error to Argus")
        return

    test_config.set_test_id_only(test_id)

    if params:
        # params available — use standard init which checks enable_argus
        try:
            test_config.init_argus_client(params)
        except Exception as init_exc:  # noqa: BLE001
            LOGGER.warning("Failed to initialize Argus client: %s", init_exc)
            return
        argus_client = test_config.argus_client()
    else:
        # params not available (config init failed) — create client directly
        # bypassing enable_argus check since we must report the critical error
        try:
            argus_client = get_argus_client(run_id=test_id, use_tunnel=get_argus_use_tunnel_from_env())
        except Exception as init_exc:  # noqa: BLE001
            LOGGER.warning("Failed to initialize Argus client: %s", init_exc)
            return

    # Create and submit error event to Argus
    error_message = (
        f"(TestFrameworkEvent Severity.CRITICAL) Failed to provision {backend} resources: {type(exc).__name__}: {exc}"
    )
    event_payload: RawEventPayload = {
        "run_id": str(test_id),
        "severity": "CRITICAL",
        "ts": time.time(),
        "message": error_message,
        "event_type": "TestFrameworkEvent",
        "received_timestamp": None,
        "nemesis_name": None,
        "duration": None,
        "node": None,
        "target_node": None,
        "known_issue": None,
        "nemesis_status": None,
    }

    try:
        argus_client.submit_event(event_payload)
        LOGGER.info("Error event submitted to Argus successfully")
    except Exception as argus_exc:  # noqa: BLE001
        LOGGER.warning("Failed to submit error event to Argus: %s", argus_exc)

    # Submit events summary so the error appears in the Argus events tab
    try:
        events_summary = [EventsInfo(severity="CRITICAL", total_events=1, messages=[error_message])]
        argus_client.submit_events(events_summary)
        LOGGER.info("Events summary submitted to Argus successfully")
    except Exception as summary_exc:  # noqa: BLE001
        LOGGER.warning("Failed to submit events summary to Argus: %s", summary_exc)

    try:
        argus_client.set_sct_run_status(TestStatus.TEST_ERROR)
    except Exception as status_exc:  # noqa: BLE001
        LOGGER.warning("Failed to set Argus run status: %s", status_exc)

    try:
        argus_client.finalize_sct_run()
    except Exception as finalize_exc:  # noqa: BLE001
        LOGGER.warning("Failed to finalize Argus run: %s", finalize_exc)


@cli.command("clean-aws-kms-aliases", help="clean AWS KMS old aliases")
@click.option("-r", "--regions", type=CloudRegion(cloud_provider="aws"), multiple=True, help="List of regions to cover")
@click.option("--time-delta-h", type=int, required=False, help="Time delta in hours. Used to detect 'old' aliases.")
@click.option("--dry-run", is_flag=True, default=False, help="Only show result of search not deleting aliases")
@click.pass_context
def clean_aws_kms_aliases(ctx, regions, time_delta_h, dry_run):
    """Clean AWS KMS old aliases."""
    add_file_logger()
    regions = regions or AWS_SUPPORTED_REGIONS
    aws_kms, kwargs = AwsKms(region_names=regions), {"dry_run": dry_run}
    if time_delta_h:
        kwargs["time_delta_h"] = time_delta_h
    aws_kms.cleanup_old_aliases(**kwargs)


@cli.command("clean-resources", help="clean tagged instances. Supported clouds: AWS|Azure|GCE|OCI")
@click.option("--post-behavior", is_flag=True, default=False, help="clean all resources according to post behavior")
@click.option("--user", type=str, help="user name to filter instances by")
@click.option("--billing-project", type=str, help="billing project to filter instances by")
@click.option("--test-id", "test_id", help="test id to filter by. Could be used multiple times", multiple=True)
@click.option("--logdir", type=str, help="directory with test run")
@click.option("--dry-run", is_flag=True, default=False, help="dry run")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option(
    "--clean-runners",
    is_flag=True,
    default=False,
    help="Include SCT runner instances in cleanup (requires --user or --test-id to avoid accidental deletion of all SCT runners)",
)
@click.pass_context
def clean_resources(ctx, post_behavior, user, billing_project, test_id, logdir, dry_run, backend, clean_runners):
    """Clean cloud resources.

    There are different options how to run clean up:
      - To clean resources for the latest run according to post behavior
        $ hydra clean-resources --post-behavior
      - The same as above but with altered logdir
        $ hydra clean-resources --post-behavior --logdir /path/to/logdir
      - To clean resources for some Test ID according to post behavior (test run status extracted from logdir)
        $ hydra clean-resources --post-behavior --test-id TESTID
      - The same as above but with altered logdir
        $ hydra clean-resources --post-behavior --test-id TESTID --logdir /path/to/logdir
      - To clean resources for the latest run ignoring post behavior
        $ hydra clean-resources
      - The same as above but with altered logdir
        $ hydra clean-resources --logdir /path/to/logdir
      - To clean all resources belong to some Test ID:
        $ hydra clean-resources --test-id TESTID
      - To clean all resources belong to some user:
        $ hydra clean-resources --user vasya.pupkin
      - To clean all resources and SCT runners for a user (similarly can be filtered by test-id):
        $ hydra clean-resources --user vasya.pupkin --clean-runners

    Also you can add --dry-run option to see what should be cleaned.
    """
    add_file_logger()

    if clean_runners and not user and not test_id:
        click.echo(
            "ERROR: --clean-runners requires --user and/or --test-id, to prevent accidentally deleting all SCT runners",
            err=True,
        )
        ctx.exit(1)

    user_param = {}
    if user:
        user_param["RunByUser"] = user
    if billing_project:
        user_param["billing_project"] = billing_project
    if user or test_id or billing_project:
        os.environ["SCT_REGION_NAME"] = os.environ.get("SCT_REGION_NAME", "")
        os.environ["SCT_GCE_DATACENTER"] = os.environ.get("SCT_GCE_DATACENTER", "")
        os.environ["SCT_AZURE_REGION_NAME"] = os.environ.get("SCT_AZURE_REGION_NAME", "")
        os.environ["SCT_OCI_REGION_NAME"] = os.environ.get("SCT_OCI_REGION_NAME", "")

    if not post_behavior and user and not test_id and not logdir and not billing_project:
        click.echo(f"Clean all resources belong to user `{user}'")
        user_param["CreatedBy"] = "SCT"
        params = (user_param,)
    else:
        if not logdir and (post_behavior or not test_id):
            logdir = get_test_config().base_logdir()

        if not test_id and (latest_test_id := search_test_id_in_latest(logdir)):
            click.echo(f"Latest TestId in {logdir} is {latest_test_id}")
            test_id = (latest_test_id,)

        if not test_id and not user_param:
            click.echo(clean_resources.get_help(ctx))
            return

        if post_behavior:
            click.echo(
                f"Clean resources according to post behavior for filters: {user_param} and Test IDs: {test_id or 'N/A'}"
            )
        else:
            click.echo(f"Clean all resources for filters: {user_param} and Test IDs: {test_id or 'N/A'}")

        params = ({"TestId": tid, **user_param} for tid in (test_id or (None,)))

    if backend is None:
        if os.environ.get("SCT_CLUSTER_BACKEND", None) is None:
            os.environ["SCT_CLUSTER_BACKEND"] = "aws"
    else:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    config = SCTConfiguration()
    if post_behavior:
        click.echo(f"Use {logdir} as a logdir")
        clean_func = partial(clean_resources_according_post_behavior, config=config, logdir=logdir)
    else:
        clean_func = partial(clean_cloud_resources, config=config)

    if dry_run:
        click.echo("Make a dry-run")

    for param in params:
        effective_param = {k: v for k, v in (param or {}).items() if v is not None}
        clean_func(effective_param, dry_run=dry_run)
        click.echo(f"Cleanup for the {effective_param} resources has been finished")

        if clean_runners:
            click.echo(f"Cleaning SCT runners for {effective_param}...")
            clean_sct_runners(
                test_status="",
                test_runner_ip=None,
                backend=backend,
                user=effective_param.get("RunByUser"),
                test_id=effective_param.get("TestId"),
                dry_run=dry_run,
                force=True,
            )
            click.echo(f"SCT runner cleanup for {effective_param} has been finished")


@cli.command("list-resources", help="list tagged instances in cloud (AWS/GCE/Azure/XCloud)")
@click.option("--user", type=str, help="user name to filter instances by")
@click.option("--billing-project", type=str, help="billing project to filter instances by")
@click.option("--get-all", is_flag=True, default=False, help="All resources")
@click.option("--get-all-running", is_flag=True, default=False, help="All running resources")
@click.option("--test-id", "test_id", help="test id to filter by")
@click.option("--verbose", is_flag=True, default=False, help="if enable, will log progress")
@click.option(
    "-b",
    "--backend",
    "backend_type",
    type=click.Choice(available_backends + ["all"]),
    default="all",
    help="use specific backend",
)
@click.option(
    "--xcloud-env",
    "xcloud_envs",
    type=str,
    multiple=True,
    default=["lab", "staging", "prod"],
    help="ScyllaDB Cloud environments to check (can be specified multiple times). Defaults to lab, staging, and prod",
)
@click.pass_context
def list_resources(ctx, user, billing_project, test_id, get_all, get_all_running, verbose, backend_type, xcloud_envs):  # noqa: PLR0912, PLR0914, PLR0915
    add_file_logger()

    params = {}

    if user:
        params["RunByUser"] = user
    if billing_project:
        params["billing_project"] = billing_project
    if test_id:
        params["TestId"] = test_id
    if all([not get_all, not get_all_running, not user, not test_id, not billing_project]):
        click.echo(list_resources.get_help(ctx))
        sys.exit(1)

    if get_all_running or user or billing_project:
        os.environ["SCT_REGION_NAME"] = os.environ.get("SCT_REGION_NAME", "")
        os.environ["SCT_GCE_DATACENTER"] = os.environ.get("SCT_GCE_DATACENTER", "")
        os.environ["SCT_AZURE_REGION_NAME"] = os.environ.get("SCT_AZURE_REGION_NAME", "")

    if get_all_running:
        table_header = ["Name", "Region-AZ", "PublicIP", "TestId", "RunByUser", "billing_project", "LaunchTime"]
    else:
        table_header = ["Name", "Region-AZ", "State", "TestId", "RunByUser", "billing_project", "LaunchTime"]

    def list_resources_on_aws():
        click.secho("Checking AWS EC2...", fg="green")
        aws_instances = list_instances_aws(tags_dict=params, running=get_all_running, verbose=verbose)

        if aws_instances:
            aws_table = Table(*table_header, show_lines=False)
            for instance in sorted(aws_instances, key=lambda i: i.get("LaunchTime", "")):
                tags = aws_tags_to_dict(instance.get("Tags"))
                name = tags.get("Name", "N/A")
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                aws_table.add_row(
                    name,
                    instance["Placement"]["AvailabilityZone"],
                    instance.get("PublicIpAddress", "N/A") if get_all_running else instance["State"]["Name"],
                    test_id,
                    run_by_user,
                    billing_project,
                    str(instance["LaunchTime"].ctime()),
                )
            click.echo(rich_table_to_string(aws_table, title="Instances used on AWS"))
        else:
            click.secho("Nothing found for selected filters in AWS!", fg="yellow")

        click.secho("Checking AWS Elastic IPs...", fg="green")
        elastic_ips_aws = list_elastic_ips_aws(tags_dict=params, verbose=verbose)
        if elastic_ips_aws:
            aws_table = Table(
                "AllocationId",
                "PublicIP",
                "TestId",
                "RunByUser",
                "billing_project",
                "InstanceId (attached to)",
                show_lines=False,
            )
            for eip in sorted(elastic_ips_aws, key=lambda e: e.get("AllocationId", "")):
                tags = aws_tags_to_dict(eip.get("Tags"))
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                aws_table.add_row(
                    eip["AllocationId"],
                    eip["PublicIp"],
                    test_id,
                    run_by_user,
                    billing_project,
                    eip.get("InstanceId", "N/A"),
                )
            click.echo(rich_table_to_string(aws_table, title="EIPs used on AWS"))
        else:
            click.secho("No elastic ips found for selected filters in AWS!", fg="yellow")

        click.secho("Checking AWS Security Groups...", fg="green")
        security_groups = list_test_security_groups(tags_dict=params, verbose=verbose)
        if security_groups:
            aws_table = Table("Name", "Id", "TestId", "RunByUser", "billing_project", show_lines=False)
            for group in sorted(security_groups, key=lambda g: g.get("GroupId", "")):
                tags = aws_tags_to_dict(group.get("Tags"))
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                name = tags.get("Name", "N/A")
                aws_table.add_row(name, group["GroupId"], test_id, run_by_user, billing_project)
            click.echo(rich_table_to_string(aws_table, title="SGs used on AWS"))
        else:
            click.secho("No security groups found for selected filters in AWS!", fg="yellow")

        click.secho("Checking AWS Placement Groups...", fg="green")
        placement_groups = list_placement_groups_aws(tags_dict=params, available=get_all_running, verbose=verbose)
        if placement_groups:
            aws_table = Table("Name", "Id", "TestId", "RunByUser", "billing_project", show_lines=False)
            for group in sorted(placement_groups, key=lambda g: g.get("GroupId", "")):
                tags = aws_tags_to_dict(group.get("Tags"))
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                name = tags.get("Name", "N/A")
                aws_table.add_row(name, group["GroupId"], test_id, run_by_user, billing_project)
            click.echo(rich_table_to_string(aws_table, title="SGs used on AWS"))
        else:
            click.secho("No placement groups found for selected filters in AWS!", fg="yellow")

    def list_resources_on_gce():
        for project in SUPPORTED_PROJECTS:
            with environment(SCT_GCE_PROJECT=project):
                click.secho(f"Checking GCE ({project})...", fg="green")
                gce_instances = list_instances_gce(tags_dict=params, running=get_all_running, verbose=verbose)
                if gce_instances:
                    gce_table = Table(*table_header, show_lines=False)
                    for instance in sorted(gce_instances, key=lambda i: i.creation_timestamp or ""):
                        tags = gce_meta_to_dict(instance.metadata)
                        public_ips = gce_public_addresses(instance)
                        public_ips = ", ".join(public_ips) if None not in public_ips else "N/A"
                        gce_table.add_row(
                            instance.name,
                            instance.zone.split("/")[-1],
                            public_ips if get_all_running else instance.status,
                            tags.get("TestId", "N/A") if tags else "N/A",
                            tags.get("RunByUser", "N/A") if tags else "N/A",
                            tags.get("billing_project", "N/A") if tags else "N/A",
                            str(instance.creation_timestamp),
                        )
                    click.echo(rich_table_to_string(gce_table, title="Resources used on GCE"))
                else:
                    click.secho("Nothing found for selected filters in GCE!", fg="yellow")

    def list_resources_on_eks():
        click.secho("Checking EKS...", fg="green")
        eks_clusters = list_clusters_eks(tags_dict=params, verbose=verbose)
        if eks_clusters:
            eks_table = Table(
                "Name", "TestId", "Region", "RunByUser", "billing_project", "CreateTime", show_lines=False
            )
            for cluster in sorted(eks_clusters, key=lambda c: str(c.create_time or "")):
                tags = cluster.metadata
                eks_table.add_row(
                    cluster.name,
                    tags.get("TestId", "N/A") if tags else "N/A",
                    cluster.region_name,
                    tags.get("RunByUser", "N/A") if tags else "N/A",
                    tags.get("billing_project", "N/A") if tags else "N/A",
                    str(cluster.create_time),
                )
            click.echo(rich_table_to_string(eks_table, title="EKS clusters"))
        else:
            click.secho("Nothing found for selected filters in EKS!", fg="yellow")

        click.secho("Checking AWS Load Balancers...", fg="green")
        load_balancers = list_load_balancers_aws(tags_dict=params, verbose=verbose)
        if load_balancers:
            aws_table = Table("Name", "Region", "TestId", "RunByUser", "billing_project", show_lines=False)
            for elb in sorted(load_balancers, key=lambda e: e.get("ResourceARN", "").split(":")[-1]):
                tags = aws_tags_to_dict(elb.get("Tags"))
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                _, _, _, region, _, name = elb["ResourceARN"].split(":")
                aws_table.add_row(
                    name,
                    region,
                    test_id,
                    run_by_user,
                    billing_project,
                )
            click.echo(rich_table_to_string(aws_table, title="ELBs used on AWS"))
        else:
            click.secho("No load balancers found for selected filters in AWS!", fg="yellow")

        click.secho("Checking AWS Cloudformation Stacks ...", fg="green")
        cfn_stacks = list_cloudformation_stacks_aws(tags_dict=params, verbose=verbose)
        if cfn_stacks:
            aws_table = Table("Name", "Region", "TestId", "RunByUser", "billing_project", show_lines=False)
            for stack in sorted(cfn_stacks, key=lambda s: s.get("ResourceARN", "").split(":")[-1]):
                tags = aws_tags_to_dict(stack.get("Tags"))
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                _, _, _, region, _, name = stack["ResourceARN"].split(":")
                aws_table.add_row(
                    name,
                    region,
                    test_id,
                    run_by_user,
                    billing_project,
                )
            click.echo(rich_table_to_string(aws_table, title="Cloudformation Stacks used on AWS"))
        else:
            click.secho("No Cloudformation stacks found for selected filters in AWS!", fg="yellow")

    def list_resources_on_gke():
        click.secho("Checking GKE...", fg="green")
        gke_clusters = list_clusters_gke(tags_dict=params, verbose=verbose)
        if gke_clusters:
            gke_table = Table(
                "Name", "Region-AZ", "TestId", "RunByUser", "billing_project", "CreateTime", show_lines=False
            )
            for cluster in sorted(gke_clusters, key=lambda c: str(c.cluster_info.get("createTime", ""))):
                tags = gce_meta_to_dict(cluster.metadata)
                gke_table.add_row(
                    cluster.name,
                    cluster.zone,
                    tags.get("TestId", "N/A") if tags else "N/A",
                    tags.get("RunByUser", "N/A") if tags else "N/A",
                    tags.get("billing_project", "N/A") if tags else "N/A",
                    str(cluster.cluster_info["createTime"]),
                )
            click.echo(rich_table_to_string(gke_table, title="GKE clusters"))
        else:
            click.secho("Nothing found for selected filters in GKE!", fg="yellow")

    def list_resources_on_docker():
        click.secho("Checking Docker...", fg="green")
        docker_resources = list_resources_docker(tags_dict=params, running=get_all_running, verbose=verbose)

        if any(docker_resources.values()):
            if docker_resources.get("containers"):
                docker_table = Table(
                    "Name",
                    "Builder",
                    "Public IP" if get_all_running else "Status",
                    "TestId",
                    "RunByUser",
                    "billing_project",
                    "Created",
                    show_lines=False,
                )
                for builder_name, docker_containers in docker_resources["containers"].items():
                    for container in docker_containers:
                        container.reload()
                        docker_table.add_row(
                            container.name,
                            builder_name,
                            get_ip_address_of_container(container) if get_all_running else container.status,
                            container.labels.get("TestId", "N/A"),
                            container.labels.get("RunByUser", "N/A"),
                            container.labels.get("billing_project", "N/A"),
                            str(container.attrs.get("Created", "N/A")),
                        )
                click.echo(rich_table_to_string(docker_table, title="Containers used on Docker"))
            if docker_resources.get("images"):
                docker_table = Table(
                    "Name", "Builder", "TestId", "RunByUser", "billing_project", "Created", show_lines=False
                )
                for builder_name, docker_images in docker_resources["images"].items():
                    for image in docker_images:
                        image.reload()
                        for tag in image.tags:
                            docker_table.add_row(
                                tag,
                                builder_name,
                                image.labels.get("TestId", "N/A"),
                                image.labels.get("RunByUser", "N/A"),
                                image.labels.get("billing_project", "N/A"),
                                str(image.attrs.get("Created", "N/A")),
                            )
                click.echo(rich_table_to_string(docker_table, title="Images used on Docker"))
        else:
            click.secho("Nothing found for selected filters in Docker!", fg="yellow")

    def list_resources_on_azure():
        click.secho("Checking Azure instances...", fg="green")
        instances: List[VmInstance] = []
        for provisioner in AzureProvisioner.discover_regions(params.get("TestId", "")):
            instances += provisioner.list_instances()
        # Filter by user and billing_project using params dict values
        if params.get("RunByUser"):
            instances = [inst for inst in instances if inst.tags.get("RunByUser") == params["RunByUser"]]
        if params.get("billing_project"):
            instances = [inst for inst in instances if inst.tags.get("billing_project") == params["billing_project"]]
        if instances:
            azure_table = Table(
                "Name",
                "Region-AZ",
                "PublicIP",
                "TestId",
                "RunByUser",
                "billing_project",
                "LaunchTime",
                show_lines=False,
            )

            for instance in sorted(instances, key=lambda i: i.tags.get("RunByUser", "")):
                creation_time = (
                    instance.creation_time.isoformat(sep=" ", timespec="seconds") if instance.creation_time else "N/A"
                )
                tags = instance.tags
                test_id = tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project_value = tags.get("billing_project", "N/A")
                azure_table.add_row(
                    instance.name,
                    instance.region,
                    str(instance.public_ip_address),
                    test_id,
                    run_by_user,
                    billing_project_value,
                    creation_time,
                )
            click.echo(rich_table_to_string(azure_table, title="Instances used on Azure"))
        else:
            click.secho("Nothing found for selected filters in Azure!", fg="yellow")

    def list_resources_on_oci():
        click.secho("Checking OCI instances...", fg="green")
        oci_instances = list_instances_oci(tags_dict=params, running=get_all_running, verbose=verbose)
        if oci_instances:
            oci_table = Table(*table_header, show_lines=False)
            for instance in sorted(oci_instances, key=lambda i: i.get("LaunchTime", "")):
                tags = instance.freeform_tags or {}
                name = instance.display_name
                region_az = f"{instance.region}-{instance.availability_domain}"

                # Instance state in OCI is capitalized (RUNNING, STOPPED, TERMINATED)
                state = instance.lifecycle_state

                test_id = tags.get("SCT_TEST_ID") or tags.get("TestId", "N/A")
                run_by_user = tags.get("RunByUser", "N/A")
                billing_project = tags.get("billing_project", "N/A")
                launch_time = instance.time_created.strftime("%Y-%m-%d %H:%M:%S")

                oci_table.add_row(
                    name,
                    region_az,
                    instance.public_ip_address if get_all_running else state,
                    test_id,
                    run_by_user,
                    billing_project,
                    launch_time,
                )
            click.echo(rich_table_to_string(oci_table, title="Instances used on OCI"))
        else:
            click.secho("Nothing found for selected filters in OCI!", fg="yellow")

    def list_resources_on_xcloud():
        """List ScyllaDB Cloud clusters across specified environments"""
        # Use environments from command line option
        environments = xcloud_envs

        for environment in environments:
            try:
                click.secho(f"Checking ScyllaDB Cloud ({environment})...", fg="green")
                credentials = KeyStore().get_cloud_rest_credentials(environment)
                api_client = ScyllaCloudAPIClient(api_url=credentials["base_url"], auth_token=credentials["api_token"])
                account_id = api_client.get_account_details().get("accountId")
                clusters = api_client.get_clusters(account_id=account_id, enriched=True)

                if clusters:
                    # Filter by test_id or user if provided
                    filtered_clusters = []
                    for cluster in clusters:
                        # Get cluster details to access tags/metadata
                        cluster_details = api_client.get_cluster_details(
                            account_id=account_id, cluster_id=cluster.get("id"), enriched=True
                        )
                        # Check if cluster matches filters
                        cluster_name = cluster_details.get("clusterName", "")
                        # Filter by test_id if provided
                        if test_id:
                            short_test_id = extract_short_test_id_from_name(cluster_name)
                            if short_test_id and not test_id.startswith(short_test_id):
                                continue
                        if user:
                            if user not in cluster_name:
                                continue

                        filtered_clusters.append(cluster_details)

                    if filtered_clusters:
                        xcloud_table = Table(
                            "Name",
                            "Environment",
                            "Status",
                            "Provider",
                            "TestId",
                            "RunByUser",
                            "billing_project",
                            "CreatedAt",
                            show_lines=False,
                        )

                        for cluster in sorted(filtered_clusters, key=lambda c: str(c.get("createdAt", ""))):
                            cluster_name = cluster.get("clusterName", "N/A")
                            cluster_status = cluster.get("status", "N/A")
                            cloud_provider = cluster.get("cloudProvider", {}).get("name", "N/A")
                            created_at = cluster.get("createdAt", "N/A")

                            # Extract test_id and user from metadata or cluster name
                            short_test_id = extract_short_test_id_from_name(cluster_name) or "N/A"
                            # TODO: Extract cluster_user from cluster name once naming convention includes username,
                            #       or from cluster tags/metadata when API provides user information.
                            cluster_user = "N/A"
                            # TODO: Extract billing_project from cluster tags/metadata when API provides it.
                            cluster_billing_project = "N/A"

                            xcloud_table.add_row(
                                cluster_name,
                                environment,
                                cluster_status,
                                cloud_provider,
                                short_test_id,
                                cluster_user,
                                cluster_billing_project,
                                str(created_at),
                            )

                        click.echo(rich_table_to_string(xcloud_table, title=f"ScyllaDB Cloud clusters ({environment})"))
                    else:
                        click.secho(
                            f"Nothing found for selected filters in ScyllaDB Cloud ({environment})!", fg="yellow"
                        )
                else:
                    click.secho(f"No clusters found in ScyllaDB Cloud ({environment})!", fg="yellow")

            except Exception as exc:  # noqa: BLE001
                click.secho(f"Failed to list resources in ScyllaDB Cloud ({environment}): {exc}", fg="red")

    backend_listing_map = {
        "aws": list_resources_on_aws,
        "gce": list_resources_on_gce,
        "oci": list_resources_on_oci,
        "k8s-gke": list_resources_on_gke,
        "k8s-eks": list_resources_on_eks,
        "docker": list_resources_on_docker,
        "azure": list_resources_on_azure,
        "xcloud": list_resources_on_xcloud,
    }
    if list_resources_per_backend_type := backend_listing_map.get(backend_type):
        list_resources_per_backend_type()
    else:
        for list_resources_per_backend_type in backend_listing_map.values():
            list_resources_per_backend_type()


@cli.command("list-images", help="List machine images")
@cloud_provider_option(default="aws", required=False, help="Cloud provided to query. Defaults to aws.")
@click.option(
    "-br",
    "--branch",
    type=str,
    help="Branch to query images for. Defaults to 'master:latest' Mutually exclusive with --version.",
)
@click.option(
    "-v",
    "--version",
    type=str,
    help="List images by version. Use '-v all' for all versions. "
    "OSS format: <4.3> Enterprise format: <enterprise-2021.1>. Mutually exclusive with --branch.",
)
@click.option("-r", "--region", "regions", type=CloudRegion(), help="Cloud region to query images in", multiple=True)
@click.option(
    "-a",
    "--arch",
    type=click.Choice([str(v) for v in VmArch]),
    default=str(VmArch.X86),
    show_choices=True,
    show_default=True,
    help="architecture of the AMI",
)
@click.option("-o", "--output-format", type=click.Choice(["table", "json"]), default="table", help="")
def list_images(  # noqa: PLR0912, PLR0914
    cloud_provider: str,
    branch: str,
    version: str,
    regions: List[str],
    arch: str,
    output_format: Literal["table", "json"] = "table",
):
    if len(regions) == 0:
        regions = [NemesisJobGenerator.BACKEND_TO_REGION[cloud_provider]]
    add_file_logger()

    # Convert arch string to VmArch enum using built-in enum value constructor
    arch_enum = VmArch(arch)

    version_fields = ["Backend", "Name", "ImageId", "CreationDate", "ScyllaVersion"]
    version_fields_aws = ["Backend", "Name", "ImageId", "CreationDate", "NameTag", "ScyllaVersion"]

    branch_fields = ["Backend", "Name", "ImageId", "CreationDate", "BuildId", "Arch", "ScyllaVersion"]
    branch_fields_aws = [
        "Backend",
        "Name",
        "ImageId",
        "CreationDate",
        "NameTag",
        "BuildId",
        "Arch",
        "ScyllaVersion",
        "OwnerId",
    ]
    if version and branch:
        click.echo("Use --version or --branch, not both.")
        return

    branch = branch or "master:latest"

    for region in regions:
        if version is not None:
            match cloud_provider:
                case "aws":
                    rows = get_ami_images_versioned(region_name=region, arch=arch_enum, version=version)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=rows, field_names=version_fields_aws),
                                title=f"AWS Machine Images by Version in region {region}",
                            )
                        )
                    elif output_format == "json":
                        ami_images_json = images_dict_in_json_format(rows=rows, field_names=version_fields_aws)
                        click.echo(ami_images_json)
                case "gce":
                    rows = get_gce_images_versioned(version=version, arch=arch_enum)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=rows, field_names=version_fields),
                                title="GCE Machine Images by version",
                            )
                        )
                    elif output_format == "json":
                        gce_images_json = images_dict_in_json_format(rows=rows, field_names=version_fields)
                        click.echo(gce_images_json)
                case "azure":
                    # Check if this is a full version tag (e.g., "2026.1.0~dev-0.20260124.edda66886e94")
                    if parse_scylla_version_tag(version):
                        # Full version tag: use get_scylla_images for exact matching in private galleries
                        azure_images = azure_utils.get_scylla_images(
                            scylla_version=version, region_name=region, arch=arch_enum
                        )
                        rows = []
                        for image in azure_images:
                            rows.append(
                                [
                                    "Azure",
                                    image.name,
                                    image.id,
                                    "N/A",
                                    image.tags.get("scylla_version", "N/A"),
                                ]
                            )
                    else:
                        # Simple version: use get_released_scylla_images for community gallery
                        azure_images = azure_utils.get_released_scylla_images(
                            scylla_version=version, region_name=region, arch=arch_enum
                        )
                        rows = []
                        for image in azure_images:
                            rows.append(
                                [
                                    "Azure",
                                    image.name,
                                    image.unique_id,
                                    "N/A",
                                    image.artifact_tags.get("scylla_version", "N/A"),
                                ]
                            )

                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=rows, field_names=version_fields),
                                title="Azure Machine Images by version",
                            )
                        )
                    elif output_format == "json":
                        azure_images_json = images_dict_in_json_format(rows=rows, field_names=version_fields)
                        click.echo(azure_images_json)

                case "oci":
                    rows = get_scylla_images_by_version(version=version, region=region, arch=arch_enum)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=rows, field_names=version_fields),
                                title=f"OCI Machine Images by version in region {region}",
                            )
                        )
                    elif output_format == "json":
                        click.echo(images_dict_in_json_format(rows=rows, field_names=version_fields))

                case _:
                    click.echo(f"Cloud provider {cloud_provider} is not supported")

        elif branch:
            if ":" not in branch:
                branch += ":all"

            match cloud_provider:
                case "aws":
                    ami_images = get_ami_images(branch=branch, region=region, arch=arch_enum)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=ami_images, field_names=branch_fields_aws),
                                title=f"AMI Machine Images for {branch} in region {region}",
                            )
                        )
                    elif output_format == "json":
                        ami_images_json = images_dict_in_json_format(rows=ami_images, field_names=branch_fields_aws)
                        click.echo(ami_images_json)
                case "gce":
                    gce_images = get_gce_images(branch=branch, arch=arch_enum)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=gce_images, field_names=branch_fields),
                                title=f"GCE Machine Images for {branch}",
                            )
                        )
                    elif output_format == "json":
                        gce_images_json = images_dict_in_json_format(rows=gce_images, field_names=branch_fields)
                        click.echo(gce_images_json)
                case "azure":
                    azure_images = azure_utils.get_scylla_images(
                        scylla_version=branch, region_name=region, arch=arch_enum
                    )
                    rows = []
                    for image in azure_images:
                        rows.append(["Azure", image.name, image.id, "N/A", image.tags.get("scylla_version", "N/A")])

                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=rows, field_names=version_fields),
                                title="Azure Machine Images by version",
                            )
                        )
                    elif output_format == "json":
                        azure_images_json = images_dict_in_json_format(rows=rows, field_names=version_fields)
                        click.echo(azure_images_json)
                case "oci":
                    oci_images = get_scylla_images_by_branch(branch=branch, region=region, arch=arch_enum)
                    if output_format == "table":
                        click.echo(
                            rich_table_to_string(
                                create_pretty_table(rows=oci_images, field_names=branch_fields),
                                title=f"OCI Machine Images for {branch} in region {region}",
                            )
                        )
                    elif output_format == "json":
                        click.echo(images_dict_in_json_format(rows=oci_images, field_names=branch_fields))
                case _:
                    click.echo(f"Cloud provider {cloud_provider} is not supported")


@cli.command("find-ami-equivalent", help="Find equivalent AMI in different region or architecture")
@click.option("--ami-id", required=True, type=str, help="Source AMI ID to find equivalents for")
@click.option("--source-region", required=True, type=str, help="AWS region where source AMI is located")
@click.option(
    "-r",
    "--target-region",
    "target_regions",
    type=str,
    multiple=True,
    help="Target region(s) to search for equivalents. Can be specified multiple times. "
    "If not specified, searches in source region only.",
)
@click.option(
    "-a",
    "--target-arch",
    type=click.Choice(AwsArchType.__args__),
    help="Target architecture (x86_64 or arm64). If not specified, uses same arch as source AMI.",
)
@click.option(
    "-o",
    "--output-format",
    type=click.Choice(["table", "json", "text"]),
    default="table",
    help="Output format: 'table' for human-readable table, 'json' for structured data, or 'text' for AMI IDs only",
)
def find_ami_equivalent(
    ami_id: str,
    source_region: str,
    target_regions: tuple[str, ...],
    target_arch: AwsArchType | None,
    output_format: str,
):
    """Find equivalent AMIs in different regions or architectures based on tags."""
    add_file_logger()

    # Convert tuple to list or None
    target_regions_list = list(target_regions) if target_regions else None

    # Find equivalent AMIs
    results = find_equivalent_ami(
        ami_id=ami_id, source_region=source_region, target_regions=target_regions_list, target_arch=target_arch
    )

    if not results:
        click.echo(f"No equivalent AMIs found for {ami_id}")
        return

    if output_format == "table":
        # Create pretty table output
        field_names = [
            "Region",
            "AMI ID",
            "Name",
            "Architecture",
            "Creation Date",
            "Name Tag",
            "Scylla Version",
            "Build ID",
            "Owner ID",
        ]
        table = Table(*field_names, show_lines=False)

        for result in results:
            table.add_row(
                result["region"],
                result["ami_id"],
                result["name"],
                result["architecture"],
                result["creation_date"],
                result["name_tag"],
                result["scylla_version"],
                result["build_id"][:6] if result["build_id"] else "N/A",
                result["owner_id"],
            )

        title = f"Equivalent AMIs for {ami_id} (source: {source_region})"
        if target_arch:
            title += f" - Target arch: {target_arch}"
        click.echo(rich_table_to_string(table, title=title))

    elif output_format == "json":
        # Create JSON output for pipeline usage
        output = {
            "source_ami_id": ami_id,
            "source_region": source_region,
            "target_arch": target_arch,
            "results": results,
        }
        click.echo(json.dumps(output, indent=2))

    elif output_format == "text":
        # Output only AMI IDs, one per line
        for result in results:
            click.echo(result["ami_id"])


@cli.command("list-repos", help="List repos url of Scylla formal versions")
@click.option(
    "-d", "--dist-type", type=click.Choice(["centos", "ubuntu", "debian"]), default="centos", help="Distribution type"
)
@click.option(
    "-v",
    "--dist-version",
    type=click.Choice(
        [
            "xenial",
            "trusty",
            "bionic",
            "focal",  # Ubuntu
            "jessie",
            "stretch",
            "buster",
            "bullseye",
        ]
    ),  # Debian
    default=None,
    help="deb style versions",
)
def list_repos(dist_type, dist_version):
    add_file_logger()

    if not dist_type == "centos" and dist_version is None:
        click.secho("when passing --dist-type=debian/ubuntu need to pass --dist-version as well", fg="red")
        sys.exit(1)

    repo_maps = get_s3_scylla_repos_mapping(dist_type, dist_version)

    tbl = Table("Version Family", "Repo Url", show_lines=False)

    for version_prefix, repo_url in repo_maps.items():
        tbl.add_row(version_prefix, repo_url)

    click.echo(rich_table_to_string(tbl, title="Scylla Repos"))


@cli.command("get-scylla-base-versions", help="Get Scylla base versions of upgrade")
@click.option("-s", "--scylla-version", type=str, help="Scylla version, eg: 4.5, 2021.1")
@click.option("-r", "--scylla-repo", type=str, help="Scylla repo")
@click.option("-d", "--linux-distro", type=str, help="Linux Distribution type")
@click.option("-o", "--only-print-versions", type=bool, default=False, required=False, help="")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option(
    "--base_version_all_sts_versions",
    is_flag=True,
    default=False,
    help="Whether to include all supported STS versions as base versions",
)
def get_scylla_base_versions(
    scylla_version, scylla_repo, linux_distro, only_print_versions, backend, base_version_all_sts_versions
):
    """
    Upgrade test try to upgrade from multiple supported base versions, this command is used to
    get the base versions according to the scylla repo and distro type, then we don't need to hardcode
    the base version for each branch.
    """
    add_file_logger()

    with Path("defaults/test_default.yaml").open(mode="r", encoding="utf-8") as test_defaults_yaml:
        test_defaults = yaml.safe_load(test_defaults_yaml)

    if not linux_distro or linux_distro == "null":
        linux_distro = test_defaults.get("scylla_linux_distro")

    version_detector = UpgradeBaseVersion(scylla_repo, linux_distro, scylla_version, base_version_all_sts_versions)

    if not version_detector.dist_type == "centos" and version_detector.dist_version is None:
        click.secho("when passing --dist-type=debian/ubuntu need to pass --dist-version as well", fg="red")
        sys.exit(1)

    # We can't detect the support versions for this distro, which shares the repo with others, eg: centos8
    # so we need to assign the start support versions for it.
    version_detector.set_start_support_version(backend)

    supported_versions, version_list = version_detector.get_version_list()
    click.echo(f"Supported Versions: {supported_versions}")

    if only_print_versions:
        click.echo(f"Base Versions: {' '.join(version_list)}")
        return

    tbl = Table("Version Family", "Repo Url", show_lines=False)
    for version in version_list:
        tbl.add_row(version, version_detector.repo_maps[version])
    click.echo(rich_table_to_string(tbl, title="Base Versions"))
    return


@cli.command("get-official-supported-versions", help="Get officially supported ScyllaDB versions")
@click.option(
    "-o",
    "--only-print-versions",
    type=bool,
    default=False,
    required=False,
    help="Print only the versions as plain output instead of the table view",
)
def get_official_supported_versions(only_print_versions):
    """
    Fetch officially supported ScyllaDB versions from the scylladb-docs-homepage repository.
    Returns versions that have status "Supported" in the upstream supported_versions.json.
    """
    add_file_logger()

    version_list = fetch_official_supported_versions()
    click.echo(f"Official Supported Versions: {' '.join(version_list)}")

    if only_print_versions:
        click.echo(" ".join(version_list))
        return

    tbl = Table("Supported Versions", show_lines=False)
    for version in version_list:
        tbl.add_row(version)
    click.echo(rich_table_to_string(tbl, title="Supported Versions"))
    return


@cli.command("output-conf", help="Output test configuration readed from the file")
@click.argument("config_files", type=str, default="")
@click.option("-b", "--backend", type=click.Choice(available_backends))
def output_conf(config_files, backend):
    add_file_logger()

    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend
    if config_files:
        os.environ["SCT_CONFIG_FILES"] = config_files
    config = SCTConfiguration()
    click.secho(config.dump_config(), fg="green")
    sys.exit(0)


@cli.command("get-db-arch", help="Detect CPU architecture (x86_64 or arm64) for the DB instance type from config")
@click.argument("config_files", type=str, default="")
@click.option("-b", "--backend", type=click.Choice(available_backends))
def get_db_arch(config_files, backend):
    """Print the CPU architecture for the configured DB instance type.

    Uses backend-specific detection for AWS (EC2 API) and Azure (naming convention).
    Other backends (GCE, OCI) always return x86_64 as ARM image support is not
    yet wired through their config resolution.
    Prints 'arm64' or 'x86_64' to stdout.
    """
    add_file_logger()

    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend
    if config_files:
        os.environ["SCT_CONFIG_FILES"] = config_files
    config = SCTConfiguration()

    backend = backend or config.get("cluster_backend")

    if backend == "docker":
        host_arch = "arm64" if platform.machine() == "aarch64" else "x86_64"
        click.echo(host_arch)
        sys.exit(0)

    # Determine the instance type parameter for this backend
    instance_type_key = {
        "aws": "instance_type_db",
        # "gce": "gce_instance_type_db",
        "azure": "azure_instance_type_db",
        # "oci": "oci_instance_type_db",
        "k8s-eks": "instance_type_db",
        # "k8s-gke": "gce_instance_type_db",
    }.get(backend)

    if not instance_type_key:
        # GCE and OCI backends don't support ARM images in SCT yet
        click.echo("x86_64")
        sys.exit(0)

    instance_type = config.get(instance_type_key)
    if not instance_type:
        click.echo("x86_64")
        sys.exit(0)

    # TODO: add handlers for GCE and OCI when available
    if backend in ("aws", "k8s-eks"):
        region_names = config.get("region_names")
        region = region_names[0] if region_names else "us-east-1"
        arch = get_arch_from_instance_type(instance_type, region_name=region)
        click.echo(arch)
        sys.exit(0)
    elif backend == "azure":
        vm_arch = azure_utils.get_arch_from_azure_instance_type(instance_type)
        click.echo("arm64" if vm_arch == VmArch.ARM else "x86_64")
        sys.exit(0)

    click.echo("x86_64")
    sys.exit(0)


def _run_yaml_test(backend, full_path, env):
    output = []
    error = False
    output.append(f"---- linting: {full_path} -----")
    while os.environ:
        os.environ.popitem()
    for key, value in env.items():
        os.environ[key] = value
    os.environ["SCT_CLUSTER_BACKEND"] = backend
    os.environ["SCT_CONFIG_FILES"] = full_path
    logging.getLogger().handlers = []
    logging.getLogger().disabled = True
    try:
        config = SCTConfiguration()
        config.verify_configuration()
        config.check_required_files()
    except Exception as exc:  # noqa: BLE001
        output.append("".join(traceback.format_exception(type(exc), exc, exc.__traceback__)))
        error = True
    return error, output


@cli.command(help="[DEPRECATED] Test yaml in test-cases directory. Use 'lint-pipelines' instead.")
@click.option("-b", "--backend", type=click.Choice(available_backends), default="aws")
@click.option("-i", "--include", type=str, default="")
@click.option("-e", "--exclude", type=str, default="")
def lint_yamls(backend, exclude: str, include: str):
    click.secho(
        "WARNING: 'lint-yamls' is deprecated. Use 'lint-pipelines' instead.\n"
        "  For single-file validation: sct.py lint-pipelines --pipeline-file <path>\n"
        "  For full validation: sct.py lint-pipelines\n",
        fg="yellow",
    )
    if not include:
        raise ValueError("You did not provide include filters")

    exclude_filters = []
    for flt in exclude.split(","):
        if not flt:
            continue
        try:
            exclude_filters.append(re.compile(flt))
        except Exception as exc:  # noqa: BLE001
            raise ValueError(f'Exclude filter "{flt}" compiling failed with: {exc}') from exc

    include_filters = []
    for flt in include.split(","):
        if not flt:
            continue
        try:
            include_filters.append(re.compile(flt))
        except Exception as exc:  # noqa: BLE001
            raise ValueError(f'Include filter "{flt}" compiling failed with: {exc}') from exc

    original_env = {**os.environ}
    process_pool = ProcessPoolExecutor(max_workers=5)

    features = []
    for root, _, files in os.walk("./test-cases"):
        for file in files:
            full_path = os.path.join(root, file)
            if not any((flt.search(file) or flt.search(full_path) for flt in include_filters)):
                continue
            if any((flt.search(file) or flt.search(full_path) for flt in exclude_filters)):
                continue
            features.append(process_pool.submit(_run_yaml_test, backend, full_path, original_env))

    failed = False
    for pp_feature in features:
        error, pp_output = pp_feature.result()
        if error:
            failed = True
            click.secho("\n".join(pp_output), fg="red")
        else:
            click.secho("\n".join(pp_output), fg="green")
    print()
    sys.exit(1 if failed else 0)


def _validate_single_pipeline(pipeline_path, env):
    """Worker function for lint-pipelines ProcessPoolExecutor."""
    # Deferred import: runs in worker process to avoid importing heavy deps in main process
    from sdcm.utils.lint.validator import validate_pipeline  # noqa: PLC0415

    return str(pipeline_path), validate_pipeline(Path(pipeline_path), env)


def _discover_and_parse_pipelines(pipeline_dir, pipeline_file, include_filter, exclude_filter):
    """Discover pipeline files, parse them, and build validation environments."""
    original_env = {**os.environ}

    include_re = re.compile(include_filter) if include_filter else None
    exclude_re = re.compile(exclude_filter) if exclude_filter else None

    if pipeline_file:
        pipeline_files = [Path(pipeline_file)]
    else:
        pipeline_files = discover_pipeline_files(Path(pipeline_dir))

    if include_re:
        pipeline_files = [f for f in pipeline_files if include_re.search(str(f))]
    if exclude_re:
        pipeline_files = [f for f in pipeline_files if not exclude_re.search(str(f))]

    tasks = []
    skipped = 0
    for path in pipeline_files:
        config = parse_jenkinsfile(path)
        if config is None:
            skipped += 1
            continue
        env = build_env(config)
        merged_env = {**original_env, **env}
        tasks.append((str(path), merged_env))

    return tasks, skipped


def _write_junit_xml(junit_xml_path, tasks, failures, total, failed_count, skipped):
    """Write lint-pipelines results as JUnit XML report."""
    testsuite = ET.Element(
        "testsuite", name="lint-pipelines", tests=str(total), failures=str(failed_count), skipped=str(skipped)
    )
    for path_str, _ in tasks:
        testcase = ET.SubElement(testsuite, "testcase", name=path_str, classname="lint-pipelines")
        if path_str in failures:
            failure = ET.SubElement(testcase, "failure", message="Validation failed")
            failure.text = failures[path_str]
    tree = ET.ElementTree(testsuite)
    ET.indent(tree)
    tree.write(junit_xml_path, xml_declaration=True, encoding="unicode")
    click.echo(f"JUnit XML report written to {junit_xml_path}")


@cli.command("lint-pipelines", help="Validate configurations from Jenkins pipeline files")
@click.option("--pipeline-dir", default="jenkins-pipelines", help="Root directory of pipeline files")
@click.option("--pipeline-file", default=None, help="Validate a single pipeline file (ad-hoc mode)")
@click.option("--workers", default=None, type=int, help="Number of parallel workers (default: CPU count)")
@click.option("-i", "--include", "include_filter", default="", help="Regex filter to include specific pipeline files")
@click.option("-e", "--exclude", "exclude_filter", default="", help="Regex filter to exclude specific pipeline files")
@click.option(
    "--junit-xml", "junit_xml_path", default=None, type=click.Path(), help="Write JUnit XML report to this path"
)
def lint_pipelines(pipeline_dir, pipeline_file, workers, include_filter, exclude_filter, junit_xml_path):
    scylla_qa_internal_path = Path(__file__).resolve().parent / "scylla-qa-internal"
    if not scylla_qa_internal_path.exists():
        click.echo("scylla-qa-internal not found, trying to clone...")
        try:
            clone_repo(
                remoter=LOCALRUNNER,
                repo_url="git@github.com:scylladb/scylla-qa-internal.git",
                destination_dir_name=str(scylla_qa_internal_path),
                clone_as_root=False,
                branch="master",
            )
            click.echo("Successfully cloned scylla-qa-internal")
        except Exception as exc:  # noqa: BLE001
            click.echo(
                f"Warning: Could not clone scylla-qa-internal: {exc}. "
                "Pipelines referencing its configs may fail validation."
            )

    tasks, skipped = _discover_and_parse_pipelines(
        pipeline_dir,
        pipeline_file,
        include_filter,
        exclude_filter,
    )

    if not tasks:
        click.echo("No pipeline files to validate.")
        sys.exit(0)

    worker_count = workers or os.cpu_count() or 4
    show_progress = sys.stderr.isatty()

    failed_count = 0
    passed_count = 0
    failures = {}
    chunksize = max(1, len(tasks) // (worker_count * 4))
    paths, envs = zip(*tasks) if tasks else ([], [])
    with ProcessPoolExecutor(max_workers=worker_count) as process_pool:
        results = process_pool.map(_validate_single_pipeline, paths, envs, chunksize=chunksize)

        for path, (is_error, error_msg) in results:
            if is_error:
                failed_count += 1
                failures[path] = error_msg
                click.secho(f"FAIL: {path}", fg="red", bold=True)
                for line in error_msg.strip().splitlines():
                    click.secho(f"  {line}", fg="red")
                click.echo()
            else:
                passed_count += 1
            if show_progress:
                print(
                    f"\r  [{passed_count + failed_count}/{len(tasks)}] validated...",
                    end="",
                    file=sys.stderr,
                    flush=True,
                )

    if show_progress:
        print(file=sys.stderr)

    total = passed_count + failed_count
    click.echo("---")
    summary = f"{passed_count}/{total} pipelines passed"
    if skipped:
        summary += f" ({skipped} skipped)"
    if failed_count:
        summary += f" ({failed_count} failed)"
    click.secho(summary, fg="green" if failed_count == 0 else "red")

    if junit_xml_path:
        _write_junit_xml(junit_xml_path, tasks, failures, total, failed_count, skipped)

    sys.exit(1 if failed_count else 0)


@cli.command("lint-test-docs", help="Validate test_metadata sections in test-case YAML files")
@click.option("--test-case-dir", default="test-cases", help="Root directory of test cases")
@click.option("--missing-only", is_flag=True, help="Only report test cases missing test_metadata")
@click.option("--test-case-file", default=None, type=click.Path(exists=True), help="Validate a single file")
def lint_test_docs(test_case_dir, missing_only, test_case_file):
    taxonomy_path = Path("docs/pipeline-labels/taxonomy.yaml")

    if test_case_file:
        files = [Path(test_case_file)]
    else:
        files = sorted(Path(test_case_dir).rglob("*.yaml"))

    if not files:
        click.echo("No test-case YAML files found.")
        sys.exit(0)

    failed_count = 0
    passed_count = 0
    missing_count = 0

    for path in files:
        result = lint_test_metadata(path, taxonomy_path if taxonomy_path.exists() else None)

        if missing_only:
            if result.missing:
                missing_count += 1
                click.secho(str(path), fg="yellow")
            continue

        if not result.passed:
            failed_count += 1
            click.secho(f"FAIL: {path}", fg="red", bold=True)
            for err in result.errors:
                click.secho(f"  ERROR: {err}", fg="red")
            for warn in result.warnings:
                click.secho(f"  WARN:  {warn}", fg="yellow")
            click.echo()
        else:
            passed_count += 1
            if result.warnings:
                click.secho(f"WARN: {path}", fg="yellow")
                for warn in result.warnings:
                    click.secho(f"  WARN:  {warn}", fg="yellow")
                click.echo()

    if missing_only:
        click.echo(f"{missing_count} test cases missing test_metadata (out of {len(files)} total)")
        sys.exit(0)

    total = passed_count + failed_count
    click.echo("---")
    summary = f"{passed_count}/{total} test cases passed"
    if failed_count:
        summary += f" ({failed_count} failed)"
    click.secho(summary, fg="green" if failed_count == 0 else "red")
    sys.exit(1 if failed_count else 0)


@cli.command(
    "preview-job-description", help="Preview the Jenkins job description that would be generated for a pipeline file"
)
@click.argument("jenkinsfile", type=click.Path(exists=True))
def preview_job_description(jenkinsfile):
    jenkins_file = Path(jenkinsfile)
    text_description = JenkinsPipelines.get_job_description(jenkins_file)

    content = jenkins_file.read_text()
    pipeline_params = {}
    for key in ("backend", "test_name", "test_config", "region"):
        match = re.search(rf"{key}:\s*['\"]([^'\"]+)['\"]", content)
        if match:
            pipeline_params[key] = match.group(1)

    if text_description:
        description = text_description
    elif pipeline_params:
        parts = []
        if "test_name" in pipeline_params:
            parts.append(f"test: {pipeline_params['test_name']}")
        if "backend" in pipeline_params:
            parts.append(f"backend: {pipeline_params['backend']}")
        if "region" in pipeline_params:
            parts.append(f"region: {pipeline_params['region']}")
        if "test_config" in pipeline_params:
            parts.append(f"config: {pipeline_params['test_config']}")
        description = " | ".join(parts)
    else:
        sct_jenkinsfile = (
            jenkins_file.relative_to(Path(__file__).resolve().parent) if jenkins_file.is_absolute() else jenkins_file
        )
        description = str(sct_jenkinsfile)

    metadata_block = JenkinsPipelines.get_metadata_description(jenkins_file)
    if metadata_block:
        description = f"{description}\n\n{metadata_block}"

    if not description.strip():
        click.secho("No description content found.", fg="yellow")
        sys.exit(1)

    click.echo(description)


@cli.command(help="Check test configuration file")
@click.argument("config_file", type=str, default="")
@click.option("-b", "--backend", type=click.Choice(available_backends), default="aws")
def conf(config_file, backend):
    add_file_logger()

    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend
    if config_file:
        os.environ["SCT_CONFIG_FILES"] = config_file
    config = SCTConfiguration()
    try:
        config.verify_configuration()
        config.check_required_files()
    except Exception as ex:
        logging.exception(str(ex))
        click.secho(str(ex), fg="red")
        sys.exit(1)
    else:
        click.secho(config.dump_config(), fg="green")
        sys.exit(0)


@cli.command("conf-docs", help="Show all available configuration in yaml/markdown format")
@click.option(
    "-o", "--output-format", type=click.Choice(["yaml", "markdown"]), default="yaml", help="type of the output"
)
def conf_docs(output_format):
    if output_format == "markdown":
        click.secho(SCTConfiguration.dump_help_config_markdown())
    elif output_format == "yaml":
        click.secho(SCTConfiguration.dump_help_config_yaml())


@cli.command("update-conf-docs", help="Update the docs configuration markdown")
def update_conf_docs():
    markdown_file = Path(__name__).parent / "docs" / "configuration_options.md"
    markdown_file.write_text(SCTConfiguration.dump_help_config_markdown())
    click.secho(f"docs written into {markdown_file}")


@click.group(help="Group of commands for investigating testrun")
def investigate():
    pass


@investigate.command("show-logs", help="Show logs collected for testrun filtered by test-id")
@click.argument("test_id")
@click.option(
    "-o", "--output-format", type=click.Choice(["table", "markdown"]), default="table", help="type of the output"
)
@click.option(
    "--update-argus/--no-update-argus", type=bool, required=False, default=False, help="Update argus with links"
)
def show_log(test_id, output_format, update_argus: bool):
    add_file_logger()

    files = list_logs_by_test_id(test_id)

    if output_format == "table":
        table = Table("Date", "Log type", "Link", show_lines=False)
        for log in files:
            table.add_row(log["date"].strftime("%Y%m%d_%H%M%S"), log["type"], log["link"])
        click.echo(rich_table_to_string(table, title=f"Log links for testrun with test id {test_id}"))
    elif output_format == "markdown":
        click.echo("\n## Logs\n")
        for log in files:
            click.echo(f"* **{log['type']}** - {log['link']}")

    if update_argus:
        try:
            store_logs_in_argus(
                test_id=test_id,
                logs=reduce(lambda acc, log: acc[log["type"]].append(log["s3_url"]) or acc, files, defaultdict(list)),
                update=True,
                use_tunnel=get_argus_use_tunnel_from_env(),
            )
        except Exception:  # noqa: BLE001
            LOGGER.error("Error updating logs in argus.", exc_info=True)


@investigate.command("download-logs", help="Download logs collected for testrun by test-id")
@click.argument("test_id")
@click.option("-d", "--dest-dir", type=click.Path(), default=".", help="Destination directory for downloaded logs")
@click.option("-t", "--log-type", type=str, multiple=True, help="Filter by log type (can be specified multiple times)")
@click.option("--list-types", is_flag=True, help="List available log types and exit")
def download_logs(test_id: str, dest_dir: str, log_type: tuple[str, ...], list_types: bool):
    """Download logs from S3 for a specific test run.

    Examples:
        # Download all logs to current directory
        hydra investigate download-logs abc123

        # Download to specific directory
        hydra investigate download-logs abc123 -d /tmp/logs

        # Download only specific log types
        hydra investigate download-logs abc123 -t sct-runner -t db-cluster

        # List available log types
        hydra investigate download-logs abc123 --list-types
    """
    add_file_logger()

    files = list_logs_by_test_id(test_id)

    if not files:
        click.secho(f"No logs found for test id: {test_id}", fg="yellow")
        return

    if list_types:
        available_types = sorted(set(log["type"] for log in files))
        click.echo("Available log types:")
        for t in available_types:
            click.echo(f"  - {t}")
        return

    # Filter by log type if specified
    if log_type:
        files = [f for f in files if f["type"] in log_type]
        if not files:
            click.secho(f"No logs found matching types: {', '.join(log_type)}", fg="yellow")
            return

    # Create destination directory
    dest_path = os.path.abspath(dest_dir)
    os.makedirs(dest_path, exist_ok=True)

    click.echo(f"Downloading {len(files)} log file(s) to {dest_path}")

    s3_storage = S3Storage()
    downloaded = []
    failed = []

    for log in files:
        log_name = os.path.basename(log["file_path"])
        click.echo(f"  Downloading [{log['type']}] {log_name}...")
        try:
            downloaded_path = s3_storage.download_file(log["s3_url"], dest_path)
            downloaded.append({"type": log["type"], "path": downloaded_path})
            click.secho(f"    -> {downloaded_path}", fg="green")
        except Exception as exc:  # noqa: BLE001
            failed.append({"type": log["type"], "name": log_name, "error": str(exc)})
            click.secho(f"    Failed: {exc}", fg="red")

    click.echo("")
    click.secho(f"Downloaded: {len(downloaded)}/{len(files)} files", fg="green" if not failed else "yellow")

    if failed:
        click.secho(f"Failed: {len(failed)} files", fg="red")
        for f in failed:
            click.echo(f"  - [{f['type']}] {f['name']}: {f['error']}")
        sys.exit(1)


@investigate.command("show-monitor", help="Run monitoring stack with saved data locally")
@click.argument("test_id")
@click.option("--cluster-name", type=str, required=False, help="Cluster name (relevant for multi-tenant test)")
@click.option("--date-time", type=str, required=False, help="Datetime of monitor-set archive is collected")
@click.option("--kill", type=bool, required=False, help="Kill and remove containers")
def show_monitor(test_id, date_time, kill, cluster_name):
    add_file_logger()

    click.echo(f"Search monitoring stack archive files for test id {test_id} and restoring...")
    containers = {}
    try:
        containers = restore_monitoring_stack(test_id, date_time)
    except Exception as details:  # noqa: BLE001
        LOGGER.error(details)

    if not containers:
        click.echo("Errors were found when restoring Scylla monitoring stack")
        kill_running_monitoring_stack_services()
        sys.exit(1)

    for cluster, containers_ports in containers.items():
        if cluster_name and cluster != cluster_name:
            continue

        click.echo(f"Monitoring stack for cluster {cluster} restored")
        table = Table("Service", "Container", "Link", show_lines=False)
        for docker in get_monitoring_stack_services(ports=containers_ports):
            table.add_row(docker["service"], docker["name"], f"http://{SCT_RUNNER_HOST}:{docker['port']}")
        click.echo(rich_table_to_string(table, title=f"Monitoring stack services for cluster {cluster}"))
        click.echo("")
        if kill:
            kill_running_monitoring_stack_services(ports=containers_ports)


@investigate.command("show-jepsen-results", help="Run a server with Jepsen results")
@click.argument("test_id")
def show_jepsen_results(test_id):
    add_file_logger()

    click.secho(message=f"\nSearch Jepsen results archive files for test id {test_id} and restoring...\n", fg="green")
    jepsen = JepsenResults()
    if jepsen.restore_jepsen_data(test_id):
        click.secho(
            message=f"\nJepsen data restored, starting web server on "
            f"http://{SCT_RUNNER_HOST}:{jepsen.jepsen_results_port}/",
            fg="green",
        )
        detach = SCT_RUNNER_HOST != "127.0.0.1"
        if not detach:
            click.secho(message="Press Ctrl-C to stop the server.", fg="green")
        click.echo("")
        jepsen.run_jepsen_web_server(detach=detach)


@investigate.command("search-builder", help="Search builder where test run with test-id located")
@click.argument("test-id")
def search_builder(test_id):
    logging.getLogger("paramiko").setLevel(logging.CRITICAL)
    add_file_logger()

    results = get_builder_by_test_id(test_id)
    tbl = Table("Builder Name", "Public IP", "path", show_lines=False)
    for result in results:
        tbl.add_row(result["builder"]["name"], result["builder"]["public_ip"], result["path"])

    click.echo(rich_table_to_string(tbl, title=f"Found builders for Test-id: {test_id}"))


@investigate.command("show-events", help="Return content of file events_log/events for running job by test-id")
@click.argument("test-id")
@click.option(
    "--follow",
    type=bool,
    required=False,
    is_flag=True,
    default=False,
    help="Follow job events log file (similar tail -f <file>)",
)
@click.option("--last-n", type=int, required=False, help="return last n lines from events.log file")
@click.option("--save-to", type=str, required=False, help="Download events.log file and save to provided dir")
def show_events(test_id: str, follow: bool = False, last_n: int = None, save_to: str = None):
    logging.getLogger("paramiko").setLevel(logging.CRITICAL)
    add_file_logger()
    builders = get_builder_by_test_id(test_id)

    if not builders:
        LOGGER.info("Builder was not found for provided test-id %s", test_id)

    for builder in builders:
        LOGGER.info(
            "Applying action for events.log on builder %s:%s...",
            builder["builder"]["name"],
            builder["builder"]["public_ip"],
        )
        remoter = builder["builder"]["remoter"]

        if follow or last_n:
            options = "-f " if follow else ""
            options += f"-n {last_n} " if last_n else ""
            try:
                remoter.run("tail %s %s/events_log/events.log", options, builder["path"])
            except KeyboardInterrupt:
                LOGGER.info("Monitoring events.log for test-id %s stopped!", test_id)
        elif save_to:
            remoter.receive_files(f"{builder['path']}/events_log/events.log", save_to)
            LOGGER.info("Events saved to %s", save_to)
        else:
            remoter.run(f"cat {builder['path']}/events_log/events.log")
    click.echo("Show events done.")


cli.add_command(investigate)


@cli.command("unit-tests", help="Run all the SCT internal unit-tests")
@click.option(
    "-t", "--test", required=False, default=[""], multiple=True, help="Run specific test file from unit-tests directory"
)
@click.option("-n", required=False, default=4, help="Sets number of parallel tests to run, default is 4")
@click.option("--junit-xml", required=False, default="", help="Path to write JUnit XML report")
def unit_tests(test, n, junit_xml):
    args = ["-v", "-m", "not integration", f"-n{n}", *(f"unit_tests/{t}" for t in test)]
    if junit_xml:
        args.append(f"--junit-xml={junit_xml}")
    sys.exit(pytest.main(args))


@cli.command("integration-tests", help="Run all the SCT internal integration-tests")
@click.option(
    "-t", "--test", required=False, default=[""], multiple=True, help="Run specific test file from unit-tests directory"
)
@click.option("-n", required=False, default=4, help="Sets number of parallel tests to run, default is 4")
@click.option("--junit-xml", required=False, default="", help="Path to write JUnit XML report")
def integration_tests(test, n, junit_xml):
    get_test_config().logdir()
    add_file_logger()

    if not running_in_podman():  # we can't do sudo commands within rootless podman
        # setup prerequisites for the integration test is identical
        # to the kind local functional tests
        # TODO: to refactor setup_prerequisites out of LocalKindCluster
        sct_config = SCTConfiguration()
        local_cluster = mini_k8s.LocalKindCluster(
            software_version="",
            user_prefix="",
            params=sct_config,
        )
        local_cluster.setup_prerequisites()

    args = ["-v", "-m", "integration", "--dist", "loadgroup", f"-n{n}", *(f"unit_tests/{t}" for t in test)]
    if junit_xml:
        args.append(f"--junit-xml={junit_xml}")
    sys.exit(pytest.main(args))


@cli.command("pre-commit", help="Run pre-commit checkers")
def pre_commit():
    result = 0
    target = "origin/$CHANGE_TARGET" if "CHANGE_TARGET" in os.environ else "upstream/master"

    if not sys.stdin.isatty():
        lint_commits = True

    else:
        click.echo(
            dedent("""
            WARNING: This command will temporarily checkout different commits to validate commit messages.
                Any uncommitted changes should be saved before running this command.
                This is primarily intended for CI usage.
        """).strip()
        )
        lint_commits = bool(click.confirm("Do you want to continue?"))

    if lint_commits:
        # Check commit messages by checking out each commit
        result += os.system(f"bash scripts/lint_commits.sh {target}")

    result += os.system("pre-commit run -a --show-diff-on-failure")
    result = 1 if result else 0
    sys.exit(result)


class OutputLogger:
    def __init__(self, filename, terminal):
        self.terminal = terminal
        self.log = open(filename, "a", encoding="utf-8")

    def write(self, message):
        self.terminal.write(message)
        self.log.write(message)

    def flush(self):
        self.terminal.flush()
        self.log.flush()

    def isatty(self):
        return False


@cli.command("run-test", help="Run SCT test using unittest")
@click.argument("argv")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option(
    "-c",
    "--config",
    multiple=True,
    type=click.Path(exists=True),
    help="Test config .yaml to use, can have multiple of those",
)
@click.option("-l", "--logdir", help="Directory to use for logs")
def run_test(argv, backend, config, logdir):
    if config:
        os.environ["SCT_CONFIG_FILES"] = str(list(config))
    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    if logdir:
        os.environ["_SCT_LOGDIR"] = logdir

    os.environ["SCT_TEST_METHOD"] = argv
    logfile = os.path.join(get_test_config().logdir(), "output.log")
    sys.stdout = OutputLogger(logfile, sys.stdout)
    sys.stderr = OutputLogger(logfile, sys.stderr)

    if "::" in argv:
        target = argv
    else:
        test_path = argv.split(".")
        test_path[0] = f"{test_path[0]}.py"
        target = "::".join(test_path)

    if not target:
        print("argv is referring to the directory or file that contain tests, it can't be empty")
        sys.exit(1)
    return_code = pytest.main(["-s", "-vv", "-rN", "-p", "no:logging", target])
    sys.exit(return_code)


@cli.command("run-pytest", help="Run tests using pytest")
@click.argument("target")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option(
    "-c",
    "--config",
    multiple=True,
    type=click.Path(exists=True),
    help="Test config .yaml to use, can have multiple of those",
)
@click.option("-l", "--logdir", help="Directory to use for logs")
def run_pytest(target, backend, config, logdir):
    if config:
        os.environ["SCT_CONFIG_FILES"] = str(list(config))
    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    if logdir:
        os.environ["_SCT_LOGDIR"] = logdir
    _logdir = Path(get_test_config().logdir())
    logfile = _logdir / "output.log"
    junit_file = _logdir / "junit.xml"
    sys.stdout = OutputLogger(logfile, sys.stdout)
    sys.stderr = OutputLogger(logfile, sys.stderr)
    if not target:
        print("argv is referring to the directory or file that contain tests, it can't be empty")
        sys.exit(1)
    return_code = pytest.main(["-s", "-v", f"--junit-xml={junit_file}", target])
    test_config = get_test_config()
    test_config.init_argus_client(params=SCTConfiguration())
    test_config.argus_client().sct_submit_junit_report(file_name=junit_file.name, raw_content=junit_file.read_text())
    sys.exit(return_code)


@cli.command("cloud-usage-report", help="Generate and send Cloud usage report")
@click.option("-e", "--emails", required=True, type=str, help="Comma separated list of emails. Example a@b.com,c@d.com")
@click.option(
    "-t",
    "--report-type",
    required=True,
    type=click.Choice(choices=["general", "last-7-days-qa", "last-7-days-non-qa"], case_sensitive=False),
    help="Type of the report",
)
@click.option(
    "-u",
    "--user",
    required=False,
    type=str,
    default="",
    help="User or instance owner. Applicable for last-7-days-* reports",
)
@click.option(
    "--billing-project", required=False, type=str, default="", help="Billing project tag filter (currently not applied)"
)
def cloud_usage_report(emails, report_type, user, billing_project):
    add_file_logger()

    email_list = emails.split(",")
    click.secho(message=f"Will send {user} Cloud Usage '{report_type}' report to {email_list}", fg="green")
    match report_type:
        case "general":
            cloud_report(mail_to=email_list)
        case "last-7-days-qa":
            cloud_qa_report(mail_to=email_list, user=user)
        case "last-7-days-non-qa":
            cloud_non_qa_report(mail_to=email_list, user=user)
    click.secho(message="Done.", fg="yellow")


@cli.command("collect-logs", help="Collect logs from cluster by test-id")
@click.option("--test-id", help="Find cluster by test-id")
@click.option("--logdir", help="Path to directory with sct results")
@click.option("--backend", help="Cloud where search nodes", default=None)
@click.option("--config-file", type=str, help="config test file path")
def collect_logs(test_id=None, logdir=None, backend=None, config_file=None):
    add_file_logger()

    logging.getLogger("paramiko").setLevel(logging.CRITICAL)
    if backend is None:
        if os.environ.get("SCT_CLUSTER_BACKEND", None) is None:
            os.environ["SCT_CLUSTER_BACKEND"] = backend = "aws"
    else:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    if config_file and not os.environ.get("SCT_CONFIG_FILES", None):
        os.environ["SCT_CONFIG_FILES"] = config_file

    config = SCTConfiguration()

    if not test_id:
        test_id = config.get("test_id")
        if test_id:
            LOGGER.info("Using test_id from SCT configuration: %s", test_id)

    collector = Collector(test_id=test_id, params=config, test_dir=logdir)

    collected_logs, collection_error = collector.run()

    table = Table("Cluster set", "Link", show_lines=False)
    for cluster_type, s3_links in collected_logs.items():
        for link in s3_links:
            current_cluster_type = cluster_type
            # Cover case when archive is created per log file not all logs in one archive.
            # Here log name is extracted from archive name. For example:
            # for link https://cloudius-jenkins-test.s3.amazonaws.com/c63a6913-6253-45a0-b5cf-d553f713fe81/20211222_
            # 101636/warning-c63a6913.log.tar.gz
            #  current_cluster_type will be "warning"
            if cluster_type == "sct-runner" and cluster_type not in link:
                current_cluster_type = link.split("/")[-1].split("-")[0]
            table.add_row(
                current_cluster_type, create_proxy_argus_s3_url(link).format(collector.test_id, link.split("/")[-1])
            )

    click.echo(rich_table_to_string(table, title=f"Collected logs by test-id: {collector.test_id}"))
    update_sct_runner_tags(backend=backend, test_id=collector.test_id, tags={"logs_collected": True})

    # Always send collected logs to Argus, even if there were collection errors
    if collector.test_id:
        store_logs_in_argus(
            test_id=UUID(collector.test_id),
            logs=collected_logs,
            use_tunnel=config.get("argus_use_ssh_tunnel") or False,
            log_dir=os.path.dirname(collector.storage_dir) if collector.storage_dir else None,
        )

    # Raise error only after logs have been sent to Argus
    if collection_error:
        raise RuntimeError(collection_error)


def store_logs_in_argus(
    test_id: UUID,
    logs: dict[str, list[list[str] | str]],
    update: bool = False,
    use_tunnel: bool = False,
    log_dir: str | None = None,
):
    try:
        argus_client = get_argus_client(run_id=test_id, use_tunnel=use_tunnel, log_dir=log_dir)
        log_links = []
        existing_links = [name for [name, _] in argus_client.get_run().get("logs", [])] if update else []
        for log_name, s3_links in logs.items():
            for link in s3_links:
                if not link:
                    LOGGER.warning("Link is missing for log %s", log_name)
                    continue
                file_name = link.split("/")[-1]
                if update and file_name not in existing_links:
                    LOGGER.info("Adding missing log link %s to Argus...", file_name)
                    log_links.append(LogLink(log_name=file_name, log_link=link))
                elif not update:
                    log_links.append(LogLink(log_name=file_name, log_link=link))
                else:
                    LOGGER.info("Skipping %s because it has been already sent to argus.", file_name)
        argus_client.submit_sct_logs(log_links)

        if not argus_client.get_run().get("events"):
            argus_offline_collect_events(client=argus_client)
    except Exception:
        LOGGER.error("Error saving logs to argus", exc_info=True)


@cli.command("send-email", help="Send email with results for testrun")
@click.option("--test-id", help="Test-id of run")
@click.option("--test-status", help="Override test status FAILED|ABORTED")
@click.option("--start-time", help="Override test start time")
@click.option("--started-by", help="Default user that started the test")
@click.option("--runner-ip", type=str, required=False, help="Sct runner ip for the running test")
@click.option("--email-recipients", help="Send email to next recipients")
@click.option("--logdir", help="Directory where to find testrun folder")
def send_email(
    test_id=None,
    test_status=None,
    start_time=None,
    started_by=None,
    runner_ip=None,
    email_recipients=None,
    logdir=None,
):
    add_file_logger()

    if not email_recipients:
        LOGGER.warning("No email recipients. Email will not be sent")
        sys.exit(1)
    LOGGER.info("Email will be sent to next recipients: %s", email_recipients)
    email_recipients = email_recipients.split(",")
    sct_config = SCTConfiguration()

    argus_test_id = os.environ.get("SCT_REUSE_CLUSTER") or os.environ.get("SCT_TEST_ID")
    LOGGER.info("Sending email for test %s...", argus_test_id)
    client = init_argus_client(argus_test_id, use_tunnel=get_argus_use_tunnel_from_env(), log_dir=logdir)
    run = client.get_run()
    title_template_data = {**dict(sct_config), **run}

    template = sct_config.get("argus_email_report_template")
    if not template:
        LOGGER.error("Argus Email Report template file is not defined.")
        sys.exit(1)

    p = Path(f"./argus_report_templates/{template}")
    if not p.exists():
        LOGGER.error("Argus Email Report template does not exist: %s", p)
        sys.exit(1)

    with p.open() as f:
        template = yaml.safe_load(f)

    title: str = (
        template["title"] if ("{" not in template["title"]) else template["title"].format(**title_template_data)
    )
    LOGGER.info("Sending email to %s with title %s and sections: %s", email_recipients, title, template)
    client.send_email(recipients=email_recipients, title=title, sections=template["sections"])


@cli.command("create-operator-test-release-jobs", help="Create pipeline jobs for a new scylla-operator branch/release")
@click.argument("branch", type=str)
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--sct_branch", default="master", type=str)
@click.option("--sct_repo", default="git@github.com:scylladb/scylla-cluster-tests.git", type=str)
@click.option("--triggers/--no-triggers", default=False)
def create_operator_test_release_jobs(branch, username, password, sct_branch, sct_repo, triggers):
    add_file_logger()

    base_job_dir = f"scylla-operator/{branch}"
    server = JenkinsPipelines(
        username=username, password=password, base_job_dir=base_job_dir, sct_branch_name=sct_branch, sct_repo=sct_repo
    )
    server.create_job_tree(
        f"{server.base_sct_dir}/jenkins-pipelines/operator",
        create_freestyle_jobs=triggers,
        template_context={"release_version": get_latest_scylla_release(product="scylla-enterprise")},
    )


@cli.command("create-manager-test-release-jobs", help="Create pipeline jobs for a new scylla-manager branch/release")
@click.argument("branch", type=str)
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--sct_branch", default="master", type=str)
@click.option("--sct_repo", default="git@github.com:scylladb/scylla-cluster-tests.git", type=str)
@click.option("--triggers/--no-triggers", default=False)
def create_manager_test_release_jobs(branch, username, password, sct_branch, sct_repo, triggers):
    add_file_logger()

    server = JenkinsPipelines(
        username=username, password=password, base_job_dir=branch, sct_branch_name=sct_branch, sct_repo=sct_repo
    )
    server.create_job_tree(
        f"{server.base_sct_dir}/jenkins-pipelines/manager",
        create_freestyle_jobs=triggers,
        template_context={"release_version": get_latest_scylla_release(product="scylla-enterprise")},
    )


@cli.command("create-qa-tools-jobs", help="Create pipeline jobs for a new scylla-operator branch/release")
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--sct_branch", default="master", type=str)
@click.option("--sct_repo", default="git@github.com:scylladb/scylla-cluster-tests.git", type=str)
@click.option("--triggers/--no-triggers", default=False)
def create_qa_tools_jobs(username, password, sct_branch, sct_repo, triggers):
    add_file_logger()

    base_job_dir = "QA-tools"
    server = JenkinsPipelines(
        username=username, password=password, base_job_dir=base_job_dir, sct_branch_name=sct_branch, sct_repo=sct_repo
    )
    server.create_job_tree(
        f"{server.base_sct_dir}/jenkins-pipelines/qa", create_freestyle_jobs=triggers, job_name_suffix=""
    )


@cli.command("create-performance-jobs", help="Create pipeline jobs for performance")
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--sct_branch", default="branch-perf-v17", type=str)
@click.option("--sct_repo", default="git@github.com:scylladb/scylla-cluster-tests.git", type=str)
@click.option("--triggers/--no-triggers", default=False)
def create_performance_jobs(username, password, sct_branch, sct_repo, triggers):
    add_file_logger()

    # we start from the root of jenkins, because we have jobs to scylla-master and scylla-enterprise
    base_job_dir = ""
    server = JenkinsPipelines(
        username=username, password=password, base_job_dir=base_job_dir, sct_branch_name=sct_branch, sct_repo=sct_repo
    )
    server.create_job_tree(
        f"{server.base_sct_dir}/jenkins-pipelines/performance/{sct_branch}",
        create_freestyle_jobs=triggers,
        job_name_suffix="",
    )


@cli.command("create-nemesis-yaml")
@click.option("--diff/--no-diff", default=True)
def create_nemesis_yaml(diff):
    error_carrier = ErrorCarrier() if diff else None
    file_opener = partial(OpenWithDiff, error_carrier=error_carrier) if diff else open

    generate_nemesis_yaml(file_opener)
    if error_carrier:
        sys.exit(1)


@cli.command("create-nemesis-pipelines")
@click.option("--base-job", default=None, type=str)
@click.option("--backend", default=NemesisJobGenerator.BACKEND_TO_REGION.keys(), multiple=True)
@click.option("--diff/--no-diff", default=True)
def create_nemesis_pipelines(base_job: str, backend: list[str], diff: bool):
    error_carrier = ErrorCarrier() if diff else None
    file_opener = partial(OpenWithDiff, error_carrier=error_carrier) if diff else open

    gen = NemesisJobGenerator(base_job=base_job, backends=backend, base_dir="", file_opener=file_opener)
    gen.render_base_job_config()
    gen.create_test_cases_from_template()
    gen.create_job_files_from_template()

    if error_carrier:
        sys.exit(1)


@cli.command("create-test-release-jobs", help="Create pipeline jobs for a new branch")
@click.argument("branch", type=str)
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--sct_branch", default="master", type=str)
@click.option("--sct_repo", default="git@github.com:scylladb/scylla-cluster-tests.git", type=str)
def create_test_release_jobs(branch, username, password, sct_branch, sct_repo):
    add_file_logger()

    base_job_dir = f"{branch}"
    server = JenkinsPipelines(
        username=username, password=password, base_job_dir=base_job_dir, sct_branch_name=sct_branch, sct_repo=sct_repo
    )
    base_path = f"{server.base_sct_dir}/jenkins-pipelines/oss"
    server.create_job_tree(base_path)

    scylla_doctor_path = f"{server.base_sct_dir}/jenkins-pipelines/scylla-doctor"
    server.create_job_tree(scylla_doctor_path)

    if branch == "scylla-master":
        base_path = f"{server.base_sct_dir}/jenkins-pipelines/master-triggers"
        server.create_job_tree(base_path)


@cli.command(
    "configure-jenkins-throttle", help="Create/update Jenkins throttle-concurrents categories for perf regions"
)
@click.argument("username", envvar="JENKINS_USERNAME", type=str, required=False)
@click.argument("password", envvar="JENKINS_PASSWORD", type=str, required=False)
@click.option("--dry-run", is_flag=True, default=False, help="Print categories without applying changes")
@click.option(
    "--suffix",
    multiple=True,
    default=["i8g"],
    help="Additional suffixes to create per region (e.g. --suffix i8g --suffix i4i). "
    "Base categories (no suffix) are always created.",
)
def configure_jenkins_throttle(username, password, dry_run, suffix):
    add_file_logger()

    # AWS regions used by performance tests
    aws_regions = [
        "us-east-1",
        "eu-west-1",
        "eu-west-2",
        "eu-west-3",
        "eu-north-1",
    ]

    categories = [ThrottleCategory(name=f"SCT-perf-{region}", max_total=1, max_per_node=1) for region in aws_regions]
    for sfx in suffix:
        categories += [
            ThrottleCategory(name=f"SCT-perf-{region}-{sfx}", max_total=1, max_per_node=1) for region in aws_regions
        ]
    if dry_run:
        click.echo("Would ensure the following throttle categories:")
        for cat in categories:
            click.echo(f"  {cat.name}: max_total={cat.max_total}, max_per_node={cat.max_per_node}")
        return

    manager = ThrottleCategoryManager(username=username, password=password)
    changed = manager.ensure_categories(categories)
    if changed:
        click.echo(f"Created/updated {len(changed)} categories: {changed}")
    else:
        click.echo("All categories already up to date")


@cli.command("prepare-regions", help="Configure all required resources for SCT runs in selected cloud region")
@cloud_provider_option
@click.option("-r", "--regions", type=CloudRegion(), help="Cloud region", multiple=True)
def prepare_regions(cloud_provider, regions):
    add_file_logger()
    regions = regions or get_all_regions(cloud_provider)

    for region in regions:
        if cloud_provider == "aws":
            region = AwsRegion(region_name=region)  # noqa: PLW2901
        elif cloud_provider == "azure":
            region = AzureRegion(region_name=region)  # noqa: PLW2901
        elif cloud_provider == "gce":
            region = GceRegion(region_name=region)  # noqa: PLW2901
        elif cloud_provider == "oci":
            region = OciRegion(region_name=region)  # noqa: PLW2901
        else:
            raise Exception(f"Unsupported Cloud provider: `{cloud_provider}")
        region.configure()


@cli.command("configure-aws-peering", help="Configure all required resources for SCT to run in multi-dc")
@click.option(
    "-r", "--regions", type=CloudRegion(cloud_provider="aws"), default=[], help="Cloud regions", multiple=True
)
def configure_aws_peering(regions):
    add_file_logger()
    peering = AwsVpcPeering(regions)
    peering.configure()


@cli.command("configure-oci-peering", help="Configure OCI cross-region VCN peering for multi-dc SCT runs")
@click.option("-r", "--regions", type=CloudRegion(cloud_provider="oci"), default=[], help="OCI regions", multiple=True)
def configure_oci_peering(regions):
    add_file_logger()
    peering = OciVcnPeering(regions or None)
    peering.configure()


@cli.command(
    "create-runner-image",
    help=f"Create an SCT runner image in the selected cloud region."
    f" If the requested region is not a source region"
    f" (aws: {AwsSctRunner.SOURCE_IMAGE_REGION}, gce: {GceSctRunner.SOURCE_IMAGE_REGION},"
    f" azure: {AzureSctRunner.SOURCE_IMAGE_REGION}, oci: {OciSctRunner.SOURCE_IMAGE_REGION}) the image will be first created in the"
    f" source region and then copied to the chosen one.",
)
@cloud_provider_option
@click.option("-r", "--region", required=True, type=CloudRegion(), help="Cloud region")
@click.option("-z", "--availability-zone", default="", type=str, help="Name of availability zone, ex. 'a'")
def create_runner_image(cloud_provider, region, availability_zone):
    if cloud_provider == "aws":
        assert len(availability_zone) == 1, f"Invalid AZ: {availability_zone}, availability-zone is one-letter a-z."
    elif cloud_provider == "oci":
        assert availability_zone, "Availability zone is required for OCI"
    add_file_logger()
    os.environ.setdefault("SCT_CLUSTER_BACKEND", cloud_provider)
    sct_config = SCTConfiguration()
    sct_runner = get_sct_runner(
        cloud_provider=cloud_provider, region_name=region, availability_zone=availability_zone, params=sct_config
    )
    sct_runner.create_image()


@cli.command("create-runner-instance", help="Create an SCT runner instance in the selected cloud region")
@cloud_provider_option
@click.option("-r", "--region", required=True, type=CloudRegion(), help="Cloud region")
@click.option("-z", "--availability-zone", default="", type=str, help="Name of availability zone, ex. 'a'")
@click.option("-i", "--instance-type", required=False, type=str, default="", help="Instance type")
@click.option("-i", "--root-disk-size-gb", required=False, type=int, default=0, help="Root disk size in Gb")
@click.option("-t", "--test-id", required=True, type=str, help="Test ID")
@click.option("-tn", "--test-name", required=False, type=str, default="", help="Test Name")
@click.option("-d", "--duration", required=True, type=int, help="Test duration in MINUTES")
@click.option(
    "-rm", "--restore-monitor", required=False, type=bool, help="Is the runner for restore monitor purpose or not"
)
@click.option(
    "-rt",
    "--restored-test-id",
    required=False,
    type=str,
    help="Test ID of the test that the runner is created for restore monitor",
)
@click.option("-p", "--address-pool", required=False, type=str, help="ElasticIP pool to use")
def create_runner_instance(
    cloud_provider,
    region,
    availability_zone,
    instance_type,
    root_disk_size_gb,
    test_id,
    test_name,
    duration,
    restore_monitor=False,
    restored_test_id="",
    address_pool=None,
):
    add_file_logger()
    sct_runner_ip_path = Path("sct_runner_ip")
    sct_runner_ip_path.unlink(missing_ok=True)

    os.environ.setdefault("SCT_CLUSTER_BACKEND", cloud_provider)
    sct_config = SCTConfiguration()
    sct_runner = get_sct_runner(
        cloud_provider=cloud_provider, region_name=region, availability_zone=availability_zone, params=sct_config
    )

    instance_type = instance_type or sct_config.get("instance_type_runner")
    root_disk_size_gb = root_disk_size_gb or sct_config.get("root_disk_size_runner")
    test_name = test_name or test_id

    instance = sct_runner.create_instance(
        instance_type=instance_type,
        root_disk_size_gb=root_disk_size_gb,
        test_id=test_id,
        test_name=test_name,
        test_duration=duration,
        restore_monitor=restore_monitor,
        restored_test_id=restored_test_id,
        address_pool=address_pool,
    )
    if not instance:
        sys.exit(1)

    LOGGER.info("Verifying SSH connectivity...")
    runner_public_ip = sct_runner.get_instance_public_ip(instance=instance)
    remoter = sct_runner.get_remoter(host=runner_public_ip, connect_timeout=240)

    @retrying(n=3, sleep_time=30, message=f"Retrying SSH connection to {runner_public_ip}")
    def verify_ssh():
        assert remoter.run("true", timeout=200, verbose=False, ignore_status=True).ok

    try:
        verify_ssh()
        LOGGER.info("Successfully connected the SCT Runner. Public IP: %s", runner_public_ip)
        with sct_runner_ip_path.open(mode="w", encoding="utf-8") as sct_runner_ip_file:
            sct_runner_ip_file.write(runner_public_ip)
    except Exception:  # noqa: BLE001
        LOGGER.error("Unable to SSH to %s! Exiting...", runner_public_ip)
        sys.exit(1)


@cli.command("set-runner-tags")
@click.argument("runner-ip", type=str)
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Cloud backend to use")
@click.option(
    "-t",
    "--tags",
    type=(str, str),
    help="Space separated key value pair to add as a new tag to the runner",
    multiple=True,
)
def set_runner_tags(runner_ip, backend, tags):
    add_file_logger()
    update_sct_runner_tags(test_runner_ip=runner_ip, backend=backend, tags=dict(tags))


@cli.command("find-runner-instance", help="Find an existing SCT runner by test ID and write its IP to sct_runner_ip")
@click.option("-t", "--test-id", required=True, type=str, help="Test ID to find the runner for")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Cloud backend to use")
@click.option(
    "-d", "--duration", required=False, type=int, default=0, help="New test duration in minutes (extends keep tag)"
)
def find_runner_instance(test_id, backend, duration):
    add_file_logger()
    sct_runner_ip_path = Path("sct_runner_ip")
    sct_runner_ip_path.unlink(missing_ok=True)

    runners = list_sct_runners(backend=backend, test_id=test_id)
    if not runners:
        LOGGER.error("No SCT runner found for test_id: %s", test_id)
        sys.exit(1)

    runner = runners[0]
    if not runner.public_ips:
        LOGGER.error("SCT runner %s has no public IPs", runner.instance_name)
        sys.exit(1)

    runner_ip = runner.public_ips[0]
    LOGGER.info(
        "Found SCT runner %s at %s for test %s",
        runner.instance_name,
        runner_ip,
        test_id,
    )

    if duration:
        # keep is relative to launch_time, so add elapsed hours to give the test enough time
        elapsed_hours = (
            int((datetime.now(UTC) - runner.launch_time).total_seconds() / 3600) if runner.launch_time else 0
        )
        tags_to_update = {"keep": str(elapsed_hours + int(duration / 60) + 6), "keep_action": "terminate"}
        update_sct_runner_tags(backend=backend, test_runner_ip=runner_ip, tags=tags_to_update)
        LOGGER.info("Updated runner tags: %s", tags_to_update)

    sct_runner_ip_path.write_text(runner_ip)
    LOGGER.info("SCT Runner IP written to %s: %s", sct_runner_ip_path, runner_ip)


@cli.command("clean-runner-instances", help="Clean all unused SCT runner instances")
@click.option("-ip", "--runner-ip", required=False, type=str, default="")
@click.option("-ts", "--test-status", type=str, help="The result of the test run")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Specific backend to use")
@click.option("--user", type=str, help="user name to filter instances by")
@click.option("--billing-project", type=str, help="billing project to filter instances by")
@click.option("--test-id", "test_id", help="test id to filter by. Could be used multiple times", multiple=True)
@click.option("--dry-run", is_flag=True, default=False, help="dry run")
@click.option("--force", is_flag=True, default=False, help="Skip cleaning logic and terminate the instance")
def clean_runner_instances(runner_ip, test_status, backend, user, billing_project, test_id, dry_run, force):
    add_file_logger()
    # Note: clean_sct_runners does not currently support billing_project; option accepted for CLI parity.
    clean_sct_runners(
        test_runner_ip=runner_ip,
        test_status=test_status,
        backend=backend,
        user=user,
        test_id=test_id,
        dry_run=dry_run,
        force=force,
    )


@cli.command("configure-jenkins-builders", help="Configure all required jenkins builders for SCT")
@cloud_provider_option
@click.option("-r", "--regions", type=CloudRegion(), default=[], help="Cloud regions", multiple=True)
def configure_jenkins_builders(cloud_provider, regions):
    add_file_logger()
    logging.basicConfig(level=logging.INFO)

    match cloud_provider:
        case "aws":
            AwsCiBuilder(AwsRegion("eu-west-1")).configure_auto_scaling_group()
            AwsFipsCiBuilder(AwsRegion("us-east-1")).configure_auto_scaling_group()
            AwsBuilder.configure_in_all_region(regions=regions)
        case "gce":
            GceBuilder.configure_in_all_region(regions=regions)
        case "oci":
            OciBuilder.configure_in_all_region(regions=regions)
        case "azure":
            raise NotImplementedError("configure_jenkins_builders doesn't support Azure yet")


@cli.command("nemesis-list", help="get the list of select disrupt function for SisyphusMonkey")
@click.option("-b", "--backend", type=click.Choice(available_backends), help="Backend to use")
@click.option(
    "-c",
    "--config",
    multiple=True,
    type=click.Path(exists=True),
    help="Test config .yaml to use, can have multiple of those",
)
def get_nemesis_list(backend, config):
    """
    # usage via command line:
    hydra nemesis-list -c test-cases/longevity/longevity-cdc-100gb-4h.yaml -c configurations/tablets.yaml

    # usage with environment variables
    export SCT_CONFIG_FILES='["test-cases/longevity/longevity-cdc-100gb-4h.yaml", "configurations/tablets.yaml"]'
    hydra nemesis-list

    """
    add_file_logger()
    logging.basicConfig(level=logging.WARNING)

    if config:
        os.environ["SCT_CONFIG_FILES"] = str(list(config))
    if backend:
        os.environ["SCT_CLUSTER_BACKEND"] = backend

    tester = FakeTester()

    tester.params = SCTConfiguration()
    sisyphus_nemesis = SisyphusMonkey(tester, None)

    collected_disrupt_methods_names = [disrupt.__name__ for disrupt in sisyphus_nemesis.disruptions_list]

    click.secho(f"config files used: {pprint.pformat(tester.params.get('config_files'))}\n\n", fg="green")
    click.secho(pprint.pformat(collected_disrupt_methods_names), fg="green")


@cli.command("create-argus-test-run", help="Initialize an argus test run.")
def create_argus_test_run():
    try:
        params = SCTConfiguration()
        git_status = get_git_status_info()
        test_config = get_test_config()
        if not params.get("test_id"):
            LOGGER.error("test_id is not set")
            return
        test_config.set_test_id_only(params.get("test_id"))
        test_config.init_argus_client(params)
        argus_client = test_config.argus_client()
        argus_client.submit_sct_run(
            job_name=get_job_name(),
            job_url=get_job_url(),
            started_by=get_username(),
            commit_id=git_status.get("branch.oid", get_git_commit_id()),
            origin_url=git_status.get("upstream.url"),
            branch_name=git_status.get("branch.upstream"),
            sct_config=None,
        )
        if isinstance(argus_client, ReplayOnlyArgusSCTClient):
            LOGGER.warning(
                "Argus is DISABLED (SCT_ENABLE_ARGUS=false) - using replay-only mode. "
                "Test id: %s. Logs will still be collected and uploaded to S3, "
                "but no Argus test run exists. Use 'hydra.sh collect-logs' output "
                "or S3 links in console to download artifacts.",
                argus_client.run_id,
            )
        else:
            LOGGER.info("Initialized Argus TestRun with test id %s", argus_client.run_id)
    except ArgusClientError:
        LOGGER.error("Failed to submit data to Argus", exc_info=True)


@cli.command("finish-argus-test-run", help="Finish argus test run if it is not finished by SCT.")
@click.option("-s", "--jenkins-status", type=str, help="jenkins build status", required=True)
def finish_argus_test_run(jenkins_status):
    try:
        LOGGER.info("Finishing Argus TestRun with jenkins status %s", jenkins_status)
        params = SCTConfiguration()
        test_config = get_test_config()
        if not params.get("test_id"):
            LOGGER.error("test_id is not set")
            return
        test_config.set_test_id_only(params.get("test_id"))
        test_config.init_argus_client(params)
        if jenkins_status == "ABORTED":
            LOGGER.info("Jenkins build status is ABORTED, setting Argus TestRun status to ABORTED")
            new_status = TestStatus.ABORTED
        else:
            status = test_config.argus_client().get_status()
            if status in [TestStatus.PASSED, TestStatus.FAILED, TestStatus.TEST_ERROR]:
                LOGGER.info("Argus TestRun already finished with status %s", status.value)
                return
            new_status = TestStatus.FAILED
        test_config.argus_client().set_sct_run_status(new_status)
        test_config.argus_client().finalize_sct_run()
    except ArgusClientError:
        LOGGER.error("Failed to submit data to Argus", exc_info=True)


@cli.command("fetch-junit-from-runner", help="copy the junit.xml back")
@click.argument("runner-ip", type=str)
@click.option(
    "-b",
    "--backend",
    type=click.Choice(available_backends),
    help="Allows to skip making backend detection API calls.",
)
def fetch_junit(runner_ip, backend):
    add_file_logger()

    runner = list_sct_runners(backend=backend, test_runner_ip=runner_ip)
    proxy_command, target_ip, target_username, target_key = sct_ssh.get_proxy_command(runner[0].instance, True)
    cmd = (
        f"ssh -tt {proxy_command}"
        f' -i {target_key} -o "UserKnownHostsFile=/dev/null" '
        f'-o "StrictHostKeyChecking=no" -o ServerAliveInterval=10 {target_username}@{target_ip} '
        f"'find ~/sct-results -iname junit.xml | xargs cat'"
    )

    output = subprocess.run(["bash", "-c", cmd], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
    junit_xml_file = Path("results") / "junit.xml"
    junit_xml_file.parent.mkdir(parents=True, exist_ok=True)
    junit_xml_file.write_text(output.stdout.strip())


@cli.command("upload", help="Upload arbitrary log/screenshot to s3 corresponding to the test_id")
@click.option("--test-id", type=str, required=True)
@click.option("--use-argus/--no-use-argus", default=True)
@click.option("--public/--no-public", default=False)
@click.argument("file-path", type=str, required=True)
def upload_artifact_file(test_id: str, file_path: str, use_argus: bool, public: bool):
    add_file_logger()
    if use_argus:
        params = SCTConfiguration()
        params["test_id"] = test_id
        test_config = get_test_config()
        test_config.set_test_id_only(params.get("test_id"))
        test_config.init_argus_client(params)
        client = test_config.argus_client()
        try:
            client.get_status()
        except ArgusClientError:
            LOGGER.error("Failed getting status for %s in Argus, aborting...", test_id)
            return
    else:
        # --no-use-argus: no HTTP calls are made, but submissions are still
        # recorded to the replay log. Set the test id first so those records are
        # attributed to this run instead of a random fallback uuid.
        test_config = get_test_config()
        test_config.set_test_id_only(test_id)
        client = test_config.argus_client()  # replay-only client when Argus is disabled

    image_exts = [".jpg", ".png"]
    file = Path(file_path)
    if file.exists():
        timestamp = datetime.now(timezone(timedelta(hours=0, minutes=0)))
        subfolder = timestamp.strftime("upload_%Y%m%d_%H%M%S")
        s3_path = f"{test_id}/{subfolder}"
        LOGGER.info("Going to upload %s to S3...", file.absolute())
        s3 = S3Storage()
        file_url = s3.upload_file(file.absolute(), s3_path, public)
        LOGGER.info("Uploaded %s to %s", file.absolute(), file_url)
        client.submit_sct_logs([LogLink(log_name=file.name, log_link=file_url)])
        if file.suffix in image_exts:
            client.submit_screenshots([file_url])
    else:
        LOGGER.error("File %s does not exist", file.absolute())
        return


@cli.command(
    "hdr-investigate",
    help="Analyze HDR file for latency spikes.\n"
    "Usage example:\n"
    "hydra hdr-investigate --stress-operation READ --throttled-load true "
    '--test-id 8732ecb1-7e1f-44e7-b109-6d789b15f4b5 --start-time "2025-09-14\\ 20:45:18" '
    "--duration-from-start-min 30",
)
@click.option(
    "--test-id",
    type=str,
    required=False,
    help="If hdr_folder is not provided, logs will be downloaded from argus using this test_id",
)
@click.option(
    "--stress-tool",
    default="cassandra-stress",
    required=False,
    type=click.Choice(["cassandra-stress", "scylla-bench", "latte"], case_sensitive=False),
    help="stress tool name. Supported tools: cassandra-stress|scylla-bench|latte",
)
@click.option(
    "--stress-operation",
    required=True,
    type=click.Choice(["READ", "WRITE"], case_sensitive=False),
    help="Supported stress operations: READ|WRITE",
)
@click.option("--throttled-load", type=bool, required=True, help="Is the load throttled or not")
@click.option("--start-time", type=str, required=True, help="Start time in format 'YYYY-MM-DD\\ HH:MM:SS'")
@click.option(
    "--duration-from-start-min",
    type=int,
    required=True,
    help="Time period in minutes in HDR file to investigate, started from start-time ",
)
@click.option(
    "--error-threshold-ms",
    type=int,
    default=10,
    required=False,
    help="Error threshold in ms for P99 to consider it as a spike",
)
@click.option("--hdr-summary-interval-sec", type=int, default=600, required=False, help="Interval in seconds for scan")
@click.option("--hdr-folder", type=str, default=None, required=False, help="Path to folder with hdr files. ")
def hdr_investigate(
    test_id: str,
    stress_tool: str,
    stress_operation: str,
    throttled_load: bool,
    start_time: str,
    duration_from_start_min: int,
    error_threshold_ms: int,
    hdr_summary_interval_sec: int,
    hdr_folder: str,
) -> None:
    """
    Analyze HDR file for latency spikes.

    This function scans HDR files to identify intervals with high latency spikes.
    It performs a scan to find intervals with the highest P99 latency
    Args:
        test_id (str): Test ID. If `hdr_folder` is not provided, logs will be downloaded from Argus using this ID to /tmp/<test-id> folder.
        stress_tool (str): Name of the stress tool. Supported: cassandra-stress, scylla-bench, latte.
        stress_operation (str): Stress operation type. Supported: READ, WRITE.
        throttled_load (bool): Whether the load is throttled.
        start_time (str): Start time in format 'YYYY-MM-DD HH:MM:SS'.
        duration_from_start_min (int): Time period in minutes in HDR file to investigate, starting from `start_time`.
        error_threshold_ms (int): Error threshold in ms for P99 to consider it as a spike.
        hdr_summary_interval_sec (int): Interval in seconds for scan.
        hdr_folder (str): Path to folder with HDR files. If not provided, logs will be downloaded from Argus.

    Returns:
        None

    Usage example:
       hydra hdr-investigate --stress-operation READ --throttled-load true --test-id 8732ecb1-7e1f-44e7-b109-6d789b15f4b5
       --start-time \"2025-09-14\\ 20:45:18\" --duration-from-start-min 30
    """
    stress_operation = stress_operation.upper()

    try:
        start_time_ms = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC).timestamp()
    except ValueError:
        raise ValueError("start_time must be in 'YYYY-MM-DD HH:MM:SS' format")

    if not hdr_folder:
        if not test_id:
            raise ValueError("Either test_id or hdr_folder must be provided")

        hdr_folder = download_and_unpack_logs(test_id, log_type="loader-set")

    hdr_tags = get_hdr_tags(stress_tool, stress_operation, throttled_load)

    # Step 1: Coarse scan
    end_time_ms = start_time_ms + duration_from_start_min * 60
    hdr_summaries = make_hdrhistogram_summary_by_interval(
        hdr_tags, stress_operation, hdr_folder, start_time_ms, end_time_ms, interval=hdr_summary_interval_sec
    )

    summaries = []
    for tag in hdr_tags:
        # Find intervals for this tag
        tag_hdr_summaries = [(summary.get(f"{stress_operation}--{tag}", {})) for summary in hdr_summaries]
        # Filter intervals where percentile_99 > error threshold. Meaning it is a spike
        tag_summaries = [
            s for s in tag_hdr_summaries if s and "percentile_99" in s and s["percentile_99"] > error_threshold_ms
        ]
        # Sort by percentile_99 descending
        tag_summaries.sort(key=lambda x: x["percentile_99"], reverse=True)
        for num, operation in enumerate(tag_summaries, start=1):
            p99 = {}
            # Convert to human readable time
            dt_start = datetime.fromtimestamp(operation["start_time"] / 1000, UTC).strftime("%Y-%m-%d %H:%M:%S")
            dt_end = datetime.fromtimestamp(operation["end_time"] / 1000, UTC).strftime("%Y-%m-%d %H:%M:%S")
            if operation["percentile_99"] > error_threshold_ms:
                p99[f"{dt_start} - {dt_end}"] = operation["percentile_99"]
            summaries.append({"tag": tag, "spikes": p99})

    hdr_table = Table("Tag", "Timeframe", "P99", show_lines=False)
    for group in summaries:
        # Example: group: {'tag': 'READ-st', 'spikes': {'2025-08-25 10:22:40 - 2025-08-25 10:32:40': 487.85}}
        for time_frame, p99 in group["spikes"].items():
            hdr_table.add_row(group["tag"], time_frame, str(p99))
    click.echo(
        f"\nFound P99 spikes higher than {error_threshold_ms} ms for tags {hdr_tags} with interval {hdr_summary_interval_sec} seconds\n"
    )
    click.echo(rich_table_to_string(hdr_table, title="HDR Latency Spikes"))


@cli.command("trigger-matrix", help="Trigger Jenkins jobs from a YAML matrix configuration file")
@click.option("--matrix", required=True, type=click.Path(exists=True), help="Path to trigger matrix YAML file")
@click.option(
    "--scylla-version",
    required=False,
    default=None,
    type=str,
    help="Scylla version (e.g., master:latest, 2024.2.5-0.20250221.xxx-1). "
    "If not provided, resolved automatically from image params",
)
@click.option("--job-folder", default=None, type=str, help="Override auto-detected Jenkins job folder")
@click.option("--labels-selector", default=None, type=str, help="Comma-separated labels to filter jobs (AND logic)")
@click.option(
    "--backend",
    default=None,
    type=click.Choice(["aws", "gce", "azure", "docker", "oci"]),
    help="Filter jobs by backend",
)
@click.option("--skip-jobs", default=None, type=str, help="Comma-separated job names to skip")
@click.option("--stress-duration", default=None, type=str, help="Override stress_duration parameter")
@click.option(
    "--region",
    default=None,
    type=str,
    help="Region for jobs that don't set one in the matrix. Jobs with their own 'region' param keep it",
)
@click.option(
    "--availability-zone",
    default=None,
    type=str,
    help="Availability zone for jobs that don't set one in the matrix. "
    "Jobs with their own 'availability_zone' param keep it",
)
@click.option(
    "--provision-type",
    default=None,
    type=click.Choice(["spot", "on_demand", "spot_fleet"]),
    help="Override provision type",
)
@click.option(
    "--scylla-ami-id",
    default=None,
    type=str,
    help="Scylla AMI ID (AWS) — also used to resolve scylla_version if not provided",
)
@click.option(
    "--gce-image-db",
    default=None,
    type=str,
    help="Scylla GCE image — also used to resolve scylla_version if not provided",
)
@click.option(
    "--azure-image-db",
    default=None,
    type=str,
    help="Scylla Azure image — also used to resolve scylla_version if not provided",
)
@click.option(
    "--oci-image-db",
    default=None,
    type=str,
    help="Scylla OCI image OCID — also used to resolve scylla_version if not provided",
)
@click.option(
    "--unified-package",
    default=None,
    type=str,
    help="Unified package URL (for PGO offline installer jobs that don't need scylla_version)",
)
@click.option(
    "--new-scylla-repo",
    default=None,
    type=str,
    help="Scylla repo URL for rolling-upgrade jobs (RPM .repo or DEB .list). Overrides per-job template values.",
)
@click.option(
    "--base-versions",
    default=None,
    type=str,
    help="Passed from scylla-pkg release triggers. Only relevant for rolling upgrade tests. "
    "Comma-separated base versions to upgrade FROM (e.g., '2025.1,2025.2'). "
    "When empty, the downstream job auto-selects base versions.",
)
@click.option(
    "--version-resolution",
    default=None,
    type=click.Choice(VERSION_RESOLUTION_STRATEGIES),
    help="How the version passed to the jobs is picked (overrides the matrix YAML): "
    "per-backend — every backend runs its own latest build; "
    "common — all jobs run the newest build published on every backend in the matrix; "
    "aws-strict — the AWS build for everyone, backends missing that build are not triggered",
)
@click.option("--dry-run", is_flag=True, default=False, help="Preview mode — do not trigger jobs")
@click.option(
    "--use-job-throttling/--no-use-job-throttling",
    default=None,
    help="Override use_job_throttling parameter for triggered jobs",
)
@click.option("--requested-by-user", default=None, type=str, help="User requesting the run")
@click.option("--billing-project", default=None, type=str, help="Billing project for resource tagging")
@click.option(
    "--email-recipients", default=None, type=str, help="Comma-separated email recipients for wait-mode results report"
)
def trigger_matrix_cmd(  # noqa: PLR0912, PLR0913
    matrix,
    scylla_version,
    job_folder,
    labels_selector,
    backend,
    skip_jobs,
    stress_duration,
    region,
    availability_zone,
    provision_type,
    scylla_ami_id,
    gce_image_db,
    azure_image_db,
    oci_image_db,
    unified_package,
    new_scylla_repo,
    base_versions,
    version_resolution,
    dry_run,
    use_job_throttling,
    requested_by_user,
    billing_project,
    email_recipients,
):
    add_file_logger()

    # Resolve scylla_version: from image tags, or from partial version via AMI lookup
    version_from_image = False
    if not scylla_version:
        has_image = any([scylla_ami_id, gce_image_db, azure_image_db, oci_image_db])
        if unified_package:
            # PGO jobs only need unified_package, no version resolution needed
            pass
        elif new_scylla_repo:
            # Rolling upgrade jobs can be triggered with just new_scylla_repo
            pass
        elif not has_image:
            click.echo(
                "Error: Either --scylla-version, an image param "
                "(--scylla-ami-id, --gce-image-db, --azure-image-db, --oci-image-db), "
                "--new-scylla-repo, or --unified-package is required",
                err=True,
            )
            sys.exit(1)
        else:
            try:
                scylla_version = resolve_scylla_version_from_image(
                    scylla_ami_id=scylla_ami_id,
                    gce_image_db=gce_image_db,
                    azure_image_db=azure_image_db,
                    oci_image_db=oci_image_db,
                    region=region,
                )
                version_from_image = True
                click.echo(f"Resolved scylla_version from image: {scylla_version}")
            except Exception as exc:  # noqa: BLE001
                click.echo(f"Error: {exc}", err=True)
                sys.exit(1)

    # For user-provided partial versions (master:latest, 2025.4), resolve to full version.
    # This is the AWS reference version — it decides pre_release job filtering, and is what
    # aws-strict stamps on every job. Each backend still resolves its own build later on,
    # according to the matrix' version_resolution strategy.
    # Skip if version was already resolved from image tags — that's already the real version.
    original_version = scylla_version
    if scylla_version and not version_from_image:
        try:
            scylla_version = resolve_to_full_version(scylla_version, region=region)
            click.echo(f"Using full version: {scylla_version}")
        except Exception as exc:  # noqa: BLE001
            if version_resolution == "aws-strict":
                click.echo(f"Error resolving full version: {exc}", err=True)
                sys.exit(1)
            click.echo(f"Warning: {exc}")
            click.echo(f"Leaving '{original_version}' unresolved — each backend resolves its own build")

    # Detect image architecture for arch-aware job filtering
    image_arch = None
    if scylla_ami_id:
        image_arch = resolve_image_architecture(scylla_ami_id=scylla_ami_id, region=region)
        if not image_arch:
            click.echo(f"Error: could not resolve architecture for AMI {scylla_ami_id}", err=True)
            sys.exit(1)
        click.echo(f"Resolved image architecture: {image_arch}")
    elif any([gce_image_db, azure_image_db, oci_image_db]):
        image_arch = resolve_image_architecture(
            gce_image_db=gce_image_db,
            azure_image_db=azure_image_db,
            oci_image_db=oci_image_db,
            region=region,
        )
        if image_arch:
            click.echo(f"Resolved image architecture: {image_arch}")

    overrides = {}
    if stress_duration:
        overrides["stress_duration"] = stress_duration
    if region:
        overrides["region"] = region
    if availability_zone:
        overrides["availability_zone"] = availability_zone
    if provision_type:
        overrides["provision_type"] = provision_type
    if requested_by_user:
        overrides["requested_by_user"] = requested_by_user
    if billing_project:
        overrides["billing_project"] = billing_project
    if unified_package:
        overrides["unified_package"] = unified_package
    if new_scylla_repo:
        overrides["new_scylla_repo"] = new_scylla_repo
    if base_versions:
        overrides["base_versions"] = base_versions
    if use_job_throttling is not None:
        overrides["use_job_throttling"] = use_job_throttling

    try:
        results = run_trigger_matrix(
            matrix_file=matrix,
            scylla_version=scylla_version or "",
            filter_version=original_version,
            job_folder=job_folder,
            labels_selector=labels_selector,
            backend=backend,
            skip_jobs=skip_jobs,
            dry_run=dry_run,
            email_recipients=email_recipients.split(",") if email_recipients else None,
            image_arch=image_arch,
            version_resolution=version_resolution,
            **overrides,
        )
    except Exception as exc:  # noqa: BLE001
        click.echo(f"Error: {exc}", err=True)
        sys.exit(1)

    click.echo(f"\nTriggered: {len(results['triggered'])} jobs")
    for job in results["triggered"]:
        click.echo(f"  + {job}")
    if results["skipped"]:
        click.echo(f"Skipped: {len(results['skipped'])} jobs")
        for job in results["skipped"]:
            click.echo(f"  - {job}")
    if results["failed"]:
        click.echo(f"Failed: {len(results['failed'])} jobs")
        for job in results["failed"]:
            click.echo(f"  ! {job}")
        sys.exit(1)


cli.add_command(sct_ssh.ssh)
cli.add_command(sct_ssh.tunnel)
cli.add_command(sct_ssh.copy_cmd)
cli.add_command(sct_ssh.attach_test_sg_cmd)
cli.add_command(sct_ssh.ssh_cmd)
cli.add_command(sct_ssh.gcp_allow_public)
cli.add_command(sct_ssh.update_scylla_packages)
cli.add_command(sct_scan_issues.scan_issue_skips)
cli.add_command(sct_sizing.sizing_group)


if __name__ == "__main__":
    cli.main(prog_name="hydra")
