|
|
@@ -1,446 +1,639 @@
|
|
|
-// client2server - Go server with WebSocket + Redpanda
|
|
|
+// client2server - Go server with WebSocket + Redpanda (Kafka-compatible)
|
|
|
// Copyright (c) 2026 Luis Rosales - MIT License
|
|
|
//
|
|
|
-// Build: go build -o client2server-server
|
|
|
-// Run: ./client2server-server
|
|
|
-// WebSocket: ws://localhost:3843
|
|
|
-// HTTP API: http://localhost:3843
|
|
|
+// Build: go build -o client2server-server .
|
|
|
+// Run: ./client2server-server
|
|
|
+// Env: REDPANDA_BROKERS=localhost:9092 TOKEN=*** PORT=3843
|
|
|
+//
|
|
|
+// WebSocket: ws://localhost:3843/ws
|
|
|
+// HTTP API: http://localhost:3843/api/{events,routers,command}, /health
|
|
|
|
|
|
package main
|
|
|
|
|
|
import (
|
|
|
"context"
|
|
|
"encoding/json"
|
|
|
+ "errors"
|
|
|
"fmt"
|
|
|
"log"
|
|
|
"net/http"
|
|
|
"os"
|
|
|
"os/signal"
|
|
|
+ "strings"
|
|
|
+ "sync"
|
|
|
"syscall"
|
|
|
"time"
|
|
|
|
|
|
+ "github.com/coder/websocket"
|
|
|
"github.com/google/uuid"
|
|
|
- "github.com/redpanda-data/redpanda-sdk-go/redpanda"
|
|
|
- "github.com/redpanda-data/redpanda-sdk-go/schema"
|
|
|
- "nhooyr.io/websocket"
|
|
|
+ "github.com/twmb/franz-go/pkg/kadm"
|
|
|
+ "github.com/twmb/franz-go/pkg/kerr"
|
|
|
+ "github.com/twmb/franz-go/pkg/kgo"
|
|
|
)
|
|
|
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
// Config
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+
|
|
|
type Config struct {
|
|
|
RedpandaBrokers []string
|
|
|
- WebsocketPort int
|
|
|
- APIPort int
|
|
|
- Token string
|
|
|
+ Port int
|
|
|
+ Token string
|
|
|
}
|
|
|
|
|
|
var cfg = Config{
|
|
|
RedpandaBrokers: []string{"localhost:9092"},
|
|
|
- WebsocketPort: 3843,
|
|
|
- APIPort: 3844,
|
|
|
- Token: "secret-token",
|
|
|
+ Port: 3843,
|
|
|
+ Token: "secret-token",
|
|
|
+}
|
|
|
+
|
|
|
+func loadConfigFromEnv() {
|
|
|
+ cfg.RedpandaBrokers = strings.Split(getEnvStr("REDPANDA_BROKERS", "localhost:9092"), ",")
|
|
|
+ cfg.Port = getEnvInt("PORT", cfg.Port)
|
|
|
+ cfg.Token = getEnvStr("TOKEN", cfg.Token)
|
|
|
+}
|
|
|
+
|
|
|
+func getEnvStr(key, def string) string {
|
|
|
+ if v := os.Getenv(key); v != "" {
|
|
|
+ return v
|
|
|
+ }
|
|
|
+ return def
|
|
|
}
|
|
|
|
|
|
-// Event from router
|
|
|
+func getEnvInt(key string, def int) int {
|
|
|
+ if v := os.Getenv(key); v != "" {
|
|
|
+ var n int
|
|
|
+ if _, err := fmt.Sscanf(v, "%d", &n); err == nil {
|
|
|
+ return n
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return def
|
|
|
+}
|
|
|
+
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// Domain types
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+
|
|
|
+// RouterEvent - generic event sent from a router to the server.
|
|
|
type RouterEvent struct {
|
|
|
- ID string `json:"id"`
|
|
|
- RouterID string `json:"router_id"`
|
|
|
- Hostname string `json:"hostname"`
|
|
|
- EventType string `json:"event_type"`
|
|
|
- Timestamp time.Time `json:"timestamp"`
|
|
|
- Payload map[string]interface{} `json:"payload"`
|
|
|
- ReceivedAt time.Time `json:"received_at"`
|
|
|
- Connection string `json:"connection"`
|
|
|
+ ID string `json:"id"`
|
|
|
+ RouterID string `json:"router_id"`
|
|
|
+ Hostname string `json:"hostname,omitempty"`
|
|
|
+ EventType string `json:"event_type"`
|
|
|
+ Timestamp time.Time `json:"timestamp"`
|
|
|
+ Payload map[string]interface{} `json:"payload"`
|
|
|
+ ReceivedAt time.Time `json:"received_at"`
|
|
|
+ Connection string `json:"connection"` // "websocket" | "http"
|
|
|
}
|
|
|
|
|
|
-// Command to router
|
|
|
+// RouterCommand - command sent from server to a router.
|
|
|
type RouterCommand struct {
|
|
|
- ID string `json:"id"`
|
|
|
- RouterID string `json:"router_id"`
|
|
|
- Command string `json:"command"`
|
|
|
+ ID string `json:"id"`
|
|
|
+ RouterID string `json:"router_id"`
|
|
|
+ Command string `json:"command"`
|
|
|
Args map[string]string `json:"args,omitempty"`
|
|
|
- SentAt time.Time `json:"sent_at"`
|
|
|
+ SentAt time.Time `json:"sent_at"`
|
|
|
}
|
|
|
|
|
|
-// Command result received from router
|
|
|
+// CommandResult - reply from a router after running a command.
|
|
|
type CommandResult struct {
|
|
|
Success bool `json:"success"`
|
|
|
- Output string `json:"output"`
|
|
|
- Error string `json:"error"`
|
|
|
+ Output string `json:"output"`
|
|
|
+ Error string `json:"error"`
|
|
|
}
|
|
|
|
|
|
-// Router state
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// Router registry
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+
|
|
|
type Router struct {
|
|
|
- ID string
|
|
|
- LastSeen time.Time
|
|
|
+ ID string
|
|
|
+ LastSeen time.Time
|
|
|
Conn *websocket.Conn
|
|
|
Connected bool
|
|
|
+ writeMu sync.Mutex // serialise writes to the WS connection
|
|
|
}
|
|
|
|
|
|
-var routers = make(map[string]*Router)
|
|
|
-
|
|
|
-// Per-router command queue (for offline routers - auto-flush on reconnect)
|
|
|
-var routerQueues = make(map[string][]RouterCommand)
|
|
|
-
|
|
|
-// Pending commands awaiting results (command_id -> result channel)
|
|
|
-var pendingCommands = make(map[string]chan CommandResult)
|
|
|
-const commandTimeout = 30 * time.Second
|
|
|
+var (
|
|
|
+ routersMu sync.RWMutex
|
|
|
+ routers = make(map[string]*Router)
|
|
|
+ routerQueuesMu sync.Mutex
|
|
|
+ routerQueues = make(map[string][]RouterCommand) // queued while offline
|
|
|
+ pendingMu sync.Mutex
|
|
|
+ pendingCmds = make(map[string]chan CommandResult) // command_id -> result chan
|
|
|
+ executedMu sync.Mutex
|
|
|
+ executedCmds = make(map[string]time.Time) // idempotency: cmd_id -> last run
|
|
|
+)
|
|
|
|
|
|
-// Idempotency: track recently executed commands (command_id -> timestamp)
|
|
|
-var executedCommands = make(map[string]time.Time)
|
|
|
-const idempotencyTTL = 5 * 60 * time.Second // 5 minutes
|
|
|
+const (
|
|
|
+ commandTimeout = 30 * time.Second
|
|
|
+ idempotencyTTL = 5 * time.Minute
|
|
|
+ wsWriteTimeout = 10 * time.Second
|
|
|
+ publishTimeout = 3 * time.Second
|
|
|
+ offlineThreshold = 60 * time.Second
|
|
|
+)
|
|
|
|
|
|
-// Redpanda
|
|
|
-var rp *redpanda.Client
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// Redpanda (Kafka) client
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
|
|
|
-func initRedpanda() error {
|
|
|
- cfg.RedpandaBrokers = getEnvComma("REDPANDA_BROKERS", "localhost:9092")
|
|
|
+var kcl *kgo.Client
|
|
|
|
|
|
- var err error
|
|
|
- rp, err = redpanda.NewClient(&redpanda.ClientConfig{
|
|
|
- Brokers: cfg.RedpandaBrokers,
|
|
|
- })
|
|
|
+func initRedpanda(ctx context.Context) error {
|
|
|
+ cl, err := kgo.NewClient(
|
|
|
+ kgo.SeedBrokers(cfg.RedpandaBrokers...),
|
|
|
+ kgo.ClientID("client2server"),
|
|
|
+ kgo.ProducerLinger(5*time.Millisecond),
|
|
|
+ kgo.ProducerBatchCompression(kgo.SnappyCompression()),
|
|
|
+ )
|
|
|
if err != nil {
|
|
|
- return fmt.Errorf("redpanda: %v", err)
|
|
|
+ return fmt.Errorf("kafka client: %w", err)
|
|
|
}
|
|
|
+ kcl = cl
|
|
|
|
|
|
- // Create topics
|
|
|
+ // Best-effort topic creation. Redpanda has auto-create enabled in dev, so
|
|
|
+ // this is just to make sure they exist with sane defaults.
|
|
|
+ adm := kadm.NewClient(cl)
|
|
|
topics := []string{"router-events", "router-commands"}
|
|
|
- for _, topic := range topics {
|
|
|
- err := rp.CreateTopic(topic, 1, 3)
|
|
|
- if err != nil && !schema.ErrTopicExists.Exists(err) {
|
|
|
- log.Printf("Topic %s: %v", topic, err)
|
|
|
+ resp, err := adm.CreateTopics(ctx, 1, 1, nil, topics...)
|
|
|
+ if err != nil {
|
|
|
+ log.Printf("create topics admin call failed (non-fatal if auto-create is on): %v", err)
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ for _, ct := range resp {
|
|
|
+ if ct.Err != nil && !errors.Is(ct.Err, kerr.TopicAlreadyExists) {
|
|
|
+ log.Printf("topic %s: %v", ct.Topic, ct.Err)
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
-func getEnvComma(key, def string) []string {
|
|
|
- val := os.Getenv(key)
|
|
|
- if val == "" {
|
|
|
- return []string{def}
|
|
|
- }
|
|
|
- return []string{val}
|
|
|
-}
|
|
|
-
|
|
|
-func getEnvStr(key, def string) string {
|
|
|
- if val := os.Getenv(key); val != "" {
|
|
|
- return val
|
|
|
+func publish(ctx context.Context, topic string, key string, value any) error {
|
|
|
+ data, err := json.Marshal(value)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("marshal: %w", err)
|
|
|
}
|
|
|
- return def
|
|
|
-}
|
|
|
-
|
|
|
-func getEnvInt(key string, def int) int {
|
|
|
- if val := os.Getenv(key); val != "" {
|
|
|
- var v int
|
|
|
- fmt.Sscanf(val, "%d", &v)
|
|
|
- return v
|
|
|
+ rec := &kgo.Record{Topic: topic, Key: []byte(key), Value: data}
|
|
|
+ // Bound how long the HTTP handler can wait for the broker.
|
|
|
+ pctx, cancel := context.WithTimeout(ctx, publishTimeout)
|
|
|
+ defer cancel()
|
|
|
+ res := kcl.ProduceSync(pctx, rec)
|
|
|
+ if err := res.FirstErr(); err != nil {
|
|
|
+ return err
|
|
|
}
|
|
|
- return def
|
|
|
-}
|
|
|
-
|
|
|
-// Publish event to Redpanda
|
|
|
-func publishEvent(event RouterEvent) error {
|
|
|
- data, _ := json.Marshal(event)
|
|
|
- return rp.Produce("router-events", []byte(event.ID), data)
|
|
|
-}
|
|
|
-
|
|
|
-// Publish command to Redpanda
|
|
|
-func publishCommand(cmd RouterCommand) error {
|
|
|
- data, _ := json.Marshal(cmd)
|
|
|
- return rp.Produce("router-commands", []byte(cmd.ID), data)
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
// WebSocket handler
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+
|
|
|
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
|
|
|
+ // Auth: token in query string (?token=...) or Authorization header
|
|
|
token := r.URL.Query().Get("token")
|
|
|
+ if token == "" {
|
|
|
+ if h := r.Header.Get("Authorization"); strings.HasPrefix(h, "Bearer ") {
|
|
|
+ token = strings.TrimPrefix(h, "Bearer ")
|
|
|
+ }
|
|
|
+ }
|
|
|
if token != cfg.Token {
|
|
|
http.Error(w, "Unauthorized", http.StatusUnauthorized)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
|
|
- CompressionMode: websocket.CompressionContextTakeover,
|
|
|
+ // Disable permessage-deflate to keep things simple on minimal routers
|
|
|
+ CompressionMode: websocket.CompressionDisabled,
|
|
|
})
|
|
|
if err != nil {
|
|
|
- log.Printf("WS accept: %v", err)
|
|
|
+ log.Printf("ws accept: %v", err)
|
|
|
return
|
|
|
}
|
|
|
- defer conn.Close(websocket.StatusNormalClosure, "")
|
|
|
+ defer conn.Close(websocket.StatusNormalClosure, "bye")
|
|
|
|
|
|
- ctx := context.Background()
|
|
|
+ ctx, cancel := context.WithCancel(r.Context())
|
|
|
+ defer cancel()
|
|
|
|
|
|
- // Read router registration
|
|
|
- var regMsg RouterEvent
|
|
|
- err = conn.Read(ctx, ®Msg)
|
|
|
+ // First message must be a registration event
|
|
|
+ firstMsg, err := readRouterMessage(ctx, conn)
|
|
|
if err != nil {
|
|
|
- log.Printf("WS read reg: %v", err)
|
|
|
+ log.Printf("ws read reg: %v", err)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ var reg RouterEvent
|
|
|
+ if err := json.Unmarshal(firstMsg, ®); err != nil {
|
|
|
+ log.Printf("ws reg parse: %v", err)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- routerID := regMsg.RouterID
|
|
|
+ routerID := reg.RouterID
|
|
|
if routerID == "" {
|
|
|
routerID = r.RemoteAddr
|
|
|
}
|
|
|
|
|
|
+ routersMu.Lock()
|
|
|
routers[routerID] = &Router{
|
|
|
- ID: routerID,
|
|
|
- LastSeen: time.Now(),
|
|
|
- Conn: conn,
|
|
|
- Connected: true,
|
|
|
- }
|
|
|
-
|
|
|
- log.Printf("Router connected: %s (flushing queue)", routerID)
|
|
|
-
|
|
|
- // Flush queued commands for this router
|
|
|
- if queuedCmds, ok := routerQueues[routerID]; ok && len(queuedCmds) > 0 {
|
|
|
- log.Printf("Flushing %d queued commands to %s", len(queuedCmds), routerID)
|
|
|
- for _, cmd := range queuedCmds {
|
|
|
- resultChan := make(chan CommandResult, 1)
|
|
|
- cmd.SentAt = time.Now()
|
|
|
- pendingCommands[cmd.ID] = resultChan
|
|
|
- publishCommand(cmd)
|
|
|
- log.Printf("Queued cmd sent: %s", cmd.Command)
|
|
|
- // Fire and forget - waiter will handle result
|
|
|
- }
|
|
|
- delete(routerQueues, routerID)
|
|
|
+ ID: routerID,
|
|
|
+ LastSeen: time.Now(),
|
|
|
+ Conn: conn,
|
|
|
+ Connected: true,
|
|
|
}
|
|
|
+ routersMu.Unlock()
|
|
|
+
|
|
|
+ log.Printf("router connected: %s (from %s)", routerID, r.RemoteAddr)
|
|
|
+ flushQueuedCommands(ctx, routerID)
|
|
|
|
|
|
// Message loop
|
|
|
for {
|
|
|
- var event RouterEvent
|
|
|
- err := conn.Read(ctx, &event)
|
|
|
+ raw, err := readRouterMessage(ctx, conn)
|
|
|
if err != nil {
|
|
|
break
|
|
|
}
|
|
|
|
|
|
- event.ID = uuid.New().String()
|
|
|
- event.ReceivedAt = time.Now()
|
|
|
- event.Connection = "websocket"
|
|
|
-
|
|
|
- // Check if this is a command result
|
|
|
- if event.EventType == "command_result" {
|
|
|
- // Find pending command and send result
|
|
|
- if cmdID := event.Payload["command_id"]; cmdID != nil {
|
|
|
- cmdIDstr, _ := cmdID.(string)
|
|
|
- if ch, ok := pendingCommands[cmdIDstr]; ok {
|
|
|
- result := CommandResult{
|
|
|
- Success: event.Payload["success"] == true,
|
|
|
- Output: func() string { s, _ := event.Payload["output"].(string); return s }(),
|
|
|
- Error: func() string { s, _ := event.Payload["error"].(string); return s }(),
|
|
|
- }
|
|
|
- ch <- result
|
|
|
- }
|
|
|
+ var ev RouterEvent
|
|
|
+ if err := json.Unmarshal(raw, &ev); err != nil {
|
|
|
+ log.Printf("[%s] bad event json: %v", routerID, err)
|
|
|
+ continue
|
|
|
+ }
|
|
|
|
|
|
- // Mark as executed (for idempotency)
|
|
|
- executedCommands[cmdIDstr] = time.Now()
|
|
|
+ // Server-assigned fields
|
|
|
+ ev.ID = uuid.New().String()
|
|
|
+ ev.ReceivedAt = time.Now()
|
|
|
+ ev.Connection = "websocket"
|
|
|
+ if ev.Timestamp.IsZero() {
|
|
|
+ ev.Timestamp = ev.ReceivedAt
|
|
|
+ }
|
|
|
+
|
|
|
+ // Command result handling
|
|
|
+ if ev.EventType == "command_result" {
|
|
|
+ if cid, _ := ev.Payload["command_id"].(string); cid != "" {
|
|
|
+ deliverCommandResult(cid, ev.Payload)
|
|
|
+ executedMu.Lock()
|
|
|
+ executedCmds[cid] = time.Now()
|
|
|
+ executedMu.Unlock()
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- // Publish to Redpanda
|
|
|
- if err := publishEvent(event); err != nil {
|
|
|
- log.Printf("Publish error: %v", err)
|
|
|
+ if err := publish(ctx, "router-events", routerID, ev); err != nil {
|
|
|
+ log.Printf("publish event: %v", err)
|
|
|
}
|
|
|
+ log.Printf("[%s] %s", routerID, ev.EventType)
|
|
|
|
|
|
- log.Printf("[%s] %s", routerID, event.EventType)
|
|
|
- routers[routerID].LastSeen = time.Now()
|
|
|
+ routersMu.Lock()
|
|
|
+ if r, ok := routers[routerID]; ok {
|
|
|
+ r.LastSeen = time.Now()
|
|
|
+ }
|
|
|
+ routersMu.Unlock()
|
|
|
}
|
|
|
|
|
|
- if routers[routerID] != nil {
|
|
|
- routers[routerID].Connected = false
|
|
|
+ routersMu.Lock()
|
|
|
+ if r, ok := routers[routerID]; ok {
|
|
|
+ r.Connected = false
|
|
|
}
|
|
|
- log.Printf("Router disconnected: %s", routerID)
|
|
|
+ routersMu.Unlock()
|
|
|
+ log.Printf("router disconnected: %s", routerID)
|
|
|
}
|
|
|
|
|
|
-// HTTP Event webhook
|
|
|
-func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
|
|
|
- if r.Method != "POST" {
|
|
|
- http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
|
|
- return
|
|
|
+func readRouterMessage(ctx context.Context, conn *websocket.Conn) ([]byte, error) {
|
|
|
+ // coder/websocket: Read returns a Message
|
|
|
+ _, data, err := conn.Read(ctx)
|
|
|
+ return data, err
|
|
|
+}
|
|
|
+
|
|
|
+// writeJSON serialises writes to the connection.
|
|
|
+func (r *Router) writeJSON(ctx context.Context, v any) error {
|
|
|
+ data, err := json.Marshal(v)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
}
|
|
|
+ wctx, cancel := context.WithTimeout(ctx, wsWriteTimeout)
|
|
|
+ defer cancel()
|
|
|
+ r.writeMu.Lock()
|
|
|
+ defer r.writeMu.Unlock()
|
|
|
+ return r.Conn.Write(wctx, websocket.MessageText, data)
|
|
|
+}
|
|
|
|
|
|
- token := r.Header.Get("Authorization")
|
|
|
- if token != "Bearer "+cfg.Token && token != cfg.Token {
|
|
|
- http.Error(w, "Unauthorized", http.StatusUnauthorized)
|
|
|
+func deliverCommandResult(cmdID string, payload map[string]interface{}) {
|
|
|
+ pendingMu.Lock()
|
|
|
+ ch, ok := pendingCmds[cmdID]
|
|
|
+ if ok {
|
|
|
+ delete(pendingCmds, cmdID)
|
|
|
+ }
|
|
|
+ pendingMu.Unlock()
|
|
|
+ if !ok {
|
|
|
return
|
|
|
}
|
|
|
+ res := CommandResult{
|
|
|
+ Success: payload["success"] == true,
|
|
|
+ }
|
|
|
+ if s, ok := payload["output"].(string); ok {
|
|
|
+ res.Output = s
|
|
|
+ }
|
|
|
+ if s, ok := payload["error"].(string); ok {
|
|
|
+ res.Error = s
|
|
|
+ }
|
|
|
+ select {
|
|
|
+ case ch <- res:
|
|
|
+ default:
|
|
|
+ }
|
|
|
+}
|
|
|
|
|
|
- var event RouterEvent
|
|
|
- if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
|
|
|
- http.Error(w, "Invalid JSON", http.StatusBadRequest)
|
|
|
+func flushQueuedCommands(ctx context.Context, routerID string) {
|
|
|
+ routerQueuesMu.Lock()
|
|
|
+ queue := routerQueues[routerID]
|
|
|
+ delete(routerQueues, routerID)
|
|
|
+ routerQueuesMu.Unlock()
|
|
|
+ if len(queue) == 0 {
|
|
|
return
|
|
|
}
|
|
|
+ log.Printf("flushing %d queued commands to %s", len(queue), routerID)
|
|
|
+ for _, cmd := range queue {
|
|
|
+ cmd.SentAt = time.Now()
|
|
|
+ pendingMu.Lock()
|
|
|
+ pendingCmds[cmd.ID] = make(chan CommandResult, 1)
|
|
|
+ pendingMu.Unlock()
|
|
|
+ if err := publish(ctx, "router-commands", routerID, cmd); err != nil {
|
|
|
+ log.Printf("queue flush publish: %v", err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
|
|
|
- event.ID = uuid.New().String()
|
|
|
- event.ReceivedAt = time.Now()
|
|
|
- event.Connection = "http"
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// HTTP handlers
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
|
|
|
- if err := publishEvent(event); err != nil {
|
|
|
- w.WriteHeader(http.StatusInternalServerError)
|
|
|
+func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
|
|
|
+ if r.Method != http.MethodPost {
|
|
|
+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
|
return
|
|
|
}
|
|
|
-
|
|
|
- json.NewEncoder(w).Encode(map[string]string{"event_id": event.ID})
|
|
|
+ if !authorised(r) {
|
|
|
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ var ev RouterEvent
|
|
|
+ if err := json.NewDecoder(r.Body).Decode(&ev); err != nil {
|
|
|
+ http.Error(w, "invalid json", http.StatusBadRequest)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ ev.ID = uuid.New().String()
|
|
|
+ ev.ReceivedAt = time.Now()
|
|
|
+ ev.Connection = "http"
|
|
|
+ if ev.Timestamp.IsZero() {
|
|
|
+ ev.Timestamp = ev.ReceivedAt
|
|
|
+ }
|
|
|
+ if err := publish(r.Context(), "router-events", ev.RouterID, ev); err != nil {
|
|
|
+ log.Printf("publish event: %v", err)
|
|
|
+ http.Error(w, "publish failed", http.StatusBadGateway)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]string{
|
|
|
+ "event_id": ev.ID,
|
|
|
+ "router_id": ev.RouterID,
|
|
|
+ "status": "accepted",
|
|
|
+ })
|
|
|
}
|
|
|
|
|
|
-// Get routers
|
|
|
func handleRouters(w http.ResponseWriter, r *http.Request) {
|
|
|
- list := []map[string]interface{}{}
|
|
|
- for id, router := range routers {
|
|
|
- list = append(list, map[string]interface{}{
|
|
|
- "id": id,
|
|
|
- "last_seen": router.LastSeen,
|
|
|
- "online": time.Since(router.LastSeen) < 60*time.Second,
|
|
|
+ type entry struct {
|
|
|
+ ID string `json:"id"`
|
|
|
+ LastSeen time.Time `json:"last_seen"`
|
|
|
+ Online bool `json:"online"`
|
|
|
+ Queued int `json:"queued_commands"`
|
|
|
+ }
|
|
|
+ routersMu.RLock()
|
|
|
+ list := make([]entry, 0, len(routers))
|
|
|
+ for id, rt := range routers {
|
|
|
+ list = append(list, entry{
|
|
|
+ ID: id,
|
|
|
+ LastSeen: rt.LastSeen,
|
|
|
+ Online: time.Since(rt.LastSeen) < offlineThreshold,
|
|
|
+ Queued: len(routerQueues[id]),
|
|
|
})
|
|
|
}
|
|
|
- json.NewEncoder(w).Encode(map[string]interface{}{"routers": list})
|
|
|
+ routersMu.RUnlock()
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]interface{}{"routers": list})
|
|
|
}
|
|
|
|
|
|
-// Send command and wait for result
|
|
|
func handleCommand(w http.ResponseWriter, r *http.Request) {
|
|
|
- if r.Method != "POST" {
|
|
|
- http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
|
|
+ if r.Method != http.MethodPost {
|
|
|
+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ if !authorised(r) {
|
|
|
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
|
return
|
|
|
}
|
|
|
-
|
|
|
var cmd RouterCommand
|
|
|
if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil {
|
|
|
- http.Error(w, "Invalid JSON", http.StatusBadRequest)
|
|
|
+ http.Error(w, "invalid json", http.StatusBadRequest)
|
|
|
return
|
|
|
}
|
|
|
-
|
|
|
routerID := r.URL.Query().Get("router_id")
|
|
|
if routerID == "" {
|
|
|
routerID = cmd.RouterID
|
|
|
}
|
|
|
-
|
|
|
if routerID == "" {
|
|
|
http.Error(w, "router_id required", http.StatusBadRequest)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- router := routers[routerID]
|
|
|
-
|
|
|
- // ──────────────────────────────────────────────────────────
|
|
|
- // IDEMPOTENCY CHECK - Dedup before publishing!
|
|
|
- // ──────────────────────────────────────────────────────────
|
|
|
- // If client provides an idempotency key, reuse it
|
|
|
+ // Idempotency
|
|
|
xecID := cmd.ID
|
|
|
if xecID == "" {
|
|
|
xecID = uuid.New().String()
|
|
|
}
|
|
|
-
|
|
|
- // Check if already executed within TTL
|
|
|
- if lastExec, exists := executedCommands[xecID]; exists {
|
|
|
- if time.Since(lastExec) < idempotencyTTL {
|
|
|
- log.Printf("Idempotent reject: %s (recently executed)", xecID)
|
|
|
- json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
- "idempotent_reject": true,
|
|
|
- "existing_command_id": xecID,
|
|
|
- "message": "Command already executed",
|
|
|
- })
|
|
|
- return
|
|
|
- }
|
|
|
- delete(executedCommands, xecID)
|
|
|
+ executedMu.Lock()
|
|
|
+ if last, exists := executedCmds[xecID]; exists && time.Since(last) < idempotencyTTL {
|
|
|
+ executedMu.Unlock()
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
+ "idempotent_reject": true,
|
|
|
+ "existing_command_id": xecID,
|
|
|
+ "message": "command already executed within TTL",
|
|
|
+ })
|
|
|
+ return
|
|
|
}
|
|
|
+ delete(executedCmds, xecID)
|
|
|
+ executedMu.Unlock()
|
|
|
|
|
|
- // ──────────────────────────────────────────────────────────
|
|
|
- // Queue vs Live delivery
|
|
|
- // ──────────────────────────────────────────────────────────
|
|
|
cmd.ID = xecID
|
|
|
cmd.RouterID = routerID
|
|
|
cmd.SentAt = time.Now()
|
|
|
|
|
|
- if router == nil || !router.Connected {
|
|
|
- // Router offline - QUEUE it!
|
|
|
+ routersMu.RLock()
|
|
|
+ router, online := routers[routerID]
|
|
|
+ routersMu.RUnlock()
|
|
|
+
|
|
|
+ if !online || router == nil || !router.Connected {
|
|
|
+ routerQueuesMu.Lock()
|
|
|
routerQueues[routerID] = append(routerQueues[routerID], cmd)
|
|
|
- log.Printf("Queued command for %s: %s (queue depth: %d)", routerID, cmd.Command, len(routerQueues[routerID]))
|
|
|
- json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
- "command_id": cmd.ID,
|
|
|
- "status": "queued",
|
|
|
- "queued_for": routerID,
|
|
|
- "queue_depth": len(routerQueues[routerID]),
|
|
|
- "message": "Router offline, command queued",
|
|
|
+ depth := len(routerQueues[routerID])
|
|
|
+ routerQueuesMu.Unlock()
|
|
|
+ log.Printf("queued cmd %s for %s (depth=%d)", cmd.Command, routerID, depth)
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
+ "command_id": cmd.ID,
|
|
|
+ "status": "queued",
|
|
|
+ "queued_for": routerID,
|
|
|
+ "queue_depth": depth,
|
|
|
})
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- // Router online - send directly
|
|
|
- resultChan := make(chan CommandResult, 1)
|
|
|
- pendingCommands[cmd.ID] = resultChan
|
|
|
-
|
|
|
- if err := publishCommand(cmd); err != nil {
|
|
|
- delete(pendingCommands, cmd.ID)
|
|
|
- log.Printf("Publish command: %v", err)
|
|
|
- http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
|
+ // Online: register waiter, publish, wait for result
|
|
|
+ resultCh := make(chan CommandResult, 1)
|
|
|
+ pendingMu.Lock()
|
|
|
+ pendingCmds[cmd.ID] = resultCh
|
|
|
+ pendingMu.Unlock()
|
|
|
+
|
|
|
+ if err := publish(r.Context(), "router-commands", routerID, cmd); err != nil {
|
|
|
+ pendingMu.Lock()
|
|
|
+ delete(pendingCmds, cmd.ID)
|
|
|
+ pendingMu.Unlock()
|
|
|
+ log.Printf("publish command: %v", err)
|
|
|
+ http.Error(w, "publish failed", http.StatusBadGateway)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- log.Printf("Command to %s: %s (waiting...)", routerID, cmd.Command)
|
|
|
+ log.Printf("cmd %s -> %s (waiting)", cmd.Command, routerID)
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
|
|
|
- // Wait for result with timeout
|
|
|
select {
|
|
|
- case result := <-resultChan:
|
|
|
- delete(pendingCommands, cmd.ID)
|
|
|
- json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
+ case res := <-resultCh:
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
"command_id": cmd.ID,
|
|
|
- "status": "completed",
|
|
|
- "success": result.Success,
|
|
|
- "output": result.Output,
|
|
|
- "error": result.Error,
|
|
|
+ "status": "completed",
|
|
|
+ "success": res.Success,
|
|
|
+ "output": res.Output,
|
|
|
+ "error": res.Error,
|
|
|
})
|
|
|
case <-time.After(commandTimeout):
|
|
|
- delete(pendingCommands, cmd.ID)
|
|
|
- json.NewEncoder(w).Encode(map[string]string{
|
|
|
+ pendingMu.Lock()
|
|
|
+ delete(pendingCmds, cmd.ID)
|
|
|
+ pendingMu.Unlock()
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]string{
|
|
|
"command_id": cmd.ID,
|
|
|
- "status": "timeout",
|
|
|
- "error": "Router did not respond",
|
|
|
+ "status": "timeout",
|
|
|
+ "error": "router did not respond",
|
|
|
})
|
|
|
}
|
|
|
-
|
|
|
- // Also mark timeout for idempotency (allows retry after TTL)
|
|
|
- // executedCommands stays, will expire naturally
|
|
|
}
|
|
|
|
|
|
-// Health
|
|
|
func handleHealth(w http.ResponseWriter, r *http.Request) {
|
|
|
- json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
- "status": "ok",
|
|
|
- "routers": len(routers),
|
|
|
+ routersMu.RLock()
|
|
|
+ onlineCount := 0
|
|
|
+ for _, rt := range routers {
|
|
|
+ if time.Since(rt.LastSeen) < offlineThreshold {
|
|
|
+ onlineCount++
|
|
|
+ }
|
|
|
+ }
|
|
|
+ total := len(routers)
|
|
|
+ routersMu.RUnlock()
|
|
|
+ w.Header().Set("Content-Type", "application/json")
|
|
|
+ _ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
+ "status": "ok",
|
|
|
+ "routers": total,
|
|
|
+ "routers_online": onlineCount,
|
|
|
+ "redpanda": cfg.RedpandaBrokers,
|
|
|
})
|
|
|
}
|
|
|
|
|
|
-func main() {
|
|
|
- cfg.Token = getEnvStr("TOKEN", cfg.Token)
|
|
|
- cfg.WebsocketPort = getEnvInt("PORT", cfg.WebsocketPort)
|
|
|
+func authorised(r *http.Request) bool {
|
|
|
+ h := r.Header.Get("Authorization")
|
|
|
+ if h == cfg.Token { // legacy: raw token
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ if strings.HasPrefix(h, "Bearer ") && strings.TrimPrefix(h, "Bearer ") == cfg.Token {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ return false
|
|
|
+}
|
|
|
+
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// Background cleanup
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+
|
|
|
+func startJanitor(ctx context.Context) {
|
|
|
+ go func() {
|
|
|
+ t := time.NewTicker(time.Minute)
|
|
|
+ defer t.Stop()
|
|
|
+ for {
|
|
|
+ select {
|
|
|
+ case <-ctx.Done():
|
|
|
+ return
|
|
|
+ case <-t.C:
|
|
|
+ now := time.Now()
|
|
|
+ executedMu.Lock()
|
|
|
+ for id, ts := range executedCmds {
|
|
|
+ if now.Sub(ts) > idempotencyTTL {
|
|
|
+ delete(executedCmds, id)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ executedMu.Unlock()
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+}
|
|
|
+
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
+// main
|
|
|
+// ----------------------------------------------------------------------------
|
|
|
|
|
|
+func main() {
|
|
|
+ loadConfigFromEnv()
|
|
|
log.SetFlags(log.LstdFlags | log.Lshortfile)
|
|
|
log.Printf("=== client2server Go Server ===")
|
|
|
- log.Printf("WebSocket: ws://localhost:%d", cfg.WebsocketPort)
|
|
|
+ log.Printf("port=%d brokers=%v", cfg.Port, cfg.RedpandaBrokers)
|
|
|
|
|
|
- if err := initRedpanda(); err != nil {
|
|
|
- log.Printf("Redpanda init failed: %v", err)
|
|
|
- }
|
|
|
+ ctx, cancel := context.WithCancel(context.Background())
|
|
|
+ defer cancel()
|
|
|
|
|
|
- http.HandleFunc("/", handleHealth)
|
|
|
- http.HandleFunc("/health", handleHealth)
|
|
|
- http.HandleFunc("/api/events", handleHTTPEvent)
|
|
|
- http.HandleFunc("/api/events/", handleHTTPEvent)
|
|
|
- http.HandleFunc("/api/routers", handleRouters)
|
|
|
- http.HandleFunc("/api/command", handleCommand)
|
|
|
- http.HandleFunc("/ws", handleWebSocket)
|
|
|
+ if err := initRedpanda(ctx); err != nil {
|
|
|
+ log.Printf("redpanda init failed (continuing): %v", err)
|
|
|
+ } else {
|
|
|
+ defer kcl.Close()
|
|
|
+ }
|
|
|
+ startJanitor(ctx)
|
|
|
+
|
|
|
+ mux := http.NewServeMux()
|
|
|
+ mux.HandleFunc("/health", handleHealth)
|
|
|
+ mux.HandleFunc("/api/events", handleHTTPEvent)
|
|
|
+ mux.HandleFunc("/api/routers", handleRouters)
|
|
|
+ mux.HandleFunc("/api/command", handleCommand)
|
|
|
+ mux.HandleFunc("/ws", handleWebSocket)
|
|
|
+ mux.HandleFunc("/", handleHealth)
|
|
|
+
|
|
|
+ srv := &http.Server{
|
|
|
+ Addr: fmt.Sprintf(":%d", cfg.Port),
|
|
|
+ Handler: mux,
|
|
|
+ ReadTimeout: 0, // WS connections are long-lived
|
|
|
+ WriteTimeout: 0,
|
|
|
+ }
|
|
|
|
|
|
+ // Graceful shutdown
|
|
|
go func() {
|
|
|
sigCh := make(chan os.Signal, 1)
|
|
|
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
|
|
<-sigCh
|
|
|
- log.Println("Shutting down...")
|
|
|
- for _, router := range routers {
|
|
|
- router.Conn.Close(websocket.StatusNormalClosure, "server shutdown")
|
|
|
+ log.Println("shutting down...")
|
|
|
+ routersMu.RLock()
|
|
|
+ for _, r := range routers {
|
|
|
+ if r.Conn != nil {
|
|
|
+ r.Conn.Close(websocket.StatusNormalClosure, "server shutdown")
|
|
|
+ }
|
|
|
}
|
|
|
+ routersMu.RUnlock()
|
|
|
+ shutdownCtx, c := context.WithTimeout(context.Background(), 5*time.Second)
|
|
|
+ defer c()
|
|
|
+ _ = srv.Shutdown(shutdownCtx)
|
|
|
os.Exit(0)
|
|
|
}()
|
|
|
|
|
|
- log.Printf("Server ready on port %d", cfg.WebsocketPort)
|
|
|
- log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", cfg.WebsocketPort), nil))
|
|
|
-}
|
|
|
+ log.Printf("server ready on :%d", cfg.Port)
|
|
|
+ if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
|
+ log.Fatal(err)
|
|
|
+ }
|
|
|
+}
|
Gogs