dandri/server
1
0
Fork 0
mirror of https://github.com/spacebarchat/server.git synced 2026-06-07 15:21:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Process lifecycle scripts

Browse Source 
#Conflicts:
#	src/api/Server.ts
#	src/bundle/Server.ts
#	src/cdn/Server.ts
#	src/gateway/Server.ts
#	src/gateway/events/Connection.ts
#	src/util/util/ipc/listener/RabbitMqSingleListener.ts
#	src/util/util/ipc/listener/UnixSocketListener.ts
#	src/util/util/ipc/writer/UnixSocketWriter.ts
#	src/webrtc/Server.ts
This commit is contained in:
Rory&
2026-05-28 10:38:39 +02:00
parent ac3160e833
commit b1b73d45e7
16 changed files with 181 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package-lock.json
Generated
BIN
View File
Binary file not shown.
package.json
+2 -1
View File
@@ -75,7 +75,8 @@
"pretty-quick": "^4.2.2",
"ts-node": "^10.9.2",
"typescript": "^6.0.3",
"typescript-json-schema": "^0.67.4"
"typescript-json-schema": "^0.67.4",
"why-is-node-running": "^3.2.2"
},
"dependencies": {
"@spacebarchat/pion-webrtc": "^0.0.4",
src/api/Server.ts
+7 -5
View File
@@ -16,15 +16,16 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import path from "node:path";
import { Request, Response, Router } from "express";
import morgan from "morgan";
import { Server, ServerOptions } from "lambert-server";
import { red } from "picocolors";
import { Config, ConnectionConfig, ConnectionLoader, Email, JSONReplacer, WebAuthn, initDatabase, initEvent, registerRoutes, getDatabase, getRevInfoOrFail } from "@spacebar/util";
import { Authentication, CORS, ImageProxy, BodyParser, ErrorHandler, initRateLimits, initTranslation } from "./middlewares";
import { Request, Response, Router } from "express";
import { Server, ServerOptions } from "lambert-server";
import morgan from "morgan";
import path from "node:path";
import { red } from "picocolors";
import { initInstance } from "./util/handlers/Instance";
import { route } from "./util";
import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
const ASSETS_FOLDER = path.join(__dirname, "..", "..", "assets");
const PUBLIC_ASSETS_FOLDER = path.join(ASSETS_FOLDER, "public");
@@ -196,6 +197,7 @@ export class SpacebarServer extends Server {
if (logRequests) console.log(red(`Warning: Request logging is enabled! This will spam your console!\nTo disable this, unset the 'LOG_REQUESTS' environment variable!`));
await ProcessLifecycle.Ready();
return super.start();
}
}
src/api/routes/stop.ts
+5 -2
View File
@@ -16,8 +16,9 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { route } from "@spacebar/api";
import { route, SpacebarServer } from "@spacebar/api";
import { Request, Response, Router } from "express";
import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
const router: Router = Router({ mergeParams: true });
@@ -35,7 +36,9 @@ router.post(
(req: Request, res: Response) => {
console.log(`/stop was called by ${req.user_id} at ${new Date()}`);
res.sendStatus(200);
process.kill(process.pid, "SIGTERM");
ProcessLifecycle.Shutdown().catch((e) => {
console.error("Failed to shut down:", e);
});
},
);
src/bundle/Server.ts
+7 -11
View File
@@ -16,21 +16,18 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import morgan from "morgan";
process.on("unhandledRejection", console.error);
process.on("uncaughtException", console.error);
import http from "node:http";
import fs from "node:fs";
import cluster from "node:cluster";
import morgan from "morgan";
import express from "express";
import { green, bold } from "picocolors";
import * as Api from "@spacebar/api";
import * as Gateway from "@spacebar/gateway";
import * as Webrtc from "@spacebar/webrtc";
import { CDNServer } from "@spacebar/cdn";
import express from "express";
import { green, bold } from "picocolors";
import { Config, initDatabase } from "@spacebar/util";
import fs from "node:fs";
import cluster from "node:cluster";
import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
const app = express();
const server = http.createServer();
@@ -48,8 +45,7 @@ const webrtc = new Webrtc.Server({
production,
});
process.on("SIGTERM", async () => {
console.log("Shutting down due to SIGTERM");
ProcessLifecycle.eventEmitter.on("stopping", async () => {
await gateway.stop();
await cdn.stop();
await api.stop();
src/cdn/Server.ts
+6 -2
View File
@@ -16,13 +16,14 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import path from "node:path";
import morgan from "morgan";
import { Server, ServerOptions } from "lambert-server";
import { Attachment, Config, initDatabase, registerRoutes } from "@spacebar/util";
import { CORS, BodyParser } from "@spacebar/api";
import path from "node:path";
import guildProfilesRoute from "./routes/guild-profiles";
import morgan from "morgan";
import { storage } from "./util";
import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
export type CDNServerOptions = ServerOptions;
@@ -71,6 +72,7 @@ export class CDNServer extends Server {
this.app.use("/guilds/:guild_id/users/:user_id/banners", guildProfilesRoute);
if (process.env.LOG_ROUTES !== "false") console.log("[Server] Route /guilds/:guild_id/users/:user_id/banners registered");
await ProcessLifecycle.Ready();
return super.start();
}
@@ -89,6 +91,8 @@ export class CDNServer extends Server {
}
async stop() {
await ProcessLifecycle.Shutdown();
await ProcessLifecycle.Finalize();
return super.stop();
}
}
src/gateway/Server.ts
+17 -13
View File
@@ -16,21 +16,21 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import dotenv from "dotenv";
dotenv.config({ quiet: true });
import { checkToken, closeDatabase, Config, initDatabase, initEvent, Rights } from "@spacebar/util";
import ws from "ws";
import { Connection, openConnections } from "./events/Connection";
import http from "node:http";
import { cleanupOnStartup } from "./util";
import { randomString } from "@spacebar/api";
import { setInterval } from "node:timers";
import ws from "ws";
import { checkToken, Config, initDatabase, initEvent, Rights } from "@spacebar/util";
import { randomString } from "@spacebar/api"; // TODO: move to util
import { Connection, openConnections } from "./events/Connection";
import { cleanupOnStartup, OPCODES, Send } from "./util";
import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
export class Server {
public ws: ws.Server;
public port: number;
public server: http.Server;
public production: boolean;
private monitoringLoop: NodeJS.Timeout;
constructor({ port, server, production }: { port: number; server?: http.Server; production?: boolean }) {
this.port = port;
@@ -42,7 +42,7 @@ export class Server {
const eluP = [1, 5, 15].map(() => performance.eventLoopUtilization());
const cpu = [1, 5, 15].map(() => process.cpuUsage());
let sec = 0;
setInterval(() => {
const monitoringLoop = setInterval(() => {
sec += 1;
// for some reason this behaves differently from cpuUsage, so we need an absolute reference as "previous"
const eluC = performance.eventLoopUtilization();
@@ -150,6 +150,8 @@ export class Server {
res.writeHead(200).end("Online");
});
ProcessLifecycle.eventEmitter.on("stopping", () => clearTimeout(monitoringLoop));
}
this.server.on("upgrade", (request, socket, head) => {
@@ -177,14 +179,16 @@ export class Server {
this.server.listen(this.port);
console.log(`[Gateway] online on 0.0.0.0:${this.port}`);
}
await ProcessLifecycle.Ready();
}
async stop() {
await ProcessLifecycle.Shutdown();
clearInterval(this.monitoringLoop);
this.ws.clients.forEach((x) => x.close());
this.ws.close(() => {
this.server.close(() => {
closeDatabase();
});
});
this.ws.close();
this.server.close();
await ProcessLifecycle.Finalize();
}
}
src/gateway/events/Close.ts
+31 -29
View File
@@ -19,6 +19,7 @@
import { WebSocket } from "@spacebar/gateway";
import { emitEvent, Member, PresenceUpdateEvent, Session, SessionsReplace, User, VoiceState, VoiceStateUpdateEvent, distributePresenceUpdate } from "@spacebar/util";
import { randomString } from "@spacebar/api";
import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
export async function Close(this: WebSocket, code: number, reason: Buffer) {
console.log("[WebSocket] closed", code, reason.toString());
@@ -32,36 +33,37 @@ export async function Close(this: WebSocket, code: number, reason: Buffer) {
const authSessionId = this.session?.session_id;
const closedAt = Date.now();
setTimeout(async () => {
console.log("Handling presence update after disconnect");
try {
if (authSessionId && this.user_id) {
const s = await Session.findOne({
where: { user_id: this.user_id, session_id: authSessionId },
});
if (s && (s.last_seen?.getTime() ?? 0) <= closedAt) {
console.log("... updating session");
await Session.update({ user_id: this.user_id, session_id: authSessionId }, { status: "offline", activities: [], client_status: {} });
this.session = await Session.findOneOrFail({ where: { session_id: this.session_id } });
console.log("... distributing PRESENCE_UPDATE");
await distributePresenceUpdate(this.user_id, {
event: "PRESENCE_UPDATE",
data: {
user: (await User.findOneOrFail({ where: { id: this.user_id } })).toPublicUser(),
status: this.session!.getPublicStatus(),
client_status: this.session!.client_status,
activities: this.session!.activities,
},
origin: "GATEWAY_CLOSE",
transaction_id: `IDENT_${this.user_id}_${randomString()}`,
} satisfies PresenceUpdateEvent);
console.log("... done!");
} else console.log("... Discarding presence update as the session reactivated");
if (!(ProcessLifecycle.state === "stopping" || ProcessLifecycle.state === "stopped"))
setTimeout(async () => {
console.log("Handling presence update after disconnect");
try {
if (authSessionId && this.user_id) {
const s = await Session.findOne({
where: { user_id: this.user_id, session_id: authSessionId },
});
if (s && (s.last_seen?.getTime() ?? 0) <= closedAt) {
console.log("... updating session");
await Session.update({ user_id: this.user_id, session_id: authSessionId }, { status: "offline", activities: [], client_status: {} });
this.session = await Session.findOneOrFail({ where: { session_id: this.session_id } });
console.log("... distributing PRESENCE_UPDATE");
await distributePresenceUpdate(this.user_id, {
event: "PRESENCE_UPDATE",
data: {
user: (await User.findOneOrFail({ where: { id: this.user_id } })).toPublicUser(),
status: this.session!.getPublicStatus(),
client_status: this.session!.client_status,
activities: this.session!.activities,
},
origin: "GATEWAY_CLOSE",
transaction_id: `IDENT_${this.user_id}_${randomString()}`,
} satisfies PresenceUpdateEvent);
console.log("... done!");
} else console.log("... Discarding presence update as the session reactivated");
}
} catch (e) {
console.error("[WebSocket] Close session cleanup failed", code, e);
}
} catch (e) {
console.error("[WebSocket] Close session cleanup failed", code, e);
}
}, 10_000);
}, 10_000);
const voiceState = await VoiceState.findOne({
where: { user_id: this.user_id },
src/gateway/events/Connection.ts
+20 -9
View File
@@ -29,6 +29,7 @@ import { Deflate, Inflate } from "fast-zlib";
import { URL } from "node:url";
import { Config } from "@spacebar/util";
import { Decoder, Encoder } from "@toondepauw/node-zstd";
import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
// TODO: check rate limit
// TODO: specify rate limit in config
@@ -43,16 +44,26 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
if (index !== -1) openConnections.splice(index, 1);
});
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
process.on(sig, async () => {
await Send(socket, {
op: OPCODES.Reconnect,
s: socket.sequence++,
d: Math.round(Math.random() * 5000),
});
socket.close(1000);
const onShutdown = async () => {
await Send(socket, {
op: OPCODES.Reconnect,
s: socket.sequence++,
d: Math.round(Math.random() * 5000),
});
}
const closeListeners = socket.listeners("close");
for (const listener of closeListeners) {
socket.off("close", listener);
// noinspection JSVoidFunctionReturnValueUsed - awaiting results
const res = listener.call(socket, 1000, 0) as void | Promise<void>;
if (res) await res;
}
socket.close(1000);
};
if (ProcessLifecycle.state == "stopping" || ProcessLifecycle.state == "stopped") return await onShutdown();
ProcessLifecycle.eventEmitter.on("stopping", onShutdown);
const forwardedFor = Config.get().security.forwardedFor;
const ipAddress = forwardedFor ? (request.headers[forwardedFor.toLowerCase()] as string) : request.socket.remoteAddress;
src/util/util/Database.ts
+5 -2
View File
@@ -23,6 +23,7 @@ import { DataSource } from "typeorm";
// noinspection ES6PreferShortImport
import { ConfigEntity } from "../entities/Config";
import fs from "node:fs";
import { ProcessLifecycle } from "./ProcessLifecycle";
// UUID extension option is only supported with postgres
// We want to generate all id's with Snowflakes that's why we have our own BaseEntity class
@@ -127,11 +128,13 @@ export async function initDatabase(): Promise<DataSource> {
}
}
console.log(`[Database] ${green("Connected")}`);
ProcessLifecycle.eventEmitter.on("stopped", async () => await closeDatabase());
console.log(`[Database] ${green("Connected")}`);
return dbConnection;
}
export async function closeDatabase() {
await dbConnection?.destroy();
if (DataSourceOptions.isInitialized) await DataSourceOptions.destroy();
if (dbConnection?.isInitialized) await dbConnection?.destroy();
}
src/util/util/ProcessLifecycle.ts
+62
View File
@@ -0,0 +1,62 @@
/*
Spacebar: A FOSS re-implementation and extension of the Discord.com backend.
Copyright (C) 2026 Spacebar and Spacebar Contributors
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published
by the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import EventEmitter from "node:events";
import whyIsNodeRunning from "why-is-node-running";
interface ProcessLifecycleEvents {
starting: unknown[];
running: unknown[];
stopping: unknown[];
stopped: unknown[];
}
export class ProcessLifecycle {
static state: keyof ProcessLifecycleEvents = "starting";
static eventEmitter: EventEmitter<ProcessLifecycleEvents> = new EventEmitter();
// to be ran after startup is finished
static async Ready() {
await this.emitAsync((this.state = "running"));
}
// to be ran at the start of shutdown
static async Shutdown() {
await this.emitAsync((this.state = "stopping"));
}
// to be ran at the end of shutdown (clean up sockets, ...)
static async Finalize() {
await this.emitAsync((this.state = "stopped"));
}
// emit, except it awaits promises
private static async emitAsync(eventName: keyof ProcessLifecycleEvents) {
for (const evt of this.eventEmitter.listeners(eventName)) {
// noinspection JSVoidFunctionReturnValueUsed - we want to handle async functions blocking aswell
const res = evt() as void | Promise<void>;
if (res) await res;
}
}
}
process.on("SIGUSR1", () => {
console.log("Handling SIGUSR1:");
whyIsNodeRunning();
console.log("\nProcess state:", ProcessLifecycle.state);
});
src/util/util/ipc/listener/RabbitMqSingleListener.ts
+3 -4
View File
@@ -17,10 +17,11 @@
*/
import EventEmitter from "node:events";
import { randomUUID } from "node:crypto";
import { BaseEventListener } from "./BaseEventListener";
import { EVENT, Event, EventOpts, sleep } from "@spacebar/util";
import amqp, { Channel, ChannelModel } from "amqplib";
import { randomUUID } from "node:crypto";
import { ProcessLifecycle } from "../../ProcessLifecycle";
export class RabbitMqSingleListener extends BaseEventListener {
private readonly host: string;
@@ -50,9 +51,7 @@ export class RabbitMqSingleListener extends BaseEventListener {
}
this.channel = await this.connection.createChannel();
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
process.on(sig, () => this.close());
}
ProcessLifecycle.eventEmitter.on("stopped", async () => await this.close());
this.connection.on("error", (err) => {
console.error("[RabbitMQSingleListener] Connection error:", err);
src/util/util/ipc/listener/UnixSocketListener.ts
+3 -3
View File
@@ -21,6 +21,7 @@ import fs from "node:fs";
import net, { Server } from "node:net";
import { BaseEventListener } from "./BaseEventListener";
import { EVENT, Event, EventOpts } from "@spacebar/util";
import { ProcessLifecycle } from "../../ProcessLifecycle";
export class UnixSocketListener extends BaseEventListener {
eventEmitter: EventEmitter;
@@ -78,9 +79,7 @@ export class UnixSocketListener extends BaseEventListener {
console.log(`[UnixSocketListener] Listening on ${this.socketPath}`);
});
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
process.on(sig, () => this.close());
}
ProcessLifecycle.eventEmitter.on("stopped", async () => await this.close());
this.isInitialized = true;
}
@@ -96,6 +95,7 @@ export class UnixSocketListener extends BaseEventListener {
try {
fs.unlinkSync(this.socketPath);
} catch (e) {
if (e instanceof Error && "errno" in e && e.errno == -2) return;
console.error("[UnixSocketListener] Failed to unlink socket file:", e);
}
}
src/util/util/ipc/writer/RabbitMqSingleWriter.ts
+2 -4
View File
@@ -19,6 +19,7 @@
import { BaseEventWriter } from "./BaseEventWriter";
import amqp, { Channel, ChannelModel } from "amqplib";
import { Event, sleep } from "@spacebar/util";
import { ProcessLifecycle } from "../../ProcessLifecycle";
export class RabbitMqSingleWriter extends BaseEventWriter {
private readonly host: string;
@@ -46,10 +47,7 @@ export class RabbitMqSingleWriter extends BaseEventWriter {
}
this.channel = await this.connection.createChannel();
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
process.on(sig, () => this.close());
}
ProcessLifecycle.eventEmitter.on("stopped", async () => await this.close());
this.connection.on("error", (err) => {
console.error("[RabbitMQSingleWriter] Connection error:", err);
});
src/util/util/ipc/writer/UnixSocketWriter.ts
+2 -4
View File
@@ -22,6 +22,7 @@ import path from "node:path";
import { red } from "picocolors";
import { BaseEventWriter } from "./BaseEventWriter";
import { Event, Stopwatch } from "@spacebar/util";
import { ProcessLifecycle } from "../../ProcessLifecycle";
export class UnixSocketWriter extends BaseEventWriter {
socketPath: string;
@@ -142,10 +143,7 @@ export class UnixSocketWriter extends BaseEventWriter {
console.error("[UnixSocketWriter] Unix socket writer failed to read directory:", err);
}
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
process.on(sig, () => this.close());
}
ProcessLifecycle.eventEmitter.on("stopped", async () => await this.close());
this.isInitializing = false;
}
src/webrtc/Server.ts
+9 -6
View File
@@ -15,14 +15,14 @@
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import dotenv from "dotenv";
dotenv.config({ quiet: true });
import { closeDatabase, Config, initDatabase, initEvent, Session, TimeSpan } from "@spacebar/util";
import http from "node:http";
import ws from "ws";
import { green, yellow } from "picocolors";
import { Config, initDatabase, initEvent } from "@spacebar/util";
import { Connection } from "./events/Connection";
import { loadWebRtcLibrary, mediaServer, WRTC_PORT_MAX, WRTC_PORT_MIN, WRTC_PUBLIC_IP } from "./util";
import { green, yellow } from "picocolors";
import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
export class Server {
public ws: ws.Server;
@@ -76,11 +76,14 @@ export class Server {
this.server.listen(this.port);
console.log(`[WebRTC] ${green(`online on 0.0.0.0:${this.port}`)}`);
}
await ProcessLifecycle.Ready();
}
async stop() {
await closeDatabase();
await ProcessLifecycle.Shutdown();
this.server.close();
mediaServer?.stop();
await mediaServer?.stop();
await ProcessLifecycle.Finalize();
}
}