path: root/broadd.c

/*
-- broadd - tcp/udp broadcast server
------ env vars:
---------- MAX_CLIENTS    -> maximum number of hypothetical connections
---------- BUFS           -> buffer size, values >MTU are not recommended on udp
---------- RATE_LIMIT_S   -> rate limit expiry window
---------- CLIENT_TIMEOUT -> how long to wait before evicting a udp client
---------- WILDCARD       -> whether or not to enable wildcard. set to 0 or 1.
---------- RELAY_TXT      -> read a pre-determined list of relays and connect to them at startup.
-------------- format is one relay per line, `<ip> <port>`
------ difference:
---------- udp mode does not send m_ok or p_ok
---------- send `N\n` to initiate connection in udp mode
---------- you get kicked after TIMEOUT seconds with no packets in udp mode
---------- send an `N\n` packet before TIMEOUT seconds are elapsed, to prevent getting your id reaped
------ protocol:
---------- all server replies except M for incoming messages are suffixed with "\n"
----------
---------- connect
-------------- if in udp mode, server sends `T<timeout_sec>\n` to tell you when timeout will occur.
-------------- server sends `IT<id>` or `IU<id>`, depending on whether it's tcp or udp
-------------- this is your id
-------------- if server is full, it instead sends "full"
---------- send
-------------- `P<8-byte password>`
------------------ server will only broadcast to you messages starting with that password
------------------ the password will be stripped before it is sent to you
-------------- `M<BUFS_max-byte buffer>`
------------------ broadcast message to all clients
------------------ only clients with `password==first_8_bytes` receive your msg
------------------ message shorter than 8 returns "Err"
------------------ default password is "00000000" ASCII
-------------- `N<whatever>`
------------------ does nothing. use this for initiating your UDP connection.
-------------- `W`
------------------ returns `W0` or `W1`, depending on whether the server has wildcard enabled or not.
---------- receive
---------------- when you send P for password
-------------------- server sends back "p_ok"
---------------- when you send M or S for message
-------------------- server sends back "m_ok"
---------------- when server is shutting down
-------------------- server sends "bye"
-------------------- close the socket when you get this msg
---------------- when an msg is broadcasted
-------------------- M<id>:{T<timestamp>}<bytes>
---------------- when admin sends you a msg
-------------------- server sends `A<BUFS-1_max-byte buffer>`
-------------------- this message is sent to all, regardless of password
------ admin panel:
-------- type the commands into server stdin to use
-------- spaces are omitted before/after args
-------- `E` -> closes the server
-------- `C <target-ip> <target-port>` -> connects to another server, relaying all its messages to the current server's clients
-------- `B <msg>` -> broadcasts message to all currently connected clients
*/

// ======== change these ========
#if 1
    // max buffer size
    #define UDP_BUFS 1024
    #define TCP_BUFS 8192

    // rate limit: BUFS bytes / RLM seconds
    #define RATE_LIMIT_S 1

    // max clients count
    #define MAX_CLIENTS 8192

    // client timeout for udp mode in seconds
    #define CLIENT_TIMEOUT 30

    // max age for timestamped message
    #define DROP_OLD 5
#endif
// ==============================

#if UDP == 1
    #define BUFS UDP_BUFS
#else
    #define BUFS TCP_BUFS
#endif

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <netdb.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/uio.h>
#include <time.h>
#include <unistd.h>

#define p_stringify(...) #__VA_ARGS__
#define stringify(...)   p_stringify(__VA_ARGS__)

#if DEBUG == 1
    #define log(...)  printf(__VA_ARGS__)
    #define verb(...) __VA_ARGS__
#else
    #define log(...)
    #define verb(...)
#endif

#if UDP == 1
    #define onlyudp(...) __VA_ARGS__
    #define onlytcp(...)
#else
    #define onlyudp(...)
    #define onlytcp(...) __VA_ARGS__
#endif

// glob consts using env vars
int env_max_clients  = MAX_CLIENTS;
int env_bufs         = BUFS;
int env_rate_limit_s = RATE_LIMIT_S;
int env_wildcard     = 1;
onlyudp(int env_client_timeout = CLIENT_TIMEOUT);

