Source/Plugin/HelperConnection.cpp

#include "HelperConnection.h"
#include "Constants.h"

#include <sys/socket.h>
#include <sys/un.h>
#include <sys/wait.h>
#include <unistd.h>
#include <signal.h>
#include <poll.h>
#include <spawn.h>

extern char** environ;

#include "json.hpp"
using json = nlohmann::json;

HelperConnection::HelperConnection()
{
    myPid = getpid();
    // Ignore SIGPIPE globally (macOS doesn't have MSG_NOSIGNAL)
    signal(SIGPIPE, SIG_IGN);
}

HelperConnection::~HelperConnection()
{
    killHelper();
}

bool HelperConnection::spawnHelper()
{
    killHelper();

    // 1. Create shared memory
    auto shmName = IPC::getShmName(myPid);
    if (!sharedAudio.create(shmName))
    {
        DBG("HelperConnection: Failed to create shared memory");
        return false;
    }

    // 2. Create semaphores
    if (!inputSem.create(IPC::getInputSemName(myPid)) ||
        !outputSem.create(IPC::getOutputSemName(myPid)))
    {
        DBG("HelperConnection: Failed to create semaphores");
        return false;
    }

    // 3. Create Unix domain socket (server side)
    auto socketPath = IPC::getSocketPath(myPid);
    unlink(socketPath.c_str());

    serverSocketFd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (serverSocketFd < 0)
    {
        DBG("HelperConnection: Failed to create socket");
        return false;
    }

    struct sockaddr_un addr;
    memset(&addr, 0, sizeof(addr));
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, socketPath.c_str(), sizeof(addr.sun_path) - 1);

    if (bind(serverSocketFd, (struct sockaddr*)&addr, sizeof(addr)) < 0 ||
        listen(serverSocketFd, 1) < 0)
    {
        DBG("HelperConnection: Failed to bind/listen socket");
        close(serverSocketFd);
        serverSocketFd = -1;
        return false;
    }

    // 4. Find helper binary
    auto helperPath = juce::File::getSpecialLocation(juce::File::currentExecutableFile)
                          .getParentDirectory()  // MacOS/
                          .getParentDirectory()  // Contents/
                          .getChildFile("Resources")
                          .getChildFile(PluginBridgeConstants::kHelperBinaryName);

    if (!helperPath.exists())
    {
        helperPath = juce::File::getSpecialLocation(juce::File::currentExecutableFile)
                         .getParentDirectory()
                         .getChildFile(PluginBridgeConstants::kHelperBinaryName);
    }

    if (!helperPath.exists())
    {
        DBG("HelperConnection: Helper binary not found at: " + helperPath.getFullPathName());
        return false;
    }

    // 5. Spawn helper process
    state.store(State::Starting);

    auto pidStr = juce::String(myPid).toStdString();
    auto helperPathStr = helperPath.getFullPathName().toStdString();

    pid_t pid;
    char* argv[] = {
        (char*)helperPathStr.c_str(),
        (char*)"--pid",
        (char*)pidStr.c_str(),
        nullptr
    };

    int spawnResult = posix_spawn(&pid, helperPathStr.c_str(), nullptr, nullptr, argv, environ);
    if (spawnResult != 0)
    {
        DBG("HelperConnection: Failed to spawn helper: " + juce::String(strerror(spawnResult)));
        state.store(State::Disconnected);
        return false;
    }

    helperPid = pid;
    helperRunning.store(true);

    // 6. Accept connection from helper (with timeout)
    struct pollfd pfd;
    pfd.fd = serverSocketFd;
    pfd.events = POLLIN;

    if (poll(&pfd, 1, IPC::kHelperTimeoutMs) <= 0)
    {
        DBG("HelperConnection: Helper didn't connect in time");
        killHelper();
        return false;
    }

    clientSocketFd = accept(serverSocketFd, nullptr, nullptr);
    if (clientSocketFd < 0)
    {
        DBG("HelperConnection: Failed to accept helper connection");
        killHelper();
        return false;
    }

    // Set SO_NOSIGPIPE on the accepted socket (macOS)
    int nosigpipe = 1;
    setsockopt(clientSocketFd, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe, sizeof(nosigpipe));

    // 7. Start IPC listen thread
    ipcRunning.store(true);
    ipcThread = std::thread([this]() { ipcListenLoop(); });

    // 8. Start crash detection timer
    startTimerHz(4);

    // Wait for "ready" notification
    int waitMs = 0;
    while (state.load() == State::Starting && waitMs < IPC::kHelperTimeoutMs)
    {
        juce::Thread::sleep(50);
        waitMs += 50;
    }

    return state.load() == State::Connected;
}

