Files
nofx/telemetry/experience.go
shinchan-zhai fd28aa7fc6 reliability: wrap all 27 bare goroutines with safe.Go/GoNamed panic recovery
Applied safe.GoNamed to:
- 9 exchange order_sync goroutines (OKX, Hyperliquid, Aster, Bybit, KuCoin, Gate, Bitget, Lighter, Binance×2)
- Drawdown monitor (auto_trader_risk.go)
- Brain news scanner + market briefs (agent/brain.go)
- Sentinel scanner (agent/sentinel.go)
- Agent scheduler (agent/scheduler.go)
- x402 idle watchdog (mcp/payment/x402.go)
- MCP stream idle watchdog (mcp/client.go)
- Rate limiter cleanup (api/rate_limiter.go)
- 3 telemetry fire-and-forget sends (telemetry/experience.go)
- CoinAnk WS handler (provider/coinank/coinank_api/kline_ws.go)
- API server goroutine (main.go)

Added manual defer/recover with error reporting to:
- Telegram AI agent handler (sends error msg to user on panic)
- Trader data fetch (returns error result on panic to prevent deadlock)

Before: a panic in ANY of these 27 goroutines would crash the entire
trading process with zero diagnostics. Now all panics are caught, logged
with stack traces, and the process continues running.
2026-03-25 01:05:54 +08:00

244 lines
5.3 KiB
Go

// Package telemetry handles product telemetry
package telemetry
import (
"bytes"
"encoding/json"
"net/http"
"nofx/safe"
"sync"
"time"
)
const (
telemetryEndpoint = "https://www.google-analytics.com/mp/collect"
tid = "G-14J8SY6F0J"
tk = "sgPLmshGTPiF-X57rzEIKA"
)
var (
client *Client
clientOnce sync.Once
httpClient = &http.Client{Timeout: 5 * time.Second}
)
type Client struct {
enabled bool
installationID string
mu sync.RWMutex
}
type TradeEvent struct {
Exchange string
TradeType string
Symbol string
AmountUSD float64
Leverage int
UserID string
TraderID string
}
type AIUsageEvent struct {
UserID string
TraderID string
ModelProvider string // openai, deepseek, anthropic, etc.
ModelName string // gpt-4o, deepseek-chat, claude-3, etc.
Channel string // payment channel: "claw402" or "native"
InputTokens int
OutputTokens int
}
type telemetryPayload struct {
ClientID string `json:"client_id"`
Events []telemetryEvent `json:"events"`
}
type telemetryEvent struct {
Name string `json:"name"`
Params map[string]interface{} `json:"params"`
}
func Init(enabled bool, installationID string) {
clientOnce.Do(func() {
client = &Client{
enabled: enabled,
installationID: installationID,
}
})
}
func SetInstallationID(id string) {
if client == nil {
return
}
client.mu.Lock()
defer client.mu.Unlock()
client.installationID = id
}
func GetInstallationID() string {
if client == nil {
return ""
}
client.mu.RLock()
defer client.mu.RUnlock()
return client.installationID
}
func SetEnabled(enabled bool) {
if client == nil {
return
}
client.mu.Lock()
defer client.mu.Unlock()
client.enabled = enabled
}
func IsEnabled() bool {
if client == nil {
return false
}
client.mu.RLock()
defer client.mu.RUnlock()
return client.enabled
}
func TrackTrade(event TradeEvent) {
if client == nil || !IsEnabled() {
return
}
// Send asynchronously to not block trading
safe.Go(func() {
_ = sendTradeEvent(event)
})
}
// sendTradeEvent sends the trade event to GA4
func sendTradeEvent(event TradeEvent) error {
client.mu.RLock()
installationID := client.installationID
client.mu.RUnlock()
payload := telemetryPayload{
ClientID: installationID,
Events: []telemetryEvent{
{
Name: "trade",
Params: map[string]interface{}{
"exchange": event.Exchange,
"trade_type": event.TradeType,
"symbol": event.Symbol,
"amount_usd": event.AmountUSD,
"leverage": event.Leverage,
"installation_id": installationID, // For counting active installations
"user_id": event.UserID, // For counting active users
"trader_id": event.TraderID, // For counting active traders
"engagement_time_msec": 1, // Required by GA4
},
},
},
}
jsonData, err := json.Marshal(payload)
if err != nil {
return err
}
url := telemetryEndpoint + "?measurement_id=" + tid + "&api_secret=" + tk
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := httpClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
return nil
}
func TrackStartup(version string) {
if client == nil || !IsEnabled() {
return
}
safe.Go(func() {
client.mu.RLock()
installationID := client.installationID
client.mu.RUnlock()
payload := telemetryPayload{
ClientID: installationID,
Events: []telemetryEvent{
{
Name: "app_startup",
Params: map[string]interface{}{
"version": version,
"installation_id": installationID,
"engagement_time_msec": 1,
},
},
},
}
jsonData, _ := json.Marshal(payload)
url := telemetryEndpoint + "?measurement_id=" + tid + "&api_secret=" + tk
req, _ := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
if req != nil {
req.Header.Set("Content-Type", "application/json")
resp, err := httpClient.Do(req)
if err == nil {
resp.Body.Close()
}
}
})
}
func TrackAIUsage(event AIUsageEvent) {
if client == nil || !IsEnabled() {
return
}
safe.Go(func() {
client.mu.RLock()
installationID := client.installationID
client.mu.RUnlock()
payload := telemetryPayload{
ClientID: installationID,
Events: []telemetryEvent{
{
Name: "ai_usage",
Params: map[string]interface{}{
"model_provider": event.ModelProvider,
"model_name": event.ModelName,
"channel": event.Channel,
"input_tokens": event.InputTokens,
"output_tokens": event.OutputTokens,
"total_tokens": event.InputTokens + event.OutputTokens,
"installation_id": installationID,
"user_id": event.UserID,
"trader_id": event.TraderID,
"engagement_time_msec": 1,
},
},
},
}
jsonData, _ := json.Marshal(payload)
url := telemetryEndpoint + "?measurement_id=" + tid + "&api_secret=" + tk
req, _ := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
if req != nil {
req.Header.Set("Content-Type", "application/json")
resp, err := httpClient.Do(req)
if err == nil {
resp.Body.Close()
}
}
})
}