// 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; each connection handle is owner-scoped to the context that created
// it (see SshConnT.ownerId).

#define _GNU_SOURCE

#include "calogSsh.h"

#include "calogHandle.h"

#include <netdb.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.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.
#define SSH_MAX_TRANSFER (64 * 1024 * 1024)

// What a connection handle owns: the socket, the libssh2 session, a lazily-opened+cached SFTP
// subsystem, and the id of the context that created it. ONLY that owner may operate on it.
typedef struct SshConnT {
    int             sock;
    LIBSSH2_SESSION *session;
    LIBSSH2_SFTP    *sftp;
    uint64_t        ownerId;
} 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;

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 int32_t sshConnect(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshDrain(LIBSSH2_CHANNEL *channel, int streamId, char **bytesOut, int64_t *lengthOut);
static int32_t sshEnsureSftp(SshConnT *conn, CalogValueT *result);
static int32_t sshExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshMapSetBool(CalogAggT *map, const char *key, bool value);
static int32_t sshMapSetInt(CalogAggT *map, const char *key, int64_t value);
static int32_t sshMapSetStr(CalogAggT *map, const char *key, const char *bytes, int64_t length);
static int32_t sshResolve(SshLibT *lib, int64_t handle, CalogValueT *result, SshConnT **out);


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;
    size_t               cap;
    size_t               length;
    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 (;;) {
        ssize_t n;
        if (length == cap) {
            size_t want;
            char  *grown;
            if (cap >= SSH_MAX_TRANSFER) {
                free(buffer);
                libssh2_sftp_close(file);
                return calogFail(result, calogErrRangeE, "sftpGet: file exceeds the maximum transfer size");
            }
            want = (cap == 0) ? 65536 : cap * CALOG_GROWTH_FACTOR;
            if (want > SSH_MAX_TRANSFER) {
                want = SSH_MAX_TRANSFER;
            }
            grown = (char *)realloc(buffer, want);
            if (grown == NULL) {
                free(buffer);
                libssh2_sftp_close(file);
                return calogFail(result, calogErrOomE, "sftpGet: out of memory");
            }
            buffer = grown;
            cap    = want;
        }
        n = libssh2_sftp_read(file, buffer + length, cap - length);
        if (n > 0) {
            length += (size_t)n;
        } else if (n == 0) {
            break;
        } else if (n == LIBSSH2_ERROR_EAGAIN) {
            continue;
        } else {
            free(buffer);
            libssh2_sftp_close(file);
            return calogFail(result, calogErrArgE, "sftpGet: read failed");
        }
    }
    libssh2_sftp_close(file);
    status = calogValueString(result, (buffer != NULL) ? 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[512];
        int                     rc;

        rc = libssh2_sftp_readdir(dir, name, sizeof(name), &attrs);
        if (rc == 0) {
            break;
        }
        if (rc == LIBSSH2_ERROR_EAGAIN) {
            continue;
        }
        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 = sshMapSetStr(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 = sshMapSetInt(entry, "size", size);
        }
        if (status == calogOkE) {
            bool isDir;
            isDir  = (attrs.flags & LIBSSH2_SFTP_ATTR_PERMISSIONS) && LIBSSH2_SFTP_S_ISDIR(attrs.permissions);
            status = sshMapSetBool(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) {
            if (n == LIBSSH2_ERROR_EAGAIN) {
                continue;
            }
            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 = sshMapSetInt(map, "size", size);
    if (status == calogOkE) {
        status = sshMapSetBool(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)");
    }
    // Peek first to enforce owner-scoping; because the owner is a single thread, the peek and
    // the remove below cannot race another context claiming this handle.
    conn = (SshConnT *)calogHandleGet(lib->handles, args[0].as.i, SSH_TYPE_CONN);
    if (conn == NULL) {
        return calogFail(result, calogErrNotFoundE, "sshClose: invalid ssh handle");
    }
    if (conn->ownerId != calogCurrentId()) {
        return calogFail(result, calogErrArgE, "sshClose: ssh handle belongs to another context");
    }
    conn = (SshConnT *)calogHandleRemove(lib->handles, args[0].as.i, SSH_TYPE_CONN);
    if (conn == NULL) {
        return calogFail(result, calogErrNotFoundE, "sshClose: invalid ssh handle");
    }
    if (conn->sftp != NULL) {
        libssh2_sftp_shutdown(conn->sftp);
    }
    libssh2_session_disconnect(conn->session, "calog: closing");
    libssh2_session_free(conn->session);
    close(conn->sock);
    free(conn);
    return calogOkE;
}


static void sshCloser(uint32_t type, void *resource) {
    SshConnT *conn;

    (void)type;
    conn = (SshConnT *)resource;
    if (conn->sftp != NULL) {
        libssh2_sftp_shutdown(conn->sftp);
    }
    libssh2_session_disconnect(conn->session, "calog: shutdown");
    libssh2_session_free(conn->session);
    close(conn->sock);
    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) {
        libssh2_session_disconnect(session, "out of memory");
        libssh2_session_free(session);
        close(fd);
        return calogFail(result, calogErrOomE, "sshConnect: out of memory");
    }
    conn->sock    = fd;
    conn->session = session;
    conn->sftp    = NULL;
    conn->ownerId = calogCurrentId();
    handle = calogHandleAdd(lib->handles, SSH_TYPE_CONN, conn);
    if (handle == 0) {
        libssh2_session_disconnect(session, "out of memory");
        libssh2_session_free(session);
        close(fd);
        free(conn);
        return calogFail(result, calogErrOomE, "sshConnect: out of memory");
    }
    calogValueInt(result, handle);
    return calogOkE;
}


// Drain one channel stream (0 = stdout, SSH_EXTENDED_DATA_STDERR = stderr) into a fresh
// growable buffer, binary-safe. Returns calogOkE with *bytesOut owned by the caller (freed on
// every path), or an error (the buffer is released here on failure).
static int32_t sshDrain(LIBSSH2_CHANNEL *channel, int streamId, char **bytesOut, int64_t *lengthOut) {
    char   *buffer;
    size_t  cap;
    size_t  length;

    buffer = NULL;
    cap    = 0;
    length = 0;
    for (;;) {
        ssize_t n;
        if (length == cap) {
            size_t want;
            char  *grown;
            if (cap >= SSH_MAX_TRANSFER) {
                free(buffer);
                return calogErrRangeE;
            }
            want = (cap == 0) ? 4096 : cap * CALOG_GROWTH_FACTOR;
            if (want > SSH_MAX_TRANSFER) {
                want = SSH_MAX_TRANSFER;
            }
            grown = (char *)realloc(buffer, want);
            if (grown == NULL) {
                free(buffer);
                return calogErrOomE;
            }
            buffer = grown;
            cap    = want;
        }
        n = libssh2_channel_read_ex(channel, streamId, buffer + length, cap - length);
        if (n > 0) {
            length += (size_t)n;
        } else if (n == 0) {
            break;
        } else if (n == LIBSSH2_ERROR_EAGAIN) {
            continue;
        } else {
            free(buffer);
            return calogErrArgE;
        }
    }
    *bytesOut  = buffer;
    *lengthOut = (int64_t)length;
    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)");
    }
    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");
    }
    if (args[1].as.s.length > SSH_MAX_TRANSFER) {
        libssh2_channel_free(channel);
        return calogFail(result, calogErrRangeE, "sshExec: command too long");
    }
    // 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");
    }
    // Blocking reads: drain stdout to EOF, then stderr (libssh2 buffers the other stream).
    status = sshDrain(channel, 0, &outBytes, &outLength);
    if (status != calogOkE) {
        libssh2_channel_free(channel);
        return calogFail(result, status, "sshExec: out of memory");
    }
    status = sshDrain(channel, SSH_EXTENDED_DATA_STDERR, &errBytes, &errLength);
    if (status != calogOkE) {
        free(outBytes);
        libssh2_channel_free(channel);
        return calogFail(result, status, "sshExec: out of memory");
    }
    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 = sshMapSetStr(map, "stdout", outBytes, outLength);
    if (status == calogOkE) {
        status = sshMapSetStr(map, "stderr", errBytes, errLength);
    }
    if (status == calogOkE) {
        status = sshMapSetInt(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;
}


// Set map[key] = boolean. On failure the (freshly built) key is released; the scalar needs
// none.
static int32_t sshMapSetBool(CalogAggT *map, const char *key, bool value) {
    CalogValueT keyValue;
    CalogValueT boolValue;
    int32_t     status;

    status = calogValueString(&keyValue, key, (int64_t)strlen(key));
    if (status != calogOkE) {
        return status;
    }
    calogValueBool(&boolValue, value);
    status = calogAggSet(map, &keyValue, &boolValue);
    if (status != calogOkE) {
        calogValueFree(&keyValue);
    }
    return status;
}


// Set map[key] = integer. On failure the (freshly built) key is released; the scalar needs
// none.
static int32_t sshMapSetInt(CalogAggT *map, const char *key, int64_t value) {
    CalogValueT keyValue;
    CalogValueT intValue;
    int32_t     status;

    status = calogValueString(&keyValue, key, (int64_t)strlen(key));
    if (status != calogOkE) {
        return status;
    }
    calogValueInt(&intValue, value);
    status = calogAggSet(map, &keyValue, &intValue);
    if (status != calogOkE) {
        calogValueFree(&keyValue);
    }
    return status;
}


// Set map[key] = binary-safe string. On failure any built values are released.
static int32_t sshMapSetStr(CalogAggT *map, const char *key, const char *bytes, int64_t length) {
    CalogValueT keyValue;
    CalogValueT stringValue;
    int32_t     status;

    status = calogValueString(&keyValue, key, (int64_t)strlen(key));
    if (status != calogOkE) {
        return status;
    }
    status = calogValueString(&stringValue, bytes, length);
    if (status != calogOkE) {
        calogValueFree(&keyValue);
        return status;
    }
    status = calogAggSet(map, &keyValue, &stringValue);
    if (status != calogOkE) {
        calogValueFree(&keyValue);
        calogValueFree(&stringValue);
    }
    return status;
}


// Look up a connection handle and enforce owner-scoping. Not-found resolves to
// calogErrNotFoundE; a handle owned by another context resolves to calogErrArgE.
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");
    }
    if (conn->ownerId != calogCurrentId()) {
        return calogFail(result, calogErrArgE, "ssh handle belongs to another context");
    }
    *out = conn;
    return calogOkE;
}