appfactory-worker/app/main.py

import json
import os
import socket
import sqlite3
import subprocess
import threading
import time
from datetime import datetime, timezone
from pathlib import Path

from fastapi import FastAPI

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

WORKER_ID = os.getenv("APPFACTORY_WORKER_ID", socket.gethostname())
POLL_SECONDS = int(os.getenv("APPFACTORY_WORKER_POLL_SECONDS", "3"))

app = FastAPI(title="AppFactory Worker")

_worker_thread = None
_worker_started = False


def utc_now():
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")


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


def append_job_log(job_id: int, stream: str, message: str):
    con = get_connection()
    con.execute(
        """
        INSERT INTO job_logs (job_id, stream, message, created_at)
        VALUES (?, ?, ?, CURRENT_TIMESTAMP)
        """,
        (job_id, stream, message),
    )
    con.commit()
    con.close()


def update_job_status(job_id: int, status: str, error_text: str | None = None, result: dict | None = None):
    con = get_connection()

    if status == "running":
        con.execute(
            """
            UPDATE jobs
            SET status = 'running',
                started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
                worker_id = ?
            WHERE id = ?
            """,
            (WORKER_ID, job_id),
        )
    elif status in ("success", "failed", "cancelled"):
        con.execute(
            """
            UPDATE jobs
            SET status = ?,
                finished_at = CURRENT_TIMESTAMP,
                error_text = ?,
                result_json = ?
            WHERE id = ?
            """,
            (
                status,
                error_text,
                json.dumps(result or {}, ensure_ascii=False),
                job_id,
            ),
        )

    con.commit()
    con.close()


def get_next_job():
    con = get_connection()

    con.execute("BEGIN IMMEDIATE")

    row = con.execute(
        """
        SELECT *
        FROM jobs
        WHERE status = 'queued'
        ORDER BY id ASC
        LIMIT 1
        """
    ).fetchone()

    if not row:
        con.commit()
        con.close()
        return None

    job_id = row["id"]

    running_same_target = con.execute(
        """
        SELECT id
        FROM jobs
        WHERE status = 'running'
          AND target_type = ?
          AND target_id = ?
          AND id != ?
        LIMIT 1
        """,
        (row["target_type"], row["target_id"], job_id),
    ).fetchone()

    if running_same_target:
        con.commit()
        con.close()
        return None

    con.execute(
        """
        UPDATE jobs
        SET status = 'running',
            started_at = CURRENT_TIMESTAMP,
            worker_id = ?
        WHERE id = ?
          AND status = 'queued'
        """,
        (WORKER_ID, job_id),
    )

    con.commit()
    con.close()

    return dict(row)


def build_env(job: dict):
    payload = json.loads(job.get("payload_json") or "{}")

    env = os.environ.copy()
    env["APPFACTORY_TRIGGER_SOURCE"] = job.get("source") or "worker"
    env["APPFACTORY_TRIGGERED_BY_USER_ID"] = str(job.get("created_by_user_id") or "")
    env["APPFACTORY_TRIGGERED_BY_USERNAME"] = str(job.get("created_by_username") or "")
    env["APPFACTORY_TRIGGERED_BY_DISPLAY_NAME"] = str(job.get("created_by_display_name") or "")
    env["APPFACTORY_COMMIT_AUTHOR"] = str(payload.get("commit_author") or "")
    env["APPFACTORY_PUSHER"] = str(payload.get("pusher") or "")
    env["APPFACTORY_COMMIT_SHA"] = str(payload.get("commit_sha") or "")

    return env


def command_for_job(job: dict):
    job_type = job["type"]
    target_id = job["target_id"]

    if job_type == "deploy_app":
        return [str(TOOLS_DIR / "deploy-app.sh"), target_id]

    if job_type == "deploy_core_service":
        return [str(TOOLS_DIR / "deploy-core-service.sh"), target_id]

    raise ValueError(f"Unsupported job type: {job_type}")


def execute_job(job: dict):
    job_id = int(job["id"])
    target = job["target_id"]

    append_job_log(job_id, "system", f"Worker {WORKER_ID} started job {job_id} for {target}")
    update_job_status(job_id, "running")

    try:
        command = command_for_job(job)
        env = build_env(job)

        append_job_log(job_id, "system", "Command: " + " ".join(command))

        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            env=env,
        )

        stdout_chunks = []
        stderr_chunks = []

        stdout, stderr = process.communicate()

        if stdout:
            stdout_chunks.append(stdout)
            append_job_log(job_id, "stdout", stdout)

        if stderr:
            stderr_chunks.append(stderr)
            append_job_log(job_id, "stderr", stderr)

        returncode = process.returncode

        if returncode == 0:
            update_job_status(
                job_id,
                "success",
                result={
                    "returncode": returncode,
                    "stdout_length": len("".join(stdout_chunks)),
                    "stderr_length": len("".join(stderr_chunks)),
                },
            )
            append_job_log(job_id, "system", "Job finished successfully")
        else:
            update_job_status(
                job_id,
                "failed",
                error_text=f"Command failed with return code {returncode}",
                result={"returncode": returncode},
            )
            append_job_log(job_id, "system", f"Job failed with return code {returncode}")

    except Exception as exc:
        update_job_status(job_id, "failed", error_text=str(exc))
        append_job_log(job_id, "stderr", str(exc))


def worker_loop():
    while True:
        try:
            job = get_next_job()

            if job:
                execute_job(job)
            else:
                time.sleep(POLL_SECONDS)

        except Exception as exc:
            print(f"Worker loop error: {exc}", flush=True)
            time.sleep(POLL_SECONDS)


@app.on_event("startup")
def start_worker():
    global _worker_thread
    global _worker_started

    if _worker_started:
        return

    _worker_started = True
    _worker_thread = threading.Thread(target=worker_loop, daemon=True)
    _worker_thread.start()


@app.get("/health")
def health():
    return {
        "status": "ok",
        "worker_id": WORKER_ID,
    }