appfactory-monitor/app/main.py

import os
import sqlite3
import threading
import time
from pathlib import Path

import requests
from fastapi import FastAPI

DB_FILE = Path("/opt/appfactory/data/appfactory/appfactory.db")

CHECK_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_MONITOR_INTERVAL_SECONDS", "30"))
REQUEST_TIMEOUT_SECONDS = int(os.getenv("APPFACTORY_MONITOR_TIMEOUT_SECONDS", "5"))
HEALTH_RETENTION_DAYS = int(os.getenv("APPFACTORY_HEALTH_RETENTION_DAYS", "7"))

CORE_SERVICES = {
    "appfactory-portal": "http://appfactory-portal:9100/health",
    "appfactory-webhook": "http://appfactory-webhook:9000/health",
    "appfactory-worker": "http://appfactory-worker:9200/health",
    "appfactory-monitor": "http://appfactory-monitor:9300/health",
}

app = FastAPI(title="AppFactory Monitor")

_monitor_thread = None
_monitor_started = False


def get_connection():
    con = sqlite3.connect(DB_FILE, timeout=30)
    con.row_factory = sqlite3.Row
    return con


def get_app_services():
    con = get_connection()

    rows = con.execute(
        """
        SELECT id
        FROM apps
        WHERE status IS NOT NULL
        ORDER BY id
        """
    ).fetchall()

    con.close()

    services = {}

    for row in rows:
        app_id = row["id"]
        services[app_id] = f"http://appfactory-caddy/apps/{app_id}/health"

    return services


def cleanup_old_health_records():
    con = get_connection()

    con.execute(
        """
        DELETE FROM service_health
        WHERE checked_at < datetime('now', ?)
        """,
        (f"-{HEALTH_RETENTION_DAYS} days",),
    )

    con.commit()
    con.close()


def save_health_result(
    service_id: str,
    status: str,
    http_status: int | None,
    response_time_ms: int | None,
    error_text: str | None,
):
    con = get_connection()

    con.execute(
        """
        INSERT INTO service_health (
            service_id,
            status,
            http_status,
            response_time_ms,
            error_text,
            checked_at
        )
        VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
        """,
        (
            service_id,
            status,
            http_status,
            response_time_ms,
            error_text,
        ),
    )

    con.commit()
    con.close()


def check_service(service_id: str, url: str):
    start = time.monotonic()

    try:
        response = requests.get(
            url,
            timeout=REQUEST_TIMEOUT_SECONDS,
        )

        elapsed_ms = int((time.monotonic() - start) * 1000)

        if 200 <= response.status_code < 300:
            save_health_result(
                service_id=service_id,
                status="healthy",
                http_status=response.status_code,
                response_time_ms=elapsed_ms,
                error_text=None,
            )
        else:
            save_health_result(
                service_id=service_id,
                status="unhealthy",
                http_status=response.status_code,
                response_time_ms=elapsed_ms,
                error_text=response.text[:500],
            )

    except Exception as exc:
        elapsed_ms = int((time.monotonic() - start) * 1000)

        save_health_result(
            service_id=service_id,
            status="unreachable",
            http_status=None,
            response_time_ms=elapsed_ms,
            error_text=str(exc),
        )


def run_check_cycle():
    cleanup_old_health_records()

    services = {}
    services.update(CORE_SERVICES)
    services.update(get_app_services())

    for service_id, url in services.items():
        check_service(service_id, url)


def monitor_loop():
    while True:
        try:
            run_check_cycle()
        except Exception as exc:
            print(f"Monitor loop error: {exc}", flush=True)

        time.sleep(CHECK_INTERVAL_SECONDS)


@app.on_event("startup")
def start_monitor():
    global _monitor_thread
    global _monitor_started

    if _monitor_started:
        return

    _monitor_started = True

    _monitor_thread = threading.Thread(
        target=monitor_loop,
        daemon=True,
    )
    _monitor_thread.start()


@app.get("/health")
def health():
    return {
        "status": "ok",
        "interval_seconds": CHECK_INTERVAL_SECONDS,
        "timeout_seconds": REQUEST_TIMEOUT_SECONDS,
        "retention_days": HEALTH_RETENTION_DAYS,
    }


@app.get("/services")
def services():
    result = {}
    result.update(CORE_SERVICES)
    result.update(get_app_services())

    return result