// Copyright (c) 2023 Tulir Asokan // // This Source Code Form is subject to the terms of the Mozilla Public // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at http://mozilla.org/MPL/2.0/. package appservice import ( "context" "encoding/json" "errors" "fmt" "net/http" "net/url" "path/filepath" "strings" "sync" "sync/atomic" "time" "github.com/gorilla/websocket" "github.com/rs/zerolog" "github.com/tidwall/gjson" "github.com/tidwall/sjson" ) type WebsocketRequest struct { ReqID int `json:"id,omitempty"` Command string `json:"command"` Data interface{} `json:"data"` Deadline time.Duration `json:"-"` } type WebsocketCommand struct { ReqID int `json:"id,omitempty"` Command string `json:"command"` Data json.RawMessage `json:"data"` Ctx context.Context `json:"-"` } func (wsc *WebsocketCommand) MakeResponse(ok bool, data interface{}) *WebsocketRequest { if wsc.ReqID == 0 || wsc.Command == "response" || wsc.Command == "error" { return nil } cmd := "response" if !ok { cmd = "error" } if err, isError := data.(error); isError { var errorData json.RawMessage var jsonErr error unwrappedErr := err var prefixMessage string for unwrappedErr != nil { errorData, jsonErr = json.Marshal(unwrappedErr) if errorData != nil && len(errorData) > 2 && jsonErr == nil { prefixMessage = strings.Replace(err.Error(), unwrappedErr.Error(), "", 1) prefixMessage = strings.TrimRight(prefixMessage, ": ") break } unwrappedErr = errors.Unwrap(unwrappedErr) } if errorData != nil { if !gjson.GetBytes(errorData, "message").Exists() { errorData, _ = sjson.SetBytes(errorData, "message", err.Error()) } // else: marshaled error contains a message already } else { errorData, _ = sjson.SetBytes(nil, "message", err.Error()) } if len(prefixMessage) > 0 { errorData, _ = sjson.SetBytes(errorData, "prefix_message", prefixMessage) } data = errorData } return &WebsocketRequest{ ReqID: wsc.ReqID, Command: cmd, Data: data, } } type WebsocketTransaction struct { Status string `json:"status"` TxnID string `json:"txn_id"` Transaction } type WebsocketTransactionResponse struct { TxnID string `json:"txn_id"` } type WebsocketMessage struct { WebsocketTransaction WebsocketCommand } const ( WebsocketCloseConnReplaced = 4001 WebsocketCloseTxnNotAcknowledged = 4002 ) type MeowWebsocketCloseCode string const ( MeowServerShuttingDown MeowWebsocketCloseCode = "server_shutting_down" MeowConnectionReplaced MeowWebsocketCloseCode = "conn_replaced" MeowTxnNotAcknowledged MeowWebsocketCloseCode = "transactions_not_acknowledged" ) var ( ErrWebsocketManualStop = errors.New("the websocket was disconnected manually") ErrWebsocketOverridden = errors.New("a new call to StartWebsocket overrode the previous connection") ErrWebsocketUnknownError = errors.New("an unknown error occurred") ErrWebsocketNotConnected = errors.New("websocket not connected") ErrWebsocketClosed = errors.New("websocket closed before response received") ) func (mwcc MeowWebsocketCloseCode) String() string { switch mwcc { case MeowServerShuttingDown: return "the server is shutting down" case MeowConnectionReplaced: return "the connection was replaced by another client" case MeowTxnNotAcknowledged: return "transactions were not acknowledged" default: return string(mwcc) } } type CloseCommand struct { Code int `json:"-"` Command string `json:"command"` Status MeowWebsocketCloseCode `json:"status"` } func (cc CloseCommand) Error() string { return fmt.Sprintf("websocket: close %d: %s", cc.Code, cc.Status.String()) } func parseCloseError(err error) error { closeError := &websocket.CloseError{} if !errors.As(err, &closeError) { return err } var closeCommand CloseCommand closeCommand.Code = closeError.Code closeCommand.Command = "disconnect" if len(closeError.Text) > 0 { jsonErr := json.Unmarshal([]byte(closeError.Text), &closeCommand) if jsonErr != nil { return err } } if len(closeCommand.Status) == 0 { if closeCommand.Code == WebsocketCloseConnReplaced { closeCommand.Status = MeowConnectionReplaced } else if closeCommand.Code == websocket.CloseServiceRestart { closeCommand.Status = MeowServerShuttingDown } } return &closeCommand } func (as *AppService) HasWebsocket() bool { return as.ws != nil } func (as *AppService) SendWebsocket(cmd *WebsocketRequest) error { ws := as.ws if cmd == nil { return nil } else if ws == nil { return ErrWebsocketNotConnected } as.wsWriteLock.Lock() defer as.wsWriteLock.Unlock() if cmd.Deadline == 0 { cmd.Deadline = 3 * time.Minute } _ = ws.SetWriteDeadline(time.Now().Add(cmd.Deadline)) return ws.WriteJSON(cmd) } func (as *AppService) clearWebsocketResponseWaiters() { as.websocketRequestsLock.Lock() for _, waiter := range as.websocketRequests { waiter <- &WebsocketCommand{Command: "__websocket_closed"} } as.websocketRequests = make(map[int]chan<- *WebsocketCommand) as.websocketRequestsLock.Unlock() } func (as *AppService) addWebsocketResponseWaiter(reqID int, waiter chan<- *WebsocketCommand) { as.websocketRequestsLock.Lock() as.websocketRequests[reqID] = waiter as.websocketRequestsLock.Unlock() } func (as *AppService) removeWebsocketResponseWaiter(reqID int, waiter chan<- *WebsocketCommand) { as.websocketRequestsLock.Lock() existingWaiter, ok := as.websocketRequests[reqID] if ok && existingWaiter == waiter { delete(as.websocketRequests, reqID) } close(waiter) as.websocketRequestsLock.Unlock() } type ErrorResponse struct { Code string `json:"code"` Message string `json:"message"` } func (er *ErrorResponse) Error() string { return fmt.Sprintf("%s: %s", er.Code, er.Message) } func (as *AppService) RequestWebsocket(ctx context.Context, cmd *WebsocketRequest, response interface{}) error { cmd.ReqID = int(atomic.AddInt32(&as.websocketRequestID, 1)) respChan := make(chan *WebsocketCommand, 1) as.addWebsocketResponseWaiter(cmd.ReqID, respChan) defer as.removeWebsocketResponseWaiter(cmd.ReqID, respChan) err := as.SendWebsocket(cmd) if err != nil { return err } select { case resp := <-respChan: if resp.Command == "__websocket_closed" { return ErrWebsocketClosed } else if resp.Command == "error" { var respErr ErrorResponse err = json.Unmarshal(resp.Data, &respErr) if err != nil { return fmt.Errorf("failed to parse error JSON: %w", err) } return &respErr } else if response != nil { err = json.Unmarshal(resp.Data, &response) if err != nil { return fmt.Errorf("failed to parse response JSON: %w", err) } return nil } else { return nil } case <-ctx.Done(): return ctx.Err() } } func (as *AppService) unknownCommandHandler(cmd WebsocketCommand) (bool, interface{}) { zerolog.Ctx(cmd.Ctx).Warn().Msg("No handler for websocket command") return false, fmt.Errorf("unknown request type") } func (as *AppService) SetWebsocketCommandHandler(cmd string, handler WebsocketHandler) { as.websocketHandlersLock.Lock() as.websocketHandlers[cmd] = handler as.websocketHandlersLock.Unlock() } func (as *AppService) consumeWebsocket(stopFunc func(error), ws *websocket.Conn) { defer stopFunc(ErrWebsocketUnknownError) ctx := context.Background() for { var msg WebsocketMessage err := ws.ReadJSON(&msg) if err != nil { as.Log.Debug().Err(err).Msg("Error reading from websocket") stopFunc(parseCloseError(err)) return } with := as.Log.With(). Int("req_id", msg.ReqID). Str("ws_command", msg.Command) if msg.TxnID != "" { with = with.Str("transaction_id", msg.TxnID) } log := with.Logger() ctx = log.WithContext(ctx) if msg.Command == "" || msg.Command == "transaction" { if msg.TxnID == "" || !as.txnIDC.IsProcessed(msg.TxnID) { as.handleTransaction(ctx, msg.TxnID, &msg.Transaction) } else { log.Debug(). Object("content", &msg.Transaction). Msg("Ignoring duplicate transaction") } go func() { err = as.SendWebsocket(msg.MakeResponse(true, &WebsocketTransactionResponse{TxnID: msg.TxnID})) if err != nil { log.Warn().Err(err).Msg("Failed to send response to websocket transaction") } else { log.Debug().Msg("Sent response to transaction") } }() } else if msg.Command == "connect" { log.Debug().Msg("Websocket connect confirmation received") } else if msg.Command == "response" || msg.Command == "error" { as.websocketRequestsLock.RLock() respChan, ok := as.websocketRequests[msg.ReqID] if ok { select { case respChan <- &msg.WebsocketCommand: default: log.Warn().Msg("Failed to handle response: channel didn't accept response") } } else { log.Warn().Msg("Dropping response to unknown request ID") } as.websocketRequestsLock.RUnlock() } else { log.Debug().Msg("Received websocket command") as.websocketHandlersLock.RLock() handler, ok := as.websocketHandlers[msg.Command] as.websocketHandlersLock.RUnlock() if !ok { handler = as.unknownCommandHandler } go func() { okResp, data := handler(msg.WebsocketCommand) err = as.SendWebsocket(msg.MakeResponse(okResp, data)) if err != nil { log.Error().Err(err).Msg("Failed to send response to websocket command") } else if okResp { log.Debug().Msg("Sent success response to websocket command") } else { log.Debug().Msg("Sent error response to websocket command") } }() } } } func (as *AppService) StartWebsocket(baseURL string, onConnect func()) error { parsed, err := url.Parse(baseURL) if err != nil { return fmt.Errorf("failed to parse URL: %w", err) } parsed.Path = filepath.Join(parsed.Path, "_matrix/client/unstable/fi.mau.as_sync") if parsed.Scheme == "http" { parsed.Scheme = "ws" } else if parsed.Scheme == "https" { parsed.Scheme = "wss" } ws, resp, err := websocket.DefaultDialer.Dial(parsed.String(), http.Header{ "Authorization": []string{fmt.Sprintf("Bearer %s", as.Registration.AppToken)}, "User-Agent": []string{as.BotClient().UserAgent}, "X-Mautrix-Process-ID": []string{as.ProcessID}, "X-Mautrix-Websocket-Version": []string{"3"}, }) if resp != nil && resp.StatusCode >= 400 { var errResp Error err = json.NewDecoder(resp.Body).Decode(&errResp) if err != nil { return fmt.Errorf("websocket request returned HTTP %d with non-JSON body", resp.StatusCode) } else { return fmt.Errorf("websocket request returned %s (HTTP %d): %s", errResp.ErrorCode, resp.StatusCode, errResp.Message) } } else if err != nil { return fmt.Errorf("failed to open websocket: %w", err) } if as.StopWebsocket != nil { as.StopWebsocket(ErrWebsocketOverridden) } closeChan := make(chan error) closeChanOnce := sync.Once{} stopFunc := func(err error) { closeChanOnce.Do(func() { closeChan <- err }) } as.ws = ws as.StopWebsocket = stopFunc as.PrepareWebsocket() as.Log.Debug().Msg("Appservice transaction websocket opened") go as.consumeWebsocket(stopFunc, ws) if onConnect != nil { onConnect() } closeErr := <-closeChan if as.ws == ws { as.clearWebsocketResponseWaiters() as.ws = nil } _ = ws.SetWriteDeadline(time.Now().Add(3 * time.Second)) err = ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseGoingAway, "")) if err != nil && !errors.Is(err, websocket.ErrCloseSent) { as.Log.Warn().Err(err).Msg("Error writing close message to websocket") } err = ws.Close() if err != nil { as.Log.Warn().Err(err).Msg("Error closing websocket") } return closeErr }