package core import ( "context" "crypto/subtle" "encoding/base64" "encoding/json" "fmt" "log/slog" "net/http" "strconv" "strings" "sync" "time" "github.com/gorilla/websocket" ) // --------------------------------------------------------------------------- // BridgeServer — global WebSocket server shared across all engines // --------------------------------------------------------------------------- // BridgeServer exposes a WebSocket endpoint for external platform adapters. // A single instance is created globally; each project engine receives a // lightweight BridgePlatform handle that delegates to this server. type BridgeServer struct { port int token string path string corsOrigins []string insecure bool // allow running without token (local dev only) server *http.Server mu sync.RWMutex adapters map[string]*bridgeAdapter // platform name → adapter enginesMu sync.RWMutex engines map[string]*bridgeEngineRef // project name → engine ref } type bridgeEngineRef struct { engine *Engine platform *BridgePlatform } type bridgeAdapter struct { platform string capabilities map[string]bool metadata map[string]any conn *websocket.Conn writeMu sync.Mutex server *BridgeServer previewMu sync.Mutex previewRequests map[string]chan string // ref_id → channel receiving preview_handle } // bridgeReplyCtx carries the information needed to route replies back to the adapter. type bridgeReplyCtx struct { Platform string `json:"platform"` SessionKey string `json:"session_key"` ReplyCtx string `json:"reply_ctx"` progressStyle string `json:"-"` supportsProgressCardPayload bool `json:"-"` } func (rc *bridgeReplyCtx) progressStyleHint() string { if rc == nil { return progressStyleLegacy } return rc.progressStyle } func (rc *bridgeReplyCtx) supportsProgressCardPayloadHint() bool { if rc == nil { return false } return rc.supportsProgressCardPayload } const bridgeReconstructReplyCtxKind = "bridge_reconstruct" // bridgeReconstructReplyCtxPayload is a forward-compatible reply envelope for // reconstruct_reply adapters. Receivers should ignore unknown fields. type bridgeReconstructReplyCtxPayload struct { Kind string `json:"kind"` Version int `json:"v"` SenderProject string `json:"sender_project"` TransportChatID string `json:"transport_chat_id"` TransportSessionKey string `json:"transport_session_key,omitempty"` } // --- Wire protocol messages --- type bridgeMsg struct { Type string `json:"type"` } type bridgeRegister struct { Type string `json:"type"` Platform string `json:"platform"` Capabilities []string `json:"capabilities"` Project string `json:"project,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } type bridgeMessage struct { Type string `json:"type"` MsgID string `json:"msg_id"` SessionKey string `json:"session_key"` UserID string `json:"user_id"` UserName string `json:"user_name,omitempty"` Content string `json:"content"` ReplyCtx string `json:"reply_ctx"` Project string `json:"project,omitempty"` Images []bridgeImageData `json:"images,omitempty"` Files []bridgeFileData `json:"files,omitempty"` Audio *bridgeAudioData `json:"audio,omitempty"` } type bridgeCardAction struct { Type string `json:"type"` SessionKey string `json:"session_key"` Action string `json:"action"` ReplyCtx string `json:"reply_ctx"` Project string `json:"project,omitempty"` } type bridgePreviewAck struct { Type string `json:"type"` RefID string `json:"ref_id"` PreviewHandle string `json:"preview_handle"` } type bridgeImageData struct { MimeType string `json:"mime_type"` Data string `json:"data"` // base64 FileName string `json:"file_name,omitempty"` } type bridgeFileData struct { MimeType string `json:"mime_type"` Data string `json:"data"` // base64 FileName string `json:"file_name"` } type bridgeAudioData struct { MimeType string `json:"mime_type"` Data string `json:"data"` // base64 Format string `json:"format"` Duration int `json:"duration,omitempty"` } func NewBridgeServer(port int, token, path string, corsOrigins []string) *BridgeServer { return newBridgeServer(port, token, path, corsOrigins, false) } // NewBridgeServerInsecure creates a BridgeServer that allows running without token. // This should only be used for local development. func NewBridgeServerInsecure(port int, token, path string, corsOrigins []string) *BridgeServer { return newBridgeServer(port, token, path, corsOrigins, true) } func newBridgeServer(port int, token, path string, corsOrigins []string, insecure bool) *BridgeServer { if port <= 0 { port = 9810 } if path == "" { path = "/bridge/ws" } if !strings.HasPrefix(path, "/") { path = "/" + path } // Validate security settings if token == "" && !insecure { slog.Error("bridge: token is required when insecure mode is not enabled", "help", "set bridge.token in config, or use insecure=true for local development only") return nil } if insecure && token == "" { slog.Warn("bridge: running in INSECURE mode without authentication - only use for local development!") } return &BridgeServer{ port: port, token: token, path: path, corsOrigins: corsOrigins, insecure: insecure, adapters: make(map[string]*bridgeAdapter), engines: make(map[string]*bridgeEngineRef), } } // NewPlatform creates a BridgePlatform for a specific project engine. func (bs *BridgeServer) NewPlatform(projectName string) *BridgePlatform { return &BridgePlatform{server: bs, project: projectName} } // RegisterEngine associates a project engine with its BridgePlatform. func (bs *BridgeServer) RegisterEngine(projectName string, engine *Engine, bp *BridgePlatform) { bs.enginesMu.Lock() defer bs.enginesMu.Unlock() if err := bp.Start(engine.handleMessage); err != nil { slog.Warn("bridge: platform start failed", "project", projectName, "error", err) } bp.SetCardNavigationHandler(engine.handleCardNav) bs.engines[projectName] = &bridgeEngineRef{engine: engine, platform: bp} } // Start launches the HTTP/WebSocket server. func (bs *BridgeServer) Start() { mux := http.NewServeMux() mux.HandleFunc(bs.path, bs.handleWS) // Session management REST endpoints (with CORS support) mux.HandleFunc("/bridge/sessions", bs.corsHTTP(bs.authHTTP(bs.handleSessions))) mux.HandleFunc("/bridge/sessions/", bs.corsHTTP(bs.authHTTP(bs.handleSessionRoutes))) addr := fmt.Sprintf(":%d", bs.port) bs.server = &http.Server{Addr: addr, Handler: mux} go func() { slog.Info("bridge: server started", "addr", addr, "path", bs.path) if err := bs.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { slog.Error("bridge: server error", "error", err) } }() } // corsHTTP wraps a handler with CORS headers. OPTIONS preflight is handled directly. func (bs *BridgeServer) corsHTTP(handler http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { bs.setCORS(w, r) if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) return } handler(w, r) } } // setCORS sets Access-Control-* headers when the request origin matches cors_origins. func (bs *BridgeServer) setCORS(w http.ResponseWriter, r *http.Request) { if len(bs.corsOrigins) == 0 { return } origin := r.Header.Get("Origin") for _, o := range bs.corsOrigins { if o == "*" || o == origin { w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PATCH, DELETE, OPTIONS") w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type") w.Header().Set("Access-Control-Max-Age", "86400") break } } } // Stop shuts down the server and closes all adapter connections. func (bs *BridgeServer) Stop() { bs.mu.Lock() for _, a := range bs.adapters { a.conn.Close() } bs.adapters = make(map[string]*bridgeAdapter) bs.mu.Unlock() if bs.server != nil { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := bs.server.Shutdown(ctx); err != nil && err != http.ErrServerClosed { slog.Debug("bridge: server shutdown failed", "error", err) } } } // ConnectedAdapters returns the names of currently connected adapters. func (bs *BridgeServer) ConnectedAdapters() []string { bs.mu.RLock() defer bs.mu.RUnlock() names := make([]string, 0, len(bs.adapters)) for name := range bs.adapters { names = append(names, name) } return names } // --------------------------------------------------------------------------- // BridgePlatform — per-engine Platform that delegates to BridgeServer // --------------------------------------------------------------------------- // BridgePlatform implements core.Platform for a single project. // It is a lightweight handle; the actual WebSocket server lives in BridgeServer. type BridgePlatform struct { server *BridgeServer project string handler MessageHandler navHandler CardNavigationHandler } // Compile-time interface checks. var ( _ Platform = (*BridgePlatform)(nil) _ CardSender = (*BridgePlatform)(nil) _ InlineButtonSender = (*BridgePlatform)(nil) _ MessageUpdater = (*BridgePlatform)(nil) _ PreviewStarter = (*BridgePlatform)(nil) _ PreviewCleaner = (*BridgePlatform)(nil) _ TypingIndicator = (*BridgePlatform)(nil) _ AudioSender = (*BridgePlatform)(nil) _ ImageSender = (*BridgePlatform)(nil) _ FileSender = (*BridgePlatform)(nil) _ CardNavigable = (*BridgePlatform)(nil) _ ReplyContextReconstructor = (*BridgePlatform)(nil) ) func (bp *BridgePlatform) Name() string { return "bridge" } func (bp *BridgePlatform) Start(handler MessageHandler) error { bp.handler = handler return nil } func (bp *BridgePlatform) Stop() error { return nil } func (bp *BridgePlatform) Reply(ctx context.Context, replyCtx any, content string) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context type %T", replyCtx) } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "reply", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "content": content, "format": "text", }) } func (bp *BridgePlatform) Send(ctx context.Context, replyCtx any, content string) error { return bp.Reply(ctx, replyCtx, content) } func (bp *BridgePlatform) ReconstructReplyCtx(sessionKey string) (any, error) { platform := bp.server.platformFromSessionKey(sessionKey) if platform == "" { return nil, fmt.Errorf("bridge: cannot determine adapter from session key %q", sessionKey) } a := bp.server.getAdapter(platform) if a == nil { return nil, fmt.Errorf("bridge: adapter %q not connected", platform) } if !a.capabilities["reconstruct_reply"] { return nil, fmt.Errorf("bridge: adapter %q does not support reconstruct_reply", platform) } replyCtx, err := buildBridgeReconstructReplyCtx(bp.project, sessionKey) if err != nil { return nil, err } return newBridgeReplyCtx(a, sessionKey, replyCtx), nil } func newBridgeReplyCtx(a *bridgeAdapter, sessionKey, replyCtx string) *bridgeReplyCtx { rc := &bridgeReplyCtx{ SessionKey: sessionKey, ReplyCtx: replyCtx, } if a == nil { return rc } rc.Platform = a.platform rc.progressStyle = bridgeProgressStyleForAdapter(a) rc.supportsProgressCardPayload = bridgeSupportsProgressCardPayloadForAdapter(a) return rc } func bridgeProgressStyleForAdapter(a *bridgeAdapter) string { if a == nil { return progressStyleLegacy } if style, ok := bridgeMetadataString(a.metadata, "progress_style"); ok { return normalizeProgressStyle(style) } if a.capabilities["preview"] && a.capabilities["update_message"] { if a.capabilities["card"] { return progressStyleCard } return progressStyleCompact } return progressStyleLegacy } func bridgeSupportsProgressCardPayloadForAdapter(a *bridgeAdapter) bool { if a == nil { return false } if supported, ok := bridgeMetadataBool(a.metadata, "supports_progress_card_payload"); ok { return supported } adapterName, _ := bridgeMetadataString(a.metadata, "adapter") return adapterName == "bot-gateway" && a.capabilities["preview"] && a.capabilities["update_message"] } func bridgeMetadataString(metadata map[string]any, key string) (string, bool) { if metadata == nil { return "", false } raw, ok := metadata[key] if !ok { return "", false } value, ok := raw.(string) if !ok { return "", false } value = strings.TrimSpace(value) if value == "" { return "", false } return value, true } func bridgeMetadataBool(metadata map[string]any, key string) (bool, bool) { if metadata == nil { return false, false } raw, ok := metadata[key] if !ok { return false, false } value, ok := raw.(bool) if !ok { return false, false } return value, true } func buildBridgeReconstructReplyCtx(project, sessionKey string) (string, error) { chatID, err := bridgeTransportChatID(sessionKey) if err != nil { return "", err } payload := bridgeReconstructReplyCtxPayload{ Kind: bridgeReconstructReplyCtxKind, Version: 1, SenderProject: project, TransportChatID: chatID, TransportSessionKey: sessionKey, } data, err := json.Marshal(payload) if err != nil { return "", fmt.Errorf("bridge: marshal reconstruct reply ctx: %w", err) } return string(data), nil } func bridgeTransportChatID(sessionKey string) (string, error) { parts := strings.SplitN(sessionKey, ":", 3) if len(parts) < 2 || parts[1] == "" { return "", fmt.Errorf("bridge: invalid session key %q", sessionKey) } return parts[1], nil } func (bp *BridgePlatform) SendCard(ctx context.Context, replyCtx any, card *Card) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["card"] { return bp.Reply(ctx, replyCtx, card.RenderText()) } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "card", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "card": serializeCard(card), }) } func (bp *BridgePlatform) ReplyCard(ctx context.Context, replyCtx any, card *Card) error { return bp.SendCard(ctx, replyCtx, card) } func (bp *BridgePlatform) SendWithButtons(ctx context.Context, replyCtx any, content string, buttons [][]ButtonOption) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["buttons"] { return bp.Reply(ctx, replyCtx, content) } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "buttons", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "content": content, "buttons": buttons, }) } func (bp *BridgePlatform) UpdateMessage(ctx context.Context, replyCtx any, content string) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["update_message"] { return ErrNotSupported } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "update_message", "session_key": rc.SessionKey, "preview_handle": rc.ReplyCtx, "content": content, }) } func (bp *BridgePlatform) SendPreviewStart(ctx context.Context, replyCtx any, content string) (previewHandle any, err error) { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return nil, fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["preview"] { return nil, ErrNotSupported } refID := fmt.Sprintf("prev-%d", time.Now().UnixNano()) ch := make(chan string, 1) a.previewMu.Lock() a.previewRequests[refID] = ch a.previewMu.Unlock() if err := bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "preview_start", "ref_id": refID, "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "content": content, }); err != nil { a.previewMu.Lock() delete(a.previewRequests, refID) a.previewMu.Unlock() return nil, err } select { case handle := <-ch: return newBridgeReplyCtx(a, rc.SessionKey, handle), nil case <-time.After(10 * time.Second): a.previewMu.Lock() delete(a.previewRequests, refID) a.previewMu.Unlock() return nil, fmt.Errorf("bridge: preview_ack timeout") case <-ctx.Done(): a.previewMu.Lock() delete(a.previewRequests, refID) a.previewMu.Unlock() return nil, ctx.Err() } } func (bp *BridgePlatform) DeletePreviewMessage(ctx context.Context, previewHandle any) error { rc, ok := previewHandle.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid preview handle") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["delete_message"] { return ErrNotSupported } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "delete_message", "session_key": rc.SessionKey, "preview_handle": rc.ReplyCtx, }) } func (bp *BridgePlatform) StartTyping(ctx context.Context, replyCtx any) (stop func()) { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return func() {} } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["typing"] { return func() {} } _ = bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "typing_start", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, }) return func() { _ = bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "typing_stop", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, }) } } func (bp *BridgePlatform) SendAudio(ctx context.Context, replyCtx any, audio []byte, format string) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["audio"] { return ErrNotSupported } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "audio", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "data": base64.StdEncoding.EncodeToString(audio), "format": format, }) } func (bp *BridgePlatform) SendImage(ctx context.Context, replyCtx any, img ImageAttachment) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["image"] { return ErrNotSupported } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "image", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "data": base64.StdEncoding.EncodeToString(img.Data), "mime_type": img.MimeType, "file_name": img.FileName, }) } func (bp *BridgePlatform) SendFile(ctx context.Context, replyCtx any, file FileAttachment) error { rc, ok := replyCtx.(*bridgeReplyCtx) if !ok { return fmt.Errorf("bridge: invalid reply context") } a := bp.server.getAdapter(rc.Platform) if a == nil || !a.capabilities["file"] { return ErrNotSupported } return bp.server.sendToAdapter(rc.Platform, map[string]any{ "type": "file", "session_key": rc.SessionKey, "reply_ctx": rc.ReplyCtx, "data": base64.StdEncoding.EncodeToString(file.Data), "mime_type": file.MimeType, "file_name": file.FileName, }) } func (bp *BridgePlatform) SetCardNavigationHandler(h CardNavigationHandler) { bp.navHandler = h } // --------------------------------------------------------------------------- // WebSocket connection handling (on BridgeServer) // --------------------------------------------------------------------------- // checkOrigin validates the WebSocket origin against CORS origins. // In insecure mode, it allows all origins. Otherwise, it checks against CORS origins or same host. func (bs *BridgeServer) checkOrigin(r *http.Request) bool { // In insecure mode, allow all origins (for local development) if bs.insecure { return true } origin := r.Header.Get("Origin") if origin == "" { // No origin header (e.g., non-browser client) - allow only if authenticated // The authentication check happens before this, so we allow return true } // If CORS origins are configured, check against them if len(bs.corsOrigins) > 0 { for _, o := range bs.corsOrigins { if o == "*" || o == origin { return true } } slog.Warn("bridge: websocket origin rejected", "origin", origin, "allowed", bs.corsOrigins) return false } // No CORS configured - require same-host (origin must match host) host := r.Host if host == "" { host = r.URL.Host } // Parse origin to get host if idx := strings.Index(origin, "://"); idx > 0 { originHost := origin[idx+3:] if originHost == host { return true } } slog.Warn("bridge: websocket origin mismatch", "origin", origin, "host", host) return false } func (bs *BridgeServer) handleWS(w http.ResponseWriter, r *http.Request) { if !bs.authenticate(r) { http.Error(w, "unauthorized", http.StatusUnauthorized) return } // Use a custom upgrader with origin checking upgrader := websocket.Upgrader{ CheckOrigin: bs.checkOrigin, } conn, err := upgrader.Upgrade(w, r, nil) if err != nil { slog.Error("bridge: websocket upgrade failed", "error", err) return } slog.Info("bridge: new connection", "remote", conn.RemoteAddr()) bs.handleConnection(conn) } func (bs *BridgeServer) handleConnection(conn *websocket.Conn) { defer conn.Close() if err := conn.SetReadDeadline(time.Now().Add(90 * time.Second)); err != nil { slog.Debug("bridge: set read deadline failed", "error", err) return } conn.SetPongHandler(func(string) error { return conn.SetReadDeadline(time.Now().Add(90 * time.Second)) }) // First message must be "register" _, raw, err := conn.ReadMessage() if err != nil { slog.Error("bridge: read register failed", "error", err) return } var reg bridgeRegister if err := json.Unmarshal(raw, ®); err != nil || reg.Type != "register" { if err := writeJSON(conn, nil, map[string]any{"type": "register_ack", "ok": false, "error": "first message must be register"}); err != nil { slog.Debug("bridge: write register ack failed", "error", err) } return } if reg.Platform == "" { if err := writeJSON(conn, nil, map[string]any{"type": "register_ack", "ok": false, "error": "platform name is required"}); err != nil { slog.Debug("bridge: write register ack failed", "error", err) } return } caps := make(map[string]bool, len(reg.Capabilities)) for _, c := range reg.Capabilities { caps[c] = true } caps["text"] = true adapter := &bridgeAdapter{ platform: reg.Platform, capabilities: caps, metadata: reg.Metadata, conn: conn, server: bs, previewRequests: make(map[string]chan string), } bs.mu.Lock() if old, exists := bs.adapters[reg.Platform]; exists { old.conn.Close() slog.Info("bridge: replaced existing adapter", "platform", reg.Platform) } bs.adapters[reg.Platform] = adapter bs.mu.Unlock() if err := writeJSON(conn, &adapter.writeMu, map[string]any{"type": "register_ack", "ok": true}); err != nil { slog.Debug("bridge: write register ack failed", "error", err) return } if bridgeMetadataStringListContains(reg.Metadata, "control_plane", bridgeCapabilitiesSnapshotProto) { if err := writeJSON(conn, &adapter.writeMu, bs.buildCapabilitiesSnapshot()); err != nil { slog.Debug("bridge: write capabilities snapshot failed", "platform", reg.Platform, "error", err) return } } slog.Info("bridge: adapter registered", "platform", reg.Platform, "capabilities", reg.Capabilities) defer func() { bs.mu.Lock() if bs.adapters[reg.Platform] == adapter { delete(bs.adapters, reg.Platform) } bs.mu.Unlock() slog.Info("bridge: adapter disconnected", "platform", reg.Platform) }() for { if err := conn.SetReadDeadline(time.Now().Add(90 * time.Second)); err != nil { slog.Debug("bridge: set read deadline failed", "platform", reg.Platform, "error", err) return } _, raw, err := conn.ReadMessage() if err != nil { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { slog.Debug("bridge: read error", "platform", reg.Platform, "error", err) } return } var base bridgeMsg if err := json.Unmarshal(raw, &base); err != nil { slog.Debug("bridge: invalid JSON", "platform", reg.Platform, "error", err) continue } switch base.Type { case "message": adapter.handleMessage(raw) case "card_action": adapter.handleCardAction(raw) case "preview_ack": adapter.handlePreviewAck(raw) case "ping": if err := writeJSON(conn, &adapter.writeMu, map[string]any{"type": "pong", "ts": time.Now().UnixMilli()}); err != nil { slog.Debug("bridge: write pong failed", "platform", reg.Platform, "error", err) return } default: slog.Debug("bridge: unknown message type", "platform", reg.Platform, "type", base.Type) } } } // --------------------------------------------------------------------------- // Adapter message handlers // --------------------------------------------------------------------------- func (a *bridgeAdapter) handleMessage(raw json.RawMessage) { var m bridgeMessage if err := json.Unmarshal(raw, &m); err != nil { slog.Debug("bridge: invalid message payload", "error", err) return } if m.SessionKey == "" || m.UserID == "" { slog.Debug("bridge: message missing required fields", "platform", a.platform) return } ref := a.server.resolveEngine(m.SessionKey, m.Project) if ref == nil { slog.Warn("bridge: no engine for session", "platform", a.platform, "session_key", m.SessionKey, "project", m.Project) return } msg := &Message{ SessionKey: m.SessionKey, Platform: a.platform, MessageID: m.MsgID, UserID: m.UserID, UserName: m.UserName, Content: m.Content, ReplyCtx: newBridgeReplyCtx(a, m.SessionKey, m.ReplyCtx), } for _, img := range m.Images { data, err := base64.StdEncoding.DecodeString(img.Data) if err != nil { slog.Debug("bridge: invalid image base64", "error", err) continue } msg.Images = append(msg.Images, ImageAttachment{ MimeType: img.MimeType, Data: data, FileName: img.FileName, }) } for _, f := range m.Files { data, err := base64.StdEncoding.DecodeString(f.Data) if err != nil { slog.Debug("bridge: invalid file base64", "error", err) continue } msg.Files = append(msg.Files, FileAttachment{ MimeType: f.MimeType, Data: data, FileName: f.FileName, }) } if m.Audio != nil { if data, err := base64.StdEncoding.DecodeString(m.Audio.Data); err == nil { msg.Audio = &AudioAttachment{ MimeType: m.Audio.MimeType, Data: data, Format: m.Audio.Format, Duration: m.Audio.Duration, } } } slog.Info("bridge: message received", "platform", a.platform, "session_key", m.SessionKey, "user", m.UserID, "content_len", len(m.Content), ) if ref.platform.handler != nil { ref.platform.handler(ref.platform, msg) } } func (a *bridgeAdapter) handleCardAction(raw json.RawMessage) { var ca bridgeCardAction if err := json.Unmarshal(raw, &ca); err != nil { slog.Debug("bridge: invalid card_action payload", "error", err) return } slog.Debug("bridge: card_action", "platform", a.platform, "action", ca.Action, "session_key", ca.SessionKey, "project", ca.Project) ref := a.server.resolveEngine(ca.SessionKey, ca.Project) if ref == nil { return } // perm: — permission response; convert to a regular message for the engine if strings.HasPrefix(ca.Action, "perm:") { var responseText string switch ca.Action { case "perm:allow": responseText = "allow" case "perm:deny": responseText = "deny" case "perm:allow_all": responseText = "allow all" default: return } a.dispatchAsMessage(ref, ca.SessionKey, ca.ReplyCtx, responseText) return } // askq: — AskUserQuestion answer; forward as a regular message if strings.HasPrefix(ca.Action, "askq:") { a.dispatchAsMessage(ref, ca.SessionKey, ca.ReplyCtx, ca.Action) return } // cmd: — command shortcut from a card button; forward as a message if strings.HasPrefix(ca.Action, "cmd:") { cmdText := strings.TrimPrefix(ca.Action, "cmd:") a.dispatchAsMessage(ref, ca.SessionKey, ca.ReplyCtx, cmdText) return } // nav: / act: — card navigation and in-place updates if ref.platform.navHandler == nil { return } card := ref.platform.navHandler(ca.Action, ca.SessionKey) if card == nil { return } if a.capabilities["card"] { _ = a.server.sendToAdapter(a.platform, map[string]any{ "type": "card", "session_key": ca.SessionKey, "reply_ctx": ca.ReplyCtx, "card": serializeCard(card), }) } else { rc := newBridgeReplyCtx(a, ca.SessionKey, ca.ReplyCtx) _ = ref.platform.Reply(context.Background(), rc, card.RenderText()) } } // dispatchAsMessage converts a card action into a regular user message // and dispatches it to the engine's message handler. func (a *bridgeAdapter) dispatchAsMessage(ref *bridgeEngineRef, sessionKey, replyCtx, content string) { if ref.platform.handler == nil { return } msg := &Message{ SessionKey: sessionKey, Platform: a.platform, UserID: "web-admin", UserName: "Web Admin", Content: content, ReplyCtx: newBridgeReplyCtx(a, sessionKey, replyCtx), } go ref.platform.handler(ref.platform, msg) } func (a *bridgeAdapter) handlePreviewAck(raw json.RawMessage) { var ack bridgePreviewAck if err := json.Unmarshal(raw, &ack); err != nil { return } a.previewMu.Lock() ch, ok := a.previewRequests[ack.RefID] if ok { delete(a.previewRequests, ack.RefID) } a.previewMu.Unlock() if ok { ch <- ack.PreviewHandle } } // --------------------------------------------------------------------------- // Session management REST API (on BridgeServer) // --------------------------------------------------------------------------- // authHTTP wraps an HTTP handler with token authentication. func (bs *BridgeServer) authHTTP(handler http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if !bs.authenticate(r) { bridgeError(w, http.StatusUnauthorized, "unauthorized") return } handler(w, r) } } func bridgeJSON(w http.ResponseWriter, status int, data any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) if err := json.NewEncoder(w).Encode(map[string]any{"ok": true, "data": data}); err != nil { slog.Debug("bridge: write JSON failed", "error", err) } } func bridgeError(w http.ResponseWriter, status int, msg string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) if err := json.NewEncoder(w).Encode(map[string]any{"ok": false, "error": msg}); err != nil { slog.Debug("bridge: write JSON failed", "error", err) } } // resolveEngineForSessionKey returns the engine ref for a given session key and optional project. func (bs *BridgeServer) resolveEngineForSessionKey(sessionKey, project string) *bridgeEngineRef { return bs.resolveEngine(sessionKey, project) } // handleSessions handles GET /bridge/sessions and POST /bridge/sessions. func (bs *BridgeServer) handleSessions(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: sessionKey := r.URL.Query().Get("session_key") if sessionKey == "" { bridgeError(w, http.StatusBadRequest, "session_key query parameter is required") return } project := r.URL.Query().Get("project") ref := bs.resolveEngineForSessionKey(sessionKey, project) if ref == nil { bridgeError(w, http.StatusNotFound, "no engine found for session key") return } sessions := ref.engine.sessions.ListSessions(sessionKey) activeID := ref.engine.sessions.ActiveSessionID(sessionKey) list := make([]map[string]any, len(sessions)) for i, s := range sessions { list[i] = map[string]any{ "id": s.ID, "name": s.GetName(), "history_count": len(s.History), } } bridgeJSON(w, http.StatusOK, map[string]any{ "sessions": list, "active_session_id": activeID, }) case http.MethodPost: var body struct { SessionKey string `json:"session_key"` Name string `json:"name"` Project string `json:"project,omitempty"` } if err := json.NewDecoder(r.Body).Decode(&body); err != nil { bridgeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) return } if body.SessionKey == "" { bridgeError(w, http.StatusBadRequest, "session_key is required") return } ref := bs.resolveEngineForSessionKey(body.SessionKey, body.Project) if ref == nil { bridgeError(w, http.StatusNotFound, "no engine found for session key") return } name := body.Name if name == "" { name = "default" } s := ref.engine.sessions.NewSession(body.SessionKey, name) bridgeJSON(w, http.StatusOK, map[string]any{ "id": s.ID, "name": s.GetName(), "message": "session created", }) default: bridgeError(w, http.StatusMethodNotAllowed, "GET or POST only") } } // handleSessionRoutes dispatches /bridge/sessions/{sub} routes. func (bs *BridgeServer) handleSessionRoutes(w http.ResponseWriter, r *http.Request) { sub := strings.TrimPrefix(r.URL.Path, "/bridge/sessions/") if sub == "" { bridgeError(w, http.StatusBadRequest, "session id required") return } // POST /bridge/sessions/switch if sub == "switch" { bs.handleSessionSwitch(w, r) return } // GET or DELETE /bridge/sessions/{id} sessionKey := r.URL.Query().Get("session_key") if sessionKey == "" { bridgeError(w, http.StatusBadRequest, "session_key query parameter is required") return } project := r.URL.Query().Get("project") ref := bs.resolveEngineForSessionKey(sessionKey, project) if ref == nil { bridgeError(w, http.StatusNotFound, "no engine found for session key") return } switch r.Method { case http.MethodGet: s := ref.engine.sessions.FindByID(sub) if s == nil { bridgeError(w, http.StatusNotFound, "session not found") return } histLimit := 50 if v := r.URL.Query().Get("history_limit"); v != "" { if n, err := strconv.Atoi(v); err == nil && n > 0 { histLimit = n } } hist := s.GetHistory(histLimit) histJSON := make([]map[string]any, len(hist)) for i, h := range hist { histJSON[i] = map[string]any{ "role": h.Role, "content": h.Content, } } bridgeJSON(w, http.StatusOK, map[string]any{ "id": s.ID, "name": s.GetName(), "history": histJSON, }) case http.MethodDelete: if ref.engine.sessions.DeleteByID(sub) { bridgeJSON(w, http.StatusOK, map[string]string{"message": "session deleted"}) } else { bridgeError(w, http.StatusNotFound, "session not found") } default: bridgeError(w, http.StatusMethodNotAllowed, "GET or DELETE only") } } // handleSessionSwitch handles POST /bridge/sessions/switch. func (bs *BridgeServer) handleSessionSwitch(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { bridgeError(w, http.StatusMethodNotAllowed, "POST only") return } var body struct { SessionKey string `json:"session_key"` Target string `json:"target"` Project string `json:"project,omitempty"` } if err := json.NewDecoder(r.Body).Decode(&body); err != nil { bridgeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) return } if body.SessionKey == "" || body.Target == "" { bridgeError(w, http.StatusBadRequest, "session_key and target are required") return } ref := bs.resolveEngineForSessionKey(body.SessionKey, body.Project) if ref == nil { bridgeError(w, http.StatusNotFound, "no engine found for session key") return } s, err := ref.engine.sessions.SwitchSession(body.SessionKey, body.Target) if err != nil { bridgeError(w, http.StatusNotFound, err.Error()) return } bridgeJSON(w, http.StatusOK, map[string]any{ "message": "session switched", "active_session_id": s.ID, }) } // --------------------------------------------------------------------------- // Internal helpers (on BridgeServer) // --------------------------------------------------------------------------- func (bs *BridgeServer) authenticate(r *http.Request) bool { // If token is not set, only allow in insecure mode if bs.token == "" { return bs.insecure } if auth := r.Header.Get("Authorization"); auth != "" { if strings.HasPrefix(auth, "Bearer ") { return subtle.ConstantTimeCompare([]byte(auth[7:]), []byte(bs.token)) == 1 } } if tok := r.Header.Get("X-Bridge-Token"); tok != "" { return subtle.ConstantTimeCompare([]byte(tok), []byte(bs.token)) == 1 } if tok := r.URL.Query().Get("token"); tok != "" { return subtle.ConstantTimeCompare([]byte(tok), []byte(bs.token)) == 1 } return false } func (bs *BridgeServer) getAdapter(platform string) *bridgeAdapter { bs.mu.RLock() defer bs.mu.RUnlock() return bs.adapters[platform] } func (bs *BridgeServer) sendToAdapter(platform string, msg map[string]any) error { bs.mu.RLock() a, ok := bs.adapters[platform] bs.mu.RUnlock() if !ok { return fmt.Errorf("bridge: adapter %q not connected", platform) } return writeJSON(a.conn, &a.writeMu, msg) } func bridgeMetadataStringListContains(metadata map[string]any, key, want string) bool { if metadata == nil || key == "" || want == "" { return false } raw, ok := metadata[key] if !ok { return false } items, ok := raw.([]any) if !ok { if stringsList, ok := raw.([]string); ok { for _, item := range stringsList { if strings.TrimSpace(item) == want { return true } } } return false } for _, item := range items { if s, ok := item.(string); ok && strings.TrimSpace(s) == want { return true } } return false } func (bs *BridgeServer) platformFromSessionKey(sessionKey string) string { if idx := strings.Index(sessionKey, ":"); idx > 0 { candidate := sessionKey[:idx] bs.mu.RLock() _, ok := bs.adapters[candidate] bs.mu.RUnlock() if ok { return candidate } } return "" } // resolveEngine finds the engine to handle a message. // It first tries to match by project name, then by session_key ownership, // and finally falls back to the single-engine case. func (bs *BridgeServer) resolveEngine(sessionKey, project string) *bridgeEngineRef { bs.enginesMu.RLock() defer bs.enginesMu.RUnlock() if project != "" { if ref, ok := bs.engines[project]; ok { return ref } } if len(bs.engines) == 1 { for _, ref := range bs.engines { return ref } } // Try to find the engine that owns sessions for this key. for _, ref := range bs.engines { if sessions := ref.engine.sessions.ListSessions(sessionKey); len(sessions) > 0 { return ref } } return nil } func writeJSON(conn *websocket.Conn, mu *sync.Mutex, v any) error { if mu != nil { mu.Lock() defer mu.Unlock() } return conn.WriteJSON(v) } // serializeCard converts a Card into a JSON-friendly map for the bridge protocol. func serializeCard(c *Card) map[string]any { result := make(map[string]any) if c.Header != nil { result["header"] = map[string]string{ "title": c.Header.Title, "color": c.Header.Color, } } var elements []map[string]any for _, elem := range c.Elements { switch e := elem.(type) { case CardMarkdown: elements = append(elements, map[string]any{"type": "markdown", "content": e.Content}) case CardDivider: elements = append(elements, map[string]any{"type": "divider"}) case CardActions: var btns []map[string]any for _, b := range e.Buttons { btns = append(btns, map[string]any{ "text": b.Text, "btn_type": b.Type, "value": b.Value, }) } elements = append(elements, map[string]any{ "type": "actions", "buttons": btns, "layout": string(e.Layout), }) case CardNote: m := map[string]any{"type": "note", "text": e.Text} if e.Tag != "" { m["tag"] = e.Tag } elements = append(elements, m) case CardListItem: elements = append(elements, map[string]any{ "type": "list_item", "text": e.Text, "btn_text": e.BtnText, "btn_type": e.BtnType, "btn_value": e.BtnValue, }) case CardSelect: var opts []map[string]string for _, o := range e.Options { opts = append(opts, map[string]string{"text": o.Text, "value": o.Value}) } elements = append(elements, map[string]any{ "type": "select", "placeholder": e.Placeholder, "options": opts, "init_value": e.InitValue, }) } } result["elements"] = elements return result }