// Part of JoeyLib, Copyright (c) 2026 Scott Duensing, under the MIT License.
// See LICENSE.md for license information.
// SPDX-License-Identifier: MIT
//
// serialStreamPeer.c - the host end of a both-directions serial stream gate.
//
//     serialStreamPeer PORT WALL MIN_BYTES
//
// The guest (examples/sermouse) transmits an endless stream and checks an
// endless stream from us, both byte-exact:
//
//     byte k = (k ^ (k >> 8) ^ mask) & 0xFF
//
// with mask GUEST_MASK for the guest's stream and PEER_MASK for ours (the
// same formula and masks as streamByte() in examples/sermouse/sermouse.c).
// A lost or repeated byte throws every byte after it off its expected
// value.
//
// This peer listens (the emulator's null_modem connects out), says so
// ("STREAM listening on ..."), checks the guest's stream in order, and
// answers it: our stream is paced byte for byte on the guest's, never more
// than WINDOW bytes ahead of what has come back. So when the guest pauses
// its stream the line goes quiet both ways, and once the guest has been
// silent for QUIET_SECONDS the counts are recorded as a quiet snapshot
// "sent:received" -- at that moment every byte either side sent has
// arrived, so a harness that reads the guest's own counts while it is
// paused can require them to equal one of these exactly.
//
// Runs until the emulator closes the link or WALL seconds pass, then prints
// one line, STREAM PASS or STREAM FAIL, with the counts and the quiet
// snapshots. PASS means the guest's stream arrived with no bad byte and at
// least MIN_BYTES long. Exit 0 on pass, 1 on fail, 2 on bad arguments.

#define _DEFAULT_SOURCE

#include <fcntl.h>
#include <poll.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

#include "gateArg.h"
#include "gateNet.h"
#include "toolBuf.h"
#include "toolFile.h"

#define TOOL_NAME     "serialStreamPeer"
#define ARG_COUNT     4
#define ARG_PORT      1
#define ARG_WALL      2
#define ARG_MIN_BYTES 3

#define GUEST_MASK    0xA5u
#define PEER_MASK     0x00u
// Our bytes allowed ahead of the guest's, and the guest silence that counts
// as a pause.
#define WINDOW        256u
#define QUIET_SECONDS 0.25
#define POLL_SECONDS  0.05
#define MS_PER_SECOND 1000
#define READ_SIZE     4096u
#define BYTE_MASK     0xFFu
#define BYTE_BITS     8u
#define NO_BAD_INDEX  (-1L)
// A snapshot: a comma, two numbers, a colon, and the end.
#define SNAPSHOT_LEN  48u
#define NO_SNAPSHOTS  "none"
// When to read and when to write: a hang-up or an error is both (the call
// reports it).
#define READABLE      (POLLIN | POLLHUP | POLLERR)
#define WRITABLE      (POLLOUT | POLLERR)

// The run's counts: the guest's bytes and the bad among them (the index of
// the first), ours, the quiet snapshots ("sent:received", comma-separated),
// and whether the emulator closed the link.
typedef struct RunT {
    size_t   received;
    size_t   sent;
    size_t   bad;
    long     firstBad;
    ToolBufT snapshots;
    bool     closed;
} RunT;

static void    run(RunT *r, int conn, double wall, double start);
static uint8_t streamByte(size_t index, uint8_t mask);


