#include "WorkerSubprocess.hpp"
#include "ConnectionWorker.hpp"
#include "SafeWebsocketDeferredSend.h"
#include "common.h"
#include "json.h"
#include <App.h>
#include <chrono>
#include <regex>
#include "SocketOwnership.hpp"
#include "RTLogger.hpp"
#include "Base64.h"
#include "DisableCrashReporter.hpp"

using namespace juce;
using namespace jsoncons;

// PROCESS_BLOCK wire format:
// FROM client:
//   frameId: uint32
//   midi:
//     repeated:
//       samplePosition: int32 where -1 breaks repetition
//       eventLength: byte
//       eventBytes: byte[eventLength]
//   automationCount: uint32
//   parameterSampleStride: uint32
//   repeated automationCount times:
//     parameterIndex: uint32
//     automation:
//       runarray(samplePosition, 1):    indexed by time of change in input samples for this block
//         value: float32
//   inputCount: byte number of stereo inputs
//   outputCount: byte number of stereo outputs expected
//   sampleCount: uint32 number of samples per input channel
//   samples: float32[inputCount * sampleCount * 2] planar input-major
// TO client:
//   frameId: uint32
//   midi:
//     repeated:
//       samplePosition: int32 where -1 breaks repetition
//       eventLength: byte
//       eventBytes: byte[eventLength]
//   samples: float32[outputCount * sampleCount * 2] planar output-major
//
// runarray(index, increment): enclosed:
//   repeated:
//     startIndex: int32 indexing index where -1 breaks repetition
//     numItems: uint32
//     repeated numItems times:
//       enclosed @ startIndex + repetitionCount * increment

struct WInputStream {
    WInputStream(uint8_t * data, size_t len)
    : data(data), len(len), pos(0)
    { }
    
    void boundsCheck(bool expr) {
        if(!expr) {
            throw std::runtime_error("message too short");
        }
    }
    
    template<typename T>
    T readT() {
        boundsCheck(pos + sizeof(T) <= len);
        T res = 0;
        memcpy(&res, data + pos, sizeof(T));
        pos += sizeof(T);
        return res;
    }
    
    template<>
    uint8_t readT() {
        boundsCheck(pos < len);
        return data[pos++];
    }
    
    void readBlock(uint8_t *dest, size_t count) {
        boundsCheck(pos + count <= len);
        memcpy(dest, data + pos, count);
        pos += count;
    }
    
    // F(int32 index) -> void
    template<typename F>
    void readRunarray(uint32_t increment, F && callback) {
        while(pos < len) {
            auto startIndex = readT<int32_t>();
            if(startIndex == -1) {
                break;
            }
            auto numItems = readT<uint32_t>();
            for(uint32_t i = 0; i < numItems; i++) {
                callback(startIndex + i * increment);
            }
        }
    }
    
    size_t remain() {
        if(pos > len) {
            throw std::runtime_error("pos exceeds len");
        }
        return len - pos;
    }
    
    uint8_t * data;
    size_t len;
    size_t pos;
};

