make it reconnect at prev cursor

Author: kristojorg

src/LabelWatcher.ts

@@ -16,11 +16,18 @@ const run = Effect.gen(function* () {
   const connect = yield* RetryingSocket;
   const agent = yield* AtpListAccountAgent;
   const cursor = yield* Cursor;
-  const initialCursor = yield* cursor.get;
   const { labelerSocketUrl } = yield* Env;
-  labelerSocketUrl.searchParams.set("cursor", initialCursor.toString());
 
-  const stream = connect({ url: labelerSocketUrl });
+  // Effect that builds URL with current cursor - re-evaluated on each reconnect
+  const getUrl = Effect.gen(function* () {
+    const currentCursor = yield* cursor.get;
+    const url = new URL(labelerSocketUrl.toString());
+    url.searchParams.set("cursor", currentCursor.toString());
+    yield* Effect.log(`Building connection URL with cursor: ${currentCursor}`);
+    return url;
+  });
+
+  const stream = connect({ getUrl });
 
   const runStream = stream.pipe(
     Stream.mapEffect(parseMessage),

src/RetryingSocket.ts

@@ -1,51 +1,49 @@
-import { Socket } from "@effect/platform"
-import type { SocketError } from "@effect/platform/Socket"
-import { Data, Effect, Schedule, Stream } from "effect"
+import { Socket } from "@effect/platform";
+import type { SocketError } from "@effect/platform/Socket";
+import { Cause, Effect, Schedule, Stream } from "effect";
 
 /**
- * A stream that will reconnect to the websocket on error
+ * A stream that will reconnect to the websocket on error.
+ * Takes an Effect that produces the URL so it can re-evaluate
+ * (e.g., get current cursor) on each reconnection attempt.
  */
-const wsStream = ({ url }: { url: string | URL }) =>
-  Stream.asyncPush<Uint8Array<ArrayBufferLike>, SocketError>((emit) =>
-    Effect.gen(function*() {
-      yield* Effect.log("Connecting to websocket at: ", url.toString())
+const wsStream = <R>({ getUrl }: { getUrl: Effect.Effect<URL, never, R> }) =>
+  Stream.asyncPush<Uint8Array<ArrayBufferLike>, SocketError, R>((emit) =>
+    Effect.gen(function* () {
+      const url = yield* getUrl;
+      yield* Effect.log("Connecting to websocket at: ", url.toString());
       const socket = yield* Socket.makeWebSocket(url.toString(), {
         closeCodeIsError: (_) => true,
-      })
+      });
 
       const e = socket
         .run((d) =>
-          Effect.gen(function*() {
-            const didEmit = emit.single(d)
+          Effect.gen(function* () {
+            const didEmit = emit.single(d);
             if (!didEmit) {
+              // this doesn't seem to work because asyncPush uses an unbounded queue internally
+              // and thus it only returns false when done.
               yield* new BufferOverflowError({
                 message: "Socket buffer overflowed, failed to emit a message.",
-              })
+              });
             }
           })
         )
         .pipe(
           Effect.catchTag("SocketError", (e) => Effect.succeed(emit.fail(e))),
-          Effect.fork,
-        )
+          Effect.fork
+        );
 
-      yield* e
+      yield* e;
     }).pipe(Effect.provide(Socket.layerWebSocketConstructorGlobal))
   ).pipe(
     Stream.tapErrorCause(Effect.logError),
-    Stream.retry(Schedule.spaced("1 second")),
-  )
-
-export class BufferOverflowError extends Data.TaggedError(
-  "BufferOverflowError",
-)<{
-  message: string
-  cause?: unknown
-}> {}
+    Stream.retry(Schedule.spaced("1 second"))
+  );
 
 export class RetryingSocket extends Effect.Service<RetryingSocket>()(
   "RetryingSocket",
   {
     succeed: wsStream,
-  },
+  }
 ) {}