tulir/gomuks

View on GitHub
matrix/sync.go

Summary

Maintainability
A
0 mins
Test Coverage
// gomuks - A terminal Matrix client written in Go.
// Copyright (C) 2020 Tulir Asokan
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.

// Based on https://github.com/matrix-org/mautrix/blob/master/sync.go

package matrix

import (
    "sync"
    "time"

    "maunium.net/go/mautrix"
    "maunium.net/go/mautrix/event"
    "maunium.net/go/mautrix/id"

    "maunium.net/go/gomuks/debug"
    ifc "maunium.net/go/gomuks/interface"
    "maunium.net/go/gomuks/matrix/rooms"
)

type GomuksSyncer struct {
    rooms             *rooms.RoomCache
    globalListeners   []mautrix.SyncHandler
    listeners         map[event.Type][]mautrix.EventHandler // event type to listeners array
    FirstSyncDone     bool
    InitDoneCallback  func()
    FirstDoneCallback func()
    Progress          ifc.SyncingModal
}

// NewGomuksSyncer returns an instantiated GomuksSyncer
func NewGomuksSyncer(rooms *rooms.RoomCache) *GomuksSyncer {
    return &GomuksSyncer{
        rooms:           rooms,
        globalListeners: []mautrix.SyncHandler{},
        listeners:       make(map[event.Type][]mautrix.EventHandler),
        FirstSyncDone:   false,
        Progress:        StubSyncingModal{},
    }
}

// ProcessResponse processes a Matrix sync response.
func (s *GomuksSyncer) ProcessResponse(res *mautrix.RespSync, since string) (err error) {
    if since == "" {
        s.rooms.DisableUnloading()
    }
    debug.Print("Received sync response")
    s.Progress.SetMessage("Processing sync response")
    steps := len(res.Rooms.Join) + len(res.Rooms.Invite) + len(res.Rooms.Leave)
    s.Progress.SetSteps(steps + 2 + len(s.globalListeners))

    wait := &sync.WaitGroup{}
    callback := func() {
        wait.Done()
        s.Progress.Step()
    }
    wait.Add(len(s.globalListeners))
    s.notifyGlobalListeners(res, since, callback)
    wait.Wait()

    s.processSyncEvents(nil, res.Presence.Events, mautrix.EventSourcePresence)
    s.Progress.Step()
    s.processSyncEvents(nil, res.AccountData.Events, mautrix.EventSourceAccountData)
    s.Progress.Step()

    wait.Add(steps)

    for roomID, roomData := range res.Rooms.Join {
        go s.processJoinedRoom(roomID, roomData, callback)
    }

    for roomID, roomData := range res.Rooms.Invite {
        go s.processInvitedRoom(roomID, roomData, callback)
    }

    for roomID, roomData := range res.Rooms.Leave {
        go s.processLeftRoom(roomID, roomData, callback)
    }

    wait.Wait()
    s.Progress.SetMessage("Finishing sync")

    if since == "" && s.InitDoneCallback != nil {
        s.InitDoneCallback()
        s.rooms.EnableUnloading()
    }
    if !s.FirstSyncDone && s.FirstDoneCallback != nil {
        s.FirstDoneCallback()
    }
    s.FirstSyncDone = true
    return
}

func (s *GomuksSyncer) notifyGlobalListeners(res *mautrix.RespSync, since string, callback func()) {
    for _, listener := range s.globalListeners {
        go func(listener mautrix.SyncHandler) {
            listener(res, since)
            callback()
        }(listener)
    }
}

func (s *GomuksSyncer) processJoinedRoom(roomID id.RoomID, roomData mautrix.SyncJoinedRoom, callback func()) {
    defer debug.Recover()
    room := s.rooms.GetOrCreate(roomID)
    room.UpdateSummary(roomData.Summary)
    s.processSyncEvents(room, roomData.State.Events, mautrix.EventSourceJoin|mautrix.EventSourceState)
    s.processSyncEvents(room, roomData.Timeline.Events, mautrix.EventSourceJoin|mautrix.EventSourceTimeline)
    s.processSyncEvents(room, roomData.Ephemeral.Events, mautrix.EventSourceJoin|mautrix.EventSourceEphemeral)
    s.processSyncEvents(room, roomData.AccountData.Events, mautrix.EventSourceJoin|mautrix.EventSourceAccountData)

    if len(room.PrevBatch) == 0 {
        room.PrevBatch = roomData.Timeline.PrevBatch
    }
    room.LastPrevBatch = roomData.Timeline.PrevBatch
    callback()
}

func (s *GomuksSyncer) processInvitedRoom(roomID id.RoomID, roomData mautrix.SyncInvitedRoom, callback func()) {
    defer debug.Recover()
    room := s.rooms.GetOrCreate(roomID)
    room.UpdateSummary(roomData.Summary)
    s.processSyncEvents(room, roomData.State.Events, mautrix.EventSourceInvite|mautrix.EventSourceState)
    callback()
}