void processblock_wire_to_shm(ShmData & res, const std::string_view & data) {
    // Be careful with memory access here! `data' should be treated as malicious
    WInputStream stream((uint8_t*)data.data(), data.size());
    res.messageType = ShmMessageType::PROCESS_BLOCK;
    auto & pb = res.processBlock;
    pb.frameId = stream.readT<uint32_t>();
    
    // MIDI
    {
        pb.numMidiEvents = 0;
        for(;;) {
            auto samplePos = stream.readT<int32_t>();
            if(samplePos == -1) {
                break;
            }
            if(pb.numMidiEvents >= MAX_MIDI_EVENTS) {
                throw std::runtime_error("too many midi events");
            }
            auto & evt = pb.midiEvents[pb.numMidiEvents++];
            evt.sampleIndex = (uint32_t)samplePos;
            evt.length = stream.readT<uint8_t>();
            if(evt.length > MAX_MIDI_EVENT_LENGTH) {
                throw std::runtime_error("midi event too long");
            }
            stream.readBlock(evt.data, evt.length);
        }
    }
    
    // Automations
    {
        pb.numParameterAutomations = stream.readT<uint32_t>();
        if(pb.numParameterAutomations > MAX_PARAMETER_AUTOMATIONS) {
            throw std::runtime_error("too many parameter automations");
        }
        pb.parameterSampleStride = stream.readT<uint32_t>();
        for(size_t i = 0; i < pb.numParameterAutomations; i++) {
            auto & aut = pb.automations[i];
            aut.parameterIndex = stream.readT<uint32_t>();
            size_t lastIdx = 0;
            float lastVal = 0.f;
            stream.readRunarray(1, [&](uint32_t sampleIndex){
                if(sampleIndex >= MAX_BUFFER_SIZE) {
                    throw std::runtime_error("parameter automation too long");
                }
                for(size_t j = lastIdx; j < sampleIndex; j++) {
                    aut.values[j] = lastVal;
                }
                lastIdx = sampleIndex;
                lastVal = stream.readT<float>();
                aut.values[sampleIndex] = lastVal;
                return true;
            });
            for(size_t j = lastIdx + 1; j < MAX_BUFFER_SIZE; j++) {
                aut.values[j] = lastVal;
            }
        }
    }
    
    // Audio
    pb.numInputs = stream.readT<uint32_t>();
    pb.numOutputs = stream.readT<uint32_t>();
    if(pb.numInputs > MAX_CHANNEL_PAIRS || pb.numOutputs > MAX_CHANNEL_PAIRS) {
        throw std::runtime_error("too many ios");
    }
    pb.numSamplesPerChannel = stream.readT<uint32_t>();
    if(pb.numSamplesPerChannel > MAX_BUFFER_SIZE) {
        throw std::runtime_error("too many samples per channel");
    }
    for(size_t i = 0; i < pb.numInputs; i++) {
        auto & inp = pb.channelss[i];
        for(size_t j = 0; j < 2; j++) {
            stream.readBlock((uint8_t*)inp.samples[j], pb.numSamplesPerChannel * sizeof(float));
        }
    }
}


std::string_view shm_to_wire(ShmData & data) {
    static std::string res;
    if(data.messageType == ShmMessageType::PROCESS_BLOCK) {
        auto & pb = data.processBlock;
        const auto channelBytes = pb.numSamplesPerChannel * sizeof(float);
        size_t midiBytes = sizeof(int32_t);
        for(size_t i = 0; i < pb.numMidiEvents; i++) {
            midiBytes += sizeof(int32_t) + 1 + pb.midiEvents[i].length;
        }
        const uint32_t frameId = pb.frameId;
        const size_t requiredSize = channelBytes * pb.numOutputs * 2 + sizeof(frameId) + midiBytes;
        if(res.size() < requiredSize) {
            res.resize(requiredSize, 0);
        }
        uint8_t * data = (uint8_t*)res.data();
        memcpy(res.data(), &frameId, sizeof(frameId));
        data += sizeof(frameId);
        for(size_t i = 0; i < pb.numMidiEvents; i++) {
            const auto & evt = pb.midiEvents[i];
            int32_t samplePos = (int32_t)evt.sampleIndex;
            memcpy(data, &samplePos, sizeof(samplePos));
            data += sizeof(samplePos);
            *(data++) = evt.length;
            memcpy(data, evt.data, evt.length);
            data += evt.length;
        }
        const int32_t endMarker = -1;
        memcpy(data, &endMarker, sizeof(endMarker));
        data += sizeof(endMarker);
        for(size_t i = 0; i < pb.numOutputs; i++) {
            for(size_t j = 0; j < 2; j++) {
                memcpy(data, pb.channelss[i].samples[j], channelBytes);
                data += channelBytes;
            }
        }
        return std::string_view(res.begin(), res.begin() + requiredSize);
    } else if(data.messageType == ShmMessageType::SET_TRANSPORT) {
        // TODO avoid allocation
        auto s = json_to_string(PluginLatencyResponse{.samplesLatency = data.setTransport.samplesLatency});
        if(res.size() < s.size()) {
            res.resize(s.size(), 0);
        }
        std::copy(s.begin(), s.end(), res.begin());
        return std::string_view(res.begin(), res.begin() + s.size());
    } else {
        throw std::runtime_error("shm_to_wire: unknown messageType");
    }
}

