// Copyright 2017 Vector Creations Ltd // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package types import ( "encoding/json" "fmt" "strconv" "strings" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/gomatrixserverlib" "github.com/tidwall/gjson" ) var ( // ErrInvalidSyncTokenType is returned when an attempt at creating a // new instance of SyncToken with an invalid type (i.e. neither "s" // nor "t"). ErrInvalidSyncTokenType = fmt.Errorf("Sync token has an unknown prefix (should be either s or t)") // ErrInvalidSyncTokenLen is returned when the pagination token is an // invalid length ErrInvalidSyncTokenLen = fmt.Errorf("Sync token has an invalid length") ) type StateDelta struct { RoomID string StateEvents []*gomatrixserverlib.HeaderedEvent Membership string // The PDU stream position of the latest membership event for this user, if applicable. // Can be 0 if there is no membership event in this delta. MembershipPos StreamPosition } // StreamPosition represents the offset in the sync stream a client is at. type StreamPosition int64 // LogPosition represents the offset in a Kafka log a client is at. type LogPosition struct { Partition int32 Offset int64 } func (p *LogPosition) IsEmpty() bool { return p.Offset == 0 } // IsAfter returns true if this position is after `lp`. func (p *LogPosition) IsAfter(lp *LogPosition) bool { if lp == nil { return false } if p.Partition != lp.Partition { return false } return p.Offset > lp.Offset } // StreamEvent is the same as gomatrixserverlib.Event but also has the PDU stream position for this event. type StreamEvent struct { *gomatrixserverlib.HeaderedEvent StreamPosition StreamPosition TransactionID *api.TransactionID ExcludeFromSync bool } // Range represents a range between two stream positions. type Range struct { // From is the position the client has already received. From StreamPosition // To is the position the client is going towards. To StreamPosition // True if the client is going backwards Backwards bool } // Low returns the low number of the range. // This represents the position the client already has and hence is exclusive. func (r *Range) Low() StreamPosition { if !r.Backwards { return r.From } return r.To } // High returns the high number of the range // This represents the position the client is going towards and hence is inclusive. func (r *Range) High() StreamPosition { if !r.Backwards { return r.To } return r.From } // SyncTokenType represents the type of a sync token. // It can be either "s" (representing a position in the whole stream of events) // or "t" (representing a position in a room's topology/depth). type SyncTokenType string const ( // SyncTokenTypeStream represents a position in the server's whole // stream of events SyncTokenTypeStream SyncTokenType = "s" // SyncTokenTypeTopology represents a position in a room's topology. SyncTokenTypeTopology SyncTokenType = "t" ) type StreamingToken struct { PDUPosition StreamPosition TypingPosition StreamPosition ReceiptPosition StreamPosition SendToDevicePosition StreamPosition InvitePosition StreamPosition AccountDataPosition StreamPosition DeviceListPosition LogPosition } // This will be used as a fallback by json.Marshal. func (s StreamingToken) MarshalText() ([]byte, error) { return []byte(s.String()), nil } // This will be used as a fallback by json.Unmarshal. func (s *StreamingToken) UnmarshalText(text []byte) (err error) { *s, err = NewStreamTokenFromString(string(text)) return err } func (t StreamingToken) String() string { posStr := fmt.Sprintf( "s%d_%d_%d_%d_%d_%d", t.PDUPosition, t.TypingPosition, t.ReceiptPosition, t.SendToDevicePosition, t.InvitePosition, t.AccountDataPosition, ) if dl := t.DeviceListPosition; !dl.IsEmpty() { posStr += fmt.Sprintf(".dl-%d-%d", dl.Partition, dl.Offset) } return posStr } // IsAfter returns true if ANY position in this token is greater than `other`. func (t *StreamingToken) IsAfter(other StreamingToken) bool { switch { case t.PDUPosition > other.PDUPosition: return true case t.TypingPosition > other.TypingPosition: return true case t.ReceiptPosition > other.ReceiptPosition: return true case t.SendToDevicePosition > other.SendToDevicePosition: return true case t.InvitePosition > other.InvitePosition: return true case t.AccountDataPosition > other.AccountDataPosition: return true case t.DeviceListPosition.IsAfter(&other.DeviceListPosition): return true } return false } func (t *StreamingToken) IsEmpty() bool { return t == nil || t.PDUPosition+t.TypingPosition+t.ReceiptPosition+t.SendToDevicePosition+t.InvitePosition+t.AccountDataPosition == 0 && t.DeviceListPosition.IsEmpty() } // WithUpdates returns a copy of the StreamingToken with updates applied from another StreamingToken. // If the latter StreamingToken contains a field that is not 0, it is considered an update, // and its value will replace the corresponding value in the StreamingToken on which WithUpdates is called. // If the other token has a log, they will replace any existing log on this token. func (t *StreamingToken) WithUpdates(other StreamingToken) StreamingToken { ret := *t ret.ApplyUpdates(other) return ret } // ApplyUpdates applies any changes from the supplied StreamingToken. If the supplied // streaming token contains any positions that are not 0, they are considered updates // and will overwrite the value in the token. func (t *StreamingToken) ApplyUpdates(other StreamingToken) { if other.PDUPosition > t.PDUPosition { t.PDUPosition = other.PDUPosition } if other.TypingPosition > t.TypingPosition { t.TypingPosition = other.TypingPosition } if other.ReceiptPosition > t.ReceiptPosition { t.ReceiptPosition = other.ReceiptPosition } if other.SendToDevicePosition > t.SendToDevicePosition { t.SendToDevicePosition = other.SendToDevicePosition } if other.InvitePosition > t.InvitePosition { t.InvitePosition = other.InvitePosition } if other.AccountDataPosition > t.AccountDataPosition { t.AccountDataPosition = other.AccountDataPosition } if other.DeviceListPosition.IsAfter(&t.DeviceListPosition) { t.DeviceListPosition = other.DeviceListPosition } } type TopologyToken struct { Depth StreamPosition PDUPosition StreamPosition } // This will be used as a fallback by json.Marshal. func (t TopologyToken) MarshalText() ([]byte, error) { return []byte(t.String()), nil } // This will be used as a fallback by json.Unmarshal. func (t *TopologyToken) UnmarshalText(text []byte) (err error) { *t, err = NewTopologyTokenFromString(string(text)) return err } func (t *TopologyToken) StreamToken() StreamingToken { return StreamingToken{ PDUPosition: t.PDUPosition, } } func (t TopologyToken) String() string { return fmt.Sprintf("t%d_%d", t.Depth, t.PDUPosition) } // Decrement the topology token to one event earlier. func (t *TopologyToken) Decrement() { depth := t.Depth pduPos := t.PDUPosition if depth-1 <= 0 { // nothing can be lower than this depth = 1 } else { // this assumes that we will never have 1000 events all with the same // depth. TODO: work out what the right PDU position is to use, probably needs a db hit. depth-- pduPos += 1000 } // The lowest token value is 1, therefore we need to manually set it to that // value if we're below it. if depth < 1 { depth = 1 } t.Depth = depth t.PDUPosition = pduPos } func NewTopologyTokenFromString(tok string) (token TopologyToken, err error) { if len(tok) < 1 { err = fmt.Errorf("empty topology token") return } if tok[0] != SyncTokenTypeTopology[0] { err = fmt.Errorf("topology token must start with 't'") return } parts := strings.Split(tok[1:], "_") var positions [2]StreamPosition for i, p := range parts { if i > len(positions) { break } var pos int pos, err = strconv.Atoi(p) if err != nil { return } positions[i] = StreamPosition(pos) } token = TopologyToken{ Depth: positions[0], PDUPosition: positions[1], } return } func NewStreamTokenFromString(tok string) (token StreamingToken, err error) { if len(tok) < 1 { err = fmt.Errorf("empty stream token") return } if tok[0] != SyncTokenTypeStream[0] { err = fmt.Errorf("stream token must start with 's'") return } categories := strings.Split(tok[1:], ".") parts := strings.Split(categories[0], "_") var positions [6]StreamPosition for i, p := range parts { if i > len(positions) { break } var pos int pos, err = strconv.Atoi(p) if err != nil { return } positions[i] = StreamPosition(pos) } token = StreamingToken{ PDUPosition: positions[0], TypingPosition: positions[1], ReceiptPosition: positions[2], SendToDevicePosition: positions[3], InvitePosition: positions[4], AccountDataPosition: positions[5], } // dl-0-1234 // $log_name-$partition-$offset for _, logStr := range categories[1:] { segments := strings.Split(logStr, "-") if len(segments) != 3 { err = fmt.Errorf("invalid log position %q", logStr) return } switch segments[0] { case "dl": // Device list syncing var partition, offset int if partition, err = strconv.Atoi(segments[1]); err != nil { return } if offset, err = strconv.Atoi(segments[2]); err != nil { return } token.DeviceListPosition.Partition = int32(partition) token.DeviceListPosition.Offset = int64(offset) default: err = fmt.Errorf("unrecognised token type %q", segments[0]) return } } return token, nil } // PrevEventRef represents a reference to a previous event in a state event upgrade type PrevEventRef struct { PrevContent json.RawMessage `json:"prev_content"` ReplacesState string `json:"replaces_state"` PrevSender string `json:"prev_sender"` } // Response represents a /sync API response. See https://matrix.org/docs/spec/client_server/r0.2.0.html#get-matrix-client-r0-sync type Response struct { NextBatch StreamingToken `json:"next_batch"` AccountData struct { Events []gomatrixserverlib.ClientEvent `json:"events,omitempty"` } `json:"account_data"` Presence struct { Events []gomatrixserverlib.ClientEvent `json:"events,omitempty"` } `json:"presence"` Rooms struct { Join map[string]JoinResponse `json:"join"` Peek map[string]JoinResponse `json:"peek"` Invite map[string]InviteResponse `json:"invite"` Leave map[string]LeaveResponse `json:"leave"` } `json:"rooms"` ToDevice struct { Events []gomatrixserverlib.SendToDeviceEvent `json:"events,omitempty"` } `json:"to_device"` DeviceLists struct { Changed []string `json:"changed,omitempty"` Left []string `json:"left,omitempty"` } `json:"device_lists"` DeviceListsOTKCount map[string]int `json:"device_one_time_keys_count,omitempty"` } // NewResponse creates an empty response with initialised maps. func NewResponse() *Response { res := Response{} // Pre-initialise the maps. Synapse will return {} even if there are no rooms under a specific section, // so let's do the same thing. Bonus: this means we can't get dreaded 'assignment to entry in nil map' errors. res.Rooms.Join = map[string]JoinResponse{} res.Rooms.Peek = map[string]JoinResponse{} res.Rooms.Invite = map[string]InviteResponse{} res.Rooms.Leave = map[string]LeaveResponse{} // Also pre-intialise empty slices or else we'll insert 'null' instead of '[]' for the value. // TODO: We really shouldn't have to do all this to coerce encoding/json to Do The Right Thing. We should // really be using our own Marshal/Unmarshal implementations otherwise this may prove to be a CPU bottleneck. // This also applies to NewJoinResponse, NewInviteResponse and NewLeaveResponse. res.AccountData.Events = []gomatrixserverlib.ClientEvent{} res.Presence.Events = []gomatrixserverlib.ClientEvent{} res.ToDevice.Events = []gomatrixserverlib.SendToDeviceEvent{} res.DeviceListsOTKCount = map[string]int{} return &res } // IsEmpty returns true if the response is empty, i.e. used to decided whether // to return the response immediately to the client or to wait for more data. func (r *Response) IsEmpty() bool { return len(r.Rooms.Join) == 0 && len(r.Rooms.Invite) == 0 && len(r.Rooms.Leave) == 0 && len(r.AccountData.Events) == 0 && len(r.Presence.Events) == 0 && len(r.ToDevice.Events) == 0 } // JoinResponse represents a /sync response for a room which is under the 'join' or 'peek' key. type JoinResponse struct { State struct { Events []gomatrixserverlib.ClientEvent `json:"events"` } `json:"state"` Timeline struct { Events []gomatrixserverlib.ClientEvent `json:"events"` Limited bool `json:"limited"` PrevBatch *TopologyToken `json:"prev_batch,omitempty"` } `json:"timeline"` Ephemeral struct { Events []gomatrixserverlib.ClientEvent `json:"events"` } `json:"ephemeral"` AccountData struct { Events []gomatrixserverlib.ClientEvent `json:"events"` } `json:"account_data"` } // NewJoinResponse creates an empty response with initialised arrays. func NewJoinResponse() *JoinResponse { res := JoinResponse{} res.State.Events = []gomatrixserverlib.ClientEvent{} res.Timeline.Events = []gomatrixserverlib.ClientEvent{} res.Ephemeral.Events = []gomatrixserverlib.ClientEvent{} res.AccountData.Events = []gomatrixserverlib.ClientEvent{} return &res } // InviteResponse represents a /sync response for a room which is under the 'invite' key. type InviteResponse struct { InviteState struct { Events []json.RawMessage `json:"events"` } `json:"invite_state"` } // NewInviteResponse creates an empty response with initialised arrays. func NewInviteResponse(event *gomatrixserverlib.HeaderedEvent) *InviteResponse { res := InviteResponse{} res.InviteState.Events = []json.RawMessage{} // First see if there's invite_room_state in the unsigned key of the invite. // If there is then unmarshal it into the response. This will contain the // partial room state such as join rules, room name etc. if inviteRoomState := gjson.GetBytes(event.Unsigned(), "invite_room_state"); inviteRoomState.Exists() { _ = json.Unmarshal([]byte(inviteRoomState.Raw), &res.InviteState.Events) } // Then we'll see if we can create a partial of the invite event itself. // This is needed for clients to work out *who* sent the invite. inviteEvent := gomatrixserverlib.ToClientEvent(event.Unwrap(), gomatrixserverlib.FormatSync) inviteEvent.Unsigned = nil if ev, err := json.Marshal(inviteEvent); err == nil { res.InviteState.Events = append(res.InviteState.Events, ev) } return &res } // LeaveResponse represents a /sync response for a room which is under the 'leave' key. type LeaveResponse struct { State struct { Events []gomatrixserverlib.ClientEvent `json:"events"` } `json:"state"` Timeline struct { Events []gomatrixserverlib.ClientEvent `json:"events"` Limited bool `json:"limited"` PrevBatch *TopologyToken `json:"prev_batch,omitempty"` } `json:"timeline"` } // NewLeaveResponse creates an empty response with initialised arrays. func NewLeaveResponse() *LeaveResponse { res := LeaveResponse{} res.State.Events = []gomatrixserverlib.ClientEvent{} res.Timeline.Events = []gomatrixserverlib.ClientEvent{} return &res } type SendToDeviceNID int type SendToDeviceEvent struct { gomatrixserverlib.SendToDeviceEvent ID SendToDeviceNID UserID string DeviceID string SentByToken *StreamingToken } type PeekingDevice struct { UserID string DeviceID string } type Peek struct { RoomID string New bool Deleted bool }