1/* cluster.rs: keys and topology. */
2#include "mesh/cluster_conf.h"
3#include "mesh/abi/peer_frames.h"
4#include "mesh/parse.h"
5#include "mesh/rust_fmt.h"
6#include "mesh/mt.h"
7
8#include <stdio.h>
9#include <stdlib.h>
10#include <string.h>
11
12mcc_tunables mcc_tunables_default(void) { return (mcc_tunables){MESH_PEER_PING_MS, MESH_PEER_DEAD_MS, msw_default()}; }
13
14/* Rust's trim_matches('"'): every leading and trailing quote. */
15static char *unquote(const char *s) {
16    size_t n = strlen(s);
17    while (n && s[0] == '"') s++, n--;
18    while (n && s[n - 1] == '"') n--;
19    return mt_strndup(s, n);
20}
21
22bool mcc_keys(const char *text, ms_map *out, char **err) {
23    *out = (ms_map){0};
24    mt_strs lines = mt_lines(text);
25    bool ok = true;
26    for (size_t i = 0; i < lines.n && ok; i++) {
27        const char *raw = lines.v[i];
28        const char *hash = strchr(raw, '#');
29        char *pre = hash ? mt_strndup(raw, (size_t)(hash - raw)) : mt_strdup(raw);
30        char *line = mt_trim(pre);
31        free(pre);
32        char *k, *v;
33        if (!*line) {
34        } else if (!mt_split_once(line, '=', &k, &v)) {
35            char *q = rf_debug_str(raw);
36            *err = mt_fmt("line %zu: expected key = value, got %s", i + 1, q);
37            free(q);
38            ok = false;
39        } else {
40            char *kt = mt_trim(k), *vt = mt_trim(v);
41            char *vq = unquote(vt);
42            if (!*kt) {
43                *err = mt_fmt("line %zu: empty key", i + 1);
44                ok = false;
45            } else if (!ms_set(out, kt, vq, false)) {
46                *err = mt_fmt("line %zu: %s given twice", i + 1, kt);
47                ok = false;
48            }
49            free(k), free(v), free(kt), free(vt), free(vq);
50        }
51        free(line);
52    }
53    mt_strs_free(&lines);
54    if (!ok) ms_free(out);
55    return ok;
56}
57
58/* `k.peer` must be host:port (a non-empty host, a u16 port after the last ':'). */
59static bool peer_addr(const ms_map *m, uint32_t k, char **out, char **err) {
60    char key[32];
61    snprintf(key, sizeof key, "%u.peer", k);
62    const char *a = ms_get(m, key);
63    if (!a) {
64        *err = mt_fmt("cluster: no `%u.peer = host:port`", k);
65        return false;
66    }
67    const char *colon = strrchr(a, ':');
68    uint16_t port;
69    if (!colon || colon == a || !mp_u16(colon + 1, &port)) {
70        char *q = rf_debug_str(a);
71        *err = mt_fmt("cluster: %u.peer = %s: expected host:port", k, q);
72        free(q);
73        return false;
74    }
75    *out = mt_strdup(a);
76    return true;
77}
78
79static bool millis(const ms_map *m, const char *k, uint64_t d, uint64_t *out, char **err) {
80    const char *v = ms_get(m, k);
81    if (!v) {
82        *out = d;
83        return true;
84    }
85    if (mp_u64(v, out)) return true;
86    char *q = rf_debug_str(v);
87    *err = mt_fmt("cluster: %s = %s: milliseconds", k, q);
88    free(q);
89    return false;
90}
91
92bool mcc_topology_parse(const char *text, uint32_t me, mcc_topology *out, char **err) {
93    ms_map m;
94    if (!mcc_keys(text, &m, err)) return false;
95    char *peers[MESH_MAX_SYSTEMS] = {0};
96    uint32_t n = 0;
97    bool ok = false;
98    const char *sys = ms_get(&m, "systems");
99    if (!sys) {
100        *err = mt_strdup("cluster: no `systems = N`");
101        goto done;
102    }
103    if (!mp_u32(sys, &n)) {
104        *err = mt_strdup("cluster: systems: not a number");
105        goto done;
106    }
107    if (n < 2 || n > MESH_MAX_SYSTEMS) {
108        *err = mt_fmt("cluster: systems = %u: 2..=%u", n, (unsigned)MESH_MAX_SYSTEMS);
109        goto done;
110    }
111    if (me >= n) {
112        *err = mt_fmt("cluster: --system %u, but the cluster has systems 0..%u", me, n - 1);
113        goto done;
114    }
115    for (uint32_t k = 0; k < n; k++)
116        if (!peer_addr(&m, k, &peers[k], err)) goto done;
117    for (uint32_t i = 0; i < n; i++)
118        for (uint32_t j = 0; j < i; j++)
119            if (strcmp(peers[j], peers[i]) == 0) {
120                *err = mt_fmt("cluster: systems %u and %u share the peer address %s", j, i, peers[i]);
121                goto done;
122            }
123    mcc_topology t = {.me = me, .n = n};
124    if (me > 0) {
125        char key[32];
126        snprintf(key, sizeof key, "%u.listen", me);
127        const char *l = ms_get(&m, key);
128        t.listen = mt_strdup(l ? l : peers[me]);
129    }
130    for (uint32_t j = me + 1; j < n; j++) {
131        t.dial_k[t.ndial] = j;
132        t.dial_addr[t.ndial++] = mt_strdup(peers[j]);
133    }
134    if (!millis(&m, "ping_ms", MESH_PEER_PING_MS, &t.tun.ping_ms, err) || !millis(&m, "dead_ms", MESH_PEER_DEAD_MS, &t.tun.dead_ms, err) ||
135        !msw_from_keys(&m, &t.tun.sched, err)) {
136        mcc_topology_free(&t);
137        goto done;
138    }
139    if (t.tun.ping_ms == 0 || t.tun.dead_ms < 3 * t.tun.ping_ms) {
140        *err = mt_fmt("cluster: dead_ms %llu must be at least 3 x ping_ms %llu", (unsigned long long)t.tun.dead_ms, (unsigned long long)t.tun.ping_ms);
141        mcc_topology_free(&t);
142        goto done;
143    }
144    *out = t;
145    ok = true;
146done:
147    for (uint32_t k = 0; k < MESH_MAX_SYSTEMS; k++) free(peers[k]);
148    ms_free(&m);
149    return ok;
150}
151
152mcc_topology mcc_pair_dialer(const char *peer) {
153    mcc_topology t = {.me = 0, .n = 2, .ndial = 1, .tun = mcc_tunables_default()};
154    t.dial_k[0] = 1;
155    t.dial_addr[0] = mt_strdup(peer);
156    return t;
157}
158
159mcc_topology mcc_pair_listener(const char *addr) {
160    return (mcc_topology){.me = 1, .n = 2, .listen = mt_strdup(addr), .tun = mcc_tunables_default()};
161}
162
163bool mcc_topology_eq(const mcc_topology *a, const mcc_topology *b) {
164    if (a->me != b->me || a->n != b->n || !mt_eq(a->listen, b->listen) || a->ndial != b->ndial) return false;
165    for (size_t i = 0; i < a->ndial; i++)
166        if (a->dial_k[i] != b->dial_k[i] || strcmp(a->dial_addr[i], b->dial_addr[i]) != 0) return false;
167    return a->tun.ping_ms == b->tun.ping_ms && a->tun.dead_ms == b->tun.dead_ms && msw_eq(&a->tun.sched, &b->tun.sched);
168}
169
170void mcc_topology_free(mcc_topology *t) {
171    free(t->listen);
172    for (size_t i = 0; i < t->ndial; i++) free(t->dial_addr[i]);
173    *t = (mcc_topology){0};
174}