class WebsocketThread : private Thread {
public:
    WebsocketThread(int bufferSize,
                    int sampleRate,
                    const std::string_view & bearerToken,
                    const std::string_view & bridgeKeySeed,
                    const std::string_view & architecture,
                    std::function<void(int)> && portCallback,
                    std::function<void()> && disarmTimeoutCallback,
                    std::function<void()> && doneCallback,
                    std::function<void(const std::string_view &)> && trustHandler,
                    std::function<void(bool)> && launchingChangedHandler,
                    std::function<void(bool)> && keyboardEnabledHandler)
      : Thread("Websocket thread"),
        bufferSize(bufferSize),
        sampleRate(sampleRate),
        bearerToken(bearerToken),
        shouldShutdown(false),
        listenSocket(nullptr),
        clientSocket(nullptr),
        eventLoop(nullptr),
        portCallback(std::move(portCallback)),
        disarmTimeoutCallback(std::move(disarmTimeoutCallback)),
        doneCallback(std::move(doneCallback)),
        trustHandler(std::move(trustHandler)),
        launchingChangedHandler(std::move(launchingChangedHandler)),
        keyboardEnabledHandler(std::move(keyboardEnabledHandler)),
        initialShouldStayOnTop(false),
        initialKeyboardEnabled(false),
        bridgeKeySeed(bridgeKeySeed),
        architecture(architecture)
    {
        shm.shouldReset = false;
    }
    
    void shutdown() {
        std::unique_lock lock(connectionStateMutex);
        shouldShutdown = true;
        if(!eventLoop) {
            return;
        }
        eventLoop->defer([this]{
            std::unique_lock lock(connectionStateMutex);
            if(listenSocket) {
                us_listen_socket_close(0, listenSocket);
                listenSocket = nullptr;
            }
            if(clientSocket) {
                us_socket_close(0, clientSocket, 0, nullptr);
                clientSocket = nullptr;
            }
        });
    }
    
    void setShouldStayOnTop(bool value) {
        std::unique_lock lock(initialShouldStayOnTopMutex);
        if(!worker) {
            initialShouldStayOnTop = value;
            return;
        }
        worker->setPluginWindowAlwaysOnTop(value);
    }

    void setKeyboardEnabled(bool value) {
        std::unique_lock lock(initialShouldStayOnTopMutex);
        if (!worker) {
            initialKeyboardEnabled = value;
            return;
        }
        worker->setKeyboardEnabled(value);
    }

    void keyboardActivity() {
        worker->keyboardActivity();
    }
    
    void trustResolution(const XmlElement & elem) {
        if(!worker) {
            return;
        }
        worker->trustResolution(elem);
    }
    
private:
    std::string id;
    
    struct ConnectionUserData {
        std::unique_ptr<SafeWebsocketDeferredSend> safeSend;
    };
    
    using clock_t = std::chrono::high_resolution_clock;
    
    void run() override {
        try {
            uint64_t tid;
#ifdef _MSC_VER
            tid = GetCurrentThreadId();
#else
            pthread_threadid_np(NULL, &tid);
#endif
            std::stringstream ss;
            ss << tid;
            id = ss.str();
            
            realRun();
        } catch(const std::exception & e) {
            RTLogger::log("Exception in realtime thread %s: %s", id.c_str(), e.what());
        } catch(...) {
            RTLogger::log("Exception in realtime thread %s (non-std)", id.c_str());
        }
        {
            std::unique_lock lock(connectionStateMutex);
            eventLoop = nullptr;
        }
        doneCallback();
        doneCallback = {};
    }
    