#define DEDUP_SIZE 65536
#define DEDUP_MASK (DEDUP_SIZE - 1)

#define DEDUP_WINDOW 5

typedef struct {
    uint32_t hash;
    time_t   seen_at;
} dedup_entry_t;

dedup_entry_t dedup_table[ DEDUP_SIZE ];

uint32_t hash_data(char* data, int len) {
    uint32_t hash = 2166136261u;
    for (int i = 0; i < len; i++) {
        hash ^= (uint8_t) data[ i ];
        hash *= 16777619;
    }
    return hash;
}

int is_duplicate(char* payload, int len) {
    if (len <= 0) return 0;

    uint32_t h   = hash_data(payload, len);
    uint32_t idx = h & DEDUP_MASK;
    time_t   now = time(NULL);

    if (dedup_table[ idx ].hash == h) {
        if (now - dedup_table[ idx ].seen_at < DEDUP_WINDOW) { return 1; }
    }

    dedup_table[ idx ].hash    = h;
    dedup_table[ idx ].seen_at = now;

    return 0;
}

#define MAX_RELAYS 64

struct relay {
    int                     fd;
    int                     is_udp;
    time_t                  last_hb; // udp: last heartbeat sent
    int                     remote_timeout;
    struct sockaddr_storage addr;    // udp
    socklen_t               addrlen; // udp
};

struct relay relays[ MAX_RELAYS ];
int          relay_count = 0;

// stdin nonblocking
#define NB_EOF   -1
#define NB_AGAIN -2

static char   nb_buffer[ BUFSIZ ];
static size_t nb_buf_pos  = 0;
static size_t nb_buf_fill = 0;

#define fold inline __attribute__((always_inline)) static

fold void set_stdin_nonblocking(void) {
    static int done = 0;
    if (!done) {
        int flags = fcntl(STDIN_FILENO, F_GETFL);
        if (flags != -1) { fcntl(STDIN_FILENO, F_SETFL, flags | O_NONBLOCK); }
        done = 1;
    }
}

fold int nb_getstring(char str[], size_t max_size) {
    if (max_size <= 1) {
        if (max_size == 1) str[ 0 ] = '\0';
        return 0;
    }

    size_t pos = 0;

    // drain buffer
    size_t remaining = max_size - 1;
    size_t buf_avail = nb_buf_fill - nb_buf_pos;
    size_t copy_len  = buf_avail < remaining ? buf_avail : remaining;
    if (copy_len > 0) {
        memcpy(str + pos, nb_buffer + nb_buf_pos, copy_len);
        nb_buf_pos += copy_len;
        pos += copy_len;
    }

    if (pos < max_size - 1 && nb_buf_pos >= nb_buf_fill) {
        ssize_t n;
        do { n = read(STDIN_FILENO, nb_buffer, sizeof(nb_buffer)); } while (n == -1 && errno == EINTR);

        if (n <= 0) {
            if (n == 0) {
                if (pos == 0) return NB_EOF;
            } else { // n == -1
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    if (pos == 0) return NB_AGAIN;
                } else {
                    if (pos == 0) return NB_EOF;
                }
            }
        } else {
            nb_buf_pos  = 0;
            nb_buf_fill = (size_t) n;

            remaining = max_size - 1 - pos;
            buf_avail = nb_buf_fill - nb_buf_pos;
            copy_len  = buf_avail < remaining ? buf_avail : remaining;
            if (copy_len > 0) {
                memcpy(str + pos, nb_buffer + nb_buf_pos, copy_len);
                nb_buf_pos += copy_len;
                pos += copy_len;
            }
        }
    }

    str[ pos ] = '\0';
    return (int) pos;
}

#define likely(x)   __builtin_expect(!!(x), 1)
#define unlikely(x) __builtin_expect(!!(x), 0)

uint64_t default_pass;  // default password for when a client joins.
uint64_t wildcard_pass; // a client with this password will receive all messages with any password.

