comfy-web/app.py

#!/usr/bin/env python3
"""
comfy-web — web GUI + payment frontend for the ComfyUI app registry.

Enumerates comfy_apps.APPS, takes payment in sats via BTCPay, runs the app
against nightmare's ComfyUI (GPU), and delivers the image.

Run: python3 app.py  (dev, port 5188)
"""
import os, json, time, sqlite3, threading, uuid, uuid as _uuid
import urllib.request, urllib.parse
from flask import Flask, request, render_template, redirect, url_for, jsonify, send_from_directory, abort

import sys
sys.path.insert(0, os.path.expanduser("~/auto-publisher/core"))
import comfy_apps as CA
import comfy_workflows_lib as CL

# ---------------------------------------------------------------------------
# config
# ---------------------------------------------------------------------------
BASE = os.path.expanduser("~/comfy-web")
DB = os.path.join(BASE, "jobs.db")
UPLOAD_DIR = os.path.join(BASE, "uploads")
RESULT_DIR = os.path.join(BASE, "results")
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(RESULT_DIR, exist_ok=True)

# BTCPay (CT133). API calls use the LAN IP (public domain has CF 1010 block for non-browser);
# checkout redirects use the public domain. Empty => MOCK mode.
BTCPAY_API = "https://10.30.20.140"          # server-to-server API (self-signed, ssl unverified)
BTCPAY_PUBLIC = "https://btcpay.thetempleofdoom.com"   # customer-facing checkout
BTCPAY_STORE_ID = "6AZoSSdADJ8NndwLDhhh3geg6BgS487ZCkW3msX2mT1"
BTCPAY_API_KEY = "18ebeb23d41e971e7a133479a7c0644f78a337fc"
WEBHOOK_URL = "https://nebula.thetempleofdoom.com"

MOCK_PAYMENTS = not (BTCPAY_API and BTCPAY_STORE_ID and BTCPAY_API_KEY)

import ssl
_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE

app = Flask(__name__)

# ---------------------------------------------------------------------------
# db
# ---------------------------------------------------------------------------
def db():
    c = sqlite3.connect(DB)
    c.row_factory = sqlite3.Row
    return c

def init_db():
    with db() as c:
        c.execute("""CREATE TABLE IF NOT EXISTS jobs (
            id TEXT PRIMARY KEY,
            app TEXT, prompt TEXT, params TEXT,
            status TEXT,          -- queued | running | done | failed | awaiting_payment
            btcpay_invoice TEXT, result_file TEXT, error TEXT,
            created_at REAL, paid_at REAL, finished_at REAL
        )""")

init_db()

# ---------------------------------------------------------------------------
# payment (BTCPay Greenfield API, or mock)
# ---------------------------------------------------------------------------
def create_invoice(job_id, app_name, sats):
    price = CA.APPS[app_name]["price_sats"]
    amount = f"{sats / 100_000_000:.8f}"  # sats -> BTC
    if MOCK_PAYMENTS:
        return f"mock-{job_id}"
    body = json.dumps({
        "amount": amount, "currency": "BTC",
        "metadata": {"job_id": job_id, "app": app_name},
        "checkout": {"redirectURL": f"{WEBHOOK_URL}/job/{job_id}", "requiresRefundEmail": False},
    }).encode()
    req = urllib.request.Request(
        f"{BTCPAY_API}/api/v1/stores/{BTCPAY_STORE_ID}/invoices",
        data=body, headers={"Authorization": f"token {BTCPAY_API_KEY}", "Content-Type": "application/json"},
    )
    r = urllib.request.urlopen(req, timeout=30, context=_CTX)
    inv = json.loads(r.read())
    return inv["id"]

def settle_job(job_id):
    """Mark paid and queue for execution (idempotent — no double-credit)."""
    with db() as c:
        cur = c.execute("UPDATE jobs SET status='queued', paid_at=? WHERE id=? AND status='awaiting_payment'",
                        (time.time(), job_id))
        if cur.rowcount == 0:
            return  # already settled (replay guard)
    with db() as c:
        row = c.execute("SELECT id, app, prompt, params FROM jobs WHERE id=?", (job_id,)).fetchone()
    if row:
        _queue.append(row)

# ---------------------------------------------------------------------------
# job worker (serializes the GPU)
# ---------------------------------------------------------------------------
_queue = []

