client2server/usr/sbin/client2server-luv.lua

--[[
    client2server-luv - Lua WebSocket Event Forwarder for OpenWrt with luv async

    Features:
    - WebSocket connection to central server
    - Auto-reconnect on disconnect
    - Local buffer (store-and-forward while offline)
    - DHCP/WiFi/Interface event tracking
    - ASYNC monitoring via luv (libuv bindings)
    - Parallel polling for 50+ devices

    Copyright (c) 2026 Luis Rosales - MIT License
]]

-- ============================================================================
-- REQUIREMENTS
-- ============================================================================

-- Try to load luv (libuv bindings for async operations)
local luv_ok, luv = pcall(require, "luv")

-- Try to load websocket library, fallback to simple HTTP
local ws_client = nil
local has_websocket, websocket = pcall(require, "websocket")

if has_websocket then
    ws_client = websocket.client.sync()
end

-- ============================================================================
-- CONFIG
-- ============================================================================

local cfg = {
    url = os.getenv("SERVER_URL") or "wss://your-server.com/ws",
    token = os.getenv("SERVER_TOKEN") or "secret-token",
    router_id = os.getenv("ROUTER_ID") or "unknown",
    reconnect_delay = 5,
    ping_interval = 30,
    buffer_file = "/tmp/event_buffer",
    max_buffer = 100,
    poll_interval = 30,        -- device poll interval (seconds)
    wan_poll_interval = 30,     -- WAN poll interval (seconds)
    dhcp_poll_interval = 5,   -- DHCP poll interval (seconds)
    poll_timeout = 5000,       -- poll timeout (ms)
}

-- Load from UCI if available
pcall(function()
    local uci = require("luci.model.uci").cursor()
    cfg.url = uci:get("event-forwarder", "server", "url") or cfg.url
    cfg.token = uci:get("event-forwarder", "server", "token") or cfg.token
    cfg.router_id = uci:get("event-forwarder", "router", "id") or cfg.router_id
end)

-- ============================================================================
-- UTILITIES
-- ============================================================================

local function log(level, msg)
    os.execute(string.format('logger -t "client2server-luv" -p user.%s "%s" 2>/dev/null', level, msg))
end

local function get_hostname()
    -- Try multiple methods to get hostname
    local f = io.popen("cat /proc/sys/kernel/hostname 2>/dev/null || hostname 2>/dev/null || cat /etc/hostname 2>/dev/null || echo unknown")
    local h = f and f:read("*a") or "unknown"
    if f then f:close() end
    return (h:gsub("%s+$", ""))
end

cfg.router_id = cfg.router_id == "unknown" and get_hostname() or cfg.router_id

local function json_encode(t)
    local parts = {}
    for k, v in pairs(t) do
        if type(v) == "string" then
            table.insert(parts, string.format('"%s": "%s"', k, v:gsub('"', '\\"')))
        elseif type(v) == "number" then
            table.insert(parts, string.format('"%s": %s', k, tostring(v)))
        elseif type(v) == "boolean" then
            table.insert(parts, string.format('"%s": %s', k, tostring(v)))
        end
    end
    return "{" .. table.concat(parts, ",") .. "}"
end

local function json_decode(str)
    local result = {}
    for k, v in str:gmatch('"([^"]+)":%s*([^},]+)') do
        v = v:gsub('[%s"]+', '')
        if v == "true" or v == "false" then
            result[k] = v == "true"
        elseif tonumber(v) then
            result[k] = tonumber(v)
        else
            result[k] = v
        end
    end
    return result
end

-- ============================================================================
-- BUFFER (OFFLINE SUPPORT)
-- ============================================================================

local buffer = {
    events = {},
    dirty = false,
}

