| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/subtle" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "strings" |
| 12 | "time" |
| 13 | ) |
| 14 | |
| 15 | const defaultControlAddr = "127.0.0.1:37913" |
| 16 | |
| 17 | type controlHTTPServer struct { |
| 18 | server *http.Server |
| 19 | addr string |
| 20 | } |
| 21 | |
| 22 | type controlStatusResponse struct { |
| 23 | Status string `json:"status"` |
| 24 | ControlAddr string `json:"control_addr,omitempty"` |
| 25 | ActiveSessions int `json:"active_sessions"` |
| 26 | RetainedSessions int `json:"retained_sessions"` |
| 27 | StartErrors []string `json:"start_errors,omitempty"` |
| 28 | Adapters []AdapterHealthSnapshot `json:"adapters"` |
| 29 | } |
| 30 | |
| 31 | type controlSendRequest struct { |
| 32 | ConnectionID string `json:"connection_id"` |
| 33 | Domain string `json:"domain,omitempty"` |
| 34 | ChatID string `json:"chat_id"` |
| 35 | ChatType ChatType `json:"chat_type,omitempty"` |
| 36 | Text string `json:"text,omitempty"` |
| 37 | MediaURLs []string `json:"media_urls,omitempty"` |
| 38 | ReplyToMsgID string `json:"reply_to_msg_id,omitempty"` |
| 39 | } |
| 40 | |
| 41 | type controlSendResponse struct { |
| 42 | MessageID string `json:"message_id,omitempty"` |
| 43 | MessageIDs []string `json:"message_ids,omitempty"` |
| 44 | Partial bool `json:"partial,omitempty"` |
| 45 | Error string `json:"error,omitempty"` |
| 46 | } |
| 47 | |
| 48 | func (gw *BotGateway) startControlServer(parent context.Context) error { |
| 49 | if !gw.cfg.ControlEnabled { |
| 50 | return nil |
| 51 | } |
| 52 | token := strings.TrimSpace(gw.cfg.ControlToken) |
| 53 | if token == "" { |
| 54 | return errors.New("bot control is enabled but control token is empty") |
| 55 | } |
| 56 | addr := strings.TrimSpace(gw.cfg.ControlAddr) |
| 57 | if addr == "" { |
| 58 | addr = defaultControlAddr |
| 59 | } |
| 60 | if err := validateLoopbackAddr(addr); err != nil { |
| 61 | return err |
| 62 | } |
| 63 | mux := http.NewServeMux() |
| 64 | mux.HandleFunc("/status", gw.controlAuth(gw.handleControlStatus)) |
| 65 | mux.HandleFunc("/health", gw.controlAuth(gw.handleControlStatus)) |
| 66 | mux.HandleFunc("/metrics", gw.controlAuth(gw.handleControlMetrics)) |
| 67 | mux.HandleFunc("/send", gw.controlAuth(gw.handleControlSend)) |
| 68 | |
| 69 | ln, err := net.Listen("tcp", addr) |
| 70 | if err != nil { |
| 71 | return fmt.Errorf("start bot control server: %w", err) |
| 72 | } |
| 73 | srv := &http.Server{ |
| 74 | Handler: mux, |
| 75 | ReadHeaderTimeout: 5 * time.Second, |
| 76 | } |
| 77 | control := &controlHTTPServer{server: srv, addr: ln.Addr().String()} |
| 78 | gw.mu.Lock() |
| 79 | gw.controlServer = control |
| 80 | gw.mu.Unlock() |
| 81 | |
| 82 | gw.gatewayWG.Add(2) |
| 83 | go func() { |
| 84 | defer gw.gatewayWG.Done() |
| 85 | <-parent.Done() |
| 86 | gw.stopControlServer() |
| 87 | }() |
| 88 | go func() { |
| 89 | defer gw.gatewayWG.Done() |
| 90 | if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { |
| 91 | gw.logger.Warn("bot control server stopped with error", "addr", control.addr, "err", err) |
| 92 | } |
| 93 | }() |
| 94 | gw.logger.Info("bot control server started", "addr", control.addr) |
| 95 | return nil |
| 96 | } |
| 97 | |
| 98 | func (gw *BotGateway) stopControlServer() { |
| 99 | gw.mu.Lock() |
| 100 | control := gw.controlServer |
| 101 | gw.controlServer = nil |
| 102 | gw.mu.Unlock() |
| 103 | if control == nil || control.server == nil { |
| 104 | return |
| 105 | } |
| 106 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 107 | defer cancel() |
| 108 | if err := control.server.Shutdown(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) { |
| 109 | gw.logger.Warn("bot control server shutdown failed", "addr", control.addr, "err", err) |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | // ControlAddr returns the bound loopback control address when enabled. |
| 114 | func (gw *BotGateway) ControlAddr() string { |
| 115 | gw.mu.Lock() |
| 116 | defer gw.mu.Unlock() |
| 117 | if gw.controlServer == nil { |
| 118 | return "" |
| 119 | } |
| 120 | return gw.controlServer.addr |
| 121 | } |
| 122 | |
| 123 | func validateLoopbackAddr(addr string) error { |
| 124 | host, _, err := net.SplitHostPort(addr) |
| 125 | if err != nil { |
| 126 | return fmt.Errorf("bot control addr must be host:port on loopback: %w", err) |
| 127 | } |
| 128 | host = strings.Trim(host, "[]") |
| 129 | if strings.EqualFold(host, "localhost") { |
| 130 | return nil |
| 131 | } |
| 132 | ip := net.ParseIP(host) |
| 133 | if ip == nil || !ip.IsLoopback() { |
| 134 | return fmt.Errorf("bot control addr must bind to loopback, got %q", addr) |
| 135 | } |
| 136 | return nil |
| 137 | } |
| 138 | |
| 139 | func (gw *BotGateway) controlAuth(next http.HandlerFunc) http.HandlerFunc { |
| 140 | return func(w http.ResponseWriter, r *http.Request) { |
| 141 | got := strings.TrimSpace(r.Header.Get("Authorization")) |
| 142 | const prefix = "Bearer " |
| 143 | if !strings.HasPrefix(got, prefix) || subtle.ConstantTimeCompare([]byte(strings.TrimSpace(strings.TrimPrefix(got, prefix))), []byte(strings.TrimSpace(gw.cfg.ControlToken))) != 1 { |
| 144 | http.Error(w, "unauthorized", http.StatusUnauthorized) |
| 145 | return |
| 146 | } |
| 147 | next(w, r) |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | func (gw *BotGateway) handleControlStatus(w http.ResponseWriter, r *http.Request) { |
| 152 | if r.Method != http.MethodGet { |
| 153 | http.Error(w, "method not allowed", http.StatusMethodNotAllowed) |
| 154 | return |
| 155 | } |
| 156 | gw.mu.Lock() |
| 157 | retained := len(gw.controllers) |
| 158 | gw.mu.Unlock() |
| 159 | startErrs := gw.StartErrors() |
| 160 | errTexts := make([]string, 0, len(startErrs)) |
| 161 | for _, err := range startErrs { |
| 162 | if err != nil { |
| 163 | errTexts = append(errTexts, err.Error()) |
| 164 | } |
| 165 | } |
| 166 | status := "running" |
| 167 | for _, health := range gw.AdapterHealth() { |
| 168 | switch health.Status { |
| 169 | case "error", "degraded", "closed": |
| 170 | status = "degraded" |
| 171 | } |
| 172 | } |
| 173 | writeControlJSON(w, controlStatusResponse{ |
| 174 | Status: status, |
| 175 | ControlAddr: gw.ControlAddr(), |
| 176 | ActiveSessions: gw.sessions.ActiveCount(), |
| 177 | RetainedSessions: retained, |
| 178 | StartErrors: errTexts, |
| 179 | Adapters: gw.AdapterHealth(), |
| 180 | }) |
| 181 | } |
| 182 | |
| 183 | func (gw *BotGateway) handleControlSend(w http.ResponseWriter, r *http.Request) { |
| 184 | if r.Method != http.MethodPost { |
| 185 | http.Error(w, "method not allowed", http.StatusMethodNotAllowed) |
| 186 | return |
| 187 | } |
| 188 | var req controlSendRequest |
| 189 | if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil { |
| 190 | http.Error(w, "invalid json: "+err.Error(), http.StatusBadRequest) |
| 191 | return |
| 192 | } |
| 193 | if strings.TrimSpace(req.ConnectionID) == "" || strings.TrimSpace(req.ChatID) == "" { |
| 194 | http.Error(w, "connection_id and chat_id are required", http.StatusBadRequest) |
| 195 | return |
| 196 | } |
| 197 | result, err := gw.SendToAdapter(r.Context(), req.ConnectionID, req.Domain, OutboundMessage{ |
| 198 | ConnectionID: req.ConnectionID, |
| 199 | Domain: req.Domain, |
| 200 | ChatID: req.ChatID, |
| 201 | ChatType: req.ChatType, |
| 202 | Text: req.Text, |
| 203 | MediaURLs: req.MediaURLs, |
| 204 | ReplyToMsgID: req.ReplyToMsgID, |
| 205 | }) |
| 206 | if err != nil { |
| 207 | if len(result.DeliveredMessageIDs()) > 0 { |
| 208 | writeControlJSONStatus(w, http.StatusMultiStatus, controlSendResponse{ |
| 209 | MessageID: result.MessageID, |
| 210 | MessageIDs: result.MessageIDs, |
| 211 | Partial: true, |
| 212 | Error: err.Error(), |
| 213 | }) |
| 214 | return |
| 215 | } |
| 216 | http.Error(w, err.Error(), http.StatusBadGateway) |
| 217 | return |
| 218 | } |
| 219 | writeControlJSON(w, controlSendResponse{MessageID: result.MessageID, MessageIDs: result.MessageIDs}) |
| 220 | } |
| 221 | |
| 222 | func (gw *BotGateway) handleControlMetrics(w http.ResponseWriter, r *http.Request) { |
| 223 | if r.Method != http.MethodGet { |
| 224 | http.Error(w, "method not allowed", http.StatusMethodNotAllowed) |
| 225 | return |
| 226 | } |
| 227 | gw.mu.Lock() |
| 228 | retained := len(gw.controllers) |
| 229 | gw.mu.Unlock() |
| 230 | w.Header().Set("Content-Type", "text/plain; version=0.0.4") |
| 231 | fmt.Fprintf(w, "# TYPE reasonix_bot_active_sessions gauge\nreasonix_bot_active_sessions %d\n", gw.sessions.ActiveCount()) |
| 232 | fmt.Fprintf(w, "# TYPE reasonix_bot_retained_sessions gauge\nreasonix_bot_retained_sessions %d\n", retained) |
| 233 | fmt.Fprintln(w, "# TYPE reasonix_bot_adapter_messages_total counter") |
| 234 | for _, health := range gw.AdapterHealth() { |
| 235 | labels := adapterMetricLabels(health) |
| 236 | fmt.Fprintf(w, "reasonix_bot_adapter_messages_total{%s} %d\n", labels, health.Messages) |
| 237 | } |
| 238 | fmt.Fprintln(w, "# TYPE reasonix_bot_adapter_sends_total counter") |
| 239 | for _, health := range gw.AdapterHealth() { |
| 240 | labels := adapterMetricLabels(health) |
| 241 | fmt.Fprintf(w, "reasonix_bot_adapter_sends_total{%s} %d\n", labels, health.Sends) |
| 242 | } |
| 243 | fmt.Fprintln(w, "# TYPE reasonix_bot_adapter_send_errors_total counter") |
| 244 | for _, health := range gw.AdapterHealth() { |
| 245 | labels := adapterMetricLabels(health) |
| 246 | fmt.Fprintf(w, "reasonix_bot_adapter_send_errors_total{%s} %d\n", labels, health.SendErrors) |
| 247 | } |
| 248 | fmt.Fprintln(w, "# TYPE reasonix_bot_adapter_status gauge") |
| 249 | for _, health := range gw.AdapterHealth() { |
| 250 | labels := adapterMetricLabels(health) |
| 251 | fmt.Fprintf(w, "reasonix_bot_adapter_status{%s,status=\"%s\"} 1\n", labels, prometheusLabelValue(health.Status)) |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | func writeControlJSON(w http.ResponseWriter, v any) { |
| 256 | writeControlJSONStatus(w, http.StatusOK, v) |
| 257 | } |
| 258 | |
| 259 | func writeControlJSONStatus(w http.ResponseWriter, status int, v any) { |
| 260 | w.Header().Set("Content-Type", "application/json") |
| 261 | w.WriteHeader(status) |
| 262 | _ = json.NewEncoder(w).Encode(v) |
| 263 | } |
| 264 | |
| 265 | func adapterMetricLabels(health AdapterHealthSnapshot) string { |
| 266 | return fmt.Sprintf("id=\"%s\",platform=\"%s\",domain=\"%s\"", |
| 267 | prometheusLabelValue(health.ID), |
| 268 | prometheusLabelValue(string(health.Platform)), |
| 269 | prometheusLabelValue(health.Domain), |
| 270 | ) |
| 271 | } |
| 272 | |
| 273 | func prometheusLabelValue(value string) string { |
| 274 | value = strings.ReplaceAll(value, "\\", "\\\\") |
| 275 | value = strings.ReplaceAll(value, "\n", "\\n") |
| 276 | value = strings.ReplaceAll(value, "\"", "\\\"") |
| 277 | return value |
| 278 | } |
| 279 |