squid-manager/log_storage.py

"""Persistent log storage for Squid access.log.

This module owns the bridge between raw access.log files on disk and the
SQLite-backed :class:`~app.LogEntry` table. It exposes:

- :func:`sync_log_to_db` - incrementally import new lines from a log file
- :func:`query_logs` - SQL-backed filter/search with time range support
- :func:`get_log_stats` - aggregated statistics (hit/miss, top clients...)
- :func:`get_distinct_hosts` / :func:`get_distinct_clients` /
  :func:`get_distinct_methods` - faceted browsing

Implementation notes:
- We rely on the existing :mod:`log_parser` for line format handling so we do
  not duplicate the access.log grammar.
- Bulk inserts are batched (default 500 rows) using
  :func:`sqlalchemy.orm.Session.execute` with a parameterised INSERT so we
  don't pay the round-trip cost of per-row commits.
- De-duplication uses the tuple ``(instance_id, log_time, client, raw_line)``;
  on conflict (we surface those as ``skipped_count`` rather than re-inserting).
"""

from __future__ import annotations

import os
import time
from collections import Counter, defaultdict
from datetime import datetime, timezone
from typing import Iterable

from sqlalchemy import and_, func, or_, select
from sqlalchemy.exc import SQLAlchemyError

import log_parser
from app import db, LogEntry  # imported lazily-friendly via app


# ---------------------------------------------------------------------------
# Sync
# ---------------------------------------------------------------------------


def _iter_log_lines(path: str, max_lines: int | None = None):
    """Yield raw lines from ``path`` as bytes-decoded strings.

    When ``max_lines`` is set we walk backwards through the file with a
    bounded per-iteration buffer (Deque-style window) and yield lines in
    chronological order. For an unbounded scan (full-history import) we
    just stream forwards. Designed to be tolerant of mid-stream truncation
    or binary garbage.
    """
    if not os.path.isfile(path):
        return

    if not max_lines or max_lines <= 0:
        try:
            with open(path, "r", encoding="utf-8", errors="replace") as fh:
                for raw in fh:
                    line = raw.rstrip("\n").rstrip("\r")
                    if line.strip():
                        yield line
        except OSError:
            return
        return

    # Bounded tail: read last ``max_lines`` lines efficiently
    try:
        with open(path, "rb") as fh:
            try:
                fh.seek(0, os.SEEK_END)
                size = fh.tell()
                chunks: list[bytes] = []
                lines_found = 0
                block = 65536
                pos = size
                while pos > 0 and lines_found <= max_lines:
                    read_size = min(block, pos)
                    pos -= read_size
                    fh.seek(pos)
                    chunks.append(fh.read(read_size))
                    lines_found += chunks[-1].count(b"\n")
                data = b"".join(reversed(chunks))
            except OSError:
                fh.seek(0)
                data = fh.read()
        text = data.decode("utf-8", errors="replace")
        lines = text.splitlines()
        if len(lines) > max_lines:
            lines = lines[-max_lines:]
        for line in lines:
            stripped = line.strip()
            if stripped:
                yield stripped
    except OSError:
        return


