CREATIVE_tg1/watchparty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
CREATIVE_tg1
2026-06-21 03:53:06 +03:00
commit f1dd557c5d
147 changed files with 34363 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/cleanup.ts
+16
View File
@@ -0,0 +1,16 @@
import { postgres } from "./utils/postgres.ts";
cleanupPostgres();
setInterval(cleanupPostgres, 5 * 60 * 1000);
async function cleanupPostgres() {
if (!postgres) {
return;
}
console.time("[CLEANUP]");
const result = await postgres?.query(
`DELETE FROM room WHERE owner IS NULL AND ("lastUpdateTime" < NOW() - INTERVAL '1 day' OR "lastUpdateTime" IS NULL)`,
);
console.log(result.command, result.rowCount);
console.timeEnd("[CLEANUP]");
}
server/config.ts
+69
View File
@@ -0,0 +1,69 @@
import { loadEnvFile } from "node:process";
try {
loadEnvFile();
} catch (e) {
console.log(e);
}
const defaults = {
REDIS_URL: "", // Optional, for metrics
DATABASE_URL: "", // Optional, for permanent rooms and VBrowser management
YOUTUBE_API_KEY: "", // Optional, provide one to enable searching YouTube
NODE_ENV: "", // Usually, you should let process.env.NODE_ENV override this
FIREBASE_ADMIN_SDK_CONFIG: "", // Optional, for features requiring sign-in/authentication
FIREBASE_DATABASE_URL: "", // Optional (unused)
STRIPE_SECRET_KEY: "", // Optional, for subscriptions
VBROWSER_SESSION_SECONDS: 10800, // Number of seconds to allow vbrowsers to run for
VBROWSER_SESSION_SECONDS_LARGE: 86400, // Number of seconds to allow large vbrowsers to run for
VM_POOL_RAMP_DOWN_HOURS: "", // Comma separated start/end UTC hours of the ramp down period
VM_POOL_RAMP_UP_HOURS: "", // Comma separated start/end UTC hours of the ramp up period
VBROWSER_TAG: "", // Optional, tag to put on VBrowser VM instances
DO_TOKEN: "", // Optional, for DigitalOcean VMs
DO_GATEWAY: "", // Gateway handling SSL termination
DO_IMAGE: "", // ID of DigitalOcean snapshot image to use for vbrowser
DO_SSH_KEYS: "", // IDs of DigitalOcean SSH keys to access vbrowsers
HETZNER_TOKEN: "", // Optional, for Hetzner VMs
HETZNER_GATEWAY: "", // Gateway handling SSL termination
HETZNER_SSH_KEYS: "", // IDs of Hetzner SSH keys to access vbrowsers
HETZNER_IMAGE: "", // ID of Hetzner snapshot image to use for vbrowser
VM_MANAGER_CONFIG: "", // Comma-separated list of the pools of VMs to run (provider:size:region:minSize:limitSize:hostname), e.g. Docker:large:US:0:1:localhost,Docker:standard:US:0:1:localhost
SCW_SECRET_KEY: "", // Optional, for Scaleway VMs
SCW_ORGANIZATION_ID: "", // Optional, for Scaleway VMs
SCW_GATEWAY: "", // Gateway handling SSL termination
SCW_IMAGE: "", // ID of Scaleway snapshot image to use for vbrowser
DOCKER_VM_HOST: "localhost", // Optional, for Docker VMs
DOCKER_VM_HOST_SSH_USER: "root", // Optional, username for Docker host
DOCKER_VM_HOST_SSH_KEY_BASE64: "", // Optional, private SSH key for Docker host, or default to ~/.ssh/id_rsa content
SSL_KEY_FILE: "", // Optional, Filename of SSL key (to use https)
SSL_CRT_FILE: "", // Optional, Filename of SSL cert (to use https)
PORT: 8080, // Port to use for server
HOST: "0.0.0.0", // Host interface to bind server to
STATS_KEY: "", // Secret string to validate viewing stats
BETA_USER_EMAILS: "", // Comma-delimited list of user emails to include in the beta
CUSTOM_SETTINGS_HOSTNAME: "", // Hostname to send different config settings to client
STREAM_PATH: "", // Path of server that supports additional video streams
CONVERT_PATH: "", // Path of server that supports video conversion
ROOM_CAPACITY: 0, // Maximum capacity of a standard room. Set to 0 for unlimited.
ROOM_CAPACITY_SUB: 0, // Maximum capacity of a sub room. Set to 0 for unlimited.
BUILD_DIRECTORY: "build", // Name of the directory where the built React UI is served from
VM_MIN_UPTIME_MINUTES: 0, // Number of minutes of the hour VMs must exist for before being eligible for termination
SHARD: undefined, // Shard ID of the web server (configure in ecosystem.config.js)
FREE_ROOM_LIMIT: 1, // The maximum number of rooms a free user can have
SUBSCRIBER_ROOM_LIMIT: 20, // The maximum number of rooms a subscriber can have
VMWORKER_PORT: 3100, // Port to use for the vmWorker HTTP server
VM_ASSIGNMENT_TIMEOUT: 75, // Number of seconds to wait for a VM before failing
DISCORD_BOT_TOKEN: "", // Token for the Discord bot that generates WatchParty links
DISCORD_ADMIN_BOT_TOKEN: "", // Optional, for Discord bot to set subscriber roles
DISCORD_ADMIN_BOT_SERVER_ID: "708181150220156929", // Optional, ID of the Discord server
DISCORD_ADMIN_BOT_SUB_ROLE_ID: "722202622345609246", // Optional, ID of subscriber role
MEDIASOUP_SERVER: "", // Optional, URL of the MediaSoup server to broadcast to for larger screen/file shares
TWITCH_PROXY_PATH: "", // Optional, URL of the server that can proxy twitch HLS stream playlists and segments
VBROWSER_ADMIN_KEY: "", // Optional, the key to hit admin endpoints on the vbrowser
OPENSUBTITLES_KEY: "", // Optional, key to OpenSubtitles API
};
export default {
...defaults,
...process.env,
};
server/discordBot.ts
+41
View File
@@ -0,0 +1,41 @@
import { Client, IntentsBitField, Events } from "discord.js";
import config from "./config.ts";
import axios from "axios";
import { redisCount } from "./utils/redis.ts";
// URL to invite bot: https://discord.com/api/oauth2/authorize?client_id=1071394728513380372&permissions=2147485696&scope=bot
const HOST_NAME = "https://www.watchparty.me";
const API_NAME = "https://backend.watchparty.me";
const client = new Client({
intents: [IntentsBitField.Flags.Guilds, IntentsBitField.Flags.GuildMessages],
});
client.on("ready", () => {
console.log("I am ready!");
console.log("bot is in %s guilds", client.guilds.cache.size);
});
client.on(Events.InteractionCreate, async (interaction) => {
if (!interaction.isChatInputCommand()) return;
if (interaction.commandName === "watch") {
const preload = interaction.options.get("video")?.value;
// Call the watchparty API to make a room
const response = await axios.post(API_NAME + "/createRoom", {
video: preload,
});
redisCount("discordBotWatch");
// Return the generated room URL
await interaction.reply({
content: `Created a new WatchParty${
preload ? ` with video ${preload}` : ""
}!
${HOST_NAME + "/watch" + response.data.name}
`,
});
}
});
client.login(config.DISCORD_BOT_TOKEN);
server/ecosystem.config.js
+81
View File
@@ -0,0 +1,81 @@
export const apps = [
// {
// name: 'server',
// script: './server/server.ts',
// log_date_format: 'YYYY-MM-DD HH:mm:ss Z',
// interpreter: 'node',
// env: {
// PORT: 80,
// },
// },
{
name: "shard1",
script: "./server/server.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
env: {
SHARD: 1,
PORT: 3001,
},
},
{
name: "shard2",
script: "./server/server.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
env: {
SHARD: 2,
PORT: 3002,
},
},
// {
// name: 'shard3',
// script: './server/server.ts',
// log_date_format: 'YYYY-MM-DD HH:mm:ss Z',
// interpreter: 'node',
// env: {
// SHARD: 3,
// PORT: 3003,
// },
// },
{
name: "vmWorker",
script: "./server/vmWorker.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
env: {
HETZNER_GATEWAY: "gateway2.watchparty.me",
HETZNER_SSH_KEYS: "1570536",
HETZNER_IMAGE: "398857306",
SCW_GATEWAY: "gateway2.watchparty.me",
SCW_IMAGE: "172bd9df-eba5-44e7-add0-f6edbb0f9c64",
DO_GATEWAY: "gateway2.watchparty.me",
DO_IMAGE: "150334605",
DO_SSH_KEYS: "cc:3d:a7:d3:99:17:fe:b7:dd:59:c4:78:14:d4:02:d1",
},
},
{
name: "syncSubs",
script: "./server/syncSubs.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
},
{
name: "timeSeries",
script: "./server/timeSeries.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
},
{
name: "cleanup",
script: "./server/cleanup.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
},
{
name: "discordBot",
script: "./server/discordBot.ts",
log_date_format: "YYYY-MM-DD HH:mm:ss Z",
interpreter: "node",
},
];
server/index.d.ts
Vendored
+41
View File
@@ -0,0 +1,41 @@
interface YoutubeResult {
kind: string;
etag: string;
snippet: {
publishedAt: string;
channelId: string;
title: string;
description: string;
thumbnails: {
default: {
url: string;
width: number;
height: number;
};
medium: {
url: string;
width: number;
height: number;
};
high: {
url: string;
width: number;
height: number;
};
standard: {
url: string;
width: number;
height: number;
};
};
channelTitle: string;
tags: string[];
categoryId: string;
liveBroadcastContent: string;
localized: {
title: string;
description: string;
};
defaultAudioLanguage: string;
};
}
server/room.ts
+1457
View File
File diff suppressed because it is too large Load Diff
server/server.ts
+793
View File
@@ -0,0 +1,793 @@
import config from "./config.ts";
import fs from "node:fs";
import express, { type Response } from "express";
import bodyParser from "body-parser";
import compression from "compression";
import cors from "cors";
import https from "node:https";
import http from "node:http";
import { Server } from "socket.io";
import { searchYoutube, youtubePlaylist } from "./utils/youtube.ts";
import { Room } from "./room.ts";
import { redis, redisCount } from "./utils/redis.ts";
import {
getCustomerByEmail,
createSelfServicePortal,
getIsSubscriberByEmail,
} from "./utils/stripe.ts";
import { deleteUser, validateUserToken } from "./utils/firebase.ts";
import path from "node:path";
import { getStartOfDay } from "./utils/time.ts";
import { getSessionLimitSeconds } from "./vm/utils.ts";
import { postgres, insertObject, upsertObject } from "./utils/postgres.ts";
import axios, { isAxiosError } from "axios";
import crypto from "node:crypto";
import { gzipSync } from "node:zlib";
import { resolveShard } from "./utils/resolveShard.ts";
import { makeRoomName, makeUserName } from "./utils/moniker.ts";
import { getStats } from "./utils/getStats.ts";
if (process.env.NODE_ENV === "development") {
axios.interceptors.request.use(
(config) => {
// console.log(config);
return config;
},
(error) => {
console.error(error);
},
);
}
const releaseInterval = 5 * 60 * 1000;
const app = express();
let server = null as https.Server | http.Server | null;
if (config.SSL_KEY_FILE && config.SSL_CRT_FILE) {
const key = fs.readFileSync(config.SSL_KEY_FILE);
const cert = fs.readFileSync(config.SSL_CRT_FILE);
server = https.createServer({ key: key, cert: cert }, app);
} else {
server = new http.Server(app);
}
server?.listen(config.PORT, config.HOST);
const io = new Server(server, { cors: {}, transports: ["websocket"] });
io.engine.use(async (req: any, res: Response, next: () => void) => {
const roomId = req._query.roomId;
if (!roomId) {
return next();
}
// Attempt to ensure the room being connected to is loaded in memory
// If it doesn't exist, we may fail later with "invalid namespace"
const shard = resolveShard(roomId);
const key = "/" + roomId;
// Check to make sure this shard should load this room
const isCorrectShard = !config.SHARD || shard === Number(config.SHARD);
// Get the room data from postgres
const persistedRoom = (
await postgres?.query<PersistentRoom>(
`SELECT * from room where "roomId" = $1`,
[key],
)
)?.rows?.[0];
// Don't await after this because we may have a race condition where 2 rquests both try to load the room
if (isCorrectShard && !rooms.has(key)) {
const data = persistedRoom?.data
? JSON.stringify(persistedRoom.data)
: undefined;
if (data) {
const room = new Room(io, key, data);
rooms.set(key, room);
console.log(
"loading room %s into memory on shard %s",
roomId,
config.SHARD,
);
}
}
next();
});
const rooms = new Map<string, Room>();
// Following functions iterate over in-memory rooms
setInterval(minuteMetrics, 60 * 1000);
setInterval(release, releaseInterval);
setInterval(saveRooms, 1000);
if (process.env.NODE_ENV === "development") {
try {
import("./vmWorker.ts");
// import('./syncSubs.ts');
// import('./timeSeries.ts');
} catch (e) {
console.error(e);
}
}
app.use(cors());
app.use(bodyParser.json());
app.use(bodyParser.raw({ type: "text/plain", limit: 1000000 }));
app.get("/ping", (_req, res) => {
res.json("pong");
});
// Data's already compressed so go before the compression middleware
app.get("/subtitle/:hash", async (req, res) => {
const key = "subtitle:" + req.params.hash;
const buf = await redis?.getBuffer(key);
if (!buf) {
res.status(404).end("not found");
return;
}
await redis?.expire(key, 24 * 60 * 60);
res.setHeader("Content-Encoding", "gzip");
res.end(buf);
});
app.use(compression());
app.post("/subtitle", async (req, res) => {
const data = req.body;
if (!redis) {
return;
}
// calculate hash, gzip and save to redis
const hash = crypto
.createHash("sha256")
.update(data, "utf8")
.digest()
.toString("hex");
let gzipData = gzipSync(data);
await redis.setex("subtitle:" + hash, 24 * 60 * 60, gzipData);
redisCount("subUploads");
res.json({ hash });
});
app.get("/downloadSubtitles", async (req, res) => {
// Request the URL from OS
try {
const urlResp = await axios<{ link: string }>({
url: "https://api.opensubtitles.com/api/v1/download",
method: "POST",
headers: {
"User-Agent": "watchparty v1",
"Api-Key": config.OPENSUBTITLES_KEY,
Accept: "application/json",
"Content-Type": "application/json",
// 'Authorization': 'Bearer ' + config.OPENSUBTITLES_KEY,
},
data: {
file_id: req.query.file_id,
// sub_format: 'srt',
},
});
redisCount("subDownloadsOS");
if (!redis) {
// Return the direct link to the user, will work for about 3 hours
res.json(urlResp.data);
return;
}
// Cache the contents in Redis (longer retention)
const subResp = await axios.get(urlResp.data.link, {
responseType: "arraybuffer",
});
const data = subResp.data;
const hash = crypto
.createHash("sha256")
.update(data, "utf8")
.digest()
.toString("hex");
let gzipData = gzipSync(data);
await redis.setex("subtitle:" + hash, 24 * 60 * 60, gzipData);
res.json({ link: "/subtitle/" + hash });
} catch (e) {
if (isAxiosError(e)) {
console.log(e.response);
}
throw e;
}
});
app.get("/searchSubtitles", async (req, res) => {
try {
const title = req.query.title ? String(req.query.title) : "";
const url = req.query.url ? String(req.query.url) : "";
let subUrl = "";
if (url) {
const startResp = await axios({
method: "get",
url: url,
headers: {
Range: "bytes=0-65535",
},
responseType: "arraybuffer",
});
const start = startResp.data;
const size = Number(startResp.headers["content-range"].split("/")[1]);
const endResp = await axios({
method: "get",
url: url,
headers: {
Range: `bytes=${size - 65536}-`,
},
responseType: "arraybuffer",
});
const end = endResp.data;
// console.log(start, end, size);
let hash = computeOpenSubtitlesHash(start, end, size);
// hash = 'f65334e75574f00f';
// Search API for subtitles by hash
subUrl = `https://api.opensubtitles.com/api/v1/subtitles?moviehash=${hash}&languages=en`;
} else if (title) {
subUrl = `https://api.opensubtitles.com/api/v1/subtitles?query=${title}&languages=en`;
}
// Alternative, web client calls this to get back some JS with the download URL embedded
// https://www.opensubtitles.com/nocache/download/7585196/subreq.js?file_name=Borgen.S04E01.en&locale=en&np=true&sub_frmt=srt&subtitle_id=6615808&ext_installed=false
// Up to 10 downloads per IP per day, but proxyable and doesn't require key
const response = await axios.get(subUrl, {
headers: {
"User-Agent": "watchparty v1",
"Api-Key": config.OPENSUBTITLES_KEY,
},
});
// console.log(subUrl, response.data);
const subtitles = response.data;
res.json(subtitles.data);
} catch (e: any) {
console.error(e.message);
res.json([]);
}
redisCount("subSearchesOS");
});
app.get("/stats", async (req, res) => {
if (req.query.key && req.query.key === config.STATS_KEY) {
const stats = await getStats();
res.json(stats);
} else {
res.status(403).json({ error: "Access Denied" });
}
});
app.get("/health/:metric", async (req, res) => {
const vmManagerStats = (
await axios.get("http://localhost:" + config.VMWORKER_PORT + "/stats")
).data;
const result = vmManagerStats[req.params.metric]?.availableVBrowsers?.length;
res.status(result ? 200 : 500).json(result);
});
app.get("/timeSeries", async (req, res) => {
if (req.query.key && req.query.key === config.STATS_KEY && redis) {
const timeSeriesData = await redis.lrange("timeSeries", 0, -1);
const timeSeries = timeSeriesData.map((entry) => JSON.parse(entry));
res.json(timeSeries);
} else {
res.status(403).json({ error: "Access Denied" });
}
});
app.get("/youtube", async (req, res) => {
if (typeof req.query.q === "string") {
try {
redisCount("youtubeSearch");
const items = await searchYoutube(req.query.q);
res.json(items);
} catch {
res.status(500).json({ error: "youtube error" });
}
} else {
res.status(500).json({ error: "query must be a string" });
}
});
app.get("/youtubePlaylist/:playlistId", async (req, res) => {
try {
const items = await youtubePlaylist(req.params.playlistId);
res.json(items);
} catch {
res.status(500).json({ error: "youtube error" });
}
});
app.post("/createRoom", async (req, res) => {
const genName = () => "/" + makeRoomName(config.SHARD);
let name = genName();
console.log("createRoom: ", name);
const newRoom = new Room(io, name);
if (postgres) {
const now = new Date();
const roomObj = {
roomId: newRoom.roomId,
lastUpdateTime: now,
creationTime: now,
};
try {
await insertObject(postgres, "room", roomObj);
} catch (e) {
redisCount("createRoomError");
throw e;
}
}
const decoded = await validateUserToken(req.body?.uid, req.body?.token);
newRoom.creator = decoded?.email;
const preload = (req.body?.video || "").slice(0, 20000);
if (preload) {
redisCount("createRoomPreload");
newRoom.video = preload;
newRoom.paused = true;
await newRoom.saveRoom();
}
const prePlaylist = Array.isArray(req.body?.playlist) && req.body?.playlist;
if (prePlaylist) {
for (let item of req.body.playlist) {
newRoom.playlistAdd(null, item);
}
}
rooms.set(name, newRoom);
res.json({ name });
});
app.post("/manageSub", async (req, res) => {
const decoded = await validateUserToken(
String(req.body?.uid),
String(req.body?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
if (!decoded.email) {
res.status(400).json({ error: "no email found" });
return;
}
const customer = await getCustomerByEmail(decoded.email);
if (!customer) {
res.status(400).json({ error: "customer not found" });
return;
}
const session = await createSelfServicePortal(
customer.id,
req.body?.return_url,
);
res.json(session);
});
app.delete("/deleteAccount", async (req, res) => {
// TODO pass this in req.query instead
const decoded = await validateUserToken(req.body?.uid, req.body?.token);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
if (postgres) {
// Delete rooms
await postgres.query("DELETE FROM room WHERE owner = $1", [decoded.uid]);
// Delete linked accounts
await postgres.query("DELETE FROM link_account WHERE uid = $1", [
decoded.uid,
]);
}
await deleteUser(decoded.uid);
redisCount("deleteAccount");
res.json({});
});
app.get("/metadata", async (req, res) => {
const decoded = await validateUserToken(
String(req.query?.uid),
String(req.query?.token),
);
let isSubscriber = await getIsSubscriberByEmail(decoded?.email);
// Has the user ever been a subscriber?
// const customer = await getCustomerByEmail(decoded.email);
let isFreePoolFull = false;
try {
isFreePoolFull = (
await axios.get(
"http://localhost:" + config.VMWORKER_PORT + "/isFreePoolFull",
)
).data.isFull;
} catch (e: any) {
console.warn("[WARNING]: free pool check failed: %s", e.code);
}
const beta =
decoded?.email != null &&
Boolean(config.BETA_USER_EMAILS.split(",").includes(decoded?.email));
const streamPath = beta ? config.STREAM_PATH : undefined;
const convertPath = isSubscriber ? config.CONVERT_PATH : undefined;
// log metrics but don't wait for it
if (postgres && decoded?.uid) {
upsertObject(
postgres,
"active_user",
{ uid: decoded?.uid, lastActiveTime: new Date() },
{ uid: true },
);
}
res.json({
isSubscriber,
isFreePoolFull,
beta,
streamPath,
convertPath,
});
});
app.get("/resolveRoom/:vanity", async (req, res) => {
const vanity = req.params.vanity;
const result = await postgres?.query(
`SELECT "roomId", vanity from room WHERE LOWER(vanity) = $1`,
[vanity?.toLowerCase() ?? ""],
);
// console.log(vanity, result.rows);
// We also use this for checking name availability, so just return null if it doesn't exist (http 200)
res.json(result?.rows[0] ?? null);
});
app.get("/roomData/:roomId", async (req, res) => {
// Returns the room data given a room ID
// Only return data if the room doesn't have a password
// If it does, we could accept it as a URL parameter but for now just don't support
const result = await postgres?.query(
`SELECT data from room WHERE "roomId" = $1 and password IS NULL`,
["/" + req.params.roomId],
);
res.json(result?.rows[0]?.data);
});
app.get("/resolveShard/:roomId", async (req, res) => {
const shardNum = resolveShard(req.params.roomId);
res.send(String(config.SHARD ? shardNum : ""));
});
app.get("/listRooms", async (req, res) => {
const decoded = await validateUserToken(
String(req.query?.uid),
String(req.query?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
const result = await postgres?.query<PersistentRoom>(
`SELECT "roomId", vanity, password from room WHERE owner = $1`,
[decoded.uid],
);
res.json(result?.rows ?? []);
});
app.delete("/deleteRoom", async (req, res) => {
const decoded = await validateUserToken(
String(req.query?.uid),
String(req.query?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
const result = await postgres?.query(
`DELETE from room WHERE owner = $1 and "roomId" = $2`,
[decoded.uid, req.query.roomId],
);
res.json(result?.rows);
});
app.get("/linkAccount", async (req, res) => {
const decoded = await validateUserToken(
String(req.query?.uid),
String(req.query?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
if (!postgres) {
res.status(400).json({ error: "invalid database client" });
return;
}
// Get the linked accounts for the user
let linkAccounts: LinkAccount[] = [];
if (decoded?.uid && postgres) {
const { rows } = await postgres.query(
"SELECT kind, accountid, accountname, discriminator FROM link_account WHERE uid = $1",
[decoded?.uid],
);
linkAccounts = rows;
}
res.json(linkAccounts);
});
app.post("/linkAccount", async (req, res) => {
const decoded = await validateUserToken(
String(req.body?.uid),
String(req.body?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
if (!postgres) {
res.status(400).json({ error: "invalid database client" });
return;
}
const kind = req.body?.kind;
if (kind === "discord") {
const tokenType = req.body?.tokenType;
const accessToken = req.body.accessToken;
// Get the token and verify the user
const response = await axios.get("https://discord.com/api/users/@me", {
headers: {
authorization: `${tokenType} ${accessToken}`,
},
});
const accountid = response.data.id;
const accountname = response.data.username;
const discriminator = response.data.discriminator;
// Store the user id, username, discriminator
await upsertObject(
postgres,
"link_account",
{
accountid: accountid,
accountname: accountname,
discriminator: discriminator,
uid: decoded.uid,
kind: kind,
},
{ uid: true, kind: true },
);
res.json({});
} else {
res.status(400).json({ error: "unsupported kind" });
}
});
app.delete("/linkAccount", async (req, res) => {
// TODO read from req.query instead
const decoded = await validateUserToken(
String(req.body?.uid),
String(req.body?.token),
);
if (!decoded) {
res.status(400).json({ error: "invalid user token" });
return;
}
if (!postgres) {
res.status(400).json({ error: "invalid database client" });
return;
}
await postgres.query(
"DELETE FROM link_account WHERE uid = $1 AND kind = $2",
[decoded.uid, req.body.kind],
);
res.json({});
});
app.get("/generateName", async (req, res) => {
res.send(makeUserName());
});
// Proxy video segments
app.get("/proxy/*splat", async (req, res) => {
redisCount("proxyReqs");
try {
const parsed = new URL("http://localhost" + req.url);
const pathname = parsed.pathname.slice("/proxy".length);
const host = parsed.searchParams.get("host");
if (pathname.endsWith("index-dvr.m3u8")) {
// VOD
// https://d2vjef5jvl6bfs.cloudfront.net/3012391a6c3e84c79ef6_gamesdonequick_41198403369_1681059003/chunked/index-dvr.m3u8
const resp = await axios.get("https://" + host + pathname);
const re2 = /(.*.ts)/g;
let repl = resp.data.replaceAll(re2, `$1?host=${host}`);
// Mark this as a VOD
repl += "#EXT-X-ENDLIST";
res.send(repl);
} else if (pathname.endsWith(".m3u8")) {
// Stream
// https://video-weaver.sea02.hls.ttvnw.net/v1/playlist/CrQEgv7Mz6nnsfJH3XtVQxeYXk8mViy1zNGWglcybvxZsI1rv3iLnjAnnqwCiVXCJ-DdD27J6RuFrLy7YUYwHUCKazIKICIupUCn9UXtaBYhBM5JIYqg9dz6NWYrCWU9HZJj2TGROv9mAOKuTR51YS82hdYL4PFZa3xxWXhgDsxXQHNDB03kY6S0aG0-EVva1xYrn5Ge6IAXRwug9QDGlb-ydtF3BtYppoTklVI7CVLySPPwbbt5Ow1JXdnKhLSwQEs4bh3BLwMnRBwUFI5nmE18BLYbkMOUivgYP5SSMgnGGlSkJO-iJNPWvepunEgyBUzB_7L-b1keTcV-Qak9IcWIITIWbRvmg6qB3ZSuWdcJgWKmdXdIn4qoRM4o16G1_0N_WRqPtMQFo0hmTlAVmHrzRArJQmaSgqAxZxRbFMd9RFeX6qjP9NtwguPbSeStdVbQxMNC34iavYUIxo8Ug812BHsG7J_kIlof2zkIqkEbP3oV3UkSByIo7xh9EEVargjaGDuQRt8zPQ6-fNBWJJe9F6IFu7lXBPIJ016lopyfcvTWjbLbBHsVkg6vG-3UISh0nud7KB5g5ipQePhtcFSI5hvjlfX1DAVHEpTWXkvlnL4wNqEqpBYL2btSXYeE1Cb-RAvrAT0s61usERcL2eI-S5aTcSO8_hxQ2afC7c9vlypOWgP6p6XNpViZHXmdXv4t-d68Z-MpLtSU7VbB3pRWnSswFFyA3W39ITic4lb97Djp3wHhGgz0Sy8aDb9r0tnphIYgASoJdXMtZWFzdC0yMKQG.m3u8
// Extract the edge URL host and add it to URL so proxy can fetch
const resp = await axios.get("https://" + host + pathname);
// const re = /https:\/\/(.*)\/v1\/segment\/(.*)/g;
// const match = re.exec(resp.data);
// const edgehost = match?.[1];
// const repl = resp.data.replaceAll(
// re,
// `/proxy/v1/segment/$2?host=${edgehost}`,
// );
const repl = resp.data;
res.send(repl);
} else if (pathname.endsWith(".ts")) {
// Segment
const resp = await axios.get("https://" + host + pathname, {
responseType: "arraybuffer",
});
res.writeHead(200, {
"Content-Type": "application/octet-stream",
"Accept-Ranges": "bytes",
"Content-Length": resp.data.length,
"Transfer-Encoding": "chunked",
});
res.write(resp.data);
res.end();
} else {
res.status(404);
res.end();
}
} catch (e) {
// console.log(e);
console.log("proxy failed: %s", req.url);
}
});
app.use(express.static(config.BUILD_DIRECTORY));
// Send index.html for all other requests (SPA)
app.use("/*splat", (_req, res) => {
res.sendFile(
path.resolve(
import.meta.dirname + `/../${config.BUILD_DIRECTORY}/index.html`,
),
);
});
async function saveRooms() {
// Unload rooms that are empty and idle
// Frees up some JS memory space when process is long-running
// On reconnect, we'll attempt to reload the room
let saveCount = 0;
let skipCount = 0;
const start = Date.now();
await Promise.all(
Array.from(rooms.entries()).map(async ([key, room]) => {
if (
room.roster.length === 0 &&
!room.vBrowser &&
Number(room.lastUpdateTime) < Date.now() - 8 * 60 * 60 * 1000
) {
console.log(
"freeing room %s from memory on shard %s",
key,
config.SHARD,
);
await room.saveRoom();
room.destroy();
rooms.delete(key);
saveCount += 1;
// Unregister the namespace to avoid dupes on reload
io._nsps.delete(key);
} else if (room.roster.length) {
room.lastUpdateTime = new Date();
await room.saveRoom();
saveCount += 1;
} else {
skipCount += 1;
}
}),
);
const end = Date.now();
console.log(
"[SAVEROOMS] %s saved in %sms, %s skipped",
saveCount,
end - start,
skipCount,
);
}
async function release() {
// Reset VMs in rooms that are:
// older than the session limit
// assigned to a room with no users
const roomArr = Array.from(rooms.values());
console.log("[RELEASE] %s rooms in batch", roomArr.length);
for (let room of roomArr) {
if (room.vBrowser && room.vBrowser.assignTime) {
const maxTime = getSessionLimitSeconds(room.vBrowser.large) * 1000;
const elapsed = Date.now() - room.vBrowser.assignTime;
const ttl = maxTime - elapsed;
const isTimedOut = ttl && ttl < releaseInterval;
const isAlmostTimedOut = ttl && ttl < releaseInterval * 2;
const isRoomEmpty = room.roster.length === 0;
const isRoomIdle =
Date.now() - Number(room.lastUpdateTime) > 5 * 60 * 1000;
if (isTimedOut || (isRoomEmpty && isRoomIdle)) {
console.log("[RELEASE] VM in room:", room.roomId);
room.stopVBrowserInternal();
if (isTimedOut) {
room.addChatMessage(null, {
id: "",
system: true,
cmd: "vBrowserTimeout",
msg: "",
});
redisCount("vBrowserTerminateTimeout");
} else if (isRoomEmpty) {
redisCount("vBrowserTerminateEmpty");
}
} else if (isAlmostTimedOut) {
room.addChatMessage(null, {
id: "",
system: true,
cmd: "vBrowserAlmostTimeout",
msg: "",
});
}
}
// We want to spread out the jobs over about half the release interval
// This gives other jobs some CPU time
const waitTime = releaseInterval / 2 / roomArr.length;
await new Promise((resolve) => setTimeout(resolve, waitTime));
}
}
async function minuteMetrics() {
const roomArr = Array.from(rooms.values());
let vbWaiting = 0;
for (let room of roomArr) {
if (room.vBrowser && room.vBrowser.id) {
// Update the heartbeat
await postgres?.query(
`UPDATE vbrowser SET "heartbeatTime" = NOW() WHERE "roomId" = $1 and vmid = $2`,
[room.roomId, room.vBrowser.id],
);
const expireTime = getStartOfDay() / 1000 + 86400;
if (room.vBrowser?.creatorClientID) {
await redis?.zincrby(
"vBrowserClientIDMinutes",
1,
room.vBrowser.creatorClientID,
);
await redis?.expireat("vBrowserClientIDMinutes", expireTime);
}
if (room.vBrowser?.creatorUID) {
await redis?.zincrby(
"vBrowserUIDMinutes",
1,
room.vBrowser?.creatorUID,
);
await redis?.expireat("vBrowserUIDMinutes", expireTime);
}
}
const users = room.roster.length;
if (users) {
await redis?.setex(`roomCounts:${room.roomId}`, 120, users);
await redis?.setex(
`roomRosters:${room.roomId}`,
120,
JSON.stringify(room.getRosterForStats()),
);
}
vbWaiting += room.vBrowserQueue ? 1 : 0;
}
// Report shard metrics
const obj: ShardMetric = {
uptime: process.uptime(),
mem: process.memoryUsage().rss,
roomCount: rooms.size,
users: io.engine.clientsCount,
vbWaiting,
};
await redis?.setex(
`shardMetrics:${config.SHARD ?? 0}`,
120,
JSON.stringify(obj),
);
}
function computeOpenSubtitlesHash(first: Buffer, last: Buffer, size: number) {
// console.log(first.length, last.length, size);
let temp = BigInt(size);
process(first);
process(last);
temp = temp & BigInt("0xffffffffffffffff");
return temp.toString(16).padStart(16, "0");
function process(chunk: Buffer) {
for (let i = 0; i < chunk.length; i += 8) {
const long = chunk.readBigUInt64LE(i);
temp += long;
}
}
}
server/syncSubs.ts
+156
View File
@@ -0,0 +1,156 @@
import config from "./config.ts";
import { getUserByEmail } from "./utils/firebase.ts";
import { insertObject, newPostgres, updateObject } from "./utils/postgres.ts";
import { getAllActiveSubscriptions, getAllCustomers } from "./utils/stripe.ts";
import { Client as DiscordClient, IntentsBitField } from "discord.js";
let lastSubs = "";
let currentSubs = "";
const postgres2 = newPostgres();
// set up the Discord admin bot
const discordBot = new DiscordClient({
intents: [IntentsBitField.Flags.Guilds, IntentsBitField.Flags.GuildMembers],
});
if (config.DISCORD_ADMIN_BOT_TOKEN) {
discordBot.login(config.DISCORD_ADMIN_BOT_TOKEN);
// discordBot.once('ready', () => {
// console.log(`Discord Bot "${discordBot?.user?.username}" ready`);
// });
}
if (process.env.NODE_ENV === "development") {
setTimeout(syncSubscribers, 1000);
}
setInterval(syncSubscribers, 60 * 1000);
async function syncSubscribers() {
if (!config.STRIPE_SECRET_KEY || !config.FIREBASE_ADMIN_SDK_CONFIG) {
return;
}
console.time("syncSubscribers");
// Fetch subs, customers from stripe
const [subs, customers] = await Promise.all([
getAllActiveSubscriptions(),
getAllCustomers(),
]);
const emailMap = new Map();
customers.forEach((cust) => {
emailMap.set(cust.id, cust.email);
});
console.log("%s subs in Stripe", subs.length);
const uidMap = new Map();
for (let i = 0; i < subs.length; i += 50) {
// Batch customers and fetch firebase data
const batch = subs.slice(i, i + 50);
const fbUsers = await Promise.all(
batch
.map((sub) =>
emailMap.get(sub.customer)
? getUserByEmail(emailMap.get(sub.customer))
: null,
)
.filter(Boolean),
);
fbUsers.forEach((user) => {
uidMap.set(user?.email, user?.uid);
});
}
let noUID = 0;
// Create sub objects
let result = subs
.map((sub) => {
let uid = uidMap.get(emailMap.get(sub.customer));
if (!uid) {
uid = emailMap.get(sub.customer);
noUID += 1;
}
return {
customerId: sub.customer,
email: emailMap.get(sub.customer),
status: sub.status,
uid,
};
})
.filter((sub) => sub.uid);
console.log("%s subs to insert", result.length);
console.log("%s subs do not have UID, using email", noUID);
const newResult = result.filter(
(sub, index) =>
index === result.findIndex((other) => sub.uid === other.uid),
);
console.log("%s deduped subs to insert", newResult.length);
if (result.length !== newResult.length) {
// Log the difference
console.log(result.filter((x) => !newResult.includes(x)));
}
result = newResult;
currentSubs = result
.map((sub) => sub.uid)
.sort()
.join();
// Upsert to DB
// console.log(result);
if (currentSubs !== lastSubs) {
try {
await postgres2?.query("BEGIN TRANSACTION");
await postgres2?.query("DELETE FROM subscriber");
await postgres2?.query('UPDATE room SET "isSubRoom" = false');
for (let row of result) {
await insertObject(postgres2, "subscriber", row);
await updateObject(
postgres2,
"room",
{ isSubRoom: true },
{ owner: row.uid },
);
}
await postgres2?.query("COMMIT");
lastSubs = currentSubs;
} catch (e) {
console.error(e);
await postgres2?.query("ROLLBACK");
}
}
if (
discordBot.isReady() &&
config.DISCORD_ADMIN_BOT_SERVER_ID &&
config.DISCORD_ADMIN_BOT_SUB_ROLE_ID
) {
console.log("setting discord roles");
// Update the sub status of users in Discord
// Join the current subs with linked accounts
const guild = discordBot.guilds.cache.get(
config.DISCORD_ADMIN_BOT_SERVER_ID,
);
const role = guild?.roles.cache.get(config.DISCORD_ADMIN_BOT_SUB_ROLE_ID);
const toUpdate = (
await postgres2.query(
`SELECT la.accountid from subscriber JOIN link_account la ON subscriber.uid = la.uid WHERE la.kind = 'discord'`,
)
).rows;
console.log("%s users to set sub role", toUpdate.length);
for (let row of toUpdate) {
try {
const user = await guild?.members.fetch(row.accountid);
if (user && role) {
console.log("assigning role %s to user %s", role, user.id);
await user.roles.add(role);
}
} catch (e: any) {
console.log(e.message);
}
}
}
console.timeEnd("syncSubscribers");
}
server/timeSeries.ts
+53
View File
@@ -0,0 +1,53 @@
import config from "./config.ts";
import axios from "axios";
import { redis } from "./utils/redis.ts";
import { getStats } from "./utils/getStats.ts";
statsTimeSeries();
setInterval(statsTimeSeries, 5 * 60 * 1000);
async function statsTimeSeries() {
if (redis) {
console.time("timeSeries");
try {
const stats = await getStats();
const isFreePoolFull = (
await axios.get(
"http://localhost:" + config.VMWORKER_PORT + "/isFreePoolFull",
)
).data.isFull;
const datapoint: AnyDict = {
time: new Date(),
currentUsers: stats.counts.currentUsers,
currentVBrowser: stats.counts.currentVBrowser,
currentVBrowserLarge: stats.counts.currentVBrowserLarge,
currentHttp: stats.counts.currentHttp,
currentScreenShare: stats.counts.currentScreenShare,
currentFileShare: stats.counts.currentFileShare,
currentVideoChat: stats.counts.currentVideoChat,
chatMessages: stats.counts.chatMessages,
redisUsage: stats.counts.redisUsage,
hetznerApiRemaining: stats.counts.hetznerApiRemaining,
avgStartMS:
(stats.vBrowserStartMS || []).map(Number).reduce((a, b) => a + b, 0) /
(stats.vBrowserStartMS?.length ?? 0),
vBrowserStarts: stats.counts.vBrowserStarts,
vBrowserLaunches: stats.counts.vBrowserLaunches,
vBrowserFails: stats.counts.vBrowserFails,
vBrowserStagingFails: stats.counts.vBrowserStagingFails,
isFreePoolFull: Number(isFreePoolFull),
};
Object.keys(stats.vmManagerStats).forEach((key) => {
if (stats.vmManagerStats[key]) {
datapoint[key] =
stats.vmManagerStats[key]?.availableVBrowsers?.length;
}
});
await redis.lpush("timeSeries", JSON.stringify(datapoint));
await redis.ltrim("timeSeries", 0, 288);
} catch (e: any) {
console.warn(`[TIMESERIES] %s when collecting stats`, e.code);
}
console.timeEnd("timeSeries");
}
}
server/tsconfig.json
+14
View File
@@ -0,0 +1,14 @@
{
"compilerOptions": {
"target": "esnext",
"module": "nodenext",
"strict": true,
"noEmit": true,
"allowJs": true,
"skipLibCheck": true,
"allowImportingTsExtensions": true,
"erasableSyntaxOnly": true,
"verbatimModuleSyntax": true
},
"include": [".", "../global.d.ts"]
}
server/utils/firebase.ts
+73
View File
@@ -0,0 +1,73 @@
import config from "../config.ts";
import admin from "firebase-admin";
if (config.FIREBASE_ADMIN_SDK_CONFIG) {
admin.initializeApp({
credential: admin.credential.cert(
JSON.parse(config.FIREBASE_ADMIN_SDK_CONFIG),
),
databaseURL: config.FIREBASE_DATABASE_URL,
});
}
export async function validateUserToken(uid: string, token: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return undefined;
}
if (!token) {
return undefined;
}
try {
const decoded = await admin.auth().verifyIdToken(token);
if (uid !== decoded.uid) {
// Valid but for wrong user
return undefined;
}
return decoded;
} catch (e) {
// Promise rejects if verification failed
console.log(e);
return undefined;
}
}
export async function writeData(key: string, value: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return;
}
await admin.database().ref(key).set(value);
}
export async function getUserByEmail(email: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return null;
}
try {
return await admin.auth().getUserByEmail(email);
} catch (e: any) {
console.log(email, e.message);
}
return null;
}
export async function getUser(uid: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return null;
}
return await admin.auth().getUser(uid);
}
export async function getUserEmail(uid: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return null;
}
const user = await admin.auth().getUser(uid);
return user.email;
}
export async function deleteUser(uid: string) {
if (!config.FIREBASE_ADMIN_SDK_CONFIG) {
return null;
}
return admin.auth().deleteUser(uid);
}
server/utils/getStats.ts
+312
View File
@@ -0,0 +1,312 @@
import type { AssignedVM } from "../vm/base.ts";
import { postgres } from "./postgres.ts";
import os from "node:os";
import { getRedisCountDay, getRedisCountDayDistinct, redis } from "./redis.ts";
import config from "../config.ts";
import { apps } from "../ecosystem.config.js";
export async function getStats() {
const now = Date.now();
let currentUsers = 0;
// Render each shard metrics as its own object
const shardMetrics: Record<string, ShardMetric> = {};
const shardKeys = new Set(
apps.map((app) => `shardMetrics:${app.env?.SHARD ?? 0}`),
);
for (let key of shardKeys) {
const resp2 = await redis?.get(key);
if (resp2) {
shardMetrics[key] = JSON.parse(resp2);
currentUsers += shardMetrics[key].users;
}
}
// Count these from postgres data
let currentHttp = 0;
let currentVBrowser = 0;
let currentVBrowserLarge = 0;
let currentScreenShare = 0;
let currentFileShare = 0;
const currentRoomSizes: Record<string, number> = {};
const result = await postgres?.query<{
roomId: string;
creationTime: Date;
lastUpdateTime: Date;
vanity: string;
isSubRoom: boolean;
roomTitle: string;
roomDescription: string;
mediaPath: string;
owner: string;
password: string;
video: string;
videoTS: number;
vBrowser: AssignedVM;
creator: string;
lock: string;
}>(
`SELECT "roomId", "creationTime", "lastUpdateTime", vanity, "isSubRoom", "roomTitle", "roomDescription", "mediaPath", owner, password,
data->'video' as video, data->'videoTS' as "videoTS", data->'vBrowser' as "vBrowser", data->'creator' as creator, data->'lock' as lock
FROM room
WHERE "lastUpdateTime" > NOW() - INTERVAL '7 day'
AND length(data->>'video') > 0
ORDER BY "creationTime" DESC`,
);
const currentRoomData = await Promise.all(
(result?.rows ?? []).map(async (dbRoom) => {
const vBrowser = dbRoom.vBrowser;
if (vBrowser) {
currentVBrowser += 1;
}
if (vBrowser?.large) {
currentVBrowserLarge += 1;
}
const rosterLength = Number(
await redis?.get(`roomCounts:${dbRoom.roomId}`),
);
let roster = [];
if (rosterLength) {
currentRoomSizes[rosterLength] =
(currentRoomSizes[rosterLength] ?? 0) + 1;
const resp = await redis?.get(`roomRosters:${dbRoom.roomId}`);
if (resp) {
roster = JSON.parse(resp);
}
}
const obj = {
roomId: dbRoom.roomId,
video: dbRoom.video || undefined,
videoTS: dbRoom.videoTS || undefined,
creationTime: dbRoom.creationTime || undefined,
lastUpdateTime: dbRoom.lastUpdateTime || undefined,
vanity: dbRoom.vanity || undefined,
isSubRoom: dbRoom.isSubRoom || undefined,
owner: dbRoom.owner || undefined,
password: dbRoom.password || undefined,
roomTitle: dbRoom.roomTitle || undefined,
roomDescription: dbRoom.roomDescription || undefined,
mediaPath: dbRoom.mediaPath || undefined,
vBrowser,
vBrowserElapsed: vBrowser?.assignTime && now - vBrowser?.assignTime,
lock: dbRoom.lock || undefined,
creator: dbRoom.creator || undefined,
rosterLength,
roster,
};
if (obj.video?.startsWith("http") && rosterLength) {
currentHttp += 1;
}
if (obj.video?.startsWith("screenshare://") && rosterLength) {
currentScreenShare += 1;
}
if (obj.video?.startsWith("fileshare://") && rosterLength) {
currentFileShare += 1;
}
return obj;
}),
);
// Singleton stats below (same for all shards)
const currentVideoChat = 0;
const cpuUsage = os.loadavg()[1] * 100;
const redisUsage = Number(
(await redis?.info())
?.split("\n")
.find((line) => line.startsWith("used_memory:"))
?.split(":")[1]
.trim(),
);
const postgresUsage = Number(
(await postgres?.query(`SELECT pg_database_size('postgres');`))?.rows[0]
.pg_database_size,
);
const numPermaRooms = Number(
(await postgres?.query("SELECT count(1) from room WHERE owner IS NOT NULL"))
?.rows[0].count,
);
const numAllRooms = Number(
(await postgres?.query("SELECT count(1) from room"))?.rows[0].count,
);
const numSubs = Number(
(await postgres?.query("SELECT count(1) from subscriber"))?.rows[0].count,
);
const discordBotWatch = await getRedisCountDay("discordBotWatch");
const createRoomErrors = await getRedisCountDay("createRoomError");
const deleteAccounts = await getRedisCountDay("deleteAccount");
const chatMessages = await getRedisCountDay("chatMessages");
const addReactions = await getRedisCountDay("addReaction");
const hetznerApiRemaining = Number(await redis?.get("hetznerApiRemaining"));
const vBrowserStarts = await getRedisCountDay("vBrowserStarts");
const vBrowserLaunches = await getRedisCountDay("vBrowserLaunches");
const vBrowserFails = await getRedisCountDay("vBrowserFails");
const vBrowserStagingFails = await getRedisCountDay("vBrowserStagingFails");
const vBrowserReimages = await getRedisCountDay("vBrowserReimage");
const vBrowserCleanups = await getRedisCountDay("vBrowserCleanup");
const vBrowserStopTimeout = await getRedisCountDay(
"vBrowserTerminateTimeout",
);
const vBrowserStopEmpty = await getRedisCountDay("vBrowserTerminateEmpty");
const vBrowserStopManual = await getRedisCountDay("vBrowserTerminateManual");
const vBrowserStartMS = await redis?.lrange("vBrowserStartMS", 0, -1);
const vBrowserStageRetries = await redis?.lrange(
"vBrowserStageRetries",
0,
-1,
);
const vBrowserStageFails = await redis?.lrange("vBrowserStageFails", 0, -1);
const vBrowserSessionMS = await redis?.lrange("vBrowserSessionMS", 0, -1);
// const vBrowserVMLifetime = await redis?.lrange('vBrowserVMLifetime', 0, -1);
const proxyReqs = await getRedisCountDay("proxyReqs");
const urlStarts = await getRedisCountDay("urlStarts");
const streamStarts = await getRedisCountDay("streamStarts");
const convertStarts = await getRedisCountDay("convertStarts");
const playlistAdds = await getRedisCountDay("playlistAdds");
const screenShareStarts = await getRedisCountDay("screenShareStarts");
const fileShareStarts = await getRedisCountDay("fileShareStarts");
const mediasoupStarts = await getRedisCountDay("mediasoupStarts");
const videoChatStarts = await getRedisCountDay("videoChatStarts");
const connectStarts = await getRedisCountDay("connectStarts");
const connectStartsDistinct = await getRedisCountDayDistinct(
"connectStartsDistinct",
);
const subUploads = await getRedisCountDay("subUploads");
const subDownloadsOS = await getRedisCountDay("subDownloadsOS");
const subSearchesOS = await getRedisCountDay("subSearchesOS");
const youtubeSearch = await getRedisCountDay("youtubeSearch");
const vBrowserClientIDsCard = await redis?.zcard("vBrowserClientIDs");
const vBrowserUIDsCard = await redis?.zcard("vBrowserUIDs");
const createRoomPreloads = await getRedisCountDay("createRoomPreload");
const vBrowserClientIDs = altArrayToObject(
await redis?.zrevrangebyscore(
"vBrowserClientIDs",
"+inf",
"0",
"WITHSCORES",
"LIMIT",
0,
20,
),
);
const vBrowserUIDs = altArrayToObject(
await redis?.zrevrangebyscore(
"vBrowserUIDs",
"+inf",
"0",
"WITHSCORES",
"LIMIT",
0,
20,
),
);
const vBrowserClientIDMinutes = altArrayToObject(
await redis?.zrevrangebyscore(
"vBrowserClientIDMinutes",
"+inf",
"0",
"WITHSCORES",
"LIMIT",
0,
20,
),
);
const vBrowserUIDMinutes = altArrayToObject(
await redis?.zrevrangebyscore(
"vBrowserUIDMinutes",
"+inf",
"0",
"WITHSCORES",
"LIMIT",
0,
20,
),
);
// Fetch VM stats from vmworker
const resp = await fetch(
"http://localhost:" + config.VMWORKER_PORT + "/stats",
);
const vmManagerStats = await resp.json();
return {
...shardMetrics,
currentRoomSizes,
counts: {
currentUsers,
currentVideoChat,
currentVBrowser,
currentVBrowserLarge,
currentHttp,
currentScreenShare,
currentFileShare,
cpuUsage,
redisUsage,
postgresUsage,
numPermaRooms,
numAllRooms,
numSubs,
discordBotWatch,
createRoomErrors,
createRoomPreloads,
deleteAccounts,
chatMessages,
addReactions,
proxyReqs,
urlStarts,
streamStarts,
convertStarts,
playlistAdds,
screenShareStarts,
fileShareStarts,
mediasoupStarts,
subUploads,
subDownloadsOS,
subSearchesOS,
youtubeSearch,
videoChatStarts,
connectStarts,
connectStartsDistinct,
hetznerApiRemaining,
vBrowserStarts,
vBrowserLaunches,
vBrowserFails,
vBrowserStagingFails,
vBrowserReimages,
vBrowserCleanups,
vBrowserStopManual,
vBrowserStopEmpty,
vBrowserStopTimeout,
vBrowserClientIDsCard,
vBrowserUIDsCard,
},
// Stats object from vmWorker (render as JSON)
vmManagerStats,
// Array of room data (render as JSON)
currentRoomData,
// Arrays of last values (render as one column table)
vBrowserStartMS,
vBrowserStageRetries,
vBrowserStageFails,
vBrowserSessionMS,
// Maps of vbrowser users
vBrowserClientIDs,
vBrowserClientIDMinutes,
vBrowserUIDs,
vBrowserUIDMinutes,
};
}
function altArrayToObject(arr: string[] | undefined) {
const result: Record<string, number> = {};
if (!arr) {
return result;
}
for (let i = 0; i < arr.length; i += 2) {
const k = arr[i];
const v = arr[i + 1];
result[k] = Number(v);
}
return result;
}
server/utils/moniker.ts
+41
View File
@@ -0,0 +1,41 @@
import fs from "node:fs";
import { resolveShard } from "./resolveShard.ts";
let adjectives = fs
.readFileSync(process.cwd() + "/words/adjectives.txt")
.toString()
.split("\n");
const nouns = fs
.readFileSync(process.cwd() + "/words/nouns.txt")
.toString()
.split("\n");
const verbs = fs
.readFileSync(process.cwd() + "/words/verbs.txt")
.toString()
.split("\n");
const randomElement = (array: string[]) =>
array[Math.floor(Math.random() * array.length)];
export function makeRoomName(shard: number | undefined) {
let filteredAdjectives = adjectives;
if (shard) {
// Filter the adjective list by shard
filteredAdjectives = adjectives.filter(
(adj) => resolveShard(adj) === Number(shard),
);
}
const adjective = randomElement(filteredAdjectives);
const noun = randomElement(nouns);
const verb = randomElement(verbs);
return `${adjective}-${noun}-${verb}`;
}
export function makeUserName() {
return `${capFirst(randomElement(adjectives))} ${capFirst(
randomElement(nouns),
)}`;
}
function capFirst(string: string) {
return string.charAt(0).toUpperCase() + string.slice(1);
}
server/utils/playlist.ts
+7
View File
@@ -0,0 +1,7 @@
export const findPlaylistVideoByUrl = (
playlist: PlaylistVideo[],
url?: string,
) => {
if (!url) return;
return playlist.find((video) => video.url === url);
};
server/utils/postgres.ts
+87
View File
@@ -0,0 +1,87 @@
import { Client, type QueryResult } from "pg";
import config from "../config.ts";
export let postgres: Client | undefined = undefined;
if (config.DATABASE_URL) {
postgres = new Client({
connectionString: config.DATABASE_URL,
ssl: { rejectUnauthorized: false },
});
postgres.connect();
}
/**
* Use this if we need a new connection instead of sharing.
* Guarantees we'll return a client because we throw if we don't have it configured
* @returns
*/
export function newPostgres() {
if (!config.DATABASE_URL) {
throw new Error("postgres not configured");
}
const postgres = new Client({
connectionString: config.DATABASE_URL,
ssl: { rejectUnauthorized: false },
});
postgres.connect();
return postgres;
}
export async function updateObject(
postgres: Client,
table: string,
object: AnyDict,
condition: AnyDict,
): Promise<QueryResult<any>> {
const columns = Object.keys(object);
const values = Object.values(object);
// TODO support compound conditions, not just one
let query = `UPDATE ${table} SET ${columns
.map((c, i) => `"${c}" = $${i + 1}`)
.join(",")}
WHERE "${Object.keys(condition)[0]}" = $${Object.keys(object).length + 1}
RETURNING *`;
//console.log(query);
const result = await postgres.query(query, [
...values,
condition[Object.keys(condition)[0]],
]);
return result;
}
export async function insertObject(
postgres: Client,
table: string,
object: AnyDict,
): Promise<QueryResult<any>> {
const columns = Object.keys(object);
const values = Object.values(object);
let query = `INSERT INTO ${table} (${columns.map((c) => `"${c}"`).join(",")})
VALUES (${values.map((_, i) => "$" + (i + 1)).join(",")})
RETURNING *`;
// console.log(query);
const result = await postgres.query(query, values);
return result;
}
export async function upsertObject(
postgres: Client,
table: string,
object: AnyDict,
conflict: BooleanDict,
): Promise<QueryResult<any>> {
const columns = Object.keys(object);
const values = Object.values(object);
let query = `INSERT INTO ${table} (${columns.map((c) => `"${c}"`).join(",")})
VALUES (${values.map((_, i) => "$" + (i + 1)).join(",")})
ON CONFLICT (${Object.keys(conflict)
.map((k) => `"${k}"`)
.join(",")})
DO UPDATE SET ${Object.keys(object)
.map((c) => `"${c}" = EXCLUDED."${c}"`)
.join(",")}
RETURNING *`;
// console.log(query);
const result = await postgres.query(query, values);
return result;
}
server/utils/redis.ts
+72
View File
@@ -0,0 +1,72 @@
import config from "../config.ts";
import { Redis } from "ioredis";
import { getStartOfHour } from "./time.ts";
export let redis: Redis | undefined = undefined;
if (config.REDIS_URL) {
redis = new Redis(config.REDIS_URL);
}
export async function redisCount(prefix: string) {
if (!redis) {
return;
}
try {
const key = `${prefix}:${getStartOfHour()}`;
await redis.incr(key);
await redis.expireat(key, getStartOfHour() + 86400 * 1000);
} catch (e) {
console.log(e);
}
}
export async function getRedisCountDay(prefix: string) {
if (!redis) {
return;
}
// Get counts for last 24 hour keys (including current partial hour)
const keyArr = [];
for (let i = 0; i < 24; i += 1) {
keyArr.push(`${prefix}:${getStartOfHour() - i * 3600 * 1000}`);
}
const values = await redis.mget(...keyArr);
return values.reduce((a, b) => (Number(a) || 0) + (Number(b) || 0), 0);
}
export async function getRedisCountHour(prefix: string) {
if (!redis) {
return;
}
// Get counts for previous full hour
const value = await redis.get(`${prefix}:${getStartOfHour() - 3600 * 1000}`);
return Number(value);
}
export async function redisCountDistinct(prefix: string, item: string) {
if (!redis) {
return;
}
const key = `${prefix}:${getStartOfHour()}`;
await redis.pfadd(key, item);
await redis.expireat(key, getStartOfHour() + 86400 * 1000);
}
export async function getRedisCountDayDistinct(prefix: string) {
if (!redis) {
return;
}
// Get counts for last 24 hour keys (including current partial hour)
const keyArr = [];
for (let i = 0; i < 24; i += 1) {
keyArr.push(`${prefix}:${getStartOfHour() - i * 3600 * 1000}`);
}
return await redis.pfcount(...keyArr);
}
export async function getRedisCountHourDistinct(prefix: string) {
if (!redis) {
return;
}
// Get counts for previous full hour
return await redis.pfcount(`${prefix}:${getStartOfHour() - 3600 * 1000}`);
}
server/utils/regex.ts
+11
View File
@@ -0,0 +1,11 @@
/** This regex searches for YouTube Video IDs from a YouTube URL */
/** example: */
/** YOUTUBE_VIDEO_ID_REGEX.exec('https://youtube.com/?v=14634524364) */
/** will return the id 14634524364 in the first exec group [1] */
export const YOUTUBE_VIDEO_ID_REGEX =
/(?:youtube\.com\/(?:[^\/]+\/.+\/|(?:v|e(?:mbed)?)\/|.*[?&]v=)|youtu\.be\/)([^"&?\/\s]{11})/;
/** These regexes allow us to find hours, minutes and seconds from a ISO 8601 time string */
export const PT_HOURS_REGEX = /(\d{1,2})H/;
export const PT_MINUTES_REGEX = /(\d{1,2})M/;
export const PT_SECONDS_REGEX = /(\d{1,2})S/;
server/utils/resolveShard.ts
+12
View File
@@ -0,0 +1,12 @@
import config from "../config.ts";
import { apps } from "../ecosystem.config.js";
export function resolveShard(roomId: string): number {
if (!config.SHARD) {
return 0;
}
const numShards = apps.filter((app) => app.env?.SHARD).length;
const letter = roomId[0];
const charCode = letter.charCodeAt(0);
return Number((charCode % numShards) + 1);
}
server/utils/stripe.ts
+61
View File
@@ -0,0 +1,61 @@
import config from "../config.ts";
import Stripe from "stripe";
const stripe = new Stripe(config.STRIPE_SECRET_KEY, {
apiVersion: "2020-08-27",
});
export async function getCustomerByEmail(email: string) {
if (!config.STRIPE_SECRET_KEY) {
return undefined;
}
const customer = await stripe.customers.list({
email,
expand: ["data.subscriptions"],
});
return customer?.data[0];
}
export async function getIsSubscriberByEmail(email: string | undefined) {
if (!config.STRIPE_SECRET_KEY) {
// If Stripe isn't set up assume everyone is a subscriber
return true;
}
if (!email) {
return false;
}
const customer = await getCustomerByEmail(email);
const isSubscriber = Boolean(
customer?.subscriptions?.data?.find((sub) => sub?.status === "active"),
);
return isSubscriber;
}
export async function createSelfServicePortal(
customerId: string,
returnUrl: string,
) {
return await stripe.billingPortal.sessions.create({
customer: customerId,
return_url: returnUrl,
});
}
export async function getAllCustomers() {
const result = [];
for await (const customer of stripe.customers.list({ limit: 100 })) {
result.push(customer);
}
return result;
}
export async function getAllActiveSubscriptions() {
const result = [];
for await (const sub of stripe.subscriptions.list({
limit: 100,
status: "active",
})) {
result.push(sub);
}
return result;
}
server/utils/time.ts
+14
View File
@@ -0,0 +1,14 @@
export function getStartOfDay() {
const now = Date.now();
return now - (now % 86400000);
}
export function getStartOfHour() {
const now = Date.now();
return now - (now % 3600000);
}
export function getStartOfMinute() {
const now = Date.now();
return now - (now % 60000);
}
server/utils/youtube.ts
+127
View File
@@ -0,0 +1,127 @@
import config from "../config.ts";
import {
PT_HOURS_REGEX,
PT_MINUTES_REGEX,
PT_SECONDS_REGEX,
YOUTUBE_VIDEO_ID_REGEX,
} from "./regex.ts";
import { youtube, youtube_v3 } from "@googleapis/youtube";
let Youtube = config.YOUTUBE_API_KEY
? youtube({
version: "v3",
auth: config.YOUTUBE_API_KEY,
})
: null;
export const mapYoutubeSearchResult = (
video: youtube_v3.Schema$SearchResult,
): PlaylistVideo => {
return {
channel: video.snippet?.channelTitle ?? "",
url: "https://www.youtube.com/watch?v=" + video?.id?.videoId,
name: video.snippet?.title ?? "",
img: video.snippet?.thumbnails?.default?.url ?? "",
duration: 0,
type: "youtube",
};
};
export const mapYoutubeListResult = (
video: youtube_v3.Schema$Video,
): PlaylistVideo => {
const videoId = video.id;
return {
url: "https://www.youtube.com/watch?v=" + videoId,
name: video.snippet?.title ?? "",
img: video.snippet?.thumbnails?.default?.url ?? "",
channel: video.snippet?.channelTitle ?? "",
duration: getVideoDuration(video.contentDetails?.duration ?? ""),
type: "youtube",
};
};
export const mapYoutubePlaylistResult = (
item: youtube_v3.Schema$PlaylistItem,
): PlaylistVideo => {
return {
url: "https://www.youtube.com/watch?v=" + item.snippet?.resourceId?.videoId,
name: item.snippet?.title ?? "",
img: item.snippet?.thumbnails?.default?.url ?? "",
channel: item.snippet?.channelTitle ?? "",
duration: 0,
// duration: getVideoDuration(video.contentDetails?.duration ?? ''),
type: "youtube",
};
};
export const searchYoutube = async (
query: string,
): Promise<PlaylistVideo[]> => {
const response = await Youtube?.search.list({
part: ["snippet"],
type: ["video"],
maxResults: 25,
q: query,
});
return response?.data?.items?.map(mapYoutubeSearchResult) ?? [];
};
export const youtubePlaylist = async (
playlistId: string,
): Promise<PlaylistVideo[]> => {
const response = await Youtube?.playlistItems.list({
part: ["snippet"],
playlistId,
maxResults: 100,
});
return response?.data?.items?.map(mapYoutubePlaylistResult) ?? [];
};
export const getYoutubeVideoID = (url: string) => {
const idParts = YOUTUBE_VIDEO_ID_REGEX.exec(url);
if (!idParts) {
return;
}
const id = idParts[1];
if (!id) {
return;
}
return id;
};
export const fetchYoutubeVideo = async (
id: string,
): Promise<PlaylistVideo | null> => {
const response = await Youtube?.videos.list({
part: ["snippet", "contentDetails"],
id: [id],
});
const top = response?.data?.items?.[0];
return top ? mapYoutubeListResult(top) : null;
};
export const getVideoDuration = (string: string): number => {
if (!string) {
return 0;
}
const hoursParts = PT_HOURS_REGEX.exec(string);
const minutesParts = PT_MINUTES_REGEX.exec(string);
const secondsParts = PT_SECONDS_REGEX.exec(string);
const hours = hoursParts ? parseInt(hoursParts[1]) : 0;
const minutes = minutesParts ? parseInt(minutesParts[1]) : 0;
const seconds = secondsParts ? parseInt(secondsParts[1]) : 0;
const totalSeconds = seconds + minutes * 60 + hours * 60 * 60;
return totalSeconds;
};
export const isYouTube = (input: string) => {
return (
input.startsWith("https://www.youtube.com/") ||
input.startsWith("https://youtu.be/")
);
};
server/vm/base.ts
+625
View File
@@ -0,0 +1,625 @@
import config from "../config.ts";
import axios from "axios";
import { redis, redisCount } from "../utils/redis.ts";
import { postgres as pg } from "../utils/postgres.ts";
import type { PoolConfig, PoolRegion } from "./utils.ts";
import type { Client } from "pg";
const incrInterval = 5 * 1000;
const decrInterval = 15 * 1000;
const cleanupInterval = 5 * 60 * 1000;
// If postgres isn't configured we can still run in stateless mode
// Only start/get/terminate can be used, otherwise exception will be thrown
const postgres = !pg ? (null as unknown as Client) : pg;
export abstract class VMManager {
protected isLarge = false;
protected region: PoolRegion = "US";
private limitSize = 0;
private minSize = 0;
protected hostname: string | undefined;
constructor({ isLarge, region, limitSize, minSize, hostname }: PoolConfig) {
this.isLarge = isLarge;
this.region = region;
this.limitSize = Number(limitSize) || 0;
this.minSize = Number(minSize) || 0;
this.hostname = hostname;
}
public getIsLarge = () => {
return this.isLarge;
};
public getRegion = () => {
return this.region;
};
public getMinSize = () => {
return this.minSize;
};
public getLimitSize = () => {
return this.limitSize;
};
public getTargetBuffer = () => {
let buffer = this.limitSize * 0.02;
// If ramping config, adjust buffer based on the hour
// During ramp down hours, keep a smaller buffer
// During ramp up hours, keep a larger buffer
// const rampDownHours = config.VM_POOL_RAMP_DOWN_HOURS.split(",").map(Number);
// const rampUpHours = config.VM_POOL_RAMP_UP_HOURS.split(",").map(Number);
// const nowHour = new Date().getUTCHours();
// const isRampDown =
// rampDownHours.length &&
// pointInInterval24(nowHour, rampDownHours[0], rampDownHours[1]);
// const isRampUp =
// rampUpHours.length &&
// pointInInterval24(nowHour, rampUpHours[0], rampUpHours[1]);
// if (isRampDown) {
// buffer *= 0.5;
// } else if (isRampUp) {
// buffer *= 1.5;
// }
return Math.ceil(buffer);
};
public getCurrentSize = async () => {
const { rows } = await postgres.query(
`SELECT count(1) FROM vbrowser WHERE pool = $1`,
[this.getPoolName()],
);
return Number(rows[0]?.count);
};
public getPoolName = () => {
return this.id + (this.isLarge ? "Large" : "") + this.region;
};
public getAvailableCount = async (): Promise<number> => {
const { rows } = await postgres.query(
`SELECT count(1) FROM vbrowser WHERE pool = $1 and state = 'available'`,
[this.getPoolName()],
);
return Number(rows[0]?.count);
};
public getStagingCount = async (): Promise<number> => {
const { rows } = await postgres.query(
`SELECT count(1) FROM vbrowser WHERE pool = $1 and state = 'staging'`,
[this.getPoolName()],
);
return Number(rows[0]?.count);
};
public getAvailableVBrowsers = async (): Promise<string[]> => {
const { rows } = await postgres.query(
`SELECT vmid from vbrowser WHERE pool = $1 and state = 'available'`,
[this.getPoolName()],
);
return rows.map((row: any) => row.vmid);
};
public getStagingVBrowsers = async (): Promise<string[]> => {
const { rows } = await postgres.query(
`SELECT vmid from vbrowser WHERE pool = $1 and state = 'staging'`,
[this.getPoolName()],
);
return rows.map((row: any) => row.vmid);
};
public getTag = () => {
return (
(config.VBROWSER_TAG || "vbrowser") +
this.region +
(this.isLarge ? "Large" : "")
);
};
public assignVM = async (
roomId: string,
uid: string,
): Promise<AssignedVM | undefined> => {
if (!roomId || !uid) {
return undefined;
}
// Update and use SKIP LOCKED to ensure each consumer gets a different one
const { rows } = await postgres.query(
`
UPDATE vbrowser
SET "roomId" = $1, uid = $2, "heartbeatTime" = NOW(), "assignTime" = NOW(), state = 'used'
WHERE id = (
SELECT id
FROM vbrowser
WHERE state = 'available'
AND pool = $3
ORDER BY "creationTime" DESC
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING data, pass`,
[roomId, uid, this.getPoolName()],
);
let vm: VM | undefined = rows[0]?.data;
let pass: string = rows[0]?.pass;
if (!vm) {
return;
}
return { ...vm, pass, assignTime: Date.now() };
};
public resetVM = async (vmid: string, roomId?: string): Promise<void> => {
if (roomId !== undefined) {
// verify the roomId matches if user initiated
const { rows } = await postgres.query(
`SELECT "roomId" FROM vbrowser WHERE pool = $1 AND vmid = $2`,
[this.getPoolName(), vmid],
);
if (rows[0]?.roomId && rows[0]?.roomId !== roomId) {
console.log(
"[RESET] %s: roomId mismatch on %s, expected %s, got %s",
this.getPoolName(),
vmid,
rows[0]?.roomId,
roomId,
);
return;
}
}
console.log("[RESET]", this.getPoolName(), vmid, roomId);
// We generally want to reuse if the provider has per-hour billing
// Since most user sessions are less than an hour
// Otherwise if it's per-second or Docker, it's easier to just terminate it on reboot
if (this.reuseVMs) {
// To reset without API calls/reimaging (logic can be reused on any cloud provider so we can avoid individual reboot implementations):
// Update vbrowser script to generate uuid password
// add admin endpoints to vbrowser to get password and reboot
// to reset, hit reboot endpoint with admin key OR call cloud reboot method
// in checkstaging, check password endpoint with admin key, update DB once available
const { rows } = await postgres.query(
`SELECT image FROM vbrowser WHERE pool = $1 AND vmid = $2`,
[this.getPoolName(), vmid],
);
const vmImageId = rows[0]?.image;
// Check if VM needs to be reimaged
if (this.imageId !== vmImageId) {
await this.reimageVM(vmid);
redisCount("vBrowserReimage");
// Update the vmImageId
await postgres.query(
`UPDATE vbrowser SET image = $3 WHERE pool = $1 AND vmid = $2`,
[this.getPoolName(), vmid, this.imageId],
);
} else {
await this.rebootVM(vmid);
}
// we could crash here and then row will remain in used state
// Once the heartbeat becomes stale cleanup will reset it again
const result = await postgres.query(
`
INSERT INTO vbrowser(pool, vmid, "creationTime", state)
VALUES($1, $2, NOW(), 'staging')
ON CONFLICT(pool, vmid) DO
UPDATE SET state = 'staging',
"roomId" = NULL, uid = NULL, retries = 0, "heartbeatTime" = NULL, "assignTime" = NULL, data = NULL
`,
[this.getPoolName(), vmid],
);
console.log("UPSERT", result.rowCount);
// Normally this should be an update, but we could insert if:
// if cleaning up a VM we didn't record in db on create
// if we resized down and deleted db row but didn't complete the termination
} else {
this.terminateVMWrapper(vmid);
}
};
public startVMWrapper = async () => {
// generate credentials and boot a VM
const password = crypto.randomUUID();
const id = await this.startVM(password);
// We might fail to record it if crashing here but cleanup will reset it
await postgres.query(
`
INSERT INTO vbrowser(pool, vmid, "creationTime", state, image)
VALUES($1, $2, NOW(), 'staging', $3)`,
[this.getPoolName(), id, this.imageId],
);
redisCount("vBrowserLaunches");
return id;
};
protected terminateVMWrapper = async (vmid: string) => {
console.log("[TERMINATE]", this.getPoolName(), vmid);
// Update the DB before calling terminate
// If we don't actually complete the termination, cleanup will reset it
const { command, rowCount } = await postgres.query(
`DELETE FROM vbrowser WHERE pool = $1 AND vmid = $2 RETURNING id`,
[this.getPoolName(), vmid],
);
console.log(command, rowCount);
// We can log the VM lifetime by returning the creationTime and diffing
await this.terminateVM(vmid);
};
checkVMReady = async (host: string) => {
// NOTE: This URL doesn't work for docker since it doesn't have /
// But since we don't reuse docker VMs it's ok
const url = "https://" + host.replace("/", "/health");
try {
// const out = execSync(`curl -i -L -v --ipv4 '${host}'`);
// if (!out.toString().startsWith('OK') && !out.toString().startsWith('404 page not found')) {
// throw new Error('mismatched response from health');
// }
const resp = await axios({
method: "GET",
url,
timeout: 1000,
});
// Check to make sure the VM was recently rebooted (we could also check on the password to ensure reset)
const timeSinceBoot = Date.now() / 1000 - Number(resp.data);
// console.log(timeSinceBoot);
return this.reuseVMs ? timeSinceBoot < 150 * 1000 : true;
} catch (e) {
// console.log(url, e.message, e.response?.status);
return false;
}
};
public runBackgroundJobs = async () => {
const resizeVMGroupIncr = async () => {
const availableCount = await this.getAvailableCount();
const stagingCount = await this.getStagingCount();
const currentSize = await this.getCurrentSize();
let launch = false;
launch =
availableCount + stagingCount < this.getTargetBuffer() &&
currentSize < (this.getLimitSize() || Infinity);
if (launch) {
console.log(
"[RESIZE-INCR]",
this.getPoolName(),
"target:",
this.getTargetBuffer(),
"available:",
availableCount,
"staging:",
stagingCount,
"currentSize:",
currentSize,
"limit:",
this.getLimitSize(),
);
try {
await this.startVMWrapper();
} catch (e: any) {
console.log(
e.response?.status,
JSON.stringify(e.response?.data),
e.config?.url,
e.config?.data,
);
}
}
};
const resizeVMGroupDecr = async () => {
const availableCount = await this.getAvailableCount();
const stagingCount = await this.getStagingCount();
let unlaunch = false;
unlaunch = availableCount + stagingCount > this.getTargetBuffer();
if (unlaunch) {
// use SKIP LOCKED to delete to avoid deleting VM that might be assigning
// filter to only VMs eligible for deletion
// they must be up for long enough
// keep the oldest min pool size number of VMs
// Hetzner/DO/Scaleway rounds up to nearest hour
let modulo = 3600;
const { rows } = await postgres.query(
`
DELETE FROM vbrowser
WHERE id = (
SELECT id
FROM vbrowser
WHERE pool = $1
AND state = 'available'
AND id >= (SELECT id from vbrowser WHERE pool = $1 ORDER BY id ASC LIMIT 1 OFFSET $2)
AND CAST(extract(epoch from now() - "creationTime") as INT) % $3 > $4
FOR UPDATE SKIP LOCKED
LIMIT 1
) RETURNING vmid`,
[
this.getPoolName(),
this.getMinSize(),
modulo,
config.VM_MIN_UPTIME_MINUTES * 60, // to seconds
],
);
const first = rows[0];
if (first) {
console.log("[RESIZE-DECR] %s: %s", this.getPoolName(), first.vmid);
await this.terminateVMWrapper(first.vmid);
}
}
};
const cleanupVMGroup = async () => {
// Reset hanging VMs
// It's possible we created a VM but lost track of it
// Take the list of VMs from API
// subtract VMs that have a heartbeat or available or staging
let allVMs = [];
try {
allVMs = await this.listVMs(this.getTag());
} catch (e) {
console.log(
"[CLEANUP] %s: failed to fetch VM list",
this.getPoolName(),
);
return;
}
const { rows } = await postgres.query(
`
SELECT vmid from vbrowser
WHERE pool = $1
AND
("heartbeatTime" > (NOW() - INTERVAL '5 minutes')
OR state = 'staging'
OR state = 'available')
`,
[this.getPoolName()],
);
const inUse = new Set(rows.map((row: any) => row.vmid));
console.log(
"[CLEANUP] %s: found %s VMs, %s to keep",
this.getPoolName(),
allVMs.length,
inUse.size,
);
for (let server of allVMs) {
if (!inUse.has(server.id)) {
redisCount("vBrowserCleanup");
console.log("[CLEANUP]", this.getPoolName(), server.id);
try {
await this.resetVM(server.id);
//this.terminateVMWrapper(server.id);
} catch (e: any) {
console.warn("[CLEANUP]", this.getPoolName(), e.response?.data);
}
await new Promise((resolve) => setTimeout(resolve, 2000));
}
}
};
const checkStaging = async () => {
// Increment retry count and return data
const { rows } = await postgres.query(
`
UPDATE vbrowser
SET retries = retries + 1
WHERE pool = $1 and state = 'staging'
RETURNING id, vmid, data, retries
`,
[this.getPoolName()],
);
const stagingPromises = rows.map(async (row: any): Promise<string> => {
const rowid = row.id;
const vmid: string = row.vmid;
const retryCount: number = row.retries;
let vm: VM | null = row.data;
if (retryCount < this.minRetries) {
if (config.NODE_ENV === "development") {
console.log(
"[CHECKSTAGING] %s: [vmid: %s] [attempt: %s] waiting for minRetries",
this.getPoolName(),
vmid,
retryCount,
);
}
// Do a minimum # of retries to give reboot time
return [vmid, retryCount, false].join(",");
}
// Fetch data on first attempt
// Try again only every once in a while to reduce load on API
const shouldFetchVM =
retryCount === this.minRetries + 1 || retryCount % 30 === 0;
// Refetch the VM every once in a while even if cached to check if it should be deleted
if (shouldFetchVM) {
try {
vm = await this.getVM(vmid);
} catch (e: any) {
console.warn(e.response?.data);
if (e.response?.status === 404) {
// Remove the VM because the provider says it doesn't exist
await postgres.query("DELETE FROM vbrowser WHERE id = $1", [
rowid,
]);
throw new Error("failed to find vm " + vmid);
}
}
if (vm?.host) {
// Save the VM data
await postgres.query(
`UPDATE vbrowser SET data = $1 WHERE id = $2`,
[vm, rowid],
);
}
}
if (!vm?.host) {
console.log(
"[CHECKSTAGING] %s: no host for vm %s",
this.getPoolName(),
vmid,
);
throw new Error("no host for vm " + vmid);
}
if (retryCount % 150 === 0) {
console.log(
"[CHECKSTAGING] %s: %s poweron, attach to network",
this.getPoolName(),
vmid,
);
this.powerOn(vmid);
//this.attachToNetwork(vmid);
}
if (retryCount % 180 === 0) {
console.log("[CHECKSTAGING]", this.getPoolName(), "giving up:", vmid);
redisCount("vBrowserStagingFails");
await redis?.lpush("vBrowserStageFails", vmid);
await redis?.ltrim("vBrowserStageFails", 0, 19);
// VM didn't come up. set image to null so we reimage
await postgres.query(
`UPDATE vbrowser SET image = NULL WHERE pool = $1 AND vmid = $2`,
[this.getPoolName(), vmid],
);
await this.resetVM(vmid);
}
if (retryCount >= 180) {
throw new Error("too many attempts on vm " + vmid);
}
const ready = await this.checkVMReady(vm.host);
if (
ready ||
retryCount % (config.NODE_ENV === "development" ? 1 : 30) === 0
) {
console.log(
"[CHECKSTAGING] %s: [ready: %s] [vmid: %s] [retries: %s] [host: %s]",
this.getPoolName(),
ready,
vmid,
retryCount,
vm?.host,
);
}
if (ready) {
const passUrl =
"https://" +
(vm.host.includes("/")
? vm.host.replace("/", "/password")
: vm.host + "/password") +
(vm.host.includes("?") ? "&" : "?") +
"key=" +
config.VBROWSER_ADMIN_KEY;
// console.log(passUrl);
const resp2 = await axios({
method: "GET",
url: passUrl,
timeout: 1000,
});
const pass = resp2.data;
// password is a uuid so it should never match the previous value, if it does we probably didn't reset properly
const { rows } = await postgres.query(
`UPDATE vbrowser SET state = 'available', pass = $2 WHERE id = $1 AND pass IS DISTINCT FROM $2`,
[rowid, pass],
);
// console.log(rows);
await redis?.lpush("vBrowserStageRetries", retryCount);
await redis?.ltrim("vBrowserStageRetries", 0, 19);
}
return [vmid, retryCount, ready].join(",");
});
// TODO log something if we timeout
const result = await Promise.race([
Promise.allSettled(stagingPromises),
new Promise((resolve) => setTimeout(resolve, 30000)),
]);
return result;
};
console.log("[VMWORKER] %s: starting background jobs", this.getPoolName());
setInterval(resizeVMGroupIncr, incrInterval);
setInterval(resizeVMGroupDecr, decrInterval);
setInterval(async () => {
console.log(
"[STATS] %s: currentSize %s, available %s, staging %s, target %s",
this.getPoolName(),
await this.getCurrentSize(),
await this.getAvailableCount(),
await this.getStagingCount(),
this.getTargetBuffer(),
);
}, 10000);
// The following may take a while per iteration
// Use while loop and delay between iterations rather than setInterval to avoid stacking requests
setImmediate(async () => {
while (true) {
console.time(this.getPoolName() + ":cleanup");
try {
await cleanupVMGroup();
} catch (e: any) {
console.warn(
"[CLEANUPVMGROUP-ERROR]",
this.getPoolName(),
e.response?.data,
);
}
console.timeEnd(this.getPoolName() + ":cleanup");
await new Promise((resolve) => setTimeout(resolve, cleanupInterval));
}
});
setImmediate(async () => {
while (true) {
// console.time(this.getPoolName() + ':checkstaging');
try {
await checkStaging();
} catch (e) {
console.warn("[CHECKSTAGING-ERROR]", this.getPoolName(), e);
}
// console.timeEnd(this.getPoolName() + ':checkstaging');
await new Promise((resolve) => setTimeout(resolve, 1000));
}
});
};
public abstract id: string;
protected abstract size: string;
protected abstract largeSize: string;
protected abstract minRetries: number;
protected abstract reuseVMs: boolean;
protected abstract imageId: string;
protected abstract startVM: (name: string) => Promise<string>;
protected abstract rebootVM: (id: string) => Promise<void>;
protected abstract reimageVM: (id: string) => Promise<void>;
protected abstract terminateVM: (id: string) => Promise<void>;
public abstract getVM: (id: string) => Promise<VM>;
protected abstract listVMs: (filter: string) => Promise<VM[]>;
protected abstract powerOn: (id: string) => Promise<void>;
protected abstract attachToNetwork: (id: string) => Promise<void>;
protected abstract mapServerObject: (server: any) => VM;
public abstract updateSnapshot: () => Promise<string>;
}
function pointInInterval24(x: number, a: number, b: number) {
return nonNegativeMod(x - a, 24) <= nonNegativeMod(b - a, 24);
}
function nonNegativeMod(n: number, m: number) {
return ((n % m) + m) % m;
}
export interface VM {
id: string;
host: string;
provider: string;
large: boolean;
region: string;
}
export interface AssignedVM extends VM {
pass: string;
assignTime: number;
controllerClient?: string;
creatorUID?: string;
creatorClientID?: string;
}
export interface VMManagers {
standard: VMManager | null;
large: VMManager | null;
US: VMManager | null;
}
server/vm/digitalocean.ts
+139
View File
@@ -0,0 +1,139 @@
import config from "../config.ts";
import axios from "axios";
import { VMManager, type VM } from "./base.ts";
const DO_TOKEN = config.DO_TOKEN;
const region = "sfo3";
const gatewayHost = config.DO_GATEWAY;
const sshKeys = config.DO_SSH_KEYS.split(",");
export class DigitalOcean extends VMManager {
size = "s-2vcpu-2gb"; // s-1vcpu-1gb, s-1vcpu-2gb, s-2vcpu-2gb, s-2vcpu-4gb, c-2, s-4vcpu-8gb
largeSize = "s-4vcpu-8gb";
minRetries = 5;
reuseVMs = true;
id = "DO";
imageId = config.DO_IMAGE;
startVM = async (name: string) => {
const response = await axios({
method: "POST",
url: `https://api.digitalocean.com/v2/droplets`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
data: {
name: name,
region: region,
size: this.isLarge ? this.largeSize : this.size,
image: Number(this.imageId),
ssh_keys: sshKeys,
private_networking: true,
// user_data: cloudInit(),
tags: [this.getTag()],
},
});
const id = response.data.droplet.id;
return id;
};
terminateVM = async (id: string) => {
const response = await axios({
method: "DELETE",
url: `https://api.digitalocean.com/v2/droplets/${id}`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
});
};
rebootVM = async (id: string) => {
// Reboot the VM
const response2 = await axios({
method: "POST",
url: `https://api.digitalocean.com/v2/droplets/${id}/actions`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
data: {
type: "reboot",
},
});
return;
};
reimageVM = async (id: string) => {
// Rebuild the VM
const response2 = await axios({
method: "POST",
url: `https://api.digitalocean.com/v2/droplets/${id}/actions`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
data: {
type: "rebuild",
image: Number(this.imageId),
},
});
return;
};
getVM = async (id: string) => {
const response = await axios({
method: "GET",
url: `https://api.digitalocean.com/v2/droplets/${id}`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
});
let server = this.mapServerObject(response.data.droplet);
return server;
};
listVMs = async (filter: string) => {
// console.log(filter, tags);
const response = await axios({
method: "GET",
url: `https://api.digitalocean.com/v2/droplets`,
headers: {
Authorization: "Bearer " + DO_TOKEN,
"Content-Type": "application/json",
},
params: {
// TODO need to update if over 100 results
per_page: 100,
tag_name: filter,
},
});
return response.data.droplets.map(this.mapServerObject);
};
powerOn = async (_id: string) => {};
attachToNetwork = async (_id: string) => {};
updateSnapshot = async () => {
return "";
};
mapServerObject = (server: any): VM => {
// const ip = server.networks.v4.find(
// (network: any) => network.type === 'private',
// )?.ip_address;
const ip = server.networks.v4.find(
(network: any) => network.type === "public",
)?.ip_address;
return {
id: server.id?.toString(),
// The gateway handles SSL termination and proxies to the private IP
host: ip ? `${gatewayHost}/?ip=${ip}` : "",
provider: this.id,
large: this.isLarge,
region: this.region,
};
};
}
server/vm/docker.ts
+134
View File
@@ -0,0 +1,134 @@
// This assumes an installation of Docker exists on the given host
// and that host is configured to accept our SSH key
import config from "../config.ts";
import { VMManager, type VM } from "./base.ts";
import { imageName } from "./utils.ts";
import fs from "node:fs";
import { homedir } from "node:os";
import { NodeSSH } from "node-ssh";
export class Docker extends VMManager {
// TODO support multiple Docker providers in the pool config with same region
size = "";
largeSize = "";
minRetries = 0;
reuseVMs = false;
id = "Docker";
ssh: NodeSSH | undefined = undefined;
imageId = imageName;
getSSH = async () => {
if (this.ssh && this.ssh.isConnected()) {
return this.ssh;
}
const sshConfig = {
username: config.DOCKER_VM_HOST_SSH_USER,
host: this.hostname,
// The private key the Docker host is configured to accept
privateKey: config.DOCKER_VM_HOST_SSH_KEY_BASE64
? Buffer.from(config.DOCKER_VM_HOST_SSH_KEY_BASE64, "base64").toString()
: fs.readFileSync(homedir() + "/.ssh/id_rsa").toString(),
};
this.ssh = new NodeSSH();
await this.ssh.connect(sshConfig);
return this.ssh;
};
startVM = async (name: string) => {
const tag = this.getTag();
const conn = await this.getSSH();
// If in development, have neko share the same SSL cert as the other services
// If in production, they are probably on different hosts and neko is behind a reverse proxy for SSL termination
const sslEnv =
config.NODE_ENV === "development" &&
config.SSL_KEY_FILE &&
config.SSL_CRT_FILE
? `-e NEKO_KEY="${config.SSL_KEY_FILE}" -e NEKO_CERT="${config.SSL_CRT_FILE}"`
: "";
const { stdout, stderr } = await conn.execCommand(
`
#!/bin/bash
set -e
PORT=$(comm -23 <(seq 5000 5063 | sort) <(ss -Htan | awk '{print $4}' | cut -d':' -f2 | sort -u) | sort -n | head -n 1)
INDEX=$(($PORT - 5000))
UDP_START=$((59000+$INDEX*100))
UDP_END=$((59099+$INDEX*100))
docker run -d --rm --name=${name} --memory="2g" --cpus="2" -p $PORT:$PORT -p $UDP_START-$UDP_END:$UDP_START-$UDP_END/udp -v /etc/letsencrypt:/etc/letsencrypt -l ${tag} -l index=$INDEX --log-opt max-size=1g --shm-size=1g --cap-add="SYS_ADMIN" ${sslEnv} -e DISPLAY=":99.0" -e NEKO_PASSWORD=${name} -e NEKO_PASSWORD_ADMIN=${name} -e NEKO_ADMIN_KEY=${config.VBROWSER_ADMIN_KEY} -e NEKO_BIND=":$PORT" -e NEKO_EPR=":$UDP_START-$UDP_END" -e NEKO_H264="1" ${imageName}
`,
);
console.log(stdout, stderr);
return stdout.trim();
};
terminateVM = async (id: string) => {
const conn = await this.getSSH();
const { stdout, stderr } = await conn.execCommand(`docker rm -fv ${id}`);
console.log(stdout, stderr);
return;
};
rebootVM = async (id: string) => {
// Docker containers aren't set to reuse, so do nothing (reset will terminate)
};
reimageVM = async (id: string) => {
const conn = await this.getSSH();
const { stdout, stderr } = await conn.execCommand(
`docker pull ${this.imageId}`,
);
console.log(stdout, stderr);
// The container is out of date. Delete it
this.terminateVMWrapper(id);
return;
};
getVM = async (id: string) => {
const conn = await this.getSSH();
const { stdout } = await conn.execCommand(`docker inspect ${id}`);
let data = null;
try {
data = JSON.parse(stdout)[0];
if (!data) {
throw new Error("no container with this ID found");
}
} catch {
console.warn(stdout);
throw new Error("failed to parse json");
}
let server = this.mapServerObject(data);
return server;
};
listVMs = async (filter: string) => {
const conn = await this.getSSH();
const listCmd = `docker inspect $(docker ps --filter label=${filter} --quiet --no-trunc)`;
const { stdout } = await conn.execCommand(listCmd);
if (!stdout) {
return [];
}
let data = [];
try {
data = JSON.parse(stdout);
} catch (e) {
console.warn(stdout);
throw new Error("failed to parse json");
}
return data.map(this.mapServerObject);
};
powerOn = async (id: string) => {};
attachToNetwork = async (id: string) => {};
updateSnapshot = async () => {
return "";
};
mapServerObject = (server: any): VM => ({
id: server.Id,
host: `${this.hostname}:${5000 + Number(server.Config?.Labels?.index)}`,
provider: this.id,
large: this.isLarge,
region: this.region,
});
}
server/vm/hetzner.ts
+258
View File
@@ -0,0 +1,258 @@
import config from "../config.ts";
import axios from "axios";
import { VMManager, type VM } from "./base.ts";
import fs from "node:fs";
import { redis } from "../utils/redis.ts";
const HETZNER_TOKEN = config.HETZNER_TOKEN;
const sshKeys = config.HETZNER_SSH_KEYS.split(",").map(Number);
export class Hetzner extends VMManager {
size = "cpx11"; // cpx11, cx22 (not available in US yet)
largeSize = "cpx31"; // cpx21, cpx31, ccx13, ccx23, cx32
minRetries = 5;
reuseVMs = true;
id = "Hetzner";
gateway = config.HETZNER_GATEWAY;
imageId = config.HETZNER_IMAGE;
private getRandomDatacenter() {
// US
let datacenters = ["ash"];
if (this.region === "USW") {
datacenters = ["hil"];
} else if (this.region === "EU") {
datacenters = ["nbg1", "fsn1", "hel1"];
}
return datacenters[Math.floor(Math.random() * datacenters.length)];
}
startVM = async (name: string) => {
const data = {
name: name,
server_type: this.isLarge ? this.largeSize : this.size,
start_after_create: true,
image: Number(this.imageId),
ssh_keys: sshKeys,
public_net: {
enable_ipv4: true,
enable_ipv6: false,
},
// networks: [
// this.networks[Math.floor(Math.random() * this.networks.length)],
// ],
// user_data: `replace with vbrowser.sh startup script if we want to boot vbrowser on instance creation (won't trigger on rebuild/restart)`
labels: {
[this.getTag()]: "1",
},
location: this.getRandomDatacenter(),
};
const response = await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"Content-Type": "application/json",
"User-Agent": "test",
},
data,
});
const id = response.data.server.id;
return id;
};
terminateVM = async (id: string) => {
await axios({
method: "DELETE",
url: `https://api.hetzner.cloud/v1/servers/${id}`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"User-Agent": "test",
},
});
};
rebootVM = async (id: string) => {
// Reboot the VM
await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers/${id}/actions/reboot`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"User-Agent": "test",
},
});
return;
};
reimageVM = async (id: string) => {
// Rebuild the VM
await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers/${id}/actions/rebuild`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"User-Agent": "test",
},
data: {
image: Number(this.imageId),
},
});
return;
};
getVM = async (id: string) => {
const response: any = await axios({
method: "GET",
url: `https://api.hetzner.cloud/v1/servers/${id}`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"User-Agent": "test",
},
});
console.log(
"[GETVM] %s: %s rate limit remaining",
id,
response?.headers["ratelimit-remaining"],
);
await redis?.set(
"hetznerApiRemaining",
response?.headers["ratelimit-remaining"],
);
const server = this.mapServerObject(response.data.server);
return server;
};
listVMs = async (filter: string) => {
const limit = this.getLimitSize();
const pageCount = Math.ceil((limit || 1) / 50);
const pages = Array.from(Array(pageCount).keys()).map((i) => i + 1);
const responses: any[] = await Promise.all(
pages.map((page) =>
axios({
method: "GET",
url: `https://api.hetzner.cloud/v1/servers`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"User-Agent": "test",
},
params: {
sort: "id:asc",
page,
per_page: 50,
label_selector: filter,
},
}),
),
);
const responsesMapped = responses.map((response) =>
response.data.servers.map(this.mapServerObject),
);
return responsesMapped.flat();
};
powerOn = async (id: string) => {
// Poweron the server (usually not needed)
try {
await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers/${id}/actions/poweron`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"Content-Type": "application/json",
"User-Agent": "test",
},
});
} catch (e) {
console.log("%s failed to poweron", id);
}
};
attachToNetwork = async (id: string) => {
// // Attach server to network (usually not needed)
// try {
// const response: any = await axios({
// method: 'GET',
// url: `https://api.hetzner.cloud/v1/servers/${id}`,
// headers: {
// Authorization: 'Bearer ' + HETZNER_TOKEN,
// },
// });
// if (response.data.server.private_net?.[0] == null) {
// await axios({
// method: 'POST',
// url: `https://api.hetzner.cloud/v1/servers/${id}/actions/attach_to_network`,
// headers: {
// Authorization: 'Bearer ' + HETZNER_TOKEN,
// 'Content-Type': 'application/json',
// },
// data: {
// network:
// this.networks[Math.floor(Math.random() * this.networks.length)],
// },
// });
// }
// } catch (e: any) {
// console.log('%s failed to attach to network', id);
// console.log(e.response?.data);
// }
};
updateSnapshot = async () => {
const response = await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"Content-Type": "application/json",
"User-Agent": "test",
},
data: {
name: "vBrowserSnapshot",
server_type: "cpx11",
start_after_create: true,
image: "docker-ce", // 15512617 for Ubuntu 20.04
ssh_keys: sshKeys,
user_data: fs
.readFileSync(import.meta.dirname + "/../../dev/vbrowser.sh")
.toString()
.replace("{VBROWSER_ADMIN_KEY}", config.VBROWSER_ADMIN_KEY),
location: this.getRandomDatacenter(),
},
});
const id = response.data.server.id;
await new Promise((resolve) => setTimeout(resolve, 4 * 60 * 1000));
// Validate snapshot server was created successfully
// const response3 = await axios(
// 'http://' + response.data.server.public_net?.ipv4?.ip + ':5000'
// );
const response2 = await axios({
method: "POST",
url: `https://api.hetzner.cloud/v1/servers/${id}/actions/create_image`,
headers: {
Authorization: "Bearer " + HETZNER_TOKEN,
"Content-Type": "application/json",
},
});
const imageId = response2.data.image.id;
await this.terminateVM(id);
return imageId;
};
mapServerObject = (server: any): VM => {
const public_ip = server.public_net?.ipv4?.ip;
// const private_ip = server.private_net?.[0]?.ip;
const ip = public_ip;
// We can use either the public or private IP for communicating between gateway and VM
// Only signaling traffic goes through here since the video is transmitted over WebRTC
// The private IP requires the server and gateway to be on the same network and there is a limit to the number of servers allowed
return {
id: server.id?.toString(),
// The gateway handles SSL termination and proxies to the private IP
host: ip ? `${this.gateway}/?ip=${ip}` : "",
provider: this.id,
large: this.isLarge,
region: this.region,
};
};
}
server/vm/scaleway.ts
+159
View File
@@ -0,0 +1,159 @@
import config from "../config.ts";
import axios from "axios";
import { VMManager, type VM } from "./base.ts";
const SCW_SECRET_KEY = config.SCW_SECRET_KEY;
const SCW_ORGANIZATION_ID = config.SCW_ORGANIZATION_ID;
const region = "nl-ams-1"; //fr-par-1
const gatewayHost = config.SCW_GATEWAY;
export class Scaleway extends VMManager {
size = "DEV1-S"; // DEV1-S, DEV1-M, DEV1-L, GP1-XS
largeSize = "DEV1-M";
minRetries = 5;
reuseVMs = true;
id = "Scaleway";
imageId = config.SCW_IMAGE;
startVM = async (name: string) => {
const response = await axios({
method: "POST",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
data: {
name: name,
dynamic_ip_required: true,
commercial_type: this.isLarge ? this.largeSize : this.size,
image: this.imageId,
volumes: {},
organization: SCW_ORGANIZATION_ID,
tags: [this.getTag()],
},
});
// console.log(response.data);
const id = response.data.server.id;
const response2 = await axios({
method: "PATCH",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers/${id}/user_data/cloud-init`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "text/plain",
},
// set userdata for boot action
//data: cloudInit(),
});
// console.log(response2.data);
// boot the instance
const response3 = await axios({
method: "POST",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers/${id}/action`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
data: {
action: "poweron",
},
});
// console.log(response3.data);
return id;
};
terminateVM = async (id: string) => {
const response = await axios({
method: "POST",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers/${id}/action`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
data: {
action: "terminate",
},
});
};
rebootVM = async (id: string) => {
// Reboot the VM (also destroys the Docker container since it has --rm flag)
const response2 = await axios({
method: "POST",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers/${id}/action`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
data: {
action: "reboot",
},
});
return;
};
reimageVM = async (id: string) => {
// Scaleway doesn't have a reimage/rebuild command. Delete the VM
await this.terminateVMWrapper(id);
};
getVM = async (id: string) => {
const response = await axios({
method: "GET",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers/${id}`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
});
let server = this.mapServerObject(response.data.server);
return server;
};
listVMs = async (filter: string) => {
const limit = this.getLimitSize();
const pageCount = Math.ceil((limit || 1) / 100);
const pages = Array.from(Array(pageCount).keys()).map((i) => i + 1);
const responses: any[] = await Promise.all(
pages.map((page) =>
axios({
method: "GET",
url: `https://api.scaleway.com/instance/v1/zones/${region}/servers`,
headers: {
"X-Auth-Token": SCW_SECRET_KEY,
"Content-Type": "application/json",
},
params: {
page,
per_page: 100,
tags: filter,
},
}),
),
);
const responsesMapped = responses.map((response) =>
response.data.servers.map(this.mapServerObject),
);
return responsesMapped.flat();
};
powerOn = async (_id: string) => {};
attachToNetwork = async (_id: string) => {};
updateSnapshot = async () => {
return "";
};
mapServerObject = (server: any): VM => {
// const ip = server.private_ip;
const ip = server.public_ip?.address;
return {
id: server.id,
// The gateway handles SSL termination and proxies to the private IP
host: ip ? `${gatewayHost}/?ip=${ip}` : "",
provider: this.id,
large: this.isLarge,
region: this.region,
};
};
}
server/vm/utils.ts
+86
View File
@@ -0,0 +1,86 @@
import { type AssignedVM, VMManager } from "./base.ts";
import config from "../config.ts";
import { Scaleway } from "./scaleway.ts";
import { Hetzner } from "./hetzner.ts";
import { DigitalOcean } from "./digitalocean.ts";
import { Docker } from "./docker.ts";
// Chromium on ARM: ghcr.io/howardchung/vbrowser/arm-chromium
export const imageName = "howardc93/vbrowser";
export type PoolRegion = "US" | "USW" | "EU";
export type PoolConfig = {
provider: string;
isLarge: boolean;
region: PoolRegion;
limitSize: number | undefined;
minSize: number | undefined;
hostname: string | undefined;
};
function createVMManager(poolConfig: PoolConfig): VMManager {
let vmManager: VMManager | null = null;
if (
config.SCW_SECRET_KEY &&
config.SCW_ORGANIZATION_ID &&
config.SCW_IMAGE &&
config.SCW_GATEWAY &&
poolConfig.provider === "Scaleway"
) {
vmManager = new Scaleway(poolConfig);
} else if (
config.HETZNER_TOKEN &&
config.HETZNER_IMAGE &&
config.HETZNER_GATEWAY &&
poolConfig.provider === "Hetzner"
) {
vmManager = new Hetzner(poolConfig);
} else if (
config.DO_TOKEN &&
config.DO_IMAGE &&
config.DO_GATEWAY &&
poolConfig.provider === "DO"
) {
vmManager = new DigitalOcean(poolConfig);
} else if (poolConfig.provider === "Docker") {
vmManager = new Docker(poolConfig);
}
if (!vmManager) {
throw new Error("failed to create vmManager");
}
return vmManager;
}
export function getVMManagerConfig(): PoolConfig[] {
return config.VM_MANAGER_CONFIG.split(",")
.filter(Boolean)
.map((c) => {
const split = c.split(":");
return {
provider: split[0],
isLarge: split[1] === "large",
region: split[2] as PoolRegion,
minSize: Number(split[3]),
limitSize: Number(split[4]),
hostname: split[5],
};
});
}
export function getBgVMManagers(): { [key: string]: VMManager } {
const result: { [key: string]: VMManager } = {};
const conf = getVMManagerConfig();
conf.forEach((c) => {
const mgr = createVMManager(c);
if (mgr) {
result[mgr.getPoolName()] = mgr;
}
});
return result;
}
export function getSessionLimitSeconds(isLarge: boolean) {
return isLarge
? config.VBROWSER_SESSION_SECONDS_LARGE
: config.VBROWSER_SESSION_SECONDS;
}
server/vmWorker.ts
+114
View File
@@ -0,0 +1,114 @@
import config from "./config.ts";
import { getBgVMManagers } from "./vm/utils.ts";
import express from "express";
import bodyParser from "body-parser";
const app = express();
const vmManagers = getBgVMManagers();
app.use(bodyParser.json());
Object.values(vmManagers).forEach((manager) => {
manager?.runBackgroundJobs();
});
app.post("/assignVM", async (req, res) => {
try {
// Find a pool that matches the size and region requirements
const pools = Object.values(vmManagers).filter((mgr) => {
return (
mgr.getIsLarge() === Boolean(req.body.isLarge) &&
(mgr.getRegion() === req.body.region || !req.body.region)
);
});
let vm = null;
// Sequentially try each to give earlier pools preference
// We might want to add the ability to load balance as well by randomly selecting between pools with same priority
for (let pool of pools) {
console.log(
"try assignVM from pool:",
pool.getPoolName(),
req.body.roomId,
req.body.uid,
);
vm = await pool.assignVM(req.body.roomId, req.body.uid);
if (vm) {
res.json(vm);
return;
}
}
res.json(null);
} catch (e) {
console.warn(e);
res.status(500).end();
}
});
app.post("/releaseVM", async (req, res) => {
try {
const pool =
vmManagers[
req.body.provider + (req.body.isLarge ? "Large" : "") + req.body.region
];
if (req.body.id) {
await pool?.resetVM(req.body.id, req.body.roomId);
}
res.end();
} catch (e) {
console.warn(e);
res.status(500).end();
}
});
app.get("/stats", async (req, res) => {
const vmManagerStats: AnyDict = {};
for (let [key, vmManager] of Object.entries(vmManagers)) {
const availableVBrowsers = await vmManager?.getAvailableVBrowsers();
const stagingVBrowsers = await vmManager?.getStagingVBrowsers();
const size = await vmManager?.getCurrentSize();
if (key && vmManager) {
vmManagerStats[key] = {
availableVBrowsers,
stagingVBrowsers,
bufferSize: vmManager?.getTargetBuffer(),
// terminationVBrowsers,
size,
};
}
}
res.json(vmManagerStats);
});
app.get("/isFreePoolFull", async (req, res) => {
const freePools = Object.values(vmManagers).filter((mgr) => {
return mgr?.getIsLarge() === false && mgr?.getLimitSize() > 0;
});
const fullResult = await Promise.all<Boolean>(
freePools.map(async (freePool) => {
let isFull = false;
if (freePool) {
const availableCount = await freePool.getAvailableCount();
const limitSize = freePool?.getLimitSize() ?? 0;
const currentSize = await freePool.getCurrentSize();
isFull = Boolean(
limitSize > 0 &&
(Number(availableCount) === 0 ||
Number(currentSize) - Number(availableCount) > limitSize * 0.95),
);
}
return isFull;
}),
);
const isFull = freePools.length && fullResult.every(Boolean);
res.json({ isFull });
});
app.post("/updateSnapshot", async (req, res) => {
const pool = vmManagers[req.body.provider + req.body.region];
const result = await pool?.updateSnapshot();
res.send(result?.toString() + "\n");
});
app.listen(config.VMWORKER_PORT, () => {
console.log("vmWorker listening on %s", config.VMWORKER_PORT);
});