Fix memory journal returns the last remote sequence instead of the next

.changeset/fix-memory-journal-next-sequence.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Return the first unused remote sequence from the in-memory event journal.

packages/effect/src/unstable/eventlog/EventJournal.ts

@@ -86,7 +86,7 @@ export class EventJournal extends Context.Service<EventJournal, {
   ) => Effect.Effect<A, EventJournalError | E, R>
 
   /**
-   * Retrieve the last known sequence number for a remote source.
+   * Retrieve the first unused sequence number for a remote source.
    */
   readonly nextRemoteSequence: (remoteId: RemoteId) => Effect.Effect<number, EventJournalError>
 
@@ -417,8 +417,8 @@ export const makeMemory: Effect.Effect<EventJournal["Service"]> = Effect.gen(fun
       for (const remoteEntry of options.entries) {
         if (byId.has(remoteEntry.entry.idString)) {
           duplicateEntries.push(remoteEntry.entry)
-          if (remoteEntry.remoteSequence > remote.sequence) {
-            remote.sequence = remoteEntry.remoteSequence
+          if (remoteEntry.remoteSequence >= remote.sequence) {
+            remote.sequence = remoteEntry.remoteSequence + 1
           }
           continue
         }
@@ -456,8 +456,8 @@ export const makeMemory: Effect.Effect<EventJournal["Service"]> = Effect.gen(fun
             target.missing.push(remoteEntry.entry)
           }
         })
-        if (remoteEntry.remoteSequence > remote.sequence) {
-          remote.sequence = remoteEntry.remoteSequence
+        if (remoteEntry.remoteSequence >= remote.sequence) {
+          remote.sequence = remoteEntry.remoteSequence + 1
         }
       }
       journal.sort((a, b) => a.createdAtMillis - b.createdAtMillis)

packages/effect/test/unstable/eventlog/EventJournal.test.ts

@@ -26,6 +26,32 @@ describe("EventJournal", () => {
       assert.deepStrictEqual(sourceMissing, [])
     }))
 
+  it.effect("returns the next unused remote sequence", () =>
+    Effect.gen(function*() {
+      const journal = yield* EventJournal.makeMemory
+      const remoteId = EventJournal.makeRemoteIdUnsafe()
+      const entry = new EventJournal.Entry({
+        id: EventJournal.makeEntryIdUnsafe(),
+        event: "Repro",
+        primaryKey: "key",
+        payload: new Uint8Array()
+      }, { disableChecks: true })
+
+      assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 0)
+      yield* journal.writeFromRemote({
+        remoteId,
+        entries: [new EventJournal.RemoteEntry({ remoteSequence: 0, entry })],
+        effect: () => Effect.void
+      })
+      assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 1)
+      yield* journal.writeFromRemote({
+        remoteId,
+        entries: [new EventJournal.RemoteEntry({ remoteSequence: 5, entry })],
+        effect: () => Effect.void
+      })
+      assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 6)
+    }))
+
   it.effect("records entries in memory and publishes local changes", () =>
     Effect.gen(function*() {
       const journal = yield* EventJournal.EventJournal