def sync_log_to_db(
    path: str,
    instance_id: int = 0,
    batch_size: int = 500,
    max_lines: int = 100000,
) -> tuple[int, int, str | None]:
    """Incrementally import new lines from ``path`` into :class:`LogEntry`.

    Steps:

    1. Compute the high-water mark ``max(existing.log_time)`` for the
       instance so we can skip already-imported lines quickly.
    2. Stream the last ``max_lines`` lines of the file
       (or the entire file when ``max_lines`` is None/0) and parse them
       through :func:`log_parser.parse_line`.
    3. Insert new rows in batches of ``batch_size`` using
       :func:`sqlalchemy.dialects.sqlite.insert` with
       ``OR (primary key uniqueness handled by auot-increment)`` semantics.
       Duplicate suppression uses an in-memory ``seen`` set keyed by
       ``(client, log_time, raw_line)`` since the file itself can repeat
       a row at rotation boundaries.
    4. Commit once per batch and return a tuple ``(inserted, skipped, err)``.

    All exceptions are caught and reported as the third tuple element so the
    caller can decide whether to surface them (e.g. on the dashboard).
    """
    if batch_size <= 0:
        batch_size = 500
    try:
        # high-water mark: only lines with log_time > wm are new
        wm = (
            db.session.query(func.max(LogEntry.log_time))
            .filter(LogEntry.instance_id == instance_id)
            .scalar()
        )
        if wm is None:
            wm = 0.0

        seen_keys: set[tuple[str, float, str]] = set()
        inserted = 0
        skipped = 0
        batch: list[dict] = []

        for raw_line in _iter_log_lines(path, max_lines=max_lines):
            e = log_parser.parse_line(raw_line)
            if not e:
                skipped += 1
                continue
            try:
                ts = float(e.get("time") or 0)
            except (TypeError, ValueError):
                skipped += 1
                continue
            if ts <= wm:
                skipped += 1
                continue
            key = (e.get("client") or "", ts, raw_line)
            if key in seen_keys:
                skipped += 1
                continue
            seen_keys.add(key)

            batch.append({
                "instance_id": instance_id,
                "log_time": ts,
                "elapsed_ms": int(e.get("elapsed_ms") or 0),
                "client": (e.get("client") or "")[:64],
                "result_code": (e.get("result_code") or "")[:32],
                "http_code": (e.get("http_code") or "")[:8],
                "size_bytes": int(e.get("size") or 0),
                "method": (e.get("method") or "")[:16],
                "url": (e.get("url") or "")[:65535],
                "host": (e.get("host") or "")[:256],
                "hier_code": (e.get("hier_code") or "")[:32],
                "content_type": (e.get("content_type") or "")[:128],
                "raw_line": raw_line[:65535],
                "created_at": datetime.now(timezone.utc),
            })
            if len(batch) >= batch_size:
                _flush_batch(batch)
                inserted += len(batch)
                batch = []

        if batch:
            _flush_batch(batch)
            inserted += len(batch)

        # record last sync time so the UI can display "上次同步"
        _record_last_sync(instance_id, path)
        return inserted, skipped, None
    except SQLAlchemyError as e:
        try:
            db.session.rollback()
        except Exception:
            pass
        return 0, 0, f"DB error: {type(e).__name__}: {e}"
    except Exception as e:
        try:
            db.session.rollback()
        except Exception:
            pass
        return 0, 0, f"{type(e).__name__}: {e}"


def _column_count() -> int:
    """Return the number of columns in the :class:`~app.LogEntry` table.

    Used to cap the per-statement row count so the single INSERT does not
    exceed SQLite's 999-boundary-variable limit per SQL statement.
    """
    return len(LogEntry.__table__.columns)


# SQLite hard limit: at most 999 ? placeholders per single SQL statement.
_SQLITE_MAX_VARIABLES = 999


def _max_rows_per_stmt() -> int:
    """Max rows that can safely fit in one parameterised INSERT."""
    cols = _column_count()
    return max(1, (_SQLITE_MAX_VARIABLES - 1) // cols)


def _flush_batch(batch: list[dict]) -> None:
    """Insert ``batch`` rows, splitting into chunks small enough for SQLite.

    SQLite rejects a single parameterised INSERT when its bound-variable
    count exceeds 999. Since :class:`~app.LogEntry` carries many columns, a
    'large' batch (e.g. 500 rows) would blow past that limit. Split the
    batch into sub-chunks of ``_max_rows_per_stmt()`` rows and execute each
    sub-statement individually, then commit once the whole batch is in.
    """
    if not batch:
        return
    from sqlalchemy.dialects.sqlite import insert as sqlite_insert

    chunk_size = _max_rows_per_stmt()
    for start in range(0, len(batch), chunk_size):
        chunk = batch[start:start + chunk_size]
        stmt = sqlite_insert(LogEntry).values(chunk)
        db.session.execute(stmt)
    db.session.commit()


# ---------------------------------------------------------------------------
# Query
# ---------------------------------------------------------------------------


def _parse_time(value) -> float | None:
    """Parse an ISO-ish datetime / unix timestamp into a unix float (UTC)."""
    if value is None or value == "":
        return None
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, datetime):
        if value.tzinfo is None:
            value = value.replace(tzinfo=timezone.utc)
        return value.timestamp()
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return None
        # datetime-local HTML input sends strings like "2024-01-31T10:00"
        # or "2024-01-31 10:00:00". Python's fromisoformat handles both.
        try:
            # Support trailing Z
            if s.endswith("Z"):
                s = s[:-1] + "+00:00"
            dt = datetime.fromisoformat(s)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return dt.timestamp()
        except ValueError:
            # fall through to float attempt
            pass
        try:
            return float(s)
        except ValueError:
            return None
    return None


