1import * as Effect from "effect/Effect";
2import * as Schema from "effect/Schema";
3import * as SqlClient from "effect/sql/SqlClient";
4
5interface LegacyV2EventRow {
6 readonly event_id: string;
7 readonly command_id: string | null;
8 readonly thread_id: string;
9 readonly run_id: string | null;
10 readonly node_id: string | null;
11 readonly driver: string | null;
12 readonly provider_instance_id: string | null;
13 readonly raw_event_id: string | null;
14 readonly event_type: string;
15 readonly occurred_at: string;
16 readonly payload_json: string;
17}
18
19interface ProjectProjectionRow {
20 readonly project_id: string;
21 readonly title: string;
22 readonly workspace_root: string;
23 readonly default_model_selection_json: string | null;
24 readonly scripts_json: string;
25 readonly created_at: string;
26 readonly updated_at: string;
27 readonly deleted_at: string | null;
28}
29
30const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
31const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
32
33// Event-store setup and V1 project baseline composed by migration 055.
34export default Effect.gen(function* () {
35 const sql = yield* SqlClient.SqlClient;
36
37 yield* sql`
38 ALTER TABLE orchestration_events
39 ADD COLUMN application_event_version INTEGER NOT NULL DEFAULT 1
40 `;
41 yield* sql`
42 CREATE INDEX idx_orchestration_events_application_sequence
43 ON orchestration_events(application_event_version, sequence)
44 `;
45 yield* sql`
46 ALTER TABLE orchestration_command_receipts
47 ADD COLUMN command_type TEXT NOT NULL DEFAULT 'legacy'
48 `;
49
50 const legacyV2Events = yield* sql<LegacyV2EventRow>`
51 SELECT
52 event_id,
53 command_id,
54 thread_id,
55 run_id,
56 node_id,
57 driver,
58 provider_instance_id,
59 raw_event_id,
60 event_type,
61 occurred_at,
62 payload_json
63 FROM orchestration_v2_events
64 ORDER BY sequence ASC
65 `;
66
67 yield* Effect.forEach(
68 legacyV2Events,
69 (event) =>
70 Effect.gen(function* () {
71 const metadata = yield* encodeJson({
72 applicationEventVersion: 2,
73 ...(event.run_id === null ? {} : { runId: event.run_id }),
74 ...(event.node_id === null ? {} : { nodeId: event.node_id }),
75 ...(event.driver === null ? {} : { driver: event.driver }),
76 ...(event.provider_instance_id === null
77 ? {}
78 : { providerInstanceId: event.provider_instance_id }),
79 ...(event.raw_event_id === null ? {} : { rawEventId: event.raw_event_id }),
80 });
81 yield* sql`
82 INSERT INTO orchestration_events (
83 event_id,
84 aggregate_kind,
85 stream_id,
86 stream_version,
87 event_type,
88 occurred_at,
89 command_id,
90 causation_event_id,
91 correlation_id,
92 actor_kind,
93 payload_json,
94 metadata_json,
95 application_event_version
96 )
97 SELECT
98 ${event.event_id},
99 'thread',
100 ${event.thread_id},
101 COALESCE(
102 (
103 SELECT MAX(stream_version) + 1
104 FROM orchestration_events
105 WHERE aggregate_kind = 'thread' AND stream_id = ${event.thread_id}
106 ),
107 0
108 ),
109 ${event.event_type},
110 ${event.occurred_at},
111 ${event.command_id},
112 NULL,
113 ${event.command_id},
114 ${event.raw_event_id === null ? "server" : "provider"},
115 ${event.payload_json},
116 ${metadata},
117 2
118 WHERE NOT EXISTS (
119 SELECT 1 FROM orchestration_events WHERE event_id = ${event.event_id}
120 )
121 `;
122 }),
123 { concurrency: 1, discard: true },
124 );
125
126 // ProjectService wrote this projection directly before the application event
127 // boundary was restored. Re-baseline every current row into the shared log so
128 // projection rebuilds preserve the exact pre-migration project state.
129 const projectRows = yield* sql<ProjectProjectionRow>`
130 SELECT
131 project_id,
132 title,
133 workspace_root,
134 default_model_selection_json,
135 scripts_json,
136 created_at,
137 updated_at,
138 deleted_at
139 FROM projection_projects
140 ORDER BY created_at ASC, project_id ASC
141 `;
142 yield* Effect.forEach(
143 projectRows,
144 (project) =>
145 Effect.gen(function* () {
146 const createdEventId = `migration:39:project:${project.project_id}:baseline`;
147 const createdPayload = {
148 projectId: project.project_id,
149 title: project.title,
150 workspaceRoot: project.workspace_root,
151 defaultModelSelection:
152 project.default_model_selection_json === null
153 ? null
154 : yield* decodeJson(project.default_model_selection_json),
155 scripts: yield* decodeJson(project.scripts_json),
156 createdAt: project.created_at,
157 updatedAt: project.updated_at,
158 };
159 const createdPayloadJson = yield* encodeJson(createdPayload);
160 const migrationMetadataJson = yield* encodeJson({
161 applicationEventVersion: 2,
162 migration: 39,
163 });
164 yield* sql`
165 INSERT INTO orchestration_events (
166 event_id,
167 aggregate_kind,
168 stream_id,
169 stream_version,
170 event_type,
171 occurred_at,
172 command_id,
173 causation_event_id,
174 correlation_id,
175 actor_kind,
176 payload_json,
177 metadata_json,
178 application_event_version
179 )
180 VALUES (
181 ${createdEventId},
182 'project',
183 ${project.project_id},
184 COALESCE(
185 (
186 SELECT MAX(stream_version) + 1
187 FROM orchestration_events
188 WHERE aggregate_kind = 'project' AND stream_id = ${project.project_id}
189 ),
190 0
191 ),
192 'project.created',
193 ${project.updated_at},
194 NULL,
195 NULL,
196 NULL,
197 'server',
198 ${createdPayloadJson},
199 ${migrationMetadataJson},
200 2
201 )
202 `;
203
204 if (project.deleted_at !== null) {
205 const deletedPayloadJson = yield* encodeJson({
206 projectId: project.project_id,
207 deletedAt: project.deleted_at,
208 });
209 yield* sql`
210 INSERT INTO orchestration_events (
211 event_id,
212 aggregate_kind,
213 stream_id,
214 stream_version,
215 event_type,
216 occurred_at,
217 command_id,
218 causation_event_id,
219 correlation_id,
220 actor_kind,
221 payload_json,
222 metadata_json,
223 application_event_version
224 )
225 VALUES (
226 ${`migration:39:project:${project.project_id}:deleted`},
227 'project',
228 ${project.project_id},
229 (
230 SELECT MAX(stream_version) + 1
231 FROM orchestration_events
232 WHERE aggregate_kind = 'project' AND stream_id = ${project.project_id}
233 ),
234 'project.deleted',
235 ${project.deleted_at},
236 NULL,
237 ${createdEventId},
238 NULL,
239 'server',
240 ${deletedPayloadJson},
241 ${migrationMetadataJson},
242 2
243 )
244 `;
245 }
246 }),
247 { concurrency: 1, discard: true },
248 );
249
250 yield* sql`
251 INSERT INTO orchestration_command_receipts (
252 command_id,
253 aggregate_kind,
254 aggregate_id,
255 accepted_at,
256 result_sequence,
257 status,
258 error,
259 command_type
260 )
261 SELECT
262 command_id,
263 'thread',
264 thread_id,
265 accepted_at,
266 COALESCE(
267 (
268 SELECT MAX(events.sequence)
269 FROM orchestration_events events
270 WHERE events.application_event_version = 2
271 AND events.command_id = receipts.command_id
272 ),
273 0
274 ),
275 status,
276 error,
277 command_type
278 FROM orchestration_v2_command_receipts receipts
279 WHERE TRUE
280 ON CONFLICT(command_id) DO NOTHING
281 `;
282});