import type { AuthAccessSnapshot, AuthAccessStreamEvent, AuthAccessStreamSnapshotEvent,} from "@t3tools/contracts";import { WS_METHODS } from "@t3tools/contracts";import * as Stream from "effect/Stream";import { Atom } from "effect/unstable/reactivity"; import type { EnvironmentRegistry } from "../connection/registry.ts";import { subscribe } from "../rpc/client.ts";import { createEnvironmentSubscriptionAtomFamily } from "./runtime.ts"; export const EMPTY_AUTH_ACCESS_SNAPSHOT: AuthAccessSnapshot = { pairingLinks: [], clientSessions: [],}; function upsertByKey<A>( values: ReadonlyArray<A>, next: A, key: (value: A) => string,): ReadonlyArray<A> { const nextKey = key(next); return [...values.filter((value) => key(value) !== nextKey), next];} export function applyAuthAccessStreamEvent( current: AuthAccessSnapshot, event: AuthAccessStreamEvent,): AuthAccessSnapshot { switch (event.type) { case "snapshot": return event.payload; case "pairingLinkUpserted": return { ...current, pairingLinks: upsertByKey(current.pairingLinks, event.payload, (value) => value.id), }; case "pairingLinkRemoved": return { ...current, pairingLinks: current.pairingLinks.filter((value) => value.id !== event.payload.id), }; case "clientUpserted": return { ...current, clientSessions: upsertByKey( current.clientSessions, event.payload, (value) => value.sessionId, ), }; case "clientRemoved": return { ...current, clientSessions: current.clientSessions.filter( (value) => value.sessionId !== event.payload.sessionId, ), }; }} export function projectAuthAccessSnapshot( current: AuthAccessSnapshot, event: AuthAccessStreamEvent,): readonly [AuthAccessSnapshot, ReadonlyArray<AuthAccessStreamEvent>] { const snapshot = applyAuthAccessStreamEvent(current, event); const projected: AuthAccessStreamSnapshotEvent = { version: 1, revision: event.revision, type: "snapshot", payload: snapshot, }; return [snapshot, [projected]];} export function createAuthEnvironmentAtoms<R, E>( runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,) { return { accessChanges: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:server:auth-access-changes", subscribe: (_input: null) => subscribe(WS_METHODS.subscribeAuthAccess, {}).pipe( Stream.mapAccum(() => EMPTY_AUTH_ACCESS_SNAPSHOT, projectAuthAccessSnapshot), ), }), };}
import type {
AuthAccessSnapshot,
AuthAccessStreamEvent,
AuthAccessStreamSnapshotEvent,
} from "@t3tools/contracts";
import { WS_METHODS } from "@t3tools/contracts";
import * as Stream from "effect/Stream";
import { Atom } from "effect/unstable/reactivity";
import type { EnvironmentRegistry } from "../connection/registry.ts";
import { subscribe } from "../rpc/client.ts";
import { createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";
export const EMPTY_AUTH_ACCESS_SNAPSHOT: AuthAccessSnapshot = {
pairingLinks: [],
clientSessions: [],
};
function upsertByKey<A>(
values: ReadonlyArray<A>,
next: A,
key: (value: A) => string,
): ReadonlyArray<A> {
const nextKey = key(next);
return [...values.filter((value) => key(value) !== nextKey), next];
}
export function applyAuthAccessStreamEvent(
current: AuthAccessSnapshot,
event: AuthAccessStreamEvent,
): AuthAccessSnapshot {
switch (event.type) {
case "snapshot":
return event.payload;
case "pairingLinkUpserted":
return {
...current,
pairingLinks: upsertByKey(current.pairingLinks, event.payload, (value) => value.id),
};
case "pairingLinkRemoved":
return {
...current,
pairingLinks: current.pairingLinks.filter((value) => value.id !== event.payload.id),
};
case "clientUpserted":
return {
...current,
clientSessions: upsertByKey(
current.clientSessions,
event.payload,
(value) => value.sessionId,
),
};
case "clientRemoved":
return {
...current,
clientSessions: current.clientSessions.filter(
(value) => value.sessionId !== event.payload.sessionId,
),
};
}
}
export function projectAuthAccessSnapshot(
current: AuthAccessSnapshot,
event: AuthAccessStreamEvent,
): readonly [AuthAccessSnapshot, ReadonlyArray<AuthAccessStreamEvent>] {
const snapshot = applyAuthAccessStreamEvent(current, event);
const projected: AuthAccessStreamSnapshotEvent = {
version: 1,
revision: event.revision,
type: "snapshot",
payload: snapshot,
};
return [snapshot, [projected]];
}
export function createAuthEnvironmentAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
) {
return {
accessChanges: createEnvironmentSubscriptionAtomFamily(runtime, {
label: "environment-data:server:auth-access-changes",
subscribe: (_input: null) =>
subscribe(WS_METHODS.subscribeAuthAccess, {}).pipe(
Stream.mapAccum(() => EMPTY_AUTH_ACCESS_SNAPSHOT, projectAuthAccessSnapshot),
),
}),
};
}