def _entry_to_dict(row: LogEntry) -> dict:
    """Convert a :class:`LogEntry` row to the dict shape used downstream.

    Mirrors :class:`log_parser.LogEntry` (which is a dict subclass) so
    existing consumers - templates, alert rules, anomaly detection - keep
    working without modification.
    """
    ts = float(row.log_time or 0)
    host = row.host or ""
    url = row.url or ""
    raw = row.raw_line or ""
    # split result_code (already stored without slash) - reconstruct full
    # Squid "result" string (e.g. "TCP_HIT/200") for backward compat
    result_full = row.result_code or ""
    if row.http_code:
        result_full = f"{result_full}/{row.http_code}"
    return {
        "id": row.id,
        "time": row.log_time,                       # float unix ts
        "timestamp": datetime.fromtimestamp(ts, tz=timezone.utc) if ts > 0 else None,
        "elapsed_ms": int(row.elapsed_ms or 0),
        "client": row.client or "",
        "result": result_full,
        "result_code": row.result_code or "",
        "http_code": row.http_code or "",
        "category": log_parser.RESULT_CATEGORIES.get(row.result_code or "", "Other"),
        "result_description": log_parser.RESULT_CODE_DESCRIPTION.get(row.result_code or "", row.result_code or ""),
        "size": int(row.size_bytes or 0),
        "method": row.method or "",
        "url": url,
        "host": host,
        "ident": "",                                # not persisted separately
        "hierarchy": row.hier_code or "",
        "hier_code": row.hier_code or "",
        "peer": "",                                 # not persisted
        "content_type": row.content_type or "",
        "raw_line": raw,
        "formatted_size": log_parser.format_bytes(int(row.size_bytes or 0)),
        "formatted_duration": log_parser.format_duration(int(row.elapsed_ms or 0)),
    }


def query_logs(filters: dict) -> list[dict]:
    """Run a filtered SQL query against the persisted log rows.

    Supported keys (all optional):

    - ``instance_id`` (int)
    - ``client``, ``method``, ``result_code``, ``host`` (exact or substring)
    - ``host_filter_mode`` "exact" | "contains" (default "contains")
    - ``url`` substring LIKE
    - ``min_size``, ``max_size`` (int bytes)
    - ``min_elapsed``, ``max_elapsed`` (int milliseconds)
    - ``start_time``, ``end_time`` (datetime or unix timestamp, inclusive lo,
      inclusive hi for end so users don't accidentally exclude the minute
      they typed)
    - ``limit`` (default 1000), ``offset`` (default 0)

    Returns a list of dicts with the same shape as
    :func:`log_parser.parse_lines`, plus ``formatted_size`` /
    ``formatted_duration`` keys.
    """
    try:
        q = db.session.query(LogEntry)
        # instance scoping
        iid = filters.get("instance_id")
        if iid is not None:
            q = q.filter(LogEntry.instance_id == iid)

        c = (filters.get("client") or "").strip()
        if c:
            q = q.filter(LogEntry.client.like(f"%{c}%"))
        m = (filters.get("method") or "").strip()
        if m:
            q = q.filter(LogEntry.method == m)
        rc = (filters.get("result_code") or "").strip()
        if rc:
            q = q.filter(LogEntry.result_code.like(f"%{rc}%"))
        h = (filters.get("host") or "").strip()
        if h:
            mode = (filters.get("host_filter_mode") or "contains").lower()
            if mode == "exact":
                q = q.filter(LogEntry.host == h)
            else:
                q = q.filter(LogEntry.host.like(f"%{h}%"))
        u = (filters.get("url") or "").strip()
        if u:
            q = q.filter(LogEntry.url.like(f"%{u}%"))

        mn = filters.get("min_size")
        if mn is not None:
            q = q.filter(LogEntry.size_bytes >= int(mn))
        mx = filters.get("max_size")
        if mx is not None:
            q = q.filter(LogEntry.size_bytes <= int(mx))
        mne = filters.get("min_elapsed")
        if mne is not None:
            q = q.filter(LogEntry.elapsed_ms >= int(mne))
        mxe = filters.get("max_elapsed")
        if mxe is not None:
            q = q.filter(LogEntry.elapsed_ms <= int(mxe))

        st = _parse_time(filters.get("start_time"))
        if st is not None:
            q = q.filter(LogEntry.log_time >= st)
        et = _parse_time(filters.get("end_time"))
        if et is not None:
            q = q.filter(LogEntry.log_time <= et)

        # default ordering: newest first
        q = q.order_by(LogEntry.log_time.desc(), LogEntry.id.desc())

        limit = int(filters.get("limit") or 1000)
        offset = int(filters.get("offset") or 0)
        if limit > 0:
            q = q.limit(limit)
        if offset > 0:
            q = q.offset(offset)

        rows = q.all()
        return [_entry_to_dict(r) for r in rows]
    except SQLAlchemyError as e:
        try:
            db.session.rollback()
        except Exception:
            pass
        return []


