126 lines
4.6 KiB
TypeScript
126 lines
4.6 KiB
TypeScript
import type { Server as IOServer, Socket, DefaultEventsMap } from "socket.io";
|
|
import { authorize } from "./authorize";
|
|
import { channelRoom } from "./channel-room";
|
|
import { matchChannelTemplate } from "./channel-template";
|
|
import type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface";
|
|
|
|
/** Shape of per-socket session data attached by Gate 1. */
|
|
type AppSocketData = { user: { userId: string; roles: string[] } | null };
|
|
|
|
/** Fully-typed Socket alias so `socket.data.user` resolves to `AppSocketData["user"]`. */
|
|
type AppSocket = Socket<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, AppSocketData>;
|
|
|
|
function parseCookies(header: string): Record<string, string> {
|
|
const out: Record<string, string> = {};
|
|
for (const part of header.split(";")) {
|
|
const [k, ...rest] = part.trim().split("=");
|
|
if (k) out[k] = decodeURIComponent(rest.join("="));
|
|
}
|
|
return out;
|
|
}
|
|
|
|
export class SocketIORealtimeServer implements IRealtimeServer {
|
|
private readonly io: IOServer;
|
|
private readonly opts: IRealtimeServerOptions;
|
|
|
|
constructor(opts: IRealtimeServerOptions) {
|
|
this.opts = opts;
|
|
this.io = opts.io;
|
|
}
|
|
|
|
async start(): Promise<void> {
|
|
const { authenticator, registry } = this.opts;
|
|
|
|
// Gate 1: connect — read cookie, authenticate, attach user.
|
|
// If the authenticator throws (e.g. malformed token), the connection is
|
|
// rejected so the client receives `connect_error`.
|
|
(this.io as IOServer<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, AppSocketData>).use(
|
|
async (socket, next) => {
|
|
try {
|
|
const cookies = parseCookies(socket.handshake.headers.cookie ?? "");
|
|
socket.data.user = await authenticator.authenticate({
|
|
cookies,
|
|
headers: socket.handshake.headers as Record<string, string>,
|
|
});
|
|
next();
|
|
} catch (err) {
|
|
next(err instanceof Error ? err : new Error(String(err)));
|
|
}
|
|
},
|
|
);
|
|
|
|
this.io.on("connection", (rawSocket) => {
|
|
const socket = rawSocket as AppSocket;
|
|
|
|
// Gate 2: subscribe.
|
|
socket.on("subscribe", async (requestedName: string, ack?: (r: unknown) => void) => {
|
|
// Find a registered descriptor whose name (or template) matches requestedName.
|
|
// listChannels() covers both inbound descriptors and outbound-only channels.
|
|
let matched: { descriptor: { name: string; scope: unknown }; params: Record<string, string> } | null = null;
|
|
for (const descriptor of registry.listChannels()) {
|
|
const m = matchChannelTemplate(descriptor.name, requestedName);
|
|
if (m) {
|
|
matched = { descriptor, params: m.params };
|
|
break;
|
|
}
|
|
}
|
|
if (!matched) {
|
|
ack?.({ ok: false, error: "unknown_channel" });
|
|
return;
|
|
}
|
|
|
|
const allowed = await authorize(
|
|
matched.descriptor as never,
|
|
matched.params,
|
|
socket.data.user ?? null,
|
|
);
|
|
if (!allowed) {
|
|
ack?.({ ok: false, error: "forbidden" });
|
|
return;
|
|
}
|
|
|
|
socket.join(channelRoom(requestedName));
|
|
ack?.({ ok: true });
|
|
});
|
|
|
|
// Gate 3: inbound — one listener per registered channel.
|
|
for (const entry of registry.list()) {
|
|
socket.on(entry.descriptor.name, async (payload: unknown, ack?: (r: unknown) => void) => {
|
|
const parsed = entry.descriptor.schema.safeParse(payload);
|
|
if (!parsed.success) {
|
|
ack?.({ ok: false, error: "invalid_input" });
|
|
return;
|
|
}
|
|
|
|
// For userScoped channels, derive params from the authenticated user
|
|
// so the owner's own socket passes the `params.userId === user.userId` check.
|
|
const scope = entry.descriptor.scope;
|
|
const params: Record<string, string> =
|
|
typeof scope === "object" && "userScoped" in scope
|
|
? { userId: socket.data.user?.userId ?? "" }
|
|
: {};
|
|
|
|
const allowed = await authorize(entry.descriptor, params, socket.data.user ?? null);
|
|
if (!allowed) {
|
|
ack?.({ ok: false, error: "forbidden" });
|
|
return;
|
|
}
|
|
try {
|
|
await entry.handler(parsed.data, {
|
|
userId: socket.data.user?.userId ?? null,
|
|
roles: socket.data.user?.roles ?? [],
|
|
});
|
|
ack?.({ ok: true });
|
|
} catch {
|
|
ack?.({ ok: false, error: "handler_error" });
|
|
}
|
|
});
|
|
}
|
|
});
|
|
}
|
|
|
|
async stop(): Promise<void> {
|
|
await new Promise<void>((resolve) => this.io.close(() => resolve()));
|
|
}
|
|
}
|