Application Routes

loglife.core.routes.webhook.routes

Webhook endpoint for inbound WhatsApp messages.

Receives POST requests, validates payloads, and enqueues messages for processing.

webhook()

Handle inbound WhatsApp messages.

Source code in src/loglife/core/routes/webhook/routes.py

@webhook_bp.route("/webhook", methods=["POST"])
def webhook() -> ResponseReturnValue:
    """Handle inbound WhatsApp messages."""
    try:
        data: dict = request.get_json()

        message = Message.from_payload(data)
        g.client_type = message.client_type  # expose client type to sender service

        enqueue_inbound_message(message)

        logger.info("Queued message type %s for %s", message.msg_type, message.sender)
        # For emulator, we don't need an explicit "queued" response message in the UI
        # We return an empty message so the emulator doesn't show "Message queued"
        return success_response(message="")
    except Exception as e:
        error = f"Error processing webhook > {e}"
        logger.exception(error)
        return error_response(error)

loglife.core.routes.emulator.routes

Web-based emulator for testing chat flows.

Serves the emulator UI and provides an SSE stream for realtime logs.

emulator()

Render the emulator HTML interface.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/")
def emulator() -> str:
    """Render the emulator HTML interface."""
    return render_template("emulator.html", db_url=SQLITE_WEB_URL)

events()

Stream realtime log events to the browser via SSE.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/events")
def events() -> Response:
    """Stream realtime log events to the browser via SSE."""

    def stream() -> Generator[str, None, None]:
        # Listen yields messages from the broadcaster
        for msg in log_broadcaster.listen():
            # Handle multiline messages for SSE
            formatted_msg = msg.replace("\n", "\ndata: ")
            yield f"data: {formatted_msg}\n\n"

    return Response(stream(), mimetype="text/event-stream")