// calogSsh.c -- calog SSH library (see calogSsh.h). SSH command execution and SFTP file
// transfer over the shared typed handle table, bound to libssh2. Every blocking call is an
// inline native, so it stalls only the calling script's context thread. Sessions run in
// libssh2 blocking mode. A connection handle has no owning context -- any context it is passed
// to may use it -- so the caller must serialize access (see the threading notes in calogSsh.h).

#define _GNU_SOURCE

#include "calogSsh.h"

#include "calogHandle.h"
#include "calogInternal.h"

#include <netdb.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <unistd.h>

#include <libssh2.h>
#include <libssh2_sftp.h>

// Sole handle type tag for this library, distinct so a stray handle of another kind fails to
// resolve here.
#define SSH_TYPE_CONN 1u

#define SSH_PORT_MAX 65535

// Upper bound on a single read-to-EOF transfer (sshExec output, sftpGet file), matching
// calogNet/calogHttp: without it a huge/endless remote source would grow an unbounded buffer
// and OOM-kill the shared host process. Exceeding it fails with calogErrRangeE; a stream of
// exactly this many bytes succeeds (see sshDrainStep's boundary probe).
#define SSH_MAX_TRANSFER (64 * 1024 * 1024)

// libssh2_sftp_readdir fails the whole listing (LIBSSH2_ERROR_BUFFER_TOO_SMALL) on a remote
// entry name at or beyond this size; real filesystems cap names at 255 bytes (NAME_MAX), so
// this leaves comfortable margin.
#define SSH_NAME_MAX 512

// What a connection handle owns: the socket, the libssh2 session, and a lazily-opened+cached
// SFTP subsystem. The handle is a plain process-wide entry with no owning context: any context
// it is passed to may operate on it (see the threading notes in calogSsh.h).
typedef struct SshConnT {
    int             sock;
    LIBSSH2_SESSION *session;
    LIBSSH2_SFTP    *sftp;
} SshConnT;

// Process-wide SSH library state shared by every runtime that registers the natives; refCount
// is one per registered runtime, so libssh2_exit runs (and the table is destroyed) only when
// the LAST runtime shuts down.
typedef struct SshLibT {
    CalogHandleTableT *handles;
    int32_t            refCount;
} SshLibT;

// Read adapter so sshDrainStep works uniformly over a channel stream (libssh2_channel_read_ex)
// and an SFTP file (libssh2_sftp_read); the SFTP adapter ignores streamId.
typedef ssize_t (*SshReadFnT)(void *handle, int streamId, char *buffer, size_t length);

// Outcome of one sshDrainStep call.
typedef enum SshStepE {
    sshStepDataE,
    sshStepEofE,
    sshStepAgainE
} SshStepE;

static pthread_mutex_t gSshLibMutex = PTHREAD_MUTEX_INITIALIZER;
static SshLibT        *gSshLib      = NULL;

