From eca6e8aaacc677b912f4d4d5d79cb69295c7ad36 Mon Sep 17 00:00:00 2001 From: wouter Date: Fri, 17 Jul 2026 02:40:43 +0200 Subject: [PATCH 1/2] feat: add typed client event API --- examples/02-client.ts | 37 +++------ packages/mxfx/src/client/client.ts | 9 +- packages/mxfx/src/client/event.test.ts | 111 +++++++++++++++++++++++++ packages/mxfx/src/client/event.ts | 90 ++++++++++++++++++++ packages/mxfx/src/client/index.ts | 1 + 5 files changed, 215 insertions(+), 33 deletions(-) create mode 100644 packages/mxfx/src/client/event.test.ts create mode 100644 packages/mxfx/src/client/event.ts diff --git a/examples/02-client.ts b/examples/02-client.ts index e4c0201..40c8e25 100644 --- a/examples/02-client.ts +++ b/examples/02-client.ts @@ -1,33 +1,18 @@ import { NodeHttpClient, NodeRuntime } from '@effect/platform-node' -import { Cause, Config, Effect, Layer, References, Stream } from 'effect' +import { Cause, Config, Effect, Layer, References, Schema } from 'effect' import { DevTools } from 'effect/unstable/devtools' import { MatrixApi, MatrixAuth, MatrixClient, MatrixConfig, Store } from 'mxfx' import { endpoints } from 'mxfx/api' -import type { RoomId } from 'mxfx/branded' -type SyncFrame = typeof endpoints.getSyncV3ResponseSchema.Type -const lastPredicate = (frame: Stream.Stream) => - frame.pipe( - Stream.flatMap(sync => Stream.fromIterable(Object.entries(sync.rooms?.join ?? {}))), - Stream.flatMap(([roomId, room]) => - Stream.fromIterable(room.timeline?.events ?? []).pipe(Stream.map(event => ({ ...event, roomId: roomId as RoomId }))), - ), - Stream.filterEffect(event => - Effect.gen(function* () { - yield* Effect.logDebug(event) - if (event.type !== 'm.room.message') return false - if (typeof event.content !== 'object' || event.content === null) return false - - const content = event.content as { msgtype?: unknown; body?: unknown } - - return ( - typeof content.body === 'string' && - content.body.trim().startsWith('!last') && - (content.msgtype === 'm.text' || content.msgtype === 'm.notice' || content.msgtype === 'm.emote') - ) - }), - ), - ) +const lastEvent = MatrixClient.makeEvent({ + type: 'm.room.message', + bucket: 'rooms.joined', + schema: Schema.Struct({ + msgtype: Schema.Union([Schema.Literal('m.text'), Schema.Literal('m.notice'), Schema.Literal('m.emote')]), + body: Schema.String, + }), + predicate: event => event.content.body.trim().startsWith('!last'), +}) const program = Effect.gen(function* () { const api = yield* MatrixApi.MatrixApi @@ -37,7 +22,7 @@ const program = Effect.gen(function* () { const { userId, deviceId, isGuest } = yield* endpoints.getAccountWhoami().pipe(Effect.andThen(api.execute)) yield* Effect.logDebug({ userId, deviceId, isGuest }) - yield* client.onEvent({ predicate: lastPredicate }, event => + yield* client.onEvent(lastEvent, event => Effect.gen(function* () { const timeline = yield* store.getRoomTimeline(event.roomId) const lastMessages = timeline.filter(e => e.type === 'm.room.message' && e.sender !== userId).slice(-5) diff --git a/packages/mxfx/src/client/client.ts b/packages/mxfx/src/client/client.ts index e75e063..4f6bc5a 100644 --- a/packages/mxfx/src/client/client.ts +++ b/packages/mxfx/src/client/client.ts @@ -1,8 +1,8 @@ import { Effect, Layer, Context, Option, PubSub, Duration, Stream } from 'effect' import { endpoints, MatrixApi } from '../api/index.ts' -import type { RoomId } from '../branded/index.ts' import { Store } from '../store/index.ts' +import type { EventLike, EventUnit } from './event.ts' import { Reducers } from './sync/reducer.ts' const make = Effect.gen(function* () { @@ -48,7 +48,7 @@ const make = Effect.gen(function* () { const syncLoop = () => Effect.forever(syncOnce) const onEvent = Effect.fnUntraced(function* (eventUnit: EventUnit, f: (event: T) => Effect.Effect) { - yield* syncStream.pipe(eventUnit.predicate, Stream.runForEach(f), Effect.forkDetach) + yield* syncStream.pipe(eventUnit.stream, Stream.runForEach(f), Effect.forkDetach) }) return { @@ -58,11 +58,6 @@ const make = Effect.gen(function* () { }) type SyncFrame = typeof endpoints.getSyncV3ResponseSchema.Type -type EventLike = { roomId: RoomId; type: string; content: unknown } - -type EventUnit = { - predicate: (frame: Stream.Stream) => Stream.Stream -} export class MatrixClient extends Context.Service< MatrixClient, diff --git a/packages/mxfx/src/client/event.test.ts b/packages/mxfx/src/client/event.test.ts new file mode 100644 index 0000000..590d14f --- /dev/null +++ b/packages/mxfx/src/client/event.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, expectTypeOf, it } from '@effect/vitest' +import { Effect, Schema, Stream } from 'effect' + +import { endpoints } from '../api/index.ts' +import { makeEvent } from './event.ts' + +const commandContent = Schema.Struct({ + command: Schema.String, + count: Schema.NumberFromString.pipe(Schema.check(Schema.isFinite())), +}) + +const commandEvent = makeEvent({ + type: 'com.example.command', + bucket: 'rooms.joined', + schema: commandContent, + predicate: event => event.content.command === 'run', +}) + +const frame = Schema.decodeUnknownSync(endpoints.getSyncV3ResponseSchema)({ + nextBatch: 'next', + rooms: { + join: { + '!room:example.org': { + stateAfter: { + events: [ + { + type: 'com.example.command', + content: { command: 'run', count: '1' }, + eventId: '$state:example.org', + originServerTs: 1, + sender: '@alice:example.org', + stateKey: '', + }, + ], + }, + timeline: { + events: [ + { + type: 'com.example.command', + content: { command: 'run', count: '2' }, + eventId: '$run:example.org', + originServerTs: 2, + sender: '@alice:example.org', + }, + { + type: 'com.example.command', + content: { command: 'skip', count: '3' }, + eventId: '$skip:example.org', + originServerTs: 3, + sender: '@alice:example.org', + }, + { + type: 'com.example.command', + content: { command: 'run', count: 'invalid' }, + eventId: '$invalid:example.org', + originServerTs: 4, + sender: '@alice:example.org', + }, + { + type: 'com.example.other', + content: { command: 'run', count: '5' }, + eventId: '$other:example.org', + originServerTs: 5, + sender: '@alice:example.org', + }, + ], + }, + }, + }, + }, +}) + +describe('makeEvent', () => { + it.effect('selects, decodes, and refines events from joined rooms', () => + Effect.gen(function* () { + const events = yield* Stream.make(frame).pipe(commandEvent.stream, Stream.runCollect) + + expect(Array.from(events)).toEqual([ + expect.objectContaining({ + roomId: '!room:example.org', + type: 'com.example.command', + stateKey: '', + content: { command: 'run', count: 1 }, + }), + expect.objectContaining({ + roomId: '!room:example.org', + type: 'com.example.command', + content: { command: 'run', count: 2 }, + }), + ]) + }), + ) + + it('infers the literal event type and decoded content', () => { + type Event = Parameters[0] + expectTypeOf().toEqualTypeOf<'com.example.command'>() + expectTypeOf().toEqualTypeOf<{ + readonly command: string + readonly count: number + }>() + }) + + it.effect('emits nothing when the joined-room bucket is absent', () => + Effect.gen(function* () { + const emptyFrame = Schema.decodeUnknownSync(endpoints.getSyncV3ResponseSchema)({ nextBatch: 'next' }) + const events = yield* Stream.make(emptyFrame).pipe(commandEvent.stream, Stream.runCollect) + + expect(Array.from(events)).toEqual([]) + }), + ) +}) diff --git a/packages/mxfx/src/client/event.ts b/packages/mxfx/src/client/event.ts new file mode 100644 index 0000000..df12447 --- /dev/null +++ b/packages/mxfx/src/client/event.ts @@ -0,0 +1,90 @@ +import { Result, Schema, Stream } from 'effect' +import type { Predicate } from 'effect/Predicate' + +import type { endpoints } from '../api/index.ts' +import type { RoomId } from '../branded/index.ts' +import type { ClientEventWithoutRoomId } from '../schema/index.ts' + +export type SyncFrame = typeof endpoints.getSyncV3ResponseSchema.Type + +export type EventLike = { + readonly roomId: RoomId + readonly type: string + readonly content: unknown +} + +/** A room event whose type and content have been refined by `makeEvent`. */ +export type ClientEventOf = Omit & { + readonly roomId: RoomId + readonly type: Type + readonly content: Content +} + +/** The high-level room collections exposed by sync. */ +export type EventBucket = 'rooms.joined' + +export interface EventUnit { + readonly stream: (frames: Stream.Stream) => Stream.Stream +} + +export interface MakeEventOptions> { + readonly type: Type + readonly bucket: EventBucket + /** Schema used to decode and refine the event's `content`. */ + readonly schema: ContentSchema + /** An optional refinement applied after type and content validation. */ + readonly predicate?: Predicate> +} + +export type EventDefinition> = EventUnit< + ClientEventOf +> & + Readonly>> + +const joinedRoomEvents = (frame: SyncFrame) => + Object.entries(frame.rooms?.join ?? {}).flatMap(([roomId, room]) => { + // `state_after` supersedes `state` when requested. Reading one or the + // other prevents the same state event from being delivered twice. + const state = room.stateAfter?.events ?? room.state?.events ?? [] + const timeline = room.timeline?.events ?? [] + + return [...state, ...timeline].map(event => ({ ...event, roomId: roomId as RoomId })) + }) + +/** + * Defines a typed subscription to Matrix room events. + * + * Events are selected from the configured sync bucket, checked against their + * Matrix event type, decoded with the content schema, and finally passed to + * the optional predicate. Invalid content is ignored rather than terminating + * the client's long-running sync subscription. + */ +export const makeEvent = >( + options: MakeEventOptions, +): EventDefinition => { + type Event = ClientEventOf + + const predicate: Predicate = options.predicate ?? (() => true) + const decodeContent = Schema.decodeUnknownOption(options.schema) + + const stream = (frames: Stream.Stream): Stream.Stream => + frames.pipe( + Stream.flatMap(frame => Stream.fromIterable(joinedRoomEvents(frame))), + Stream.filterMap(event => { + if (event.type !== options.type) return Result.fail(event) + + return Result.fromOption(decodeContent(event.content), () => event).pipe( + Result.map(content => ({ ...event, type: options.type, content }) as Event), + ) + }), + Stream.filter(predicate), + ) + + return { + type: options.type, + bucket: options.bucket, + schema: options.schema, + predicate, + stream, + } +} diff --git a/packages/mxfx/src/client/index.ts b/packages/mxfx/src/client/index.ts index 2fd994d..1872090 100644 --- a/packages/mxfx/src/client/index.ts +++ b/packages/mxfx/src/client/index.ts @@ -1 +1,2 @@ export * from './client.ts' +export * from './event.ts' From 4bf911b3856d6dbf8c4ba885a60a075171c7a4f3 Mon Sep 17 00:00:00 2001 From: wouter Date: Fri, 17 Jul 2026 02:53:26 +0200 Subject: [PATCH 2/2] fix: deduplicate joined room events --- packages/mxfx/src/client/event.test.ts | 19 +++++++++++++++++++ packages/mxfx/src/client/event.ts | 15 ++++++++++++--- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/packages/mxfx/src/client/event.test.ts b/packages/mxfx/src/client/event.test.ts index 590d14f..ee1daf6 100644 --- a/packages/mxfx/src/client/event.test.ts +++ b/packages/mxfx/src/client/event.test.ts @@ -42,6 +42,14 @@ const frame = Schema.decodeUnknownSync(endpoints.getSyncV3ResponseSchema)({ originServerTs: 2, sender: '@alice:example.org', }, + { + type: 'com.example.command', + content: { command: 'run', count: '99' }, + eventId: '$state:example.org', + originServerTs: 1, + sender: '@alice:example.org', + stateKey: '', + }, { type: 'com.example.command', content: { command: 'skip', count: '3' }, @@ -91,6 +99,17 @@ describe('makeEvent', () => { }), ) + it.effect('deduplicates state-after events also present in the timeline', () => + Effect.gen(function* () { + const events = yield* Stream.make(frame).pipe(commandEvent.stream, Stream.runCollect) + + expect(Array.from(events).map(event => [event.eventId, event.content.count])).toEqual([ + ['$state:example.org', 1], + ['$run:example.org', 2], + ]) + }), + ) + it('infers the literal event type and decoded content', () => { type Event = Parameters[0] expectTypeOf().toEqualTypeOf<'com.example.command'>() diff --git a/packages/mxfx/src/client/event.ts b/packages/mxfx/src/client/event.ts index df12447..864427a 100644 --- a/packages/mxfx/src/client/event.ts +++ b/packages/mxfx/src/client/event.ts @@ -43,12 +43,21 @@ export type EventDefinition Object.entries(frame.rooms?.join ?? {}).flatMap(([roomId, room]) => { - // `state_after` supersedes `state` when requested. Reading one or the - // other prevents the same state event from being delivered twice. + // `state_after` supersedes `state` when requested. State events precede + // timeline events, so the state copy wins when the same event is present + // in both while timeline-only (including intermediate) events are kept. const state = room.stateAfter?.events ?? room.state?.events ?? [] const timeline = room.timeline?.events ?? [] + const seenEventIds = new Set() - return [...state, ...timeline].map(event => ({ ...event, roomId: roomId as RoomId })) + return [...state, ...timeline] + .filter(event => { + if (seenEventIds.has(event.eventId)) return false + + seenEventIds.add(event.eventId) + return true + }) + .map(event => ({ ...event, roomId: roomId as RoomId })) }) /**