@samitouri / QOSAMI-X402Tunnel / commits / 36a5f7ee

add chatting example

Kim committed Oct 21, 2025 at 16:27 UTC 36a5f7ee152b7fb67fc003aad27064079e37b765
12 files changed +492 -129
Makefile
+4 -4
@@ -28,10 +28,10 @@ CLIENT_FLAGS := \
28 $(CLIENT_BOOTSTRAPS_FLAGS)
29
30 client-run:
31 - go run ./cmd/example_client $(CLIENT_FLAGS)
31 + go run ./cmd/example_http_client $(CLIENT_FLAGS)
32
33 client-build:
34 - go build -trimpath -o bin/relaydns-client ./cmd/example_client
34 + go build -trimpath -o bin/relaydns-client ./cmd/example_http_client
35
36 # ---------- Dev helpers ----------
37 fmt:
@@ -45,7 +45,7 @@ help:
45 @echo " make server-up # build and start relayserver (docker compose)"
46 @echo " make server-down # stop and remove containers"
47 @echo "\nClient:"
48 - @echo " make client-run # run example_client locally with minimal flags"
49 - @echo " make client-build # build example_client to ./bin/relaydns-client"
48 + @echo " make client-run # run example_http_client locally with minimal flags"
49 + @echo " make client-build # build example_http_client to ./bin/relaydns-client"
50 @echo "\nFlags (override with make VAR=value):"
51 @echo " SERVER_URL BACKEND_HTTP BOOTSTRAPS"
cmd/example_chat/main.go new
+91
@@ -0,0 +1,91 @@
1 +package main
2 +
3 +import (
4 + "context"
5 + "flag"
6 + "fmt"
7 + "net"
8 + "net/http"
9 + "os"
10 + "os/signal"
11 + "syscall"
12 + "time"
13 +
14 + "github.com/gosuda/relaydns/relaydns"
15 + "github.com/rs/zerolog/log"
16 +)
17 +
18 +var (
19 + flagServerURL string
20 + flagBootstraps relaydns.StringSlice
21 + flagAddr string
22 + flagName string
23 +)
24 +
25 +func init() {
26 + flag.StringVar(&flagServerURL, "server-url", "http://localhost:8080", "relayserver base URL to fetch multiaddrs from /health")
27 + flag.Var(&flagBootstraps, "bootstrap", "multiaddr with /p2p/ (repeat). supports /dnsaddr/")
28 + flag.StringVar(&flagAddr, "addr", ":8091", "local chat HTTP listen address")
29 + flag.StringVar(&flagName, "name", "demo-chat", "backend display name")
30 +}
31 +
32 +func main() {
33 + flag.Parse()
34 + if err := run(); err != nil {
35 + log.Fatal().Err(err).Msg("example_chat")
36 + }
37 +}
38 +
39 +func run() error {
40 + ctx, cancel := context.WithCancel(context.Background())
41 + defer cancel()
42 +
43 + // 1) start local chat HTTP backend
44 + ln, err := net.Listen("tcp", flagAddr)
45 + if err != nil {
46 + return fmt.Errorf("listen chat: %w", err)
47 + }
48 + hub := newHub()
49 + mux := http.NewServeMux()
50 + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { serveIndex(w, r, flagName) })
51 + mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) { handleWS(w, r, hub) })
52 + srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 60 * time.Second}
53 + go func() {
54 + log.Info().Msgf("[chat] http listening on %s", ln.Addr().String())
55 + if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {
56 + log.Error().Err(err).Msg("chat http error")
57 + cancel()
58 + }
59 + }()
60 +
61 + // 2) advertise over RelayDNS (HTTP tunneled via server /peer route)
62 + client, err := relaydns.NewClient(ctx, relaydns.ClientConfig{
63 + Protocol: "/relaydns/http/1.0",
64 + Topic: "relaydns.backends",
65 + AdvertiseEvery: 3 * time.Second,
66 + Name: flagName,
67 + TargetTCP: relaydns.AddrToTarget(flagAddr),
68 +
69 + ServerURL: flagServerURL,
70 + Bootstraps: flagBootstraps,
71 + HTTPTimeout: 5 * time.Second,
72 + PreferQUIC: true,
73 + PreferLocal: true,
74 + })
75 + if err != nil {
76 + return fmt.Errorf("new client: %w", err)
77 + }
78 + if err := client.Start(ctx); err != nil {
79 + return fmt.Errorf("start client: %w", err)
80 + }
81 + defer client.Close()
82 +
83 + sig := make(chan os.Signal, 1)
84 + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
85 + <-sig
86 + cancel()
87 + shutCtx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second)
88 + _ = srv.Shutdown(shutCtx)
89 + cancelFn()
90 + return nil
91 +}
cmd/example_chat/view.go new
+225
@@ -0,0 +1,225 @@
1 +package main
2 +
3 +import (
4 + "context"
5 + "html/template"
6 + "net/http"
7 + "sync"
8 + "time"
9 +
10 + "github.com/coder/websocket"
11 + "github.com/coder/websocket/wsjson"
12 +)
13 +
14 +// simple in-memory chat hub
15 +type hub struct {
16 + mu sync.RWMutex
17 + messages []message
18 + conns map[*websocket.Conn]struct{}
19 +}
20 +
21 +type message struct {
22 + TS time.Time `json:"ts"`
23 + User string `json:"user"`
24 + Text string `json:"text"`
25 +}
26 +
27 +func newHub() *hub {
28 + return &hub{conns: map[*websocket.Conn]struct{}{}, messages: make([]message, 0, 64)}
29 +}
30 +
31 +func (h *hub) broadcast(m message) {
32 + h.mu.Lock()
33 + h.messages = append(h.messages, m)
34 + conns := make([]*websocket.Conn, 0, len(h.conns))
35 + for c := range h.conns {
36 + conns = append(conns, c)
37 + }
38 + h.mu.Unlock()
39 + for _, c := range conns {
40 + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
41 + _ = wsjson.Write(ctx, c, m)
42 + cancel()
43 + }
44 +}
45 +
46 +func handleWS(w http.ResponseWriter, r *http.Request, h *hub) {
47 + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
48 + // Allow any origin for demo simplicity. Consider tightening in production.
49 + OriginPatterns: []string{"*"},
50 + })
51 + if err != nil {
52 + return
53 + }
54 + // Use a connection-scoped context not tied to the HTTP request lifecycle.
55 + connCtx, cancelConn := context.WithCancel(context.Background())
56 + h.mu.Lock()
57 + h.conns[conn] = struct{}{}
58 + backlog := append([]message(nil), h.messages...)
59 + h.mu.Unlock()
60 + if len(backlog) > 20 {
61 + backlog = backlog[len(backlog)-20:]
62 + }
63 + for _, m := range backlog {
64 + _ = wsjson.Write(connCtx, conn, m)
65 + }
66 + go func() {
67 + defer func() {
68 + h.mu.Lock()
69 + delete(h.conns, conn)
70 + h.mu.Unlock()
71 + conn.Close(websocket.StatusNormalClosure, "")
72 + cancelConn()
73 + }()
74 + for {
75 + var req struct {
76 + User string `json:"user"`
77 + Text string `json:"text"`
78 + }
79 + if err := wsjson.Read(connCtx, conn, &req); err != nil {
80 + return
81 + }
82 + if req.User == "" {
83 + req.User = "anon"
84 + }
85 + if req.Text == "" {
86 + continue
87 + }
88 + h.broadcast(message{TS: time.Now().UTC(), User: req.User, Text: req.Text})
89 + }
90 + }()
91 +}
92 +
93 +func serveIndex(w http.ResponseWriter, r *http.Request, name string) {
94 + w.Header().Set("Content-Type", "text/html; charset=utf-8")
95 + _ = indexTmpl.Execute(w, struct{ Name string }{Name: name})
96 +}
97 +
98 +var indexTmpl = template.Must(template.New("chat").Parse(`<!DOCTYPE html>
99 +<html lang="en">
100 +<head>
101 + <meta charset="utf-8" />
102 + <meta name="viewport" content="width=device-width, initial-scale=1" />
103 + <title>RelayDNS Chat — {{.Name}}</title>
104 + <style>
105 + :root{
106 + --bg: #0d1117;
107 + --panel: #111827;
108 + --border: #1f2937;
109 + --fg: #e5e7eb;
110 + --muted: #9ca3af;
111 + --accent: #22c55e;
112 + --cursor: #22c55e;
113 + }
114 + *{ box-sizing: border-box }
115 + body { margin:0; padding:24px; background:var(--bg); color:var(--fg); font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial }
116 + .wrap { max-width: 920px; margin: 0 auto }
117 + h1 { margin:0 0 12px 0; font-weight:700 }
118 + .term { border:1px solid var(--border); border-radius:10px; background:var(--panel); overflow:hidden }
119 + .termbar { display:flex; align-items:center; justify-content:space-between; padding:10px 12px; border-bottom:1px solid var(--border); font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size:14px }
120 + .dots { display:flex; gap:6px }
121 + .dot { width:10px; height:10px; border-radius:50%; }
122 + .dot.red{ background:#ef4444 }
123 + .dot.yellow{ background:#f59e0b }
124 + .dot.green{ background:#22c55e }
125 + .nick { display:flex; align-items:center; gap:8px }
126 + .nick input{ background:transparent; border:1px solid var(--border); color:var(--fg); padding:6px 8px; border-radius:6px; font-family:inherit; font-size:13px; width:180px }
127 + .nick button{ background:transparent; border:1px solid var(--border); color:var(--fg); padding:6px 8px; border-radius:6px; font-family:inherit; font-size:13px; cursor:pointer }
128 + .screen { height:420px; overflow:auto; padding:14px; font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size:14px; line-height:1.5; }
129 + .line { white-space: pre-wrap; word-break: break-word }
130 + .ts { color:var(--muted) }
131 + .usr { color:#60a5fa }
132 + .promptline { display:flex; align-items:center; gap:8px; padding:12px 14px; border-top:1px solid var(--border); font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; }
133 + #prompt { color:var(--accent) }
134 + #cmd { flex:1; background:transparent; border:none; outline:none; color:var(--fg); font-family: inherit; font-size:14px; caret-color: var(--cursor) }
135 + small{ color:var(--muted); display:block; margin-top:10px }
136 + </style>
137 +</head>
138 +<body>
139 + <div class="wrap">
140 + <h1>🔐 Chatting — {{.Name}}</h1>
141 + <div class="term">
142 + <div class="termbar">
143 + <div class="dots"><span class="dot red"></span><span class="dot yellow"></span><span class="dot green"></span></div>
144 + <div style="opacity:.9">relaychat@relaydns</div>
145 + <div class="nick">
146 + <label for="user" style="color:var(--muted)">nick</label>
147 + <input id="user" type="text" placeholder="anon" />
148 + <button id="roll" title="randomize nickname">🎲</button>
149 + </div>
150 + </div>
151 + <div id="log" class="screen"></div>
152 + <div class="promptline">
153 + <span id="prompt"></span>
154 + <input id="cmd" type="text" autocomplete="off" spellcheck="false" placeholder="type a message and press Enter" />
155 + </div>
156 + </div>
157 + <small>Tip: Enter to send • Nickname persists locally</small>
158 + </div>
159 + <script>
160 + const log = document.getElementById('log');
161 + const user = document.getElementById('user');
162 + const cmd = document.getElementById('cmd');
163 + const roll = document.getElementById('roll');
164 + const promptEl = document.getElementById('prompt');
165 +
166 + function setPrompt(){
167 + const nick = (user.value || 'anon').replace(/\s+/g,'').slice(0,24) || 'anon';
168 + promptEl.textContent = nick + '@chat:~$';
169 + }
170 + function randomNick(){
171 + const adjs = ['brisk','calm','clever','daring','eager','gentle','merry','rapid','vivid','bright'];
172 + const animals = ['fox','lion','panda','otter','eagle','whale','koala','lynx','squid','yak'];
173 + const a = adjs[Math.floor(Math.random()*adjs.length)];
174 + const b = animals[Math.floor(Math.random()*animals.length)];
175 + const id = Math.random().toString(36).slice(2,6);
176 + return a + '-' + b + '-' + id;
177 + }
178 + // Restore nickname or initialize randomly
179 + let savedNick = null;
180 + try { savedNick = localStorage.getItem('relaydns_nick'); } catch(_) {}
181 + if(savedNick){
182 + user.value = savedNick;
183 + } else {
184 + user.value = randomNick();
185 + try { localStorage.setItem('relaydns_nick', user.value); } catch(_) {}
186 + }
187 + setPrompt();
188 + user.addEventListener('input', () => { try{ localStorage.setItem('relaydns_nick', user.value); }catch(_){}; setPrompt(); });
189 + roll.addEventListener('click', () => {
190 + user.value = randomNick();
191 + try{ localStorage.setItem('relaydns_nick', user.value); }catch(_){}
192 + setPrompt();
193 + user.focus();
194 + });
195 +
196 + function append(msg){
197 + const div = document.createElement('div');
198 + div.className = 'line';
199 + const ts = new Date(msg.ts).toLocaleTimeString();
200 + div.innerHTML = '<span class="ts">[' + ts + ']</span> <span class="usr">' +
201 + (msg.user || 'anon') + '</span>: ' + escapeHTML(msg.text || '');
202 + log.appendChild(div);
203 + log.scrollTop = log.scrollHeight;
204 + }
205 + function escapeHTML(s){
206 + return s.replace(/[&<>\"]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','\"':'&quot;'}[c]));
207 + }
208 +
209 + const wsProto = location.protocol === 'https:' ? 'wss' : 'ws';
210 + const ws = new WebSocket(wsProto + '://' + location.host + location.pathname + 'ws');
211 + ws.onmessage = (e) => { try{ append(JSON.parse(e.data)); }catch(_){ } };
212 + function send(){
213 + const payload = { user: (user.value || 'anon'), text: cmd.value.trim() };
214 + if(!payload.text) return;
215 + ws.send(JSON.stringify(payload));
216 + cmd.value='';
217 + }
218 + cmd.addEventListener('keydown', e => {
219 + if(e.key === 'Enter') { e.preventDefault(); send(); }
220 + });
221 + // Focus command line on load
222 + setTimeout(()=>cmd.focus(), 0);
223 + </script>
224 +</body>
225 +</html>`))
cmd/example_http_client/main.go renamed
+5 -12
@@ -76,11 +76,11 @@ func runClient(cmd *cobra.Command, args []string) error {
76
77 // 2) libp2p host
78 client, err := relaydns.NewClient(ctx, relaydns.ClientConfig{
79 - Protocol: flagProtocol,
80 - Topic: flagTopic,
81 - Advertise: 3 * time.Second,
82 - Name: flagClientName,
83 - TargetTCP: addrToTarget(flagAddr),
79 + Protocol: flagProtocol,
80 + Topic: flagTopic,
81 + AdvertiseEvery: 3 * time.Second,
82 + Name: flagClientName,
83 + TargetTCP: relaydns.AddrToTarget(flagAddr),
84
85 ServerURL: flagServerURL,
86 Bootstraps: flagBootstraps,
@@ -105,10 +105,3 @@ func runClient(cmd *cobra.Command, args []string) error {
105 time.Sleep(200 * time.Millisecond)
106 return nil
107 }
108 -
109 -func addrToTarget(listen string) string {
110 - if len(listen) > 0 && listen[0] == ':' {
111 - return "127.0.0.1" + listen
112 - }
113 - return listen
114 -}
cmd/example_http_client/view.go renamed
cmd/server/view.go
+9 -16
@@ -14,7 +14,7 @@ import (
14 "github.com/rs/zerolog/log"
15 )
16
17 -// serveHTTP builds the HTTP mux and starts serving admin UI + per-peer proxy.
17 +// serveHTTP builds the HTTP mux.
18 func serveHTTP(ctx context.Context, addr string, d *relaydns.Director, h host.Host, cancel context.CancelFunc) {
19 if addr == "" {
20 return
@@ -33,12 +33,13 @@ func serveHTTP(ctx context.Context, addr string, d *relaydns.Director, h host.Ho
33 if v.Info.TTL > 0 {
34 ttl = fmt.Sprintf("%ds", v.Info.TTL)
35 }
36 + // Derive a friendly kind from protocol id. Be lenient to variations.
37 + p := strings.ToLower(v.Info.Proto)
38 kind := "TCP"
37 - if strings.Contains(v.Info.Proto, "/http/") {
38 - kind = "HTTP"
39 - }
40 - if strings.Contains(v.Info.Proto, "/ssh/") {
39 + if strings.Contains(p, "ssh") {
40 kind = "SSH"
41 + } else if strings.Contains(p, "http") {
42 + kind = "HTTP"
43 }
44 rows = append(rows, row{
45 Peer: v.Info.Peer,
@@ -55,7 +56,7 @@ func serveHTTP(ctx context.Context, addr string, d *relaydns.Director, h host.Ho
56 log.Debug().Int("clients", len(rows)).Msg("render admin index")
57 _ = adminIndexTmpl.Execute(w, page{
58 NodeID: h.ID().String(),
58 - Addrs: buildAddrs(h),
59 + Addrs: relaydns.BuildAddrs(h),
60 Rows: rows,
61 })
62 })
@@ -87,12 +88,12 @@ func serveHTTP(ctx context.Context, addr string, d *relaydns.Director, h host.Ho
88 Status string `json:"status"`
89 Addrs []string `json:"multiaddrs"`
90 }
90 - resp := info{Status: "ok", Addrs: buildAddrs(h)}
91 + resp := info{Status: "ok", Addrs: relaydns.BuildAddrs(h)}
92 w.Header().Set("Content-Type", "application/json")
93 _ = json.NewEncoder(w).Encode(resp)
94 })
95
95 - log.Info().Msgf("[server] http (admin+proxy): %s", addr)
96 + log.Info().Msgf("[server] http: %s", addr)
97 if err := http.ListenAndServe(addr, mux); err != nil {
98 log.Error().Err(err).Msg("[server] http error")
99 cancel()
@@ -117,14 +118,6 @@ type page struct {
118 Rows []row
119 }
120
120 -func buildAddrs(h host.Host) []string {
121 - out := make([]string, 0)
122 - for _, a := range h.Addrs() {
123 - out = append(out, fmt.Sprintf("%s/p2p/%s", a.String(), h.ID().String()))
124 - }
125 - return out
126 -}
127 -
121 var adminIndexTmpl = template.Must(template.New("admin-index").Parse(`<!doctype html>
122 <html lang="ko">
123 <head>
go.mod
+1
@@ -3,6 +3,7 @@ module github.com/gosuda/relaydns
3 go 1.25.0
4
5 require (
6 + github.com/coder/websocket v1.8.14
7 github.com/libp2p/go-libp2p v0.44.0
8 github.com/libp2p/go-libp2p-pubsub v0.15.0
9 github.com/multiformats/go-multiaddr v0.16.0
go.sum
+2
@@ -19,6 +19,8 @@ github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23/go.mod h1:bbYlZJ7
19 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
20 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
21 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
22 +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
23 +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
24 github.com/coreos/go-systemd v0.0.0-20181012123002-c6f51f82210d/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
25 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
26 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
relaydns/client.go
+9 -87
@@ -3,14 +3,9 @@ package relaydns
3 import (
4 "context"
5 "encoding/json"
6 - "errors"
6 "fmt"
7 "io"
8 "net"
10 - "net/http"
11 - "net/url"
12 - "sort"
13 - "strings"
9 "sync"
10 "time"
11
@@ -33,7 +28,7 @@ type ClientConfig struct {
28 // pubsub topic for backend adverts (e.g. "relaydns.backends")
29 Topic string
30 // advertise interval
36 - Advertise time.Duration
31 + AdvertiseEvery time.Duration
32 // advertise TTL (how long server should keep this entry alive)
33 AdvertiseTTL time.Duration
34 // how often to refresh server health/bootstraps (if ServerURL set)
@@ -66,11 +61,11 @@ type RelayClient struct {
61 // NewClient constructs a client with defaults applied and an initialized libp2p host.
62 // It does not start networking. Call Start(ctx) to begin handlers, pubsub, and advertising.
63 func NewClient(ctx context.Context, cfg ClientConfig) (*RelayClient, error) {
69 - if cfg.Advertise <= 0 {
70 - cfg.Advertise = 5 * time.Second
64 + if cfg.AdvertiseEvery <= 0 {
65 + cfg.AdvertiseEvery = 5 * time.Second
66 }
67 if cfg.AdvertiseTTL <= 0 {
73 - cfg.AdvertiseTTL = 10 * cfg.Advertise
68 + cfg.AdvertiseTTL = 10 * cfg.AdvertiseEvery
69 }
70 if cfg.HTTPTimeout <= 0 {
71 cfg.HTTPTimeout = 3 * time.Second
@@ -162,18 +157,14 @@ func (b *RelayClient) Start(ctx context.Context) error {
157 b.wg.Add(1)
158 go func() {
159 defer b.wg.Done()
165 - ticker := time.NewTicker(b.cfg.Advertise)
160 + ticker := time.NewTicker(b.cfg.AdvertiseEvery)
161 defer ticker.Stop()
162 for {
163 select {
164 case <-advCtx.Done():
165 return
166 case <-ticker.C:
172 - addrs := b.h.Addrs()
173 - enc := make([]string, 0, len(addrs))
174 - for _, a := range addrs {
175 - enc = append(enc, fmt.Sprintf("%s/p2p/%s", a.String(), b.h.ID().String()))
176 - }
167 + enc := BuildAddrs(b.h)
168 ad := Advertise{
169 Peer: b.h.ID().String(),
170 Name: b.cfg.Name,
@@ -220,9 +211,9 @@ func (b *RelayClient) Start(ctx context.Context) error {
211 }()
212 }
213
223 - if addrs := b.Host().Addrs(); len(addrs) > 0 {
224 - for _, a := range addrs {
225 - log.Info().Msgf("[client] host addr: %s/p2p/%s", a.String(), b.Host().ID().String())
214 + if addrs := BuildAddrs(b.Host()); len(addrs) > 0 {
215 + for _, s := range addrs {
216 + log.Info().Msgf("[client] host addr: %s", s)
217 }
218 } else {
219 log.Info().Msgf("[client] host peer: %s (no listen addrs yet)", b.Host().ID().String())
@@ -264,72 +255,3 @@ func (b *RelayClient) ServerStatus() string {
255 }
256 return "Connecting..."
257 }
267 -
268 -func fetchMultiaddrsFromHealth(base string, timeout time.Duration) ([]string, error) {
269 - u, err := url.Parse(base)
270 - if err != nil {
271 - return nil, fmt.Errorf("parse server-url: %w", err)
272 - }
273 - // ensure path ends with /health
274 - if !strings.HasSuffix(u.Path, "/health") {
275 - if u.Path == "" || u.Path == "/" {
276 - u.Path = "/health"
277 - } else {
278 - u.Path = strings.TrimSuffix(u.Path, "/") + "/health"
279 - }
280 - }
281 - client := &http.Client{Timeout: timeout}
282 - req, _ := http.NewRequest(http.MethodGet, u.String(), nil)
283 - resp, err := client.Do(req)
284 - if err != nil {
285 - return nil, err
286 - }
287 - defer resp.Body.Close()
288 -
289 - var payload struct {
290 - Status string `json:"status"`
291 - PeerID string `json:"peerId"`
292 - Multiaddrs []string `json:"multiaddrs"`
293 - }
294 - if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
295 - return nil, err
296 - }
297 - if payload.Status != "ok" {
298 - return nil, errors.New("health not ok")
299 - }
300 - addrs := make([]string, 0, len(payload.Multiaddrs))
301 - for _, s := range payload.Multiaddrs {
302 - // sanity check
303 - if strings.Contains(s, "/p2p/") && (strings.Contains(s, "/ip4/") || strings.Contains(s, "/ip6/")) {
304 - addrs = append(addrs, s)
305 - }
306 - }
307 - return addrs, nil
308 -}
309 -
310 -func sortMultiaddrs(addrs []string, preferQUIC, preferLocal bool) {
311 - score := func(a string) int {
312 - sc := 0
313 - if preferQUIC && strings.Contains(a, "/quic-v1") {
314 - sc += 2
315 - }
316 - if preferLocal && (strings.Contains(a, "/ip4/127.0.0.1/") || strings.Contains(a, "/ip6/::1/")) {
317 - sc += 1
318 - }
319 - return sc
320 - }
321 - sort.SliceStable(addrs, func(i, j int) bool { return score(addrs[i]) > score(addrs[j]) })
322 -}
323 -
324 -func uniq(ss []string) []string {
325 - seen := map[string]struct{}{}
326 - out := make([]string, 0, len(ss))
327 - for _, s := range ss {
328 - if _, ok := seen[s]; ok {
329 - continue
330 - }
331 - seen[s] = struct{}{}
332 - out = append(out, s)
333 - }
334 - return out
335 -}
relaydns/director.go
+42 -2
@@ -10,12 +10,14 @@ import (
10 "net/http"
11 "net/url"
12 "sort"
13 + "strings"
14 "sync"
15 "time"
16
17 pubsub "github.com/libp2p/go-libp2p-pubsub"
18 "github.com/libp2p/go-libp2p/core/host"
19 "github.com/libp2p/go-libp2p/core/peer"
20 + "github.com/libp2p/go-libp2p/core/protocol"
21 ma "github.com/multiformats/go-multiaddr"
22 "github.com/rs/zerolog/log"
23 )
@@ -187,7 +189,7 @@ func (d *Director) ProxyHTTP(w http.ResponseWriter, r *http.Request, peerID, pat
189 http.Error(w, "upstream connect failed", http.StatusBadGateway)
190 return
191 }
190 - s, err := d.h.NewStream(d.ctx, entry.AddrInfo.ID, protocolID(d.protocol))
192 + s, err := d.h.NewStream(d.ctx, entry.AddrInfo.ID, protocol.ID(d.protocol))
193 if err != nil {
194 log.Error().Err(err).Msg("new stream")
195 http.Error(w, "open stream failed", http.StatusBadGateway)
@@ -210,12 +212,50 @@ func (d *Director) ProxyHTTP(w http.ResponseWriter, r *http.Request, peerID, pat
212 return
213 }
214 defer resp.Body.Close()
215 +
216 + // Handle WebSocket Upgrade: write 101 response and then raw-tunnel bytes
217 + if resp.StatusCode == http.StatusSwitchingProtocols && strings.Contains(strings.ToLower(resp.Header.Get("Upgrade")), "websocket") {
218 + if hj, ok := w.(http.Hijacker); ok {
219 + clientConn, clientBuf, err := hj.Hijack()
220 + if err != nil {
221 + log.Error().Err(err).Msg("hijack client conn")
222 + return
223 + }
224 + defer clientConn.Close()
225 + // Write upstream 101 response (headers)
226 + if err := resp.Write(clientBuf); err == nil {
227 + _ = clientBuf.Flush()
228 + }
229 + // Raw byte tunnel between client and upstream stream
230 + go io.Copy(s, clientConn)
231 + io.Copy(clientConn, s)
232 + return
233 + }
234 + }
235 for k, vv := range resp.Header {
236 for _, v := range vv {
237 w.Header().Add(k, v)
238 }
239 }
240 w.WriteHeader(resp.StatusCode)
241 + // If upstream is SSE, forward with Flush to avoid buffering
242 + if strings.Contains(strings.ToLower(resp.Header.Get("Content-Type")), "text/event-stream") {
243 + if f, ok := w.(http.Flusher); ok {
244 + buf := make([]byte, 4096)
245 + for {
246 + n, err := resp.Body.Read(buf)
247 + if n > 0 {
248 + if _, werr := w.Write(buf[:n]); werr == nil {
249 + f.Flush()
250 + }
251 + }
252 + if err != nil {
253 + break
254 + }
255 + }
256 + return
257 + }
258 + }
259 _, _ = io.Copy(w, resp.Body)
260 }
261
@@ -232,7 +272,7 @@ func (d *Director) ProxyTCP(c net.Conn, peerID string) error {
272 if err := d.h.Connect(d.ctx, *entry.AddrInfo); err != nil {
273 return err
274 }
235 - s, err := d.h.NewStream(d.ctx, entry.AddrInfo.ID, protocolID(d.protocol))
275 + s, err := d.h.NewStream(d.ctx, entry.AddrInfo.ID, protocol.ID(d.protocol))
276 if err != nil {
277 return err
278 }
relaydns/types.go
+1 -8
@@ -4,7 +4,6 @@ import (
4 "time"
5
6 "github.com/libp2p/go-libp2p/core/peer"
7 - "github.com/libp2p/go-libp2p/core/protocol"
7 )
8
9 type Advertise struct {
@@ -15,7 +14,7 @@ type Advertise struct {
14 Ready bool `json:"ready"`
15 Load float64 `json:"load"`
16 TS time.Time `json:"ts"`
18 - TTL int `json:"ttl,omitempty"` // seconds
17 + TTL int `json:"ttl,omitempty"`
18 Proto string `json:"proto,omitempty"`
19 }
20
@@ -25,9 +24,3 @@ type HostEntry struct {
24 LastSeen time.Time
25 Connected bool
26 }
28 -
29 -// Removed Picker: selection is explicit via /peer/{peerID}/...
30 -
31 -func protocolID(s string) protocol.ID {
32 - return protocol.ID(s)
33 -}
relaydns/utils.go new
+103
@@ -0,0 +1,103 @@
1 +package relaydns
2 +
3 +import (
4 + "encoding/json"
5 + "errors"
6 + "fmt"
7 + "net/http"
8 + "net/url"
9 + "sort"
10 + "strings"
11 + "time"
12 +
13 + "github.com/libp2p/go-libp2p/core/host"
14 +)
15 +
16 +type StringSlice []string
17 +
18 +func (s *StringSlice) String() string { return fmt.Sprint([]string(*s)) }
19 +func (s *StringSlice) Set(v string) error { *s = append(*s, v); return nil }
20 +
21 +func AddrToTarget(listen string) string {
22 + if len(listen) > 0 && listen[0] == ':' {
23 + return "127.0.0.1" + listen
24 + }
25 + return listen
26 +}
27 +
28 +func BuildAddrs(h host.Host) []string {
29 + out := make([]string, 0)
30 + for _, a := range h.Addrs() {
31 + out = append(out, fmt.Sprintf("%s/p2p/%s", a.String(), h.ID().String()))
32 + }
33 + return out
34 +}
35 +
36 +func sortMultiaddrs(addrs []string, preferQUIC, preferLocal bool) {
37 + score := func(a string) int {
38 + sc := 0
39 + if preferQUIC && strings.Contains(a, "/quic-v1") {
40 + sc += 2
41 + }
42 + if preferLocal && (strings.Contains(a, "/ip4/127.0.0.1/") || strings.Contains(a, "/ip6/::1/")) {
43 + sc += 1
44 + }
45 + return sc
46 + }
47 + sort.SliceStable(addrs, func(i, j int) bool { return score(addrs[i]) > score(addrs[j]) })
48 +}
49 +
50 +func uniq(ss []string) []string {
51 + seen := map[string]struct{}{}
52 + out := make([]string, 0, len(ss))
53 + for _, s := range ss {
54 + if _, ok := seen[s]; ok {
55 + continue
56 + }
57 + seen[s] = struct{}{}
58 + out = append(out, s)
59 + }
60 + return out
61 +}
62 +
63 +func fetchMultiaddrsFromHealth(base string, timeout time.Duration) ([]string, error) {
64 + u, err := url.Parse(base)
65 + if err != nil {
66 + return nil, fmt.Errorf("parse server-url: %w", err)
67 + }
68 + // ensure path ends with /health
69 + if !strings.HasSuffix(u.Path, "/health") {
70 + if u.Path == "" || u.Path == "/" {
71 + u.Path = "/health"
72 + } else {
73 + u.Path = strings.TrimSuffix(u.Path, "/") + "/health"
74 + }
75 + }
76 + client := &http.Client{Timeout: timeout}
77 + req, _ := http.NewRequest(http.MethodGet, u.String(), nil)
78 + resp, err := client.Do(req)
79 + if err != nil {
80 + return nil, err
81 + }
82 + defer resp.Body.Close()
83 +
84 + var payload struct {
85 + Status string `json:"status"`
86 + PeerID string `json:"peerId"`
87 + Multiaddrs []string `json:"multiaddrs"`
88 + }
89 + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
90 + return nil, err
91 + }
92 + if payload.Status != "ok" {
93 + return nil, errors.New("health not ok")
94 + }
95 + addrs := make([]string, 0, len(payload.Multiaddrs))
96 + for _, s := range payload.Multiaddrs {
97 + // sanity check
98 + if strings.Contains(s, "/p2p/") && (strings.Contains(s, "/ip4/") || strings.Contains(s, "/ip6/")) {
99 + addrs = append(addrs, s)
100 + }
101 + }
102 + return addrs, nil
103 +}