func (s *GomuksSyncer) processLeftRoom(roomID id.RoomID, roomData mautrix.SyncLeftRoom, callback func()) {
    defer debug.Recover()
    room := s.rooms.GetOrCreate(roomID)
    room.HasLeft = true
    room.UpdateSummary(roomData.Summary)
    s.processSyncEvents(room, roomData.State.Events, mautrix.EventSourceLeave|mautrix.EventSourceState)
    s.processSyncEvents(room, roomData.Timeline.Events, mautrix.EventSourceLeave|mautrix.EventSourceTimeline)

    if len(room.PrevBatch) == 0 {
        room.PrevBatch = roomData.Timeline.PrevBatch
    }
    room.LastPrevBatch = roomData.Timeline.PrevBatch
    callback()
}

func (s *GomuksSyncer) processSyncEvents(room *rooms.Room, events []*event.Event, source mautrix.EventSource) {
    for _, evt := range events {
        s.processSyncEvent(room, evt, source)
    }
}

func (s *GomuksSyncer) processSyncEvent(room *rooms.Room, evt *event.Event, source mautrix.EventSource) {
    if room != nil {
        evt.RoomID = room.ID
    }
    // Ensure the type class is correct. It's safe to mutate since it's 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 == mautrix.EventSourcePresence, source&mautrix.EventSourceEphemeral != 0:
        evt.Type.Class = event.EphemeralEventType
    case source&mautrix.EventSourceAccountData != 0:
        evt.Type.Class = event.AccountDataEventType
    case source == mautrix.EventSourceToDevice:
        evt.Type.Class = event.ToDeviceEventType
    default:
        evt.Type.Class = event.MessageEventType
    }

    err := evt.Content.ParseRaw(evt.Type)
    if err != nil {
        debug.Printf("Failed to unmarshal content of event %s (type %s) by %s in %s: %v\n%s", evt.ID, evt.Type.Repr(), evt.Sender, evt.RoomID, err, string(evt.Content.VeryRaw))
        // TODO might be good to let these pass to allow handling invalid events too
        return
    }

    if room != nil && evt.Type.IsState() {
        room.UpdateState(evt)
    }
    s.notifyListeners(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 *GomuksSyncer) OnEventType(eventType event.Type, callback mautrix.EventHandler) {
    _, exists := s.listeners[eventType]
    if !exists {
        s.listeners[eventType] = []mautrix.EventHandler{}
    }
    s.listeners[eventType] = append(s.listeners[eventType], callback)
}

func (s *GomuksSyncer) OnSync(callback mautrix.SyncHandler) {
    s.globalListeners = append(s.globalListeners, callback)
}

func (s *GomuksSyncer) notifyListeners(source mautrix.EventSource, evt *event.Event) {
    listeners, exists := s.listeners[evt.Type]
    if !exists {
        return
    }
    for _, fn := range listeners {
        fn(source, evt)
    }
}

// OnFailedSync always returns a 10 second wait period between failed /syncs, never a fatal error.
func (s *GomuksSyncer) OnFailedSync(res *mautrix.RespSync, err error) (time.Duration, error) {
    debug.Printf("Sync failed: %v", err)
    return 10 * time.Second, nil
}

// GetFilterJSON returns a filter with a timeline limit of 50.
func (s *GomuksSyncer) GetFilterJSON(_ id.UserID) *mautrix.Filter {
    stateEvents := []event.Type{
        event.StateMember,
        event.StateRoomName,
        event.StateTopic,
        event.StateCanonicalAlias,
        event.StatePowerLevels,
        event.StateTombstone,
        event.StateEncryption,
    }
    messageEvents := []event.Type{
        event.EventMessage,
        event.EventRedaction,
        event.EventEncrypted,
        event.EventSticker,
        event.EventReaction,
    }
    return &mautrix.Filter{
        Room: mautrix.RoomFilter{
            IncludeLeave: false,
            State: mautrix.FilterPart{
                LazyLoadMembers: true,
                Types:           stateEvents,
            },
            Timeline: mautrix.FilterPart{
                LazyLoadMembers: true,
                Types:           append(messageEvents, stateEvents...),
                Limit:           50,
            },
            Ephemeral: mautrix.FilterPart{
                Types: []event.Type{event.EphemeralEventTyping, event.EphemeralEventReceipt},
            },
            AccountData: mautrix.FilterPart{
                Types: []event.Type{event.AccountDataRoomTags},
            },
        },
        AccountData: mautrix.FilterPart{
            Types: []event.Type{event.AccountDataPushRules, event.AccountDataDirectChats, AccountDataGomuksPreferences},
        },
        Presence: mautrix.FilterPart{
            NotTypes: []event.Type{event.NewEventType("*")},
        },
    }
}