plumbing/servicem8_webhook_receiver.py

import json
import logging
import os
import sqlite3
from contextlib import closing
from datetime import datetime, timezone

from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
from servicem8_quote_template_parser import (
    QUOTE_TEMPLATE_FORM_UUID,
    STATE_DB_PATH,
    init_state_db as init_quote_state_db,
    parse_quote_template_form_response,
)

DB_PATH = os.getenv("WEBHOOK_DB_PATH", "./servicem8_webhooks.db")
APP_HOST = os.getenv("WEBHOOK_HOST", "0.0.0.0")
APP_PORT = int(os.getenv("WEBHOOK_PORT", "18354"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("servicem8-webhook-receiver")

app = FastAPI(title="ServiceM8 Webhook Receiver", version="1.2.1")


def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    with closing(get_conn()) as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS webhook_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                client_host TEXT,
                method TEXT NOT NULL,
                path TEXT NOT NULL,
                headers_json TEXT,
                payload_json TEXT NOT NULL
            )
            """
        )

        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS webhook_objects (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                client_host TEXT,
                method TEXT NOT NULL,
                path TEXT NOT NULL,
                object_type TEXT,
                object_uuid TEXT,
                changed_fields_json TEXT,
                object_time_utc TEXT,
                resource_url TEXT,
                headers_json TEXT,
                payload_json TEXT NOT NULL
            )
            """
        )

        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS webhook_form_responses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                client_host TEXT,
                method TEXT NOT NULL,
                path TEXT NOT NULL,
                headers_json TEXT,
                payload_json TEXT NOT NULL
            )
            """
        )

        conn.commit()

    init_quote_state_db(STATE_DB_PATH)


@app.on_event("startup")
async def startup_event():
    init_db()
    logger.info("SQLite DB ready at %s", DB_PATH)


@app.get("/health")
async def health():
    return {"ok": True}


async def parse_request_payload(request: Request):
    content_type = request.headers.get("content-type", "").lower()

    if "application/json" in content_type:
        try:
            return await request.json()
        except Exception:
            body = await request.body()
            return {"_raw_body": body.decode("utf-8", errors="replace")}

    if "application/x-www-form-urlencoded" in content_type or "multipart/form-data" in content_type:
        try:
            form = await request.form()
            return dict(form)
        except Exception:
            body = await request.body()
            return {"_raw_body": body.decode("utf-8", errors="replace")}

    try:
        return await request.json()
    except Exception:
        body = await request.body()
        return {"_raw_body": body.decode("utf-8", errors="replace")}


def maybe_handle_challenge(payload, client_host):
    if isinstance(payload, dict) and payload.get("mode") == "subscribe" and payload.get("challenge"):
        challenge = str(payload["challenge"])
        logger.info("ServiceM8 webhook verification received from %s", client_host)
        return PlainTextResponse(challenge, status_code=200)
    return None


def store_simple_event(table_name, request: Request, client_host: str, received_at: str, headers: dict, payload):
    with closing(get_conn()) as conn:
        conn.execute(
            f"""
            INSERT INTO {table_name} (
                received_at,
                client_host,
                method,
                path,
                headers_json,
                payload_json
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                received_at,
                client_host,
                request.method,
                request.url.path,
                json.dumps(headers),
                json.dumps(payload),
            ),
        )
        conn.commit()


def queue_quote_template_jobmaterials(payload, received_at: str):
    try:
        parsed = parse_quote_template_form_response(payload)
    except Exception as exc:
        logger.warning("Quote template parser failed: %s", exc)
        return

    form_uuid = parsed.get("form_uuid", "")
    if form_uuid != QUOTE_TEMPLATE_FORM_UUID:
        return

    queue_path = os.path.join(os.path.dirname(DB_PATH), "quote-template-jobmaterials-queue.jsonl")
    queue_record = {
        "queued_at": received_at,
        "form_uuid": form_uuid,
        "form_response_uuid": parsed.get("form_response_uuid", ""),
        "job_uuid": parsed.get("job_uuid", ""),
        "author_name": parsed.get("author_name", ""),
        "description": parsed.get("description", ""),
        "desired_job_materials": parsed.get("desired_job_materials", []),
    }
    with open(queue_path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(queue_record, ensure_ascii=False) + "\n")

    logger.info(
        "Queued quote-template jobmaterials: form_response_uuid=%s job_uuid=%s rows=%s",
        queue_record["form_response_uuid"],
        queue_record["job_uuid"],
        len(queue_record["desired_job_materials"]),
    )


@app.post("/webhooks/servicem8-job-updated")
async def servicem8_event_webhook(request: Request):
    payload = await parse_request_payload(request)
    headers = dict(request.headers)
    client_host = request.client.host if request.client else None
    received_at = datetime.now(timezone.utc).isoformat()

    challenge_response = maybe_handle_challenge(payload, client_host)
    if challenge_response is not None:
        return challenge_response

    store_simple_event("webhook_events", request, client_host, received_at, headers, payload)

    logger.info("Event webhook received from %s and stored", client_host)
    return PlainTextResponse("OK", status_code=200)


@app.post("/webhooks/servicem8/form-response")
async def servicem8_form_response_webhook(request: Request):
    payload = await parse_request_payload(request)
    headers = dict(request.headers)
    client_host = request.client.host if request.client else None
    received_at = datetime.now(timezone.utc).isoformat()

    challenge_response = maybe_handle_challenge(payload, client_host)
    if challenge_response is not None:
        return challenge_response

    store_simple_event("webhook_form_responses", request, client_host, received_at, headers, payload)
    queue_quote_template_jobmaterials(payload, received_at)

    logger.info("Form response webhook received from %s and stored", client_host)
    return PlainTextResponse("OK", status_code=200)


@app.post("/webhooks/servicem8-object")
async def servicem8_object_webhook(request: Request):
    payload = await parse_request_payload(request)
    headers = dict(request.headers)
    client_host = request.client.host if request.client else None
    received_at = datetime.now(timezone.utc).isoformat()

    challenge_response = maybe_handle_challenge(payload, client_host)
    if challenge_response is not None:
        return challenge_response

    object_type = None
    object_uuid = None
    changed_fields = None
    object_time_utc = None
    resource_url = None

    if isinstance(payload, dict):
        object_type = payload.get("object")
        resource_url = payload.get("resource_url")

        entry = payload.get("entry")
        if isinstance(entry, list) and len(entry) > 0 and isinstance(entry[0], dict):
            first_entry = entry[0]
            object_uuid = first_entry.get("uuid")
            changed_fields = first_entry.get("changed_fields")
            object_time_utc = first_entry.get("time")

    with closing(get_conn()) as conn:
        conn.execute(
            """
            INSERT INTO webhook_objects (
                received_at,
                client_host,
                method,
                path,
                object_type,
                object_uuid,
                changed_fields_json,
                object_time_utc,
                resource_url,
                headers_json,
                payload_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                received_at,
                client_host,
                request.method,
                request.url.path,
                object_type,
                object_uuid,
                json.dumps(changed_fields) if changed_fields is not None else None,
                object_time_utc,
                resource_url,
                json.dumps(headers),
                json.dumps(payload),
            ),
        )
        conn.commit()

    logger.info(
        "Object webhook received from %s: object=%s uuid=%s",
        client_host,
        object_type,
        object_uuid,
    )
    return PlainTextResponse("OK", status_code=200)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("servicem8_webhook_receiver:app", host=APP_HOST, port=APP_PORT, reload=False)