client2server/package/src/client2server-unified.lua

--[[
    client2server-unified.lua - Bidirectional event forwarder for OpenWrt
    Copyright (c) 2026 Luis Rosales - MIT License
    
    Uses coroutines for concurrent event monitoring + command execution
]]

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

local cfg = {
    server_url = os.getenv("SERVER_URL") or "wss://your-server.com:3843",
    server_token = os.getenv("SERVER_TOKEN") or "secret-token",
    router_id = os.getenv("ROUTER_ID") or "",
    check_interval = 5,
    wan_interface = "wan",
    wan_device = "eth0",
    buffer_file = "/tmp/event_buffer",
    pid_file = "/var/run/client2server.pid",
    max_buffer = 100,
}

-- Load UCI config
pcall(function()
    local uci = require("luci.model.uci").cursor()
    cfg.server_url = uci:get("client2server", "server", "url") or cfg.server_url
    cfg.server_token = uci:get("client2server", "server", "token") or cfg.server_token
    cfg.router_id = uci:get("client2server", "router", "id") or cfg.router_id
    cfg.wan_interface = uci:get("client2server", "wan", "interface") or cfg.wan_interface
    cfg.wan_device = uci:get("client2server", "wan", "device") or cfg.wan_device
end)

if cfg.router_id == "" then
    local f = io.popen("hostname")
    cfg.router_id = f and (f:read("*a") or ""):gsub("%s+$", "") or "unknown"
    if f then f:close() end
end

-- ============================================================================
-- LOGGING
-- ============================================================================

local function log(level, msg)
    os.execute(string.format('logger -t "client2server" -p user.%s "%s"', level, msg:gsub('"', '\\"')))
end

local function log_info(msg)  log("info", msg) end
local function log_err(msg)  log("err", msg) end

-- ============================================================================
-- JSON
-- ============================================================================

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)))
        elseif type(v) == "table" then
            table.insert(parts, string.format('"%s": %s', k, json_encode(v)))
        end
    end
    return "{" .. table.concat(parts, ",") .. "}"
end

local function json_decode(str)
    local result = {}
    for k, v in str:gmatch('"([^"]+)":%s*"([^"]*)"') do result[k] = v end
    for k, v in str:gmatch('"([^"]+)":%s*(%d+)') do result[k] = tonumber(v) end
    for k, v in str:gmatch('"([^"]+)":%s*(%a+)') do result[k] = v end
    return result
end

-- ============================================================================
-- BUFFER
-- ============================================================================

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

function buffer.init()
    local f = io.open(cfg.buffer_file, "r")
    if f then
        for line in f:lines() do
            if line and line ~= "" then
                table.insert(buffer.events, line)
            end
        end
        f:close()
    end
end

function buffer.save()
    if not buffer.dirty then return end
    local f = io.open(cfg.buffer_file, "w")
    if not f then return end
    for _, ev in ipairs(buffer.events) do f:write(ev .. "\n") end
    f:close()
    buffer.dirty = false
end