def _worker():
    while True:
        if not _queue:
            time.sleep(1)
            continue
        row = _queue.pop(0)
        job_id, app_name, prompt, params = row["id"], row["app"], row["prompt"], json.loads(row["params"] or "{}")
        try:
            with db() as c:
                c.execute("UPDATE jobs SET status='running' WHERE id=?", (job_id,))
            # free the GPU (unload Ollama + switch to ComfyUI OC) before rendering
            CL.free_gpu()
            # run the app
            fn = CA.APPS[app_name]["fn"]
            params = dict(params)
            if prompt:
                params["prompt"] = prompt
            wf = fn(**params)
            pid, err = CL.submit(wf)
            if err:
                raise RuntimeError(str(err)[:400])
            paths = CL.fetch_result(pid, timeout=900)
            if not paths:
                raise RuntimeError("job timed out")
            # move result into results dir under job id
            src = paths[0]
            ext = os.path.splitext(src)[1] or ".png"
            dest = os.path.join(RESULT_DIR, job_id + ext)
            os.rename(src, dest)
            with db() as c:
                c.execute("UPDATE jobs SET status='done', result_file=?, finished_at=? WHERE id=?",
                          (os.path.basename(dest), time.time(), job_id))
            CL.restore_gpu()
        except Exception as e:
            CL.restore_gpu()
            with db() as c:
                c.execute("UPDATE jobs SET status='failed', error=?, finished_at=? WHERE id=?",
                          (str(e)[:500], time.time(), job_id))

threading.Thread(target=_worker, daemon=True).start()

# ---------------------------------------------------------------------------
# routes
# ---------------------------------------------------------------------------
@app.route("/")
def index():
    apps = CA.list_apps()
    return render_template("index.html", apps=apps)

@app.route("/app/<name>")
def app_detail(name):
    if name not in CA.APPS:
        abort(404)
    meta = CA.list_apps()[name]
    return render_template("app.html", name=name, meta=meta)

@app.route("/run", methods=["POST"])
def run_job():
    app_name = request.form.get("app")
    if app_name not in CA.APPS:
        abort(400)
    prompt = request.form.get("prompt", "")
    # gather params + handle uploaded source image
    params = {}
    img = request.files.get("image")
    if img and img.filename:
        local = os.path.join(UPLOAD_DIR, uuid.uuid4().hex + os.path.splitext(img.filename)[1])
        img.save(local)
        remote = os.path.basename(local)
        CL.upload_image(local, name=remote)
        params["image_name"] = remote
        # for inpaint, a mask may come too
        mask = request.files.get("mask")
        if mask and mask.filename:
            mlocal = os.path.join(UPLOAD_DIR, uuid.uuid4().hex + os.path.splitext(mask.filename)[1])
            mask.save(mlocal)
            CL.upload_image(mlocal, name=os.path.basename(mlocal))
            params["mask_name"] = os.path.basename(mlocal)
    # collect numeric/select params
    for key in ("seed", "steps", "denoise", "upscaler", "control_type", "weight", "left", "right", "top", "bottom",
                "w", "h", "cfg", "strength"):
        if key in request.form and request.form[key] != "":
            v = request.form[key]
            params[key] = int(v) if v.lstrip("-").isdigit() else v
    if request.form.get("negative", "").strip():
        params["negative"] = request.form["negative"].strip()
    job_id = uuid.uuid4().hex
    with db() as c:
        c.execute("INSERT INTO jobs (id, app, prompt, params, status, created_at) VALUES (?,?,?,?,?,?)",
                  (job_id, app_name, prompt, json.dumps(params), "awaiting_payment", time.time()))
    # payment
    sats = CA.APPS[app_name]["price_sats"]
    if MOCK_PAYMENTS:
        settle_job(job_id)
        return redirect(url_for("job", job_id=job_id))
    inv = create_invoice(job_id, app_name, sats)
    with db() as c:
        c.execute("UPDATE jobs SET btcpay_invoice=? WHERE id=?", (inv, job_id))
    return redirect(f"{BTCPAY_PUBLIC}/i/{inv}")

@app.route("/webhook/btcpay", methods=["POST"])
def btcpay_webhook():
    data = request.get_json(force=True)
    # metadata is TOP-LEVEL in the BTCPay webhook payload, and job_id is nested inside it
    if data.get("type") in ("InvoiceSettled", "InvoiceProcessing"):
        metadata = (data.get("metadata") or {})
        job_id = metadata.get("job_id") or data.get("job_id")
        if job_id:
            settle_job(job_id)
    return "", 200

@app.route("/job/<job_id>")
def job(job_id):
    with db() as c:
        row = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()
    if not row:
        abort(404)
    return render_template("job.html", job=row)

@app.route("/jobs")
def jobs():
    with db() as c:
        rows = c.execute("SELECT * FROM jobs ORDER BY created_at DESC LIMIT 100").fetchall()
    return render_template("jobs.html", jobs=rows)

@app.route("/image/<filename>")
def image(filename):
    return send_from_directory(RESULT_DIR, filename)

@app.route("/health")
def health():
    return jsonify({"ok": True, "apps": len(CA.APPS), "mock_payments": MOCK_PAYMENTS})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5188, debug=False)