Queue webhook deployments as jobs

app/main.py

@@ -2,12 +2,14 @@ import hashlib
 import hmac
 import json
 import os
-import subprocess
+import sqlite3
-import threading
 
 from fastapi import FastAPI, Header, HTTPException, Request
 
 
+DB_FILE = "/opt/appfactory/data/appfactory/appfactory.db"
+
+
 def read_env_file_value(key: str, default: str = "") -> str:
     try:
         with open("/opt/appfactory/config/appfactory.env", "r", encoding="utf-8") as f:
@@ -23,12 +25,10 @@ def read_env_file_value(key: str, default: str = "") -> str:
 
 WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", read_env_file_value("WEBHOOK_SECRET", ""))
 
-APP_DEPLOY_SCRIPT = "/tools/deploy-app.sh"
-CORE_DEPLOY_SCRIPT = "/tools/deploy-core-service.sh"
-
 CORE_SERVICES = {
     "appfactory-portal",
     "appfactory-webhook",
+    "appfactory-worker",
 }
 
 app = FastAPI(title="AppFactory Webhook")
@@ -59,6 +59,7 @@ def first_commit(payload: dict) -> dict:
     commits = payload.get("commits") or []
     if commits:
         return commits[-1] or {}
+
     return payload.get("head_commit") or {}
 
 
@@ -104,28 +105,78 @@ def extract_attribution(payload: dict) -> dict:
     }
 
 
-def run_command(command: list[str], repo_name: str, deploy_env: dict[str, str]):
+def create_job(
-    env = os.environ.copy()
+    job_type: str,
-    env.update(deploy_env)
+    target_type: str,
+    target_id: str,
+    payload: dict,
+    created_by_username: str,
+    created_by_display_name: str,
+    source: str,
+) -> int:
+    con = sqlite3.connect(DB_FILE)
 
-    print(f"Deploy started: {repo_name}", flush=True)
+    cur = con.execute(
-    print(f"Command: {' '.join(command)}", flush=True)
+        """
-    print(f"Trigger source: {env.get('APPFACTORY_TRIGGER_SOURCE')}", flush=True)
+        INSERT INTO jobs (
-    print(f"Pusher: {env.get('APPFACTORY_PUSHER')}", flush=True)
+            type,
-    print(f"Commit author: {env.get('APPFACTORY_COMMIT_AUTHOR')}", flush=True)
+            target_type,
+            target_id,
+            payload_json,
+            status,
+            created_by_username,
+            created_by_display_name,
+            source
+        )
+        VALUES (?, ?, ?, ?, 'queued', ?, ?, ?)
+        """,
+        (
+            job_type,
+            target_type,
+            target_id,
+            json.dumps(payload, ensure_ascii=False),
+            created_by_username,
+            created_by_display_name,
+            source,
+        ),
+    )
 
-    result = subprocess.run(command, capture_output=True, text=True, env=env)
+    job_id = cur.lastrowid
 
-    print(f"Deploy finished: {repo_name}", flush=True)
+    con.execute(
-    print(f"Return code: {result.returncode}", flush=True)
+        """
+        INSERT INTO audit_events (
+            username,
+            action,
+            target_type,
+            target_id,
+            source,
+            metadata,
+            created_at
+        )
+        VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+        """,
+        (
+            created_by_username,
+            "webhook.deploy.queued",
+            target_type,
+            target_id,
+            "webhook",
+            json.dumps(
+                {
+                    "job_id": job_id,
+                    "job_type": job_type,
+                    **payload,
+                },
+                ensure_ascii=False,
+            ),
+        ),
+    )
 
-    if result.stdout:
+    con.commit()
-        print("STDOUT:", flush=True)
+    con.close()
-        print(result.stdout, flush=True)
 
-    if result.stderr:
+    return job_id
-        print("STDERR:", flush=True)
-        print(result.stderr, flush=True)
 
 
 @app.post("/gitea")
@@ -149,27 +200,34 @@ async def gitea_webhook(
     attr = extract_attribution(payload)
 
     if repo_name in CORE_SERVICES:
-        command = [CORE_DEPLOY_SCRIPT, repo_name]
+        job_type = "deploy_core_service"
+        target_type = "core_service"
     else:
-        command = [APP_DEPLOY_SCRIPT, repo_name]
+        job_type = "deploy_app"
+        target_type = "app"
 
-    deploy_env = {
+    ref = payload.get("ref") or ""
-        "APPFACTORY_TRIGGER_SOURCE": "webhook",
+    job_payload = {
-        "APPFACTORY_TRIGGERED_BY_USERNAME": attr["pusher_username"],
+        "repository": repo_name,
-        "APPFACTORY_TRIGGERED_BY_DISPLAY_NAME": attr["pusher"],
+        "ref": ref,
-        "APPFACTORY_COMMIT_AUTHOR": attr["commit_author"],
+        "commit_sha": attr["commit_sha"],
-        "APPFACTORY_PUSHER": attr["pusher"],
+        "commit_author": attr["commit_author"],
-        "APPFACTORY_COMMIT_SHA": attr["commit_sha"],
+        "pusher": attr["pusher"],
     }
 
-    threading.Thread(
+    job_id = create_job(
-        target=run_command,
+        job_type=job_type,
-        args=(command, repo_name, deploy_env),
+        target_type=target_type,
-        daemon=True,
+        target_id=repo_name,
-    ).start()
+        payload=job_payload,
+        created_by_username=attr["pusher_username"],
+        created_by_display_name=attr["pusher"],
+        source="webhook",
+    )
 
     return {
-        "status": "accepted",
+        "status": "queued",
+        "job_id": job_id,
         "repo": repo_name,
         "core": repo_name in CORE_SERVICES,
         "triggered_by": attr["pusher_username"],