static int32_t sftpGet(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpList(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpMkdir(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpPut(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpRemove(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpStat(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshAuthKey(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshAuthPassword(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static void    sshCloser(uint32_t type, void *resource);
static void    sshConnDestroy(SshConnT *conn, const char *reason);
static int32_t sshConnect(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshDrainStep(SshReadFnT readFn, void *handle, int streamId, char **bytes, int64_t *cap, int64_t *length, SshStepE *step);
static int32_t sshEnsureSftp(SshConnT *conn, CalogValueT *result);
static int32_t sshExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshExecDrain(int sock, LIBSSH2_SESSION *session, LIBSSH2_CHANNEL *channel, char **outBytes, int64_t *outLength, char **errBytes, int64_t *errLength);
static ssize_t sshReadChannel(void *handle, int streamId, char *buffer, size_t length);
static ssize_t sshReadSftpFile(void *handle, int streamId, char *buffer, size_t length);
static int32_t sshResolve(SshLibT *lib, int64_t handle, CalogValueT *result, SshConnT **out);
static void    sshSessionDestroy(LIBSSH2_SESSION *session, int sock, const char *reason);
static void    sshWaitSocket(int sock, LIBSSH2_SESSION *session);


int32_t calogSshRegister(CalogT *calog) {
    pthread_mutex_lock(&gSshLibMutex);
    if (gSshLib == NULL) {
        SshLibT *lib;
        lib = (SshLibT *)calloc(1, sizeof(*lib));
        if (lib == NULL) {
            pthread_mutex_unlock(&gSshLibMutex);
            return calogErrOomE;
        }
        lib->handles = calogHandleTableCreate();
        if (lib->handles == NULL) {
            free(lib);
            pthread_mutex_unlock(&gSshLibMutex);
            return calogErrOomE;
        }
        // libssh2_init uses global state and is not thread safe; the lib mutex serialises it.
        if (libssh2_init(0) != 0) {
            calogHandleTableDestroy(lib->handles, NULL);
            free(lib);
            pthread_mutex_unlock(&gSshLibMutex);
            return calogErrUnsupportedE;
        }
        gSshLib = lib;
    }
    gSshLib->refCount++;
    pthread_mutex_unlock(&gSshLibMutex);
    calogRegisterInline(calog, "sshConnect", sshConnect, gSshLib);
    calogRegisterInline(calog, "sshAuthPassword", sshAuthPassword, gSshLib);
    calogRegisterInline(calog, "sshAuthKey", sshAuthKey, gSshLib);
    calogRegisterInline(calog, "sshExec", sshExec, gSshLib);
    calogRegisterInline(calog, "sshClose", sshClose, gSshLib);
    calogRegisterInline(calog, "sftpGet", sftpGet, gSshLib);
    calogRegisterInline(calog, "sftpPut", sftpPut, gSshLib);
    calogRegisterInline(calog, "sftpList", sftpList, gSshLib);
    calogRegisterInline(calog, "sftpStat", sftpStat, gSshLib);
    calogRegisterInline(calog, "sftpRemove", sftpRemove, gSshLib);
    calogRegisterInline(calog, "sftpMkdir", sftpMkdir, gSshLib);
    return calogOkE;
}


void calogSshShutdown(void) {
    pthread_mutex_lock(&gSshLibMutex);
    if (gSshLib == NULL) {
        pthread_mutex_unlock(&gSshLibMutex);
        return;
    }
    gSshLib->refCount--;
    if (gSshLib->refCount <= 0) {
        calogHandleTableDestroy(gSshLib->handles, sshCloser);
        libssh2_exit();
        free(gSshLib);
        gSshLib = NULL;
    }
    pthread_mutex_unlock(&gSshLibMutex);
}


static int32_t sftpGet(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT             *lib;
    SshConnT            *conn;
    LIBSSH2_SFTP_HANDLE *file;
    char                *buffer;
    int64_t              cap;
    int64_t              length;
    SshStepE             step;
    int32_t              status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpGet expects (handle, remotePath)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    file = libssh2_sftp_open(conn->sftp, args[1].as.s.bytes, LIBSSH2_FXF_READ, 0);
    if (file == NULL) {
        return calogFail(result, calogErrNotFoundE, "sftpGet: could not open the remote file");
    }
    buffer = NULL;
    cap    = 0;
    length = 0;
    for (;;) {
        status = sshDrainStep(sshReadSftpFile, file, 0, &buffer, &cap, &length, &step);
        if (status != calogOkE) {
            free(buffer);
            libssh2_sftp_close(file);
            if (status == calogErrRangeE) {
                return calogFail(result, status, "sftpGet: file exceeds the maximum transfer size");
            }
            if (status == calogErrOomE) {
                return calogFail(result, status, "sftpGet: out of memory");
            }
            return calogFail(result, status, "sftpGet: read failed");
        }
        if (step == sshStepEofE) {
            break;
        }
    }
    libssh2_sftp_close(file);
    status = calogValueString(result, buffer, (int64_t)length);
    free(buffer);
    return status;
}


static int32_t sftpList(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT             *lib;
    SshConnT            *conn;
    LIBSSH2_SFTP_HANDLE *dir;
    CalogAggT           *list;
    int32_t              status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpList expects (handle, path)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    dir = libssh2_sftp_opendir(conn->sftp, args[1].as.s.bytes);
    if (dir == NULL) {
        return calogFail(result, calogErrArgE, "sftpList: could not open the directory");
    }
    status = calogAggCreate(&list, calogListE);
    if (status != calogOkE) {
        libssh2_sftp_closedir(dir);
        return calogFail(result, status, "sftpList: out of memory");
    }
    for (;;) {
        LIBSSH2_SFTP_ATTRIBUTES attrs;
        CalogAggT              *entry;
        CalogValueT             entryValue;
        char                    name[SSH_NAME_MAX];
        int                     rc;

        rc = libssh2_sftp_readdir(dir, name, sizeof(name), &attrs);
        if (rc == 0) {
            break;
        }
        if (rc < 0) {
            calogAggFree(list);
            libssh2_sftp_closedir(dir);
            return calogFail(result, calogErrArgE, "sftpList: read failed");
        }
        // Skip "." and ".." so callers see only real entries.
        if ((rc == 1 && name[0] == '.') || (rc == 2 && name[0] == '.' && name[1] == '.')) {
            continue;
        }
        status = calogAggCreate(&entry, calogMapE);
        if (status != calogOkE) {
            calogAggFree(list);
            libssh2_sftp_closedir(dir);
            return calogFail(result, status, "sftpList: out of memory");
        }
        status = calogMapSetStr(entry, "name", name, (int64_t)rc);
        if (status == calogOkE) {
            int64_t size;
            size   = (attrs.flags & LIBSSH2_SFTP_ATTR_SIZE) ? (int64_t)attrs.filesize : 0;
            status = calogMapSetInt(entry, "size", size);
        }
        if (status == calogOkE) {
            bool isDir;
            isDir  = (attrs.flags & LIBSSH2_SFTP_ATTR_PERMISSIONS) && LIBSSH2_SFTP_S_ISDIR(attrs.permissions);
            status = calogMapSetBool(entry, "isDir", isDir);
        }
        if (status != calogOkE) {
            calogAggFree(entry);
            calogAggFree(list);
            libssh2_sftp_closedir(dir);
            return calogFail(result, status, "sftpList: failed to build an entry");
        }
        calogValueAgg(&entryValue, entry);
        status = calogAggPush(list, &entryValue);
        if (status != calogOkE) {
            calogValueFree(&entryValue);
            calogAggFree(list);
            libssh2_sftp_closedir(dir);
            return calogFail(result, status, "sftpList: out of memory");
        }
    }
    libssh2_sftp_closedir(dir);
    calogValueAgg(result, list);
    return calogOkE;
}


static int32_t sftpMkdir(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT  *lib;
    SshConnT *conn;
    int32_t   status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpMkdir expects (handle, path)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    if (libssh2_sftp_mkdir(conn->sftp, args[1].as.s.bytes, 0755) != 0) {
        return calogFail(result, calogErrArgE, "sftpMkdir: could not create the directory");
    }
    return calogOkE;
}


static int32_t sftpPut(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT             *lib;
    SshConnT            *conn;
    LIBSSH2_SFTP_HANDLE *file;
    int64_t              total;
    int32_t              status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 3 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpPut expects (handle, remotePath, data)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    file = libssh2_sftp_open(conn->sftp, args[1].as.s.bytes,
                             LIBSSH2_FXF_WRITE | LIBSSH2_FXF_CREAT | LIBSSH2_FXF_TRUNC, 0644);
    if (file == NULL) {
        return calogFail(result, calogErrArgE, "sftpPut: could not open the remote file");
    }
    total = 0;
    while (total < args[2].as.s.length) {
        ssize_t n;
        n = libssh2_sftp_write(file, args[2].as.s.bytes + total, (size_t)(args[2].as.s.length - total));
        if (n < 0) {
            libssh2_sftp_close(file);
            return calogFail(result, calogErrArgE, "sftpPut: write failed");
        }
        total += (int64_t)n;
    }
    libssh2_sftp_close(file);
    return calogOkE;
}


static int32_t sftpRemove(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT  *lib;
    SshConnT *conn;
    int32_t   status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpRemove expects (handle, path)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    if (libssh2_sftp_unlink(conn->sftp, args[1].as.s.bytes) != 0) {
        return calogFail(result, calogErrArgE, "sftpRemove: could not remove the path");
    }
    return calogOkE;
}


static int32_t sftpStat(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT                *lib;
    SshConnT               *conn;
    LIBSSH2_SFTP_ATTRIBUTES attrs;
    CalogAggT              *map;
    int64_t                 size;
    bool                    isDir;
    int                     rc;
    int32_t                 status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sftpStat expects (handle, path)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    status = sshEnsureSftp(conn, result);
    if (status != calogOkE) {
        return status;
    }
    rc = libssh2_sftp_stat(conn->sftp, args[1].as.s.bytes, &attrs);
    if (rc < 0) {
        // A missing path (or any stat failure) reports as nil rather than an error.
        return calogOkE;
    }
    status = calogAggCreate(&map, calogMapE);
    if (status != calogOkE) {
        return calogFail(result, status, "sftpStat: out of memory");
    }
    size  = (attrs.flags & LIBSSH2_SFTP_ATTR_SIZE) ? (int64_t)attrs.filesize : 0;
    isDir = (attrs.flags & LIBSSH2_SFTP_ATTR_PERMISSIONS) && LIBSSH2_SFTP_S_ISDIR(attrs.permissions);
    status = calogMapSetInt(map, "size", size);
    if (status == calogOkE) {
        status = calogMapSetBool(map, "isDir", isDir);
    }
    if (status != calogOkE) {
        calogAggFree(map);
        return calogFail(result, status, "sftpStat: failed to build the result");
    }
    calogValueAgg(result, map);
    return calogOkE;
}


static int32_t sshAuthKey(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT    *lib;
    SshConnT   *conn;
    const char *publicKey;
    const char *passphrase;
    int         rc;
    int32_t     status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount < 3 || argCount > 5 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sshAuthKey expects (handle, user, privateKeyPath[, publicKeyPath, passphrase])");
    }
    publicKey  = NULL;
    passphrase = NULL;
    if (argCount >= 4) {
        if (args[3].type != calogStringE) {
            return calogFail(result, calogErrArgE, "sshAuthKey: publicKeyPath must be a string");
        }
        if (args[3].as.s.length > 0) {
            publicKey = args[3].as.s.bytes;
        }
    }
    if (argCount == 5) {
        if (args[4].type != calogStringE) {
            return calogFail(result, calogErrArgE, "sshAuthKey: passphrase must be a string");
        }
        if (args[4].as.s.length > 0) {
            passphrase = args[4].as.s.bytes;
        }
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    rc = libssh2_userauth_publickey_fromfile_ex(conn->session, args[1].as.s.bytes, (unsigned int)args[1].as.s.length,
                                                publicKey, args[2].as.s.bytes, passphrase);
    calogValueBool(result, rc == 0);
    return calogOkE;
}


static int32_t sshAuthPassword(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT  *lib;
    SshConnT *conn;
    int       rc;
    int32_t   status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 3 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sshAuthPassword expects (handle, user, password)");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    rc = libssh2_userauth_password_ex(conn->session, args[1].as.s.bytes, (unsigned int)args[1].as.s.length,
                                      args[2].as.s.bytes, (unsigned int)args[2].as.s.length, NULL);
    calogValueBool(result, rc == 0);
    return calogOkE;
}


static int32_t sshClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT  *lib;
    SshConnT *conn;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 1 || args[0].type != calogIntE) {
        return calogFail(result, calogErrArgE, "sshClose expects (handle)");
    }
    // Atomically claim the handle -- a concurrent sshClose of the same handle gets NULL and must
    // not double-destroy. ssh handles are shareable across contexts (no owner check); the caller
    // is responsible for ensuring nobody is mid-operation on the connection when it is closed
    // (see the threading notes in calogSsh.h).
    conn = (SshConnT *)calogHandleRemove(lib->handles, args[0].as.i, SSH_TYPE_CONN);
    if (conn == NULL) {
        return calogFail(result, calogErrNotFoundE, "sshClose: invalid ssh handle");
    }
    sshConnDestroy(conn, "calog: closing");
    return calogOkE;
}


static void sshCloser(uint32_t type, void *resource) {
    (void)type;
    sshConnDestroy((SshConnT *)resource, "calog: shutdown");
}


// Tear down a connection handle: shut down any cached SFTP subsystem, then the session and
// socket, then free the handle itself. Shared by sshClose and the handle table's closer so
// the release order lives in exactly one place.
static void sshConnDestroy(SshConnT *conn, const char *reason) {
    if (conn->sftp != NULL) {
        libssh2_sftp_shutdown(conn->sftp);
    }
    sshSessionDestroy(conn->session, conn->sock, reason);
    free(conn);
}


static int32_t sshConnect(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT         *lib;
    SshConnT        *conn;
    LIBSSH2_SESSION *session;
    struct addrinfo  hints;
    struct addrinfo *res;
    struct addrinfo *rp;
    char             portBuffer[8];
    int64_t          port;
    int64_t          handle;
    int              fd;
    int              rc;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount < 1 || argCount > 2 || args[0].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sshConnect expects (host[, port])");
    }
    port = 22;
    if (argCount == 2) {
        if (args[1].type != calogIntE) {
            return calogFail(result, calogErrArgE, "sshConnect expects (host[, port])");
        }
        port = args[1].as.i;
    }
    if (port < 1 || port > SSH_PORT_MAX) {
        return calogFail(result, calogErrArgE, "sshConnect: port out of range");
    }
    memset(&hints, 0, sizeof(hints));
    hints.ai_family   = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    snprintf(portBuffer, sizeof(portBuffer), "%u", (unsigned int)port);
    rc = getaddrinfo(args[0].as.s.bytes, portBuffer, &hints, &res);
    if (rc != 0) {
        return calogFail(result, calogErrArgE, gai_strerror(rc));
    }
    fd = -1;
    for (rp = res; rp != NULL; rp = rp->ai_next) {
        fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if (fd < 0) {
            continue;
        }
        if (connect(fd, rp->ai_addr, rp->ai_addrlen) == 0) {
            break;
        }
        close(fd);
        fd = -1;
    }
    freeaddrinfo(res);
    if (fd < 0) {
        return calogFail(result, calogErrArgE, "sshConnect: could not connect");
    }
    session = libssh2_session_init();
    if (session == NULL) {
        close(fd);
        return calogFail(result, calogErrOomE, "sshConnect: could not create a session");
    }
    libssh2_session_set_blocking(session, 1);
    if (libssh2_session_handshake(session, fd) != 0) {
        libssh2_session_free(session);
        close(fd);
        return calogFail(result, calogErrArgE, "sshConnect: SSH handshake failed");
    }
    conn = (SshConnT *)calloc(1, sizeof(*conn));
    if (conn == NULL) {
        sshSessionDestroy(session, fd, "out of memory");
        return calogFail(result, calogErrOomE, "sshConnect: out of memory");
    }
    conn->sock    = fd;
    conn->session = session;
    conn->sftp    = NULL;
    handle = calogHandleAdd(lib->handles, SSH_TYPE_CONN, conn);
    if (handle == 0) {
        sshConnDestroy(conn, "out of memory");
        return calogFail(result, calogErrOomE, "sshConnect: out of memory");
    }
    calogValueInt(result, handle);
    return calogOkE;
}


// Attempt one read of *bytes/*cap/*length via readFn, growing the buffer first if it is
// full. At the SSH_MAX_TRANSFER cap, instead of failing outright, probes with a single-byte
// read so a stream of exactly SSH_MAX_TRANSFER bytes still succeeds -- only a stream with
// MORE than the cap fails (calogErrRangeE). Shared by sftpGet (a single SFTP file stream)
// and sshExecDrain (two interleaved channel streams). Never frees *bytes on failure -- that
// stays the caller's job, since callers may be juggling more than one buffer.
static int32_t sshDrainStep(SshReadFnT readFn, void *handle, int streamId, char **bytes, int64_t *cap, int64_t *length, SshStepE *step) {
    void   *grown;
    char    probe[1];
    ssize_t n;
    int32_t status;

    if (*length == *cap) {
        if (*cap >= SSH_MAX_TRANSFER) {
            n = readFn(handle, streamId, probe, sizeof(probe));
            if (n == 0) {
                *step = sshStepEofE;
                return calogOkE;
            }
            if (n == LIBSSH2_ERROR_EAGAIN) {
                *step = sshStepAgainE;
                return calogOkE;
            }
            if (n < 0) {
                return calogErrArgE;
            }
            return calogErrRangeE;
        }
        grown  = *bytes;
        status = calogGrow(&grown, cap, *length + 1, 1);
        if (status != calogOkE) {
            return status;
        }
        *bytes = (char *)grown;
        if (*cap > SSH_MAX_TRANSFER) {
            *cap = SSH_MAX_TRANSFER;
        }
    }
    n = readFn(handle, streamId, *bytes + *length, (size_t)(*cap - *length));
    if (n > 0) {
        *length += n;
        *step    = sshStepDataE;
    } else if (n == 0) {
        *step = sshStepEofE;
    } else if (n == LIBSSH2_ERROR_EAGAIN) {
        *step = sshStepAgainE;
    } else {
        return calogErrArgE;
    }
    return calogOkE;
}


// Open the SFTP subsystem on the connection if not already open, caching it for reuse.
static int32_t sshEnsureSftp(SshConnT *conn, CalogValueT *result) {
    LIBSSH2_SFTP *sftp;

    if (conn->sftp != NULL) {
        return calogOkE;
    }
    sftp = libssh2_sftp_init(conn->session);
    if (sftp == NULL) {
        return calogFail(result, calogErrArgE, "could not start an SFTP session");
    }
    conn->sftp = sftp;
    return calogOkE;
}


static int32_t sshExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    SshLibT         *lib;
    SshConnT        *conn;
    LIBSSH2_CHANNEL *channel;
    CalogAggT       *map;
    char            *outBytes;
    char            *errBytes;
    int64_t          outLength;
    int64_t          errLength;
    int              exitCode;
    int32_t          status;

    lib = (SshLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "sshExec expects (handle, command)");
    }
    if (args[1].as.s.length > SSH_MAX_TRANSFER) {
        return calogFail(result, calogErrRangeE, "sshExec: command too long");
    }
    status = sshResolve(lib, args[0].as.i, result, &conn);
    if (status != calogOkE) {
        return status;
    }
    channel = libssh2_channel_open_session(conn->session);
    if (channel == NULL) {
        return calogFail(result, calogErrArgE, "sshExec: could not open a channel");
    }
    // Not libssh2_channel_exec (it derives the length with strlen, truncating a command with an
    // embedded NUL); pass the binary-safe length explicitly.
    if (libssh2_channel_process_startup(channel, "exec", 4u, args[1].as.s.bytes, (unsigned int)args[1].as.s.length) != 0) {
        libssh2_channel_free(channel);
        return calogFail(result, calogErrArgE, "sshExec: could not start the command");
    }
    status = sshExecDrain(conn->sock, conn->session, channel, &outBytes, &outLength, &errBytes, &errLength);
    if (status != calogOkE) {
        libssh2_channel_free(channel);
        if (status == calogErrRangeE) {
            return calogFail(result, status, "sshExec: command output exceeds the maximum transfer size");
        }
        if (status == calogErrOomE) {
            return calogFail(result, status, "sshExec: out of memory");
        }
        return calogFail(result, status, "sshExec: channel read failed");
    }
    libssh2_channel_close(channel);
    exitCode = libssh2_channel_get_exit_status(channel);
    libssh2_channel_free(channel);
    status = calogAggCreate(&map, calogMapE);
    if (status != calogOkE) {
        free(outBytes);
        free(errBytes);
        return calogFail(result, status, "sshExec: out of memory");
    }
    status = calogMapSetStr(map, "stdout", outBytes, outLength);
    if (status == calogOkE) {
        status = calogMapSetStr(map, "stderr", errBytes, errLength);
    }
    if (status == calogOkE) {
        status = calogMapSetInt(map, "exitCode", (int64_t)exitCode);
    }
    free(outBytes);
    free(errBytes);
    if (status != calogOkE) {
        calogAggFree(map);
        return calogFail(result, status, "sshExec: failed to build the result");
    }
    calogValueAgg(result, map);
    return calogOkE;
}


// Run the remote command's stdout and stderr to completion, interleaving reads of both
// streams in one loop (never draining one to EOF before starting the other). The two streams
// share one flow-control window, so fully draining stdout first while the remote blocks
// writing a flood of stderr can hang forever waiting on stdout, which never arrives because
// the remote process never gets to exit. Drops the session into non-blocking mode for the
// duration (restored before every return) so a stream with nothing ready right now cannot
// block progress on the other; sshWaitSocket parks the calling thread only when neither
// stream can make progress right now. On failure both output buffers are freed here.
static int32_t sshExecDrain(int sock, LIBSSH2_SESSION *session, LIBSSH2_CHANNEL *channel, char **outBytes, int64_t *outLength, char **errBytes, int64_t *errLength) {
    char    *outBuf;
    char    *errBuf;
    int64_t  outCap;
    int64_t  errCap;
    int64_t  outLen;
    int64_t  errLen;
    bool     outDone;
    bool     errDone;
    bool     progressed;
    SshStepE step;
    int32_t  status;

    outBuf  = NULL;
    errBuf  = NULL;
    outCap  = 0;
    errCap  = 0;
    outLen  = 0;
    errLen  = 0;
    outDone = false;
    errDone = false;
    libssh2_session_set_blocking(session, 0);
    while (!outDone || !errDone) {
        progressed = false;
        if (!outDone) {
            status = sshDrainStep(sshReadChannel, channel, 0, &outBuf, &outCap, &outLen, &step);
            if (status != calogOkE) {
                libssh2_session_set_blocking(session, 1);
                free(outBuf);
                free(errBuf);
                return status;
            }
            if (step == sshStepEofE) {
                outDone    = true;
                progressed = true;
            } else if (step == sshStepDataE) {
                progressed = true;
            }
        }
        if (!errDone) {
            status = sshDrainStep(sshReadChannel, channel, SSH_EXTENDED_DATA_STDERR, &errBuf, &errCap, &errLen, &step);
            if (status != calogOkE) {
                libssh2_session_set_blocking(session, 1);
                free(outBuf);
                free(errBuf);
                return status;
            }
            if (step == sshStepEofE) {
                errDone    = true;
                progressed = true;
            } else if (step == sshStepDataE) {
                progressed = true;
            }
        }
        if (!progressed && (!outDone || !errDone)) {
            sshWaitSocket(sock, session);
        }
    }
    libssh2_session_set_blocking(session, 1);
    *outBytes  = outBuf;
    *outLength = outLen;
    *errBytes  = errBuf;
    *errLength = errLen;
    return calogOkE;
}


static ssize_t sshReadChannel(void *handle, int streamId, char *buffer, size_t length) {
    return libssh2_channel_read_ex((LIBSSH2_CHANNEL *)handle, streamId, buffer, length);
}


static ssize_t sshReadSftpFile(void *handle, int streamId, char *buffer, size_t length) {
    (void)streamId;
    return libssh2_sftp_read((LIBSSH2_SFTP_HANDLE *)handle, buffer, length);
}


// Look up a connection handle. Not-found resolves to calogErrNotFoundE. There is NO owner
// check: an ssh handle may be used by any context it is passed to (see the threading notes in
// calogSsh.h) -- the caller must serialize access, since a libssh2 session is not thread-safe.
static int32_t sshResolve(SshLibT *lib, int64_t handle, CalogValueT *result, SshConnT **out) {
    SshConnT *conn;

    conn = (SshConnT *)calogHandleGet(lib->handles, handle, SSH_TYPE_CONN);
    if (conn == NULL) {
        return calogFail(result, calogErrNotFoundE, "invalid ssh handle");
    }
    *out = conn;
    return calogOkE;
}


// Tear down a raw session and its socket (disconnect, free the session, close the fd). Used
// by sshConnDestroy and by sshConnect's post-handshake failure paths, before a SshConnT
// exists to own the session.
static void sshSessionDestroy(LIBSSH2_SESSION *session, int sock, const char *reason) {
    libssh2_session_disconnect(session, reason);
    libssh2_session_free(session);
    close(sock);
}


// Block the calling thread until the socket is ready in whatever direction libssh2 is
// currently waiting on. Used only by sshExecDrain's non-blocking interleave; every other
// native in this file runs its session in blocking mode and never calls this.
static void sshWaitSocket(int sock, LIBSSH2_SESSION *session) {
    fd_set  fdSet;
    fd_set *readSet;
    fd_set *writeSet;
    int     direction;

    FD_ZERO(&fdSet);
    FD_SET(sock, &fdSet);
    readSet   = NULL;
    writeSet  = NULL;
    direction = libssh2_session_block_directions(session);
    if (direction & LIBSSH2_SESSION_BLOCK_INBOUND) {
        readSet = &fdSet;
    }
    if (direction & LIBSSH2_SESSION_BLOCK_OUTBOUND) {
        writeSet = &fdSet;
    }
    select(sock + 1, readSet, writeSet, NULL, NULL);
}