mirror of
https://github.com/spacebarchat/server.git
synced 2026-07-21 04:40:57 +00:00
Experimental unix socket mode
This commit is contained in:
Rory&
Generated
+1
-1
@@ -5,7 +5,7 @@
|
||||
<scripts>
|
||||
<script value="start:cdn" />
|
||||
</scripts>
|
||||
<node-interpreter value="/nix/store/dcdc33kdjdhjnzg6rkmd0cx4kpwl8cac-nodejs-20.17.0/bin/node" />
|
||||
<node-interpreter value="project" />
|
||||
<package-manager value="npm" />
|
||||
<envs />
|
||||
<EXTENSION ID="com.fapiko.jetbrains.plugins.better_direnv.runconfigs.NodeRunConfiguration">
|
||||
|
||||
Generated
+53
-43
@@ -24,12 +24,19 @@
|
||||
<option name="RECENT_GIT_ROOT_PATH" value="$PROJECT_DIR$" />
|
||||
</component>
|
||||
<component name="HighlightingSettingsPerFile">
|
||||
<setting file="file://$USER_HOME$/.cache/JetBrains/WebStorm2024.3/javascript/typings/node/20.17.10/node_modules/@types/node/crypto.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/amqplib/index.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/amqplib/properties.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/jsonwebtoken/index.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/node/events.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/result/node_modules/lambert-server/src/Server.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/result/node_modules/lambert-server/src/Utils.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/ws/index.d.mts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/ws/index.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/@types/ws/package.json" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/data-source/DataSource.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/driver/types/ReplicationMode.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/query-builder/Brackets.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/query-builder/SelectQueryBuilder.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/query-builder/WhereExpressionBuilder.d.ts" root0="SKIP_INSPECTION" />
|
||||
<setting file="file://$PROJECT_DIR$/node_modules/typeorm/repository/BaseEntity.d.ts" root0="SKIP_INSPECTION" />
|
||||
</component>
|
||||
<component name="ProblemsViewState">
|
||||
<option name="selectedTabId" value="ProjectErrors" />
|
||||
@@ -43,56 +50,56 @@
|
||||
<option name="hideEmptyMiddlePackages" value="true" />
|
||||
<option name="showLibraryContents" value="true" />
|
||||
</component>
|
||||
<component name="PropertiesComponent"><![CDATA[{
|
||||
"keyToString": {
|
||||
"NIXITCH_NIXPKGS_CONFIG": "/etc/nix/nixpkgs-config.nix",
|
||||
"NIXITCH_NIX_CONF_DIR": "",
|
||||
"NIXITCH_NIX_OTHER_STORES": "",
|
||||
"NIXITCH_NIX_PATH": "/home/Rory/.nix-defexpr/channels:nixpkgs=/nix/store/wb6agba4kfsxpbnb5hzlq58vkjzvbsk6-source",
|
||||
"NIXITCH_NIX_PROFILES": "/run/current-system/sw /nix/var/nix/profiles/default /etc/profiles/per-user/Rory /home/Rory/.local/state/nix/profile /nix/profile /home/Rory/.nix-profile",
|
||||
"NIXITCH_NIX_REMOTE": "",
|
||||
"NIXITCH_NIX_USER_PROFILE_DIR": "/nix/var/nix/profiles/per-user/Rory",
|
||||
"Node.js.Server.ts.executor": "Debug",
|
||||
"RunOnceActivity.ShowReadmeOnStart": "true",
|
||||
"RunOnceActivity.git.unshallow": "true",
|
||||
"git-widget-placeholder": "master",
|
||||
"javascript.nodejs.core.library.configured.version": "24.8.0",
|
||||
"javascript.nodejs.core.library.typings.version": "24.7.0",
|
||||
"last_opened_file_path": "/home/Rory/git/spacebar/server-master/src/schemas/api/users",
|
||||
"node.js.detected.package.eslint": "true",
|
||||
"node.js.detected.package.standard": "true",
|
||||
"node.js.selected.package.eslint": "(autodetect)",
|
||||
"node.js.selected.package.standard": "",
|
||||
"node.js.selected.package.tslint": "(autodetect)",
|
||||
"nodejs_interpreter_path": "node",
|
||||
"nodejs_package_manager_path": "npm",
|
||||
"npm.Start API.executor": "Run",
|
||||
"npm.Start CDN.executor": "Run",
|
||||
"npm.Start Gateway.executor": "Run",
|
||||
"npm.build.executor": "Run",
|
||||
"npm.start.executor": "Debug",
|
||||
"prettierjs.PrettierConfiguration.Package": "/home/Rory/git/spacebar/server-master/node_modules/prettier",
|
||||
"settings.editor.selected.configurable": "com.github.copilot.settings.customInstructions.CopilotInstructionsConfigurable",
|
||||
"ts.external.directory.path": "/home/Rory/git/spacebar/server-master/node_modules/typescript/lib"
|
||||
<component name="PropertiesComponent">{
|
||||
"keyToString": {
|
||||
"NIXITCH_NIXPKGS_CONFIG": "/etc/nix/nixpkgs-config.nix",
|
||||
"NIXITCH_NIX_CONF_DIR": "",
|
||||
"NIXITCH_NIX_OTHER_STORES": "",
|
||||
"NIXITCH_NIX_PATH": "/home/Rory/.nix-defexpr/channels:nixpkgs=/nix/store/wb6agba4kfsxpbnb5hzlq58vkjzvbsk6-source",
|
||||
"NIXITCH_NIX_PROFILES": "/run/current-system/sw /nix/var/nix/profiles/default /etc/profiles/per-user/Rory /home/Rory/.local/state/nix/profile /nix/profile /home/Rory/.nix-profile",
|
||||
"NIXITCH_NIX_REMOTE": "",
|
||||
"NIXITCH_NIX_USER_PROFILE_DIR": "/nix/var/nix/profiles/per-user/Rory",
|
||||
"Node.js.Server.ts.executor": "Debug",
|
||||
"RunOnceActivity.ShowReadmeOnStart": "true",
|
||||
"RunOnceActivity.git.unshallow": "true",
|
||||
"git-widget-placeholder": "master",
|
||||
"javascript.nodejs.core.library.configured.version": "24.8.0",
|
||||
"javascript.nodejs.core.library.typings.version": "24.7.0",
|
||||
"last_opened_file_path": "/home/Rory/git/spacebar/server-master/src/util/migration/postgres",
|
||||
"node.js.detected.package.eslint": "true",
|
||||
"node.js.detected.package.standard": "true",
|
||||
"node.js.selected.package.eslint": "(autodetect)",
|
||||
"node.js.selected.package.standard": "",
|
||||
"node.js.selected.package.tslint": "(autodetect)",
|
||||
"nodejs_interpreter_path": "node",
|
||||
"nodejs_package_manager_path": "npm",
|
||||
"npm.Start API.executor": "Run",
|
||||
"npm.Start CDN.executor": "Run",
|
||||
"npm.Start Gateway.executor": "Run",
|
||||
"npm.build.executor": "Run",
|
||||
"npm.start.executor": "Debug",
|
||||
"prettierjs.PrettierConfiguration.Package": "/home/Rory/git/spacebar/server-master/node_modules/prettier",
|
||||
"settings.editor.selected.configurable": "com.github.copilot.settings.customInstructions.CopilotInstructionsConfigurable",
|
||||
"ts.external.directory.path": "/home/Rory/git/spacebar/server-master/node_modules/typescript/lib"
|
||||
},
|
||||
"keyToStringList": {
|
||||
"DatabaseDriversLRU": [
|
||||
"postgresql"
|
||||
"keyToStringList": {
|
||||
"DatabaseDriversLRU": [
|
||||
"postgresql"
|
||||
],
|
||||
"GitStage.ChangesTree.GroupingKeys": [
|
||||
"directory",
|
||||
"module",
|
||||
"repository"
|
||||
"GitStage.ChangesTree.GroupingKeys": [
|
||||
"directory",
|
||||
"module",
|
||||
"repository"
|
||||
]
|
||||
}
|
||||
}]]></component>
|
||||
}</component>
|
||||
<component name="RecentsManager">
|
||||
<key name="CopyFile.RECENT_KEYS">
|
||||
<recent name="$PROJECT_DIR$/src/util/migration/postgres" />
|
||||
<recent name="$PROJECT_DIR$/src/schemas/api/users" />
|
||||
<recent name="$PROJECT_DIR$/src/util/entities" />
|
||||
<recent name="$PROJECT_DIR$/src/gateway/opcodes" />
|
||||
<recent name="$PROJECT_DIR$/src/api/routes/users/@me/billing" />
|
||||
<recent name="$PROJECT_DIR$/src/util/util/networking/abuseipdb" />
|
||||
</key>
|
||||
<key name="MoveFile.RECENT_KEYS">
|
||||
<recent name="$PROJECT_DIR$/src/schemas/api/guilds" />
|
||||
@@ -117,6 +124,7 @@
|
||||
<entry key="Start CDN" value="STOPPED" />
|
||||
<entry key="Start Gateway" value="STOPPED" />
|
||||
<entry key="Start bundle" value="STOPPED" />
|
||||
<entry key="build" value="STOPPED" />
|
||||
</map>
|
||||
</value>
|
||||
</entry>
|
||||
@@ -165,6 +173,8 @@
|
||||
<workItem from="1764432507485" duration="30779000" />
|
||||
<workItem from="1764521899659" duration="22029000" />
|
||||
<workItem from="1765142585081" duration="16912000" />
|
||||
<workItem from="1765849511651" duration="7921000" />
|
||||
<workItem from="1765861713706" duration="18045000" />
|
||||
</task>
|
||||
<servers />
|
||||
</component>
|
||||
|
||||
@@ -187,9 +187,7 @@ async function consume(this: WebSocket, opts: EventOpts) {
|
||||
}
|
||||
break;
|
||||
case "CHANNEL_CREATE":
|
||||
if (!permission.overwriteChannel(data.permission_overwrites).has("VIEW_CHANNEL")) {
|
||||
return;
|
||||
}
|
||||
if (!permission.overwriteChannel(data.permission_overwrites).has("VIEW_CHANNEL")) return;
|
||||
this.events[id] = await listenEvent(id, consumer, listenOpts);
|
||||
break;
|
||||
case "RELATIONSHIP_ADD":
|
||||
|
||||
@@ -21,7 +21,14 @@ import { RabbitMQ } from "./RabbitMQ";
|
||||
import EventEmitter from "events";
|
||||
import { EVENT, Event } from "../interfaces";
|
||||
import { randomUUID } from "crypto";
|
||||
import path from "path";
|
||||
import { Socket } from "node:net";
|
||||
import { FSWatcher } from "node:fs";
|
||||
import { Stopwatch } from "./Stopwatch";
|
||||
import { Config } from "./Config";
|
||||
export const events = new EventEmitter();
|
||||
let unixSocketListener: UnixSocketListener | null = null;
|
||||
let unixSocketWriter: UnixSocketWriter | null = null;
|
||||
|
||||
export async function emitEvent(payload: Omit<Event, "created_at">) {
|
||||
const id = (payload.guild_id || payload.channel_id || payload.user_id) as string;
|
||||
@@ -42,6 +49,12 @@ export async function emitEvent(payload: Omit<Event, "created_at">) {
|
||||
// todo: should we retry publishng the event?
|
||||
console.log("[RabbitMQ] ", e);
|
||||
}
|
||||
} else if (process.env.EVENT_TRANSMISSION === "unix" && process.env.EVENT_SOCKET_PATH) {
|
||||
if (!unixSocketWriter) {
|
||||
unixSocketWriter = new UnixSocketWriter(process.env.EVENT_SOCKET_PATH);
|
||||
await unixSocketWriter.init();
|
||||
}
|
||||
await unixSocketWriter.emit(payload);
|
||||
} else if (process.env.EVENT_TRANSMISSION === "process") {
|
||||
process.send?.({ type: "event", event: payload, id } as ProcessEvent);
|
||||
} else {
|
||||
@@ -57,6 +70,14 @@ export async function initEvent() {
|
||||
// use event emitter
|
||||
// use process messages
|
||||
}
|
||||
|
||||
await listenEvent("spacebar", async (event) => {
|
||||
console.log("[Event] Received spacebar event:", event);
|
||||
if ((event.event as string) === "SB_RELOAD_CONFIG") {
|
||||
console.log("[Event] Reloading config due to RELOAD_CONFIG event");
|
||||
await Config.init(true);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export interface EventOpts extends Event {
|
||||
@@ -84,6 +105,12 @@ export async function listenEvent(event: string, callback: (event: EventOpts) =>
|
||||
return await rabbitListen(channel, event, callback, {
|
||||
acknowledge: opts?.acknowledge,
|
||||
});
|
||||
} else if (process.env.EVENT_TRANSMISSION === "unix" && process.env.EVENT_SOCKET_PATH) {
|
||||
if (!unixSocketListener) {
|
||||
unixSocketListener = new UnixSocketListener(path.join(process.env.EVENT_SOCKET_PATH, `${process.pid}.sock`));
|
||||
await unixSocketListener.init();
|
||||
}
|
||||
return await unixSocketListener.listen(event, callback);
|
||||
} else if (process.env.EVENT_TRANSMISSION === "process") {
|
||||
const cancel = async () => {
|
||||
process.removeListener("message", listener);
|
||||
@@ -155,3 +182,126 @@ async function rabbitListen(channel: Channel, id: string, callback: (event: Even
|
||||
|
||||
return cancel;
|
||||
}
|
||||
|
||||
class UnixSocketListener {
|
||||
eventEmitter: EventEmitter;
|
||||
socketPath: string;
|
||||
|
||||
constructor(socketPath: string) {
|
||||
this.eventEmitter = new EventEmitter();
|
||||
this.socketPath = socketPath;
|
||||
}
|
||||
|
||||
async init() {
|
||||
const net = await import("net");
|
||||
const server = net.createServer((socket) => {
|
||||
socket.on("connect", () => {
|
||||
console.log("[Events] Unix socket client connected");
|
||||
});
|
||||
socket.on("data", (data) => {
|
||||
try {
|
||||
const payload = JSON.parse(data.toString());
|
||||
this.eventEmitter.emit(payload.id, payload.event);
|
||||
} catch (e) {
|
||||
console.error("[Events] Failed to parse unix socket data:", e);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
server.listen(this.socketPath, () => {
|
||||
console.log(`Unix socket server listening on ${this.socketPath}`);
|
||||
});
|
||||
|
||||
const shutdown = () => {
|
||||
console.log("[Events] Closing unix socket server");
|
||||
server.close();
|
||||
process.exit(0);
|
||||
};
|
||||
for (const sig of ["SIGINT", "SIGTERM", "SIGQUIT"] as const) {
|
||||
process.on(sig, shutdown);
|
||||
}
|
||||
}
|
||||
|
||||
async listen(event: string, callback: (event: EventOpts) => unknown): Promise<() => Promise<void>> {
|
||||
const listener = (data: any) => {
|
||||
callback({
|
||||
...data,
|
||||
cancel,
|
||||
});
|
||||
};
|
||||
|
||||
this.eventEmitter.addListener(event, listener);
|
||||
|
||||
const cancel = async () => {
|
||||
this.eventEmitter.removeListener(event, listener);
|
||||
this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() - 1);
|
||||
};
|
||||
|
||||
this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() + 1);
|
||||
|
||||
return cancel;
|
||||
}
|
||||
}
|
||||
|
||||
class UnixSocketWriter {
|
||||
socketPath: string;
|
||||
clients: { [key: string]: Socket } = {};
|
||||
watcher: FSWatcher;
|
||||
|
||||
constructor(socketPath: string) {
|
||||
this.socketPath = socketPath;
|
||||
}
|
||||
|
||||
async init() {
|
||||
const net = await import("net");
|
||||
const fs = await import("fs");
|
||||
|
||||
if (!fs.opendirSync(this.socketPath)) throw new Error("Unix socket path does not exist or is not a directory: " + this.socketPath);
|
||||
|
||||
console.log("[Events] Unix socket writer initializing for", this.socketPath);
|
||||
|
||||
const connect = (file: string) => {
|
||||
const fullPath = path.join(this.socketPath, file);
|
||||
this.clients[fullPath] = net.createConnection(fullPath, () => {
|
||||
console.log("[Events] Unix socket client connected to", fullPath);
|
||||
});
|
||||
};
|
||||
|
||||
// connect to all sockets, now and in the future
|
||||
this.watcher = fs.watch(this.socketPath, {}, (eventType, filename) => {
|
||||
console.log("[Events] Unix socket writer received watch sig", eventType, filename);
|
||||
connect(filename!);
|
||||
});
|
||||
|
||||
// connect to existing sockets if any
|
||||
fs.readdir(this.socketPath, (err, files) => {
|
||||
if (err) return console.error("[Events] Unix socket writer failed to read directory:", err);
|
||||
|
||||
console.log("[Events] Unix socket writer found existing sockets:", files);
|
||||
files.forEach((file) => {
|
||||
connect(file);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async emit(event: Event) {
|
||||
if (!this.clients) throw new Error("UnixSocketWriter not initialized");
|
||||
|
||||
const tsw = Stopwatch.startNew();
|
||||
const payload = JSON.stringify({ id: (event.guild_id || event.channel_id || event.user_id) as string, event });
|
||||
for (const socket of Object.entries(this.clients)) {
|
||||
if(socket[1].destroyed) {
|
||||
console.log("[Events] Unix socket writer found destroyed socket, removing:", socket[0]);
|
||||
delete this.clients[socket[0]];
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
socket[1].write(payload);
|
||||
} catch (e) {
|
||||
console.error("[Events] Unix socket writer failed to write to socket", socket[0], ":", e);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[Events] Unix socket writer emitted to ${Object.entries(this.clients).length} sockets in ${tsw.elapsed().totalMilliseconds}ms`);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user