FabledSteward/steward/timeline/service.py

"""Reading the timeline, and keeping it: the filtered list the page, the host
card and get_timeline share, the first fill from records that predate it,
its retention, and a host rename carried onto its entries."""
from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

from sqlalchemy import delete, func, select, text, update
from sqlalchemy.dialects.postgresql import insert

from . import catalog
from .config import Config, kind_types
from steward.models.timeline import TimelineEntry

logger = logging.getLogger(__name__)

# How far back the first fill reaches: the default retention of the records it reads.
BACKFILL_DAYS = 90


def query(cfg: Config, *, host: str = "", area: str = "", kind: str = "", type_: str = "",
          since: datetime | None = None, include_hidden: bool = False):
    """The entries matching the filters, unordered (core.paging orders them).
    Hidden types are left out unless `include_hidden`, or a `type_` asks for one."""
    stmt = select(TimelineEntry)
    if host:
        stmt = stmt.where(TimelineEntry.hosts.contains([host]))
    if area:
        stmt = stmt.where(cfg.area_filter(area))
    if kind:
        stmt = stmt.where(TimelineEntry.type.in_(kind_types(kind)))
    if type_:
        stmt = stmt.where(TimelineEntry.type == type_)
    elif not include_hidden:
        hidden = cfg.hidden_types()
        if hidden:
            stmt = stmt.where(TimelineEntry.type.notin_(hidden))
    if since is not None:
        stmt = stmt.where(TimelineEntry.at >= since)
    return stmt


# The full records an entry can point to, and whether one is still kept.
def _ref_models() -> dict:
    from steward.models.alerts import AlertEvent
    from steward.models.ansible import AnsibleRun
    from steward.models.audit import AuditEvent
    return {"alert_event": AlertEvent, "ansible_run": AnsibleRun, "audit_event": AuditEvent}


async def kept_refs(db, entries) -> set[tuple[str, str]]:
    """The (ref_type, ref_id) of `entries` whose full record still exists."""
    wanted: dict[str, set[str]] = {}
    for e in entries:
        if e.ref_type and e.ref_id:
            wanted.setdefault(e.ref_type, set()).add(e.ref_id)
    kept = set()
    models = _ref_models()
    for ref_type, ids in wanted.items():
        model = models.get(ref_type)
        if model is None:           # a kind only a plugin knows: assume it is kept
            kept |= {(ref_type, i) for i in ids}
            continue
        found = (await db.execute(select(model.id).where(model.id.in_(ids)))).scalars()
        kept |= {(ref_type, str(i)) for i in found}
    return kept


def view(entry: TimelineEntry, cfg: Config, kept: set) -> dict:
    """One entry as the page and the MCP tool read it."""
    known = catalog.TYPES.get(entry.type)
    area = cfg.area_of(entry)
    detail = entry.detail or {}
    url = detail.get("url")
    gone = bool(entry.ref_type) and (entry.ref_type, entry.ref_id) not in kept
    return {
        "entry": entry,
        "type_label": known.label if known else entry.type,
        "kind": known.kind if known else None,
        "kind_label": catalog.KINDS.get(known.kind) if known else None,
        "area": area,
        "area_label": cfg.area_label(area),
        "note": detail.get("note"),
        "url": None if gone else url,
        "gone": gone,
    }


async def views(db, cfg: Config, entries) -> list[dict]:
    kept = await kept_refs(db, entries)
    return [view(e, cfg, kept) for e in entries]


async def recent_for_host(db, cfg: Config, host: str, limit: int = 8) -> list[dict]:
    """A host's newest shown entries, for its page."""
    rows = (await db.execute(
        query(cfg, host=host).order_by(TimelineEntry.at.desc(), TimelineEntry.id.desc())
        .limit(limit))).scalars().all()
    return await views(db, cfg, rows)


# ── Keeping it ──────────────────────────────────────────────────────────────

