fwt_software/src/core/RemoteCommands.cpp

#include "fgc/RemoteCommands.h"

#include "fgc/Logger.h"

#include <algorithm>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>

namespace fgc {

namespace {

CommandReply refusal(const RemoteCommand& rc, const std::string& why) {
    CommandReply r;
    r.id    = rc.id;
    r.cmd   = rc.cmd;
    r.level = "error";
    r.lines = {why};
    return r;
}

bool containsAny(const std::string& line, const std::vector<std::string>& needles) {
    for (const auto& n : needles)
        if (line.find(n) != std::string::npos) return true;
    return false;
}

}  // namespace

// What the tap collects. The tap runs on whichever thread logs, under the
// logger's lock, so it only records - under its own small lock, since poll()
// reads it on the control thread while the tap may still be adding.
struct RemoteCommandRunner::Capture {
    std::mutex               m;
    std::thread::id          tid = std::this_thread::get_id();
    bool                     handling = true;   // the command is still running
    std::optional<Await>     await;
    bool                     ended = false;     // the awaited answer has arrived
    std::vector<std::string> lines;
    LogLevel                 worst = LogLevel::Info;

    void take(LogLevel level, const std::string& text) {
        std::lock_guard<std::mutex> lock(m);
        // While the command runs: everything it logs, on its own thread - the
        // camera, the MQTT client and the sensors log from theirs meanwhile, and
        // none of that is its answer. From any thread: the awaited answer.
        const bool mine    = handling && std::this_thread::get_id() == tid;
        const bool answer  = await && containsAny(text, await->collect);
        const bool closing = await && containsAny(text, await->ends);
        if (!mine && !answer && !closing) return;
        // A multi-line entry (the firmware dump) becomes one reply line per line.
        size_t from = 0;
        while (from <= text.size()) {
            const size_t nl = text.find('\n', from);
            const std::string part = text.substr(from, nl == std::string::npos ? nl : nl - from);
            if (!part.empty() || nl != std::string::npos) lines.push_back(part);
            if (nl == std::string::npos) break;
            from = nl + 1;
        }
        if (!lines.empty() && lines.back().empty()) lines.pop_back();
        if (level > worst) worst = level;
        if (closing) ended = true;
    }
};

struct RemoteCommandRunner::Pending {
    RemoteCommand            rc;
    std::shared_ptr<Capture> cap;
    long long                deadline_ms = 0;
};

std::optional<CommandReply> RemoteCommandRunner::run(const RemoteCommand& rc,
                                                     const Handler& handle,
                                                     const Await* await, long long now_ms) {
    const std::string tag = rc.id.empty() ? "" : " [" + rc.id + "]";

    if (!rc.error.empty()) {
        LOG_WARN << "remote command" << tag << " not run: " << rc.error;
        return refusal(rc, "not run: " + rc.error);
    }
    if (rc.retained) {
        LOG_WARN << "remote command" << tag << " ignored: it was a retained message ('" << rc.cmd
                 << "'), which the broker re-delivers on every reconnect";
        return refusal(rc, "not run: retained messages are never run - publish commands "
                           "without the retain flag");
    }
    if (!rc.id.empty()) {
        auto it = std::find_if(recent_.begin(), recent_.end(),
                               [&](const std::pair<std::string, CommandReply>& p) {
                                   return p.first == rc.id;
                               });
        if (it != recent_.end()) {
            LOG_INFO << "remote command" << tag << " is a repeat; answered again, not run again";
            return it->second;
        }
    }

    LOG_INFO << "remote command" << tag << ": " << rc.cmd;

    // The tap is armed before the command runs, so an answer that comes back
    // quickly is not missed; reading the buffer after setTap({}) is ordered by
    // the logger's lock.
    auto cap = std::make_shared<Capture>();
    if (await) cap->await = *await;
    Logger::setTap([cap](LogLevel level, const std::string& line) { cap->take(level, line); });
    try {
        handle(rc.cmd);
    } catch (...) {
        Logger::setTap({});
        throw;
    }

    bool wait = false;
    {
        std::lock_guard<std::mutex> lock(cap->m);
        cap->handling = false;
        // A command that warned (refused, busy, bad arguments) sent nothing to
        // wait for.
        wait = await && !cap->ended && cap->worst < LogLevel::Warn;
    }
    if (wait) {
        pending_ = std::make_shared<Pending>(Pending{rc, cap, now_ms + await->timeout_ms});
        return std::nullopt;
    }
    return finish(rc, *cap);
}

std::optional<CommandReply> RemoteCommandRunner::poll(long long now_ms) {
    if (!pending_) return std::nullopt;
    Capture& cap = *pending_->cap;
    {
        std::lock_guard<std::mutex> lock(cap.m);
        if (!cap.ended) {
            if (now_ms < pending_->deadline_ms) return std::nullopt;
            cap.lines.push_back("no answer from the firmware within " +
                                std::to_string(cap.await->timeout_ms / 1000) + " s");
            if (cap.worst < LogLevel::Warn) cap.worst = LogLevel::Warn;
        }
    }
    const std::shared_ptr<Pending> done = std::move(pending_);
    return finish(done->rc, *done->cap);
}

CommandReply RemoteCommandRunner::finish(const RemoteCommand& rc, Capture& cap) {
    Logger::setTap({});

    CommandReply r;
    r.id    = rc.id;
    r.cmd   = rc.cmd;
    {
        std::lock_guard<std::mutex> lock(cap.m);
        r.lines = std::move(cap.lines);
        r.level = cap.worst >= LogLevel::Error ? "error"
                : cap.worst >= LogLevel::Warn  ? "warn"
                                               : "info";
    }

    if (!rc.id.empty()) {
        recent_.emplace_back(rc.id, r);
        while (recent_.size() > kRememberIds) recent_.pop_front();
    }
    return r;
}

}  // namespace fgc