"""单工作线程同步调度；持久待同步记录由调用方管理。"""
import math
import threading
import time


class SyncScheduler:
    MAX_PENDING_NAMES = 128

    def __init__(self, sync, interval, on_error, cooldown=None):
        self._sync = sync
        self._interval = interval
        self._on_error = on_error
        self._cooldown = cooldown or (lambda: 0)
        self._condition = threading.Condition()
        self._state = 'new'
        self._thread = None
        self._pending = set()
        self._all_pending = False

    def _delay(self):
        delay = float(self._interval())
        if not math.isfinite(delay) or delay <= 0:
            raise ValueError('invalid sync interval')
        return delay

    def _start_locked(self, restart):
        if self._state == 'running':
            return False
        if self._state == 'stopping' or (self._state == 'stopped' and not restart):
            return False
        self._delay()
        self._state = 'running'
        self._thread = threading.Thread(target=self._run, name='ranking-hf-sync', daemon=True)
        try:
            self._thread.start()
        except Exception:
            self._state = 'new'
            self._thread = None
            raise
        return True

    def start(self, restart=False):
        with self._condition:
            return self._start_locked(restart)

    def request(self, file_name):
        with self._condition:
            if self._state in {'stopping', 'stopped'}:
                return False
            if not self._all_pending:
                self._pending.add(file_name)
                if len(self._pending) > self.MAX_PENDING_NAMES:
                    self._pending.clear()
                    self._all_pending = True
            self._start_locked(False)
            self._condition.notify_all()
            return True

    def close(self):
        with self._condition:
            thread = self._thread
            if thread is threading.current_thread():
                raise RuntimeError('sync worker cannot join itself')
            self._state = 'stopping'
            self._pending.clear()
            self._all_pending = False
            self._condition.notify_all()
        if thread is not None:
            thread.join()
        with self._condition:
            if self._thread is thread:
                self._state = 'stopped'
                self._thread = None

    def _run(self):
        try:
            deadline = time.monotonic() + self._delay()
            while True:
                with self._condition:
                    while self._state == 'running':
                        periodic = time.monotonic() >= deadline
                        if periodic or self._pending or self._all_pending:
                            cooldown = float(self._cooldown())
                            if not math.isfinite(cooldown) or cooldown < 0:
                                raise ValueError('invalid sync cooldown')
                            if cooldown > 0:
                                # 远期服务端日期也不能超过平台锁等待上限。
                                self._condition.wait(min(cooldown, 60))
                                continue
                            names = None if periodic or self._all_pending else sorted(self._pending)
                            self._pending.clear()
                            self._all_pending = False
                            break
                        self._condition.wait(min(60, max(0, deadline - time.monotonic())))
                    else:
                        return
                try:
                    self._sync(names)
                except Exception as error:
                    self._on_error(error)
                # 即时请求不能无限推迟正常批量窗口。
                if periodic:
                    deadline = time.monotonic() + self._delay()
        finally:
            with self._condition:
                if self._state == 'running':
                    self._state = 'stopped'
                self._condition.notify_all()