Feat/encryption

.gitignore

@@ -1,6 +1,16 @@
-SCHEMA.md
-
+# Generated build artifacts
 dist
+**/*.tsbuildinfo
+
+# Generated dependencies
 node_modules
+
+# Environment specific files
 .env
-db.sqlite
+*.sqlite*
+
+# Apple specific files
+.DS_Store
+
+# scratchpad
+**/scratchpad/**/*

.vscode/mxfx-snippets.code-snippets

@@ -31,30 +31,24 @@
     "body": "Effect.fn(function* () {${0}})",
   },
 
-  "mxfx-get-endpoint": {
+  "mxfx-endpoint": {
     "scope": "typescript",
-    "prefix": "mxfx-endpoint-get",
+    "prefix": "mxfx-endpoint",
     "body": [
-      "import { Schema } from 'effect.ts'",
-      "import { makeEndpoint } from '@/matrix-endpoint.ts'",
+      "import { Schema } from 'effect'",
+      "import { makeEndpoint } from '../endpoint.ts'",
       "",
-      "const responseSchema = Schema.Struct({${0}})",
+      "const schema = Schema.Struct({${0}})",
       "",
       "/**",
-      " * `GET /_matrix/client/${1:path}`",
+      " * `${1:GET} /_matrix/client/${2:path}`",
       " *",
       " * @description",
-      " * ${2:description}",
+      " * ${3:description}",
       " *",
-      " * @see ${3:documentationUrl} ",
+      " * @see ${4:documentationUrl} ",
       " */",
-      "export const ${4:getEndpoint} = () =>",
-      "  makeEndpoint({",
-      "    path: '/${1:path}',",
-      "    method: 'GET',",
-      "    auth: ${5:true},",
-      "    schema: responseSchema,",
-      "  })",
+      "export const ${5:getEndpoint} = () => makeEndpoint('${1:GET}', { auth: ${6:true}, schema })`'/${2:path}'`",
       "",
     ],
   },

examples/02-client.ts

@@ -1,12 +1,12 @@
 import { NodeHttpClient, NodeRuntime } from '@effect/platform-node'
 import { Cause, Config, Effect, Layer, References, Stream } from 'effect'
 import { DevTools } from 'effect/unstable/devtools'
-import { MatrixApi, MatrixAuth, MatrixClient, MatrixConfig, Store } from 'mxfx'
+import { MatrixApi, MatrixAuth, MatrixClient, MatrixConfig } 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>) =>
+const pingPredicate = (frame: Stream.Stream<SyncFrame>) =>
   frame.pipe(
     Stream.flatMap(sync => Stream.fromIterable(Object.entries(sync.rooms?.join ?? {}))),
     Stream.flatMap(([roomId, room]) =>
@@ -22,7 +22,7 @@ const lastPredicate = (frame: Stream.Stream<SyncFrame>) =>
 
         return (
           typeof content.body === 'string' &&
-          content.body.trim().startsWith('!last') &&
+          content.body.trim().startsWith('!ping') &&
           (content.msgtype === 'm.text' || content.msgtype === 'm.notice' || content.msgtype === 'm.emote')
         )
       }),
@@ -32,23 +32,14 @@ const lastPredicate = (frame: Stream.Stream<SyncFrame>) =>
 const program = Effect.gen(function* () {
   const api = yield* MatrixApi.MatrixApi
   const client = yield* MatrixClient.MatrixClient
-  const store = yield* Store.Store
 
   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({ predicate: pingPredicate }, 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)
-      yield* Effect.log(lastMessages)
-
       yield* endpoints
-        .putRoomsSendV3({
-          content: { msgtype: 'm.text', body: lastMessages.map(x => `${x.content['body']}`).join('\n\n') },
-          eventType: 'm.room.message',
-          roomId: event.roomId,
-        })
+        .putRoomsSendV3({ content: { msgtype: 'm.text', body: 'pong' }, eventType: 'm.room.message', roomId: event.roomId })
         .pipe(Effect.andThen(api.execute))
     }).pipe(Effect.catchCause(cause => Effect.log(cause))),
   )