function buffer.load()
    local f = io.open(cfg.buffer_file, "r")
    if not f then return end

    for line in f:lines() do
        if line and line ~= "" then
            table.insert(buffer.events, line)
        end
    end
    f:close()
    log("info", "Loaded " .. #buffer.events .. " buffered events")
end

function buffer.save()
    local f = io.open(cfg.buffer_file, "w")
    if not f then return end

    for _, event in ipairs(buffer.events) do
        f:write(event .. "\n")
    end
    f:close()
    buffer.dirty = false
end

function buffer.add(json_event)
    table.insert(buffer.events, json_event)

    -- Limit buffer size
    while #buffer.events > cfg.max_buffer do
        table.remove(buffer.events, 1)
    end

    buffer.dirty = true
end

function buffer.flush(send_fn)
    if #buffer.events == 0 then return end

    local to_send = buffer.events
    buffer.events = {}
    buffer.dirty = false

    for _, event in ipairs(to_send) do
        local ok, err = pcall(send_fn, event)
        if not ok or err then
            -- Re-add to buffer on failure
            table.insert(buffer.events, event)
        end
    end

    buffer.save()
end

function buffer.clear()
    buffer.events = {}
    buffer.dirty = false
    os.remove(cfg.buffer_file)
end

-- ============================================================================
-- WEBSOCKET (with fallback)
-- ============================================================================

local ws = {}

function ws.connect(url)
    if ws_client then
        local sock, err = ws_client.connect(url)
        if err then
            log("err", "WS connect error: " .. err)
            return nil
        end
        return sock
    end
    return nil
end

function ws.send(sock, data)
    if sock then
        return sock:send(data)
    end
    return false, "no socket"
end

function ws.close(sock)
    if sock then
        sock:close()
    end
end

function ws.connected(sock)
    return sock ~= nil
end

function build_event(event_type, payload)
    return json_encode({
        type = event_type,
        router_id = cfg.router_id,
        timestamp = os.time(),
        payload = payload
    })
end

-- ============================================================================
-- ASYNC POLLING WITH LUV
-- ============================================================================

-- Device polling results storage
local poll_results = {}
local poll_count = 0
local poll_total = 0

-- Poll a single device via ubus
local function poll_single_device(device_id, device_ip)
    local cmd = string.format(
        'ubus call network.interface.%s status 2>/dev/null',
        device_id
    )

    local f = io.popen(cmd)
    if not f then
        return { status = "error", ip = device_ip }
    end

    local result = f:read("*all")
    f:close()

    local parsed = json_decode(result)
    parsed.status = "online"
    parsed.ip = device_ip

    return parsed
end

-- Async parallel polling with luv
local function poll_devices_parallel(device_list)
    if not luv then
        -- Fallback: sequential
        for _, dev in ipairs(device_list) do
            poll_results[dev.id] = poll_single_device(dev.id, dev.ip)
        end
        return
    end

    poll_results = {}
    poll_count = 0
    poll_total = #device_list

    log("info", "Starting parallel poll of " .. poll_total .. " devices")

    -- Poll each device in parallel using luv async
    for _, dev in ipairs(device_list) do
        local device_id = dev.id
        local device_ip = dev.ip

        -- Run in async task
        luv.new_task(function()
            local result = poll_single_device(device_id, device_ip)
            poll_results[device_id] = result

            poll_count = poll_count + 1
            if poll_count == poll_total then
                log("info", "All " .. poll_total .. " devices polled")
                -- Process results here if needed
            end
        end)
    end
end

-- ============================================================================
-- WIFI EVENTS (hostapd via ubus)
-- ============================================================================

local function monitor_wifi_events()
    if not luv then
        log("warn", "luv not available for WiFi monitoring")
        return
    end

    -- Use luv to watch ubus for wireless events
    -- Note: ubus doesn't support event subscription directly,
    -- so we poll hostapd status periodically

    local timer = luv.new_timer()
    local last_clients = {}

    luv.timer_start(timer, 10000, 10000, function()
        -- Poll wireless clients
        local f = io.popen("ubus call hostapd.wlan0-1 get_clients 2>/dev/null")
        if f then
            local result = f:read("*all")
            f:close()

            if result and result ~= "" then
                -- Parse clients and detect changes
                local current_clients = {}
                for mac in result:gmatch('"([^"]+)":') do
                    current_clients[mac] = true
                end

                -- Detect new connections
                for mac, _ in pairs(current_clients) do
                    if not last_clients[mac] then
                        local ev = build_event("wifi_connect", {
                            mac = mac,
                            interface = "wlan0-1"
                        })
                        log("info", "WiFi connected: " .. mac)
                        buffer.add(ev)
                    end
                end

                -- Detect disconnections
                for mac, _ in pairs(last_clients) do
                    if not current_clients[mac] then
                        local ev = build_event("wifi_disconnect", {
                            mac = mac,
                            interface = "wlan0-1"
                        })
                        log("info", "WiFi disconnected: " .. mac)
                        buffer.add(ev)
                    end
                end

                last_clients = current_clients
            end
        end
    end)

    log("info", "WiFi event monitor started")
end

-- ============================================================================
-- DHCP LEASES HELPERS (must be before monitor functions)
-- ============================================================================

-- Helper: read current leases
local function read_leases()
    local leases = {}
    local f = io.open("/var/lib/dnsmasq/dnsmasq.leases", "r")
    if not f then return leases end

    for line in f:lines() do
        local ts, mac, ip, name = line:match("(%d+)%s+(%S+)%s+(%S+)%s+(%S+)")
        if mac then
            leases[mac] = { ip = ip, hostname = name, time = tonumber(ts) }
        end
    end
    f:close()
    return leases
end

-- Helper: process leases and detect changes
local function process_leases(old_leases, callback)
    local leases = read_leases()

    -- New leases
    for mac, info in pairs(leases) do
        if not old_leases[mac] then
            callback(mac, info.ip, info.hostname, "new")
        end
    end

    -- Expired leases
    for mac, info in pairs(old_leases) do
        if not leases[mac] then
            callback(mac, info.ip, info.hostname, "expired")
        end
    end
end

-- ============================================================================
-- DHCP LEASES MONITOR
-- ============================================================================

local function monitor_dhcp_leases()
    if not luv then
        -- Fallback: sequential file polling
        monitor_dhcp_sequential()
        return
    end
    
    -- Use timer-based polling instead of fs_event (more compatible)
    local timer = luv.new_timer()
    local old_leases = {}
    
    luv.timer_start(timer, cfg.dhcp_poll_interval * 1000, cfg.dhcp_poll_interval * 1000, function()
        -- File changed, process leases
        process_leases(old_leases, function(mac, ip, hostname, action)
            local ev = build_event("dhcp_lease", {
                mac = mac,
                ip = ip,
                hostname = hostname,
                action = action
            })
            log("info", "DHCP: " .. action .. " - " .. mac .. " -> " .. ip)
            buffer.add(ev)
        end)
        
        old_leases = read_leases()
    end)
    
    log("info", "DHCP lease monitor started")
end

-- Fallback: sequential DHCP monitoring
local function monitor_dhcp_sequential()
    local old_leases = {}

    while true do
        process_leases(old_leases, function(mac, ip, hostname, action)
            local ev = build_event("dhcp_lease", {
                mac = mac,
                ip = ip,
                hostname = hostname,
                action = action
            })
            log("info", "DHCP: " .. action .. " - " .. mac .. " -> " .. ip)
            buffer.add(ev)
        end)

        old_leases = read_leases()
        os.execute("sleep " .. cfg.dhcp_poll_interval)
    end
end

-- ============================================================================
-- WAN MONITORING
-- ============================================================================

local last_wan_state = nil

local function monitor_wan()
    if not luv then
        -- Fallback: sequential WAN polling
        monitor_wan_sequential()
        return
    end

    local timer = luv.new_timer()

    luv.timer_start(timer, cfg.wan_poll_interval * 1000, cfg.wan_poll_interval * 1000, function()
        local f = io.popen("ubus call network.interface.wan status 2>/dev/null")
        if f then
            local status = f:read("*all")
            f:close()

            local is_up = status:match('"up":%s*true') ~= nil

            if is_up ~= last_wan_state then
                local ev = build_event("wan_status", {
                    up = is_up
                })
                log("info", "WAN: " .. (is_up and "up" or "down"))
                buffer.add(ev)
                last_wan_state = is_up
            end
        end
    end)

    log("info", "WAN monitor started")
end

-- Fallback: sequential WAN monitoring
local function monitor_wan_sequential()
    while true do
        local f = io.popen("ubus call network.interface.wan status 2>/dev/null")
        if f then
            local status = f:read("*all")
            f:close()

            local is_up = status:match('"up":%s*true') ~= nil

            if is_up ~= last_wan_state then
                local ev = build_event("wan_status", {
                    up = is_up
                })
                log("info", "WAN: " .. (is_up and "up" or "down"))
                buffer.add(ev)
                last_wan_state = is_up
            end
        end

        os.execute("sleep " .. cfg.wan_poll_interval)
    end
end

-- ============================================================================
-- NETWORK STATUS POLLING
-- ============================================================================

local function poll_network_status()
    if not luv then
        -- Sequential fallback
        local f = io.popen("ubus call network getStatus 2>/dev/null")
        if f then f:close() end
        return
    end

    local timer = luv.new_timer()

    luv.timer_start(timer, cfg.poll_interval * 1000, cfg.poll_interval * 1000, function()
        local f = io.popen("ubus call network getStatus 2>/dev/null")
        if f then
            local status = f:read("*all")
            f:close()

            if status and status ~= "" then
                local ev = build_event("network_status", {
                    status = status
                })
                buffer.add(ev)
            end
        end
    end)

    log("info", "Network status poller started")
end

-- ============================================================================
-- MAIN EVENT LOOP
-- ============================================================================

local function main()
    log("info", "client2server-luv starting...")
    log("info", "Router: " .. cfg.router_id)
    log("info", "Server: " .. cfg.url)

    if luv_ok then
        log("info", "luv available - using async mode")
    else
        log("warn", "luv NOT available - using blocking mode")
    end

    -- Load buffered events
    buffer.load()

    -- Save PID
    local pf = io.open("/var/run/client2server-luv.pid", "w")
    if pf then
        local f = io.popen("echo $$")
        local pid = f and f:read("*a") or "0"
        if f then f:close() end
        pf:write(pid:gsub("%s+", ""))
        pf:close()
    end

    local sock = nil
    local retries = 0

    if luv_ok then
        -- Use luv event loop
        -- luv.run() starts the event loop. The callbacks we registered
        -- (timers, fs_events) will run automatically.
        local function start_monitors()
            -- Start all monitors in parallel
            monitor_wifi_events()
            monitor_dhcp_leases()
            monitor_wan()
            poll_network_status()

            -- Keep the event loop running
            local idle = luv.new_idle()
            luv.idle_start(idle, function()
                -- Idle work - just keep loop alive
            end)
        end

        -- Start monitors first, then run the event loop with WebSocket
        local function run_event_loop()
            -- Register all monitors
            monitor_wifi_events()
            monitor_dhcp_leases()
            monitor_wan()
            poll_network_status()
            
            -- Keep the event loop running with idle
            local idle = luv.new_idle()
            luv.idle_start(idle, function()
                -- Idle work - check WebSocket periodically
            end)
            
            -- WebSocket connection - try to connect and stay connected
            local function ws_loop()
                log("info", "Connecting to server...")
                
                if ws_client then
                    sock = ws.connect(cfg.url)
                end
                
                if sock then
                    log("info", "Connected!")
                    retries = 0
                    
                    -- Flush buffer
                    buffer.flush(function(data)
                        return ws.send(sock, data)
                    end)
                    
                    -- Stay connected - reconnect timer will handle reconnection
                    local ws_timer = luv.new_timer()
                    luv.timer_start(ws_timer, 5000, 5000, function()
                        if ws.connected(sock) then
                            -- Periodic flush
                            buffer.flush(function(data)
                                return ws.send(sock, data)
                            end)
                        else
                            -- Disconnected, reconnect
                            ws.close(sock)
                            sock = nil
                            buffer.save()
                            ws_loop()  -- Reconnect
                        end
                    end)
                else
                    log("err", "Connection failed")
                    retries = retries + 1
                    
                    -- Retry after delay
                    local retry_timer = luv.new_timer()
                    luv.timer_start(retry_timer, cfg.reconnect_delay * 1000, 0, function()
                        ws_loop()
                    end)
                end
            end
            
            -- Start WebSocket loop
            ws_loop()
        end

        -- Run the event loop
        local ok, err = pcall(function() luv.run(run_event_loop) end)
        if not ok then
            log("err", "luv.run error: " .. tostring(err))
            -- Fallback: sequential mode
            while true do
                os.execute("sleep 60")
            end
        end
    else
        -- Fallback: sequential mode
        while true do
            -- Sequential monitoring
            monitor_dhcp_sequential()
            monitor_wan_sequential()

            -- WebSocket connection
            log("info", "Connecting to server...")

            if ws_client then
                sock = ws.connect(cfg.url)
            end

            if sock then
                log("info", "Connected!")
                retries = 0

                buffer.flush(function(data)
                    return ws.send(sock, data)
                end)

                local loop_count = 0
                while ws.connected(sock) and loop_count < (cfg.ping_interval / 5) do
                    os.execute("sleep 5")
                    loop_count = loop_count + 1

                    buffer.flush(function(data)
                        return ws.send(sock, data)
                    end)
                end
            else
                log("err", "Connection failed")
                retries = retries + 1
            end

            ws.close(sock)
            buffer.save()
            os.execute("sleep " .. cfg.reconnect_delay)
        end
    end
end

-- ============================================================================
-- START
-- ============================================================================

main()