mirror of
https://github.com/Termix-SSH/Termix.git
synced 2026-10-29 06:35:24 +00:00
feat: move session sharing and collab rooms into a plugin (B12)
This commit is contained in:
1 parent
791d8656a1
commit
a276f6898c
207 files changed
+35771
-11072
No files matched your search
@@ -0,0 +1,51 @@
|
||||
import type { PluginContext } from "@termix/plugin-sdk/backend";
|
||||
import type { LiveSessions } from "./live.js";
|
||||
import type { RateLimiter } from "./rate-limit.js";
|
||||
import type {
|
||||
Directory,
|
||||
RoomRepository,
|
||||
ShareRepository,
|
||||
} from "./repositories.js";
|
||||
import type { CollabRoomHub } from "./room-hub.js";
|
||||
import type { CollabRuntimeStore } from "./runtime-store.js";
|
||||
|
||||
/** Everything the routes and services share, built once in activate. */
|
||||
export interface SharingDeps {
|
||||
ctx: PluginContext;
|
||||
shares: ShareRepository;
|
||||
rooms: RoomRepository;
|
||||
directory: Directory;
|
||||
live: LiveSessions;
|
||||
hub: CollabRoomHub;
|
||||
store: CollabRuntimeStore;
|
||||
/** /resolve/:linkToken, 30 a minute per IP. */
|
||||
resolveLimiter: RateLimiter;
|
||||
/** Room guest links, 60 a minute per IP. */
|
||||
guestLimiter: RateLimiter;
|
||||
isSharingEnabledForHost: (hostId: number) => Promise<boolean>;
|
||||
}
|
||||
|
||||
export const PROTOCOLS = ["ssh", "rdp", "vnc", "telnet"] as const;
|
||||
|
||||
/** Where SSH shares are joined: the terminal's own socket. */
|
||||
export const TERMINAL_WS_PATH = "/plugin-ws/ssh-terminal/terminal";
|
||||
/** Where remote desktop viewers connect. */
|
||||
export const DISPLAY_WS_PATH = "/plugin-ws/remote-desktop/display";
|
||||
|
||||
export function clientIp(req: {
|
||||
ip?: string;
|
||||
socket?: { remoteAddress?: string };
|
||||
}): string {
|
||||
return req.ip || req.socket?.remoteAddress || "unknown";
|
||||
}
|
||||
|
||||
export function isNonEmptyString(value: unknown): value is string {
|
||||
return typeof value === "string" && value.trim().length > 0;
|
||||
}
|
||||
|
||||
/** The acting user, which core set from the authenticated request. */
|
||||
export function actorOf(ctx: PluginContext): string {
|
||||
const userId = ctx.currentActor();
|
||||
if (!userId) throw new Error("No acting user");
|
||||
return userId;
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
import type { Router } from "express";
|
||||
import type { PluginContext } from "@termix/plugin-sdk/backend";
|
||||
import type { SharingDeps } from "./deps.js";
|
||||
import { createLiveSessions, isSharingEnabledForHost } from "./live.js";
|
||||
import { createRateLimiter } from "./rate-limit.js";
|
||||
import {
|
||||
createDirectory,
|
||||
createRoomRepository,
|
||||
createShareRepository,
|
||||
type CoreRefs,
|
||||
type Tables,
|
||||
} from "./repositories.js";
|
||||
import { CollabRoomHub } from "./room-hub.js";
|
||||
import { createRoomActions, registerRoomRoutes } from "./room-routes.js";
|
||||
import { CollabRuntimeStore } from "./runtime-store.js";
|
||||
import {
|
||||
SESSION_GUESTS_KEY,
|
||||
createSessionGuests,
|
||||
createSessionSharingService,
|
||||
} from "./services.js";
|
||||
import { registerShareRoutes } from "./share-routes.js";
|
||||
import { roomMembers, rooms, shareParticipants, shares } from "./tables.js";
|
||||
|
||||
export async function activate(ctx: PluginContext) {
|
||||
const tables: Tables = {
|
||||
shares: await ctx.db.define(shares),
|
||||
participants: await ctx.db.define(shareParticipants),
|
||||
rooms: await ctx.db.define(rooms),
|
||||
members: await ctx.db.define(roomMembers),
|
||||
};
|
||||
const refs = () => ctx.db.refs<CoreRefs>();
|
||||
|
||||
const directory = createDirectory(ctx.db, refs);
|
||||
const store = new CollabRuntimeStore(ctx.log);
|
||||
const hub = new CollabRoomHub(store);
|
||||
const resolveLimiter = createRateLimiter({
|
||||
windowMs: 60_000,
|
||||
maxAttempts: 30,
|
||||
});
|
||||
const guestLimiter = createRateLimiter({ windowMs: 60_000, maxAttempts: 60 });
|
||||
ctx.disposables.add(() => {
|
||||
hub.dispose();
|
||||
resolveLimiter.dispose();
|
||||
guestLimiter.dispose();
|
||||
return store.close();
|
||||
});
|
||||
|
||||
const deps: SharingDeps = {
|
||||
ctx,
|
||||
shares: createShareRepository(ctx.db, tables),
|
||||
rooms: createRoomRepository(ctx.db, tables, refs),
|
||||
directory,
|
||||
live: createLiveSessions(ctx),
|
||||
hub,
|
||||
store,
|
||||
resolveLimiter,
|
||||
guestLimiter,
|
||||
isSharingEnabledForHost: (hostId) =>
|
||||
isSharingEnabledForHost(ctx, directory.hostExists, hostId),
|
||||
};
|
||||
|
||||
const actions = createRoomActions(deps);
|
||||
store.onEvent(actions.onRuntimeEvent);
|
||||
|
||||
// Guest links carry their own token, checked in the handlers.
|
||||
const router = ctx.http.router<Router>({
|
||||
public: ["/resolve/:linkToken", "/guest/:token"],
|
||||
});
|
||||
registerShareRoutes(router, deps);
|
||||
registerRoomRoutes(router, deps, actions);
|
||||
|
||||
ctx.services.provide("sessions.sharing", createSessionSharingService(deps));
|
||||
ctx.registry.provide(SESSION_GUESTS_KEY, createSessionGuests(deps));
|
||||
|
||||
ctx.log.info("Session sharing mounted at /plugin-api/session-sharing");
|
||||
}
|
||||
|
||||
export async function deactivate() {
|
||||
// Everything above was registered through ctx and is disposed by core.
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
import type { PluginContext } from "@termix/plugin-sdk/backend";
|
||||
import type { LiveProtocol } from "./repositories.js";
|
||||
|
||||
export const LIVE_SESSIONS_SERVICE = "sessions.live";
|
||||
|
||||
/** What every sessions.live provider reports about one session. */
|
||||
export interface LiveSessionInfo {
|
||||
id: string;
|
||||
userId: string;
|
||||
hostId: number;
|
||||
hostName: string;
|
||||
isConnected: boolean;
|
||||
createdAt: number;
|
||||
tabInstanceId: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* sessions.live v1, keyed by session type. ssh-terminal provides "ssh";
|
||||
* remote-desktop provides "rdp", "vnc" and "telnet". Through a service handle
|
||||
* every method answers with a promise. The control methods are the terminal's
|
||||
* (a remote desktop viewer cannot be kicked or handed control), and
|
||||
* createViewerToken is remote desktop's: a join token for a viewer.
|
||||
*/
|
||||
export interface LiveSessionProvider {
|
||||
getSession: (sessionId: string) => LiveSessionInfo | null;
|
||||
ownerEndSession?: (sessionId: string, reason: string) => void;
|
||||
disconnectParticipants?: (
|
||||
sessionId: string,
|
||||
shareId: string,
|
||||
options: { reason: string; userId?: string | null },
|
||||
) => number;
|
||||
setRoomShareControl?: (
|
||||
sessionId: string,
|
||||
shareId: string,
|
||||
controllerUserId: string | null,
|
||||
) => void;
|
||||
createViewerToken?: (sessionId: string, readOnly: boolean) => string;
|
||||
}
|
||||
|
||||
type Async<T> = {
|
||||
[K in keyof T]-?: T[K] extends ((...args: infer A) => infer R) | undefined
|
||||
? (...args: A) => Promise<R>
|
||||
: never;
|
||||
};
|
||||
|
||||
export type LiveSessions = ReturnType<typeof createLiveSessions>;
|
||||
|
||||
/**
|
||||
* Reaches live sessions through sessions.live as the acting user. A caller
|
||||
* with no actor (a public guest route) passes the share owner, who owns the
|
||||
* session being looked at.
|
||||
*/
|
||||
export function createLiveSessions(ctx: PluginContext) {
|
||||
async function run<T>(
|
||||
asUser: string | undefined,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
if (asUser && ctx.currentActor() !== asUser) {
|
||||
return ctx.asUser(asUser, fn);
|
||||
}
|
||||
return fn();
|
||||
}
|
||||
|
||||
function provider(protocol: string): Async<LiveSessionProvider> | null {
|
||||
const handle = ctx.services.get<Async<LiveSessionProvider>>(
|
||||
LIVE_SESSIONS_SERVICE,
|
||||
{ provider: protocol },
|
||||
);
|
||||
return "getSession" in handle ? handle : null;
|
||||
}
|
||||
|
||||
async function getSession(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
asUser?: string,
|
||||
): Promise<LiveSessionInfo | null> {
|
||||
const live = provider(protocol);
|
||||
if (!live) return null;
|
||||
try {
|
||||
return (await run(asUser, () => live.getSession(sessionId))) ?? null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
getSession,
|
||||
|
||||
async isLive(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
asUser?: string,
|
||||
): Promise<boolean> {
|
||||
const session = await getSession(protocol, sessionId, asUser);
|
||||
return !!session?.isConnected;
|
||||
},
|
||||
|
||||
async isOwnedBy(
|
||||
protocol: LiveProtocol,
|
||||
sessionId: string,
|
||||
userId: string,
|
||||
): Promise<boolean> {
|
||||
const session = await getSession(protocol, sessionId);
|
||||
return !!session && session.isConnected && session.userId === userId;
|
||||
},
|
||||
|
||||
/** Ends the session for every guest. Best effort, like before. */
|
||||
async ownerEndSession(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
reason: string,
|
||||
asUser?: string,
|
||||
): Promise<void> {
|
||||
const live = provider(protocol);
|
||||
if (!live || !("ownerEndSession" in live)) return;
|
||||
await run(asUser, () => live.ownerEndSession(sessionId, reason)).catch(
|
||||
() => {},
|
||||
);
|
||||
},
|
||||
|
||||
async disconnectParticipants(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
shareId: string,
|
||||
options: { reason: string; userId?: string | null },
|
||||
asUser?: string,
|
||||
): Promise<void> {
|
||||
const live = provider(protocol);
|
||||
if (!live || !("disconnectParticipants" in live)) return;
|
||||
await run(asUser, () =>
|
||||
live.disconnectParticipants(sessionId, shareId, options),
|
||||
).catch(() => {});
|
||||
},
|
||||
|
||||
async setRoomShareControl(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
shareId: string,
|
||||
controllerUserId: string | null,
|
||||
asUser?: string,
|
||||
): Promise<void> {
|
||||
const live = provider(protocol);
|
||||
if (!live || !("setRoomShareControl" in live)) return;
|
||||
await run(asUser, () =>
|
||||
live.setRoomShareControl(sessionId, shareId, controllerUserId),
|
||||
).catch(() => {});
|
||||
},
|
||||
|
||||
/** A viewer token for a remote desktop session. Throws while unavailable. */
|
||||
async createViewerToken(
|
||||
protocol: string,
|
||||
sessionId: string,
|
||||
readOnly: boolean,
|
||||
asUser?: string,
|
||||
): Promise<string> {
|
||||
const live = provider(protocol);
|
||||
if (!live || !("createViewerToken" in live)) {
|
||||
throw new Error(`No ${protocol} session provider is running`);
|
||||
}
|
||||
return run(asUser, () => live.createViewerToken(sessionId, readOnly));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether sessions on a host may be shared: the admin switch first, then the
|
||||
* host's own setting. Both default to on. A host that does not exist cannot
|
||||
* be shared.
|
||||
*/
|
||||
export async function isSharingEnabledForHost(
|
||||
ctx: PluginContext,
|
||||
hostExists: (hostId: number) => Promise<boolean>,
|
||||
hostId: number,
|
||||
): Promise<boolean> {
|
||||
if ((await ctx.settings.get<boolean>("globallyEnabled")) === false) {
|
||||
return false;
|
||||
}
|
||||
if (!(await hostExists(hostId))) return false;
|
||||
return (
|
||||
(await ctx.settings.getHost<boolean>(hostId, "allowSessionSharing")) !==
|
||||
false
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
/**
|
||||
* A per-IP fixed window, for the public guest endpoints. `dispose` stops the
|
||||
* sweep that forgets old windows.
|
||||
*/
|
||||
export function createRateLimiter(options: {
|
||||
windowMs: number;
|
||||
maxAttempts: number;
|
||||
}) {
|
||||
const attempts = new Map<string, { count: number; windowStart: number }>();
|
||||
const sweep = setInterval(() => {
|
||||
const now = Date.now();
|
||||
for (const [ip, entry] of attempts) {
|
||||
if (now - entry.windowStart > options.windowMs) attempts.delete(ip);
|
||||
}
|
||||
}, 5 * options.windowMs);
|
||||
sweep.unref();
|
||||
|
||||
return {
|
||||
/** Counts an attempt and says whether this IP is over the limit. */
|
||||
isLimited(ip: string): boolean {
|
||||
const now = Date.now();
|
||||
const entry = attempts.get(ip);
|
||||
if (!entry || now - entry.windowStart > options.windowMs) {
|
||||
attempts.set(ip, { count: 1, windowStart: now });
|
||||
return false;
|
||||
}
|
||||
entry.count += 1;
|
||||
return entry.count > options.maxAttempts;
|
||||
},
|
||||
dispose(): void {
|
||||
clearInterval(sweep);
|
||||
attempts.clear();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export type RateLimiter = ReturnType<typeof createRateLimiter>;
|
||||
@@ -0,0 +1,580 @@
|
||||
import { and, desc, eq, gt, inArray, isNull } from "drizzle-orm";
|
||||
import type { PluginDatabase } from "@termix/plugin-sdk/backend";
|
||||
|
||||
/* eslint-disable @typescript-eslint/no-explicit-any */
|
||||
// Tables come from ctx.db.define, which the SDK hands back untyped, and the
|
||||
// drizzle handle is the server's own. Typed at this module's edge instead.
|
||||
type Table = any;
|
||||
type Drizzle = any;
|
||||
/* eslint-enable @typescript-eslint/no-explicit-any */
|
||||
|
||||
export type ShareType = "link" | "user" | "room";
|
||||
export type PermissionLevel = "read-only" | "read-write";
|
||||
export type LiveProtocol = "ssh" | "rdp" | "vnc" | "telnet";
|
||||
|
||||
export interface ShareRecord {
|
||||
id: string;
|
||||
hostId: number;
|
||||
ownerUserId: string;
|
||||
protocol: string;
|
||||
sessionId: string;
|
||||
tabInstanceId: string | null;
|
||||
shareType: string;
|
||||
targetUserId: string | null;
|
||||
linkToken: string | null;
|
||||
permissionLevel: string;
|
||||
createdAt: string;
|
||||
expiresAt: string;
|
||||
revokedAt: string | null;
|
||||
lastJoinedAt: string | null;
|
||||
joinCount: number;
|
||||
}
|
||||
|
||||
export interface ShareCreateInput {
|
||||
id: string;
|
||||
hostId: number;
|
||||
ownerUserId: string;
|
||||
protocol: string;
|
||||
sessionId: string;
|
||||
tabInstanceId?: string | null;
|
||||
shareType: ShareType;
|
||||
targetUserId?: string | null;
|
||||
linkToken?: string | null;
|
||||
permissionLevel: PermissionLevel;
|
||||
expiresAt: string;
|
||||
}
|
||||
|
||||
export interface RoomRecord {
|
||||
id: string;
|
||||
name: string;
|
||||
ownerUserId: string;
|
||||
persistent: boolean;
|
||||
presenterUserId: string | null;
|
||||
stageProtocol: string | null;
|
||||
stageHostId: number | null;
|
||||
stageShareId: string | null;
|
||||
guestLinkToken: string | null;
|
||||
createdAt: string;
|
||||
endedAt: string | null;
|
||||
}
|
||||
|
||||
export interface RoomMemberRecord {
|
||||
id: number;
|
||||
roomId: string;
|
||||
userId: string;
|
||||
roomRole: string;
|
||||
addedBy: string | null;
|
||||
createdAt: string;
|
||||
}
|
||||
|
||||
export interface RoomMemberWithUser {
|
||||
userId: string;
|
||||
username: string;
|
||||
roomRole: string;
|
||||
createdAt: string;
|
||||
}
|
||||
|
||||
export interface RoomStage {
|
||||
presenterUserId: string | null;
|
||||
stageProtocol: string | null;
|
||||
stageHostId: number | null;
|
||||
stageShareId: string | null;
|
||||
}
|
||||
|
||||
export interface Tables {
|
||||
shares: Table;
|
||||
participants: Table;
|
||||
rooms: Table;
|
||||
members: Table;
|
||||
}
|
||||
|
||||
export interface CoreRefs {
|
||||
users: Table;
|
||||
hosts: Table;
|
||||
roles: Table;
|
||||
userRoles: Table;
|
||||
}
|
||||
|
||||
const nowIso = () => new Date().toISOString();
|
||||
|
||||
/** Not revoked and not expired. Expiry is compared as ISO text, as before. */
|
||||
function activeShare(shares: Table, now: string) {
|
||||
return and(isNull(shares.revokedAt), gt(shares.expiresAt, now));
|
||||
}
|
||||
|
||||
export type ShareRepository = ReturnType<typeof createShareRepository>;
|
||||
|
||||
export function createShareRepository(db: PluginDatabase, tables: Tables) {
|
||||
const client = () => db.client<Drizzle>();
|
||||
const { shares, participants } = tables;
|
||||
|
||||
async function findById(id: string): Promise<ShareRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(shares)
|
||||
.where(eq(shares.id, id))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
return {
|
||||
findById,
|
||||
|
||||
async create(input: ShareCreateInput): Promise<ShareRecord> {
|
||||
await (await client()).insert(shares).values({
|
||||
id: input.id,
|
||||
hostId: input.hostId,
|
||||
ownerUserId: input.ownerUserId,
|
||||
protocol: input.protocol,
|
||||
sessionId: input.sessionId,
|
||||
tabInstanceId: input.tabInstanceId ?? null,
|
||||
shareType: input.shareType,
|
||||
targetUserId: input.targetUserId ?? null,
|
||||
linkToken: input.linkToken ?? null,
|
||||
permissionLevel: input.permissionLevel,
|
||||
expiresAt: input.expiresAt,
|
||||
});
|
||||
await db.persist();
|
||||
return (await findById(input.id))!;
|
||||
},
|
||||
|
||||
async findActiveById(
|
||||
id: string,
|
||||
now = nowIso(),
|
||||
): Promise<ShareRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(shares)
|
||||
.where(and(eq(shares.id, id), activeShare(shares, now)))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
},
|
||||
|
||||
async findByLinkToken(
|
||||
linkToken: string,
|
||||
now = nowIso(),
|
||||
): Promise<ShareRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(shares)
|
||||
.where(and(eq(shares.linkToken, linkToken), activeShare(shares, now)))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
},
|
||||
|
||||
async findActiveSharesForHost(
|
||||
hostId: number,
|
||||
ownerUserId: string,
|
||||
now = nowIso(),
|
||||
): Promise<ShareRecord[]> {
|
||||
return (await client())
|
||||
.select()
|
||||
.from(shares)
|
||||
.where(
|
||||
and(
|
||||
eq(shares.hostId, hostId),
|
||||
eq(shares.ownerUserId, ownerUserId),
|
||||
activeShare(shares, now),
|
||||
),
|
||||
);
|
||||
},
|
||||
|
||||
/** Active user shares aimed at this user. */
|
||||
async findSharesTargetingUser(
|
||||
userId: string,
|
||||
now = nowIso(),
|
||||
): Promise<ShareRecord[]> {
|
||||
return (await client())
|
||||
.select()
|
||||
.from(shares)
|
||||
.where(
|
||||
and(
|
||||
eq(shares.shareType, "user"),
|
||||
eq(shares.targetUserId, userId),
|
||||
activeShare(shares, now),
|
||||
),
|
||||
);
|
||||
},
|
||||
|
||||
/** Revokes a share the given user owns. False when it is not theirs. */
|
||||
async revoke(shareId: string, ownerUserId: string): Promise<boolean> {
|
||||
const share = await findById(shareId);
|
||||
if (!share || share.ownerUserId !== ownerUserId) return false;
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(shares)
|
||||
.set({ revokedAt: nowIso() })
|
||||
.where(eq(shares.id, shareId));
|
||||
await db.persist();
|
||||
return true;
|
||||
},
|
||||
|
||||
async revokeAny(shareId: string): Promise<boolean> {
|
||||
if (!(await findById(shareId))) return false;
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(shares)
|
||||
.set({ revokedAt: nowIso() })
|
||||
.where(eq(shares.id, shareId));
|
||||
await db.persist();
|
||||
return true;
|
||||
},
|
||||
|
||||
async touchUsage(shareId: string): Promise<void> {
|
||||
const current = await findById(shareId);
|
||||
if (!current) return;
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(shares)
|
||||
.set({
|
||||
lastJoinedAt: nowIso(),
|
||||
joinCount: (current.joinCount ?? 0) + 1,
|
||||
})
|
||||
.where(eq(shares.id, shareId));
|
||||
await db.persist();
|
||||
},
|
||||
|
||||
async recordParticipantJoin(
|
||||
shareId: string,
|
||||
userId: string | null,
|
||||
guestLabel: string | null,
|
||||
): Promise<void> {
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.insert(participants)
|
||||
.values({ shareId, userId, guestLabel });
|
||||
await db.persist();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export type RoomRepository = ReturnType<typeof createRoomRepository>;
|
||||
|
||||
export function createRoomRepository(
|
||||
db: PluginDatabase,
|
||||
tables: Tables,
|
||||
refs: () => Promise<CoreRefs>,
|
||||
) {
|
||||
const client = () => db.client<Drizzle>();
|
||||
const { rooms, members } = tables;
|
||||
|
||||
async function findById(id: string): Promise<RoomRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(rooms)
|
||||
.where(eq(rooms.id, id))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
async function findMember(
|
||||
roomId: string,
|
||||
userId: string,
|
||||
): Promise<RoomMemberRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(members)
|
||||
.where(and(eq(members.roomId, roomId), eq(members.userId, userId)))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
async function updateStage(roomId: string, stage: RoomStage): Promise<void> {
|
||||
await (await client()).update(rooms).set(stage).where(eq(rooms.id, roomId));
|
||||
await db.persist();
|
||||
}
|
||||
|
||||
return {
|
||||
findById,
|
||||
findMember,
|
||||
updateStage,
|
||||
|
||||
async createRoom(input: {
|
||||
id: string;
|
||||
name: string;
|
||||
ownerUserId: string;
|
||||
persistent: boolean;
|
||||
}): Promise<RoomRecord> {
|
||||
await (await client()).insert(rooms).values(input);
|
||||
await db.persist();
|
||||
return (await findById(input.id))!;
|
||||
},
|
||||
|
||||
/** The live room whose stage points at this share, if any. */
|
||||
async findByStageShareId(shareId: string): Promise<RoomRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(rooms)
|
||||
.where(and(eq(rooms.stageShareId, shareId), isNull(rooms.endedAt)))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
},
|
||||
|
||||
async findByGuestToken(token: string): Promise<RoomRecord | null> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(rooms)
|
||||
.where(and(eq(rooms.guestLinkToken, token), isNull(rooms.endedAt)))
|
||||
.limit(1);
|
||||
return rows[0] ?? null;
|
||||
},
|
||||
|
||||
async setGuestToken(roomId: string, token: string | null): Promise<void> {
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(rooms)
|
||||
.set({ guestLinkToken: token })
|
||||
.where(eq(rooms.id, roomId));
|
||||
await db.persist();
|
||||
},
|
||||
|
||||
async listForUser(userId: string): Promise<RoomRecord[]> {
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ room: rooms })
|
||||
.from(members)
|
||||
.innerJoin(rooms, eq(members.roomId, rooms.id))
|
||||
.where(and(eq(members.userId, userId), isNull(rooms.endedAt)))
|
||||
.orderBy(desc(rooms.createdAt));
|
||||
return rows.map((row: { room: RoomRecord }) => row.room);
|
||||
},
|
||||
|
||||
async addMember(input: {
|
||||
roomId: string;
|
||||
userId: string;
|
||||
roomRole: "host" | "member";
|
||||
addedBy: string | null;
|
||||
}): Promise<boolean> {
|
||||
if (await findMember(input.roomId, input.userId)) return false;
|
||||
await (await client()).insert(members).values(input);
|
||||
await db.persist();
|
||||
return true;
|
||||
},
|
||||
|
||||
async removeMember(roomId: string, userId: string): Promise<void> {
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.delete(members)
|
||||
.where(and(eq(members.roomId, roomId), eq(members.userId, userId)));
|
||||
await db.persist();
|
||||
},
|
||||
|
||||
async listMembers(roomId: string): Promise<RoomMemberWithUser[]> {
|
||||
const rows: RoomMemberRecord[] = await (
|
||||
await client()
|
||||
)
|
||||
.select()
|
||||
.from(members)
|
||||
.where(eq(members.roomId, roomId));
|
||||
const names = await usernames(rows.map((row) => row.userId));
|
||||
return rows
|
||||
.filter((row) => names.has(row.userId))
|
||||
.map((row) => ({
|
||||
userId: row.userId,
|
||||
username: names.get(row.userId)!,
|
||||
roomRole: row.roomRole,
|
||||
createdAt: row.createdAt,
|
||||
}))
|
||||
.sort((a, b) => a.username.localeCompare(b.username));
|
||||
},
|
||||
|
||||
/**
|
||||
* Sets the stage only if it still points at `expectedShareId`, so two
|
||||
* members taking the stage at once cannot both win.
|
||||
*/
|
||||
async replaceStage(
|
||||
roomId: string,
|
||||
expectedShareId: string | null,
|
||||
stage: RoomStage,
|
||||
): Promise<boolean> {
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(rooms)
|
||||
.set(stage)
|
||||
.where(
|
||||
and(
|
||||
eq(rooms.id, roomId),
|
||||
expectedShareId
|
||||
? eq(rooms.stageShareId, expectedShareId)
|
||||
: isNull(rooms.stageShareId),
|
||||
),
|
||||
);
|
||||
const after = await findById(roomId);
|
||||
const changed = !!after && after.stageShareId === stage.stageShareId;
|
||||
if (changed) await db.persist();
|
||||
return changed;
|
||||
},
|
||||
|
||||
async clearStage(roomId: string): Promise<void> {
|
||||
await updateStage(roomId, {
|
||||
presenterUserId: null,
|
||||
stageProtocol: null,
|
||||
stageHostId: null,
|
||||
stageShareId: null,
|
||||
});
|
||||
},
|
||||
|
||||
async endRoom(roomId: string): Promise<void> {
|
||||
await (
|
||||
await client()
|
||||
)
|
||||
.update(rooms)
|
||||
.set({
|
||||
endedAt: nowIso(),
|
||||
presenterUserId: null,
|
||||
stageProtocol: null,
|
||||
stageHostId: null,
|
||||
stageShareId: null,
|
||||
})
|
||||
.where(eq(rooms.id, roomId));
|
||||
await db.persist();
|
||||
},
|
||||
|
||||
async deleteRoom(roomId: string): Promise<void> {
|
||||
const drizzle = await client();
|
||||
// Not left to the foreign key: SQLite only cascades with foreign_keys on.
|
||||
await drizzle.delete(members).where(eq(members.roomId, roomId));
|
||||
await drizzle.delete(rooms).where(eq(rooms.id, roomId));
|
||||
await db.persist();
|
||||
},
|
||||
};
|
||||
|
||||
async function usernames(ids: string[]): Promise<Map<string, string>> {
|
||||
if (ids.length === 0) return new Map();
|
||||
const { users } = await refs();
|
||||
const rows: Array<{ id: string; username: string }> = await (
|
||||
await client()
|
||||
)
|
||||
.select({ id: users.id, username: users.username })
|
||||
.from(users)
|
||||
.where(inArray(users.id, ids));
|
||||
return new Map(rows.map((row) => [row.id, row.username]));
|
||||
}
|
||||
}
|
||||
|
||||
export type Directory = ReturnType<typeof createDirectory>;
|
||||
|
||||
/** Users and roles, read through ctx.db.refs, for names and invites. */
|
||||
export function createDirectory(
|
||||
db: PluginDatabase,
|
||||
refs: () => Promise<CoreRefs>,
|
||||
) {
|
||||
const client = () => db.client<Drizzle>();
|
||||
|
||||
return {
|
||||
async username(userId: string): Promise<string> {
|
||||
const { users } = await refs();
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ username: users.username })
|
||||
.from(users)
|
||||
.where(eq(users.id, userId))
|
||||
.limit(1);
|
||||
return rows[0]?.username ?? userId;
|
||||
},
|
||||
|
||||
async userExists(userId: string): Promise<boolean> {
|
||||
const { users } = await refs();
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ id: users.id })
|
||||
.from(users)
|
||||
.where(eq(users.id, userId))
|
||||
.limit(1);
|
||||
return rows.length > 0;
|
||||
},
|
||||
|
||||
async isAdmin(userId: string): Promise<boolean> {
|
||||
const { users } = await refs();
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ isAdmin: users.isAdmin })
|
||||
.from(users)
|
||||
.where(eq(users.id, userId))
|
||||
.limit(1);
|
||||
return !!rows[0]?.isAdmin;
|
||||
},
|
||||
|
||||
async hostExists(hostId: number): Promise<boolean> {
|
||||
const { hosts } = await refs();
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ id: hosts.id })
|
||||
.from(hosts)
|
||||
.where(eq(hosts.id, hostId))
|
||||
.limit(1);
|
||||
return rows.length > 0;
|
||||
},
|
||||
|
||||
async roleExists(roleId: number): Promise<boolean> {
|
||||
const { roles } = await refs();
|
||||
const rows = await (
|
||||
await client()
|
||||
)
|
||||
.select({ id: roles.id })
|
||||
.from(roles)
|
||||
.where(eq(roles.id, roleId))
|
||||
.limit(1);
|
||||
return rows.length > 0;
|
||||
},
|
||||
|
||||
async roleUserIds(roleId: number): Promise<string[]> {
|
||||
const { userRoles } = await refs();
|
||||
const rows: Array<{ userId: string }> = await (
|
||||
await client()
|
||||
)
|
||||
.select({ userId: userRoles.userId })
|
||||
.from(userRoles)
|
||||
.where(eq(userRoles.roleId, roleId));
|
||||
return rows.map((row) => row.userId);
|
||||
},
|
||||
|
||||
async listUsers(): Promise<Array<{ id: string; username: string }>> {
|
||||
const { users } = await refs();
|
||||
return (await client())
|
||||
.select({ id: users.id, username: users.username })
|
||||
.from(users)
|
||||
.orderBy(users.username);
|
||||
},
|
||||
|
||||
async listRoles(): Promise<
|
||||
Array<{ id: number; name: string; displayName: string | null }>
|
||||
> {
|
||||
const { roles } = await refs();
|
||||
return (await client())
|
||||
.select({
|
||||
id: roles.id,
|
||||
name: roles.name,
|
||||
displayName: roles.displayName,
|
||||
})
|
||||
.from(roles)
|
||||
.orderBy(roles.name);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
import type { WebSocket } from "ws";
|
||||
import type { CollabRuntimeStore } from "./runtime-store.js";
|
||||
|
||||
const PRESENCE_HEARTBEAT_MS = 15_000;
|
||||
|
||||
export interface CollabRoomClient {
|
||||
ws: WebSocket;
|
||||
userId: string;
|
||||
username: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Local WebSocket fan-out plus the runtime store's pub/sub for room events and
|
||||
* presence across instances. The terminal socket feeds subscribe and
|
||||
* unsubscribe through the sessions.sharing service.
|
||||
*/
|
||||
export class CollabRoomHub {
|
||||
private rooms = new Map<string, Set<CollabRoomClient>>();
|
||||
private readonly heartbeat: NodeJS.Timeout;
|
||||
|
||||
constructor(private readonly store: CollabRuntimeStore) {
|
||||
store.onEvent((roomId, message) => {
|
||||
if (
|
||||
"type" in message &&
|
||||
typeof message.type === "string" &&
|
||||
message.type.startsWith("collab_internal_")
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.broadcastLocal(roomId, message);
|
||||
});
|
||||
this.heartbeat = setInterval(() => {
|
||||
for (const roomId of this.rooms.keys()) void this.refreshPresence(roomId);
|
||||
}, PRESENCE_HEARTBEAT_MS);
|
||||
this.heartbeat.unref();
|
||||
}
|
||||
|
||||
subscribe(roomId: string, client: CollabRoomClient): void {
|
||||
let clients = this.rooms.get(roomId);
|
||||
if (!clients) {
|
||||
clients = new Set();
|
||||
this.rooms.set(roomId, clients);
|
||||
}
|
||||
for (const existing of clients) {
|
||||
if (existing.ws === client.ws) return;
|
||||
}
|
||||
clients.add(client);
|
||||
void this.refreshPresence(roomId);
|
||||
}
|
||||
|
||||
/** Drops the socket from one room, or from every room when roomId is omitted. */
|
||||
unsubscribe(ws: WebSocket, roomId?: string): void {
|
||||
for (const [id, clients] of this.rooms) {
|
||||
if (roomId && id !== roomId) continue;
|
||||
let removed = false;
|
||||
for (const client of clients) {
|
||||
if (client.ws === ws) {
|
||||
clients.delete(client);
|
||||
removed = true;
|
||||
}
|
||||
}
|
||||
if (clients.size === 0) this.rooms.delete(id);
|
||||
if (removed) void this.refreshPresence(id);
|
||||
}
|
||||
}
|
||||
|
||||
broadcast(roomId: string, message: object): void {
|
||||
this.broadcastLocal(roomId, message);
|
||||
void this.store.publish(roomId, message);
|
||||
}
|
||||
|
||||
async onlineUsers(
|
||||
roomId: string,
|
||||
): Promise<Array<{ userId: string; username: string }>> {
|
||||
return this.store.onlineUsers(roomId, this.localOnlineUsers(roomId));
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
clearInterval(this.heartbeat);
|
||||
this.rooms.clear();
|
||||
}
|
||||
|
||||
private broadcastLocal(roomId: string, message: object): void {
|
||||
const clients = this.rooms.get(roomId);
|
||||
if (!clients) return;
|
||||
const payload = JSON.stringify(message);
|
||||
for (const client of clients) {
|
||||
if (client.ws.readyState !== client.ws.OPEN) continue;
|
||||
try {
|
||||
client.ws.send(payload);
|
||||
} catch {
|
||||
/* keep broadcasting to the rest */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private localOnlineUsers(
|
||||
roomId: string,
|
||||
): Array<{ userId: string; username: string }> {
|
||||
const clients = this.rooms.get(roomId);
|
||||
if (!clients) return [];
|
||||
const seen = new Map<string, string>();
|
||||
for (const client of clients) {
|
||||
seen.set(client.userId, client.username);
|
||||
}
|
||||
return Array.from(seen, ([userId, username]) => ({ userId, username }));
|
||||
}
|
||||
|
||||
private async refreshPresence(roomId: string): Promise<void> {
|
||||
const localUsers = this.localOnlineUsers(roomId);
|
||||
await this.store.updatePresence(roomId, localUsers);
|
||||
this.broadcast(roomId, {
|
||||
type: "collab_online",
|
||||
roomId,
|
||||
users: await this.onlineUsers(roomId),
|
||||
});
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,289 @@
|
||||
import crypto from "node:crypto";
|
||||
import { createClient } from "redis";
|
||||
import type { PluginLogger } from "@termix/plugin-sdk/backend";
|
||||
|
||||
export interface CollabControlRequest {
|
||||
userId: string;
|
||||
username: string;
|
||||
requestedAt: string;
|
||||
}
|
||||
|
||||
interface PresenceEntry {
|
||||
instanceId: string;
|
||||
userId: string;
|
||||
username: string;
|
||||
}
|
||||
|
||||
interface EventEnvelope {
|
||||
source: string;
|
||||
roomId: string;
|
||||
message: object;
|
||||
}
|
||||
|
||||
const KEY_PREFIX = process.env.TERMIX_REDIS_PREFIX?.trim() || "termix:collab";
|
||||
const EVENT_CHANNEL = `${KEY_PREFIX}:events`;
|
||||
const PRESENCE_TTL_MS = 45_000;
|
||||
const STATE_TTL_SECONDS = 12 * 60 * 60;
|
||||
const CONNECT_RETRY_MS = 15_000;
|
||||
|
||||
/**
|
||||
* Room state that has to be shared between Termix instances: who controls a
|
||||
* stage, pending control requests, presence and room events. Redis-backed when
|
||||
* REDIS_URL is set, with local state for a single instance.
|
||||
*/
|
||||
export class CollabRuntimeStore {
|
||||
constructor(private readonly log: PluginLogger) {}
|
||||
|
||||
private readonly instanceId = crypto.randomUUID();
|
||||
private publisher: ReturnType<typeof createClient> | null = null;
|
||||
private subscriber: ReturnType<typeof createClient> | null = null;
|
||||
private connecting: Promise<boolean> | null = null;
|
||||
private nextConnectAttempt = 0;
|
||||
private eventListeners = new Set<(roomId: string, message: object) => void>();
|
||||
private localControllers = new Map<string, string>();
|
||||
private localRequests = new Map<string, Map<string, CollabControlRequest>>();
|
||||
private presenceMembers = new Map<string, Set<string>>();
|
||||
|
||||
onEvent(listener: (roomId: string, message: object) => void): void {
|
||||
this.eventListeners.add(listener);
|
||||
void this.ensureConnected();
|
||||
}
|
||||
|
||||
async publish(roomId: string, message: object): Promise<void> {
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
await this.publisher
|
||||
.publish(
|
||||
EVENT_CHANNEL,
|
||||
JSON.stringify({ source: this.instanceId, roomId, message }),
|
||||
)
|
||||
.catch((error) => this.logRedisFailure("publish", error));
|
||||
}
|
||||
|
||||
async getController(roomId: string): Promise<string | null> {
|
||||
if (await this.ensureConnected()) {
|
||||
const value = await this.publisher
|
||||
?.get(`${KEY_PREFIX}:controller:${roomId}`)
|
||||
.catch((error) => {
|
||||
this.logRedisFailure("get_controller", error);
|
||||
return undefined;
|
||||
});
|
||||
if (value !== undefined) return value;
|
||||
}
|
||||
return this.localControllers.get(roomId) ?? null;
|
||||
}
|
||||
|
||||
async setController(roomId: string, userId: string | null): Promise<void> {
|
||||
if (userId) this.localControllers.set(roomId, userId);
|
||||
else this.localControllers.delete(roomId);
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
const key = `${KEY_PREFIX}:controller:${roomId}`;
|
||||
const operation = userId
|
||||
? this.publisher.set(key, userId, { EX: STATE_TTL_SECONDS })
|
||||
: this.publisher.del(key);
|
||||
await operation.catch((error) =>
|
||||
this.logRedisFailure("set_controller", error),
|
||||
);
|
||||
}
|
||||
|
||||
async listRequests(roomId: string): Promise<CollabControlRequest[]> {
|
||||
if (await this.ensureConnected()) {
|
||||
const values = await this.publisher
|
||||
?.hVals(`${KEY_PREFIX}:requests:${roomId}`)
|
||||
.catch((error) => {
|
||||
this.logRedisFailure("list_requests", error);
|
||||
return null;
|
||||
});
|
||||
if (values) {
|
||||
return values
|
||||
.flatMap((value) => this.parse<CollabControlRequest>(value) ?? [])
|
||||
.sort((a, b) => a.requestedAt.localeCompare(b.requestedAt));
|
||||
}
|
||||
}
|
||||
return Array.from(this.localRequests.get(roomId)?.values() ?? []).sort(
|
||||
(a, b) => a.requestedAt.localeCompare(b.requestedAt),
|
||||
);
|
||||
}
|
||||
|
||||
async upsertRequest(
|
||||
roomId: string,
|
||||
request: CollabControlRequest,
|
||||
): Promise<void> {
|
||||
let requests = this.localRequests.get(roomId);
|
||||
if (!requests) {
|
||||
requests = new Map();
|
||||
this.localRequests.set(roomId, requests);
|
||||
}
|
||||
requests.set(request.userId, request);
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
const key = `${KEY_PREFIX}:requests:${roomId}`;
|
||||
await this.publisher
|
||||
.multi()
|
||||
.hSet(key, request.userId, JSON.stringify(request))
|
||||
.expire(key, STATE_TTL_SECONDS)
|
||||
.exec()
|
||||
.catch((error) => this.logRedisFailure("upsert_request", error));
|
||||
}
|
||||
|
||||
async removeRequest(roomId: string, userId: string): Promise<void> {
|
||||
const requests = this.localRequests.get(roomId);
|
||||
requests?.delete(userId);
|
||||
if (requests?.size === 0) this.localRequests.delete(roomId);
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
await this.publisher
|
||||
.hDel(`${KEY_PREFIX}:requests:${roomId}`, userId)
|
||||
.catch((error) => this.logRedisFailure("remove_request", error));
|
||||
}
|
||||
|
||||
async clearRequests(roomId: string): Promise<void> {
|
||||
this.localRequests.delete(roomId);
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
await this.publisher
|
||||
.del(`${KEY_PREFIX}:requests:${roomId}`)
|
||||
.catch((error) => this.logRedisFailure("clear_requests", error));
|
||||
}
|
||||
|
||||
async updatePresence(
|
||||
roomId: string,
|
||||
users: Array<{ userId: string; username: string }>,
|
||||
): Promise<void> {
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return;
|
||||
const key = `${KEY_PREFIX}:presence:${roomId}`;
|
||||
const previous = this.presenceMembers.get(roomId) ?? new Set<string>();
|
||||
const expiresAt = Date.now() + PRESENCE_TTL_MS;
|
||||
const current = new Set(
|
||||
users.map((user) =>
|
||||
JSON.stringify({
|
||||
instanceId: this.instanceId,
|
||||
...user,
|
||||
} satisfies PresenceEntry),
|
||||
),
|
||||
);
|
||||
const transaction = this.publisher.multi();
|
||||
if (previous.size > 0) transaction.zRem(key, Array.from(previous));
|
||||
for (const member of current) {
|
||||
transaction.zAdd(key, { score: expiresAt, value: member });
|
||||
}
|
||||
transaction.expire(key, Math.ceil(PRESENCE_TTL_MS / 1000) * 2);
|
||||
await transaction
|
||||
.exec()
|
||||
.then(() => {
|
||||
if (current.size > 0) this.presenceMembers.set(roomId, current);
|
||||
else this.presenceMembers.delete(roomId);
|
||||
})
|
||||
.catch((error) => this.logRedisFailure("update_presence", error));
|
||||
}
|
||||
|
||||
async onlineUsers(
|
||||
roomId: string,
|
||||
localUsers: Array<{ userId: string; username: string }>,
|
||||
): Promise<Array<{ userId: string; username: string }>> {
|
||||
if (!(await this.ensureConnected()) || !this.publisher) return localUsers;
|
||||
const key = `${KEY_PREFIX}:presence:${roomId}`;
|
||||
const now = Date.now();
|
||||
const transaction = this.publisher
|
||||
.multi()
|
||||
.zRemRangeByScore(key, 0, now)
|
||||
.zRangeByScore(key, now + 1, "+inf");
|
||||
const result = await transaction.exec().catch((error) => {
|
||||
this.logRedisFailure("online_users", error);
|
||||
return null;
|
||||
});
|
||||
const members = result?.[1] as string[] | undefined;
|
||||
if (!members) return localUsers;
|
||||
const seen = new Map<string, string>();
|
||||
for (const member of members) {
|
||||
const entry = this.parse<PresenceEntry>(member);
|
||||
if (entry) seen.set(entry.userId, entry.username);
|
||||
}
|
||||
return Array.from(seen, ([userId, username]) => ({ userId, username }));
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.eventListeners.clear();
|
||||
this.nextConnectAttempt = Number.POSITIVE_INFINITY;
|
||||
await Promise.allSettled([
|
||||
this.publisher?.isOpen ? this.publisher.quit() : Promise.resolve(),
|
||||
this.subscriber?.isOpen ? this.subscriber.quit() : Promise.resolve(),
|
||||
]);
|
||||
this.publisher = null;
|
||||
this.subscriber = null;
|
||||
}
|
||||
|
||||
private async ensureConnected(): Promise<boolean> {
|
||||
const url = process.env.REDIS_URL?.trim();
|
||||
if (!url) return false;
|
||||
if (this.publisher?.isReady && this.subscriber?.isReady) return true;
|
||||
if (Date.now() < this.nextConnectAttempt) return false;
|
||||
if (this.connecting) return this.connecting;
|
||||
this.connecting = this.connect(url).finally(() => {
|
||||
this.connecting = null;
|
||||
});
|
||||
return this.connecting;
|
||||
}
|
||||
|
||||
private async connect(url: string): Promise<boolean> {
|
||||
try {
|
||||
this.publisher = createClient({
|
||||
url,
|
||||
socket: { connectTimeout: 1500, reconnectStrategy: false },
|
||||
});
|
||||
this.subscriber = this.publisher.duplicate();
|
||||
this.publisher.on("error", (error) =>
|
||||
this.logRedisFailure("client", error),
|
||||
);
|
||||
this.subscriber.on("error", (error) =>
|
||||
this.logRedisFailure("subscriber", error),
|
||||
);
|
||||
await Promise.all([this.publisher.connect(), this.subscriber.connect()]);
|
||||
await this.subscriber.subscribe(EVENT_CHANNEL, (raw) => {
|
||||
const envelope = this.parse<EventEnvelope>(raw);
|
||||
if (
|
||||
!envelope ||
|
||||
typeof envelope.source !== "string" ||
|
||||
typeof envelope.roomId !== "string" ||
|
||||
!envelope.message ||
|
||||
typeof envelope.message !== "object" ||
|
||||
envelope.source === this.instanceId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
for (const listener of this.eventListeners) {
|
||||
try {
|
||||
listener(envelope.roomId, envelope.message);
|
||||
} catch (error) {
|
||||
this.logRedisFailure("event_listener", error);
|
||||
}
|
||||
}
|
||||
});
|
||||
this.log.info("Collaboration Redis runtime connected");
|
||||
this.nextConnectAttempt = 0;
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.nextConnectAttempt = Date.now() + CONNECT_RETRY_MS;
|
||||
this.logRedisFailure("connect", error);
|
||||
await Promise.allSettled([
|
||||
this.publisher?.disconnect(),
|
||||
this.subscriber?.disconnect(),
|
||||
]);
|
||||
this.publisher = null;
|
||||
this.subscriber = null;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private parse<T>(value: string): T | null {
|
||||
try {
|
||||
return JSON.parse(value) as T;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private logRedisFailure(operation: string, error: unknown): void {
|
||||
this.log.warn(
|
||||
`Collaboration Redis runtime unavailable (${operation}), using local state: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
import type { WebSocket } from "ws";
|
||||
import type { SharingDeps } from "./deps.js";
|
||||
import type { ShareRecord } from "./repositories.js";
|
||||
|
||||
/**
|
||||
* What the terminal consumes. The shapes match
|
||||
* plugins/ssh-terminal/src/backend/services.ts, which is the consumer's copy.
|
||||
*/
|
||||
export interface SharedSessionRef {
|
||||
id: string;
|
||||
sessionId: string;
|
||||
permissionLevel: "read-write" | "read-only";
|
||||
}
|
||||
|
||||
/** sessions.sharing v1: signed-in joins. Every call runs as the acting user. */
|
||||
export interface SessionSharingV1 {
|
||||
authorizeJoin: (
|
||||
shareId: string,
|
||||
) => Promise<{ share: SharedSessionRef; displayName: string } | null>;
|
||||
recordJoin: (shareId: string) => Promise<void>;
|
||||
subscribeRoom: (roomId: string, socket: unknown) => Promise<boolean>;
|
||||
unsubscribeRoom: (socket: unknown, roomId?: string) => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Guest joins through a share link or a room guest link, on ctx.registry as
|
||||
* SESSION_GUESTS_KEY. A guest has no user, so a service's per-call
|
||||
* permission check cannot apply: the token is the authority.
|
||||
*/
|
||||
export interface SessionGuestsV1 {
|
||||
resolve: (request: {
|
||||
shareToken?: string;
|
||||
roomGuestToken?: string;
|
||||
clientIp: string;
|
||||
}) => Promise<
|
||||
{ ok: true; share: SharedSessionRef } | { ok: false; reason: string }
|
||||
>;
|
||||
recordJoin: (shareId: string) => Promise<void>;
|
||||
}
|
||||
|
||||
export const SESSION_GUESTS_KEY = "sessions.sharing.guests";
|
||||
|
||||
function toRef(share: ShareRecord): SharedSessionRef {
|
||||
return {
|
||||
id: share.id,
|
||||
sessionId: share.sessionId,
|
||||
permissionLevel:
|
||||
share.permissionLevel === "read-write" ? "read-write" : "read-only",
|
||||
};
|
||||
}
|
||||
|
||||
export function createSessionSharingService(
|
||||
deps: SharingDeps,
|
||||
): SessionSharingV1 {
|
||||
const { ctx, shares, rooms, directory, hub } = deps;
|
||||
|
||||
return {
|
||||
async authorizeJoin(shareId) {
|
||||
const userId = ctx.currentActor();
|
||||
if (!userId) return null;
|
||||
const share = await shares.findActiveById(shareId);
|
||||
if (!share || share.protocol !== "ssh") return null;
|
||||
|
||||
if (share.shareType === "user") {
|
||||
// A user share needs its target, who can still reach the host.
|
||||
if (share.targetUserId !== userId) return null;
|
||||
const access = await ctx.hosts.checkAccess(share.hostId, "connect");
|
||||
if (!access.hasAccess) return null;
|
||||
} else if (share.shareType === "room") {
|
||||
// Room membership is the authorization for a room stage: it is
|
||||
// read-only and never exposes host credentials or config.
|
||||
const room = await rooms.findByStageShareId(share.id);
|
||||
if (!room || !(await rooms.findMember(room.id, userId))) return null;
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
|
||||
return {
|
||||
share: toRef(share),
|
||||
displayName: await directory.username(userId),
|
||||
};
|
||||
},
|
||||
|
||||
async recordJoin(shareId) {
|
||||
const userId = ctx.currentActor() ?? null;
|
||||
await shares.touchUsage(shareId);
|
||||
await shares.recordParticipantJoin(shareId, userId, null);
|
||||
},
|
||||
|
||||
async subscribeRoom(roomId, socket) {
|
||||
const userId = ctx.currentActor();
|
||||
if (!userId) return false;
|
||||
const room = await rooms.findById(roomId);
|
||||
if (!room || room.endedAt) return false;
|
||||
if (!(await rooms.findMember(roomId, userId))) return false;
|
||||
hub.subscribe(roomId, {
|
||||
ws: socket as WebSocket,
|
||||
userId,
|
||||
username: await directory.username(userId),
|
||||
});
|
||||
return true;
|
||||
},
|
||||
|
||||
async unsubscribeRoom(socket, roomId) {
|
||||
hub.unsubscribe(socket as WebSocket, roomId);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function createSessionGuests(deps: SharingDeps): SessionGuestsV1 {
|
||||
const { shares, rooms, guestLimiter } = deps;
|
||||
|
||||
/** Sharing still allowed, and an SSH share: the only kind the socket joins. */
|
||||
async function joinable(
|
||||
share: ShareRecord,
|
||||
): Promise<
|
||||
{ ok: true; share: SharedSessionRef } | { ok: false; reason: string }
|
||||
> {
|
||||
if (share.protocol !== "ssh") {
|
||||
return { ok: false, reason: "Unsupported share protocol" };
|
||||
}
|
||||
if (!(await deps.isSharingEnabledForHost(share.hostId))) {
|
||||
return { ok: false, reason: "Session sharing is disabled for this host" };
|
||||
}
|
||||
return { ok: true, share: toRef(share) };
|
||||
}
|
||||
|
||||
return {
|
||||
async resolve({ shareToken, roomGuestToken, clientIp }) {
|
||||
if (shareToken) {
|
||||
const share = await shares.findByLinkToken(shareToken);
|
||||
if (!share || share.shareType !== "link") {
|
||||
return { ok: false, reason: "Invalid or expired share link" };
|
||||
}
|
||||
return joinable(share);
|
||||
}
|
||||
|
||||
if (roomGuestToken) {
|
||||
if (guestLimiter.isLimited(clientIp)) {
|
||||
return { ok: false, reason: "Too many requests" };
|
||||
}
|
||||
const room = await rooms.findByGuestToken(roomGuestToken);
|
||||
const share = room?.stageShareId
|
||||
? await shares.findActiveById(room.stageShareId)
|
||||
: null;
|
||||
if (!share) return { ok: false, reason: "Nothing is being presented" };
|
||||
return joinable(share);
|
||||
}
|
||||
|
||||
return { ok: false, reason: "Authentication required" };
|
||||
},
|
||||
|
||||
async recordJoin(shareId) {
|
||||
await shares.touchUsage(shareId);
|
||||
await shares.recordParticipantJoin(shareId, null, "Guest");
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,523 @@
|
||||
import crypto from "node:crypto";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import {
|
||||
DISPLAY_WS_PATH,
|
||||
PROTOCOLS,
|
||||
TERMINAL_WS_PATH,
|
||||
actorOf,
|
||||
clientIp,
|
||||
type SharingDeps,
|
||||
} from "./deps.js";
|
||||
import type { LiveProtocol, PermissionLevel } from "./repositories.js";
|
||||
|
||||
const DEFAULT_EXPIRY_HOURS = 24;
|
||||
const MAX_EXPIRY_HOURS = 24 * 30;
|
||||
|
||||
function computeExpiresAt(expiryHours: number | undefined): string {
|
||||
const hours = Math.min(
|
||||
Math.max(expiryHours ?? DEFAULT_EXPIRY_HOURS, 1),
|
||||
MAX_EXPIRY_HOURS,
|
||||
);
|
||||
return new Date(Date.now() + hours * 60 * 60 * 1000).toISOString();
|
||||
}
|
||||
|
||||
/**
|
||||
* The share routes. `/resolve/:linkToken` is public (an anonymous guest link);
|
||||
* the rest run behind core auth and the plugin's `use` permission, which the
|
||||
* caller applies to the router first.
|
||||
*/
|
||||
export function registerShareRoutes(router: Router, deps: SharingDeps): void {
|
||||
const { ctx, shares, directory, live, resolveLimiter } = deps;
|
||||
const requireUse = ctx.rbac.require("use") as never;
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/resolve/{linkToken}:
|
||||
* get:
|
||||
* summary: Resolve a guest share link
|
||||
* description: Public, unauthenticated endpoint for anonymous share-link guests. Never returns host name, IP, username, or hostId. Rate-limited per IP.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* parameters:
|
||||
* - in: path
|
||||
* name: linkToken
|
||||
* required: true
|
||||
* schema:
|
||||
* type: string
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Resolved share connection info
|
||||
* 404:
|
||||
* description: Link not found, expired, revoked, or sharing disabled
|
||||
* 429:
|
||||
* description: Too many requests
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.get("/resolve/:linkToken", async (req: Request, res: Response) => {
|
||||
try {
|
||||
if (resolveLimiter.isLimited(clientIp(req))) {
|
||||
return res.status(429).json({ error: "Too many requests" });
|
||||
}
|
||||
|
||||
const linkToken = String(req.params.linkToken);
|
||||
const share = await shares.findByLinkToken(linkToken);
|
||||
if (!share || share.shareType !== "link") {
|
||||
return res.status(404).json({ error: "Link not found or expired" });
|
||||
}
|
||||
if (!(await deps.isSharingEnabledForHost(share.hostId))) {
|
||||
return res.status(404).json({ error: "Link not found or expired" });
|
||||
}
|
||||
|
||||
const protocol = share.protocol as LiveProtocol;
|
||||
if (!(await live.isLive(protocol, share.sessionId, share.ownerUserId))) {
|
||||
return res.status(404).json({ error: "Session is no longer active" });
|
||||
}
|
||||
|
||||
// Field by field on purpose: anonymous guests never see the host's
|
||||
// name, address, username or id.
|
||||
const response: {
|
||||
protocol: LiveProtocol;
|
||||
permissionLevel: PermissionLevel;
|
||||
wsPath: string;
|
||||
connectParams?: Record<string, string>;
|
||||
} = {
|
||||
protocol,
|
||||
permissionLevel: share.permissionLevel as PermissionLevel,
|
||||
// Advisory only: the client builds its own socket URL per deployment.
|
||||
wsPath:
|
||||
protocol === "ssh"
|
||||
? `${TERMINAL_WS_PATH}?shareToken=${encodeURIComponent(linkToken)}`
|
||||
: DISPLAY_WS_PATH,
|
||||
};
|
||||
|
||||
if (protocol !== "ssh") {
|
||||
response.connectParams = {
|
||||
token: await live.createViewerToken(
|
||||
protocol,
|
||||
share.sessionId,
|
||||
share.permissionLevel === "read-only",
|
||||
share.ownerUserId,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
await shares.touchUsage(share.id);
|
||||
await shares.recordParticipantJoin(share.id, null, "Guest");
|
||||
} catch {
|
||||
// Bookkeeping never fails the resolve.
|
||||
}
|
||||
|
||||
res.json(response);
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to resolve session share link",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to resolve share link" });
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/create:
|
||||
* post:
|
||||
* summary: Create a session share (link or targeted user)
|
||||
* description: Mints a share grant for a live terminal/RDP/VNC/Telnet session. Caller must own the live session.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* requestBody:
|
||||
* required: true
|
||||
* content:
|
||||
* application/json:
|
||||
* schema:
|
||||
* type: object
|
||||
* required:
|
||||
* - hostId
|
||||
* - sessionId
|
||||
* - protocol
|
||||
* - shareType
|
||||
* - permissionLevel
|
||||
* properties:
|
||||
* hostId:
|
||||
* type: integer
|
||||
* sessionId:
|
||||
* type: string
|
||||
* tabInstanceId:
|
||||
* type: string
|
||||
* protocol:
|
||||
* type: string
|
||||
* enum: [ssh, rdp, vnc, telnet]
|
||||
* shareType:
|
||||
* type: string
|
||||
* enum: [link, user]
|
||||
* targetUserId:
|
||||
* type: string
|
||||
* permissionLevel:
|
||||
* type: string
|
||||
* enum: [read-only, read-write]
|
||||
* expiryHours:
|
||||
* type: number
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Share created
|
||||
* 400:
|
||||
* description: Invalid request
|
||||
* 403:
|
||||
* description: Sharing disabled, or caller does not own the session
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.post("/create", requireUse, async (req: Request, res: Response) => {
|
||||
try {
|
||||
const userId = actorOf(ctx);
|
||||
const {
|
||||
hostId,
|
||||
sessionId,
|
||||
tabInstanceId,
|
||||
protocol,
|
||||
shareType,
|
||||
targetUserId,
|
||||
permissionLevel,
|
||||
expiryHours,
|
||||
} = req.body ?? {};
|
||||
|
||||
if (
|
||||
!hostId ||
|
||||
!sessionId ||
|
||||
!protocol ||
|
||||
!shareType ||
|
||||
!permissionLevel
|
||||
) {
|
||||
return res.status(400).json({ error: "Missing required fields" });
|
||||
}
|
||||
if (!PROTOCOLS.includes(protocol)) {
|
||||
return res.status(400).json({ error: "Invalid protocol" });
|
||||
}
|
||||
if (!["link", "user"].includes(shareType)) {
|
||||
return res.status(400).json({ error: "Invalid shareType" });
|
||||
}
|
||||
if (!["read-only", "read-write"].includes(permissionLevel)) {
|
||||
return res.status(400).json({ error: "Invalid permissionLevel" });
|
||||
}
|
||||
if (shareType === "user" && !targetUserId) {
|
||||
return res
|
||||
.status(400)
|
||||
.json({ error: "targetUserId is required for user shares" });
|
||||
}
|
||||
|
||||
const numericHostId = Number(hostId);
|
||||
if (!(await deps.isSharingEnabledForHost(numericHostId))) {
|
||||
return res
|
||||
.status(403)
|
||||
.json({ error: "Session sharing is disabled for this host" });
|
||||
}
|
||||
if (!(await live.isOwnedBy(protocol, String(sessionId), userId))) {
|
||||
return res
|
||||
.status(403)
|
||||
.json({ error: "You do not own this live session" });
|
||||
}
|
||||
|
||||
if (shareType === "user") {
|
||||
const target = String(targetUserId);
|
||||
const access = (await directory.userExists(target))
|
||||
? await ctx.asUser(target, () =>
|
||||
ctx.hosts.checkAccess(numericHostId, "connect"),
|
||||
)
|
||||
: { hasAccess: false };
|
||||
if (!access.hasAccess) {
|
||||
return res.status(403).json({
|
||||
error: "Target user does not have access to this host",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const created = await shares.create({
|
||||
id: crypto.randomUUID(),
|
||||
hostId: numericHostId,
|
||||
ownerUserId: userId,
|
||||
protocol,
|
||||
sessionId: String(sessionId),
|
||||
tabInstanceId: tabInstanceId ?? null,
|
||||
shareType,
|
||||
targetUserId: shareType === "user" ? String(targetUserId) : null,
|
||||
linkToken:
|
||||
shareType === "link"
|
||||
? crypto.randomBytes(24).toString("base64url")
|
||||
: null,
|
||||
permissionLevel,
|
||||
expiresAt: computeExpiresAt(expiryHours),
|
||||
});
|
||||
|
||||
res.json({
|
||||
shareId: created.id,
|
||||
linkToken: created.linkToken,
|
||||
expiresAt: created.expiresAt,
|
||||
});
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to create session share",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to create session share" });
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/host/{hostId}/active:
|
||||
* get:
|
||||
* summary: List active session shares for a host
|
||||
* description: Returns active (non-revoked, non-expired) shares owned by the caller for the given host.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* parameters:
|
||||
* - in: path
|
||||
* name: hostId
|
||||
* required: true
|
||||
* schema:
|
||||
* type: integer
|
||||
* responses:
|
||||
* 200:
|
||||
* description: List of active shares
|
||||
* 400:
|
||||
* description: Invalid host id
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.get(
|
||||
"/host/:hostId/active",
|
||||
requireUse,
|
||||
async (req: Request, res: Response) => {
|
||||
try {
|
||||
const userId = actorOf(ctx);
|
||||
const hostId = Number.parseInt(String(req.params.hostId), 10);
|
||||
if (!hostId || Number.isNaN(hostId)) {
|
||||
return res.status(400).json({ error: "Invalid host ID" });
|
||||
}
|
||||
res.json({
|
||||
shares: await shares.findActiveSharesForHost(hostId, userId),
|
||||
});
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to list session shares",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to list session shares" });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/shared-with-me:
|
||||
* get:
|
||||
* summary: List live SSH sessions other users shared with the caller
|
||||
* description: Active user shares aimed at the caller whose session is still live, in the shape of /open-tabs/active-sessions.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Shared sessions
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.get(
|
||||
"/shared-with-me",
|
||||
requireUse,
|
||||
async (_req: Request, res: Response) => {
|
||||
try {
|
||||
const userId = actorOf(ctx);
|
||||
const result = [];
|
||||
for (const share of await shares.findSharesTargetingUser(userId)) {
|
||||
if (share.protocol !== "ssh") continue;
|
||||
const session = await live.getSession("ssh", share.sessionId);
|
||||
if (!session?.isConnected) continue;
|
||||
result.push({
|
||||
sessionId: session.id,
|
||||
hostId: session.hostId,
|
||||
hostName: session.hostName,
|
||||
tabInstanceId: session.tabInstanceId,
|
||||
isConnected: session.isConnected,
|
||||
createdAt: session.createdAt,
|
||||
isOwnSession: false,
|
||||
sharedByUsername: await directory.username(share.ownerUserId),
|
||||
permissionLevel: share.permissionLevel,
|
||||
shareId: share.id,
|
||||
});
|
||||
}
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to list shared sessions",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to list shared sessions" });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/directory:
|
||||
* get:
|
||||
* summary: Users and roles to share with or invite
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Users (id, username) and roles (id, name, displayName)
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.get("/directory", requireUse, async (_req: Request, res: Response) => {
|
||||
try {
|
||||
res.json({
|
||||
users: await directory.listUsers(),
|
||||
roles: await directory.listRoles(),
|
||||
});
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to list users and roles",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to list users and roles" });
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/{shareId}:
|
||||
* delete:
|
||||
* summary: Revoke a session share
|
||||
* description: Revokes a share. Owner or admin only. Best-effort kick of live SSH participants; remote desktop viewers are not force-disconnected.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* parameters:
|
||||
* - in: path
|
||||
* name: shareId
|
||||
* required: true
|
||||
* schema:
|
||||
* type: string
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Share revoked
|
||||
* 403:
|
||||
* description: Not authorized to revoke this share
|
||||
* 404:
|
||||
* description: Share not found
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.delete(
|
||||
"/:shareId",
|
||||
requireUse,
|
||||
async (req: Request, res: Response) => {
|
||||
try {
|
||||
const userId = actorOf(ctx);
|
||||
const shareId = String(req.params.shareId);
|
||||
const share = await shares.findById(shareId);
|
||||
if (!share) {
|
||||
return res.status(404).json({ error: "Share not found" });
|
||||
}
|
||||
|
||||
let revoked = await shares.revoke(shareId, userId);
|
||||
if (!revoked && (await directory.isAdmin(userId))) {
|
||||
revoked = await shares.revokeAny(shareId);
|
||||
}
|
||||
if (!revoked) {
|
||||
return res
|
||||
.status(403)
|
||||
.json({ error: "Not authorized to revoke this share" });
|
||||
}
|
||||
|
||||
// guacamole-lite has no kick, so a revoked remote desktop link only
|
||||
// blocks new resolves until the viewer's own socket ends.
|
||||
if (share.protocol === "ssh") {
|
||||
await live.ownerEndSession(
|
||||
"ssh",
|
||||
share.sessionId,
|
||||
"Session share revoked by owner",
|
||||
share.ownerUserId,
|
||||
);
|
||||
}
|
||||
|
||||
res.json({ success: true });
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to revoke session share",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to revoke session share" });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
/**
|
||||
* @openapi
|
||||
* /plugin-api/session-sharing/{shareId}/end:
|
||||
* post:
|
||||
* summary: End a shared session for all participants
|
||||
* description: Owner-only. Terminates the underlying session and notifies joined participants. Remote desktop viewers are not force-disconnected.
|
||||
* tags:
|
||||
* - Session Sharing
|
||||
* security:
|
||||
* - bearerAuth: []
|
||||
* parameters:
|
||||
* - in: path
|
||||
* name: shareId
|
||||
* required: true
|
||||
* schema:
|
||||
* type: string
|
||||
* responses:
|
||||
* 200:
|
||||
* description: Session ended
|
||||
* 403:
|
||||
* description: Not the owner of this share
|
||||
* 404:
|
||||
* description: Share not found
|
||||
* 500:
|
||||
* description: Server error
|
||||
*/
|
||||
router.post(
|
||||
"/:shareId/end",
|
||||
requireUse,
|
||||
async (req: Request, res: Response) => {
|
||||
try {
|
||||
const userId = actorOf(ctx);
|
||||
const share = await shares.findById(String(req.params.shareId));
|
||||
if (!share) {
|
||||
return res.status(404).json({ error: "Share not found" });
|
||||
}
|
||||
if (share.ownerUserId !== userId) {
|
||||
return res.status(403).json({ error: "Not the owner of this share" });
|
||||
}
|
||||
if (share.protocol === "ssh") {
|
||||
await live.ownerEndSession(
|
||||
"ssh",
|
||||
share.sessionId,
|
||||
"Session ended by owner",
|
||||
);
|
||||
}
|
||||
res.json({ success: true });
|
||||
} catch (error) {
|
||||
ctx.log.error(
|
||||
"Failed to end shared session",
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
res.status(500).json({ error: "Failed to end shared session" });
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
import {
|
||||
adoptLegacyTable,
|
||||
boolean,
|
||||
defineTable,
|
||||
id,
|
||||
integer,
|
||||
refHost,
|
||||
refUser,
|
||||
text,
|
||||
timestamp,
|
||||
varchar,
|
||||
} from "@termix/plugin-sdk/db";
|
||||
|
||||
/*
|
||||
* Adopted from core's session_shares, session_share_participants,
|
||||
* collab_rooms and collab_room_members, so column and index names are the
|
||||
* legacy ones and the rename carries every row across.
|
||||
*
|
||||
* Columns that point at a user or host but may be null or set null on delete,
|
||||
* and the links between these four tables, are plain columns here: the SDK
|
||||
* only references users and ssh_data, cascading. The adoption migrations keep
|
||||
* the legacy FOREIGN KEY clauses by hand, as fleets does.
|
||||
*/
|
||||
|
||||
/**
|
||||
* A grant to join one live session: a link for anonymous guests, a user, or
|
||||
* a collab room's stage ("room").
|
||||
*/
|
||||
export const shares = adoptLegacyTable(
|
||||
"session_shares",
|
||||
defineTable(
|
||||
"shares",
|
||||
{
|
||||
id: varchar().primaryKey(),
|
||||
hostId: refHost(),
|
||||
ownerUserId: refUser(),
|
||||
protocol: text().notNull(),
|
||||
// The terminal's session id for SSH, guacd's connection id otherwise.
|
||||
// Neither is a row, so there is no foreign key.
|
||||
sessionId: varchar().notNull(),
|
||||
tabInstanceId: text(),
|
||||
shareType: text().notNull(),
|
||||
targetUserId: varchar(),
|
||||
linkToken: varchar().unique(),
|
||||
permissionLevel: text().notNull().default("read-only"),
|
||||
createdAt: timestamp().notNull().defaultNow(),
|
||||
expiresAt: text().notNull(),
|
||||
revokedAt: text(),
|
||||
lastJoinedAt: text(),
|
||||
joinCount: integer().notNull().default(0),
|
||||
},
|
||||
{
|
||||
indexes: [
|
||||
{ name: "idx_session_shares_session_id", columns: ["sessionId"] },
|
||||
{ name: "idx_session_shares_host_id", columns: ["hostId"] },
|
||||
],
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
/** Who joined through a share, for the owner's record. */
|
||||
export const shareParticipants = adoptLegacyTable(
|
||||
"session_share_participants",
|
||||
defineTable("share_participants", {
|
||||
id: id(),
|
||||
shareId: varchar().notNull(),
|
||||
userId: varchar(),
|
||||
guestLabel: text(),
|
||||
joinedAt: timestamp().notNull().defaultNow(),
|
||||
leftAt: text(),
|
||||
}),
|
||||
);
|
||||
|
||||
/**
|
||||
* A collaboration room: members watching one stage, the live session the
|
||||
* current presenter shows through a "room" share.
|
||||
*/
|
||||
export const rooms = adoptLegacyTable(
|
||||
"collab_rooms",
|
||||
defineTable(
|
||||
"rooms",
|
||||
{
|
||||
id: varchar().primaryKey(),
|
||||
name: text().notNull(),
|
||||
ownerUserId: refUser(),
|
||||
persistent: boolean().notNull().default(false),
|
||||
presenterUserId: varchar(),
|
||||
stageProtocol: text(),
|
||||
stageHostId: integer(),
|
||||
stageShareId: varchar(),
|
||||
guestLinkToken: varchar(),
|
||||
createdAt: timestamp().notNull().defaultNow(),
|
||||
endedAt: text(),
|
||||
},
|
||||
{
|
||||
indexes: [{ name: "idx_collab_rooms_owner", columns: ["ownerUserId"] }],
|
||||
uniques: [
|
||||
{ name: "idx_collab_rooms_guest_token", columns: ["guestLinkToken"] },
|
||||
],
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
export const roomMembers = adoptLegacyTable(
|
||||
"collab_room_members",
|
||||
defineTable(
|
||||
"room_members",
|
||||
{
|
||||
id: id(),
|
||||
roomId: varchar().notNull(),
|
||||
userId: refUser(),
|
||||
roomRole: text().notNull().default("member"),
|
||||
addedBy: varchar(),
|
||||
createdAt: timestamp().notNull().defaultNow(),
|
||||
},
|
||||
{
|
||||
uniques: [
|
||||
{
|
||||
name: "idx_collab_room_members_room_user",
|
||||
columns: ["roomId", "userId"],
|
||||
},
|
||||
],
|
||||
indexes: [{ name: "idx_collab_room_members_user", columns: ["userId"] }],
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
export const tables = [shares, shareParticipants, rooms, roomMembers];
|
||||
Reference in new issue
Block a user