void HelperConnection::killHelper()
{
    stopTimer();

    ipcRunning.store(false);
    if (clientSocketFd >= 0) { close(clientSocketFd); clientSocketFd = -1; }
    if (serverSocketFd >= 0) { close(serverSocketFd); serverSocketFd = -1; }
    if (ipcThread.joinable()) ipcThread.join();

    if (helperPid > 0 && helperRunning.load())
    {
        kill(helperPid, SIGTERM);
        int status;
        waitpid(helperPid, &status, WNOHANG);
        usleep(100000);
        kill(helperPid, SIGKILL);
        waitpid(helperPid, &status, 0);
    }
    helperPid = 0;
    helperRunning.store(false);

    sharedAudio.destroy();
    inputSem.destroy();
    outputSem.destroy();

    unlink(IPC::getSocketPath(myPid).c_str());
    state.store(State::Disconnected);
}

std::string HelperConnection::sendCommand(const std::string& jsonCmd)
{
    std::lock_guard<std::mutex> lock(commandMutex);

    if (clientSocketFd < 0 || !helperRunning.load())
        return "{\"ok\":false,\"error\":\"not connected\"}";

    auto data = jsonCmd + "\n";
    if (::send(clientSocketFd, data.c_str(), data.size(), 0) < 0)
        return "{\"ok\":false,\"error\":\"send failed\"}";

    responseReady.store(false);
    int waitMs = 0;
    while (!responseReady.load() && waitMs < 5000)
    {
        juce::Thread::sleep(10);
        waitMs += 10;
    }

    if (!responseReady.load())
        return "{\"ok\":false,\"error\":\"timeout\"}";

    return lastResponse;
}

void HelperConnection::timerCallback()
{
    if (helperPid <= 0) return;

    int status;
    pid_t result = waitpid(helperPid, &status, WNOHANG);

    if (result == helperPid)
    {
        helperRunning.store(false);
        state.store(State::Crashed);
        stopTimer();
        DBG("HelperConnection: Helper process died");
        if (onHelperCrashed)
            onHelperCrashed();
    }
}

void HelperConnection::ipcListenLoop()
{
    std::string buffer;
    char chunk[4096];

    while (ipcRunning.load())
    {
        struct pollfd pfd;
        pfd.fd = clientSocketFd;
        pfd.events = POLLIN;

        int ret = poll(&pfd, 1, 200);
        if (ret <= 0) continue;

        ssize_t n = recv(clientSocketFd, chunk, sizeof(chunk) - 1, 0);
        if (n <= 0)
        {
            if (ipcRunning.load())
            {
                helperRunning.store(false);
                state.store(State::Crashed);
                if (onHelperCrashed)
                    juce::MessageManager::callAsync([this]() { if (onHelperCrashed) onHelperCrashed(); });
            }
            return;
        }

        chunk[n] = '\0';
        buffer += chunk;

        size_t pos;
        while ((pos = buffer.find('\n')) != std::string::npos)
        {
            auto msg = buffer.substr(0, pos);
            buffer.erase(0, pos + 1);

            if (msg.empty()) continue;

            try
            {
                auto j = json::parse(msg);
                if (j.contains("notify"))
                    handleNotification(msg);
                else
                {
                    lastResponse = msg;
                    responseReady.store(true);
                }
            }
            catch (...) {}
        }
    }
}

void HelperConnection::handleNotification(const std::string& jsonStr)
{
    try
    {
        auto j = json::parse(jsonStr);
        auto notify = j.value("notify", "");

        if (notify == "ready")
        {
            state.store(State::Connected);
            if (onHelperReady)
                juce::MessageManager::callAsync([this]() { if (onHelperReady) onHelperReady(); });
        }
        else if (notify == "param_changed")
        {
            int index = j.value("index", -1);
            float value = j.value("value", 0.0f);
            if (onParamChanged)
                onParamChanged(index, value);
        }
        else if (notify == "crashed")
        {
            state.store(State::Crashed);
            if (onHelperCrashed)
                juce::MessageManager::callAsync([this]() { if (onHelperCrashed) onHelperCrashed(); });
        }
    }
    catch (...) {}
}