-- pushfs_client.lua
-- Pocket computer client for "pushfs-v1" turtles/computers.
-- Subcommands:
--   ping [--all | --id N | --label L]
--   put <src> <dst> [--all|--id N|--label L]
--   exec "<cmd line>" [--capture] [--timeout S] [--cwd /] [--all|--id N|--label L]
--   reboot [--all|--id N|--label L]
--
-- Reporting: this client aggregates and prints ALL replies it receives
-- within a short window for each request (fan-in reporting).

--------------------------
-- Config
--------------------------
local PROTOCOL = "pushfs-v1"
local SHARED_SECRET = nil      -- set if servers require a token
local ACK_WINDOW = 2.5         -- seconds to collect multiple ACKs
local STEP_WINDOW = 2.5        -- per-step window during chunked PUT
local CHUNK = 4096
local MAX_INLINE = 48*1024

-- Lua 5.1/5.2 compat
local unpack = table.unpack or unpack

--------------------------
-- Utilities
--------------------------
local function die(msg) print(msg) error(msg, 0) end

local function openModems()
  local ok = false
  for _, s in ipairs(rs.getSides()) do
    if peripheral.getType(s) == "modem" then
      if not rednet.isOpen(s) then rednet.open(s) end
      ok = true
    end
  end
  if not ok then die("No modem. Attach a wireless modem.") end
end

local function readAll(path)
  if not fs.exists(path) or fs.isDir(path) then return nil, "missing or dir" end
  local h = fs.open(path, "rb"); if not h then return nil, "open failed" end
  local d = h.readAll() or ""; h.close(); return d
end

local function adler32(str)
  local MOD = 65521
  local a,b = 1,0
  for i=1,#str do
    a = (a + string.byte(str,i)) % MOD
    b = (b + a) % MOD
  end
  return string.format("%08x", bit32.bor(bit32.lshift(b,16), a))
end

local function gen_txid()
  return tostring(os.epoch("utc")) .. "-" .. tostring(math.random(1,1e9))
end

local function parse_args(argv)
  local cmd = argv[1]; if not cmd then die("Usage: pushfs_client <ping|put|exec|reboot> ...") end
  local out = { cmd = cmd, args = {}, flags = {} }
  local i = 2
  while i <= #argv do
    local a = argv[i]
    if a:sub(1,2) == "--" then
      local k = a:sub(3); local v = true
      if i < #argv and argv[i+1]:sub(1,2) ~= "--" then v = argv[i+1]; i = i + 1 end
      out.flags[k] = v
    else table.insert(out.args, a) end
    i = i + 1
  end
  return out
end

local function build_target(flags)
  if flags["all"] == true or flags["all"] == "true" then return "all" end
  if flags["id"] then return { id = tonumber(flags["id"]) } end
  if flags["label"] then return { label = tostring(flags["label"]) } end
  die("Target required: use --all or --id N or --label L")
end

-- Collect ALL matching replies for a txid/op during window_s seconds.
local function collect_acks(txid, want_op, window_s)
  local end_at = os.epoch("utc") + math.floor((window_s or ACK_WINDOW) * 1000)
  local seen = {}   -- sender_id -> msg
  while true do
    local now = os.epoch("utc"); if now >= end_at then break end
    local timeout = (end_at - now) / 1000
    local sender, msg, proto = rednet.receive(PROTOCOL, timeout)
    if sender and type(msg)=="table" and msg.txid==txid and msg.op==want_op then
      msg.__sender = sender
      seen[sender] = msg
    end
  end
  -- flatten
  local list = {}
  for _, m in pairs(seen) do table.insert(list, m) end
  table.sort(list, function(a,b) return (a.__sender or 0) < (b.__sender or 0) end)
  return list
end

-- Send, then collect all acks in a window.
local function send_and_collect(msg, want_op, window_s)
  rednet.broadcast(msg, PROTOCOL)
  return collect_acks(msg.txid, want_op, window_s)
end