async def prune(session, now: datetime) -> tuple[int, int]:
    """Delete entries past timeline.retention_days; (rows, days), with 0 days
    keeping everything. Call inside a transaction."""
    from steward.core.settings import get_setting
    days = int(await get_setting(session, "timeline.retention_days") or 0)
    if days <= 0:
        return 0, 0
    result = await session.execute(
        delete(TimelineEntry).where(TimelineEntry.at < now - timedelta(days=days)))
    return result.rowcount or 0, days


async def carry_host_rename(session, old: str, new: str) -> int:
    """Point a renamed host's entries at its new name, so its page still
    shows its history."""
    if not old or old == new:
        return 0
    result = await session.execute(
        update(TimelineEntry).where(TimelineEntry.hosts.contains([old]))
        .values(hosts=func.array_replace(TimelineEntry.hosts, old, new))
        .execution_options(synchronize_session=False))
    return result.rowcount or 0


async def backfill(app, now: datetime | None = None) -> int:
    """Fill the timeline once from the alert events, Ansible runs and audit
    rows that predate it, BACKFILL_DAYS back, through the same builders the
    live writers use. An entry for a record already covered is skipped, so
    it is safe to run again. Records the run in timeline.backfilled_at."""
    from steward.core.settings import get_setting, set_setting
    from steward.models.alerts import AlertEvent, AlertRule
    from steward.models.ansible import AnsibleRun
    from steward.models.audit import AuditEvent
    from .record import from_alert, from_audit, from_run

    now = now or datetime.now(timezone.utc)
    async with app.db_sessionmaker() as session:
        async with session.begin():
            if await get_setting(session, "timeline.backfilled_at"):
                return 0
        since = now - timedelta(days=BACKFILL_DAYS)
        built: list[TimelineEntry] = []
        async with session.begin():
            # The builders add each entry to the session; no flush may write
            # them before the insert below, which skips any already covered.
            session.sync_session.autoflush = False

            async def build(what: str, coro) -> None:
                # One record that can't be read is skipped, not the whole fill.
                try:
                    built.append(await coro)
                except Exception:
                    logger.exception("Timeline fill: %s skipped", what)

            for event, rule in (await session.execute(
                    select(AlertEvent, AlertRule).join(AlertRule, AlertEvent.rule_id == AlertRule.id)
                    .where(AlertEvent.transitioned_at >= since))).all():
                await build(f"alert event {event.id}", from_alert(session, rule, event))
            for run in (await session.execute(
                    select(AnsibleRun).where(AnsibleRun.started_at >= since,
                                             AnsibleRun.finished_at.is_not(None)))).scalars().all():
                await build(f"Ansible run {run.id}", from_run(session, run))
            for event in (await session.execute(
                    select(AuditEvent).where(AuditEvent.timestamp >= since))).scalars().all():
                await build(f"audit row {event.id}", from_audit(session, event, _audit_detail(event)))
            rows = [e for e in built if e is not None]
            for entry in rows:
                session.expunge(entry)
            for chunk in range(0, len(rows), 500):
                await session.execute(
                    insert(TimelineEntry).values([_row(e) for e in rows[chunk:chunk + 500]])
                    .on_conflict_do_nothing(index_elements=["ref_type", "ref_id"],
                                            index_where=text("ref_id IS NOT NULL")))
            await set_setting(session, "timeline.backfilled_at", now.isoformat())
    logger.info("Timeline: filled %d entries from the last %d days of records", len(rows), BACKFILL_DAYS)
    return len(rows)


def _audit_detail(event) -> dict:
    import json
    try:
        return json.loads(event.detail_json) if event.detail_json else {}
    except (TypeError, ValueError):
        return {}


def _row(entry: TimelineEntry) -> dict:
    from steward.core.ids import time_ordered_id
    return {"id": entry.id or time_ordered_id(), "at": entry.at, "until": entry.until,
            "type": entry.type, "area": entry.area, "hosts": entry.hosts or [],
            "subject": entry.subject, "summary": entry.summary, "detail": entry.detail,
            "actor": entry.actor or "system", "ref_type": entry.ref_type, "ref_id": entry.ref_id}