Mqtt_api/main.py

from fastapi import FastAPI
import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
import json
import threading
import os
import time
import requests
import datetime
from concurrent.futures import ThreadPoolExecutor

app = FastAPI(title="Nearle MQTT Webhook Bridge API")

# Configuration
MQTT_BROKER = os.getenv("MQTT_BROKER", "66.116.225.226")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_USER = os.getenv("MQTT_USER", "admin")
MQTT_PASS = os.getenv("MQTT_PASS", "Package@321#")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "nearle/#")

# Define your specific external POST APIs here
API_STATUS_URL = os.getenv("API_STATUS_URL", "https://jupiter.nearle.app/live/api/v1/utils/createriderstatus")
API_LOGS_URL = os.getenv("API_LOGS_URL", "https://jupiter.nearle.app/live/api/v1/utils/createriderperiodiclog")
API_PROFILE_URL = os.getenv("API_PROFILE_URL", "https://jupiter.nearle.app/live/api/v1/utils/createuserredis")

# Bounded worker pool. Previously every MQTT message spawned its own thread, so a
# stall on the Jupiter API (the 5s read timeouts seen in the logs) let threads pile
# up without limit. A fixed pool keeps memory flat and applies natural backpressure.
FORWARD_WORKERS = int(os.getenv("FORWARD_WORKERS", "16"))
_executor = ThreadPoolExecutor(max_workers=FORWARD_WORKERS, thread_name_prefix="forward")

# Liveness state, exposed on / so a monitor can tell a live bridge from a dead one.
_state = {
    "mqtt_connected": False,
    "last_connect_at": None,
    "last_disconnect_at": None,
    "last_message_at": None,
    "connect_attempts": 0,
    "messages_received": 0,
}
_state_lock = threading.Lock()


def _now():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


# Function to forward data to a specific URL
def forward_to_api(target_url: str, payload_str: str):
    if not target_url:
        return

    # Determine the content type and prepare data
    headers = {}

    # Check if the payload is JSON or plain string
    try:
        payload_data = json.loads(payload_str)
        headers['Content-Type'] = 'application/json'

        # We send the exact parsed JSON so the receiving API sees it cleanly
        data_kwargs = {"json": payload_data}
    except:
        # If it's a plain string like "Offline", send as text/plain
        headers['Content-Type'] = 'text/plain'
        data_kwargs = {"data": payload_str}

    try:
        response = requests.post(target_url, headers=headers, timeout=5, **data_kwargs)
        if response.status_code in (200, 201):
            print(f"SUCCESS ({response.status_code}) -> {target_url} | Sent: {data_kwargs}", flush=True)
        else:
            print(f"WARNING: HTTP {response.status_code} from {target_url} | Response: {response.text}", flush=True)
    except Exception as e:
        print(f"ERROR: Failed to send to {target_url} | Exception: {e}", flush=True)


def route_payload(data_type: str, rider_id: str, payload_str: str):
    payload_str = payload_str.strip()
    if data_type == "status":
        # Ensure userid is an integer if possible so Jupiter API doesn't reject it
        r_id = int(rider_id) if rider_id.isdigit() else rider_id
        status_json = json.dumps({"userid": r_id, "status": payload_str})
        forward_to_api(API_STATUS_URL, status_json)
    elif data_type == "logs":
        forward_to_api(API_LOGS_URL, payload_str)
    elif data_type == "profile":
        forward_to_api(API_PROFILE_URL, payload_str)
    else:
        # Ignore other data types or add additional routes here
        pass


# MQTT Callbacks
def on_connect(client, userdata, connect_flags, reason_code, properties=None):
    if reason_code == 0:
        with _state_lock:
            _state["mqtt_connected"] = True
            _state["last_connect_at"] = _now()
        # Subscribing here (not once after connect) is the important part: the
        # session is clean, so every reconnect starts with no subscriptions. Doing
        # it in on_connect means the bridge resubscribes automatically after any
        # broker restart or network blip instead of sitting connected and deaf.
        client.subscribe(MQTT_TOPIC)
        print(f"MQTT connected to {MQTT_BROKER}:{MQTT_PORT}, subscribed to {MQTT_TOPIC}", flush=True)
    else:
        print(f"MQTT connect refused: {reason_code}", flush=True)


def on_disconnect(client, userdata, disconnect_flags, reason_code, properties=None):
    with _state_lock:
        _state["mqtt_connected"] = False
        _state["last_disconnect_at"] = _now()
    print(f"MQTT disconnected: {reason_code} (will auto-reconnect)", flush=True)


def on_message(client, userdata, msg):
    try:
        topic = msg.topic
        payload = msg.payload.decode()

        with _state_lock:
            _state["last_message_at"] = _now()
            _state["messages_received"] += 1

        # Topic structure: nearle/riders/{id}/{data_type}/...
        parts = topic.split('/')
        if len(parts) >= 4 and parts[0] == "nearle" and parts[1] == "riders":
            rider_id = parts[2]
            data_type = parts[3]

            # Hand off to the pool so the MQTT network loop is never blocked
            _executor.submit(route_payload, data_type, rider_id, payload)

    except Exception as e:
        print(f"Error processing MQTT message: {e}", flush=True)


def start_mqtt():
    # This loop is why the bridge no longer dies silently. The old version called
    # connect() inside a try/except: one failure ("No route to host" on 2026-06-10)
    # killed the thread for good while uvicorn kept answering 200 OK on /, so the
    # bridge looked healthy while forwarding nothing until someone restarted it.
    backoff = 5
    while True:
        try:
            with _state_lock:
                _state["connect_attempts"] += 1

            client = mqtt.Client(CallbackAPIVersion.VERSION2)
            client.username_pw_set(MQTT_USER, MQTT_PASS)
            client.on_connect = on_connect
            client.on_disconnect = on_disconnect
            client.on_message = on_message
            client.reconnect_delay_set(min_delay=1, max_delay=60)

            print(f"Connecting to MQTT Broker at {MQTT_BROKER}...", flush=True)

            # connect_async + retry_first_connection means a broker that is down at
            # startup is retried instead of being a fatal error.
            client.connect_async(MQTT_BROKER, MQTT_PORT, keepalive=60)
            backoff = 5
            client.loop_forever(retry_first_connection=True)

            print("MQTT loop returned unexpectedly; restarting.", flush=True)
        except Exception as e:
            print(f"MQTT loop crashed: {e} | retrying in {backoff}s", flush=True)
        finally:
            with _state_lock:
                _state["mqtt_connected"] = False

        time.sleep(backoff)
        backoff = min(backoff * 2, 60)


# Start MQTT in a background thread
@app.on_event("startup")
async def startup_event():
    thread = threading.Thread(target=start_mqtt, daemon=True)
    thread.start()


# simple health check endpoint
@app.get("/")
def read_root():
    with _state_lock:
        snapshot = dict(_state)
    return {
        "status": "Nearle Webhook Bridge is running",
        "description": "Listening to MQTT and forwarding to external POST APIs.",
        "mqtt": snapshot,
    }


@app.get("/healthz")
def healthz():
    """Returns 200 only when the MQTT side is actually connected, so an external
    monitor can detect a bridge that is up but not receiving anything."""
    with _state_lock:
        connected = _state["mqtt_connected"]
        snapshot = dict(_state)
    from fastapi.responses import JSONResponse
    return JSONResponse(
        status_code=200 if connected else 503,
        content={"healthy": connected, "mqtt": snapshot},
    )