import { WsPush, type WsPushChannel, type WsPushData, type WsPushEnvelopeBase,} from "@t3tools/contracts";import { Deferred, Effect, Queue, Ref, Schema } from "effect";import type { Scope } from "effect"; export interface ServerPushClient { readonly send: (message: string) => Effect.Effect<boolean>;} type PushTarget = | { readonly kind: "all" } | { readonly kind: "client"; readonly client: ServerPushClient }; interface PushJob<C extends WsPushChannel = WsPushChannel> { readonly channel: C; readonly data: WsPushData<C>; readonly target: PushTarget; readonly delivered: Deferred.Deferred<boolean> | null;} export interface ServerPushBus { readonly publishAll: <C extends WsPushChannel>( channel: C, data: WsPushData<C>, ) => Effect.Effect<void>; readonly publishClient: <C extends WsPushChannel>( client: ServerPushClient, channel: C, data: WsPushData<C>, ) => Effect.Effect<boolean>;} export const makeServerPushBus = (input: { readonly clients: Ref.Ref<Set<ServerPushClient>>; readonly logOutgoingPush: (push: WsPushEnvelopeBase, recipients: number) => void;}): Effect.Effect<ServerPushBus, never, Scope.Scope> => Effect.gen(function* () { const nextSequence = yield* Ref.make(0); const queue = yield* Queue.unbounded<PushJob>(); const encodePush = Schema.encodeUnknownEffect(Schema.fromJsonString(WsPush)); const settleDelivery = (job: PushJob, delivered: boolean) => job.delivered === null ? Effect.void : Deferred.succeed(job.delivered, delivered).pipe(Effect.orDie); const send = Effect.fnUntraced(function* (job: PushJob) { const sequence = yield* Ref.updateAndGet(nextSequence, (current) => current + 1); const push: WsPushEnvelopeBase = { type: "push", sequence, channel: job.channel, data: job.data, }; const recipients = job.target.kind === "all" ? yield* Ref.get(input.clients) : new Set([job.target.client]); return yield* encodePush(push).pipe( Effect.flatMap((message) => Effect.forEach(recipients, (client) => client.send(message), { concurrency: "unbounded", }).pipe( Effect.map((deliveries) => { const recipientCount = deliveries.filter(Boolean).length; input.logOutgoingPush(push, recipientCount); return recipientCount > 0; }), ), ), ); }); yield* Effect.forkScoped( Effect.forever( Queue.take(queue).pipe( Effect.flatMap((job) => send(job).pipe( Effect.tap((delivered) => settleDelivery(job, delivered)), Effect.tapCause(() => settleDelivery(job, false)), Effect.ignoreCause({ log: true }), ), ), ), ), ); const publish = (target: PushTarget) => <C extends WsPushChannel>(channel: C, data: WsPushData<C>) => Queue.offer(queue, { channel, data, target, delivered: null, }).pipe(Effect.asVoid); return { publishAll: publish({ kind: "all" }), publishClient: (client, channel, data) => Effect.gen(function* () { const delivered = yield* Deferred.make<boolean>(); yield* Queue.offer(queue, { channel, data, target: { kind: "client", client }, delivered, }).pipe(Effect.asVoid); return yield* Deferred.await(delivered); }), } satisfies ServerPushBus; });
import {
WsPush,
type WsPushChannel,
type WsPushData,
type WsPushEnvelopeBase,
} from "@t3tools/contracts";
import { Deferred, Effect, Queue, Ref, Schema } from "effect";
import type { Scope } from "effect";
export interface ServerPushClient {
readonly send: (message: string) => Effect.Effect<boolean>;
}
type PushTarget =
| { readonly kind: "all" }
| { readonly kind: "client"; readonly client: ServerPushClient };
interface PushJob<C extends WsPushChannel = WsPushChannel> {
readonly channel: C;
readonly data: WsPushData<C>;
readonly target: PushTarget;
readonly delivered: Deferred.Deferred<boolean> | null;
}
export interface ServerPushBus {
readonly publishAll: <C extends WsPushChannel>(
channel: C,
data: WsPushData<C>,
) => Effect.Effect<void>;
readonly publishClient: <C extends WsPushChannel>(
client: ServerPushClient,
channel: C,
data: WsPushData<C>,
) => Effect.Effect<boolean>;
}
export const makeServerPushBus = (input: {
readonly clients: Ref.Ref<Set<ServerPushClient>>;
readonly logOutgoingPush: (push: WsPushEnvelopeBase, recipients: number) => void;
}): Effect.Effect<ServerPushBus, never, Scope.Scope> =>
Effect.gen(function* () {
const nextSequence = yield* Ref.make(0);
const queue = yield* Queue.unbounded<PushJob>();
const encodePush = Schema.encodeUnknownEffect(Schema.fromJsonString(WsPush));
const settleDelivery = (job: PushJob, delivered: boolean) =>
job.delivered === null
? Effect.void
: Deferred.succeed(job.delivered, delivered).pipe(Effect.orDie);
const send = Effect.fnUntraced(function* (job: PushJob) {
const sequence = yield* Ref.updateAndGet(nextSequence, (current) => current + 1);
const push: WsPushEnvelopeBase = {
type: "push",
sequence,
channel: job.channel,
data: job.data,
};
const recipients =
job.target.kind === "all" ? yield* Ref.get(input.clients) : new Set([job.target.client]);
return yield* encodePush(push).pipe(
Effect.flatMap((message) =>
Effect.forEach(recipients, (client) => client.send(message), {
concurrency: "unbounded",
}).pipe(
Effect.map((deliveries) => {
const recipientCount = deliveries.filter(Boolean).length;
input.logOutgoingPush(push, recipientCount);
return recipientCount > 0;
}),
),
),
);
});
yield* Effect.forkScoped(
Effect.forever(
Queue.take(queue).pipe(
Effect.flatMap((job) =>
send(job).pipe(
Effect.tap((delivered) => settleDelivery(job, delivered)),
Effect.tapCause(() => settleDelivery(job, false)),
Effect.ignoreCause({ log: true }),
),
),
),
),
);
const publish =
(target: PushTarget) =>
<C extends WsPushChannel>(channel: C, data: WsPushData<C>) =>
Queue.offer(queue, {
channel,
data,
target,
delivered: null,
}).pipe(Effect.asVoid);
return {
publishAll: publish({ kind: "all" }),
publishClient: (client, channel, data) =>
Effect.gen(function* () {
const delivered = yield* Deferred.make<boolean>();
yield* Queue.offer(queue, {
channel,
data,
target: { kind: "client", client },
delivered,
}).pipe(Effect.asVoid);
return yield* Deferred.await(delivered);
}),
} satisfies ServerPushBus;
});