@@ -63,7 +54,7 @@ const program = Effect.gen(function* () {
 
 const mxfxLive = MatrixClient.layerMatrixClient.pipe(
   Layer.provideMerge(MatrixApi.layer),
-  Layer.provideMerge(MatrixAuth.layerLegacyConfig({ accessToken: Config.string('MATRIX_ACCESS_TOKEN') })),
+  Layer.provideMerge(MatrixAuth.layerAccessToken({ accessToken: Config.string('MATRIX_ACCESS_TOKEN') })),
   Layer.provideMerge(MatrixConfig.layerConfig({ serverName: Config.string('MATRIX_HOME_SERVER') })),
   Layer.provideMerge(NodeHttpClient.layerNodeHttp),
   Layer.provideMerge(DevTools.layer()),

examples/03-encrypted.ts

@@ -0,0 +1,155 @@
+import { NodeHttpClient, NodeRuntime } from '@effect/platform-node'
+import { Crypto, CryptoNode } from '@mxfx/crypto-node'
+import { Effect, Layer, Option, PubSub, Duration, Stream, Cause, Config, Record, References, Redacted, Schema } from 'effect'
+import { Filter, Kv, MatrixApi, MatrixAuth, MatrixConfig, Event } from 'mxfx'
+import { endpoints } from 'mxfx/api'
+
+const emptyFilter = Filter.make({
+  room: {
+    timeline: { limit: 10, lazyLoadMembers: false },
+    state: { lazyLoadMembers: false },
+    accountData: { lazyLoadMembers: false, notTypes: ['*'] },
+    ephemeral: { lazyLoadMembers: false, notTypes: ['*'] },
+  },
+  presence: { notTypes: ['*'] },
+  accountData: { notTypes: ['*'] },
+})
+
+type SyncFrame = typeof endpoints.getSyncV3ResponseSchema.Type
+
+const syncLoop = Effect.gen(function* () {
+  const api = yield* MatrixApi.MatrixApi
+  const kv = yield* Kv.Kv
+  const crypto = yield* Crypto.Crypto
+
+  const { userId, deviceId } = yield* endpoints.getAccountWhoami().pipe(Effect.andThen(api.execute))
+
+  if (!deviceId) return yield* Effect.die('No deviceId')
+
+  const machine = yield* crypto.makeMachine({
+    userId,
+    deviceId: deviceId,
+    storage: { type: 'sqlite', passphrase: Redacted.make('passphrase'), path: './03-encrypted-db' },
+  })
+
+  const syncHub = yield* PubSub.unbounded<SyncFrame>()
+  const syncStream = Stream.fromPubSub(syncHub)
+
+  const syncOnce = kv.getString('syncToken').pipe(
+    Effect.flatMap(nextBatch =>
+      endpoints
+        .getSyncV3({
+          useStateAfter: true,
+          setPresence: 'online',
+          timeout: Duration.seconds(30),
+          since: nextBatch.pipe(Option.getOrUndefined),
+          fullState: nextBatch.pipe(Option.isNone),
+          filter: nextBatch.pipe(Option.match({ onNone: () => emptyFilter, onSome: () => undefined })),
+        })
+        .pipe(
+          Effect.andThen(api.execute),
+          Effect.tap(sync => crypto.receiveSyncChanges(machine, sync)),
+          Effect.tap(sync => PubSub.publish(syncHub, sync)),
+          Effect.tap(() => crypto.sendOutgoingRequests(machine)),
+          Effect.tap(sync => kv.set('syncToken', sync.nextBatch)),
+        ),
+    ),
+  )
+
+  yield* Effect.forever(syncOnce).pipe(
+    Effect.catchCause(cause => Effect.logError(Cause.pretty(cause))),
+    Effect.forkDetach(),
+  )
+
+  return { eventStream: syncStream, machine }
+})
+
+const handleMessages = Effect.fn(function* (f: SyncFrame) {
+  if (!f.rooms?.join) return
+  const joinedRooms = f.rooms.join
+
+  const keys = Record.keys(f.rooms?.join)
+  yield* Effect.forEach(keys, key =>
+    Effect.gen(function* () {
+      const room = joinedRooms[key]
+
+      if (!room?.timeline?.events) return
+      const events = room.timeline.events
+
+      yield* Effect.forEach(events, event =>
+        Effect.gen(function* () {
+          if (event.type !== 'm.room.message') return
+          yield* Effect.log('new message!', { content: event.content })
+        }),
+      )
+    }),
+  )
+})
+
+const handleEncryptedEvents = Effect.fn(function* (f: SyncFrame, machine: Crypto.Machine) {
+  const crypto = yield* Crypto.Crypto
+
+  if (!f.rooms?.join) return
+  const joinedRooms = f.rooms.join
+  const roomIds = Record.keys(f.rooms?.join)
+  yield* Effect.forEach(roomIds, roomId =>
+    Effect.gen(function* () {
+      const room = joinedRooms[roomId]
+
+      if (!room?.timeline?.events) return
+      const events = room.timeline.events
+
+      yield* Effect.forEach(events, event =>
+        Effect.gen(function* () {
+          if (event.type !== 'm.room.encrypted') return
+
+          const wireEvent = yield* Schema.encodeUnknownEffect(
+            Event.roomEventWithoutRoomId.pipe(MatrixApi.EncodeCase.encodeSnakeCaseSchema),
+          )(event)
+
+          const decrypted = yield* crypto.decryptRoomEvent(machine, JSON.stringify(wireEvent), roomId)
+
+          yield* Effect.log('new encrypted event!', { content: event.content, decrypted: decrypted.event })
+        }),
+      )
+    }),
+  )
+})
+
+const handleInvites = Effect.fn(function* (f: SyncFrame) {
+  const api = yield* MatrixApi.MatrixApi
+
+  if (!f.rooms?.invite) return
+  const invitedRooms = f.rooms?.invite
+
+  const keys = Record.keys(invitedRooms)
+  yield* Effect.forEach(keys, roomId => endpoints.postRoomsJoinV3({ roomId }).pipe(Effect.andThen(api.execute)))
+})
+
+const program = Effect.gen(function* () {
+  const { eventStream, machine } = yield* syncLoop
+
+  yield* eventStream.pipe(
+    Stream.onFirst(() => Effect.log('Client Started')),
+    Stream.runHead,
+  )
+
+  yield* Effect.all(
+    [
+      eventStream.pipe(Stream.runForEach(handleInvites)), //
+      eventStream.pipe(Stream.runForEach(handleMessages)),
+      eventStream.pipe(Stream.runForEach(sync => handleEncryptedEvents(sync, machine))),
+    ],
+    { concurrency: 'unbounded' },
+  )
+})
+
+const matrixLayer = CryptoNode.layer.pipe(
+  Layer.provideMerge(MatrixApi.layer),
+  Layer.provideMerge(MatrixAuth.layerAccessToken({ accessToken: Config.string('MATRIX_ACCESS_TOKEN') })),
+  Layer.provide(MatrixConfig.layerConfig({ serverName: Config.string('MATRIX_HOME_SERVER') })),
+  Layer.provide(NodeHttpClient.layerNodeHttp),
+  Layer.provideMerge(Layer.succeed(References.MinimumLogLevel, 'Debug')),
+)
+
+NodeRuntime.runMain(program.pipe(Effect.provide(matrixLayer), Effect.scoped))

examples/package.json

@@ -1,12 +1,10 @@
 {
-  "name": "example",
-  "version": "1.0.0",
-  "keywords": [],
-  "author": "",
+  "private": "true",
   "type": "module",
   "dependencies": {
-    "@effect/platform-node": "4.0.0-beta.84",
-    "effect": "4.0.0-beta.84",
+    "@effect/platform-node": "4.0.0-beta.99",
+    "@mxfx/crypto-node": "workspace:^",
+    "effect": "4.0.0-beta.99",
     "mxfx": "workspace:^"
   }
 }

package.json

@@ -10,6 +10,7 @@
     "test": "vitest"
   },
   "devDependencies": {
+    "@effect/vitest": "4.0.0-beta.99",
     "oxfmt": "^0.55.0",
     "oxlint": "^1.70.0",
     "typescript": "^6.0.3",

packages/crypto-node/package.json

@@ -0,0 +1,26 @@
+{
+  "name": "@mxfx/crypto-node",
+  "version": "0.0.1",
+  "license": "MIT",
+  "author": "Wouter de Bruijn <wouter@debruijn.dev>",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/wouter173/mxfx.git",
+    "directory": "packages/crypto-node"
+  },
+  "type": "module",
+  "sideEffects": [],
+  "exports": {
+    ".": "./src/index.ts"
+  },
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "dependencies": {
+    "@matrix-org/matrix-sdk-crypto-nodejs": "^0.6.1"
+  },
+  "peerDependencies": {
+    "effect": "4.0.0-beta.99",
+    "mxfx": "workspace:*"
+  }
+}