1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/sql/SqlClient";
3
4/**
5 * Production-facing V2 persistence setup composed by migration 055.
6 *
7 * The original V2 schema used a single `provider` column for both configured
8 * instance routing and driver identity. Keep those columns in place for
9 * migration compatibility. They remain write-through shadows where the old
10 * schema declared them NOT NULL; all V2 reads and indexes use the explicit
11 * columns added here.
12 */
13export default Effect.gen(function* () {
14 const sql = yield* SqlClient.SqlClient;
15
16 yield* sql`ALTER TABLE orchestration_v2_events ADD COLUMN driver TEXT`;
17 yield* sql`ALTER TABLE orchestration_v2_events ADD COLUMN provider_instance_id TEXT`;
18 yield* sql`
19 UPDATE orchestration_v2_events
20 SET
21 driver = COALESCE(json_extract(payload_json, '$.driver'), provider),
22 provider_instance_id = COALESCE(
23 json_extract(payload_json, '$.providerInstanceId'),
24 provider
25 )
26 `;
27 yield* sql`CREATE INDEX orchestration_v2_events_instance_sequence_idx ON orchestration_v2_events(provider_instance_id, sequence)`;
28
29 yield* sql`ALTER TABLE orchestration_v2_projection_threads ADD COLUMN provider_instance_id TEXT`;
30 yield* sql`
31 UPDATE orchestration_v2_projection_threads
32 SET provider_instance_id = COALESCE(
33 json_extract(payload_json, '$.providerInstanceId'),
34 default_provider
35 )
36 `;
37
38 yield* sql`ALTER TABLE orchestration_v2_projection_runs ADD COLUMN provider_instance_id TEXT`;
39 yield* sql`
40 UPDATE orchestration_v2_projection_runs
41 SET provider_instance_id = COALESCE(
42 json_extract(payload_json, '$.providerInstanceId'),
43 provider
44 )
45 `;
46
47 yield* sql`ALTER TABLE orchestration_v2_projection_run_attempts ADD COLUMN provider_instance_id TEXT`;
48 yield* sql`
49 UPDATE orchestration_v2_projection_run_attempts
50 SET provider_instance_id = COALESCE(
51 json_extract(payload_json, '$.providerInstanceId'),
52 provider
53 )
54 `;
55
56 yield* sql`ALTER TABLE orchestration_v2_projection_provider_sessions ADD COLUMN driver TEXT`;
57 yield* sql`ALTER TABLE orchestration_v2_projection_provider_sessions ADD COLUMN provider_instance_id TEXT`;
58 yield* sql`
59 UPDATE orchestration_v2_projection_provider_sessions
60 SET
61 driver = COALESCE(json_extract(payload_json, '$.driver'), provider),
62 provider_instance_id = COALESCE(
63 json_extract(payload_json, '$.providerInstanceId'),
64 provider
65 )
66 `;
67 yield* sql`CREATE INDEX orchestration_v2_projection_provider_sessions_instance_status_idx ON orchestration_v2_projection_provider_sessions(provider_instance_id, status)`;
68
69 yield* sql`ALTER TABLE orchestration_v2_projection_provider_threads ADD COLUMN driver TEXT`;
70 yield* sql`ALTER TABLE orchestration_v2_projection_provider_threads ADD COLUMN provider_instance_id TEXT`;
71 yield* sql`
72 UPDATE orchestration_v2_projection_provider_threads
73 SET
74 driver = COALESCE(json_extract(payload_json, '$.driver'), provider),
75 provider_instance_id = COALESCE(
76 json_extract(payload_json, '$.providerInstanceId'),
77 provider
78 )
79 `;
80 yield* sql`CREATE INDEX orchestration_v2_projection_provider_threads_instance_status_idx ON orchestration_v2_projection_provider_threads(provider_instance_id, status)`;
81
82 yield* sql`ALTER TABLE orchestration_v2_projection_subagents ADD COLUMN driver TEXT`;
83 yield* sql`ALTER TABLE orchestration_v2_projection_subagents ADD COLUMN provider_instance_id TEXT`;
84 yield* sql`
85 UPDATE orchestration_v2_projection_subagents
86 SET
87 driver = json_extract(payload_json, '$.driver'),
88 provider_instance_id = COALESCE(
89 json_extract(payload_json, '$.providerInstanceId'),
90 provider
91 )
92 `;
93
94 yield* sql`ALTER TABLE orchestration_v2_projection_context_transfers ADD COLUMN source_provider_instance_id TEXT`;
95 yield* sql`ALTER TABLE orchestration_v2_projection_context_transfers ADD COLUMN target_provider_instance_id TEXT`;
96 yield* sql`
97 UPDATE orchestration_v2_projection_context_transfers
98 SET
99 source_provider_instance_id = COALESCE(
100 json_extract(payload_json, '$.sourceProviderInstanceId'),
101 source_provider
102 ),
103 target_provider_instance_id = COALESCE(
104 json_extract(payload_json, '$.targetProviderInstanceId'),
105 target_provider
106 )
107 `;
108
109 yield* sql`
110 CREATE TABLE orchestration_v2_effect_outbox (
111 effect_id TEXT PRIMARY KEY,
112 command_id TEXT NOT NULL,
113 thread_id TEXT NOT NULL,
114 effect_type TEXT NOT NULL,
115 payload_json TEXT NOT NULL,
116 status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'succeeded', 'failed')),
117 attempt_count INTEGER NOT NULL DEFAULT 0,
118 available_at TEXT NOT NULL,
119 lease_owner TEXT,
120 lease_expires_at TEXT,
121 created_at TEXT NOT NULL,
122 updated_at TEXT NOT NULL,
123 completed_at TEXT,
124 last_error TEXT
125 )
126 `;
127 yield* sql`CREATE INDEX orchestration_v2_effect_outbox_claim_idx ON orchestration_v2_effect_outbox(status, available_at, lease_expires_at, created_at)`;
128 yield* sql`CREATE INDEX orchestration_v2_effect_outbox_command_idx ON orchestration_v2_effect_outbox(command_id, effect_id)`;
129
130 yield* sql`
131 CREATE TABLE orchestration_v2_turn_item_positions (
132 thread_id TEXT NOT NULL,
133 turn_item_id TEXT NOT NULL,
134 ordinal INTEGER NOT NULL,
135 PRIMARY KEY (thread_id, turn_item_id),
136 UNIQUE (thread_id, ordinal)
137 )
138 `;
139 yield* sql`
140 INSERT INTO orchestration_v2_turn_item_positions (
141 thread_id,
142 turn_item_id,
143 ordinal
144 )
145 SELECT
146 item.thread_id,
147 item.turn_item_id,
148 COALESCE(run.ordinal, 0) * 1000000 + ROW_NUMBER() OVER (
149 PARTITION BY item.thread_id, COALESCE(run.ordinal, 0)
150 ORDER BY item.ordinal, item.turn_item_id
151 )
152 FROM orchestration_v2_projection_turn_items AS item
153 LEFT JOIN orchestration_v2_projection_runs AS run
154 ON run.thread_id = item.thread_id
155 AND run.run_id = item.run_id
156 `;
157
158 yield* sql`
159 CREATE TABLE orchestration_v2_projection_metadata (
160 projection_name TEXT PRIMARY KEY,
161 schema_version INTEGER NOT NULL,
162 last_sequence INTEGER NOT NULL,
163 updated_at TEXT NOT NULL
164 )
165 `;
166 yield* sql`
167 INSERT INTO orchestration_v2_projection_metadata (
168 projection_name,
169 schema_version,
170 last_sequence,
171 updated_at
172 )
173 VALUES (
174 'thread-projections',
175 1,
176 COALESCE((SELECT MAX(sequence) FROM orchestration_v2_events), 0),
177 strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
178 )
179 `;
180});