mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-16 23:48:27 +00:00
1338 lines
47 KiB
TypeScript
1338 lines
47 KiB
TypeScript
import { describe, expect } from "bun:test"
|
|
import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Ref, Schema, Stream } from "effect"
|
|
import { EventV2 } from "@opencode-ai/core/event"
|
|
import { Event } from "@opencode-ai/schema/event"
|
|
import { EventManifest } from "@opencode-ai/schema/event-manifest"
|
|
import { McpEvent } from "@opencode-ai/schema/mcp-event"
|
|
import { Session } from "@opencode-ai/schema/session"
|
|
import { SessionEvent } from "@opencode-ai/schema/session-event"
|
|
import { SessionV1 } from "@opencode-ai/schema/session-v1"
|
|
import { Database } from "@opencode-ai/core/database/database"
|
|
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
|
|
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
|
|
import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
|
|
import { Location } from "@opencode-ai/core/location"
|
|
import { AbsolutePath } from "@opencode-ai/core/schema"
|
|
import { WorkspaceV2 } from "@opencode-ai/core/workspace"
|
|
import { eq } from "drizzle-orm"
|
|
import { location } from "./fixture/location"
|
|
import { testEffect } from "./lib/effect"
|
|
|
|
const locationLayer = Layer.succeed(
|
|
Location.Service,
|
|
Location.Service.of(
|
|
location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }),
|
|
),
|
|
)
|
|
const Message = EventV2.ephemeral({
|
|
type: "test.message",
|
|
schema: {
|
|
text: Schema.String,
|
|
},
|
|
})
|
|
|
|
const SyncMessage = EventV2.durable({
|
|
type: "test.sync",
|
|
durable: {
|
|
version: 1,
|
|
aggregate: "id",
|
|
},
|
|
schema: {
|
|
id: Schema.String,
|
|
text: Schema.String,
|
|
},
|
|
})
|
|
|
|
const SyncSent = EventV2.durable({
|
|
type: "test.sent",
|
|
durable: {
|
|
version: 1,
|
|
aggregate: "messageID",
|
|
},
|
|
schema: {
|
|
messageID: Schema.String,
|
|
text: Schema.String,
|
|
},
|
|
})
|
|
|
|
const GlobalMessage = EventV2.ephemeral({
|
|
type: "test.global",
|
|
schema: {
|
|
text: Schema.String,
|
|
},
|
|
})
|
|
const CountMessage = EventV2.ephemeral({
|
|
type: "test.count",
|
|
schema: {
|
|
count: Schema.Number,
|
|
},
|
|
})
|
|
|
|
const VersionedMessage = EventV2.durable({
|
|
type: "test.versioned",
|
|
durable: {
|
|
version: 2,
|
|
aggregate: "id",
|
|
},
|
|
schema: {
|
|
id: Schema.String,
|
|
text: Schema.String,
|
|
},
|
|
})
|
|
|
|
const DurableMessage = SessionV1.Event.MessageRemoved
|
|
const durableData = (sessionID: Session.ID, text: string) => ({
|
|
sessionID,
|
|
messageID: SessionV1.MessageID.ascending(`msg_${text}`),
|
|
})
|
|
|
|
/** Followed log read without markers: the old `durable` stream shape. */
|
|
const tail = (events: EventV2.Interface, input: { aggregateID: string; after?: number }) =>
|
|
events.log({ ...input, follow: true }).pipe(Stream.filter((item): item is EventV2.Payload => !EventV2.isSynced(item)))
|
|
|
|
const it = testEffect(
|
|
AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]),
|
|
)
|
|
const itWithoutLocation = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node])))
|
|
|
|
describe("EventV2", () => {
|
|
it.effect("subscribes to multiple event definitions with a discriminated payload union", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
// @ts-expect-error multi-definition subscriptions require at least one definition
|
|
events.subscribe([])
|
|
const fiber = yield* events
|
|
.subscribe([Message, CountMessage])
|
|
.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
|
|
yield* events.publish(Message, { text: "hello" })
|
|
yield* events.publish(CountMessage, { count: 2 })
|
|
|
|
const received = Array.from(yield* Fiber.join(fiber)).map((event) =>
|
|
event.type === "test.message" ? event.data.text : event.data.count,
|
|
)
|
|
expect(received).toEqual(["hello", 2])
|
|
}),
|
|
)
|
|
|
|
it.effect("publishes events with the current location", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const fiber = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
const event = yield* events.publish(Message, { text: "hello" })
|
|
const received = Array.from(yield* Fiber.join(fiber))
|
|
|
|
expect(received).toEqual([event])
|
|
expect(event.type).toBe("test.message")
|
|
expect(event).not.toHaveProperty("version")
|
|
expect(event.data).toEqual({ text: "hello" })
|
|
expect(event.location).toEqual({
|
|
directory: AbsolutePath.make("project"),
|
|
workspaceID: WorkspaceV2.ID.make("wrk_test"),
|
|
})
|
|
}),
|
|
)
|
|
|
|
itWithoutLocation.effect("omits location when no location is available", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const event = yield* events.publish(GlobalMessage, { text: "hello" })
|
|
|
|
expect(event).not.toHaveProperty("location")
|
|
expect(event.type).toBe("test.global")
|
|
}),
|
|
)
|
|
|
|
it.effect("publishes definition version", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const event = yield* events.publish(VersionedMessage, { id: "one", text: "hello" })
|
|
|
|
expect(event.type).toBe("test.versioned")
|
|
expect(event.durable?.version).toBe(EventV2.Version.make(2))
|
|
}),
|
|
)
|
|
|
|
it.effect("selects the latest durable definition independent of declaration order", () =>
|
|
Effect.sync(() => {
|
|
const latest = EventV2.durable({
|
|
type: "test.out-of-order",
|
|
durable: { version: 2, aggregate: "id" },
|
|
schema: { id: Schema.String },
|
|
})
|
|
const historical = EventV2.durable({
|
|
type: "test.out-of-order",
|
|
durable: { version: 1, aggregate: "id" },
|
|
schema: { id: Schema.String },
|
|
})
|
|
|
|
expect(Event.latest([latest, historical]).get("test.out-of-order")).toBe(latest)
|
|
expect(Event.latest([historical, latest]).get("test.out-of-order")).toBe(latest)
|
|
}),
|
|
)
|
|
|
|
it.effect("publishes to typed and wildcard subscriptions", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const typed = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
const wildcard = yield* events.subscribe().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
const event = yield* events.publish(Message, { text: "hello" })
|
|
|
|
expect(Array.from(yield* Fiber.join(typed))).toEqual([event])
|
|
expect(Array.from(yield* Fiber.join(wildcard))).toEqual([event])
|
|
}),
|
|
)
|
|
|
|
it.effect("runs projectors inline", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
yield* events.project(SyncMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
|
|
const event = yield* events.publish(SyncMessage, { id: "one", text: "hello" })
|
|
yield* events.publish(SyncMessage, { id: "one", text: "after unsubscribe" })
|
|
|
|
expect(received[0]).toEqual(event)
|
|
expect(received[1]?.data).toEqual({ id: "one", text: "after unsubscribe" })
|
|
}),
|
|
)
|
|
|
|
it.effect("commits local operational state inside a new durable event transaction", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<string>()
|
|
const aggregateID = EventV2.ID.create()
|
|
yield* events.project(SyncMessage, () => Effect.sync(() => received.push("projector")))
|
|
|
|
yield* events.publish(
|
|
SyncMessage,
|
|
{ id: aggregateID, text: "hello" },
|
|
{ commit: (seq) => Effect.sync(() => received.push(`commit:${seq}`)) },
|
|
)
|
|
|
|
expect(received).toEqual(["projector", "commit:0"])
|
|
}),
|
|
)
|
|
|
|
it.effect("rolls back the durable event and projector when the local commit fails", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
yield* db.run("CREATE TABLE IF NOT EXISTS event_commit_probe (value text NOT NULL)")
|
|
yield* db.run("DELETE FROM event_commit_probe")
|
|
yield* events.project(SyncMessage, () =>
|
|
db.run("INSERT INTO event_commit_probe (value) VALUES ('projected')").pipe(Effect.orDie, Effect.asVoid),
|
|
)
|
|
|
|
const exit = yield* events
|
|
.publish(SyncMessage, { id: aggregateID, text: "hello" }, { commit: () => Effect.die("commit failed") })
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("commit failed")
|
|
expect(yield* db.all("SELECT value FROM event_commit_probe")).toEqual([])
|
|
expect(yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all()).toEqual([])
|
|
expect(
|
|
yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).all(),
|
|
).toEqual([])
|
|
}),
|
|
)
|
|
|
|
it.effect("rejects local commit hooks on live-only events", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const exit = yield* events.publish(Message, { text: "hello" }, { commit: () => Effect.void }).pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Local commit hooks require a durable event")
|
|
}),
|
|
)
|
|
|
|
it.effect("runs projectors before publishing to streams", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<string>()
|
|
const fiber = yield* events.subscribe().pipe(
|
|
Stream.take(1),
|
|
Stream.runForEach(() => Effect.sync(() => received.push("stream"))),
|
|
Effect.forkScoped,
|
|
)
|
|
yield* events.project(SyncMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event.type)
|
|
}),
|
|
)
|
|
|
|
yield* Effect.yieldNow
|
|
yield* events.publish(SyncMessage, { id: "one", text: "hello" })
|
|
yield* Fiber.join(fiber)
|
|
|
|
expect(received).toEqual([SyncMessage.type, "stream"])
|
|
}),
|
|
)
|
|
|
|
it.effect("runs listeners inline after projectors", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<string>()
|
|
yield* events.project(SyncMessage, () =>
|
|
Effect.sync(() => {
|
|
received.push("projector")
|
|
}),
|
|
)
|
|
const unsubscribe = yield* events.listen(() =>
|
|
Effect.sync(() => {
|
|
received.push("listener")
|
|
}),
|
|
)
|
|
|
|
yield* events.publish(SyncMessage, { id: "one", text: "hello" })
|
|
yield* unsubscribe
|
|
yield* events.publish(SyncMessage, { id: "one", text: "after unsubscribe" })
|
|
|
|
expect(received).toEqual(["projector", "listener", "projector"])
|
|
}),
|
|
)
|
|
|
|
it.effect("isolates observer defects after durable events commit", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<string>()
|
|
yield* events.listen(() => {
|
|
throw new Error("listener defect")
|
|
})
|
|
yield* events.listen((event) =>
|
|
Effect.sync(() => {
|
|
received.push(event.type)
|
|
}),
|
|
)
|
|
|
|
const event = yield* events.publish(SyncMessage, { id: "one", text: "hello" })
|
|
|
|
expect(received).toEqual([SyncMessage.type])
|
|
expect(event.durable?.seq).toBeNumber()
|
|
}),
|
|
)
|
|
|
|
it.effect("notifies global listeners only after a durable event is committed", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
const observed = new Array<{ id: string; seq: number }>()
|
|
yield* events.listen((event) =>
|
|
event.type !== SyncMessage.type
|
|
? Effect.void
|
|
: db
|
|
.select({ id: EventTable.id, seq: EventTable.seq })
|
|
.from(EventTable)
|
|
.where(eq(EventTable.id, event.id))
|
|
.get()
|
|
.pipe(
|
|
Effect.orDie,
|
|
Effect.tap((row) =>
|
|
Effect.sync(() => {
|
|
if (row) observed.push(row)
|
|
}),
|
|
),
|
|
Effect.asVoid,
|
|
),
|
|
)
|
|
|
|
const event = yield* events.publish(SyncMessage, { id: aggregateID, text: "committed" })
|
|
if (!event.durable) throw new Error("Expected durable event metadata")
|
|
|
|
expect(observed).toEqual([{ id: event.id, seq: event.durable.seq }])
|
|
}),
|
|
)
|
|
|
|
it.effect("ends only an overflowing bounded subscriber without blocking other listeners", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const consuming = yield* Deferred.make<void>()
|
|
const release = yield* Deferred.make<void>()
|
|
const slowStream = yield* EventV2.liveBounded(events, { capacity: 1 })
|
|
const fastStream = yield* EventV2.liveBounded(events, { capacity: 8 })
|
|
const slow = yield* slowStream.pipe(
|
|
Stream.runForEach(() => Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Deferred.await(release)))),
|
|
Effect.forkScoped,
|
|
)
|
|
const fast = yield* fastStream.pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
|
|
|
|
yield* events.publish(Message, { text: "one" })
|
|
yield* Deferred.await(consuming)
|
|
yield* events.publish(Message, { text: "two" })
|
|
yield* events.publish(Message, { text: "overflow" })
|
|
const last = yield* events.publish(Message, { text: "still delivered" })
|
|
yield* Deferred.succeed(release, undefined)
|
|
|
|
const slowExit = yield* Fiber.await(slow)
|
|
expect(Exit.findErrorOption(slowExit).pipe(Option.getOrUndefined)).toBeInstanceOf(EventV2.SubscriberOverflowError)
|
|
expect(Array.from(yield* Fiber.join(fast))).toEqual([
|
|
expect.objectContaining({ data: { text: "one" } }),
|
|
expect.objectContaining({ data: { text: "two" } }),
|
|
expect.objectContaining({ data: { text: "overflow" } }),
|
|
last,
|
|
])
|
|
}),
|
|
)
|
|
|
|
it.effect("filters internal events before they enter a bounded server stream", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const stream = yield* EventV2.liveBounded(events, { capacity: 1, accept: EventManifest.isServer })
|
|
const received = yield* stream.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
|
|
yield* events.publish(McpEvent.ToolsChanged, { server: "one" })
|
|
yield* events.publish(McpEvent.ToolsChanged, { server: "two" })
|
|
const published = yield* events.publish(McpEvent.StatusChanged, { server: "example" })
|
|
|
|
expect(Array.from(yield* Fiber.join(received))).toEqual([published])
|
|
}),
|
|
)
|
|
|
|
it.effect("preserves observer interruption", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
yield* events.listen(() => Effect.interrupt)
|
|
|
|
const exit = yield* events.publish(SyncMessage, { id: "interrupted", text: "hello" }).pipe(Effect.exit)
|
|
const committed = yield* db
|
|
.select({ id: EventTable.id })
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, "interrupted"))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause)).toBeTrue()
|
|
expect(committed).toBeDefined()
|
|
}),
|
|
)
|
|
|
|
it.effect("keeps live-only listener defects fail-fast", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const defect = new Error("listener defect")
|
|
yield* events.listen(() => Effect.die(defect))
|
|
|
|
expect(yield* events.publish(Message, { text: "hello" }).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
|
|
}),
|
|
)
|
|
|
|
it.effect("inserts durable event rows on publish", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
|
|
yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows).toHaveLength(1)
|
|
expect(rows[0]?.type).toBe(EventV2.versionedType(SyncMessage.type, 1))
|
|
expect(rows[0]?.aggregate_id).toBe(aggregateID)
|
|
}),
|
|
)
|
|
|
|
it.effect("increments durable event seq per aggregate", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
|
|
yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
|
|
yield* events.publish(SyncMessage, { id: aggregateID, text: "second" })
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows.map((row) => row.seq)).toEqual([0, 1])
|
|
}),
|
|
)
|
|
|
|
it.effect("replays durable aggregate events after a sequence and tails new events", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
const fiber = yield* tail(events, { aggregateID, after: 0 }).pipe(
|
|
Stream.take(2),
|
|
Stream.runCollect,
|
|
Effect.forkScoped,
|
|
)
|
|
yield* Effect.yieldNow
|
|
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "two"))
|
|
|
|
expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
|
|
[1, durableData(aggregateID, "one")],
|
|
[2, durableData(aggregateID, "two")],
|
|
])
|
|
}),
|
|
)
|
|
|
|
it.effect("catches durable aggregate events published during replay handoff", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
|
|
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
|
|
expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
|
|
[0, durableData(aggregateID, "zero")],
|
|
[1, durableData(aggregateID, "one")],
|
|
])
|
|
}),
|
|
)
|
|
|
|
it.effect("retains a durable wake committed while historical replay is paused", () =>
|
|
Effect.gen(function* () {
|
|
const readStarted = yield* Deferred.make<void>()
|
|
const continueRead = yield* Deferred.make<void>()
|
|
let pause = true
|
|
const eventLayer = EventV2.layerWith({
|
|
beforeAggregateRead: () =>
|
|
pause
|
|
? Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(continueRead)))
|
|
: Effect.void,
|
|
}).pipe(Layer.provide(LayerNode.compile(Database.node)))
|
|
|
|
yield* Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
yield* Deferred.await(readStarted)
|
|
|
|
pause = false
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "during handoff"))
|
|
yield* Deferred.succeed(continueRead, undefined)
|
|
|
|
expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
|
|
[0, durableData(aggregateID, "during handoff")],
|
|
])
|
|
}).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer)))
|
|
}),
|
|
)
|
|
|
|
it.effect("coalesces durable aggregate wakes while draining every committed event", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const count = 64
|
|
const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(count), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
|
|
for (let index = 0; index < count; index++) {
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, String(index)))
|
|
}
|
|
|
|
expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual(
|
|
Array.from({ length: count }, (_, index) => [index, durableData(aggregateID, String(index))]),
|
|
)
|
|
}),
|
|
)
|
|
|
|
it.effect("omits live-only events from durable aggregate streams", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
|
|
yield* events.publish(Message, { text: "live only" })
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "durable"))
|
|
|
|
expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.type)).toEqual([DurableMessage.type])
|
|
}),
|
|
)
|
|
|
|
it.effect("uses custom sync aggregate field", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
|
|
yield* events.publish(SyncSent, { messageID: aggregateID, text: "sent" })
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows).toHaveLength(1)
|
|
expect(rows[0]?.aggregate_id).toBe(aggregateID)
|
|
}),
|
|
)
|
|
|
|
it.effect("replays durable events through projectors", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
yield* events.project(DurableMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
const aggregateID = Session.ID.create()
|
|
|
|
yield* events.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "hello"),
|
|
})
|
|
|
|
expect(received[0]?.type).toBe(DurableMessage.type)
|
|
expect(received[0]?.data).toEqual(durableData(aggregateID, "hello"))
|
|
}),
|
|
)
|
|
|
|
it.effect("replay inserts external event rows", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
|
|
yield* events.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "replayed"),
|
|
})
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows).toHaveLength(1)
|
|
expect(rows[0]?.aggregate_id).toBe(aggregateID)
|
|
}),
|
|
)
|
|
|
|
it.effect(
|
|
"replay rejects an envelope aggregate that differs from its payload without mutating the payload aggregate",
|
|
() =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const envelopeAggregateID = Session.ID.create()
|
|
const payloadAggregateID = Session.ID.create()
|
|
const received = new Array<EventV2.Payload>()
|
|
yield* events.publish(DurableMessage, durableData(payloadAggregateID, "seed"))
|
|
yield* events.project(DurableMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
|
|
const exit = yield* events
|
|
.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID: envelopeAggregateID,
|
|
data: durableData(payloadAggregateID, "replayed"),
|
|
})
|
|
.pipe(Effect.exit)
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, payloadAggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
const sequence = yield* db
|
|
.select({ seq: EventSequenceTable.seq })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, payloadAggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(String(exit)).toContain("Aggregate mismatch")
|
|
expect(received).toHaveLength(0)
|
|
expect(rows).toHaveLength(1)
|
|
expect(sequence).toEqual({ seq: 0 })
|
|
}),
|
|
)
|
|
|
|
it.effect("replay defects on sequence mismatch", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
|
|
yield* events.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "first"),
|
|
})
|
|
const exit = yield* events
|
|
.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 5,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "bad"),
|
|
})
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Sequence mismatch")
|
|
}),
|
|
)
|
|
|
|
it.effect("replay decodes synchronized transformed values before projection", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const received = new Array<typeof SessionEvent.InstructionsUpdated.Type>()
|
|
yield* events.project(SessionEvent.InstructionsUpdated, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
|
|
yield* events.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(SessionEvent.InstructionsUpdated.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: { sessionID: aggregateID, text: "context" },
|
|
})
|
|
|
|
expect(received[0]?.created).toEqual(DateTime.makeUnsafe(0))
|
|
}),
|
|
)
|
|
|
|
it.effect("replay defects on unknown event type", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const exit = yield* events
|
|
.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: "unknown.event.1",
|
|
seq: 0,
|
|
aggregateID: EventV2.ID.create(),
|
|
data: {},
|
|
})
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Unknown durable event type")
|
|
}),
|
|
)
|
|
|
|
it.effect("replayAll validates contiguous aggregate events", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const source = yield* events.replayAll([
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "one"),
|
|
},
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "two"),
|
|
},
|
|
])
|
|
|
|
expect(source).toBe(aggregateID)
|
|
}),
|
|
)
|
|
|
|
it.effect("replayAll accepts later chunks after the first batch", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
|
|
const one = yield* events.replayAll([
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "one"),
|
|
},
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "two"),
|
|
},
|
|
])
|
|
const two = yield* events.replayAll([
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 2,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "three"),
|
|
},
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 3,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "four"),
|
|
},
|
|
])
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(one).toBe(aggregateID)
|
|
expect(two).toBe(aggregateID)
|
|
expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3])
|
|
}),
|
|
)
|
|
|
|
it.effect("claim fences replay owners", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "seed"))
|
|
yield* events.claim(aggregateID, "owner-a")
|
|
yield* events.project(DurableMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "ignored"),
|
|
},
|
|
{ ownerID: "owner-b" },
|
|
)
|
|
|
|
expect(received).toHaveLength(0)
|
|
}),
|
|
)
|
|
|
|
it.effect("strict owner fences exact replay", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const id = EventV2.ID.create()
|
|
const replayed = {
|
|
id,
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "owned"),
|
|
}
|
|
yield* events.replay(replayed, { ownerID: "owner-a" })
|
|
|
|
const exit = yield* events.replay(replayed, { ownerID: "owner-b", strictOwner: true }).pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Replay owner mismatch")
|
|
}),
|
|
)
|
|
|
|
it.effect("exact replay claims an unowned aggregate", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
const published = yield* events.publish(DurableMessage, durableData(aggregateID, "owned"))
|
|
const replayed = {
|
|
id: published.id,
|
|
created: published.created,
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: published.durable!.seq,
|
|
aggregateID,
|
|
data: published.data,
|
|
}
|
|
|
|
yield* events.replay(replayed, { ownerID: "owner-a", strictOwner: true })
|
|
const row = yield* db
|
|
.select({ ownerID: EventSequenceTable.owner_id })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(row?.ownerID).toBe("owner-a")
|
|
const exit = yield* events
|
|
.replay(
|
|
{ ...replayed, id: EventV2.ID.create(), seq: 1, data: durableData(aggregateID, "conflict") },
|
|
{ ownerID: "owner-b", strictOwner: true },
|
|
)
|
|
.pipe(Effect.exit)
|
|
expect(String(exit)).toContain("Replay owner mismatch")
|
|
}),
|
|
)
|
|
|
|
it.effect("replay with owner claims an unowned sequence", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "owned"),
|
|
},
|
|
{ ownerID: "owner-1" },
|
|
)
|
|
const row = yield* db
|
|
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(row).toEqual({ seq: 0, ownerID: "owner-1" })
|
|
}),
|
|
)
|
|
|
|
it.effect("replay claims an existing unowned sequence before fencing a different owner", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "local"))
|
|
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "claimed"),
|
|
},
|
|
{ ownerID: "owner-1" },
|
|
)
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 2,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "fenced"),
|
|
},
|
|
{ ownerID: "owner-2" },
|
|
)
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
const sequence = yield* db
|
|
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows.map((row) => row.seq)).toEqual([0, 1])
|
|
expect(sequence).toEqual({ seq: 1, ownerID: "owner-1" })
|
|
}),
|
|
)
|
|
|
|
it.effect("strict replay rejects an owner conflict instead of silently skipping it", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "claimed"),
|
|
},
|
|
{ ownerID: "owner-1" },
|
|
)
|
|
|
|
const exit = yield* events
|
|
.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "conflict"),
|
|
},
|
|
{ ownerID: "owner-2", strictOwner: true },
|
|
)
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Replay owner mismatch")
|
|
}),
|
|
)
|
|
|
|
it.effect("publishes accepted replay with its durable sequence and suppresses stale replay", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.listen((event) => Effect.sync(() => received.push(event)))
|
|
const replayed = {
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "replayed"),
|
|
}
|
|
|
|
yield* events.replay(replayed, { publish: true })
|
|
yield* events.replay(replayed, { publish: true })
|
|
|
|
expect(received).toMatchObject([{ id: replayed.id, durable: { seq: 0, version: 1 }, data: replayed.data }])
|
|
}),
|
|
)
|
|
|
|
it.effect("rejects divergent stale replay without publishing it", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
const aggregateID = Session.ID.create()
|
|
const replayed = {
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "original"),
|
|
}
|
|
yield* events.listen((event) => Effect.sync(() => received.push(event)))
|
|
yield* events.replay(replayed, { publish: true })
|
|
|
|
const exit = yield* events
|
|
.replay({ ...replayed, data: durableData(aggregateID, "divergent") }, { publish: true })
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain("Replay diverged")
|
|
expect(received).toHaveLength(1)
|
|
}),
|
|
)
|
|
|
|
it.effect("rejects an event ID reused at another aggregate position", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
const id = EventV2.ID.create()
|
|
yield* events.replay({
|
|
id,
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "first"),
|
|
})
|
|
|
|
const exit = yield* events
|
|
.replay({
|
|
id,
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "second"),
|
|
})
|
|
.pipe(Effect.exit)
|
|
|
|
expect(String(exit)).toContain(`Event ${id} already exists`)
|
|
}),
|
|
)
|
|
|
|
it.effect("replay from a different owner leaves claimed sequence unchanged", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = Session.ID.create()
|
|
const received = new Array<EventV2.Payload>()
|
|
yield* events.listen((event) => Effect.sync(() => received.push(event)))
|
|
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "first"),
|
|
},
|
|
{ ownerID: "owner-1" },
|
|
)
|
|
yield* events.replay(
|
|
{
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 1,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "ignored"),
|
|
},
|
|
{ ownerID: "owner-2", publish: true },
|
|
)
|
|
const rows = yield* db
|
|
.select()
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, aggregateID))
|
|
.all()
|
|
.pipe(Effect.orDie)
|
|
const sequence = yield* db
|
|
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(rows).toHaveLength(1)
|
|
expect(sequence).toEqual({ seq: 0, ownerID: "owner-1" })
|
|
expect(received).toHaveLength(0)
|
|
}),
|
|
)
|
|
|
|
it.effect("claim updates the event sequence owner", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const { db } = yield* Database.Service
|
|
const aggregateID = EventV2.ID.create()
|
|
|
|
yield* events.publish(SyncMessage, { id: aggregateID, text: "claimed" })
|
|
yield* events.claim(aggregateID, "owner-1")
|
|
yield* events.claim(aggregateID, "owner-2")
|
|
const row = yield* db
|
|
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
|
|
.from(EventSequenceTable)
|
|
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
|
|
.get()
|
|
.pipe(Effect.orDie)
|
|
|
|
expect(row).toEqual({ seq: 0, ownerID: "owner-2" })
|
|
}),
|
|
)
|
|
|
|
it.effect("remove clears durable event sequence", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const received = new Array<EventV2.Payload>()
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "seed"))
|
|
yield* events.remove(aggregateID)
|
|
yield* events.project(DurableMessage, (event) =>
|
|
Effect.sync(() => {
|
|
received.push(event)
|
|
}),
|
|
)
|
|
|
|
yield* events.replay({
|
|
id: EventV2.ID.create(),
|
|
created: DateTime.makeUnsafe(0),
|
|
type: EventV2.versionedType(DurableMessage.type, 1),
|
|
seq: 0,
|
|
aggregateID,
|
|
data: durableData(aggregateID, "replayed"),
|
|
})
|
|
|
|
expect(received[0]?.data).toEqual(durableData(aggregateID, "replayed"))
|
|
}),
|
|
)
|
|
|
|
it.effect("log without follow replays events and completes with a synced marker", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
|
|
const items = Array.from(yield* Stream.runCollect(events.log({ aggregateID })))
|
|
|
|
expect(items.map((item) => (EventV2.isSynced(item) ? item.type : item.durable?.seq))).toEqual([
|
|
EventV2.Seq.make(0),
|
|
EventV2.Seq.make(1),
|
|
"log.synced",
|
|
])
|
|
expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID, seq: EventV2.Seq.make(1) })
|
|
}),
|
|
)
|
|
|
|
it.effect("log synced marker omits seq for an empty log and keeps the cursor otherwise", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
|
|
const empty = Array.from(yield* Stream.runCollect(events.log({ aggregateID })))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
const drained = Array.from(yield* Stream.runCollect(events.log({ aggregateID, after: 0 })))
|
|
|
|
expect(empty).toEqual([{ type: "log.synced", aggregateID }])
|
|
expect(empty[0]).not.toHaveProperty("seq")
|
|
expect(drained).toEqual([{ type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) }])
|
|
}),
|
|
)
|
|
|
|
it.effect("log with follow emits the synced marker at the replay-to-live boundary", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
const fiber = yield* events
|
|
.log({ aggregateID, follow: true })
|
|
.pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
|
|
yield* Effect.yieldNow
|
|
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
|
|
const items = Array.from(yield* Fiber.join(fiber))
|
|
expect(items.map((item) => (EventV2.isSynced(item) ? item : item.durable?.seq))).toEqual([
|
|
EventV2.Seq.make(0),
|
|
{ type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) },
|
|
EventV2.Seq.make(1),
|
|
])
|
|
}),
|
|
)
|
|
|
|
it.effect("log replays across configured read pages", () =>
|
|
Effect.gen(function* () {
|
|
const eventLayer = EventV2.layerWith({ logReadPageSize: 2 }).pipe(Layer.provide(LayerNode.compile(Database.node)))
|
|
|
|
yield* Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "two"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "three"))
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "four"))
|
|
|
|
const items = Array.from(yield* Stream.runCollect(events.log({ aggregateID })))
|
|
|
|
expect(items.map((item) => (EventV2.isSynced(item) ? item.type : item.durable?.seq))).toEqual([
|
|
EventV2.Seq.make(0),
|
|
EventV2.Seq.make(1),
|
|
EventV2.Seq.make(2),
|
|
EventV2.Seq.make(3),
|
|
EventV2.Seq.make(4),
|
|
"log.synced",
|
|
])
|
|
expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID, seq: EventV2.Seq.make(4) })
|
|
}).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer)))
|
|
}),
|
|
)
|
|
|
|
it.effect("log with follow emits events committed during replay after the synced marker", () =>
|
|
Effect.gen(function* () {
|
|
const readStarted = yield* Deferred.make<void>()
|
|
const releaseRead = yield* Deferred.make<void>()
|
|
const firstRead = yield* Ref.make(true)
|
|
const eventLayer = EventV2.layerWith({
|
|
beforeAggregateRead: () =>
|
|
Ref.getAndSet(firstRead, false).pipe(
|
|
Effect.flatMap((shouldBlock) => {
|
|
if (!shouldBlock) return Effect.void
|
|
return Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseRead)))
|
|
}),
|
|
),
|
|
}).pipe(Layer.provide(LayerNode.compile(Database.node)))
|
|
|
|
yield* Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const aggregateID = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "zero"))
|
|
const fiber = yield* events
|
|
.log({ aggregateID, follow: true })
|
|
.pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
|
|
|
|
yield* Deferred.await(readStarted)
|
|
yield* events.publish(DurableMessage, durableData(aggregateID, "one"))
|
|
yield* Deferred.succeed(releaseRead, undefined)
|
|
|
|
const items = Array.from(yield* Fiber.join(fiber))
|
|
expect(items.map((item) => (EventV2.isSynced(item) ? item : item.durable?.seq))).toEqual([
|
|
EventV2.Seq.make(0),
|
|
{ type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) },
|
|
EventV2.Seq.make(1),
|
|
])
|
|
}).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer)))
|
|
}),
|
|
)
|
|
|
|
it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2.Service
|
|
const first = Session.ID.create()
|
|
const second = Session.ID.create()
|
|
yield* events.publish(DurableMessage, durableData(first, "zero"))
|
|
yield* events.publish(DurableMessage, durableData(first, "one"))
|
|
yield* events.publish(DurableMessage, durableData(second, "zero"))
|
|
|
|
const sequences = yield* events.sequences([first, second, Session.ID.create()])
|
|
|
|
expect(sequences).toEqual(
|
|
new Map([
|
|
[first, EventV2.Seq.make(1)],
|
|
[second, EventV2.Seq.make(0)],
|
|
]),
|
|
)
|
|
expect(yield* events.sequences([])).toEqual(new Map())
|
|
}),
|
|
)
|
|
})
|