beeperstream: add framework for streaming using to-device events by batuhan · Pull Request #473 · mautrix/go

batuhan · 2026-03-17T10:33:19Z

No description provided.

helper.mach is already accessible via helper, no need to pass it separately as an embedded field. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Store streams.HandleSyncResponse as a func field rather than a *CryptoHelper back-reference, keeping all existing OlmMachine lines identical to main. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Several small fixes and cleanups across beeperstream and event packages: - beeperstream/crypto.go: JSON-marshal ciphertext instead of manual quoting (removed strconv) and return marshal error to ensure valid JSON payload for encrypted events. - beeperstream/helper.go: Simplify resolveSubscribeExpiry by returning min(descriptor expiry, fallback) directly. - beeperstream/publisher.go: Use helper's logger (h.log.Debug()) instead of zerolog.Ctx(ctx); remove zerolog import. - beeperstream/receiver.go: Add nil-check for subscription descriptor encryption to avoid nil dereference when handling encrypted events. - event/beeper.go: Add BeeperEncodedOrder type. These changes improve JSON correctness for encrypted content, tighten nil-safety, and simplify logging and expiry logic.

tulir

Structs should be preferred more over raw maps

tulir · 2026-03-23T11:19:41Z

+	})
+	on(event.ToDeviceEncrypted, func(ctx context.Context, evt *event.Event) {
+		if normalized := h.handleEvent(ctx, evt); normalized != nil && dispatch != nil {
+			dispatch(ctx, normalized)


This can probably just call h.handleEvent rather than going through the main dispatcher again? These encrypted events never have other data anyway

beeperstream: use raw bytes for stream crypto Switch beeper stream encryption to use raw byte slices (jsonbytes.UnpaddedBytes) instead of base64 strings. Remove base64 encode/decode paths and update encrypt/decrypt to accept []byte keys; store IV and ciphertext as byte fields on EncryptedEventContent. Rename algorithm constant to AlgorithmBeeperStreamV1 and validate/propagate room_id/event_id routing inside encrypted content. Add helpers for marshaling/normalizing content and routing validation, update publisher/receiver/helper logic and tests to use the new format, and adjust OlmMachine to skip beeper-stream encrypted to-device events. Also expose BeeperStreamPublisher via the Crypto interface and ensure deep-copy of binary keys in event cloning. Rename BeeperStreamCiphertext to MegolmCiphertext Replace the BeeperStreamCiphertext field with MegolmCiphertext in event.EncryptedEventContent and update all usages across beeperstream (crypto.go, helper.go, publisher.go). Adjust JSON marshal/unmarshal to read/write MegolmCiphertext for AlgorithmBeeperStreamV1 and use MegolmCiphertext for validation and GCM encryption/decryption. This unifies naming and removes the old field to avoid mismatches. Use HMAC-derived stream_id for beeper streams Replace explicit room_id/event_id routing in encrypted beeper stream events with an HMAC-derived stream_id. - Add deriveStreamID(key, roomID, eventID) which uses HMAC-SHA256 + base64.RawStdEncoding to compute an opaque stream_id. - Update encryptLogicalEvent to set StreamID instead of RoomID/EventID in EncryptedEventContent. - Change event.EncryptedEventContent to include StreamID (string) and remove RoomID/EventID fields. - Helper: track encryptedPubl and encryptedSubs maps mapping stream_id -> streamKey; store streamID on publishedStream and subscription; match incoming encrypted events by stream_id; update pending subscribe logic to carry/compare stream_id and reject old routing shape (room_id/event_id). - Publisher/Receiver: set and maintain streamID on register/subscribe/unregister/unsubscribe and cleanup maps. - Add unit tests to validate StreamID round-trip and helper behaviors, and to reject missing/old encrypted routing shapes. This change scopes routing information to an opaque identifier derived from the encryption key and original identifiers, improving privacy and simplifying matching of encrypted beeper stream events.

Introduce MaxBufferedUpdates to BeeperStreamInfo to allow per-stream replay buffer sizing and validate it on descriptors. Add resolveMaxBufferedUpdates and a maxBufferedUpdates field on publishedStream; Register now initializes maxBufferedUpdates and Publish trims updates using that value instead of the global max. Remove dependence on DispatchableSyncer/Dispatch from Helper.Init/InitAppservice and simplify the ingress adapter to only use On handlers (normalized events are handled internally). Update descriptorEqual to compare MaxBufferedUpdates and adjust tests, including a new test for latest-only replay buffer behavior and tweaks to Init idempotency checks.

Add support for batching multiple beeper stream updates into a single update payload and expanding them on receipt. Key changes: - Add Updates field to BeeperStreamUpdateEventContent to represent batched updates. - Implement expandStreamUpdateEvent and stripUpdateRouting in helper to expand batched update payloads into individual events and to produce raw maps without routing fields. - Change several handlers to return []*event.Event (handleEvent, handleEncryptedEvent, handleEncryptedForPublisher, handleEncryptedForSubscriber) and wire expansion so batched updates are normalized into individual update events. - Add makeReplayUpdateContent and sendReplayUpdates in publisher to bundle multiple updates into a single replay payload and send it to subscribers. - Update send/subscribe logic to use the new batching functions. - Add tests exercising batching behavior for both plain and encrypted replay paths and for sync handling (beeperstream/helper_test.go). This enables efficient replay of buffered updates by sending them as a single message and ensures the receiver expands them back into individual logical update events.

Simplify decryptLogicalEvent by returning specific ToDevice event types directly for subscribe/update payloads and remove the unused logicalType variable. Add a default case that returns an error for unknown beeper stream event types to avoid falling through with an invalid value.

tulir · 2026-03-24T17:27:33Z

 		}
 		content.MegolmCiphertext = content.Ciphertext[1 : len(content.Ciphertext)-1]
+	case id.AlgorithmBeeperStreamV1:
+		return json.Unmarshal(content.Ciphertext, &content.MegolmCiphertext)


I meant that this would be treated exactly the same way as AlgorithmMegolmV1, although maybe it'd be better to have a new field that's a jsonbytes.UnpaddedBytes to avoid having to manually base64 decode later 🤔

Shouldn't use the MegolmCiphertext for that though because marshaling/unmarshaling a []byte will make padded base64 rather than unpadded

tulir · 2026-03-24T17:34:41Z

+	h.registerIngressAdapter(
+		func(evtType event.Type, handler func(context.Context, *event.Event)) {
+			syncer.OnEventType(evtType, handler)
+		},
+	)


I think this is just a no-op wrapper?

Suggested change

h.registerIngressAdapter(

func(evtType event.Type, handler func(context.Context, *event.Event)) {

syncer.OnEventType(evtType, handler)

},

)

h.registerIngressAdapter(syncer.OnEventType)

tulir · 2026-03-24T17:37:47Z

+	case event.ToDeviceBeeperStreamSubscribe:
+		h.handleSubscribeEvent(ctx, evt)
+	case event.ToDeviceBeeperStreamUpdate:
+		return expandStreamUpdateEvent(evt)


I think I missed this last time, but the stream update handling code doesn't seem right. The return value doesn't seem to be going anywhere? Not entirely certain what's the best way to handle it, maybe just a new way to register a stream update listener in the helper where all updates are sent? registerIngressAdapter would then also register a listener for (unencrypted) stream updates and handle that

Rename the beeper-stream ciphertext field and add update dispatching. EncryptedEventContent now stores beeper stream payloads in BeeperStreamCiphertext (with updated Marshal/Unmarshal), and all beeperstream crypto/publisher code now reads/writes that field instead of MegolmCiphertext. Helper gains OnUpdate, an atomic updateHandler, dispatchUpdate and handleAndDispatch to call a registered handler for expanded update events; registerIngressAdapter signature adjusted to accept mautrix.EventHandler and ToDeviceBeeperStreamUpdate/ToDeviceEncrypted events are dispatched. HandleSyncResponse now dispatches expanded events, and bridge initialization registers streams.OnUpdate with the bridge event processor and ensures streams are closed on Stop. These changes centralize beeper stream handling and provide a hook for downstream consumers to receive expanded update events.

tulir · 2026-03-25T10:29:13Z

+	}
+	h.registerIngressAdapter(func(evtType event.Type, handler mautrix.EventHandler) {
+		ep.On(evtType, handler)
+	})


Why did this wrapper come back 🙈

if there's some simple type mismatch, the types should be fixed instead of adding wrappers (might just be that one of the types needs to be changed to be a type alias rather than a custom type, as in type EventHandler = func(...) instead of type EventHandler func(...))

It's the type mismatch :( I'll check how type aliases work in Go

as in type EventHandler = func(...) instead of type EventHandler func(…))