    void realRun() {
        RTLogger::log("Realtime thread %s staring: bufferSize=%d, sampleRate=%d", id.c_str(), bufferSize, sampleRate);

        uWS::Loop::get()->setLockFree(true);
        loadPluginLists(pluginListByArchitecture);

        auto app = uWS::App().ws<ConnectionUserData>("/*", {
            /* Settings */
            .maxPayloadLength = 100 * 1024 * 1024,
            .idleTimeout = 16,
            .maxBackpressure = 100 * 1024 * 1024,
            .closeOnBackpressureLimit = false,
            .resetIdleTimeoutOnSend = false,
            .sendPingsAutomatically = true,
            /* Handlers */
            .upgrade = [this](auto *res, auto *req, auto *context) {
                if(!validateOriginHeader(req)) {
                    res->close();
                    return;
                }
                
                std::string accessTokenProtocol = "access_token";
                std::set<std::string> protocols;
                {
                    auto wsProtocol = std::string(req->getHeader("sec-websocket-protocol"));
                    std::regex protoSplitRegex(",\\s*");
                    std::sregex_token_iterator iter(wsProtocol.begin(), wsProtocol.end(), protoSplitRegex, -1);
                    std::sregex_token_iterator end;
                    for (; iter != end; ++iter) {
                        protocols.insert(*iter);
                    }
                }
                if(protocols.find(accessTokenProtocol) == protocols.end() || protocols.find(bearerToken) == protocols.end()) {
                    res->writeStatus("403 Forbidden")->end();
                    return;
                }
                
                res->template upgrade<ConnectionUserData>({},
                    req->getHeader("sec-websocket-key"),
                    accessTokenProtocol,
                    req->getHeader("sec-websocket-extensions"),
                    context);
            },
            .open = [=,this](auto * ws) {
                std::unique_lock lock(connectionStateMutex);
                if(shouldShutdown) {
                    ws->end();
                    return;
                }
                auto ctx = ws->getUserData();
                
                auto [peerAddress, peerPort] = getPeerAddressAndPort(ws);
                this->peerAddress = peerAddress;
                this->peerPort = peerPort;
                
                if(peerAddress != "127.0.0.1") {
                    ws->end();
                    return;
                }

#ifdef _MSC_VER
                auto maybeMySsid = getStringSidForProcess(GetCurrentProcess());
                auto maybeSsid = findLocalhostRemoteSocketOwner(peerPort);
                if (!maybeMySsid.has_value() || !maybeSsid.has_value() || maybeMySsid.value() != maybeSsid.value()) {
                    ws->end();
                    return;
                }
#else
                auto uid = findLocalhostRemoteSocketOwner(peerPort);
                if(!uid.has_value() || uid.value() != getuid()) {
                    ws->end();
                    return;
                }
#endif
                
                ctx->safeSend.reset(new SafeWebsocketDeferredSendT(ws, uWS::Loop::get()));
                
                disarmTimeoutCallback();
                disarmTimeoutCallback = {};
                if(listenSocket) {
                    us_listen_socket_close(0, listenSocket);
                    listenSocket = nullptr;
                }
                
                clientSocket = (us_socket_t*)ws;
                RTLogger::log("Realtime thread %s new connection from %s:%d", id.c_str(), peerAddress.c_str(), peerPort);
            },
            .message = [this](auto *ws, std::string_view message, uWS::OpCode opCode) {
                auto startTime = clock_t::now();
                
                if(opCode == uWS::OpCode::TEXT) {
                    // control message
                    try {
                        handleControlMessage(message, ws);
                    } catch(std::exception & e) {
                        ws->send(json_to_string(ErrorResponse{.error = e.what()}), uWS::OpCode::TEXT);
                    }
                } else if(opCode == uWS::OpCode::BINARY) {
                    // midi + audio frames
                    if(!worker) {
                        return;
                    }
                    
                    processblock_wire_to_shm(shm, message);
                    worker->handleBlock(shm);
                    ws->send(shm_to_wire(shm), uWS::OpCode::BINARY);
                }
                
                observeMessageHandlerDuration(clock_t::now() - startTime);
            },
            .dropped = [](auto */*ws*/, std::string_view /*message*/, uWS::OpCode /*opCode*/) {
                /* A message was dropped due to set maxBackpressure and closeOnBackpressureLimit limit */
            },
            .drain = [](auto */*ws*/) {
                /* Check ws->getBufferedAmount() here */
            },
            .ping = [](auto */*ws*/, std::string_view) {
                /* Not implemented yet */
            },
            .pong = [](auto */*ws*/, std::string_view) {
                /* Not implemented yet */
            },
            .close = [this](auto * ws, int /*code*/, std::string_view /*message*/) {
                // TODO
                RTLogger::log("Realtime thread %s connection closed: %s:%d", id.c_str(), peerAddress.c_str(), peerPort);
            }
        });
        
        app.any("/*", [](auto *res, auto *req) {
            res->close();
        });
        
        {
            std::unique_lock lock(connectionStateMutex);
            if(!shouldShutdown) {
                eventLoop = uWS::Loop::get();
                listenSocket = nullptr;
                for(int p = 9020; p < 9050 && !listenSocket; p++) {
                    app.listen("127.0.0.1", p, LIBUS_LISTEN_EXCLUSIVE_PORT, [&,this](auto *listen_socket) {
                        if (listen_socket) {
                            RTLogger::log("Realtime thread %s listening on %d", id.c_str(), p);
                            listenSocket = listen_socket;
                            portCallback(p);
                            portCallback = {};
                        }
                    });
                }
            } else {
                return;
            }
        }
        
        if(!listenSocket) {
            throw std::runtime_error("Failed to listen on any port in range");
        }
        
        app.run();
    }
    
