dendrite/syncapi/streams/stream_invite.go

package streams

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"math"
	"strconv"
	"time"

	"github.com/matrix-org/gomatrixserverlib/spec"

	"github.com/matrix-org/dendrite/roomserver/api"
	"github.com/matrix-org/dendrite/syncapi/storage"
	"github.com/matrix-org/dendrite/syncapi/synctypes"
	"github.com/matrix-org/dendrite/syncapi/types"
)

type InviteStreamProvider struct {
	DefaultStreamProvider
	rsAPI api.SyncRoomserverAPI
}

func (p *InviteStreamProvider) Setup(
	ctx context.Context, snapshot storage.DatabaseTransaction,
) {
	p.DefaultStreamProvider.Setup(ctx, snapshot)

	p.latestMutex.Lock()
	defer p.latestMutex.Unlock()

	id, err := snapshot.MaxStreamPositionForInvites(ctx)
	if err != nil {
		panic(err)
	}
	p.latest = id
}

func (p *InviteStreamProvider) CompleteSync(
	ctx context.Context,
	snapshot storage.DatabaseTransaction,
	req *types.SyncRequest,
) types.StreamPosition {
	return p.IncrementalSync(ctx, snapshot, req, 0, p.LatestPosition(ctx))
}

func (p *InviteStreamProvider) IncrementalSync(
	ctx context.Context,
	snapshot storage.DatabaseTransaction,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition {
	r := types.Range{
		From: from,
		To:   to,
	}

	invites, retiredInvites, maxID, err := snapshot.InviteEventsInRange(
		ctx, req.Device.UserID, r,
	)
	if err != nil {
		req.Log.WithError(err).Error("p.DB.InviteEventsInRange failed")
		return from
	}

	eventFormat := synctypes.FormatSync
	if req.Filter.EventFormat == synctypes.EventFormatFederation {
		eventFormat = synctypes.FormatSyncFederation
	}

	for roomID, inviteEvent := range invites {
		user := spec.UserID{}
		validRoomID, err := spec.NewRoomID(inviteEvent.RoomID())
		if err != nil {
			continue
		}
		sender, err := p.rsAPI.QueryUserIDForSender(ctx, *validRoomID, inviteEvent.SenderID())
		if err == nil && sender != nil {
			user = *sender
		}

		sk := inviteEvent.StateKey()
		if sk != nil && *sk != "" {
			skUserID, err := p.rsAPI.QueryUserIDForSender(ctx, *validRoomID, spec.SenderID(*inviteEvent.StateKey()))
			if err == nil && skUserID != nil {
				skString := skUserID.String()
				sk = &skString
			}
		}

		// skip ignored user events
		if _, ok := req.IgnoredUsers.List[user.String()]; ok {
			continue
		}
		ir := types.NewInviteResponse(inviteEvent, user, sk, eventFormat)
		req.Response.Rooms.Invite[roomID] = ir
	}

	// When doing an initial sync, we don't want to add retired invites, as this
	// can add rooms we were invited to, but already left.
	if from == 0 {
		return to
	}
	for roomID := range retiredInvites {
		membership, _, err := snapshot.SelectMembershipForUser(ctx, roomID, req.Device.UserID, math.MaxInt64)
		// Skip if the user is an existing member of the room.
		// Otherwise, the NewLeaveResponse will eject the user from the room unintentionally
		if membership == spec.Join ||
			err != nil {
			continue
		}

		lr := types.NewLeaveResponse()
		h := sha256.Sum256(append([]byte(roomID), []byte(strconv.FormatInt(int64(to), 10))...))
		lr.Timeline.Events = append(lr.Timeline.Events, synctypes.ClientEvent{
			// fake event ID which muxes in the to position
			EventID:        "$" + base64.RawURLEncoding.EncodeToString(h[:]),
			OriginServerTS: spec.AsTimestamp(time.Now()),
			RoomID:         roomID,
			Sender:         req.Device.UserID,
			StateKey:       &req.Device.UserID,
			Type:           "m.room.member",
			Content:        spec.RawJSON(`{"membership":"leave"}`),
		})
		req.Response.Rooms.Leave[roomID] = lr
	}

	return maxID
}