client,event,bridgev2: add support for Beeper's custom ephemeral events and AI stream events

appservice/intent.go

@@ -222,6 +222,17 @@ func (intent *IntentAPI) SendMessageEvent(ctx context.Context, roomID id.RoomID,
 	return intent.Client.SendMessageEvent(ctx, roomID, eventType, contentJSON, extra...)
 }
 
+func (intent *IntentAPI) BeeperSendEphemeralEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, contentJSON any, extra ...mautrix.ReqSendEvent) (*mautrix.RespSendEvent, error) {
+	if err := intent.EnsureJoined(ctx, roomID); err != nil {
+		return nil, err
+	}
+	if !intent.SpecVersions.Supports(mautrix.BeeperFeatureEphemeralEvents) {
+		return nil, mautrix.MUnrecognized.WithMessage("Homeserver does not advertise com.beeper.ephemeral support")
+	}
+	contentJSON = intent.AddDoublePuppetValue(contentJSON)
+	return intent.Client.BeeperSendEphemeralEvent(ctx, roomID, eventType, contentJSON, extra...)
+}
+
 // Deprecated: use SendMessageEvent with mautrix.ReqSendEvent.Timestamp instead
 func (intent *IntentAPI) SendMassagedMessageEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, contentJSON interface{}, ts int64) (*mautrix.RespSendEvent, error) {
 	return intent.SendMessageEvent(ctx, roomID, eventType, contentJSON, mautrix.ReqSendEvent{Timestamp: ts})

bridgev2/errors.go

@@ -75,6 +75,7 @@ var (
 	ErrMediaConvertFailed               error = WrapErrorInStatus(errors.New("failed to convert media")).WithMessage("failed to convert media").WithIsCertain(true).WithSendNotice(true)
 	ErrMembershipNotSupported           error = WrapErrorInStatus(errors.New("this bridge does not support changing group membership")).WithIsCertain(true).WithErrorAsMessage().WithSendNotice(false).WithErrorReason(event.MessageStatusUnsupported)
 	ErrDeleteChatNotSupported           error = WrapErrorInStatus(errors.New("this bridge does not support deleting chats")).WithIsCertain(true).WithErrorAsMessage().WithSendNotice(false).WithErrorReason(event.MessageStatusUnsupported)
+	ErrBeeperAIStreamNotSupported       error = WrapErrorInStatus(errors.New("this bridge does not support Beeper AI stream events")).WithIsCertain(true).WithErrorAsMessage().WithSendNotice(false).WithErrorReason(event.MessageStatusUnsupported)
 	ErrPowerLevelsNotSupported          error = WrapErrorInStatus(errors.New("this bridge does not support changing group power levels")).WithIsCertain(true).WithErrorAsMessage().WithSendNotice(false).WithErrorReason(event.MessageStatusUnsupported)
 	ErrRemoteEchoTimeout                      = WrapErrorInStatus(errors.New("remote echo timed out")).WithIsCertain(false).WithSendNotice(true).WithErrorReason(event.MessageStatusTooOld)
 	ErrRemoteAckTimeout                       = WrapErrorInStatus(errors.New("remote ack timed out")).WithIsCertain(false).WithSendNotice(true).WithErrorReason(event.MessageStatusTooOld)

bridgev2/matrix/connector.go

@@ -144,6 +144,7 @@ func (br *Connector) Init(bridge *bridgev2.Bridge) {
 	br.EventProcessor.On(event.EventReaction, br.handleRoomEvent)
 	br.EventProcessor.On(event.EventRedaction, br.handleRoomEvent)
 	br.EventProcessor.On(event.EventEncrypted, br.handleEncryptedEvent)
+	br.EventProcessor.On(event.EphemeralEventEncrypted, br.handleEncryptedEvent)
 	br.EventProcessor.On(event.StateMember, br.handleRoomEvent)
 	br.EventProcessor.On(event.StatePowerLevels, br.handleRoomEvent)
 	br.EventProcessor.On(event.StateRoomName, br.handleRoomEvent)
@@ -156,6 +157,7 @@ func (br *Connector) Init(bridge *bridgev2.Bridge) {
 	br.EventProcessor.On(event.BeeperAcceptMessageRequest, br.handleRoomEvent)
 	br.EventProcessor.On(event.EphemeralEventReceipt, br.handleEphemeralEvent)
 	br.EventProcessor.On(event.EphemeralEventTyping, br.handleEphemeralEvent)
+	br.EventProcessor.On(event.BeeperEphemeralEventAIStream, br.handleEphemeralEvent)
 	br.Bot = br.AS.BotIntent()
 	br.Crypto = NewCryptoHelper(br)
 	br.Bridge.Commands.(*commands.Processor).AddHandlers(

bridgev2/matrix/intent.go

@@ -43,6 +43,7 @@ type ASIntent struct {
 
 var _ bridgev2.MatrixAPI = (*ASIntent)(nil)
 var _ bridgev2.MarkAsDMMatrixAPI = (*ASIntent)(nil)
+var _ bridgev2.EphemeralSendingMatrixAPI = (*ASIntent)(nil)
 
 func (as *ASIntent) SendMessage(ctx context.Context, roomID id.RoomID, eventType event.Type, content *event.Content, extra *bridgev2.MatrixSendExtra) (*mautrix.RespSendEvent, error) {
 	if extra == nil {
@@ -84,6 +85,21 @@ func (as *ASIntent) SendMessage(ctx context.Context, roomID id.RoomID, eventType
 	return as.Matrix.SendMessageEvent(ctx, roomID, eventType, content, mautrix.ReqSendEvent{Timestamp: extra.Timestamp.UnixMilli()})
 }
 
+func (as *ASIntent) BeeperSendEphemeralEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, content *event.Content, txnID string) (*mautrix.RespSendEvent, error) {
+	if !as.Connector.SpecVersions.Supports(mautrix.BeeperFeatureEphemeralEvents) {
+		return nil, mautrix.MUnrecognized.WithMessage("Homeserver does not advertise com.beeper.ephemeral support")
+	}
+	if encrypted, err := as.Matrix.StateStore.IsEncrypted(ctx, roomID); err != nil {
+		return nil, fmt.Errorf("failed to check if room is encrypted: %w", err)
+	} else if encrypted && as.Connector.Crypto != nil {
+		if err = as.Connector.Crypto.Encrypt(ctx, roomID, eventType, content); err != nil {
+			return nil, err
+		}
+		eventType = event.EventEncrypted
+	}
+	return as.Matrix.BeeperSendEphemeralEvent(ctx, roomID, eventType, content, mautrix.ReqSendEvent{TransactionID: txnID})
+}
+
 func (as *ASIntent) fillMemberEvent(ctx context.Context, roomID id.RoomID, userID id.UserID, content *event.Content) {
 	targetContent, ok := content.Parsed.(*event.MemberEventContent)
 	if !ok || targetContent.Displayname != "" || targetContent.AvatarURL != "" {

bridgev2/matrix/matrix.go

@@ -68,6 +68,10 @@ func (br *Connector) handleEphemeralEvent(ctx context.Context, evt *event.Event)
 	case event.EphemeralEventTyping:
 		typingContent := evt.Content.AsTyping()
 		typingContent.UserIDs = slices.DeleteFunc(typingContent.UserIDs, br.shouldIgnoreEventFromUser)
+	case event.BeeperEphemeralEventAIStream:
+		if br.shouldIgnoreEvent(evt) {
+			return
+		}
 	}
 	br.Bridge.QueueMatrixEvent(ctx, evt)
 }
@@ -231,7 +235,6 @@ func (br *Connector) postDecrypt(ctx context.Context, original, decrypted *event
 	go br.sendSuccessCheckpoint(ctx, decrypted, status.MsgStepDecrypted, retryCount)
 	decrypted.Mautrix.CheckpointSent = true
 	decrypted.Mautrix.DecryptionDuration = duration
-	decrypted.Mautrix.EventSource |= event.SourceDecrypted
 	br.EventProcessor.Dispatch(ctx, decrypted)
 	if errorEventID != nil && *errorEventID != "" {
 		_, _ = br.Bot.RedactEvent(ctx, decrypted.RoomID, *errorEventID)

bridgev2/matrixinterface.go

@@ -217,3 +217,8 @@ type MarkAsDMMatrixAPI interface {
 	MatrixAPI
 	MarkAsDM(ctx context.Context, roomID id.RoomID, otherUser id.UserID) error
 }
+
+type EphemeralSendingMatrixAPI interface {
+	MatrixAPI
+	BeeperSendEphemeralEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, content *event.Content, txnID string) (*mautrix.RespSendEvent, error)
+}

bridgev2/networkinterface.go

@@ -726,6 +726,11 @@ type MessageRequestAcceptingNetworkAPI interface {
 	HandleMatrixAcceptMessageRequest(ctx context.Context, msg *MatrixAcceptMessageRequest) error
 }
 
+type BeeperAIStreamHandlingNetworkAPI interface {
+	NetworkAPI
+	HandleMatrixBeeperAIStream(ctx context.Context, msg *MatrixBeeperAIStream) error
+}
+
 type ResolveIdentifierResponse struct {
 	// Ghost is the ghost of the user that the identifier resolves to.
 	// This field should be set whenever possible. However, it is not required,
@@ -1439,6 +1444,7 @@ type MatrixViewingChat struct {
 
 type MatrixDeleteChat = MatrixEventBase[*event.BeeperChatDeleteEventContent]
 type MatrixAcceptMessageRequest = MatrixEventBase[*event.BeeperAcceptMessageRequestEventContent]
+type MatrixBeeperAIStream = MatrixEventBase[*event.BeeperAIStreamEventContent]
 type MatrixMarkedUnread = MatrixRoomMeta[*event.MarkedUnreadEventContent]
 type MatrixMute = MatrixRoomMeta[*event.BeeperMuteEventContent]
 type MatrixRoomTag = MatrixRoomMeta[*event.TagEventContent]

bridgev2/portal.go

@@ -697,6 +697,8 @@ func (portal *Portal) handleMatrixEvent(ctx context.Context, sender *User, evt *
 			return portal.handleMatrixReceipts(ctx, evt)
 		case event.EphemeralEventTyping:
 			return portal.handleMatrixTyping(ctx, evt)
+		case event.BeeperEphemeralEventAIStream:
+			return portal.handleMatrixAIStream(ctx, sender, evt)
 		default:
 			return EventHandlingResultIgnored
 		}
@@ -941,6 +943,50 @@ func (portal *Portal) handleMatrixTyping(ctx context.Context, evt *event.Event)
 	return EventHandlingResultSuccess
 }
 
+func (portal *Portal) handleMatrixAIStream(ctx context.Context, sender *User, evt *event.Event) EventHandlingResult {
+	log := zerolog.Ctx(ctx)
+	if sender == nil {
+		log.Error().Msg("Missing sender for Matrix AI stream event")
+		return EventHandlingResultIgnored
+	}
+	login, _, err := portal.FindPreferredLogin(ctx, sender, true)
+	if err != nil {
+		log.Err(err).Msg("Failed to get user login to handle Matrix AI stream event")
+		return EventHandlingResultFailed.WithMSSError(err)
+	}
+	var origSender *OrigSender
+	if login == nil {
+		if portal.Relay == nil {
+			return EventHandlingResultIgnored
+		}
+		login = portal.Relay
+		origSender = &OrigSender{
+			User:   sender,
+			UserID: sender.MXID,
+		}
+	}
+	content, ok := evt.Content.Parsed.(*event.BeeperAIStreamEventContent)
+	if !ok {
+		log.Error().Type("content_type", evt.Content.Parsed).Msg("Unexpected parsed content type")
+		return EventHandlingResultFailed.WithMSSError(fmt.Errorf("%w: %T", ErrUnexpectedParsedContentType, evt.Content.Parsed))
+	}
+	api, ok := login.Client.(BeeperAIStreamHandlingNetworkAPI)
+	if !ok {
+		return EventHandlingResultIgnored.WithMSSError(ErrBeeperAIStreamNotSupported)
+	}
+	err = api.HandleMatrixBeeperAIStream(ctx, &MatrixBeeperAIStream{
+		Event:      evt,
+		Content:    content,
+		Portal:     portal,
+		OrigSender: origSender,
+	})
+	if err != nil {
+		log.Err(err).Msg("Failed to handle Matrix AI stream event")
+		return EventHandlingResultFailed.WithMSSError(err)
+	}
+	return EventHandlingResultSuccess.WithMSS()
+}
+
 func (portal *Portal) sendTypings(ctx context.Context, userIDs []id.UserID, typing bool) {
 	for _, userID := range userIDs {
 		login, ok := portal.currentlyTypingLogins[userID]

client.go

@@ -1359,6 +1359,48 @@ func (cli *Client) SendMessageEvent(ctx context.Context, roomID id.RoomID, event
 	return
 }
 
+// BeeperSendEphemeralEvent sends an ephemeral event into a room using Beeper's unstable endpoint.
+// contentJSON should be a value that can be encoded as JSON using json.Marshal.
+func (cli *Client) BeeperSendEphemeralEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, contentJSON any, extra ...ReqSendEvent) (resp *RespSendEvent, err error) {
+	var req ReqSendEvent
+	if len(extra) > 0 {
+		req = extra[0]
+	}
+
+	var txnID string
+	if len(req.TransactionID) > 0 {
+		txnID = req.TransactionID
+	} else {
+		txnID = cli.TxnID()
+	}
+
+	queryParams := map[string]string{}
+	if req.Timestamp > 0 {
+		queryParams["ts"] = strconv.FormatInt(req.Timestamp, 10)
+	}
+
+	if !req.DontEncrypt && cli != nil && cli.Crypto != nil && eventType != event.EventEncrypted {
+		var isEncrypted bool
+		isEncrypted, err = cli.StateStore.IsEncrypted(ctx, roomID)
+		if err != nil {
+			err = fmt.Errorf("failed to check if room is encrypted: %w", err)
+			return
+		}
+		if isEncrypted {
+			if contentJSON, err = cli.Crypto.Encrypt(ctx, roomID, eventType, contentJSON); err != nil {
+				err = fmt.Errorf("failed to encrypt event: %w", err)
+				return
+			}
+			eventType = event.EventEncrypted
+		}
+	}
+
+	urlData := ClientURLPath{"unstable", "com.beeper.ephemeral", "rooms", roomID, "ephemeral", eventType.String(), txnID}
+	urlPath := cli.BuildURLWithQuery(urlData, queryParams)
+	_, err = cli.MakeRequest(ctx, http.MethodPut, urlPath, contentJSON, &resp)
+	return
+}
+
 // SendStateEvent sends a state event into a room. See https://spec.matrix.org/v1.16/client-server-api/#put_matrixclientv3roomsroomidstateeventtypestatekey
 // contentJSON should be a pointer to something that can be encoded as JSON using json.Marshal.
 func (cli *Client) SendStateEvent(ctx context.Context, roomID id.RoomID, eventType event.Type, stateKey string, contentJSON any, extra ...ReqSendEvent) (resp *RespSendEvent, err error) {

client_ephemeral_test.go

@@ -0,0 +1,158 @@
+// Copyright (c) 2026 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_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"maunium.net/go/mautrix"
+	"maunium.net/go/mautrix/event"
+	"maunium.net/go/mautrix/id"
+)
+
+func TestClient_SendEphemeralEvent_UsesUnstablePathTxnAndTS(t *testing.T) {
+	roomID := id.RoomID("!room:example.com")
+	evtType := event.Type{Type: "com.example.ephemeral", Class: event.EphemeralEventType}
+	txnID := "txn-123"
+
+	var gotPath string
+	var gotQueryTS string
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		gotQueryTS = r.URL.Query().Get("ts")
+		assert.Equal(t, http.MethodPut, r.Method)
+		_, _ = w.Write([]byte(`{"event_id":"$evt"}`))
+	}))
+	defer ts.Close()
+
+	cli, err := mautrix.NewClient(ts.URL, "", "")
+	require.NoError(t, err)
+
+	_, err = cli.BeeperSendEphemeralEvent(
+		context.Background(),
+		roomID,
+		evtType,
+		map[string]any{"foo": "bar"},
+		mautrix.ReqSendEvent{TransactionID: txnID, Timestamp: 1234},
+	)
+	require.NoError(t, err)
+
+	assert.True(t, strings.Contains(gotPath, "/_matrix/client/unstable/com.beeper.ephemeral/rooms/"))
+	assert.True(t, strings.HasSuffix(gotPath, "/ephemeral/com.example.ephemeral/"+txnID))
+	assert.Equal(t, "1234", gotQueryTS)
+}
+
+func TestClient_SendEphemeralEvent_UnsupportedReturnsMUnrecognized(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+		_, _ = w.Write([]byte(`{"errcode":"M_UNRECOGNIZED","error":"Unrecognized endpoint"}`))
+	}))
+	defer ts.Close()
+
+	cli, err := mautrix.NewClient(ts.URL, "", "")
+	require.NoError(t, err)
+
+	_, err = cli.BeeperSendEphemeralEvent(
+		context.Background(),
+		id.RoomID("!room:example.com"),
+		event.Type{Type: "com.example.ephemeral", Class: event.EphemeralEventType},
+		map[string]any{"foo": "bar"},
+	)
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, mautrix.MUnrecognized))
+}
+
+func TestClient_SendEphemeralEvent_EncryptsInEncryptedRooms(t *testing.T) {
+	roomID := id.RoomID("!room:example.com")
+	evtType := event.Type{Type: "com.example.ephemeral", Class: event.EphemeralEventType}
+	txnID := "txn-encrypted"
+
+	stateStore := mautrix.NewMemoryStateStore()
+	err := stateStore.SetEncryptionEvent(context.Background(), roomID, &event.EncryptionEventContent{
+		Algorithm: id.AlgorithmMegolmV1,
+	})
+	require.NoError(t, err)
+
+	fakeCrypto := &fakeCryptoHelper{
+		encryptedContent: &event.EncryptedEventContent{
+			Algorithm:        id.AlgorithmMegolmV1,
+			MegolmCiphertext: []byte("ciphertext"),
+		},
+	}
+
+	var gotPath string
+	var gotBody map[string]any
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		assert.Equal(t, http.MethodPut, r.Method)
+		err := json.NewDecoder(r.Body).Decode(&gotBody)
+		require.NoError(t, err)
+		_, _ = w.Write([]byte(`{"event_id":"$evt"}`))
+	}))
+	defer ts.Close()
+
+	cli, err := mautrix.NewClient(ts.URL, "", "")
+	require.NoError(t, err)
+	cli.StateStore = stateStore
+	cli.Crypto = fakeCrypto
+
+	_, err = cli.BeeperSendEphemeralEvent(
+		context.Background(),
+		roomID,
+		evtType,
+		map[string]any{"foo": "bar"},
+		mautrix.ReqSendEvent{TransactionID: txnID},
+	)
+	require.NoError(t, err)
+
+	assert.True(t, strings.HasSuffix(gotPath, "/ephemeral/m.room.encrypted/"+txnID))
+	assert.Equal(t, string(id.AlgorithmMegolmV1), gotBody["algorithm"])
+	assert.Equal(t, 1, fakeCrypto.encryptCalls)
+	assert.Equal(t, roomID, fakeCrypto.lastRoomID)
+	assert.Equal(t, evtType, fakeCrypto.lastEventType)
+}
+
+type fakeCryptoHelper struct {
+	encryptCalls     int
+	lastRoomID       id.RoomID
+	lastEventType    event.Type
+	lastEncryptInput any
+	encryptedContent *event.EncryptedEventContent
+}
+
+func (f *fakeCryptoHelper) Encrypt(_ context.Context, roomID id.RoomID, eventType event.Type, content any) (*event.EncryptedEventContent, error) {
+	f.encryptCalls++
+	f.lastRoomID = roomID
+	f.lastEventType = eventType
+	f.lastEncryptInput = content
+	return f.encryptedContent, nil
+}
+
+func (f *fakeCryptoHelper) Decrypt(context.Context, *event.Event) (*event.Event, error) {
+	return nil, nil
+}
+
+func (f *fakeCryptoHelper) WaitForSession(context.Context, id.RoomID, id.SenderKey, id.SessionID, time.Duration) bool {
+	return false
+}
+
+func (f *fakeCryptoHelper) RequestSession(context.Context, id.RoomID, id.SenderKey, id.SessionID, id.UserID, id.DeviceID) {
+}
+
+func (f *fakeCryptoHelper) Init(context.Context) error {
+	return nil
+}

crypto/decryptmegolm.go

@@ -213,6 +213,7 @@ func (mach *OlmMachine) DecryptMegolmEvent(ctx context.Context, evt *event.Event
 			TrustSource:   device,
 			ForwardedKeys: forwardedKeys,
 			WasEncrypted:  true,
+			EventSource:   evt.Mautrix.EventSource | event.SourceDecrypted,
 			ReceivedAt:    evt.Mautrix.ReceivedAt,
 		},
 	}, nil