exploration: add typed client event API

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<SyncFrame>) =>
-  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)

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* <T extends EventLike>(eventUnit: EventUnit<T>, f: (event: T) => Effect.Effect<void, never>) {
-    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<T extends EventLike> = {
-  predicate: (frame: Stream.Stream<SyncFrame>) => Stream.Stream<T>
-}
 
 export class MatrixClient extends Context.Service<
   MatrixClient,

packages/mxfx/src/client/event.test.ts

@@ -0,0 +1,130 @@
+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: 'run', count: '99' },
+              eventId: '$state:example.org',
+              originServerTs: 1,
+              sender: '@alice:example.org',
+              stateKey: '',
+            },
+            {
+              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.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<typeof commandEvent.predicate>[0]
+    expectTypeOf<Event['type']>().toEqualTypeOf<'com.example.command'>()
+    expectTypeOf<Event['content']>().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([])
+    }),
+  )
+})

packages/mxfx/src/client/event.ts

@@ -0,0 +1,99 @@
+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<Type extends string, Content> = Omit<typeof ClientEventWithoutRoomId.Type, 'type' | 'content'> & {
+  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<Event extends EventLike> {
+  readonly stream: (frames: Stream.Stream<SyncFrame>) => Stream.Stream<Event>
+}
+
+export interface MakeEventOptions<Type extends string, ContentSchema extends Schema.Decoder<unknown>> {
+  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<ClientEventOf<Type, ContentSchema['Type']>>
+}
+
+export type EventDefinition<Type extends string, ContentSchema extends Schema.Decoder<unknown>> = EventUnit<
+  ClientEventOf<Type, ContentSchema['Type']>
+> &
+  Readonly<Required<MakeEventOptions<Type, ContentSchema>>>
+
+const joinedRoomEvents = (frame: SyncFrame) =>
+  Object.entries(frame.rooms?.join ?? {}).flatMap(([roomId, room]) => {
+    // `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<string>()
+
+    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 }))
+  })
+
+/**
+ * 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 = <const Type extends string, ContentSchema extends Schema.Decoder<unknown>>(
+  options: MakeEventOptions<Type, ContentSchema>,
+): EventDefinition<Type, ContentSchema> => {
+  type Event = ClientEventOf<Type, ContentSchema['Type']>
+
+  const predicate: Predicate<Event> = options.predicate ?? (() => true)
+  const decodeContent = Schema.decodeUnknownOption(options.schema)
+
+  const stream = (frames: Stream.Stream<SyncFrame>): Stream.Stream<Event> =>
+    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,
+  }
+}

packages/mxfx/src/client/index.ts

@@ -1 +1,2 @@
 export * from './client.ts'
+export * from './event.ts'