    void observeMessageHandlerDuration(const clock_t::duration & dt) {
        static clock_t::time_point last_report_time = clock_t::now();
        static float tavg = 0.f, tmin = std::numeric_limits<float>::infinity(), tmax = 0.f;
        auto dtMs = (float)std::chrono::duration_cast<std::chrono::microseconds>(dt).count() / 1000.f;
        tavg += dtMs;
        tmin = std::min(tmin, dtMs);
        tmax = std::max(tmax, dtMs);
        if(clock_t::now() - last_report_time > std::chrono::seconds(5)) {
            RTLogger::log("RT %s: avg %.2fms, min %.2fms, max %.2fms", id.c_str(), tavg / 1000.0, (double)tmin, (double)tmax);
            tavg = 0.f;
            tmin = std::numeric_limits<float>::infinity();
            tmax = 0.f;
            last_report_time = clock_t::now();
        }
    }
    
    template<typename WebSocket>
    void handleControlMessage(const std::string_view & message, WebSocket * ws) {
        auto j = decode_json<ClientControlMessage>(message);
        auto safeSend = ws->getUserData()->safeSend->get();
        
        switch(j.op) {
            case ClientOp::listInstanceParameters:
            {
                // ...
                break;
            }
            case ClientOp::setTransportState:
            {
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                
                shm.messageType = ShmMessageType::SET_TRANSPORT;
                shm.setTransport = decode_json<ShmSetTransportData>(message);
                worker->handleBlock(shm);
                ws->send(shm_to_wire(shm), uWS::OpCode::TEXT);
                
                break;
            }
            case ClientOp::resetPlugin:
            {
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                
                shm.shouldReset = true;
                break;
            }
            case ClientOp::instantiatePlugin:
            {
                if(worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "Plugin already instantiated"}), uWS::OpCode::TEXT);
                    return;
                }
                