function buffer.add(json_event)
    -- Warn if buffer is getting full
    if #buffer.events > cfg.max_buffer * 0.8 then
        log_err("Buffer almost full: " .. #buffer.events .. "/" .. cfg.max_buffer)
    end
    
    table.insert(buffer.events, json_event)
    if #buffer.events > cfg.max_buffer then
        -- Buffer full - drop oldest to make room
        table.remove(buffer.events, 1)
        log_err("Buffer overflow: dropping oldest event")
    end
    buffer.dirty = true
end

function buffer.flush(send_fn)
    if #buffer.events == 0 then return end
    local i = 1
    while i <= #buffer.events do
        if send_fn(buffer.events[i]) then
            table.remove(buffer.events, i)
            buffer.dirty = true
        else i = i + 1 end
    end
    buffer.save()
end

-- ============================================================================
-- WEBSOCKET
-- ============================================================================

local ws = { sock = nil, connected = false }

function ws.connect(url)
    local host = url:match("wss?://([^:/]+)")
    local port = url:match(":%d+") or ":443"
    if not host then return nil end
    
    local sock = require("socket").tcp()
    sock:settimeout(10)
    
    if not pcall(sock.connect, sock, host, tonumber(port:sub(2)) then return nil end
    ws.sock = sock
    ws.connected = true
    return sock
end

function ws.send(data)
    if not ws.connected then buffer.add(data); return false end
    local frame = string.format("\x81\x80%s", data)
    if not pcall(function() ws.sock:send(frame) end) then
        ws.connected = false
        buffer.add(data)
        return false
    end
    return true
end

-- Read from websocket (non-blocking-ish)
function ws.recv()
    if not ws.connected then return nil end
    
    -- Set non-blocking
    ws.sock:settimeout(0.1)
    
    local data, err = ws.sock:receive("*l")
    ws.sock:settimeout(10)
    
    if err and err ~= "timeout" then
        ws.connected = false
        return nil
    end
    
    return data
end

function ws.close()
    if ws.sock then pcall(ws.sock.close, ws.sock) end
    ws.sock = nil
    ws.connected = false
end

-- ============================================================================
-- COMMAND EXECUTOR
-- ============================================================================

function execute_command(cmd_obj)
    local cmd = cmd_obj.command or ""
    local args = cmd_obj.args or {}
    
    log_info("Executing: " .. cmd)
    
    local result = { success = false, output = "", error = "" }
    
    if cmd == "uci_set" then
        local config = args.config
        local section = args.section
        local option = args.option
        local value = args.value
        
        if config and section and option and value then
            local c = string.format("uci set %s.%s.%s='%s'", config, section, option, value)
            local f = io.popen(c)
            result.output = f and f:read("*a") or ""
            if f then f:close() end
            os.execute("uci commit " .. config)
            result.success = true
            log_info("UCI set: " .. config .. "." .. section .. "." .. option .. " = " .. value)
        else
            result.error = "Missing params"
        end
        
    elseif cmd == "shell" then
        local shell_cmd = args.command
        if shell_cmd then
            local f = io.popen(shell_cmd)
            result.output = f and f:read("*a") or ""
            if f then f:close() end
            result.success = true
        else
            result.error = "No command"
        end
        
    elseif cmd == "reboot" then
        os.execute("sync && reboot &")
        result.success = true
        result.output = "Reboot scheduled"
        
    elseif cmd == "wifi_restart" then
        os.execute("/etc/init.d/network restart")
        os.execute("/etc/init.d/wireless restart")
        result.success = true
        
    elseif cmd == "status" then
        local f = io.popen("ubus call network getStatus")
        result.output = f and f:read("*a") or "{}"
        if f then f:close() end
        result.success = true
        
    else
        result.error = "Unknown: " .. cmd
    end
    
    return result
end

-- ============================================================================
-- COROUTINES
-- ============================================================================

-- Monitor DHCP
local function co_dhcp()
    local leases = {}
    
    while true do
        local f = io.open("/var/lib/dnsmasq/dnsmasq.leases", "r")
        if f then
            local current = {}
            for line in f:lines() do
                local ts, mac, ip, name = line:match("(%d+)%s+(%S+)%s+(%S+)%s+(%S+)")
                if mac then
                    current[mac] = { ip = ip, hostname = name }
                    if not leases[mac] then
                        local ev = json_encode({
                            router_id = cfg.router_id, hostname = cfg.router_id,
                            event_type = "dhcp_lease_new",
                            payload = { mac = mac, ip = ip, hostname = name }
                        })
                        ws.send(ev)
                        log_info("DHCP: " .. mac .. " -> " .. ip)
                    end
                end
            end
            f:close()
            
            for mac in pairs(leases) do
                if not current[mac] then
                    local ev = json_encode({
                        router_id = cfg.router_id, hostname = cfg.router_id,
                        event_type = "dhcp_lease_expire",
                        payload = { mac = mac, old_ip = leases[mac].ip }
                    })
                    ws.send(ev)
                    log_info("DHCP expire: " .. mac)
                end
            end
            leases = current
        end
        coroutine.yield(cfg.check_interval)
    end
end

-- Monitor WAN
local function co_wan()
    local link_last, ip_last = nil, nil
    
    local f = io.open("/sys/class/net/" .. cfg.wan_device .. "/carrier", "r")
    if f then link_last = (f:read("*a") or "":find("^1") == 1; f:close() end
    
    while true do
        -- Link
        f = io.open("/sys/class/net/" .. cfg.wan_device .. "/carrier", "r")
        local link_now = f and ((f:read("*a") or ""):find("^1") == 1) or false
        if f then f:close() end
        
        if link_now ~= link_last then
            link_last = link_now
            local ev = json_encode({
                router_id = cfg.router_id, hostname = cfg.router_id,
                event_type = link_now and "wan_link_up" or "wan_link_down",
                payload = { device = cfg.wan_device }
            })
            ws.send(ev)
            log_info("WAN link: " .. (link_now and "up" or "down"))
        end
        
        -- IP
        f = io.popen("ubus call network.interface." .. cfg.wan_interface .. " status 2>/dev/null")
        local ip_now = nil
        if f then
            local st = f:read("*a")
            f:close()
            ip_now = st and st:match('"address"%s*:%s*"([^"]+)"')
        end
        
        if ip_now and ip_now ~= ip_last then
            local ev = json_encode({
                router_id = cfg.router_id, hostname = cfg.router_id,
                event_type = ip_last and "wan_dhcp_changed" or "wan_dhcp_new",
                payload = { old_ip = ip_last, new_ip = ip_now }
            })
            ws.send(ev)
            log_info("WAN IP: " .. (ip_last or "none") .. " -> " .. ip_now)
            ip_last = ip_now
        end
        
        coroutine.yield(cfg.check_interval)
    end
end

-- 🔥 COROUTINE: Command Listener (handles server → router commands)
local function co_commands()
    while true do
        if ws.connected then
            -- Check for incoming data
            local data = ws.recv()
            
            if data and data:match("^{") then
                log_info("Received command: " .. data:sub(1, 100))
                
                -- Parse JSON command
                local cmd_obj = json_decode(data)
                
                if cmd_obj.command then
                    -- Execute command
                    local result = execute_command(cmd_obj)
                    
                    -- Send result back
                    local resp = json_encode({
                        router_id = cfg.router_id,
                        event_type = "command_result",
                        payload = {
                            command = cmd_obj.command,
                            command_id = cmd_obj.id or "",
                            success = result.success,
                            output = result.output,
                            error = result.error
                        }
                    })
                    ws.send(resp)
                    
                    log_info("Command done: " .. cmd_obj.command .. " = " .. (result.success and "OK" or result.error))
                end
            end
        end
        
        coroutine.yield(1)  -- Check every second for commands
    end
end

-- COROUTINE: Auto-reconnect (keeps connection alive)
local function co_connect()
    local retry_delay = 30  -- Start at 30s
    local max_delay = 300    -- Max 5 minutes
    
    while true do
        -- Urgency based on buffer state
        local buffer_fullness = #buffer.events / cfg.max_buffer
        
        if buffer_fullness > 0.8 then
            log_err("Buffer critical at " .. string.format("%.0f%%", buffer_fullness * 100) .. " - aggressive reconnect")
            retry_delay = 10  -- Aggressive when buffer filling
        end
        
        if not ws.connected then
            log_info("Connecting to " .. cfg.server_url .. "...")
            
            local sock = ws.connect(cfg.server_url)
            
            if sock then
                ws.connected = true
                retry_delay = 30  -- Reset on success
                log_info("Connected!")
                buffer.flush(function(d) return ws.send(d) end)
            else
                log_err("Connection failed, retry in " .. retry_delay .. "s")
                coroutine.yield(retry_delay)
                retry_delay = math.min(retry_delay * 2, max_delay)
            end
        else
            -- Even when connected, periodically flush to clear buffer buildup
            buffer.flush(function(d) return ws.send(d) end)
            coroutine.yield(30)
        end
    end
end

-- ============================================================================
-- MAIN
-- ============================================================================

local function scheduler()
    local cos = {
        co_dhcp = coroutine.create(co_dhcp),
        co_wan = coroutine.create(co_wan),
        co_commands = coroutine.create(co_commands),
        co_connect = coroutine.create(co_connect),  -- Auto-reconnect!
    }
    
    while true do
        -- Resume all coroutines
        for name, co in pairs(cos) do
            if coroutine.status(co) == "suspended" then
                local _, delay = coroutine.resume(co)
                -- Small stagger to avoid spikes
                if delay then os.execute("sleep 0.1") end
            end
        end
        
        -- Flush buffer when connected
        if ws.connected then
            buffer.flush(function(d) return ws.send(d) end)
        end
        
        os.execute("sleep 1")
    end
end

local function main()
    log_info("Starting client2server...")
    log_info("Router: " .. cfg.router_id)
    log_info("Server: " .. cfg.server_url)
    
    buffer.init()
    
    local pf = io.open(cfg.pid_file, "w")
    if pf then pf:write(tostring(os.getpid())); pf:close() end
    
    -- Connect
    log_info("Initial connection...")
    local sock = ws.connect(cfg.server_url)
    
    if sock then
        ws.connected = true
        log_info("Connected!")
        buffer.flush(function(d) return ws.send(d) end)
    else
        log_err("Connection failed")
    end
    
    scheduler()
end

main()