local function print_acks(title, acks)
  print(title .. (": %d node(s)"):format(#acks))
  for _, m in ipairs(acks) do
    local who = ("id=%s"):format(tostring(m.__sender))
    if m.id or m.label then who = ("%s label=%s id=%s"):format(tostring(m.label or ""), tostring(m.id or "?"), tostring(m.__sender)) end
    local note = m.note or ""
    print(("- %s: %s%s"):format(who, m.ok and "ok" or "err", note ~= "" and (" ("..note..")") or ""))
  end
end

--------------------------
-- Ops
--------------------------
local function op_ping(flags)
  local target = build_target(flags)
  local txid = gen_txid()
  local req = { op="ping", txid=txid, target=target, token=SHARED_SECRET }
  local acks = send_and_collect(req, "ping", ACK_WINDOW)
  print_acks("ping", acks)
end

local function op_put(args, flags)
  if #args < 2 then die("put <src> <dst> [--all|--id N|--label L]") end
  local src, dst = args[1], args[2]
  local data, e = readAll(src); if not data then die("read: "..e) end
  local target = build_target(flags)
  local sum = adler32(data)

  if #data <= MAX_INLINE then
    local txid = gen_txid()
    local req = { op="put", txid=txid, target=target, fname=dst, data=data, checksum=sum, token=SHARED_SECRET }
    local acks = send_and_collect(req, "put", ACK_WINDOW)
    print_acks(("put %s -> %s"):format(src, dst), acks)
    return
  end

  -- chunked: begin
  local txid = gen_txid()
  local begin = { op="begin", txid=txid, target=target, fname=dst, size=#data, checksum=sum, token=SHARED_SECRET }
  local a_begin = send_and_collect(begin, "begin", STEP_WINDOW)
  print_acks(("begin %s -> %s"):format(src, dst), a_begin)

  -- chunk loop
  local sent = 0
  while sent < #data do
    local chunk = data:sub(sent+1, math.min(sent + CHUNK, #data))
    rednet.broadcast({ op="chunk", txid=txid, target=target, data=chunk }, PROTOCOL)
    local a_chunk = collect_acks(txid, "chunk", STEP_WINDOW)
    print(("[chunk %d..%d] acks=%d"):format(sent+1, sent+#chunk, #a_chunk))
    sent = sent + #chunk
  end

  -- end
  local endm = { op="end", txid=txid, target=target }
  local a_end = send_and_collect(endm, "end", STEP_WINDOW)
  print_acks("end", a_end)
end

local function op_exec(args, flags)
  if #args < 1 then
    die([[exec "<cmd line>" [--capture] [--timeout S] [--cwd /] [--all|--id N|--label L]]])
  end
  local line = args[1]
  local target = build_target(flags)
  local capture = flags["capture"] == true or flags["capture"] == "true"
  local timeout = tonumber(flags["timeout"] or "0")
  local cwd = flags["cwd"]

  local txid = gen_txid()
  local req = {
    op="exec", txid=txid, target=target, line=line,
    capture=capture, timeout=timeout, cwd=cwd, token=SHARED_SECRET
  }
  rednet.broadcast(req, PROTOCOL)
  local acks = collect_acks(txid, "exec", ACK_WINDOW)

  if capture then
    -- print each node's output
    for _, m in ipairs(acks) do
      local who = ("id=%s"):format(tostring(m.__sender))
      if m.id or m.label then who = ("%s label=%s id=%s"):format(tostring(m.label or ""), tostring(m.id or "?"), tostring(m.__sender)) end
      print(("--- %s ---"):format(who))
      if m.ok then print(m.output or "") else print("[error] "..tostring(m.note or "")) end
    end
  else
    print_acks(("exec %q"):format(line), acks)
  end
end

local function op_reboot(flags)
  -- Use exec to request reboot on targets. No capture.
  local target = build_target(flags)
  local txid = gen_txid()
  local req = { op="exec", txid=txid, target=target, line="reboot", capture=false, token=SHARED_SECRET }
  local acks = send_and_collect(req, "exec", 1.0)  -- quick window; they will drop off after
  print_acks("reboot", acks)
  print("Note: nodes may vanish from the network immediately after ACK.")
end

--------------------------
-- Main
--------------------------
local function main(...)
  openModems()
  math.randomseed(os.epoch("utc") % 2147483647)
  local argv = { ... }
  local parsed = parse_args(argv)
  if parsed.cmd == "ping" then
    op_ping(parsed.flags)
  elseif parsed.cmd == "put" then
    op_put(parsed.args, parsed.flags)
  elseif parsed.cmd == "exec" then
    op_exec(parsed.args, parsed.flags)
  elseif parsed.cmd == "reboot" then
    op_reboot(parsed.flags)
  else
    die("Unknown subcommand: "..tostring(parsed.cmd))
  end
end

local ok, err = pcall(main, ...)
if not ok then print("Error: "..tostring(err)) end