1import type {
2 AuthAccessSnapshot,
3 AuthAccessStreamEvent,
4 AuthAccessStreamSnapshotEvent,
5} from "@t3tools/contracts";
6import { WS_METHODS } from "@t3tools/contracts";
7import * as Stream from "effect/Stream";
8import { Atom } from "effect/reactivity";
9
10import type { EnvironmentRegistry } from "../connection/registry.ts";
11import { subscribe } from "../rpc/client.ts";
12import { createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";
13
14export const EMPTY_AUTH_ACCESS_SNAPSHOT: AuthAccessSnapshot = {
15 pairingLinks: [],
16 clientSessions: [],
17};
18
19function upsertByKey<A>(
20 values: ReadonlyArray<A>,
21 next: A,
22 key: (value: A) => string,
23): ReadonlyArray<A> {
24 const nextKey = key(next);
25 return [...values.filter((value) => key(value) !== nextKey), next];
26}
27
28export function applyAuthAccessStreamEvent(
29 current: AuthAccessSnapshot,
30 event: AuthAccessStreamEvent,
31): AuthAccessSnapshot {
32 switch (event.type) {
33 case "snapshot":
34 return event.payload;
35 case "pairingLinkUpserted":
36 return {
37 ...current,
38 pairingLinks: upsertByKey(current.pairingLinks, event.payload, (value) => value.id),
39 };
40 case "pairingLinkRemoved":
41 return {
42 ...current,
43 pairingLinks: current.pairingLinks.filter((value) => value.id !== event.payload.id),
44 };
45 case "clientUpserted":
46 return {
47 ...current,
48 clientSessions: upsertByKey(
49 current.clientSessions,
50 event.payload,
51 (value) => value.sessionId,
52 ),
53 };
54 case "clientRemoved":
55 return {
56 ...current,
57 clientSessions: current.clientSessions.filter(
58 (value) => value.sessionId !== event.payload.sessionId,
59 ),
60 };
61 }
62}
63
64function projectAuthAccessSnapshot(
65 current: AuthAccessSnapshot,
66 event: AuthAccessStreamEvent,
67): readonly [AuthAccessSnapshot, ReadonlyArray<AuthAccessStreamEvent>] {
68 const snapshot = applyAuthAccessStreamEvent(current, event);
69 const projected: AuthAccessStreamSnapshotEvent = {
70 version: 1,
71 revision: event.revision,
72 type: "snapshot",
73 payload: snapshot,
74 };
75 return [snapshot, [projected]];
76}
77
78export function createAuthEnvironmentAtoms<R, E>(
79 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
80) {
81 return {
82 accessChanges: createEnvironmentSubscriptionAtomFamily(runtime, {
83 label: "environment-data:server:auth-access-changes",
84 subscribe: (_input: null) =>
85 subscribe(WS_METHODS.subscribeAuthAccess, {}).pipe(
86 Stream.mapAccum(() => EMPTY_AUTH_ACCESS_SNAPSHOT, projectAuthAccessSnapshot),
87 ),
88 }),
89 };
90}