agaltsovav/opencode
1
0
Fork 0
Code Issues Pull Requests Actions 79 Packages Projects Releases Wiki Activity

test(core): cover sync event replay controls

Browse Source
This commit is contained in:
Dax Raad
2026-05-25 00:44:04 -04:00
parent 13e9e317b1
commit 9440f7f5b9
1 changed files with 256 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/test/event.test.ts
+256 -2
View File
@@ -1,16 +1,20 @@
import { describe, expect } from "bun:test"
import { Effect, Fiber, Layer, Schema, Stream } from "effect"
import { EventV2 } from "@opencode-ai/core/event"
import { Database } from "@opencode-ai/core/database/database"
import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
import { Location } from "@opencode-ai/core/location"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"
const locationLayer = Layer.succeed(
Location.Service,
Location.Service.of({ directory: AbsolutePath.make("project"), workspaceID: "workspace" }),
)
const it = testEffect(EventV2.defaultLayer.pipe(Layer.provideMerge(locationLayer)))
const itWithoutLocation = testEffect(EventV2.defaultLayer)
const eventLayer = Layer.mergeAll(EventV2.defaultLayer, Database.defaultLayer)
const it = testEffect(eventLayer.pipe(Layer.provideMerge(locationLayer)))
const itWithoutLocation = testEffect(eventLayer)
const Message = EventV2.define({
type: "test.message",
@@ -31,6 +35,18 @@ const SyncMessage = EventV2.define({
},
})
const SyncSent = EventV2.define({
type: "test.sent",
sync: {
version: 1,
aggregate: "messageID",
},
schema: {
messageID: Schema.String,
text: Schema.String,
},
})
const GlobalMessage = EventV2.define({
type: "test.global",
schema: {
@@ -164,6 +180,49 @@ describe("EventV2", () => {
}),
)
it.effect("inserts sync 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 sync 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("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 sync events through projectors", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
@@ -188,6 +247,69 @@ describe("EventV2", () => {
}),
)
it.effect("replay inserts external event rows", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const aggregateID = EventV2.ID.create()
yield* events.replay({
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 0,
aggregateID,
data: { id: aggregateID, text: "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 defects on sequence mismatch", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const aggregateID = EventV2.ID.create()
yield* events.replay({
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 0,
aggregateID,
data: { id: aggregateID, text: "first" },
})
const exit = yield* events
.replay({
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 5,
aggregateID,
data: { id: aggregateID, text: "bad" },
})
.pipe(Effect.exit)
expect(String(exit)).toContain("Sequence mismatch")
}),
)
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(),
type: "unknown.event.1",
seq: 0,
aggregateID: EventV2.ID.create(),
data: {},
})
.pipe(Effect.exit)
expect(String(exit)).toContain("Unknown sync event type")
}),
)
it.effect("replayAll validates contiguous aggregate events", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
@@ -213,6 +335,52 @@ describe("EventV2", () => {
}),
)
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 = EventV2.ID.create()
const one = yield* events.replayAll([
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 0,
aggregateID,
data: { id: aggregateID, text: "one" },
},
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 1,
aggregateID,
data: { id: aggregateID, text: "two" },
},
])
const two = yield* events.replayAll([
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 2,
aggregateID,
data: { id: aggregateID, text: "three" },
},
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 3,
aggregateID,
data: { id: aggregateID, text: "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
@@ -241,6 +409,92 @@ describe("EventV2", () => {
}),
)
it.effect("replay with owner claims an unowned sequence", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const aggregateID = EventV2.ID.create()
yield* events.replay(
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 0,
aggregateID,
data: { id: aggregateID, text: "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 from a different owner leaves claimed sequence unchanged", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const aggregateID = EventV2.ID.create()
yield* events.replay(
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 0,
aggregateID,
data: { id: aggregateID, text: "first" },
},
{ ownerID: "owner-1" },
)
yield* events.replay(
{
id: EventV2.ID.create(),
type: EventV2.versionedType(SyncMessage.type, 1),
seq: 1,
aggregateID,
data: { id: aggregateID, text: "ignored" },
},
{ 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).toHaveLength(1)
expect(sequence).toEqual({ seq: 0, ownerID: "owner-1" })
}),
)
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 sync event sequence", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service