did not know this existed/worked like that

Change EventHandler to a type alias so it is assignment-compatible with mautrix.EventHandler, and update InitAppservice to accept an On method using mautrix.EventHandler. Also simplify registration by passing ep.On directly to registerIngressAdapter, removing the wrapper. These changes allow direct use of mautrix event handlers and avoid unnecessary adapter closures.

tulir · 2026-03-25T12:46:57Z

+		return err
+	}
+	helper.streams = streams
+	helper.streams.OnUpdate(helper.bridge.EventProcessor.Dispatch)


Hmm, this will probably create a loop since the helper is listening on the same event processor

tulir · 2026-03-25T12:47:16Z

+}
+
+// InitAppservice attaches beeper stream handling to an appservice event processor.
+func (h *Helper) InitAppservice(_ context.Context, ep interface {


Why do this and Init have an unused context parameter 🤔

Remove the Helper.InitAppservice method and related test scaffolding (testEventProcessor and TestHelperInitAppserviceForwardsEncryptedBridgeSubscribe). Update OnUpdate comment to reference the Init path only. Also stop calling InitAppservice from CryptoHelper.Start so the appservice code path no longer attempts to initialize beeper streams.

Add InitAppservice to beeperstream.Helper to attach beeper stream handling to an appservice event processor (validates helper and processor, registers ingress adapter, and marks initialized). Add test scaffolding (testEventProcessor) and TestHelperInitAppserviceForwardsEncryptedBridgeSubscribe to verify encrypted subscribe events are forwarded and handled. Change bridgev2/matrix/crypto.go to avoid registering streams.OnUpdate when running in appservice mode and to call streams.InitAppservice during Start() when appservice encryption is enabled, with error logging if initialization fails.

Remove unused context.Context parameters from beeperstream Helper methods: Init() and InitAppservice(). Update call sites accordingly in beeperstream/helper_test.go and bridgev2/matrix/crypto.go to pass the event processor or call Init() without a context. No behavioral changes intended — this just removes an unused argument from the API and updates tests and the bridge initialization to match.

Simplify ingress handling in beeperstream.Helper by removing the OnUpdate API and related dispatch helpers (updateHandler, dispatchUpdate, handleAndDispatch). Introduce handleIngressEvent which simply calls handleEvent without dispatching, update registerIngressAdapter to use it, and change HandleSyncResponse to collect normalized events directly. This decouples per-event dispatching from ingress processing and centralizes normalization behavior.

tulir

Probably good enough for the initial version

Add encrypted beeperstream to-device transport

0dcdd83

batuhan force-pushed the batuhan/ephemeral-to-device branch from bfc1b6e to 0dcdd83 Compare March 22, 2026 01:47

batuhan and others added 3 commits March 22, 2026 03:02

Drop redundant OlmMachine embed from cryptoSyncer

d44e2a2

helper.mach is already accessible via helper, no need to pass it separately as an embedded field. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Use function value instead of back-reference in cryptoSyncer

ec5e2b9

Store streams.HandleSyncResponse as a func field rather than a *CryptoHelper back-reference, keeping all existing OlmMachine lines identical to main. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

batuhan requested a review from tulir March 22, 2026 13:55

tulir reviewed Mar 23, 2026

View reviewed changes

batuhan added 6 commits March 23, 2026 15:26

Update helper.go

4812649

Update crypto.go

a33a671

batuhan requested a review from tulir March 24, 2026 17:33

tulir reviewed Mar 24, 2026

View reviewed changes

tulir reviewed Mar 25, 2026

View reviewed changes

batuhan added 6 commits March 25, 2026 14:09

Update crypto.go

4672b31

Update helper.go

ef1c452

tulir changed the title ~~ephemeral events using to_device~~ beeperstream: add framework for streaming using to-device events Mar 25, 2026

tulir approved these changes Mar 25, 2026

View reviewed changes

batuhan merged commit 96bcf8a into main Mar 25, 2026
10 checks passed

batuhan deleted the batuhan/ephemeral-to-device branch March 25, 2026 14:25

bradtgmurray mentioned this pull request May 14, 2026

MSC4471: Streaming ephemeral event updates for room events matrix-org/matrix-spec-proposals#4471

Open

Uh oh!

Uh oh!

Conversation

batuhan commented Mar 17, 2026

Uh oh!

tulir left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tulir left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Milestone

Development

Uh oh!

2 participants