                auto pluginListIt = pluginListByArchitecture.find(architecture);
                if(pluginListIt == pluginListByArchitecture.end()) {
                    ws->send(json_to_string(ErrorResponse{.error = "Architecture not found"}), uWS::OpCode::TEXT);
                    return;
                }
                auto & pluginList = pluginListIt->second;
                
                auto j2 = decode_json<ClientInstantiatePluginMessage>(message);
                std::unique_ptr<PluginDescription> desc;
                
                for(auto & type : pluginList->getTypes()) {
                    if(type.fileOrIdentifier == j2.pluginIdentifier) {
                        desc = std::make_unique<PluginDescription>(type);
                        break;
                    }
                }
                
                if(!desc) {
                    ws->send(json_to_string(ErrorResponse{.error = "Plugin not found"}), uWS::OpCode::TEXT);
                    return;
                }
                
                auto loadedHandler = [=] {
                    safeSend->send(json_to_string(EventResponse{.event = "loaded"}), uWS::OpCode::TEXT);
                };
                
                auto errorHandler = [=](const std::string & message) {
                    RTLogger::log("errorHandler called: %s", message.c_str());
                    safeSend->send(json_to_string(ErrorResponse{.error = message}), uWS::OpCode::TEXT);
                };
                
                auto saveHandler = [=](const std::string_view & data, bool periodic) {
                    EncodedStateMessage msg;
                    msg.encodedState = convertToBase64(data);
                    msg.periodic = periodic;
                    safeSend->send(json_to_string(msg), uWS::OpCode::TEXT);
                };
                
                auto parameterGestureHandler = [=](const ParameterDescription & desc) {
                    safeSend->send(json_to_string(desc), uWS::OpCode::TEXT);
                };
                
                auto keypressHandler = [=](const KeyDescription & desc) {
                    safeSend->send(json_to_string(desc), uWS::OpCode::TEXT);
                };
                
                auto editorClosedHandler = [=]() {
                    safeSend->send(json_to_string(EditorClosedResponse()), uWS::OpCode::TEXT);
                };
                
                auto editorMovedHandler = [=](std::pair<int,int> pos) {
                    safeSend->send(json_to_string(EditorMovedResponse(pos.first, pos.second)), uWS::OpCode::TEXT);
                };
                
                auto latencyChangedHandler = [=](int samplesLatency) {
                    safeSend->send(json_to_string(PluginLatencyResponse{.samplesLatency = samplesLatency}), uWS::OpCode::TEXT);
                };
                
                std::unique_lock lock(initialShouldStayOnTopMutex);
                
                worker = std::make_unique<ConnectionWorker>(*desc,
                                                            bufferSize,
                                                            sampleRate,
                                                            std::move(bridgeKeySeed),
                                                            std::move(loadedHandler),
                                                            std::move(errorHandler),
                                                            std::move(saveHandler),
                                                            std::move(parameterGestureHandler),
                                                            std::move(keypressHandler),
                                                            std::move(editorClosedHandler),
                                                            std::move(editorMovedHandler),
                                                            std::move(trustHandler),
                                                            std::move(latencyChangedHandler),
                                                            std::move(launchingChangedHandler),
                                                            std::move(keyboardEnabledHandler));
                worker->setPluginWindowAlwaysOnTop(initialShouldStayOnTop);
                worker->setKeyboardEnabled(initialKeyboardEnabled);
                trustHandler = {};
                
