1import type {
2 RelayClientEnvironmentRecord,
3 RelayEnvironmentLinkProofPayload,
4 RelayEnvironmentLinkRequest,
5 RelayManagedEndpoint,
6} from "@t3tools/contracts/relay";
7import * as Context from "effect/Context";
8import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Schema from "effect/Schema";
12import { and, eq, isNull, or, sql } from "drizzle-orm";
13
14import * as RelayDb from "../db.ts";
15import { relayEnvironmentLinks } from "../persistence/schema.ts";
16
17export interface RelayLinkedEnvironmentRecord extends RelayClientEnvironmentRecord {
18 readonly environmentPublicKey: string;
19}
20
21export interface AgentAwarenessDeliveryUserRecord {
22 readonly userId: string;
23 readonly notificationsEnabled: boolean;
24 readonly liveActivitiesEnabled: boolean;
25}
26
27export class EnvironmentLinkUpsertPersistenceError extends Schema.TaggedError<EnvironmentLinkUpsertPersistenceError>()(
28 "EnvironmentLinkUpsertPersistenceError",
29 {
30 userId: Schema.String,
31 environmentId: Schema.String,
32 deviceId: Schema.optionalKey(Schema.String),
33 cause: Schema.Defect(),
34 },
35) {
36 override get message(): string {
37 return `Failed to persist environment link for user '${this.userId}', environment '${this.environmentId}'`;
38 }
39}
40
41export class EnvironmentLinkUserListPersistenceError extends Schema.TaggedError<EnvironmentLinkUserListPersistenceError>()(
42 "EnvironmentLinkUserListPersistenceError",
43 {
44 environmentId: Schema.String,
45 cause: Schema.Defect(),
46 },
47) {
48 override get message(): string {
49 return `Environment link user query 'list-delivery-users' failed for environment '${this.environmentId}'`;
50 }
51}
52
53export class EnvironmentLinkListPersistenceError extends Schema.TaggedError<EnvironmentLinkListPersistenceError>()(
54 "EnvironmentLinkListPersistenceError",
55 {
56 userId: Schema.String,
57 cause: Schema.Defect(),
58 },
59) {
60 override get message(): string {
61 return `Failed to list environment links for user '${this.userId}'`;
62 }
63}
64
65export class EnvironmentLinkLookupPersistenceError extends Schema.TaggedError<EnvironmentLinkLookupPersistenceError>()(
66 "EnvironmentLinkLookupPersistenceError",
67 {
68 userId: Schema.String,
69 environmentId: Schema.String,
70 cause: Schema.Defect(),
71 },
72) {
73 override get message(): string {
74 return `Failed to look up environment link for user '${this.userId}', environment '${this.environmentId}'`;
75 }
76}
77
78export class EnvironmentLinkEnvironmentLookupPersistenceError extends Schema.TaggedError<EnvironmentLinkEnvironmentLookupPersistenceError>()(
79 "EnvironmentLinkEnvironmentLookupPersistenceError",
80 {
81 environmentId: Schema.String,
82 cause: Schema.Defect(),
83 },
84) {
85 override get message(): string {
86 return `Failed to look up active managed links for environment '${this.environmentId}'`;
87 }
88}
89
90export class EnvironmentLinkRevokePersistenceError extends Schema.TaggedError<EnvironmentLinkRevokePersistenceError>()(
91 "EnvironmentLinkRevokePersistenceError",
92 {
93 userId: Schema.String,
94 environmentId: Schema.String,
95 cause: Schema.Defect(),
96 },
97) {
98 override get message(): string {
99 return `Failed to revoke environment link for user '${this.userId}', environment '${this.environmentId}'`;
100 }
101}
102
103export class EnvironmentLinks extends Context.Service<
104 EnvironmentLinks,
105 {
106 readonly upsert: (input: {
107 readonly userId: string;
108 readonly request: RelayEnvironmentLinkRequest;
109 readonly proof: RelayEnvironmentLinkProofPayload;
110 readonly endpoint: RelayManagedEndpoint;
111 }) => Effect.Effect<void, EnvironmentLinkUpsertPersistenceError>;
112 readonly listDeliveryUsersForEnvironment: (input: {
113 readonly environmentId: string;
114 readonly environmentPublicKey: string;
115 }) => Effect.Effect<
116 ReadonlyArray<AgentAwarenessDeliveryUserRecord>,
117 EnvironmentLinkUserListPersistenceError
118 >;
119 readonly listForUser: (input: {
120 readonly userId: string;
121 }) => Effect.Effect<
122 ReadonlyArray<RelayClientEnvironmentRecord>,
123 EnvironmentLinkListPersistenceError
124 >;
125 readonly getForUser: (input: {
126 readonly userId: string;
127 readonly environmentId: string;
128 }) => Effect.Effect<RelayLinkedEnvironmentRecord | null, EnvironmentLinkLookupPersistenceError>;
129 /**
130 * Active relay-managed links for an environment, narrowed to one user or to
131 * links proven by one environment key. The environment id alone is public
132 * and any account can link it, so callers acting on it must narrow.
133 */
134 readonly findActiveManagedForEnvironment: (input: {
135 readonly environmentId: string;
136 readonly userId?: string;
137 readonly environmentPublicKey?: string;
138 }) => Effect.Effect<
139 ReadonlyArray<
140 RelayLinkedEnvironmentRecord & {
141 readonly userId: string;
142 readonly holdWebhooksWhileOffline: boolean;
143 }
144 >,
145 EnvironmentLinkEnvironmentLookupPersistenceError
146 >;
147 /** Sets the webhook-hold opt-in on the active links proven by one environment key. */
148 readonly setHoldWebhooksWhileOffline: (input: {
149 readonly environmentId: string;
150 readonly environmentPublicKey: string;
151 readonly holdWebhooksWhileOffline: boolean;
152 }) => Effect.Effect<void, EnvironmentLinkEnvironmentLookupPersistenceError>;
153 readonly revokeForUser: (input: {
154 readonly userId: string;
155 readonly environmentId: string;
156 }) => Effect.Effect<boolean, EnvironmentLinkRevokePersistenceError>;
157 }
158>()("t3code-relay/environments/EnvironmentLinks") {}
159
160function agentAwarenessDeliveryUserKeyCondition(input: {
161 readonly environmentId: string;
162 readonly environmentPublicKey: string;
163}) {
164 return and(
165 eq(relayEnvironmentLinks.environmentId, input.environmentId),
166 isNull(relayEnvironmentLinks.revokedAt),
167 or(
168 eq(relayEnvironmentLinks.notificationsEnabled, true),
169 eq(relayEnvironmentLinks.liveActivitiesEnabled, true),
170 ),
171 eq(relayEnvironmentLinks.environmentPublicKey, input.environmentPublicKey),
172 );
173}
174
175const make = Effect.gen(function* () {
176 const db = yield* RelayDb.RelayDb;
177
178 return EnvironmentLinks.of({
179 upsert: Effect.fn("relay.environment_links.upsert")(function* (input) {
180 yield* Effect.annotateCurrentSpan({
181 "relay.environment_id": input.proof.environmentId,
182 });
183 const now = DateTime.formatIso(yield* DateTime.now);
184 const { request, proof } = input;
185 const environmentId = proof.environmentId;
186 // The webhook-hold opt-in belongs to the environment: a new or re-made
187 // link carries it over from the environment's other active links. Only
188 // links proven by the same key count; an environment id is public, so
189 // anyone can link one and switch the opt-in on for their own link.
190 const inheritedHoldWebhooks = sql<boolean>`EXISTS (
191 SELECT 1 FROM ${relayEnvironmentLinks} AS other
192 WHERE other.environment_id = ${environmentId}
193 AND other.environment_public_key = ${proof.environmentPublicKey}
194 AND other.revoked_at IS NULL
195 AND other.hold_webhooks_while_offline
196 )`;
197 const { endpoint } = input;
198 yield* db
199 .insert(relayEnvironmentLinks)
200 .values({
201 userId: input.userId,
202 environmentId,
203 holdWebhooksWhileOffline: inheritedHoldWebhooks,
204 environmentLabel: proof.descriptor.label,
205 environmentPublicKey: proof.environmentPublicKey,
206 endpointHttpBaseUrl: endpoint.httpBaseUrl,
207 endpointWsBaseUrl: endpoint.wsBaseUrl,
208 endpointProviderKind: endpoint.providerKind,
209 notificationsEnabled: request.notificationsEnabled,
210 liveActivitiesEnabled: request.liveActivitiesEnabled,
211 managedTunnelsEnabled: request.managedTunnelsEnabled,
212 createdByDeviceId: request.deviceId ?? null,
213 revokedAt: null,
214 createdAt: now,
215 updatedAt: now,
216 })
217 .onConflictDoUpdate({
218 target: [relayEnvironmentLinks.userId, relayEnvironmentLinks.environmentId],
219 set: {
220 environmentPublicKey: proof.environmentPublicKey,
221 environmentLabel: proof.descriptor.label,
222 endpointHttpBaseUrl: endpoint.httpBaseUrl,
223 endpointWsBaseUrl: endpoint.wsBaseUrl,
224 endpointProviderKind: endpoint.providerKind,
225 notificationsEnabled: request.notificationsEnabled,
226 liveActivitiesEnabled: request.liveActivitiesEnabled,
227 managedTunnelsEnabled: request.managedTunnelsEnabled,
228 createdByDeviceId: request.deviceId ?? null,
229 revokedAt: null,
230 updatedAt: now,
231 holdWebhooksWhileOffline: inheritedHoldWebhooks,
232 },
233 })
234 .pipe(
235 Effect.mapError(
236 (cause) =>
237 new EnvironmentLinkUpsertPersistenceError({
238 userId: input.userId,
239 environmentId,
240 ...(request.deviceId === undefined ? {} : { deviceId: request.deviceId }),
241 cause,
242 }),
243 ),
244 );
245 }),
246
247 listDeliveryUsersForEnvironment: Effect.fn(
248 "relay.environment_links.list_delivery_users_for_environment",
249 )(function* (input) {
250 yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId });
251 return yield* db
252 .select({
253 userId: relayEnvironmentLinks.userId,
254 notificationsEnabled: relayEnvironmentLinks.notificationsEnabled,
255 liveActivitiesEnabled: relayEnvironmentLinks.liveActivitiesEnabled,
256 })
257 .from(relayEnvironmentLinks)
258 .where(agentAwarenessDeliveryUserKeyCondition(input))
259 .pipe(
260 Effect.map((rows) =>
261 rows.map((row) => ({
262 userId: row.userId,
263 notificationsEnabled: row.notificationsEnabled,
264 liveActivitiesEnabled: row.liveActivitiesEnabled,
265 })),
266 ),
267 Effect.mapError(
268 (cause) =>
269 new EnvironmentLinkUserListPersistenceError({
270 environmentId: input.environmentId,
271 cause,
272 }),
273 ),
274 );
275 }),
276
277 listForUser: Effect.fn("relay.environment_links.list_for_user")(function* (input) {
278 return yield* db
279 .select({
280 environmentId: relayEnvironmentLinks.environmentId,
281 environmentLabel: relayEnvironmentLinks.environmentLabel,
282 endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl,
283 endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl,
284 endpointProviderKind: relayEnvironmentLinks.endpointProviderKind,
285 createdAt: relayEnvironmentLinks.createdAt,
286 })
287 .from(relayEnvironmentLinks)
288 .where(
289 and(
290 eq(relayEnvironmentLinks.userId, input.userId),
291 isNull(relayEnvironmentLinks.revokedAt),
292 ),
293 )
294 .pipe(
295 Effect.map((rows) =>
296 rows.map((row) => ({
297 environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"],
298 label:
299 row.environmentLabel.trim().length > 0 ? row.environmentLabel : row.environmentId,
300 endpoint: {
301 httpBaseUrl: row.endpointHttpBaseUrl,
302 wsBaseUrl: row.endpointWsBaseUrl,
303 providerKind:
304 row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"],
305 },
306 linkedAt: row.createdAt,
307 })),
308 ),
309 Effect.mapError(
310 (cause) =>
311 new EnvironmentLinkListPersistenceError({
312 userId: input.userId,
313 cause,
314 }),
315 ),
316 );
317 }),
318
319 getForUser: Effect.fn("relay.environment_links.get_for_user")(function* (input) {
320 yield* Effect.annotateCurrentSpan({
321 "relay.environment_id": input.environmentId,
322 });
323 return yield* db
324 .select({
325 environmentId: relayEnvironmentLinks.environmentId,
326 environmentLabel: relayEnvironmentLinks.environmentLabel,
327 environmentPublicKey: relayEnvironmentLinks.environmentPublicKey,
328 endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl,
329 endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl,
330 endpointProviderKind: relayEnvironmentLinks.endpointProviderKind,
331 createdAt: relayEnvironmentLinks.createdAt,
332 })
333 .from(relayEnvironmentLinks)
334 .where(
335 and(
336 eq(relayEnvironmentLinks.userId, input.userId),
337 eq(relayEnvironmentLinks.environmentId, input.environmentId),
338 isNull(relayEnvironmentLinks.revokedAt),
339 ),
340 )
341 .limit(1)
342 .pipe(
343 Effect.map((rows) => {
344 const row = rows[0];
345 return row
346 ? {
347 environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"],
348 label:
349 row.environmentLabel.trim().length > 0
350 ? row.environmentLabel
351 : row.environmentId,
352 endpoint: {
353 httpBaseUrl: row.endpointHttpBaseUrl,
354 wsBaseUrl: row.endpointWsBaseUrl,
355 providerKind:
356 row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"],
357 },
358 environmentPublicKey: row.environmentPublicKey,
359 linkedAt: row.createdAt,
360 }
361 : null;
362 }),
363 Effect.mapError(
364 (cause) =>
365 new EnvironmentLinkLookupPersistenceError({
366 userId: input.userId,
367 environmentId: input.environmentId,
368 cause,
369 }),
370 ),
371 );
372 }),
373
374 findActiveManagedForEnvironment: Effect.fn(
375 "relay.environment_links.find_active_managed_for_environment",
376 )(function* (input) {
377 yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId });
378 return yield* db
379 .select({
380 userId: relayEnvironmentLinks.userId,
381 environmentId: relayEnvironmentLinks.environmentId,
382 environmentLabel: relayEnvironmentLinks.environmentLabel,
383 environmentPublicKey: relayEnvironmentLinks.environmentPublicKey,
384 endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl,
385 endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl,
386 endpointProviderKind: relayEnvironmentLinks.endpointProviderKind,
387 createdAt: relayEnvironmentLinks.createdAt,
388 holdWebhooksWhileOffline: relayEnvironmentLinks.holdWebhooksWhileOffline,
389 })
390 .from(relayEnvironmentLinks)
391 .where(
392 and(
393 eq(relayEnvironmentLinks.environmentId, input.environmentId),
394 isNull(relayEnvironmentLinks.revokedAt),
395 eq(relayEnvironmentLinks.endpointProviderKind, "cloudflare_tunnel"),
396 eq(relayEnvironmentLinks.managedTunnelsEnabled, true),
397 input.userId === undefined ? undefined : eq(relayEnvironmentLinks.userId, input.userId),
398 input.environmentPublicKey === undefined
399 ? undefined
400 : eq(relayEnvironmentLinks.environmentPublicKey, input.environmentPublicKey),
401 ),
402 )
403 // At most one row per user who linked this environment.
404 .pipe(
405 Effect.map((rows) =>
406 rows.map((row) => ({
407 userId: row.userId,
408 environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"],
409 label:
410 row.environmentLabel.trim().length > 0 ? row.environmentLabel : row.environmentId,
411 endpoint: {
412 httpBaseUrl: row.endpointHttpBaseUrl,
413 wsBaseUrl: row.endpointWsBaseUrl,
414 providerKind:
415 row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"],
416 },
417 environmentPublicKey: row.environmentPublicKey,
418 linkedAt: row.createdAt,
419 holdWebhooksWhileOffline: row.holdWebhooksWhileOffline,
420 })),
421 ),
422 Effect.mapError(
423 (cause) =>
424 new EnvironmentLinkEnvironmentLookupPersistenceError({
425 environmentId: input.environmentId,
426 cause,
427 }),
428 ),
429 );
430 }),
431
432 setHoldWebhooksWhileOffline: Effect.fn(
433 "relay.environment_links.set_hold_webhooks_while_offline",
434 )(function* (input) {
435 yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId });
436 yield* db
437 .update(relayEnvironmentLinks)
438 .set({ holdWebhooksWhileOffline: input.holdWebhooksWhileOffline })
439 .where(
440 and(
441 eq(relayEnvironmentLinks.environmentId, input.environmentId),
442 eq(relayEnvironmentLinks.environmentPublicKey, input.environmentPublicKey),
443 isNull(relayEnvironmentLinks.revokedAt),
444 ),
445 )
446 .pipe(
447 Effect.mapError(
448 (cause) =>
449 new EnvironmentLinkEnvironmentLookupPersistenceError({
450 environmentId: input.environmentId,
451 cause,
452 }),
453 ),
454 );
455 }),
456
457 revokeForUser: Effect.fn("relay.environment_links.revoke_for_user")(function* (input) {
458 yield* Effect.annotateCurrentSpan({
459 "relay.environment_id": input.environmentId,
460 });
461 const revokedAt = DateTime.formatIso(yield* DateTime.now);
462 const rows = yield* db
463 .update(relayEnvironmentLinks)
464 .set({
465 revokedAt,
466 updatedAt: revokedAt,
467 })
468 .where(
469 and(
470 eq(relayEnvironmentLinks.userId, input.userId),
471 eq(relayEnvironmentLinks.environmentId, input.environmentId),
472 isNull(relayEnvironmentLinks.revokedAt),
473 ),
474 )
475 .returning({ environmentId: relayEnvironmentLinks.environmentId })
476 .pipe(
477 Effect.mapError(
478 (cause) =>
479 new EnvironmentLinkRevokePersistenceError({
480 userId: input.userId,
481 environmentId: input.environmentId,
482 cause,
483 }),
484 ),
485 );
486 return rows.length > 0;
487 }),
488 });
489});
490
491export const layer = Layer.effect(EnvironmentLinks, make);