onlytcp(struct client {
    int      fd;       // descriptor
    uint64_t id;       // identifier
    uint64_t pass;     // password
    time_t   last_rl;  // ratelimit timestamp
    int      rl_bytes; // bytes used
});
onlyudp(struct client {
    struct sockaddr_storage addr;      // client address
    socklen_t               addrlen;   // address len
    uint64_t                id;        // identifier
    uint64_t                pass;      // password
    time_t                  last_rl;   // ratelimit timestamp
    int                     rl_bytes;  // bytes used
    time_t                  last_seen; // for kicking
});
struct client* clients = NULL;
int            active  = 0;
int            cap     = 0;
uint64_t       nid     = 2; // next id to assign

// accept new client
onlytcp(void add_client(int fd) {
    if (active >= env_max_clients) {
        // refuse: too many clients
        write(fd, "full\n", 5);
        close(fd);
        return;
    }
    if (active >= cap) {
        cap = cap ? cap * 2 : 64;
        if (cap > env_max_clients) cap = env_max_clients;
        clients = realloc(clients, cap * sizeof(struct client));
    }
    onlytcp(clients[ active ].fd = fd);
    clients[ active ].id       = nid++;
    clients[ active ].pass     = default_pass;
    clients[ active ].last_rl  = time(NULL);
    clients[ active ].rl_bytes = 0;
    char buf[ 64 ];
    int  len = sprintf(buf, "IT%" PRIu64 "\n", clients[ active ].id);
    write(fd, buf, len);
    active++;
    log("clients:%d\n", active);
});

// disconnect client at index i
void del_client(int i) {
    onlytcp(close(clients[ i ].fd));
    if (i < active - 1) clients[ i ] = clients[ active - 1 ];
    active--;
    log("clients:%d\n", active);
}

int sock;

// broadcast to all clients whose password matches
void broadcast_msg(uint64_t sender_id, char* payload, int msg_len) {
    if (is_duplicate(payload, msg_len)) return;

    uint64_t p;

    // separate
    if (msg_len < 8) return;
    memcpy(&p, payload, 8);
    char* raw_body     = payload + 8;
    int   raw_body_len = msg_len - 8;

    // sign with timestamp
    char   temp_body_buf[ BUFS + 32 ];
    char*  final_body;
    int    final_body_len;
    time_t now            = time(NULL);
    int    is_timestamped = 0;

    // dont dupe
    if (raw_body_len >= 13 && strncmp(raw_body, "{T", 2) == 0 && raw_body[ 12 ] == '}') {
        // get time
        char t_str[ 11 ];
        memcpy(t_str, raw_body + 2, 10);
        t_str[ 10 ]   = '\0';
        long msg_time = atol(t_str);

        // check
        if (now - msg_time > DROP_OLD) {
            // old
            return;
        }

        // valid
        is_timestamped = 1;
        final_body     = raw_body;
        final_body_len = raw_body_len;
    }

    if (!is_timestamped) {
        // add stamp
        if (raw_body_len + 13 > BUFS) {
            // too big, drop
            return;
        }

        // write
        sprintf(temp_body_buf, "{T%010ld}", now);
        // append body
        memcpy(temp_body_buf + 13, raw_body, raw_body_len);

        final_body     = temp_body_buf;
        final_body_len = raw_body_len + 13;
    }

    int sent = 0;

    // bake normal
    char normal_buf[ 64 + BUFS ];
    int  normal_len = sprintf(normal_buf, "M%" PRIu64 ":", sender_id);
    memcpy(normal_buf + normal_len, final_body, final_body_len);
    normal_len += final_body_len;

    // bake wildcard
    char wild_buf[ 1 + 8 + BUFS ];
    wild_buf[ 0 ] = 'M';
    memcpy(wild_buf + 1, &p, 8);
    memcpy(wild_buf + 9, final_body, final_body_len);
    int wild_len = 1 + 8 + final_body_len;

    // send
    for (int j = 0; j < active; j++) {
        if (clients[ j ].id == sender_id) continue;
        int is_wildcard = env_wildcard && clients[ j ].pass == wildcard_pass;
        if (p != clients[ j ].pass && !is_wildcard) continue;

        char* send_buf = is_wildcard ? wild_buf : normal_buf;
        int   send_len = is_wildcard ? wild_len : normal_len;

        onlytcp(write(clients[ j ].fd, send_buf, send_len));
        onlyudp(sendto(sock, send_buf, send_len, 0, (struct sockaddr*) &clients[ j ].addr, clients[ j ].addrlen));
        sent++;
    }

    if (sent > 0) log("sentto:%d\n", sent);
}

