import time
import logging
from typing import Callable, Optional, NamedTuple, Union

from fabric.runners import Result

from sdcm.cluster import BaseNode, BaseCluster, BaseScyllaCluster
from sdcm.rest.storage_service_client import StorageServiceClient

LOGGER = logging.getLogger(__name__)


class CompactionTypes(NamedTuple):
    COMPACTION: str = "COMPACTION"
    CLEANUP: str = "CLEANUP"
    SCRUB: str = "SCRUB"
    RESHAPE: str = "RESHAPE"
    UPGRADE: str = "UPGRADE"


class ScrubModes(NamedTuple):
    ABORT: str = "ABORT"
    SKIP: str = "SKIP"
    SEGREGATE: str = "SEGREGATE"
    VALIDATE: str = "VALIDATE"


class NodetoolCommands(NamedTuple):
    flush: str = "flush"
    stop_major_compaction: str = "stop COMPACTION"
    stop_scrub_compaction: str = "stop SCRUB"
    stop_cleanup_compaction: str = "stop CLEANUP"
    stop_upgrade_compaction: str = "stop UPGRADE"
    stop_reshape_compaction: str = "stop RESHAPE"


class CompactionOps:
    NODETOOL_CMD = NodetoolCommands()
    SCRUB_MODES = ScrubModes()

    def __init__(self, cluster: Union[BaseCluster, BaseScyllaCluster], node: Optional[BaseNode] = None):
        self.cluster = cluster
        self.node = node if node and not node._is_zero_token_node else self.cluster.data_nodes[0]
        self.storage_service_client = StorageServiceClient(node=self.node)

    def trigger_major_compaction(self, keyspace: str = "keyspace1", cf: str = "standard1") -> Result:
        return self.storage_service_client.compact_ks_cf(keyspace=keyspace, cf=cf)

    def trigger_scrub_compaction(
        self, keyspace: str = "keyspace1", cf: str = "standard1", scrub_mode: Optional[str] = None
    ) -> Result:
        params = {"keyspace": keyspace, "cf": cf, "scrub_mode": scrub_mode}

        return self.storage_service_client.scrub_ks_cf(**params)

    def trigger_cleanup_compaction(
        self, keyspace: str = "keyspace1", cf: str = "standard1", timeout: int = 600
    ) -> Result:
        return self.storage_service_client.cleanup_ks_cf(keyspace=keyspace, cf=cf, timeout=timeout)

    def trigger_validation_compaction(self, keyspace: str = "keyspace1", cf: str = "standard1") -> Result:
        return self.storage_service_client.scrub_ks_cf(keyspace=keyspace, cf=cf, scrub_mode=self.SCRUB_MODES.VALIDATE)

    def trigger_upgrade_compaction(self, keyspace: str = "keyspace1", cf: str = "standard1") -> Result:
        return self.storage_service_client.upgrade_sstables(keyspace=keyspace, cf=cf)

    def trigger_flush(self):
        self.node.run_nodetool(self.NODETOOL_CMD.flush)

    def stop_major_compaction(self) -> Result:
        LOGGER.info("Stopping major compaction with <nodetool stop>")
        return self._stop_compaction(self.NODETOOL_CMD.stop_major_compaction)

    def stop_scrub_compaction(self) -> Result:
        return self._stop_compaction(self.NODETOOL_CMD.stop_scrub_compaction)

    def stop_cleanup_compaction(self) -> Result:
        return self._stop_compaction(self.NODETOOL_CMD.stop_cleanup_compaction)

    def stop_upgrade_compaction(self) -> Result:
        return self._stop_compaction(self.NODETOOL_CMD.stop_upgrade_compaction)

    def stop_reshape_compaction(self) -> Result:
        return self._stop_compaction(self.NODETOOL_CMD.stop_reshape_compaction)

    def stop_validation_compaction(self) -> Result:
        return self.stop_scrub_compaction()

    def disable_autocompaction_on_ks_cf(self, node: BaseNode, keyspace: str = "", cf: Optional[str] = ""):
        node = node if node else self.node
        node.run_nodetool(f"disableautocompaction {keyspace} {cf}")

    def _stop_compaction(self, nodetool_cmd: str) -> Result:
        LOGGER.info("Stopping compaction with nodetool %s", nodetool_cmd)
        return self.node.run_nodetool(nodetool_cmd)

    @staticmethod
    def stop_on_user_compaction_logged(
        node: BaseNode, watch_for: str, timeout: int, stop_func: Callable, mark: Optional[int] = None
    ) -> Result:
        LOGGER.info("Starting to watch for user compaction logged...")
        start_time = time.time()
        with open(node.system_log, encoding="utf-8") as log_file:
            if mark:
                log_file.seek(mark)

            while time.time() - start_time < timeout:
                line = log_file.readline()

                if watch_for in line:
                    func_return = stop_func()
                    LOGGER.info("Grepped expression found: %s in log line %s", watch_for, line)
                    return func_return


class StartStopCompactionArgs(NamedTuple):
    """
    NamedTuple for passing around common arguments
    in the start-stop compaction tests.
    """

    keyspace: str
    columnfamily: str
    timeout: int
    target_node: BaseNode
    compaction_ops: CompactionOps


COMPACTION_TYPES = CompactionTypes()