def count_logs(filters: dict | None = None) -> int:
    """Return the number of rows matching ``filters`` (same shape as ``query_logs``).

    Used by ``logs_view`` to render total / pagination accurately without
    transferring every row over the wire.
    """
    f = dict(filters or {})
    f["limit"] = 0
    f["offset"] = 0
    try:
        # rebuild only the WHERE portion - run the same filter logic but
        # count instead of fetch
        q = _build_base_query(f)
        return q.with_entities(func.count(LogEntry.id)).scalar() or 0
    except SQLAlchemyError:
        try:
            db.session.rollback()
        except Exception:
            pass
        return 0


def _build_base_query(filters: dict):
    """Internal: same WHERE-clause as :func:`query_logs` but returns the
    SQLAlchemy query for further composition (e.g. count)."""
    q = db.session.query(LogEntry)
    iid = filters.get("instance_id")
    if iid is not None:
        q = q.filter(LogEntry.instance_id == iid)
    c = (filters.get("client") or "").strip()
    if c:
        q = q.filter(LogEntry.client.like(f"%{c}%"))
    m = (filters.get("method") or "").strip()
    if m:
        q = q.filter(LogEntry.method == m)
    rc = (filters.get("result_code") or "").strip()
    if rc:
        q = q.filter(LogEntry.result_code.like(f"%{rc}%"))
    h = (filters.get("host") or "").strip()
    if h:
        q = q.filter(LogEntry.host.like(f"%{h}%"))
    u = (filters.get("url") or "").strip()
    if u:
        q = q.filter(LogEntry.url.like(f"%{u}%"))
    mn = filters.get("min_size")
    if mn is not None:
        q = q.filter(LogEntry.size_bytes >= int(mn))
    mx = filters.get("max_size")
    if mx is not None:
        q = q.filter(LogEntry.size_bytes <= int(mx))
    mne = filters.get("min_elapsed")
    if mne is not None:
        q = q.filter(LogEntry.elapsed_ms >= int(mne))
    mxe = filters.get("max_elapsed")
    if mxe is not None:
        q = q.filter(LogEntry.elapsed_ms <= int(mxe))
    st = _parse_time(filters.get("start_time"))
    if st is not None:
        q = q.filter(LogEntry.log_time >= st)
    et = _parse_time(filters.get("end_time"))
    if et is not None:
        q = q.filter(LogEntry.log_time <= et)
    return q


# ---------------------------------------------------------------------------
# Aggregate stats
# ---------------------------------------------------------------------------


