tether/server/main.go

// tether-server: HTTP+SSE relay for phone↔client clipboard sync.
// MVP: single broadcast bus, no auth, no rendezvous, no WebRTC yet.
package main

import (
	"embed"
	"encoding/json"
	"flag"
	"fmt"
	"io/fs"
	"log"
	"net/http"
	"sync"
	"time"
)

//go:embed web/index.html
var webFS embed.FS

type Message struct {
	Text   string `json:"text"`
	Source string `json:"source,omitempty"`
	TS     int64  `json:"ts"`
}

type bus struct {
	mu      sync.Mutex
	clients map[chan Message]string // chan → label
	history []Message               // last N (for new subscribers)
}

func newBus() *bus {
	return &bus{clients: map[chan Message]string{}}
}

func (b *bus) subscribe(label string) chan Message {
	ch := make(chan Message, 16)
	b.mu.Lock()
	b.clients[ch] = label
	// replay last few so newcomers see something
	for _, m := range b.history {
		select {
		case ch <- m:
		default:
		}
	}
	b.mu.Unlock()
	return ch
}

func (b *bus) unsubscribe(ch chan Message) {
	b.mu.Lock()
	delete(b.clients, ch)
	b.mu.Unlock()
	close(ch)
}

func (b *bus) publish(m Message) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.history = append(b.history, m)
	if len(b.history) > 10 {
		b.history = b.history[len(b.history)-10:]
	}
	for ch := range b.clients {
		select {
		case ch <- m:
		default: // drop on slow consumer
		}
	}
}

func main() {
	addr := flag.String("addr", ":8765", "listen address")
	flag.Parse()

	b := newBus()

	// Serve embedded HTML
	sub, _ := fs.Sub(webFS, "web")
	mux := http.NewServeMux()
	mux.Handle("/", http.FileServer(http.FS(sub)))

	// POST /api/send — accept a message, broadcast to subscribers
	mux.HandleFunc("/api/send", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "POST only", http.StatusMethodNotAllowed)
			return
		}
		var m Message
		if err := json.NewDecoder(r.Body).Decode(&m); err != nil {
			http.Error(w, "bad json", http.StatusBadRequest)
			return
		}
		if m.Source == "" {
			m.Source = r.Header.Get("X-Tether-Source")
			if m.Source == "" {
				m.Source = "phone"
			}
		}
		m.TS = time.Now().UnixMilli()
		b.publish(m)
		log.Printf("publish: %s len=%d", m.Source, len(m.Text))
		w.WriteHeader(http.StatusNoContent)
	})

	// GET /api/stream — SSE feed of all published messages
	mux.HandleFunc("/api/stream", func(w http.ResponseWriter, r *http.Request) {
		fl, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "no flusher", http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("Connection", "keep-alive")
		w.Header().Set("X-Accel-Buffering", "no")

		label := r.Header.Get("X-Tether-Client")
		if label == "" {
			label = r.RemoteAddr
		}
		ch := b.subscribe(label)
		defer b.unsubscribe(ch)
		log.Printf("subscribe: %s", label)

		// keepalive
		ka := time.NewTicker(30 * time.Second)
		defer ka.Stop()

		for {
			select {
			case <-r.Context().Done():
				log.Printf("unsubscribe: %s", label)
				return
			case m := <-ch:
				bs, _ := json.Marshal(m)
				fmt.Fprintf(w, "event: clipboard\ndata: %s\n\n", bs)
				fl.Flush()
			case <-ka.C:
				fmt.Fprintf(w, ": keepalive\n\n")
				fl.Flush()
			}
		}
	})

	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("ok"))
	})

	log.Printf("tether-server listening on %s", *addr)
	log.Fatal(http.ListenAndServe(*addr, mux))
}