                break;
            }
            case ClientOp::loadState:
            {
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                auto j2 = decode_json<EncodedStateMessage>(message);
                worker->loadState(convertFromBase64(j2.encodedState));
                break;
            }
            case ClientOp::saveState:
            {
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                worker->requestSaveState(false);
                break;
            }
            case ClientOp::openEditor:
            {
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                auto j2 = decode_json<ClientOpenEditorMessage>(message);
                std::optional<std::pair<int,int>> pos;
                if(j2.x.has_value() && j2.y.has_value()) {
                    pos = std::make_pair(j2.x.value(), j2.y.value());
                }
                worker->openPluginWindow(pos);
                break;
            }
            case ClientOp::closeEditor:
                if(!worker) {
                    ws->send(json_to_string(ErrorResponse{.error = "No plugin instance"}), uWS::OpCode::TEXT);
                    return;
                }
                worker->closePluginWindow();
                break;
            default:
                ws->send(json_to_string(ErrorResponse{.error = "Operation must be handled by host"}), uWS::OpCode::TEXT);
                break;
        }
    }
    
    friend class WorkerSubprocess;
    const int bufferSize;
    const int sampleRate;
    const std::string bearerToken;
    
    ShmData shm;
    std::unique_ptr<ConnectionWorker> worker;
    std::map<std::string, std::unique_ptr<KnownPluginList>> pluginListByArchitecture;
    std::string peerAddress;
    int peerPort;
    
    std::mutex connectionStateMutex;
    bool shouldShutdown;
    us_listen_socket_t * listenSocket;
    us_socket_t * clientSocket;
    uWS::Loop * eventLoop;
    
    std::function<void(int)> portCallback;
    std::function<void()> disarmTimeoutCallback;
    std::function<void()> doneCallback;
    std::function<void(const std::string_view&)> trustHandler;
    std::function<void(bool)> launchingChangedHandler;
    std::function<void(bool)> keyboardEnabledHandler;
    
    std::mutex initialShouldStayOnTopMutex;
    bool initialShouldStayOnTop;
    bool initialKeyboardEnabled;
    
    std::string bridgeKeySeed;
    std::string architecture;
};

WorkerSubprocess::WorkerSubprocess()
  : lastForegroundMessage(false),
    shuttingDown(false)
{

}

WorkerSubprocess::~WorkerSubprocess()
{

}

bool WorkerSubprocess::initialiseFromCommandLine (const String& commandLine,
                                                  const String& commandLineUniqueID,
                                                  int timeoutMs)
{
    auto res = ChildProcessWorker::initialiseFromCommandLine(commandLine, commandLineUniqueID, timeoutMs);
    if(res) {
        disableCrashReporter();
    }
    return res;
}

void WorkerSubprocess::handleMessageFromCoordinator (const MemoryBlock& mb)
{
    if (mb.isEmpty()) {
        return;
    }
    
    auto elem = parseXML(mb.toString());
    if (!elem) {
        throw std::runtime_error("Failed to parse message to worker subprocess");
    }
    
    if(elem->getTagName() == "WorkerSubprocess") {
        startup(std::move(elem));
    } else if(elem->getTagName() == "ShouldStayOnTop") {
        if(!workerThread) {
            return;
        }
        workerThread->setShouldStayOnTop(elem->getBoolAttribute("value"));
    } else if(elem->getTagName() == "TrustResolution") {
        auto trustStoreElem = elem->getFirstChildElement();
        if(!workerThread || !trustStoreElem) {
            return;
        }
        workerThread->trustResolution(*trustStoreElem);
    }
    else if (elem->getTagName() == "KeyboardEnabled") {
        if (!workerThread) {
            return;
        }
        workerThread->setKeyboardEnabled(elem->getBoolAttribute("value"));
    }
    else if (elem->getTagName() == "KeyboardActivity") {
        if (!workerThread) {
            return;
        }
        workerThread->keyboardActivity();
    }
}
    
