Adds new tile on the admin home page with system data. (#301)

* Adds new tile on the admin home page with system data. Also fixes the active users bug in the pie chart

* Fixes missing parentheses

* Updates user stats cache when signing in

* Reads active number of users from session provider

* Removes unused variable

* Small improvements

* Removes acl properties from system data websocket and performs initial push of data

* fix: remove acl fetch

---------

Co-authored-by: DecDuck <declanahofmeyr@gmail.com>

server/api/v1/admin/system-data/ws.get.ts

@@ -0,0 +1,36 @@
+import systemManager from "~/server/internal/system-data";
+import aclManager from "~/server/internal/acls";
+import { logger } from "~/server/internal/logging";
+
+// TODO add web socket sessions for horizontal scaling
+// Peer ID to user ID
+const socketSessions = new Map<string, string>();
+
+export default defineWebSocketHandler({
+  async open(peer) {
+    const h3 = { headers: peer.request?.headers ?? new Headers() };
+    const userId = await aclManager.getUserIdACL(h3, ["system-data:listen"]);
+    if (!userId) {
+      peer.send("unauthenticated");
+      return;
+    }
+
+    socketSessions.set(peer.id, userId);
+
+    systemManager.listen(userId, peer.id, (systemData) => {
+      peer.send(JSON.stringify(systemData));
+    });
+  },
+
+  async close(peer, _details) {
+    const userId = socketSessions.get(peer.id);
+    if (!userId) {
+      logger.info(`skipping websocket close for ${peer.id}`);
+      return;
+    }
+
+    systemManager.unlisten(userId, peer.id);
+    systemManager.unlisten("system", peer.id); // In case we were listening as 'system'
+    socketSessions.delete(peer.id);
+  },
+});

server/api/v1/auth/odic/callback.get.ts

@@ -1,6 +1,7 @@
 import sessionHandler from "~/server/internal/session";
 import authManager from "~/server/internal/auth";
 import type { Session } from "~/server/internal/session/types";
+import userStatsManager from "~/server/internal/userstats";
 
 defineRouteMeta({
   openAPI: {
@@ -61,6 +62,7 @@ export default defineEventHandler(async (h3) => {
       `/auth/mfa?redirect=${result.options.redirect ? encodeURIComponent(result.options.redirect) : "/"}`,
     );
   }
+  await userStatsManager.cacheUserSessions();
 
   if (result.options.redirect) {
     return sendRedirect(h3, result.options.redirect);

server/internal/acls/descriptions.ts

@@ -43,6 +43,8 @@ export const userACLDescriptions: ObjectFromList<typeof userACLs> = {
   "emoji:read": "Read built in emojis",
 
   "settings:read": "Read system settings.",
+  "system-data:listen":
+    "Connect to a websocket to receive system data updates.",
 };
 
 export const systemACLDescriptions: ObjectFromList<typeof systemACLs> = {
@@ -108,4 +110,7 @@ export const systemACLDescriptions: ObjectFromList<typeof systemACLs> = {
 
   "depot:new": "Create a new download depot",
   "depot:delete": "Remove a download depot",
+
+  "system-data:listen":
+    "Connect to a websocket to receive system data updates.",
 };

server/internal/acls/index.ts

@@ -37,6 +37,8 @@ export const userACLs = [
   "news:read",
 
   "settings:read",
+
+  "system-data:listen",
 ] as const;
 const userACLPrefix = "user:";
 
@@ -100,6 +102,8 @@ export const systemACLs = [
   "maintenance:read",
 
   "settings:update",
+
+  "system-data:listen",
 ] as const;
 const systemACLPrefix = "system:";
 

server/internal/notifications/index.ts

@@ -56,7 +56,7 @@ class NotificationSystem {
       },
     });
     for (const notification of notifications) {
-      await listener.callback(notification);
+      listener.callback(notification);
     }
   }
 
@@ -69,7 +69,7 @@ class NotificationSystem {
         notification.acls.findIndex(
           (e) => listener.acls.findIndex((v) => v === e) != -1,
         ) != -1;
-      if (hasSome) await listener.callback(notification);
+      if (hasSome) listener.callback(notification);
     }
   }
 

server/internal/session/cache.ts

@@ -23,6 +23,11 @@ export default function createCacheSessionProvider() {
       const session = await sessions.get(token);
       return session ? (session as T) : undefined; // Ensure undefined is returned if session is not found
     },
+    async getNumberActiveSessions() {
+      const now = new Date();
+      const allSessions = await sessions.getItems(await sessions.getKeys());
+      return allSessions.filter(({ value }) => value.expiresAt > now).length;
+    },
     async updateSession(token, data) {
       return (await this.setSession(token, data)) !== undefined;
     },

server/internal/session/db.ts

@@ -1,3 +1,5 @@
+import { DateTime } from "luxon";
+
 import prisma from "../db/database";
 import type { SessionProvider, SessionWithToken } from "./types";
 import cacheHandler from "../cache";
@@ -76,6 +78,20 @@ export default function createDBSessionHandler(): SessionProvider {
         },
       });
     },
