feat: switch out store for kv

.gitignore

@@ -1,6 +1,9 @@
-SCHEMA.md
-
 dist
+**/*.tsbuildinfo
+
 node_modules
+
 .env
 db.sqlite
+
+.DS_Store

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,20 +32,15 @@ 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') },
+          content: { msgtype: 'm.text', body: 'pong' },
           eventType: 'm.room.message',
           roomId: event.roomId,
         })

packages/mxfx/package.json

@@ -18,6 +18,7 @@
     "./config": "./src/config/index.ts",
     "./vault": "./src/vault/index.ts",
     "./branded": "./src/branded/index.ts",
+    "./kv": "./src/kv/index.ts",
     ".": "./src/index.ts"
   },
   "scripts": {

packages/mxfx/src/client/client.ts

@@ -2,19 +2,17 @@ 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 { Reducers } from './sync/reducer.ts'
+import { Kv } from '../kv/index.ts'
 
 const make = Effect.gen(function* () {
   const matrixApi = yield* MatrixApi
 
-  const reducers = yield* Reducers
-  const store = yield* Store
+  const kv = yield* Kv
 
   const syncHub = yield* PubSub.unbounded<SyncFrame>()
   const syncStream = Stream.fromPubSub(syncHub)
 
-  const syncOnce = store.getNextBatch().pipe(
+  const syncOnce = kv.getString('syncToken').pipe(
     Effect.flatMap(nextBatch =>
       endpoints
         .getSyncV3({
@@ -41,7 +39,7 @@ const make = Effect.gen(function* () {
         .pipe(
           Effect.andThen(matrixApi.execute),
           Effect.tap(syncResponse => PubSub.publish(syncHub, syncResponse)),
-          Effect.tap(syncResponse => Effect.forEach(reducers, reducer => reducer.reduce(syncResponse))),
+          Effect.andThen(syncResponse => kv.set('syncToken', syncResponse.nextBatch)),
         ),
     ),
   )
@@ -72,4 +70,4 @@ export class MatrixClient extends Context.Service<
   }
 >()('mxfx/client') {}
 
-export const layerMatrixClient = Layer.effect(MatrixClient, make).pipe(Layer.provideMerge(Reducers.layer))
+export const layerMatrixClient = Layer.effect(MatrixClient, make)

packages/mxfx/src/index.ts

@@ -3,5 +3,5 @@ export * as MatrixApi from './api/index.ts'
 export * as MatrixClient from './client/index.ts'
 export * as MatrixConfig from './config/index.ts'
 export * as Vault from './vault/index.ts'
-export * as Store from './store/index.ts'
+export * as Kv from './kv/index.ts'
 export * from './branded/index.ts'

packages/mxfx/src/kv/in-memory-kv.ts

@@ -0,0 +1,16 @@
+import { Effect, Schema } from 'effect'
+
+import { KvGetError, type KvShape } from './kv.ts'
+
+export const makeInMemoryKv = (): KvShape => {
+  const kv = new Map()
+
+  return {
+    set: (key: string, value: string) => Effect.sync(() => kv.set(key, value)),
+    getString: (key: string) =>
+      Effect.try(() => kv.get(key)).pipe(
+        Effect.andThen(Schema.decodeUnknownEffect(Schema.OptionFromOptional(Schema.String))),
+        Effect.mapError(e => new KvGetError({ cause: e })),
+      ),
+  }
+}

packages/mxfx/src/kv/index.ts

@@ -0,0 +1,2 @@
+export * from './kv.ts'
+export * from './in-memory-kv.ts'

packages/mxfx/src/kv/kv.ts

@@ -0,0 +1,15 @@
+import { Context, Data, Effect, Option } from 'effect'
+
+import { makeInMemoryKv } from './in-memory-kv.ts'
+
+export class KvGetError extends Data.TaggedError('mxfx/kv/get-error')<{
+  cause?: unknown
+}> {}
+
+export type KvShape = {
+  set: (key: string, value: string) => Effect.Effect<void>
+  getString: (key: string) => Effect.Effect<Option.Option<string>, KvGetError>
+}
+
+export const Kv = Context.Reference<KvShape>('mxfx/kv', { defaultValue: makeInMemoryKv })
+export type Kv = typeof Kv