void connect_relay(const char* ip, const char* port) {
    if (relay_count >= MAX_RELAYS) {
        printf("relay(full)\n");
        return;
    }

    // try tcp
    struct addrinfo hint = {}, *result;
    hint.ai_family       = AF_UNSPEC;
    hint.ai_socktype     = SOCK_STREAM;
    if (getaddrinfo(ip, port, &hint, &result) != 0) {
        perror("relay: getaddrinfo failed\n");
        return;
    }

    int  is_udp = 0;
    int  fd     = socket(result->ai_family, result->ai_socktype, result->ai_protocol);
    char gbuf[ 64 ];

    if (fd >= 0 && connect(fd, result->ai_addr, result->ai_addrlen) == 0) {
        // read greeting
        int n = read(fd, gbuf, sizeof(gbuf) - 1);
        if (n > 0 && gbuf[ 0 ] == 'I' && gbuf[ 1 ] == 'T') {
            // confirmed tcp
            char pbuf[ 9 ] = "P";
            memcpy(pbuf + 1, &wildcard_pass, 8);
            write(fd, pbuf, 9);
            read(fd, gbuf, sizeof(gbuf)); // p_ok
        } else {
            // not a broadd tcp server
            close(fd);
            fd = -1;
        }
    } else {
        if (fd >= 0) {
            close(fd);
            fd = -1;
        }
    }
    freeaddrinfo(result);

    // udp fallback
    if (fd < 0) {
        hint.ai_socktype = SOCK_DGRAM;
        if (getaddrinfo(ip, port, &hint, &result) != 0) {
            printf("relay: getaddrinfo failed\n");
            return;
        }
        fd = socket(result->ai_family, result->ai_socktype, result->ai_protocol);
        if (fd < 0) {
            freeaddrinfo(result);
            printf("relay: socket failed\n");
            return;
        }

        // prevent hanging
        struct timeval tv;
        tv.tv_sec  = 2;
        tv.tv_usec = 0;
        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv);

        sendto(fd, "N\n", 2, 0, result->ai_addr, result->ai_addrlen);

        if (connect(fd, result->ai_addr, result->ai_addrlen) < 0) {
            close(fd);
            freeaddrinfo(result);
            printf("relay: udp connect failed\n");
            return;
        }

        struct sockaddr_storage src;
        socklen_t               srclen = sizeof(src);

        int rn = read(fd, gbuf, sizeof(gbuf) - 1);
        if (rn <= 0) {
            printf("relay: udp timeout\n");
            close(fd);
            freeaddrinfo(result);
            return;
        }
        gbuf[ rn ]         = '\0';
        int remote_timeout = (gbuf[ 0 ] == 'T') ? atoi(gbuf + 1) : 30;

        rn = read(fd, gbuf, sizeof(gbuf) - 1);
        if (rn <= 0 || gbuf[ 0 ] != 'I' || gbuf[ 1 ] != 'U') {
            close(fd);
            freeaddrinfo(result);
            printf("relay: invalid handshake\n");
            return;
        }

        char pbuf[ 9 ] = "P";
        memcpy(pbuf + 1, &wildcard_pass, 8);
        sendto(fd, pbuf, 9, 0, result->ai_addr, result->ai_addrlen);
        is_udp = 1;

        struct relay* r   = &relays[ relay_count++ ];
        r->fd             = fd;
        r->is_udp         = 1;
        r->last_hb        = time(NULL);
        r->remote_timeout = remote_timeout;
        memcpy(&r->addr, result->ai_addr, result->ai_addrlen);
        r->addrlen = result->ai_addrlen;
        freeaddrinfo(result);

        tv.tv_sec = 0;
        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv);
    } else {
        struct relay* r   = &relays[ relay_count++ ];
        r->fd             = fd;
        r->is_udp         = 0;
        r->last_hb        = time(NULL);
        r->remote_timeout = 0;
    }

    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
    printf("relay-connect %s:%s (%s)\n", ip, port, is_udp ? "udp" : "tcp");
}