def get_log_stats(
    instance_id: int = 0,
    start_time: float | None = None,
    end_time: float | None = None,
) -> dict:
    """Compute aggregate statistics in SQL.

    Compatible with :func:`log_parser.aggregate_stats`. When there are no
    rows in scope we return the same ``_empty_stats()`` structure so the
    caller doesn't need a special case.
    """
    try:
        q = _build_base_query({
            "instance_id": instance_id,
            "start_time": start_time,
            "end_time": end_time,
        })
        total = q.with_entities(func.count(LogEntry.id)).scalar() or 0
        if total == 0:
            stats = _copy_empty_stats()
        else:
            agg = q.with_entities(
                func.coalesce(func.sum(LogEntry.size_bytes), 0),
                func.coalesce(func.sum(LogEntry.elapsed_ms), 0),
                func.min(LogEntry.log_time),
                func.max(LogEntry.log_time),
            ).one()
            total_bytes, total_elapsed, ts_min, ts_max = agg

            total = int(total)
            total_bytes = int(total_bytes)
            total_elapsed = int(total_elapsed)
            avg_elapsed = round(total_elapsed / total, 2) if total else 0.0
            avg_size = round(total_bytes / total, 2) if total else 0.0

            # result_code distribution
            by_result = dict(
                q.with_entities(
                    LogEntry.result_code,
                    func.count(LogEntry.id),
                ).group_by(LogEntry.result_code).all()
            )
            by_result = {k or "": v for k, v in by_result.items()}
            # http_code distribution (only rows with a value)
            by_http = dict(
                q.filter(LogEntry.http_code != "").with_entities(
                    LogEntry.http_code,
                    func.count(LogEntry.id),
                ).group_by(LogEntry.http_code).all()
            )
            # method distribution
            by_method = dict(
                q.with_entities(
                    LogEntry.method,
                    func.count(LogEntry.id),
                ).group_by(LogEntry.method).all()
            )
            # hierarchy distribution
            by_hier = dict(
                q.with_entities(
                    LogEntry.hier_code,
                    func.count(LogEntry.id),
                ).group_by(LogEntry.hier_code).all()
            )

            # top clients: most active + bytes
            top_clients = q.with_entities(
                LogEntry.client,
                func.count(LogEntry.id).label("c"),
                func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"),
            ).group_by(LogEntry.client).order_by(func.count(LogEntry.id).desc()).limit(50).all()
            top_clients = [{"client": c or "", "count": int(c_), "bytes": int(b_)} for c, c_, b_ in top_clients]
            # top hosts
            top_hosts_rows = q.filter(LogEntry.host != "").with_entities(
                LogEntry.host,
                func.count(LogEntry.id).label("c"),
                func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"),
            ).group_by(LogEntry.host).order_by(func.count(LogEntry.id).desc()).limit(50).all()
            top_hosts = [{"host": h, "count": int(c_), "bytes": int(b_)} for h, c_, b_ in top_hosts_rows]
            # top urls
            top_urls_rows = q.with_entities(
                LogEntry.url,
                func.count(LogEntry.id).label("c"),
            ).group_by(LogEntry.url).order_by(func.count(LogEntry.id).desc()).limit(50).all()
            top_urls = [{"url": u, "count": int(c_)} for u, c_ in top_urls_rows]

            # content-type: simpler approach - just count distinct top values
            by_ctype_rows = q.with_entities(
                LogEntry.content_type,
                func.count(LogEntry.id).label("c"),
            ).group_by(LogEntry.content_type).order_by(func.count(LogEntry.id).desc()).limit(20).all()
            by_ctype = []
            for ct, c_ in by_ctype_rows:
                ct_main = (ct or "-").split(";")[0].strip() or "-"
                by_ctype.append((ct_main, int(c_)))
            # collapse same main types
            merged: dict[str, int] = defaultdict(int)
            for ct_main, c_ in by_ctype:
                merged[ct_main] += c_
            by_ctype = list(merged.items())[:20]

            # hit/miss ratio using result_code categories
            hits = sum(c for code, c in by_result.items()
                       if log_parser.RESULT_CATEGORIES.get(code or "") == "Hit")
            misses = sum(c for code, c in by_result.items()
                         if log_parser.RESULT_CATEGORIES.get(code or "") == "Miss")
            denied = sum(c for code, c in by_result.items()
                         if log_parser.RESULT_CATEGORIES.get(code or "") == "Denied")
            aborted = by_result.get("TCP_MISS_ABORTED", 0)
            denied_rate = denied / total if total else 0.0
            aborted_rate = aborted / total if total else 0.0
            hit_ratio = hits / (hits + misses) if (hits + misses) > 0 else 0.0
            errors_4xx = sum(c for code, c in by_http.items() if code.startswith("4"))
            errors_5xx = sum(c for code, c in by_http.items() if code.startswith("5"))
            errors_4xx_rate = errors_4xx / total if total else 0.0
            errors_5xx_rate = errors_5xx / total if total else 0.0

            # unique clients / hosts
            unique_clients = q.with_entities(func.count(func.distinct(LogEntry.client))).scalar() or 0
            unique_hosts = q.filter(LogEntry.host != "").with_entities(
                func.count(func.distinct(LogEntry.host))
            ).scalar() or 0

            # hourly distribution (UTC hour of day)
            by_hour_counts: dict[int, int] = defaultdict(int)
            by_hour_bytes: dict[int, int] = defaultdict(int)
            # we already have min/max so size of an in-Python bucket is small
            # but to avoid pulling all rows we can do an UNION-ish query per
            # hour; the simpler approach: fetch (time, size) projection.
            rows = q.with_entities(LogEntry.log_time, LogEntry.size_bytes).all()
            for ts, size in rows:
                if not ts:
                    continue
                hr = datetime.fromtimestamp(float(ts), tz=timezone.utc).hour
                by_hour_counts[hr] += 1
                by_hour_bytes[hr] += int(size or 0)

            time_start = datetime.fromtimestamp(float(ts_min), tz=timezone.utc) if ts_min else None
            time_end = datetime.fromtimestamp(float(ts_max), tz=timezone.utc) if ts_max else None

            stats = {
                "total_requests": total,
                "total_bytes": total_bytes,
                "total_elapsed_ms": total_elapsed,
                "unique_clients": int(unique_clients),
                "unique_hosts": int(unique_hosts),
                "hit_ratio": hit_ratio,
                "denied_rate": denied_rate,
                "aborted_rate": aborted_rate,
                "errors_4xx_rate": errors_4xx_rate,
                "errors_5xx_rate": errors_5xx_rate,
                "errors_4xx_count": errors_4xx,
                "errors_5xx_count": errors_5xx,
                "avg_elapsed_ms": avg_elapsed,
                "avg_size_bytes": avg_size,
                "hits": hits,
                "misses": misses,
                "denied": denied,
                "aborted": aborted,
                "time_start": time_start,
                "time_end": time_end,
                "by_result": log_parser._top_n(Counter(by_result), 50),
                "by_http": log_parser._top_n(Counter(by_http), 50),
                "by_method": log_parser._top_n(Counter(by_method), 20),
                "by_category": [],  # we don't store category; compute inline if needed
                "by_hierarchy": log_parser._top_n(Counter(by_hier), 20),
                "by_ctype": by_ctype,
                "top_clients": top_clients,
                "top_hosts": top_hosts,
                "top_urls": top_urls,
                "hourly": [
                    {"hour": h, "requests": int(by_hour_counts.get(h, 0)),
                     "bytes": int(by_hour_bytes.get(h, 0))}
                    for h in range(24)
                ],
                "result_code_legend": log_parser.RESULT_CODE_DESCRIPTION,
                "hierarchy_legend": log_parser.HIERARCHY_LABELS,
            }

        # augment with sync timestamp for the dashboard UI
        try:
            from app import AppConfig
            ts_str = AppConfig.query.filter_by(key="_log_last_sync_at").first()
            stats["last_sync_at"] = ts_str.value if ts_str else None
        except Exception:
            stats["last_sync_at"] = None
        return stats
    except SQLAlchemyError as e:
        try:
            db.session.rollback()
        except Exception:
            pass
        s = _copy_empty_stats()
        s["last_sync_at"] = None
        s["error"] = f"{type(e).__name__}: {e}"
        return s


