---

proto::RequestAck Handle(const proto::WhiteListApplyReq& req) {
    proto::RequestAck ack;
    ack.set_request_id(req.request_id());

    std::vector<Live> active;
    if (int r = systemd_.ListActiveInstances(&active); r < 0) {
        ack.set_result(proto::RESULT_ERROR);
        ack.set_detail("systemd query failed");
        return ack;
    }

    std::vector<std::string> entries(req.vbs_ip_address_list().begin(),
                                     req.vbs_ip_address_list().end());
    Plan plan = MakePlan(entries, active);

    // предохранитель 1: не опознали — не действуем
    if (!plan.unresolved.empty()) {
        ack.set_result(proto::RESULT_ERROR);
        ack.set_detail("unresolved: " + Join(plan.unresolved, ","));
        return ack;
    }

    // предохранитель 2: массовая остановка
    if (!active.empty() && plan.to_stop.size() * 2 > active.size()
        && !entries.empty()) {
        ack.set_result(proto::RESULT_ERROR);
        ack.set_detail("refusing to stop " + std::to_string(plan.to_stop.size())
                       + " of " + std::to_string(active.size()));
        return ack;
    }

    for (uint16_t p : plan.to_start) {
        if (!systemd_.Start(UnitName(p, IpForPort(entries, p)))) { /* ERROR */ }
    }
    for (uint16_t p : plan.to_stop) {
        if (!systemd_.Stop(UnitNameOf(active, p))) { /* ERROR */ }
    }

    ack.set_result(proto::RESULT_OK);
    return ack;
}

Plan MakePlan(const std::vector<std::string>& raw_entries,
              const std::vector<Live>&        active)
{
    Plan plan;
    std::map<std::string, uint16_t> ip_to_port;   // из active
    std::set<uint16_t> active_ports;
    for (const auto& l : active) {
        ip_to_port[l.ip] = l.port;
        active_ports.insert(l.port);
    }

    std::set<uint16_t> desired_ports;
    for (const auto& raw : raw_entries) {
        auto e = ParseEntry(raw);
        if (!e) { plan.unresolved.push_back(raw); continue; }

        if (e->port) {
            desired_ports.insert(*e->port);          // порт пришёл
        } else if (auto it = ip_to_port.find(e->ip); it != ip_to_port.end()) {
            desired_ports.insert(it->second);        // порт известен из active
        } else {
            plan.unresolved.push_back(e->ip);        // опознать нечем
        }
    }

    std::set_difference(desired_ports.begin(), desired_ports.end(),
                        active_ports.begin(), active_ports.end(),
                        std::back_inserter(plan.to_start));
    std::set_difference(active_ports.begin(), active_ports.end(),
                        desired_ports.begin(), desired_ports.end(),
                        std::back_inserter(plan.to_stop));
    return plan;
}