# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""
This module contains RequestProcessor class
"""

import logging
import sys
import time
import signal
import traceback
from datetime import datetime, timedelta, timezone
from threading import Thread, RLock, current_thread
from typing import Callable, Any
from queue import Queue, Empty, Full
from sqlalchemy.exc import OperationalError

from .autotracer import AutoTracer
from .common import Common
from .decision_maker import DecisionMaker
from .stat_sender import StatisticsSender
from ..db import (
    session_scope,
    setup_database,
    RequestResult,
    cleanup_old_data,
    ensure_indexes,
    restore_database,
    is_malformed_database,
)
from ..internal.exceptions import SSAError
from ..internal.utils import singleton, url_split, switch_schedstats


@singleton
class RequestProcessor(Common):
    """
    SSA Request processor implementation.
    Only one instance is allowed to be created
    """

    BUFFER_SIZE = 100
    DB_ACCESS_RETRIES = 5
    # Bound the ingest queue so a tenant telemetry flood cannot grow the
    # root agent's RSS without limit; scales with the flush batch.
    QUEUE_MAXSIZE = BUFFER_SIZE * 10
    # Per-sender fairness cap: a single sender (keyed by kernel-trusted
    # peer UID) may hold at most this many of the shared queue's slots, so
    # one tenant's telemetry flood cannot occupy the whole queue and starve
    # other tenants' accepted requests.
    PER_SENDER_QUEUE_SHARE = QUEUE_MAXSIZE // 10
    # Rate-limit the drop warning so a sustained flood does not also spam the log.
    DROP_LOG_INTERVAL = 1000

    def __init__(self, engine=None):
        super().__init__()
        self.logger = logging.getLogger("req_processor")
        self.logger.info("Processor enabled: %s", __package__)
        # enable throttling detection kernel mechanism on service start
        switch_schedstats(enabled=True)

        self.engine = engine if engine else setup_database()
        self._lock = RLock()
        self.decision_maker = DecisionMaker(engine=self.engine)
        self.sender = StatisticsSender()
        self.auto_tracer = AutoTracer(engine=self.engine)
        self._queue = Queue(maxsize=self.QUEUE_MAXSIZE)
        self._dropped_requests = 0
        # Per-sender in-flight slot accounting for enqueue fairness.
        # Guarded by its own lock (not self._lock) so the enqueue hot path
        # never contends with the long-held DB-restore critical section.
        self._sender_counts = {}
        self._sender_counts_lock = RLock()
        self._buffer = []
        self.start_background_routine()
        self.start_flush_worker()

        # Catch the shutdown signal for save last data from buffer if < BUFFER_SIZE
        signal.signal(signal.SIGTERM, self.shutdown_handler)
        signal.signal(signal.SIGINT, self.shutdown_handler)

    @property
    def configured_duration(self):
        """
        Return config file value multiplied by 1000000,
        as we receive duration in microseconds
        """
        return self.requests_duration * 1000000

    def send_stats(self, report: dict):
        """
        Call Statistics Sender
        """
        try:
            self.sender.send(report)
        except SSAError as e:
            self.logger.error("StatisticsSender failed: %s", str(e))

    def start_background_routine(self) -> None:
        """
        Start dumper|DecisionMaker thread in background
        """
        t = Thread(target=self.background_routine, daemon=True)
        t.start()
        self.logger.info("[%s] Routine started", t.name)

    def start_flush_worker(self) -> None:
        """
        Start flush worker thread
        """
        t = Thread(target=self.flush_worker, daemon=True)
        t.start()
        self.logger.info("[%s] Flush worker started", t.name)

    def background_routine(self) -> None:
        """
        Dumps collected stats to file once an hour.
        Runs DecisionMaker once a day.
        """
        while True:
            tick = datetime.now(timezone.utc)
            if tick.minute == 0:
                if tick.hour == 0:
                    self._daily_routine(tick)
                self._simple_sleep(60)
            else:
                self._sleep_till_next_hour(tick.minute)

    def _daily_routine(self, tick) -> None:
        """
        Once-a-day maintenance: flush buffer, prune retention, run
        AutoTracer and DecisionMaker, then send stats.

        Cleanup runs FIRST and is wrapped in ``_safe_exec`` so that
        retention is always enforced even if a later step
        (AutoTracer / DecisionMaker) is killed by the cgroup OOM killer
        on a large ``ssa.db`` (CLPRO-3077). Otherwise the DB keeps
        growing across restarts and the OOM becomes self-reinforcing.
        """
        if is_malformed_database(self.engine):
            # Pass the callable and its arg separately: calling it inline
            # would run restore_database OUTSIDE _safe_exec (and then invoke
            # _safe_exec(None) -> TypeError), so a restore failure would
            # escape into background_routine's `while True` (CLPRO-3077).
            self._safe_exec(self.restore_db_with_lock, self.engine)
            self.logger.info(
                "[%s] Routine thread found Database disk image is malformed and now restored (%s)",
                current_thread().name,
                tick,
            )
        self.logger.info(
            "[%s] Routine thread launching buffer flushing (%s)",
            current_thread().name,
            tick,
        )
        self._safe_exec(self.flush_remaining_objects)
        self.logger.info("[%s] Routine thread launching cleanup (%s)", current_thread().name, tick)
        self._safe_exec(cleanup_old_data, self.engine)
        # Build the (domain, path) index on the post-cleanup (smaller)
        # table, before the AutoTracer / DecisionMaker scans that rely on
        # it to stream in index order instead of an external ORDER BY sort
        # whose temp spill could be charged against the cgroup on tmpfs
        # (CLPRO-3077).
        self.logger.info("[%s] Routine thread ensuring indexes (%s)", current_thread().name, tick)
        self._safe_exec(ensure_indexes, self.engine)
        self.logger.info("[%s] Routine thread launching AutoTracer (%s)", current_thread().name, tick)
        self._safe_exec(self.auto_tracer)
        self.logger.info(
            "[%s] Routine thread launching DecisionMaker (%s)",
            current_thread().name,
            tick,
        )
        report = self._safe_exec(self.decision_maker)
        self._safe_exec(self.send_stats, report)
        # attempt to enable throttling detection kernel mechanism
        # in case it was accidentally switched off. Wrapped in _safe_exec
        # like every other daily step so a failure here can't escape into
        # background_routine's unguarded `while True` and kill the thread.
        self._safe_exec(switch_schedstats, True)

    def _safe_exec(self, action: Callable, *args) -> Any:
        """Call requested Callable with given args and capture any exception"""
        try:
            return action(*args)
        except Exception:
            et, ev, _ = sys.exc_info()
            self.logger.exception(
                "%s failed with exception %s, %s",
                str(action),
                et,
                ev,
                extra={"orig_traceback": traceback.format_exc()},
            )

    def _simple_sleep(self, to_sleep: int = 15 * 60):
        """
        Log and sleep given number of seconds or 15 minutes by default
        """
        self.logger.info("[%s] Routine thread sleeping for (%s)", current_thread().name, to_sleep)
        time.sleep(to_sleep)

    def _sleep_till_next_hour(self, start_minute):
        """
        Sleep the number of minutes remaining till next hour
        """
        sleep_for = (timedelta(hours=1) - timedelta(minutes=start_minute)).total_seconds()
        self._simple_sleep(int(sleep_for))

    def restore_db_with_lock(self, engine):
        with self._lock:
            restore_database(engine)

    @staticmethod
    def get_interval_for(timestamp: int) -> int:
        """
        Takes an hour of a day, to which the given timestamp belongs
        """
        return datetime.fromtimestamp(timestamp, timezone.utc).hour

    def flush_worker(self):
        """
        Continuously flush queued request results to the database
        """
        while True:
            try:
                obj = self._queue.get()
                self._release_sender_slot(obj)
                self._buffer.append(obj)
                if len(self._buffer) >= self.BUFFER_SIZE:
                    self._flush_objects(self._buffer)
                    self._buffer = []
            except Exception:
                self.logger.exception("Flush worker failed")

    def flush_remaining_objects(self):
        """
        Flush all remaining objects even if less than BUFFER_SIZE.
        Should be called once a day.
        """
        if self._buffer:
            try:
                self._flush_objects(self._buffer)
            except Exception:
                self.logger.exception("Flush remaining objects failed")
            finally:
                self._buffer = []

    def _flush_objects(self, objects):
        for attempt in range(self.DB_ACCESS_RETRIES):
            try:
                with session_scope(self.engine) as db:
                    db.bulk_save_objects(objects)
                break
            except OperationalError as e:
                if "database is locked" in str(e):
                    self.logger.warning(
                        "Database is locked, retrying attempt %s/%s...",
                        attempt + 1,
                        self.DB_ACCESS_RETRIES,
                    )
                    time.sleep(0.1)
                else:
                    raise

    def handle(self, data: dict, peer_uid: int) -> None:
        """
        Process given request data.

        ``peer_uid`` is the kernel-trusted SO_PEERCRED UID of the sender
        (resolved by the socket layer); it keys the per-sender fairness cap.
        """
        if not data:
            self.logger.info("[%s] has empty request, skipping", current_thread().name)
            return
        url = data.get("url")
        if self.is_ignored(url):
            self.logger.debug("%s ignored", url)
            return
        domain, uri = url_split(url)

        request = RequestResult(
            domain=domain,
            path=url,
            timestamp=data["timestamp"],
            duration=data["duration"],
            is_slow_request=data["duration"] > self.configured_duration,
            hitting_limits=data["hitting_limits"],
            throttled_time=data["throttled_time"],
            io_throttled_time=data["io_throttled_time"],
            wordpress=data["wordpress"],
        )
        # Carry the sender UID on the queued item (transient, not persisted)
        # so the flush worker releases the right per-sender reservation.
        request._peer_uid = peer_uid

        # Non-blocking enqueue with per-sender fairness. Key the cap on the
        # kernel-trusted peer UID, NOT the attacker-chosen domain: a local
        # sender has exactly one UID and cannot multiply it by rotating the
        # reported URL, so its flood cannot occupy the whole queue and
        # starve other tenants' accepted telemetry. If this UID already
        # holds its share, drop its excess; otherwise, if the queue is full,
        # drop rather than block the socket handler thread (which would
        # stall the root agent) or grow it unbounded.
        with self._sender_counts_lock:
            if self._sender_counts.get(peer_uid, 0) >= self.PER_SENDER_QUEUE_SHARE:
                self._record_drop()
                return
            try:
                self._queue.put_nowait(request)
            except Full:
                self._record_drop()
                return
            self._sender_counts[peer_uid] = self._sender_counts.get(peer_uid, 0) + 1

    def _record_drop(self) -> None:
        """Count a shed request and log periodically, without spamming."""
        self._dropped_requests += 1
        if self._dropped_requests % self.DROP_LOG_INTERVAL == 1:
            self.logger.warning(
                "Request queue full (maxsize=%s); dropped %s request(s) so far",
                self.QUEUE_MAXSIZE,
                self._dropped_requests,
            )

    def _release_sender_slot(self, obj) -> None:
        """
        Free the per-sender queue reservation once an item leaves the queue.
        Pruning zeroed entries keeps the accounting map bounded by the number
        of currently-queued senders (distinct local UIDs), not the number
        ever seen.
        """
        peer_uid = getattr(obj, "_peer_uid", None)
        with self._sender_counts_lock:
            count = self._sender_counts.get(peer_uid, 0)
            if count <= 1:
                self._sender_counts.pop(peer_uid, None)
            else:
                self._sender_counts[peer_uid] = count - 1

    def shutdown_handler(self, signum, frame):
        """
        Handle shutdown signals to flush remaining objects before exit.
        """
        self.logger.info(f"Received shutdown signal {signum}, flushing queue and buffer before shutdown")
        try:
            self._drain_queue_to_buffer()
            self.flush_remaining_objects()
        except Exception:
            self.logger.exception("Failed to flush remaining objects during shutdown")
        finally:
            sys.exit(0)

    def _drain_queue_to_buffer(self):
        """
        Drain all remaining objects from queue to buffer.
        """
        while not self._queue.empty():
            try:
                obj = self._queue.get_nowait()
                self._release_sender_slot(obj)
                self._buffer.append(obj)
            except Empty:
                break