import asyncio
import os
from pathlib import Path
from typing import Callable, Awaitable, Optional


class LogWatcher:
    POLL_INTERVAL = 0.1
    FILE_RETRY_INTERVAL = 1.0
    SEED_BYTES = 8192  # read last ~8 KB on open to seed initial state

    def __init__(
        self,
        path: str,
        on_line: Callable[[str], Awaitable[None]],
        on_offline: Optional[Callable[[], Awaitable[None]]] = None,
        offline_timeout: float = 60.0,
        reopen: bool = False,
    ):
        self._path = Path(path)
        self._on_line = on_line
        self._on_offline = on_offline
        self._offline_timeout = offline_timeout
        self._reopen = reopen
        self._task: Optional[asyncio.Task] = None

    def start(self) -> asyncio.Task:
        self._task = asyncio.create_task(self._run())
        return self._task

    def stop(self):
        if self._task:
            self._task.cancel()

    async def _run(self):
        offline_since: Optional[float] = None

        if self._reopen:
            await self._run_reopen()
        else:
            await self._run_keepopen()

    async def _run_keepopen(self):
        """Default: keep file open and poll readline(). Works on local filesystems."""
        offline_since: Optional[float] = None

        while True:
            try:
                with open(self._path, 'r', errors='replace') as f:
                    f.seek(0, 2)
                    end = f.tell()
                    f.seek(max(0, end - self.SEED_BYTES))
                    if f.tell() > 0:
                        f.readline()  # discard partial first line
                    offline_since = None
                    while True:
                        line = f.readline()
                        if line:
                            try:
                                await self._on_line(line)
                            except Exception:
                                pass  # keep tailing even if a line causes a parser error
                        else:
                            await asyncio.sleep(self.POLL_INTERVAL)
            except OSError:
                now = asyncio.get_event_loop().time()
                if offline_since is None:
                    offline_since = now
                elif now - offline_since >= self._offline_timeout and self._on_offline:
                    await self._on_offline()
                    offline_since = None
                await asyncio.sleep(min(self.FILE_RETRY_INTERVAL, self._offline_timeout))
            except asyncio.CancelledError:
                return

    async def _run_reopen(self):
        """Close and reopen each poll cycle. Use for network filesystems (e.g. sshfs)
        that cache file content at open time and don't propagate remote appends to
        held-open file descriptors."""
        offline_since: Optional[float] = None
        pos: int = -1  # -1 = not yet seeded

        while True:
            try:
                with open(self._path, 'r', errors='replace') as f:
                    f.seek(0, 2)
                    end = f.tell()

                    if pos < 0:
                        f.seek(max(0, end - self.SEED_BYTES))
                        if f.tell() > 0:
                            f.readline()  # discard partial first line
                        pos = f.tell()
                    elif end < pos:
                        pos = 0  # file truncated / rotated

                    f.seek(pos)
                    while True:
                        line = f.readline()
                        if line:
                            pos = f.tell()
                            try:
                                await self._on_line(line)
                            except Exception:
                                pass  # keep tailing even if a line causes a parser error
                        else:
                            break

                offline_since = None
                await asyncio.sleep(self.POLL_INTERVAL)
            except OSError:
                now = asyncio.get_event_loop().time()
                if offline_since is None:
                    offline_since = now
                elif now - offline_since >= self._offline_timeout and self._on_offline:
                    await self._on_offline()
                    offline_since = None
                await asyncio.sleep(min(self.FILE_RETRY_INTERVAL, self._offline_timeout))
            except asyncio.CancelledError:
                return