284 lines
9.5 KiB
Go
284 lines
9.5 KiB
Go
|
// Copyright (c) 2020 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 mautrix
|
||
|
|
|
||
|
|
import (
|
||
|
|
"fmt"
|
||
|
|
"runtime/debug"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"maunium.net/go/mautrix/event"
|
||
|
|
"maunium.net/go/mautrix/id"
|
||
|
|
)
|
||
|
|
|
||
|
|
// EventSource represents the part of the sync response that an event came from.
|
||
|
|
type EventSource int
|
||
|
|
|
||
|
|
const (
|
||
|
|
EventSourcePresence EventSource = 1 << iota
|
||
|
|
EventSourceJoin
|
||
|
|
EventSourceInvite
|
||
|
|
EventSourceLeave
|
||
|
|
EventSourceAccountData
|
||
|
|
EventSourceTimeline
|
||
|
|
EventSourceState
|
||
|
|
EventSourceEphemeral
|
||
|
|
EventSourceToDevice
|
||
|
|
)
|
||
|
|
|
||
|
|
func (es EventSource) String() string {
|
||
|
|
switch {
|
||
|
|
case es == EventSourcePresence:
|
||
|
|
return "presence"
|
||
|
|
case es == EventSourceAccountData:
|
||
|
|
return "user account data"
|
||
|
|
case es == EventSourceToDevice:
|
||
|
|
return "to-device"
|
||
|
|
case es&EventSourceJoin != 0:
|
||
|
|
es -= EventSourceJoin
|
||
|
|
switch es {
|
||
|
|
case EventSourceState:
|
||
|
|
return "joined state"
|
||
|
|
case EventSourceTimeline:
|
||
|
|
return "joined timeline"
|
||
|
|
case EventSourceEphemeral:
|
||
|
|
return "room ephemeral (joined)"
|
||
|
|
case EventSourceAccountData:
|
||
|
|
return "room account data (joined)"
|
||
|
|
}
|
||
|
|
case es&EventSourceInvite != 0:
|
||
|
|
es -= EventSourceInvite
|
||
|
|
switch es {
|
||
|
|
case EventSourceState:
|
||
|
|
return "invited state"
|
||
|
|
}
|
||
|
|
case es&EventSourceLeave != 0:
|
||
|
|
es -= EventSourceLeave
|
||
|
|
switch es {
|
||
|
|
case EventSourceState:
|
||
|
|
return "left state"
|
||
|
|
case EventSourceTimeline:
|
||
|
|
return "left timeline"
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return fmt.Sprintf("unknown (%d)", es)
|
||
|
|
}
|
||
|
|
|
||
|
|
// EventHandler handles a single event from a sync response.
|
||
|
|
type EventHandler func(source EventSource, evt *event.Event)
|
||
|
|
|
||
|
|
// SyncHandler handles a whole sync response. If the return value is false, handling will be stopped completely.
|
||
|
|
type SyncHandler func(resp *RespSync, since string) bool
|
||
|
|
|
||
|
|
// Syncer is an interface that must be satisfied in order to do /sync requests on a client.
|
||
|
|
type Syncer interface {
|
||
|
|
// Process the /sync response. The since parameter is the since= value that was used to produce the response.
|
||
|
|
// This is useful for detecting the very first sync (since=""). If an error is return, Syncing will be stopped
|
||
|
|
// permanently.
|
||
|
|
ProcessResponse(resp *RespSync, since string) error
|
||
|
|
// OnFailedSync returns either the time to wait before retrying or an error to stop syncing permanently.
|
||
|
|
OnFailedSync(res *RespSync, err error) (time.Duration, error)
|
||
|
|
// GetFilterJSON for the given user ID. NOT the filter ID.
|
||
|
|
GetFilterJSON(userID id.UserID) *Filter
|
||
|
|
}
|
||
|
|
|
||
|
|
type ExtensibleSyncer interface {
|
||
|
|
OnSync(callback SyncHandler)
|
||
|
|
OnEvent(callback EventHandler)
|
||
|
|
OnEventType(eventType event.Type, callback EventHandler)
|
||
|
|
}
|
||
|
|
|
||
|
|
// DefaultSyncer is the default syncing implementation. You can either write your own syncer, or selectively
|
||
|
|
// replace parts of this default syncer (e.g. the ProcessResponse method). The default syncer uses the observer
|
||
|
|
// pattern to notify callers about incoming events. See DefaultSyncer.OnEventType for more information.
|
||
|
|
type DefaultSyncer struct {
|
||
|
|
// syncListeners want the whole sync response, e.g. the crypto machine
|
||
|
|
syncListeners []SyncHandler
|
||
|
|
// globalListeners want all events
|
||
|
|
globalListeners []EventHandler
|
||
|
|
// listeners want a specific event type
|
||
|
|
listeners map[event.Type][]EventHandler
|
||
|
|
// ParseEventContent determines whether or not event content should be parsed before passing to handlers.
|
||
|
|
ParseEventContent bool
|
||
|
|
// ParseErrorHandler is called when event.Content.ParseRaw returns an error.
|
||
|
|
// If it returns false, the event will not be forwarded to listeners.
|
||
|
|
ParseErrorHandler func(evt *event.Event, err error) bool
|
||
|
|
}
|
||
|
|
|
||
|
|
var _ Syncer = (*DefaultSyncer)(nil)
|
||
|
|
var _ ExtensibleSyncer = (*DefaultSyncer)(nil)
|
||
|
|
|
||
|
|
// NewDefaultSyncer returns an instantiated DefaultSyncer
|
||
|
|
func NewDefaultSyncer() *DefaultSyncer {
|
||
|
|
return &DefaultSyncer{
|
||
|
|
listeners: make(map[event.Type][]EventHandler),
|
||
|
|
syncListeners: []SyncHandler{},
|
||
|
|
globalListeners: []EventHandler{},
|
||
|
|
ParseEventContent: true,
|
||
|
|
ParseErrorHandler: func(evt *event.Event, err error) bool {
|
||
|
|
return false
|
||
|
|
},
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// ProcessResponse processes the /sync response in a way suitable for bots. "Suitable for bots" means a stream of
|
||
|
|
// unrepeating events. Returns a fatal error if a listener panics.
|
||
|
|
func (s *DefaultSyncer) ProcessResponse(res *RespSync, since string) (err error) {
|
||
|
|
defer func() {
|
||
|
|
if r := recover(); r != nil {
|
||
|
|
err = fmt.Errorf("ProcessResponse panicked! since=%s panic=%s\n%s", since, r, debug.Stack())
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
for _, listener := range s.syncListeners {
|
||
|
|
if !listener(res, since) {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
s.processSyncEvents("", res.Presence.Events, EventSourcePresence)
|
||
|
|
s.processSyncEvents("", res.AccountData.Events, EventSourceAccountData)
|
||
|
|
|
||
|
|
for roomID, roomData := range res.Rooms.Join {
|
||
|
|
s.processSyncEvents(roomID, roomData.State.Events, EventSourceJoin|EventSourceState)
|
||
|
|
s.processSyncEvents(roomID, roomData.Timeline.Events, EventSourceJoin|EventSourceTimeline)
|
||
|
|
s.processSyncEvents(roomID, roomData.Ephemeral.Events, EventSourceJoin|EventSourceEphemeral)
|
||
|
|
s.processSyncEvents(roomID, roomData.AccountData.Events, EventSourceJoin|EventSourceAccountData)
|
||
|
|
}
|
||
|
|
for roomID, roomData := range res.Rooms.Invite {
|
||
|
|
s.processSyncEvents(roomID, roomData.State.Events, EventSourceInvite|EventSourceState)
|
||
|
|
}
|
||
|
|
for roomID, roomData := range res.Rooms.Leave {
|
||
|
|
s.processSyncEvents(roomID, roomData.State.Events, EventSourceLeave|EventSourceState)
|
||
|
|
s.processSyncEvents(roomID, roomData.Timeline.Events, EventSourceLeave|EventSourceTimeline)
|
||
|
|
}
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *DefaultSyncer) processSyncEvents(roomID id.RoomID, events []*event.Event, source EventSource) {
|
||
|
|
for _, evt := range events {
|
||
|
|
s.processSyncEvent(roomID, evt, source)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *DefaultSyncer) processSyncEvent(roomID id.RoomID, evt *event.Event, source EventSource) {
|
||
|
|
evt.RoomID = roomID
|
||
|
|
|
||
|
|
// Ensure the type class is correct. It's safe to mutate the class since the event type is not a pointer.
|
||
|
|
// Listeners are keyed by type structs, which means only the correct class will pass.
|
||
|
|
switch {
|
||
|
|
case evt.StateKey != nil:
|
||
|
|
evt.Type.Class = event.StateEventType
|
||
|
|
case source == EventSourcePresence, source&EventSourceEphemeral != 0:
|
||
|
|
evt.Type.Class = event.EphemeralEventType
|
||
|
|
case source&EventSourceAccountData != 0:
|
||
|
|
evt.Type.Class = event.AccountDataEventType
|
||
|
|
case source == EventSourceToDevice:
|
||
|
|
evt.Type.Class = event.ToDeviceEventType
|
||
|
|
default:
|
||
|
|
evt.Type.Class = event.MessageEventType
|
||
|
|
}
|
||
|
|
|
||
|
|
if s.ParseEventContent {
|
||
|
|
err := evt.Content.ParseRaw(evt.Type)
|
||
|
|
if err != nil && !s.ParseErrorHandler(evt, err) {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
s.notifyListeners(source, evt)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *DefaultSyncer) notifyListeners(source EventSource, evt *event.Event) {
|
||
|
|
for _, fn := range s.globalListeners {
|
||
|
|
fn(source, evt)
|
||
|
|
}
|
||
|
|
listeners, exists := s.listeners[evt.Type]
|
||
|
|
if exists {
|
||
|
|
for _, fn := range listeners {
|
||
|
|
fn(source, evt)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// OnEventType allows callers to be notified when there are new events for the given event type.
|
||
|
|
// There are no duplicate checks.
|
||
|
|
func (s *DefaultSyncer) OnEventType(eventType event.Type, callback EventHandler) {
|
||
|
|
_, exists := s.listeners[eventType]
|
||
|
|
if !exists {
|
||
|
|
s.listeners[eventType] = []EventHandler{}
|
||
|
|
}
|
||
|
|
s.listeners[eventType] = append(s.listeners[eventType], callback)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *DefaultSyncer) OnSync(callback SyncHandler) {
|
||
|
|
s.syncListeners = append(s.syncListeners, callback)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *DefaultSyncer) OnEvent(callback EventHandler) {
|
||
|
|
s.globalListeners = append(s.globalListeners, callback)
|
||
|
|
}
|
||
|
|
|
||
|
|
// OnFailedSync always returns a 10 second wait period between failed /syncs, never a fatal error.
|
||
|
|
func (s *DefaultSyncer) OnFailedSync(res *RespSync, err error) (time.Duration, error) {
|
||
|
|
return 10 * time.Second, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// GetFilterJSON returns a filter with a timeline limit of 50.
|
||
|
|
func (s *DefaultSyncer) GetFilterJSON(userID id.UserID) *Filter {
|
||
|
|
return &Filter{
|
||
|
|
Room: RoomFilter{
|
||
|
|
Timeline: FilterPart{
|
||
|
|
Limit: 50,
|
||
|
|
},
|
||
|
|
},
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// OldEventIgnorer is an utility struct for bots to ignore events from before the bot joined the room.
|
||
|
|
// Create a struct and call Register with your DefaultSyncer to register the sync handler.
|
||
|
|
type OldEventIgnorer struct {
|
||
|
|
UserID id.UserID
|
||
|
|
}
|
||
|
|
|
||
|
|
func (oei *OldEventIgnorer) Register(syncer ExtensibleSyncer) {
|
||
|
|
syncer.OnSync(oei.DontProcessOldEvents)
|
||
|
|
}
|
||
|
|
|
||
|
|
// DontProcessOldEvents returns true if a sync response should be processed. May modify the response to remove
|
||
|
|
// stuff that shouldn't be processed.
|
||
|
|
func (oei *OldEventIgnorer) DontProcessOldEvents(resp *RespSync, since string) bool {
|
||
|
|
if since == "" {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
// This is a horrible hack because /sync will return the most recent messages for a room
|
||
|
|
// as soon as you /join it. We do NOT want to process those events in that particular room
|
||
|
|
// because they may have already been processed (if you toggle the bot in/out of the room).
|
||
|
|
//
|
||
|
|
// Work around this by inspecting each room's timeline and seeing if an m.room.member event for us
|
||
|
|
// exists and is "join" and then discard processing that room entirely if so.
|
||
|
|
// TODO: We probably want to process messages from after the last join event in the timeline.
|
||
|
|
for roomID, roomData := range resp.Rooms.Join {
|
||
|
|
for i := len(roomData.Timeline.Events) - 1; i >= 0; i-- {
|
||
|
|
evt := roomData.Timeline.Events[i]
|
||
|
|
if evt.Type == event.StateMember && evt.GetStateKey() == string(oei.UserID) {
|
||
|
|
membership, _ := evt.Content.Raw["membership"].(string)
|
||
|
|
if membership == "join" {
|
||
|
|
_, ok := resp.Rooms.Join[roomID]
|
||
|
|
if !ok {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
delete(resp.Rooms.Join, roomID) // don't re-process messages
|
||
|
|
delete(resp.Rooms.Invite, roomID) // don't re-process invites
|
||
|
|
break
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return true
|
||
|
|
}
|