def _copy_empty_stats() -> dict:
    """Return a fresh empty-stats dict (avoid mutating log_parser._empty_stats)."""
    return log_parser._empty_stats()


# ---------------------------------------------------------------------------
# Distinct facets
# ---------------------------------------------------------------------------


def get_distinct_hosts(instance_id: int = 0, limit: int = 200) -> list[str]:
    try:
        rows = (
            db.session.query(LogEntry.host)
            .filter(LogEntry.host != "", LogEntry.instance_id == instance_id)
            .distinct()
            .order_by(LogEntry.host.asc())
            .limit(int(limit))
            .all()
        )
        return [r[0] for r in rows if r[0]]
    except SQLAlchemyError:
        try:
            db.session.rollback()
        except Exception:
            pass
        return []


def get_distinct_clients(instance_id: int = 0, limit: int = 200) -> list[str]:
    try:
        rows = (
            db.session.query(LogEntry.client)
            .filter(LogEntry.client != "", LogEntry.instance_id == instance_id)
            .distinct()
            .order_by(LogEntry.client.asc())
            .limit(int(limit))
            .all()
        )
        return [r[0] for r in rows if r[0]]
    except SQLAlchemyError:
        try:
            db.session.rollback()
        except Exception:
            pass
        return []


def get_distinct_methods(instance_id: int = 0) -> list[str]:
    try:
        rows = (
            db.session.query(LogEntry.method)
            .filter(LogEntry.method != "", LogEntry.instance_id == instance_id)
            .distinct()
            .order_by(LogEntry.method.asc())
            .all()
        )
        return [r[0] for r in rows if r[0]]
    except SQLAlchemyError:
        try:
            db.session.rollback()
        except Exception:
            pass
        return []