void WorkerSubprocess::startup(std::unique_ptr<juce::XmlElement> startupArgs)
{
    if (workerThread) {
        throw std::runtime_error("Attempted reinitialization of WorkerSubprocess");
    }

#ifdef _MSC_VER
    // assume active while launching otherwise all editors close
    updateLaunching(true);
#endif
    
    if (!startupArgs) {
        throw std::runtime_error("Failed to parse startup args");
    }
    int bufferSize = startupArgs->getIntAttribute("bufferSize");
    int sampleRate = startupArgs->getIntAttribute("sampleRate");
    auto bearerToken = startupArgs->getStringAttribute("bearerToken").toStdString();
    auto bridgeKeySeed = convertFromBase64(startupArgs->getStringAttribute("bridgeKeySeed"));
    auto architecture = startupArgs->getStringAttribute("architecture").toStdString();
    
    if(bufferSize == 0 || sampleRate == 0 || bearerToken.empty()) {
        throw std::runtime_error("Invalid startup args");
    }
    
    workerThread = std::make_unique<WebsocketThread>(bufferSize,
                                                     sampleRate,
                                                     bearerToken,
                                                     bridgeKeySeed,
                                                     architecture,
    [this](int port) {
        auto message = std::make_unique<XmlElement>("WorkerSubprocessStarted");
        message->setAttribute("port", port);
        auto xmlString = message->toString();
        sendMessageToCoordinator({ xmlString.toRawUTF8(), xmlString.getNumBytesAsUTF8() });
    }, [this] {
        timeoutArmed = false;
    }, [this] {
        triggerAsyncUpdate();
    }, [this](const std::string_view & publicKey) {
        auto message = std::make_unique<XmlElement>("WorkerSubprocessTrustQuery");
        message->setAttribute("publicKey", convertToBase64(publicKey));
        auto xmlString = message->toString();
        sendMessageToCoordinator({ xmlString.toRawUTF8(), xmlString.getNumBytesAsUTF8() });
    }, [this](bool isLaunching) {
        updateLaunching(isLaunching);
    }, [this](bool keyboardEnabled) {
        auto message = std::make_unique<XmlElement>("WorkerKeyboardEnabled");
        message->setAttribute("value", keyboardEnabled);
        auto xmlString = message->toString();
        sendMessageToCoordinator({ xmlString.toRawUTF8(), xmlString.getNumBytesAsUTF8() });
    });
    
    // allow 50% of the frame budget in this VST, to a maximum of 30ms
    auto perFrameTimeMs = std::fmin(1000.0 * 0.5 * (double)bufferSize / (double)sampleRate, 30);
    auto periodMs = 1000.0 * (double)bufferSize / (double)sampleRate;
    
    workerThread->startRealtimeThread(Thread::RealtimeOptions{}.withMaximumProcessingTimeMs(perFrameTimeMs).withPeriodMs(periodMs));
    timeoutArmed = true;
    startTime = std::chrono::steady_clock::now();
    startTimer(50);
}

void WorkerSubprocess::handleConnectionLost()
{
    triggerAsyncUpdate();
}

void WorkerSubprocess::handleAsyncUpdate()
{
    if(shuttingDown) {
        return;
    }
    shuttingDown = true;
    stopTimer();
    if(workerThread) {
        workerThread->shutdown();
        workerThread->waitForThreadToExit(5000);
        workerThread.reset();
    }
    JUCEApplicationBase::quit();
}

void WorkerSubprocess::updateLaunching(bool foregroundMessage)
{
    bool expected = lastForegroundMessage;
    if (foregroundMessage == expected) {
        return;
    }
    if (!lastForegroundMessage.compare_exchange_strong(expected, foregroundMessage)) {
        return;
    }
    auto message = std::make_unique<XmlElement>("WorkerSubprocessForegroundChanged");
    message->setAttribute("foreground", foregroundMessage);
    auto xmlString = message->toString();
    sendMessageToCoordinator({ xmlString.toRawUTF8(), xmlString.getNumBytesAsUTF8() });
}

void WorkerSubprocess::timerCallback()
{
    if(timeoutArmed && std::chrono::steady_clock::now() - startTime > std::chrono::seconds(10)) {
        triggerAsyncUpdate();
    }
    
#ifndef _MSC_VER
    updateLaunching(Process::isForegroundProcess());
#endif
}