Skip to content

Commit caea930

Browse files
kitlangtonkitlangton
authored
fix(core): reset pre-launch session projections (#30728)
1 parent 69cfc44 commit caea930

4 files changed

Lines changed: 41 additions & 38 deletions

File tree

‎packages/core/migration/20260603040000_session_message_projection_order/migration.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
DELETE FROM `session_message`;--> statement-breakpoint
12
ALTER TABLE `session_message` ADD `seq` integer NOT NULL;--> statement-breakpoint
23
DROP INDEX IF EXISTS `session_message_session_time_created_id_idx`;--> statement-breakpoint
34
DROP INDEX IF EXISTS `session_message_session_type_time_created_id_idx`;--> statement-breakpoint

‎packages/core/src/database/migration/20260603040000_session_message_projection_order.ts‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,16 +5,10 @@ export default {
55
id: "20260603040000_session_message_projection_order",
66
up(tx) {
77
return Effect.gen(function* () {
8-
yield* tx.run(`ALTER TABLE \`session_message\` ADD COLUMN \`seq\` integer NOT NULL DEFAULT 0;`)
9-
yield* tx.run(
10-
`UPDATE \`session_message\` SET \`seq\` = COALESCE((SELECT \`seq\` + 1 FROM \`event\` WHERE \`event\`.\`id\` = \`session_message\`.\`id\`), 0);`,
11-
)
12-
const unmatched = yield* tx.get<{ count: number }>(
13-
`SELECT COUNT(*) AS \`count\` FROM \`session_message\` WHERE \`seq\` = 0;`,
14-
)
15-
if ((unmatched?.count ?? 0) > 0)
16-
return yield* Effect.die("Cannot migrate session_message projections without matching durable events")
17-
yield* tx.run(`UPDATE \`session_message\` SET \`seq\` = \`seq\` - 1;`)
8+
// Pre-launch Session projections were written before durable event persistence
9+
// became unconditional, so they cannot be assigned truthful aggregate order.
10+
yield* tx.run(`DELETE FROM \`session_message\`;`)
11+
yield* tx.run(`ALTER TABLE \`session_message\` ADD COLUMN \`seq\` integer NOT NULL;`)
1812
yield* tx.run(`DROP INDEX IF EXISTS \`session_message_session_type_time_created_id_idx\`;`)
1913
yield* tx.run(`CREATE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`)
2014
yield* tx.run(

‎packages/core/test/database-migration.test.ts‎

Lines changed: 33 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -79,54 +79,61 @@ describe("DatabaseMigration", () => {
7979
)
8080
})
8181

82-
test("backfills projected Session message order from durable event sequence", async () => {
82+
test("resets incompatible projected Session messages before adding sequence order", async () => {
8383
await run(
8484
Effect.gen(function* () {
8585
const db = yield* makeDb
86+
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
87+
yield* db.run(
88+
sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
89+
)
90+
yield* db.run(
91+
sql`CREATE TABLE part (id text PRIMARY KEY, message_id text NOT NULL, session_id text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
92+
)
8693
yield* db.run(sql`CREATE TABLE event (id text PRIMARY KEY, seq integer NOT NULL)`)
8794
yield* db.run(
88-
sql`CREATE TABLE session_message (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, time_created integer NOT NULL, data text NOT NULL)`,
95+
sql`CREATE TABLE session_message (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
8996
)
9097
yield* db.run(
9198
sql`CREATE INDEX session_message_session_time_created_id_idx ON session_message (session_id, time_created, id)`,
9299
)
93100
yield* db.run(
94101
sql`CREATE INDEX session_message_session_type_time_created_id_idx ON session_message (session_id, type, time_created, id)`,
95102
)
96-
yield* db.run(sql`INSERT INTO event (id, seq) VALUES ('evt_z', 0), ('evt_a', 1)`)
103+
yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
104+
yield* db.run(
105+
sql`INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES ('legacy_message', 'session', 1, 1, '{"role":"user"}')`,
106+
)
107+
yield* db.run(
108+
sql`INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES ('legacy_part', 'legacy_message', 'session', 1, 1, '{"type":"text","text":"hello"}')`,
109+
)
97110
yield* db.run(
98-
sql`INSERT INTO session_message (id, session_id, type, time_created, data) VALUES ('evt_z', 'session', 'user', 0, '{}'), ('evt_a', 'session', 'user', 0, '{}')`,
111+
sql`INSERT INTO session_message (id, session_id, type, time_created, time_updated, data) VALUES ('stale_projection', 'session', 'user', 1, 1, '{}')`,
99112
)
100113

101114
yield* DatabaseMigration.applyOnly(db, [sessionMessageProjectionOrderMigration])
102115

103-
expect(yield* db.all(sql`SELECT id, seq FROM session_message ORDER BY seq`)).toEqual([
104-
{ id: "evt_z", seq: 0 },
105-
{ id: "evt_a", seq: 1 },
116+
expect(yield* db.all(sql`SELECT id, session_id, data FROM message`)).toEqual([
117+
{ id: "legacy_message", session_id: "session", data: '{"role":"user"}' },
106118
])
119+
expect(yield* db.all(sql`SELECT id, message_id, session_id, data FROM part`)).toEqual([
120+
{
121+
id: "legacy_part",
122+
message_id: "legacy_message",
123+
session_id: "session",
124+
data: '{"type":"text","text":"hello"}',
125+
},
126+
])
127+
expect(yield* db.all(sql`SELECT id FROM session_message`)).toEqual([])
128+
129+
yield* db.run(
130+
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('fresh_projection', 'session', 'user', 7, 2, 2, '{}')`,
131+
)
132+
expect(yield* db.get(sql`SELECT id, seq FROM session_message`)).toEqual({ id: "fresh_projection", seq: 7 })
107133
}),
108134
)
109135
})
110136

111-
test("fails projected Session message order backfill without a durable event", async () => {
112-
await expect(
113-
run(
114-
Effect.gen(function* () {
115-
const db = yield* makeDb
116-
yield* db.run(sql`CREATE TABLE event (id text PRIMARY KEY, seq integer NOT NULL)`)
117-
yield* db.run(
118-
sql`CREATE TABLE session_message (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, time_created integer NOT NULL, data text NOT NULL)`,
119-
)
120-
yield* db.run(
121-
sql`INSERT INTO session_message (id, session_id, type, time_created, data) VALUES ('evt_missing', 'session', 'user', 0, '{}')`,
122-
)
123-
124-
yield* DatabaseMigration.applyOnly(db, [sessionMessageProjectionOrderMigration])
125-
}),
126-
),
127-
).rejects.toThrow("Cannot migrate session_message projections without matching durable events")
128-
})
129-
130137
test("runs session usage backfill in order with schema changes", async () => {
131138
await run(
132139
Effect.gen(function* () {

‎specs/v2/schema-changelog.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ Affected schema:
103103

104104
Change:
105105

106-
- Add and backfill `session_message.seq` from matching synchronized events.
106+
- Reset pre-launch Session-message projections and add `session_message.seq` for newly projected synchronized events.
107107
- Add event aggregate-sequence and aggregate-type-sequence indexes.
108108
- Add Session-message sequence, type-sequence, and compatibility timestamp indexes.
109109

@@ -114,7 +114,8 @@ Reason:
114114

115115
Compatibility:
116116

117-
- Migration fails rather than inventing chronology if an existing projected Session message has no matching durable event.
117+
- Pre-launch Session-message projections are disposable because historical versions could write them without durable creator events.
118+
- The migration resets those projections rather than inventing chronology or blocking startup.
118119
- The timestamp compatibility index remains for legacy or transitional query shapes.
119120

120121
### Structured Tool Registry And Canonical Output

0 commit comments

Comments
 (0)