gmessages/libgm/session_handler.go

204 lines
5.1 KiB
Go
Raw Normal View History

2023-06-30 11:05:33 +00:00
package libgm
2023-06-30 09:54:08 +00:00
import (
"fmt"
2023-07-15 12:57:07 +00:00
"sync"
2023-06-30 09:54:08 +00:00
"time"
2023-07-16 10:23:44 +00:00
"github.com/google/uuid"
2023-06-30 09:54:08 +00:00
"golang.org/x/exp/slices"
"google.golang.org/protobuf/proto"
"go.mau.fi/mautrix-gmessages/libgm/pblite"
2023-06-30 09:54:08 +00:00
2023-07-17 13:51:31 +00:00
"go.mau.fi/mautrix-gmessages/libgm/gmproto"
"go.mau.fi/mautrix-gmessages/libgm/routes"
2023-06-30 09:54:08 +00:00
"go.mau.fi/mautrix-gmessages/libgm/util"
)
2023-07-15 22:45:57 +00:00
type SessionHandler struct {
client *Client
2023-06-30 09:54:08 +00:00
responseWaiters map[string]chan<- *IncomingRPCMessage
2023-07-15 22:45:57 +00:00
responseWaitersLock sync.Mutex
2023-06-30 09:54:08 +00:00
2023-07-15 12:57:07 +00:00
ackMapLock sync.Mutex
ackMap []string
ackTicker *time.Ticker
2023-06-30 09:54:08 +00:00
2023-07-15 22:45:57 +00:00
sessionID string
2023-06-30 09:54:08 +00:00
responseTimeout time.Duration
}
2023-07-15 22:45:57 +00:00
func (s *SessionHandler) ResetSessionID() {
2023-07-16 10:23:44 +00:00
s.sessionID = uuid.NewString()
2023-06-30 09:54:08 +00:00
}
2023-07-17 13:51:31 +00:00
func (s *SessionHandler) sendMessageNoResponse(actionType gmproto.ActionType, encryptedData proto.Message) error {
2023-07-18 00:08:01 +00:00
_, payload, err := s.buildMessage(actionType, encryptedData)
2023-07-15 22:45:57 +00:00
if err != nil {
return err
}
2023-07-15 23:21:53 +00:00
_, err = s.client.rpc.sendMessageRequest(util.SendMessageURL, payload)
2023-07-15 22:45:57 +00:00
return err
2023-06-30 09:54:08 +00:00
}
func (s *SessionHandler) sendAsyncMessage(actionType gmproto.ActionType, encryptedData proto.Message) (<-chan *IncomingRPCMessage, error) {
2023-07-18 00:08:01 +00:00
requestID, payload, err := s.buildMessage(actionType, encryptedData)
if err != nil {
return nil, err
2023-06-30 09:54:08 +00:00
}
2023-07-15 22:45:57 +00:00
ch := s.waitResponse(requestID)
2023-07-15 23:21:53 +00:00
_, reqErr := s.client.rpc.sendMessageRequest(util.SendMessageURL, payload)
if reqErr != nil {
2023-07-15 22:45:57 +00:00
s.cancelResponse(requestID, ch)
return nil, reqErr
}
return ch, nil
}
2023-07-17 23:11:43 +00:00
func typedResponse[T proto.Message](resp *IncomingRPCMessage, err error) (casted T, retErr error) {
if err != nil {
retErr = err
return
}
var ok bool
casted, ok = resp.DecryptedMessage.(T)
if !ok {
retErr = fmt.Errorf("unexpected response type %T, expected %T", resp.DecryptedMessage, casted)
}
return
}
func (s *SessionHandler) sendMessage(actionType gmproto.ActionType, encryptedData proto.Message) (*IncomingRPCMessage, error) {
2023-07-15 22:45:57 +00:00
ch, err := s.sendAsyncMessage(actionType, encryptedData)
if err != nil {
return nil, err
2023-06-30 09:54:08 +00:00
}
2023-07-15 22:45:57 +00:00
// TODO add timeout
return <-ch, nil
}
2023-06-30 09:54:08 +00:00
2023-07-18 00:08:01 +00:00
func (s *SessionHandler) buildMessage(actionType gmproto.ActionType, data proto.Message) (string, []byte, error) {
2023-07-15 22:45:57 +00:00
var requestID string
2023-07-18 00:08:01 +00:00
var err error
sessionID := s.client.sessionHandler.sessionID
2023-06-30 09:54:08 +00:00
routeInfo, ok := routes.Routes[actionType]
if !ok {
2023-07-18 00:08:01 +00:00
return "", nil, fmt.Errorf("failed to build message: could not find route %d", actionType)
2023-06-30 09:54:08 +00:00
}
if routeInfo.UseSessionID {
2023-07-15 22:45:57 +00:00
requestID = s.sessionID
} else {
2023-07-16 10:23:44 +00:00
requestID = uuid.NewString()
2023-06-30 09:54:08 +00:00
}
2023-07-18 00:08:01 +00:00
message := &gmproto.OutgoingRPCMessage{
Mobile: s.client.AuthData.Mobile,
Data: &gmproto.OutgoingRPCMessage_Data{
RequestID: requestID,
BugleRoute: routeInfo.BugleRoute,
MessageTypeData: &gmproto.OutgoingRPCMessage_Data_Type{
EmptyArr: &gmproto.EmptyArr{},
MessageType: routeInfo.MessageType,
},
},
Auth: &gmproto.OutgoingRPCMessage_Auth{
RequestID: requestID,
TachyonAuthToken: s.client.AuthData.TachyonAuthToken,
ConfigVersion: util.ConfigMessage,
},
EmptyArr: &gmproto.EmptyArr{},
2023-06-30 09:54:08 +00:00
}
2023-07-18 00:08:01 +00:00
var encryptedData []byte
if data != nil {
var serializedData []byte
serializedData, err = proto.Marshal(data)
if err != nil {
return "", nil, err
}
encryptedData, err = s.client.AuthData.RequestCrypto.Encrypt(serializedData)
if err != nil {
return "", nil, err
}
2023-06-30 09:54:08 +00:00
}
2023-07-18 00:08:01 +00:00
message.Data.MessageData, err = proto.Marshal(&gmproto.OutgoingRPCData{
RequestID: requestID,
Action: actionType,
EncryptedProtoData: encryptedData,
SessionID: sessionID,
})
if err != nil {
return "", nil, err
2023-06-30 09:54:08 +00:00
}
2023-07-18 00:08:01 +00:00
var marshaledMessage []byte
marshaledMessage, err = pblite.Marshal(message)
return requestID, marshaledMessage, err
2023-06-30 09:54:08 +00:00
}
2023-07-15 22:45:57 +00:00
func (s *SessionHandler) queueMessageAck(messageID string) {
2023-07-15 12:57:07 +00:00
s.ackMapLock.Lock()
defer s.ackMapLock.Unlock()
2023-07-15 22:45:57 +00:00
if !slices.Contains(s.ackMap, messageID) {
s.ackMap = append(s.ackMap, messageID)
s.client.Logger.Trace().Any("message_id", messageID).Msg("Queued ack for message")
} else {
s.client.Logger.Trace().Any("message_id", messageID).Msg("Ack for message was already queued")
2023-06-30 09:54:08 +00:00
}
}
func (s *SessionHandler) startAckInterval() {
if s.ackTicker != nil {
s.ackTicker.Stop()
}
ticker := time.NewTicker(5 * time.Second)
s.ackTicker = ticker
go func() {
for range ticker.C {
s.sendAckRequest()
}
}()
}
func (s *SessionHandler) sendAckRequest() {
2023-07-15 12:57:07 +00:00
s.ackMapLock.Lock()
dataToAck := s.ackMap
s.ackMap = nil
2023-07-15 12:57:07 +00:00
s.ackMapLock.Unlock()
if len(dataToAck) == 0 {
return
}
2023-07-17 23:57:20 +00:00
ackMessages := make([]*gmproto.AckMessageRequest_Message, len(dataToAck))
for i, reqID := range dataToAck {
2023-07-17 23:57:20 +00:00
ackMessages[i] = &gmproto.AckMessageRequest_Message{
RequestID: reqID,
2023-07-16 12:55:30 +00:00
Device: s.client.AuthData.Browser,
}
}
2023-07-17 23:57:20 +00:00
ackMessagePayload := &gmproto.AckMessageRequest{
2023-07-17 13:51:31 +00:00
AuthData: &gmproto.AuthMessage{
2023-07-16 10:23:44 +00:00
RequestID: uuid.NewString(),
2023-07-16 12:55:30 +00:00
TachyonAuthToken: s.client.AuthData.TachyonAuthToken,
ConfigVersion: util.ConfigMessage,
2023-06-30 09:54:08 +00:00
},
2023-07-17 13:51:31 +00:00
EmptyArr: &gmproto.EmptyArr{},
Acks: ackMessages,
2023-06-30 09:54:08 +00:00
}
jsonData, err := pblite.Marshal(ackMessagePayload)
2023-06-30 09:54:08 +00:00
if err != nil {
2023-07-09 20:32:19 +00:00
panic(err)
2023-06-30 09:54:08 +00:00
}
2023-07-15 23:21:53 +00:00
_, err = s.client.rpc.sendMessageRequest(util.AckMessagesURL, jsonData)
2023-06-30 09:54:08 +00:00
if err != nil {
2023-07-09 20:32:19 +00:00
panic(err)
2023-06-30 09:54:08 +00:00
}
2023-07-15 15:49:28 +00:00
s.client.Logger.Debug().Strs("message_ids", dataToAck).Msg("Sent acks")
2023-06-30 09:54:08 +00:00
}