Line data Source code
1 : import * as connection from '@ankoh/dashql-jsonschema/connection.js';
2 : import * as auth from '@ankoh/dashql-jsonschema/auth.js';
3 :
4 : import type { DetailedError } from '../../utils/error.js';
5 :
6 : import { VariantKind } from '../../utils/variant.js';
7 : import { TRINO_CONNECTOR } from '../connector_info.js';
8 : import {
9 : ConnectionHealth,
10 : ConnectionStatus,
11 : ConnectionState,
12 : RESET_CONNECTION,
13 : DELETE_CONNECTION,
14 : } from '../connection_state.js';
15 : import { TrinoChannelInterface } from "./trino_channel.js";
16 : import { dateToTimestamp } from "../proto_helper.js";
17 :
18 : // OAuth flow action symbols
19 1 : export const OAUTH_STARTED = Symbol('OAUTH_STARTED');
20 1 : export const OAUTH_CANCELLED = Symbol('OAUTH_CANCELLED');
21 1 : export const OAUTH_FAILED = Symbol('OAUTH_FAILED');
22 1 : export const GENERATING_PKCE_CHALLENGE = Symbol('GENERATING_PKCE_CHALLENGE');
23 1 : export const GENERATED_PKCE_CHALLENGE = Symbol('GENERATED_PKCE_CHALLENGE');
24 1 : export const OAUTH_BROWSER_OPENED = Symbol('OAUTH_BROWSER_OPENED');
25 1 : export const RECEIVED_OAUTH_CODE = Symbol('RECEIVED_OAUTH_CODE');
26 1 : export const REQUESTING_ACCESS_TOKEN = Symbol('REQUESTING_ACCESS_TOKEN');
27 1 : export const RECEIVED_ACCESS_TOKEN = Symbol('RECEIVED_ACCESS_TOKEN');
28 :
29 : export interface TrinoConnectionStateDetails {
30 : /// The proto
31 : proto: connection.TrinoConnectionDetails;
32 : /// The channel
33 : channel: TrinoChannelInterface | null;
34 : }
35 :
36 1 : export function createTrinoConnectionStateDetails(params: connection.TrinoConnectionParams): TrinoConnectionStateDetails {
37 1 : return {
38 1 : proto: {
39 1 : setupTimings: {},
40 1 : setupParams: params,
41 1 : },
42 1 : channel: null
43 1 : };
44 1 : }
45 :
46 1 : export function getTrinoConnectionDetails(state: ConnectionState | null): TrinoConnectionStateDetails | null {
47 1 : if (state == null) return null;
48 1 : switch (state.details.type) {
49 1 : case TRINO_CONNECTOR: return state.details.value;
50 1 : default: return null;
51 1 : }
52 1 : }
53 :
54 1 : export const TRINO_CHANNEL_SETUP_CANCELLED = Symbol('TRINO_CHANNEL_SETUP_CANCELLED');
55 1 : export const TRINO_CHANNEL_SETUP_FAILED = Symbol('TRINO_CHANNEL_SETUP_FAILED');
56 1 : export const TRINO_CHANNEL_SETUP_STARTED = Symbol('TRINO_CHANNEL_SETUP_STARTED');
57 1 : export const TRINO_CHANNEL_READY = Symbol('TRINO_CHANNEL_READY');
58 :
59 : export type TrinoConnectorAction =
60 : | VariantKind<typeof RESET_CONNECTION, null>
61 : | VariantKind<typeof DELETE_CONNECTION, null>
62 : // OAuth flow actions
63 : | VariantKind<typeof OAUTH_STARTED, connection.TrinoConnectionParams>
64 : | VariantKind<typeof OAUTH_CANCELLED, DetailedError>
65 : | VariantKind<typeof OAUTH_FAILED, DetailedError>
66 : | VariantKind<typeof GENERATING_PKCE_CHALLENGE, null>
67 : | VariantKind<typeof GENERATED_PKCE_CHALLENGE, auth.OAuthPKCEChallenge>
68 : | VariantKind<typeof OAUTH_BROWSER_OPENED, null>
69 : | VariantKind<typeof RECEIVED_OAUTH_CODE, auth.TemporaryToken>
70 : | VariantKind<typeof REQUESTING_ACCESS_TOKEN, null>
71 : | VariantKind<typeof RECEIVED_ACCESS_TOKEN, connection.TrinoAccessToken>
72 : // Channel setup actions
73 : | VariantKind<typeof TRINO_CHANNEL_SETUP_STARTED, connection.TrinoConnectionParams>
74 : | VariantKind<typeof TRINO_CHANNEL_SETUP_CANCELLED, DetailedError>
75 : | VariantKind<typeof TRINO_CHANNEL_SETUP_FAILED, DetailedError>
76 : | VariantKind<typeof TRINO_CHANNEL_READY, TrinoChannelInterface>
77 : // Health check actions
78 : ;
79 :
80 1 : export function reduceTrinoConnectorState(state: ConnectionState, action: TrinoConnectorAction): ConnectionState | null {
81 1 : const details = state.details.value as TrinoConnectionStateDetails;
82 1 : let next: ConnectionState | null = null;
83 1 : switch (action.type) {
84 1 : case DELETE_CONNECTION:
85 1 : case RESET_CONNECTION:
86 1 : details.channel?.close();
87 1 : next = {
88 1 : ...state,
89 1 : details: {
90 1 : type: TRINO_CONNECTOR,
91 1 : value: {
92 1 : ...details,
93 1 : proto: {
94 1 : ...details.proto,
95 1 : setupTimings: {},
96 1 : channelError: undefined,
97 1 : healthCheckError: undefined,
98 1 : oauthState: undefined,
99 1 : },
100 1 : channel: null
101 1 : }
102 1 : },
103 1 : };
104 1 : break;
105 :
106 : // OAuth flow actions
107 1 : case OAUTH_STARTED: {
108 1 : const newDetails: TrinoConnectionStateDetails = {
109 1 : proto: {
110 1 : ...details.proto,
111 1 : setupTimings: {
112 1 : oauthStartedAt: dateToTimestamp(new Date()),
113 1 : },
114 1 : setupParams: action.value,
115 1 : channelError: undefined,
116 1 : healthCheckError: undefined,
117 1 : oauthState: {
118 1 : oauthPkce: { value: "", verifier: "" } as auth.OAuthPKCEChallenge,
119 1 : authCode: { token: "", createdAt: new Date().toISOString() } as auth.TemporaryToken,
120 1 : accessToken: {} as connection.TrinoAccessToken,
121 1 : },
122 1 : },
123 1 : channel: null,
124 1 : };
125 1 : next = {
126 1 : ...state,
127 1 : connectionStatus: ConnectionStatus.AUTH_STARTED,
128 1 : connectionHealth: ConnectionHealth.CONNECTING,
129 1 : details: {
130 1 : type: TRINO_CONNECTOR,
131 1 : value: newDetails,
132 1 : },
133 1 : };
134 1 : break;
135 1 : }
136 1 : case OAUTH_CANCELLED:
137 1 : next = {
138 1 : ...state,
139 1 : connectionStatus: ConnectionStatus.AUTH_CANCELLED,
140 1 : connectionHealth: ConnectionHealth.CANCELLED,
141 1 : details: {
142 1 : type: TRINO_CONNECTOR,
143 1 : value: {
144 1 : ...details,
145 1 : proto: {
146 1 : ...details.proto,
147 1 : setupTimings: {
148 1 : ...details.proto.setupTimings,
149 1 : oauthCancelledAt: dateToTimestamp(new Date())
150 1 : },
151 1 : },
152 1 : channel: null,
153 1 : }
154 1 : },
155 1 : };
156 1 : break;
157 1 : case OAUTH_FAILED:
158 1 : next = {
159 1 : ...state,
160 1 : connectionStatus: ConnectionStatus.AUTH_FAILED,
161 1 : connectionHealth: ConnectionHealth.FAILED,
162 1 : details: {
163 1 : type: TRINO_CONNECTOR,
164 1 : value: {
165 1 : ...details,
166 1 : proto: {
167 1 : ...details.proto,
168 1 : setupTimings: {
169 1 : ...details.proto.setupTimings,
170 1 : oauthFailedAt: dateToTimestamp(new Date())
171 1 : },
172 1 : channelError: action.value,
173 1 : },
174 1 : channel: null,
175 1 : }
176 1 : },
177 1 : };
178 1 : break;
179 1 : case GENERATING_PKCE_CHALLENGE:
180 1 : next = {
181 1 : ...state,
182 1 : connectionStatus: ConnectionStatus.PKCE_GENERATION_STARTED,
183 1 : connectionHealth: ConnectionHealth.CONNECTING,
184 1 : details: {
185 1 : type: TRINO_CONNECTOR,
186 1 : value: {
187 1 : ...details,
188 1 : proto: {
189 1 : ...details.proto,
190 1 : setupTimings: {
191 1 : ...details.proto.setupTimings,
192 1 : pkceGenStartedAt: dateToTimestamp(new Date())
193 1 : },
194 1 : },
195 1 : }
196 1 : },
197 1 : };
198 1 : break;
199 1 : case GENERATED_PKCE_CHALLENGE:
200 1 : next = {
201 1 : ...state,
202 1 : connectionStatus: ConnectionStatus.PKCE_GENERATED,
203 1 : connectionHealth: ConnectionHealth.CONNECTING,
204 1 : details: {
205 1 : type: TRINO_CONNECTOR,
206 1 : value: {
207 1 : ...details,
208 1 : proto: {
209 1 : ...details.proto,
210 1 : setupTimings: {
211 1 : ...details.proto.setupTimings,
212 1 : pkceGenFinishedAt: dateToTimestamp(new Date())
213 1 : },
214 1 : oauthState: {
215 1 : oauthPkce: action.value,
216 1 : authCode: details.proto.oauthState?.authCode ?? ({ token: "", createdAt: new Date().toISOString() } as auth.TemporaryToken),
217 1 : accessToken: details.proto.oauthState?.accessToken ?? ({} as connection.TrinoAccessToken),
218 1 : },
219 1 : },
220 1 : }
221 1 : },
222 1 : };
223 1 : break;
224 1 : case OAUTH_BROWSER_OPENED:
225 1 : next = {
226 1 : ...state,
227 1 : connectionStatus: ConnectionStatus.WAITING_FOR_OAUTH_CODE_VIA_LINK,
228 1 : connectionHealth: ConnectionHealth.CONNECTING,
229 1 : details: {
230 1 : type: TRINO_CONNECTOR,
231 1 : value: {
232 1 : ...details,
233 1 : proto: {
234 1 : ...details.proto,
235 1 : setupTimings: {
236 1 : ...details.proto.setupTimings,
237 1 : oauthBrowserOpenedAt: dateToTimestamp(new Date())
238 1 : },
239 1 : },
240 1 : }
241 1 : },
242 1 : };
243 1 : break;
244 1 : case RECEIVED_OAUTH_CODE:
245 1 : next = {
246 1 : ...state,
247 1 : connectionStatus: ConnectionStatus.OAUTH_CODE_RECEIVED,
248 1 : connectionHealth: ConnectionHealth.CONNECTING,
249 1 : details: {
250 1 : type: TRINO_CONNECTOR,
251 1 : value: {
252 1 : ...details,
253 1 : proto: {
254 1 : ...details.proto,
255 1 : setupTimings: {
256 1 : ...details.proto.setupTimings,
257 1 : oauthCodeReceivedAt: dateToTimestamp(new Date())
258 1 : },
259 1 : oauthState: {
260 1 : oauthPkce: details.proto.oauthState?.oauthPkce ?? ({ value: "", verifier: "" } as auth.OAuthPKCEChallenge),
261 1 : authCode: action.value,
262 1 : accessToken: details.proto.oauthState?.accessToken ?? ({} as connection.TrinoAccessToken),
263 1 : },
264 1 : },
265 1 : }
266 1 : },
267 1 : };
268 1 : break;
269 1 : case REQUESTING_ACCESS_TOKEN:
270 1 : next = {
271 1 : ...state,
272 1 : connectionStatus: ConnectionStatus.ACCESS_TOKEN_REQUESTED,
273 1 : connectionHealth: ConnectionHealth.CONNECTING,
274 1 : details: {
275 1 : type: TRINO_CONNECTOR,
276 1 : value: {
277 1 : ...details,
278 1 : proto: {
279 1 : ...details.proto,
280 1 : setupTimings: {
281 1 : ...details.proto.setupTimings,
282 1 : accessTokenRequestedAt: dateToTimestamp(new Date())
283 1 : },
284 1 : },
285 1 : }
286 1 : },
287 1 : };
288 1 : break;
289 1 : case RECEIVED_ACCESS_TOKEN:
290 1 : next = {
291 1 : ...state,
292 1 : connectionStatus: ConnectionStatus.ACCESS_TOKEN_RECEIVED,
293 1 : connectionHealth: ConnectionHealth.CONNECTING,
294 1 : details: {
295 1 : type: TRINO_CONNECTOR,
296 1 : value: {
297 1 : ...details,
298 1 : proto: {
299 1 : ...details.proto,
300 1 : setupTimings: {
301 1 : ...details.proto.setupTimings,
302 1 : accessTokenReceivedAt: dateToTimestamp(new Date())
303 1 : },
304 1 : oauthState: {
305 1 : oauthPkce: details.proto.oauthState?.oauthPkce ?? ({ value: "", verifier: "" } as auth.OAuthPKCEChallenge),
306 1 : authCode: details.proto.oauthState?.authCode ?? ({ token: "", createdAt: new Date().toISOString() } as auth.TemporaryToken),
307 1 : accessToken: action.value,
308 1 : },
309 1 : },
310 1 : }
311 1 : },
312 1 : };
313 1 : break;
314 :
315 : // Channel setup actions
316 1 : case TRINO_CHANNEL_SETUP_CANCELLED:
317 1 : next = {
318 1 : ...state,
319 1 : connectionStatus: ConnectionStatus.CHANNEL_SETUP_CANCELLED,
320 1 : connectionHealth: ConnectionHealth.CANCELLED,
321 1 : details: {
322 1 : type: TRINO_CONNECTOR,
323 1 : value: {
324 1 : ...details,
325 1 : proto: {
326 1 : ...details.proto,
327 1 : setupTimings: {
328 1 : ...details.proto.setupTimings,
329 1 : channelSetupCancelledAt: dateToTimestamp(new Date())
330 1 : },
331 1 : },
332 1 : channel: null
333 1 : }
334 1 : },
335 1 : };
336 1 : break;
337 1 : case TRINO_CHANNEL_SETUP_STARTED: {
338 1 : const newDetails: TrinoConnectionStateDetails = {
339 1 : ...details,
340 1 : proto: {
341 1 : ...details.proto,
342 1 : setupTimings: {
343 1 : ...details.proto.setupTimings,
344 1 : channelSetupStartedAt: dateToTimestamp(new Date())
345 1 : },
346 1 : setupParams: action.value,
347 1 : channelError: undefined,
348 1 : schemaResolutionError: undefined,
349 1 : healthCheckError: undefined,
350 1 : },
351 1 : channel: null
352 1 : };
353 1 : next = {
354 1 : ...state,
355 1 : connectionStatus: ConnectionStatus.CHANNEL_SETUP_STARTED,
356 1 : connectionHealth: ConnectionHealth.CONNECTING,
357 1 : details: {
358 1 : type: TRINO_CONNECTOR,
359 1 : value: newDetails,
360 1 : },
361 1 : };
362 1 : break;
363 1 : }
364 1 : case TRINO_CHANNEL_SETUP_FAILED:
365 1 : next = {
366 1 : ...state,
367 1 : connectionStatus: ConnectionStatus.CHANNEL_SETUP_FAILED,
368 1 : connectionHealth: ConnectionHealth.FAILED,
369 1 : details: {
370 1 : type: TRINO_CONNECTOR,
371 1 : value: {
372 1 : ...details,
373 1 : proto: {
374 1 : ...details.proto,
375 1 : setupTimings: {
376 1 : ...details.proto.setupTimings,
377 1 : channelSetupFailedAt: dateToTimestamp(new Date())
378 1 : },
379 1 : channelError: action.value,
380 1 : },
381 1 : channel: null
382 1 : }
383 1 : },
384 1 : };
385 1 : break;
386 1 : case TRINO_CHANNEL_READY:
387 1 : next = {
388 1 : ...state,
389 1 : connectionStatus: ConnectionStatus.CHANNEL_READY,
390 1 : connectionHealth: ConnectionHealth.CONNECTING,
391 1 : details: {
392 1 : type: TRINO_CONNECTOR,
393 1 : value: {
394 1 : ...details,
395 1 : proto: {
396 1 : ...details.proto,
397 1 : setupTimings: {
398 1 : ...details.proto.setupTimings,
399 1 : channelReadyAt: dateToTimestamp(new Date())
400 1 : },
401 1 : },
402 1 : channel: action.value
403 1 : }
404 1 : },
405 1 : };
406 : break;
407 1 : }
408 1 : return next;
409 1 : }
|