// Check the guest's stream and answer it until the link closes or wall
// seconds from start have gone.
static void run(RunT *r, int conn, double wall, double start) {
    struct pollfd watch;
    ToolBufT      chunk;
    uint8_t       data[READ_SIZE];
    char          snapshot[SNAPSHOT_LEN];
    ssize_t       got;
    ssize_t       put;
    size_t        want;
    size_t        i;
    double        now;
    double        lastRx;
    bool          quietArmed;

    memset(&chunk, 0, sizeof(chunk));
    lastRx     = 0.0;
    quietArmed = false;
    fcntl(conn, F_SETFL, fcntl(conn, F_GETFL) | O_NONBLOCK);
    while (gateNetNow() - start < wall) {
        // Our stream starts once the guest's has: it is open and listening.
        want          = r->received > 0u ? r->received + WINDOW - r->sent : 0u;
        watch.fd      = conn;
        watch.events  = (short)(POLLIN | (want > 0u ? POLLOUT : 0));
        watch.revents = 0;
        // An interrupted wait leaves revents 0: nothing ready.
        poll(&watch, 1, (int)(POLL_SECONDS * MS_PER_SECOND));
        now = gateNetNow();
        if ((watch.revents & READABLE) != 0) {
            got = recv(conn, data, sizeof(data), 0);
            if (got <= 0) {
                r->closed = true;
                break;
            }
            for (i = 0; i < (size_t)got; i++) {
                if (data[i] != streamByte(r->received, GUEST_MASK)) {
                    if (r->firstBad == NO_BAD_INDEX) {
                        r->firstBad = (long)r->received;
                    }
                    r->bad++;
                }
                r->received++;
            }
            lastRx     = now;
            quietArmed = true;
        }
        if ((watch.revents & WRITABLE) != 0 && want > 0u) {
            chunk.len = 0u;
            for (i = 0; i < want; i++) {
                toolBufByte(&chunk, streamByte(r->sent + i, PEER_MASK));
            }
            put = send(conn, chunk.data, chunk.len, 0);
            if (put > 0) {
                r->sent += (size_t)put;
            }
        }
        if (quietArmed && now - lastRx >= QUIET_SECONDS) {
            snprintf(snapshot, sizeof(snapshot), "%s%zu:%zu", r->snapshots.len > 0u ? "," : "", r->sent, r->received);
            toolBufAdd(&r->snapshots, snapshot, strlen(snapshot));
            quietArmed = false;
        }
    }
    toolBufFree(&chunk);
}


static uint8_t streamByte(size_t index, uint8_t mask) {
    return (uint8_t)((index ^ (index >> BYTE_BITS) ^ mask) & BYTE_MASK);
}


int main(int argc, char **argv) {
    RunT     r;
    long     minBytes;
    double   wall;
    double   start;
    uint16_t port;
    int      listener;
    int      conn;
    bool     ok;

    gateNetStart(TOOL_NAME);
    if (argc != ARG_COUNT) {
        fprintf(stderr, "usage: %s PORT WALL MIN_BYTES\n", TOOL_NAME);
        return GATE_USAGE_EXIT;
    }
    port     = gateArgPort(argv[ARG_PORT]);
    wall     = gateArgDouble("wall", argv[ARG_WALL]);
    minBytes = gateArgLong("minimum bytes", argv[ARG_MIN_BYTES]);
    start    = gateNetNow();

    listener = gateNetListen(port, GATE_NET_ONE_PEER);
    printf("STREAM listening on %s:%u\n", GATE_NET_HOST, (unsigned)port);
    conn = gateNetAccept(listener, wall, NULL);
    close(listener);
    if (conn < 0) {
        printf("STREAM FAIL: the emulator never connected\n");
        return 1;
    }
    memset(&r, 0, sizeof(r));
    r.firstBad = NO_BAD_INDEX;
    run(&r, conn, wall, start);
    close(conn);

    ok = r.bad == 0u && (long)r.received >= minBytes;
    if (r.snapshots.len == 0u) {
        toolBufAdd(&r.snapshots, NO_SNAPSHOTS, strlen(NO_SNAPSHOTS));
    }
    printf("STREAM %s received=%zu sent=%zu bad=%zu firstBad=%ld quiet=%.*s end=%s\n", ok ? "PASS" : "FAIL", r.received, r.sent, r.bad, r.firstBad, (int)r.snapshots.len, (const char *)r.snapshots.data, r.closed ? "closed" : "wall");
    return ok ? 0 : 1;
}