+    async getNumberActiveSessions() {
+      return (
+        (
+          await prisma.session.groupBy({
+            by: ["userId"],
+            where: {
+              expiresAt: {
+                gt: DateTime.now().toJSDate(),
+              },
+            },
+          })
+        ).length || 0
+      );
+    },
     async findSessions(options) {
       const search: SessionWhereInput[] = [];
       if (options.userId) {

server/internal/session/index.ts

@@ -10,8 +10,10 @@ import { parse as parseCookies } from "cookie-es";
 import type { MinimumRequestObject } from "~/server/h3";
 import type { DurationLike } from "luxon";
 import { DateTime } from "luxon";
-import createDBSessionHandler from "./db";
 import prisma from "../db/database";
+// import createMemorySessionHandler from "./memory";
+import createDBSessionHandler from "./db";
+// import createCacheSessionProvider from "./cache";
 
 /*
 This implementation may need work.
@@ -49,7 +51,7 @@ export class SessionHandler {
     // Create a new provider
     // this.sessionProvider = createCacheSessionProvider();
     this.sessionProvider = createDBSessionHandler();
-    // this.sessionProvider = createMemorySessionProvider();
+    // this.sessionProvider = createMemorySessionHandler();
   }
 
   async signin(
@@ -217,6 +219,10 @@ export class SessionHandler {
     return await this.sessionProvider.findSessions(terms);
   }
 
+  async getNumberActiveSessions() {
+    return this.sessionProvider.getNumberActiveSessions();
+  }
+
   /**
    * Update session info
    * @param token session token

server/internal/session/memory.ts

@@ -22,6 +22,15 @@ export default function createMemorySessionHandler() {
       sessions.delete(token);
       return true;
     },
+    async getNumberActiveSessions() {
+      let activeSessions = 0;
+      for (const [_key, session] of sessions) {
+        if (session.expiresAt.getDate() > Date.now()) {
+          activeSessions += 1;
+        }
+      }
+      return activeSessions;
+    },
     async cleanupSessions() {
       const now = new Date();
       for (const [token, session] of sessions) {

server/internal/session/types.d.ts

@@ -50,4 +50,5 @@ export interface SessionProvider {
   removeSession: (token: string) => Promise<boolean>;
   cleanupSessions: () => Promise<void>;
   findSessions: (options: SessionSearchTerms) => Promise<SessionWithToken[]>;
+  getNumberActiveSessions: () => Promise<number>;
 }

server/internal/system-data/index.ts

@@ -0,0 +1,58 @@
+import os from "os";
+
+export type SystemData = {
+  totalRam: number;
+  freeRam: number;
+  cpuLoad: number;
+  cpuCores: number;
+};
+
+class SystemManager {
+  // userId to acl to listenerId
+  private listeners = new Map<
+    string,
+    Map<string, { callback: (systemData: SystemData) => void }>
+  >();
+
+  listen(
+    userId: string,
+    id: string,
+    callback: (systemData: SystemData) => void,
+  ) {
+    if (!this.listeners.has(userId)) this.listeners.set(userId, new Map());
+    // eslint-disable-next-line @typescript-eslint/no-extra-non-null-assertion
+    this.listeners.get(userId)!!.set(id, { callback });
+    this.pushUpdate(userId, id);
+    setInterval(() => this.pushUpdate(userId, id), 3000);
+  }
+
+  unlisten(userId: string, id: string) {
+    this.listeners.get(userId)?.delete(id);
+  }
+
+  private async pushUpdate(userId: string, id: string) {
+    const listener = this.listeners.get(userId)?.get(id);
+    if (!listener) {
+      throw new Error("Failed to catch-up listener: callback does not exist");
+    }
+    listener.callback(this.getSystemData());
+  }
+
+  getSystemData(): SystemData {
+    return {
+      cpuLoad: this.cpuLoad(),
+      totalRam: os.totalmem(),
+      freeRam: os.freemem(),
+      cpuCores: os.cpus().length,
+    };
+  }
+
+  private cpuLoad() {
+    const [oneMinLoad, _fiveMinLoad, _fiftenMinLoad] = os.loadavg();
+    const numberCpus = os.cpus().length;
+    return 100 - ((numberCpus - oneMinLoad) / numberCpus) * 100;
+  }
+}
+
+export const systemManager = new SystemManager();
+export default systemManager;

server/internal/userstats/index.ts

@@ -4,25 +4,14 @@ Handles managing collections
 
 import cacheHandler from "../cache";
 import prisma from "../db/database";
-import { DateTime } from "luxon";
+import sessionHandler from "../session";
 
 class UserStatsManager {
   // Caches the user's core library
   private userStatsCache = cacheHandler.createCache<number>("userStats");
 
   async cacheUserSessions() {
-    const activeSessions =
-      (
-        await prisma.client.groupBy({
-          by: ["userId"],
-          where: {
-            id: { not: "system" },
-            lastConnected: {
-              gt: DateTime.now().minus({ months: 1 }).toISO(),
-            },
-          },
-        })
-      ).length || 0;
+    const activeSessions = await sessionHandler.getNumberActiveSessions();
     await this.userStatsCache.set("activeSessions", activeSessions);
   }