char* admin_msg;

int main(int argc, char** argv) {
    if (argc < 3) {
        printf("more args.\n%s <ip> <port>\n", argv[ 0 ]);
        exit(1);
    }

    onlytcp(printf("hi from tcp;" verb(" verbose mode") "\n"));
    onlyudp(printf("hi from udp;" verb(" verbose mode") "\n"));

    // parse args
    {
        const char* e;
        if ((e = getenv("MAX_CLIENTS"))) {
            int v = atoi(e);
            if (v > 0 && v <= 65536) env_max_clients = v;
            else
                printf("MAX_CLIENTS: ignored (must be 1...65536)\n");
        }
        if ((e = getenv("BUFS"))) {
            int v = atoi(e);
            if (v > 0 && v <= 65536) env_bufs = v;
            else
                printf("BUFS: ignored (must be 1...65536)\n");
        }
        if ((e = getenv("RATE_LIMIT_S"))) {
            int v = atoi(e);
            if (v > 0) env_rate_limit_s = v;
            else
                printf("RATE_LIMIT_S: ignored (must be >0)\n");
        }
        onlyudp(if ((e = getenv("CLIENT_TIMEOUT"))) {
            int v = atoi(e);
            if (v > 0) env_client_timeout = v;
            else
                printf("CLIENT_TIMEOUT: ignored (must be >0)\n");
        });
        if ((e = getenv("WILDCARD"))) {
            int v        = atoi(e);
            env_wildcard = (v != 0);
        }
    }

    admin_msg = malloc(env_bufs);

    set_stdin_nonblocking();

    admin_msg[ 0 ] = 'A';

    memcpy(&default_pass, "00000000", 8);
    memcpy(&wildcard_pass, "00000001", 8);

    struct addrinfo hint = {}, *result;
    hint.ai_family       = AF_UNSPEC;
    onlytcp(hint.ai_socktype = SOCK_STREAM);
    onlyudp(hint.ai_socktype = SOCK_DGRAM);
    hint.ai_flags = AI_PASSIVE;
    if (getaddrinfo(argv[ 1 ], argv[ 2 ], &hint, &result) != 0) {
        perror("getaddrinfo failed");
        exit(1);
    }

    sock = socket(result->ai_family, result->ai_socktype, result->ai_protocol);

    int option = 1;
    setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(option));
    if (bind(sock, result->ai_addr, result->ai_addrlen) < 0) {
        perror("bind failed");
        exit(1);
    }

    freeaddrinfo(result);
    onlytcp(listen(sock, 32));
    fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK);

    // fds[0] = server socket, fds[1] = stdin, fds[2..] = clients
    struct pollfd fds[ env_max_clients + 2 + MAX_RELAYS ];
    fds[ 1 ].fd     = STDIN_FILENO;
    fds[ 1 ].events = POLLIN;

    // parse relays.txt file
    const char* rpath = getenv("RELAY_TXT");
    if (rpath) {
        FILE* f = fopen(rpath, "r");
        if (f) {
            char line[ 256 ];
            while (fgets(line, sizeof(line), f)) {
                char ip[ 64 ], port[ 16 ];
                if (sscanf(line, "%63s %15s", ip, port) != 2) continue;
                connect_relay(ip, port);
            }
            fclose(f);
        } else {
            perror("unable to open RELAY_TXT");
        }
    }

    while (1) {
        int run_shutdown = 0;
        int got_msg      = 0;
        int resval       = 0;

        fds[ 0 ].fd     = sock;
        fds[ 0 ].events = POLLIN;
        int n           = 2;
        onlytcp(for (int i = 0; i < active; i++) {
            fds[ n ].fd     = clients[ i ].fd;
            fds[ n ].events = POLLIN;
            n++;
        });

        for (int i = 0; i < relay_count; i++) {
            fds[ n ].fd     = relays[ i ].fd;
            fds[ n ].events = POLLIN;
            n++;
        }

        int relay_base = n - relay_count;

        int poll_timeout = onlyudp(env_client_timeout * 1000) onlytcp(-1);
        // find min poll timeout
        for (int i = 0; i < relay_count; i++) {
            if (relays[ i ].is_udp && relays[ i ].remote_timeout > 0) {
                int t = relays[ i ].remote_timeout * 500;
                if (poll_timeout < 0 || t < poll_timeout) poll_timeout = t;
            }
        }
        poll(fds, n, poll_timeout);

        time_t now = time(NULL);

        // sweep udp connections
        onlyudp(for (int i = active - 1; i >= 0; i--) {
            if (now - clients[ i ].last_seen > env_client_timeout) {
                log("evict:%" PRIu64 "\n", clients[ i ].id);
                del_client(i);
            }
        });

        // check admin stdin
        if (fds[ 1 ].revents & POLLIN) {
            char admin_buf[ BUFS ];
            resval = nb_getstring(admin_buf, sizeof(admin_buf));
            if (resval == NB_EOF || (resval > 0 && admin_buf[ 0 ] == 'E')) {
                printf("dying\n");
                run_shutdown = 1;
            } else if (resval > 0) {
                if (admin_buf[ 0 ] == 'B') {
                    // skip 'B' and any spaces
                    char* msg = admin_buf + 1;
                    while (*msg == ' ') msg++;
                    int mlen = resval - (int) (msg - admin_buf);
                    if (mlen > 0) {
                        admin_msg[ 0 ] = 'A';
                        memcpy(admin_msg + 1, msg, mlen);
                        got_msg = 1;
                        resval  = mlen;
                    }
                } else if (admin_buf[ 0 ] == 'C') {
                    char ip[ 64 ], port[ 16 ];
                    if (sscanf(admin_buf + 1, " %63s %15s", ip, port) == 2) connect_relay(ip, port);
                    else
                        printf("usage: C <ip> <port>\n");
                } else {
                    printf("unknown admin command\n");
                }
            }
        }

        // shutdown: notify all clients and exit
        if (unlikely(run_shutdown)) {
            for (int i = 0; i < active; i++) {
                onlytcp(write(clients[ i ].fd, "bye\n", 4));
                onlytcp(close(clients[ i ].fd));
                onlyudp(sendto(sock, "bye\n", 4, 0, (struct sockaddr*) &clients[ i ].addr, clients[ i ].addrlen));
            }
            active = 0;
            break;
        }

        // accept new connections
        onlytcp(if (fds[ 0 ].revents & POLLIN) {
            int c;
            while ((c = accept(sock, 0, 0)) >= 0) add_client(c);
        });

        onlyudp(
            struct sockaddr_storage udp_src; //
            socklen_t               udp_srclen = sizeof(udp_src);
            char                    buf[ env_bufs ];
            int                     rd;
            int                     j = -1);

        // process existing clients
        onlytcp(for (int i = relay_base - 1; i >= 2; i--)) //
            onlyudp(
                while (fds[ 0 ].revents & POLLIN && //
                       (udp_srclen = sizeof(udp_src))
                       && (rd = recvfrom(sock, buf, sizeof(buf), 0, (struct sockaddr*) &udp_src, &udp_srclen)) > 0)) //
        {
            onlytcp(int j = i - 2);

            onlyudp(
                // find existing client
                j = -1; for (int k = 0; k < active; k++) {
                    if (clients[ k ].addrlen == udp_srclen && memcmp(&clients[ k ].addr, &udp_src, udp_srclen) == 0) {
                        j = k;
                        break;
                    }
                }
                // register new client
                if (j < 0) {
                    if (active >= env_max_clients) {
                        sendto(sock, "full\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen);
                        continue;
                    }
                    if (active >= cap) {
                        cap = cap ? cap * 2 : 64;
                        if (cap > env_max_clients) cap = env_max_clients;
                        clients = realloc(clients, cap * sizeof(struct client));
                    }
                    j                     = active;
                    clients[ j ].addr     = udp_src;
                    clients[ j ].addrlen  = udp_srclen;
                    clients[ j ].id       = nid++;
                    clients[ j ].pass     = default_pass;
                    clients[ j ].last_rl  = now;
                    clients[ j ].rl_bytes = 0;
                    active++;
                    // send timeout amount
                    char tbuf[ 32 ];
                    int  tlen = sprintf(tbuf, "T%d\n", env_client_timeout);
                    sendto(sock, tbuf, tlen, 0, (struct sockaddr*) &udp_src, udp_srclen);
                    char ibuf[ 32 ];
                    int  ilen = sprintf(ibuf, "IU%" PRIu64 "\n", clients[ j ].id);
                    sendto(sock, ibuf, ilen, 0, (struct sockaddr*) &udp_src, udp_srclen);

                    log("clients:%d\n", active);
                }

                onlyudp(clients[ j ].last_seen = now); //
            );

            onlytcp(if (likely(fds[ i ].revents & POLLIN))) {
                onlytcp(char buf[ env_bufs ]; int rd = read(fds[ i ].fd, buf, sizeof(buf)); if (rd <= 0) {
                    del_client(j);
                    if (got_msg && j < active) write(clients[ j ].fd, admin_msg, resval + 1);
                    continue;
                });

                if (buf[ 0 ] == 'P' && rd >= 9) {
                    // set password
                    memcpy(&clients[ j ].pass, buf + 1, 8);
                    onlytcp(write(clients[ j ].fd, "p_ok\n", 5));

                } else if (buf[ 0 ] == 'M' && rd >= 9) {
                    // M: broadcast with password
                    if (likely(now - clients[ j ].last_rl >= env_rate_limit_s)) {
                        clients[ j ].last_rl  = now;
                        clients[ j ].rl_bytes = 0;
                    }
                    int plen = rd - 1;
                    if (unlikely(clients[ j ].rl_bytes + plen > env_bufs)) {
                        onlytcp(write(clients[ j ].fd, "Rate\n", 5));
                        onlyudp(sendto(sock, "Rate\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen));
                    } else {
                        clients[ j ].rl_bytes += plen;
                        broadcast_msg(clients[ j ].id, buf + 1, plen);
                        onlytcp(write(clients[ j ].fd, "m_ok\n", 5));
                    }
                } else if (buf[ 0 ] == 'N') {
                    // no-op
                } else if (buf[ 0 ] == 'W') {
                    char str_to_send[] = { 'W', env_wildcard + '0', '\n' };
                    onlytcp(write(clients[ j ].fd, str_to_send, 3));
                    onlyudp(sendto(sock, str_to_send, 3, 0, (struct sockaddr*) &udp_src, udp_srclen));
                } else {
                    onlytcp(write(clients[ j ].fd, "Err\n", 4));
                    onlyudp(sendto(sock, "Err\n", 4, 0, (struct sockaddr*) &udp_src, udp_srclen));
                }
            }

            // send admin message
            onlytcp(if (unlikely(got_msg)) write(clients[ j ].fd, admin_msg, resval + 1));
        }

        // process relay fds
        for (int i = relay_count - 1; i >= 0; i--) {
            struct relay*  r  = &relays[ i ];
            struct pollfd* pf = &fds[ relay_base + i ];

            // udp heartbeat
            if (r->is_udp && r->remote_timeout > 0 && now - r->last_hb >= r->remote_timeout / 2) {
                write(r->fd, "N\n", 2);
                r->last_hb = now;
            }

            if (!(pf->revents & POLLIN)) continue;

            char rbuf[ BUFS ];
            int  rrd;

            rrd = read(r->fd, rbuf, sizeof(rbuf));
            if (rrd <= 0) {
                printf("leave(relay)\n");
                close(r->fd);
                if (i < relay_count - 1) relays[ i ] = relays[ relay_count - 1 ];
                relay_count--;
                continue;
            }
            rbuf[ rrd ] = 0;

            // broadcast relay msg with id1
            if (rbuf[ 0 ] == 'M' && rrd >= 9) { broadcast_msg(1, rbuf + 1, rrd - 1); }
        }

        onlyudp(if (unlikely(got_msg)) {
            for (int i = 0; i < active; i++)
                sendto(sock, admin_msg, resval + 1, 0, (struct sockaddr*) &clients[ i ].addr, clients[ i ].addrlen);
        });
    }

    close(sock);
    free(clients);
    return 0;
}