// Package realtime publishes events to a Soketi (Pusher-compatible) // WebSocket server so the frontend can receive realtime notifications. // // We implement the Pusher REST "trigger" protocol directly (HMAC-SHA256 // signed POST to /apps/{id}/events) rather than pull in a client library — // it's a small, stable wire format and avoids a new dependency. // // Channel auth is intentionally NOT handled here: we publish to PUBLIC // channels for now (see the notification service). When Soketi Apps are // configured for connection/channel authentication, switch the channel // names to the `private-` prefix and add a /broadcasting/auth endpoint — // the trigger code below is unchanged by that. package realtime import ( "bytes" "context" "crypto/hmac" "crypto/md5" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "net/http" "sort" "strconv" "strings" "time" "go.uber.org/zap" ) // SoketiClient triggers events on a Soketi/Pusher app. A nil *SoketiClient // is valid and makes Trigger a no-op, so callers can hold an optional // client without nil-checks. type SoketiClient struct { appID string key string secret string baseURL string // scheme://host[:port] http *http.Client logger *zap.Logger } // NewSoketiClient returns a configured client, or nil when any required // credential is missing (realtime disabled — callers degrade gracefully). func NewSoketiClient(appID, key, secret, host string, port int, useTLS bool, logger *zap.Logger) *SoketiClient { if appID == "" || key == "" || secret == "" || host == "" { logger.Info("realtime: Soketi not configured — notifications won't be pushed over WebSocket") return nil } scheme := "http" if useTLS { scheme = "https" } hostport := host // Omit the port when it's the scheme default (cleaner, and matches // what most reverse-proxied setups expect). if port != 0 && !((useTLS && port == 443) || (!useTLS && port == 80)) { hostport = fmt.Sprintf("%s:%d", host, port) } return &SoketiClient{ appID: appID, key: key, secret: secret, baseURL: scheme + "://" + hostport, http: &http.Client{Timeout: 5 * time.Second}, logger: logger, } } // Trigger publishes `event` with `data` (JSON-marshalled) to `channel`. // Best-effort: on a nil client it no-ops; errors are returned for the // caller to log-and-ignore. func (c *SoketiClient) Trigger(ctx context.Context, channel, event string, data any) error { if c == nil { return nil } dataJSON, err := json.Marshal(data) if err != nil { return fmt.Errorf("marshal event data: %w", err) } // Pusher wraps the payload as a JSON STRING in the `data` field. body, err := json.Marshal(map[string]any{ "name": event, "channels": []string{channel}, "data": string(dataJSON), }) if err != nil { return fmt.Errorf("marshal trigger body: %w", err) } bodyMD5 := fmt.Sprintf("%x", md5.Sum(body)) path := "/apps/" + c.appID + "/events" // Auth params (all values are signature-safe, no URL-encoding needed), // sorted alphabetically by key for the canonical string-to-sign. params := map[string]string{ "auth_key": c.key, "auth_timestamp": strconv.FormatInt(time.Now().Unix(), 10), "auth_version": "1.0", "body_md5": bodyMD5, } keys := make([]string, 0, len(params)) for k := range params { keys = append(keys, k) } sort.Strings(keys) var qs strings.Builder for i, k := range keys { if i > 0 { qs.WriteByte('&') } qs.WriteString(k) qs.WriteByte('=') qs.WriteString(params[k]) } query := qs.String() stringToSign := "POST\n" + path + "\n" + query mac := hmac.New(sha256.New, []byte(c.secret)) mac.Write([]byte(stringToSign)) signature := hex.EncodeToString(mac.Sum(nil)) fullURL := c.baseURL + path + "?" + query + "&auth_signature=" + signature req, err := http.NewRequestWithContext(ctx, http.MethodPost, fullURL, bytes.NewReader(body)) if err != nil { return fmt.Errorf("build trigger request: %w", err) } req.Header.Set("Content-Type", "application/json") resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("trigger request: %w", err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return fmt.Errorf("soketi trigger %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) } return nil }