# ---------------------------------------------------------------------------
# Sync bookkeeping
# ---------------------------------------------------------------------------


_LAST_SYNC_KEY = "_log_last_sync_at"


def _record_last_sync(instance_id: int, path: str) -> None:
    """Stamp ``AppConfig`` rows with the last successful sync timestamp.

    Two values are stored: ``_log_last_sync_at`` (ISO timestamp) and
    ``_log_last_sync_path`` (the file we synced from). We use ``AppConfig``
    rather than ``LogEntry`` so the timestamp survives even if log rows
    are later rotated out.
    """
    try:
        from app import AppConfig
        ts_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        for key, value in (
            (_LAST_SYNC_KEY, ts_iso),
            ("_log_last_sync_path", path or ""),
            ("_log_last_sync_instance", str(instance_id)),
        ):
            row = AppConfig.query.filter_by(key=key).first()
            if row is None:
                row = AppConfig(key=key, value=value)
                db.session.add(row)
            else:
                row.value = value
        db.session.commit()
    except Exception:
        try:
            db.session.rollback()
        except Exception:
            pass


def get_last_sync_at() -> str | None:
    """Return the ISO timestamp of the last sync, or ``None``."""
    try:
        from app import AppConfig
        row = AppConfig.query.filter_by(key=_LAST_SYNC_KEY).first()
        return row.value if row else None
    except Exception:
        return None


def get_last_sync_path() -> str | None:
    try:
        from app import AppConfig
        row = AppConfig.query.filter_by(key="_log_last_sync_path").first()
        return row.value if row else None
    except Exception:
        return None


def get_last_sync_at_local() -> str | None:
    """Return the last-sync timestamp formatted in local time (not UTC).

    :class:`~app.AppConfig` stores the sync time as an ISO-8601 UTC string
    (e.g. ``2026-07-16T07:43:36Z``). This converts it to the system's local
    timezone so the UI shows a human-friendly time to the operator.
    """
    ts = get_last_sync_at()
    if not ts:
        return None
    try:
        dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
        return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        return ts


def import_full_history(
    path: str,
    instance_id: int = 0,
    batch_size: int = 500,
    progress_cb=None,
) -> tuple[int, int, str | None]:
    """Import the entire ``access.log`` from byte 0 (not just the tail).

    Used by the "Import history" button on the logs page. Calls
    :func:`sync_log_to_db` with a very high ``max_lines`` (50M) so we
    stream every line. Progress is reported via ``progress_cb(processed)``
    so the caller can render a status indicator.
    """
    try:
        # 50M lines ceiling - enough for multi-year logs; streaming means
        # we never actually load them all into memory.
        return sync_log_to_db(
            path=path,
            instance_id=instance_id,
            batch_size=batch_size,
            max_lines=50_000_000,
        )
    except Exception as e:
        return 0, 0, f"{type(e).__name__}: {e}"