Shortcuts migration, performance fix

apps/backend/openapi.yaml

@@ -17,7 +17,7 @@ tags:
   - name: news
   - name: notifications
   - name: pageConfig
-  - name: search
+  - name: shortcuts
   - name: sessions
   - name: test
   - name: wallpapers
@@ -392,11 +392,11 @@ paths:
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
-  /jobs/searchItems:
+  /jobs/shortcuts:
     get:
       tags:
         - jobs
-      summary: Search items job
+      summary: Shortcuts indexing job
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
@@ -824,11 +824,11 @@ paths:
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
-  /searchItems:
+  /shortcuts:
     get:
       tags:
-        - search
-      summary: Search items
+        - shortcuts
+      summary: List shortcuts
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
@@ -1370,19 +1370,57 @@ paths:
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
-  /searchItems/frequentlyUsed:
+  /shortcuts/apps:
+    post:
+      tags:
+        - shortcuts
+      summary: Create an on-demand shortcut app
+      requestBody:
+        $ref: "#/components/requestBodies/JsonBody"
+      responses:
+        "200":
+          $ref: "#/components/responses/JsonOk"
+        "400":
+          $ref: "#/components/responses/JsonBadRequest"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+  /shortcuts/on-demand/{appId}:
+    put:
+      tags:
+        - shortcuts
+      summary: Replace an on-demand app's shortcuts
+      parameters:
+        - name: appId
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        $ref: "#/components/requestBodies/JsonBody"
+      responses:
+        "200":
+          $ref: "#/components/responses/JsonOk"
+        "400":
+          $ref: "#/components/responses/JsonBadRequest"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+        "404":
+          $ref: "#/components/responses/JsonNotFound"
+        "409":
+          $ref: "#/components/responses/JsonConflict"
+  /shortcuts/frequentlyUsed:
     get:
       tags:
-        - search
-      summary: List frequently used search items
+        - shortcuts
+      summary: List frequently used shortcuts
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
-  /searchItems/usageStats:
+  /shortcuts/usageStats:
     post:
       tags:
-        - search
-      summary: Log search item usage
+        - shortcuts
+      summary: Log shortcut usage
       requestBody:
         $ref: "#/components/requestBodies/JsonBody"
       responses:
@@ -1429,6 +1467,18 @@ components:
         application/json:
           schema:
             $ref: "#/components/schemas/Error"
+    JsonNotFound:
+      description: Not Found
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/Error"
+    JsonConflict:
+      description: Conflict
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/Error"
   securitySchemes:
     bearerAuth:
       type: http

apps/backend/src/index.ts

@@ -6,7 +6,11 @@ import { cors } from "hono/cors";
 import { Client as SshClient } from "ssh2";
 
 import { config } from "./lib/config";
-import { subscribeActivity } from "./lib/activity";
+import {
+  handleActivityMessage,
+  registerSessionConnection,
+  subscribeActivity,
+} from "./lib/activity";
 import { ensureSession } from "./lib/data/sessions";
 import { jobsApi, registerJobsCron } from "./jobs/index";
 import { startPocketbase } from "./pocketbase";
@@ -92,28 +96,51 @@ app.get("/health", (c) => c.json({ status: "ok" }));
 app.get("/api/v1/activity", upgradeWebSocket((c) => {
   let refreshTimer: ReturnType<typeof setInterval> | undefined;
   let unsubscribeActivity: (() => void) | undefined;
+  let unregisterSessionConnection: (() => void) | undefined;
+  let connectedUserId = "";
+  let connectedSessionId = "";
 
   return {
     async onOpen(_event, ws) {
       const token = c.req.query("token") || "";
       const sessionId = c.req.query("sessionId") || c.req.header("x-session-id") || null;
       try {
         const { userId, pb } = await requireAuth({ token, sessionId });
-        await ensureSession(pb, userId, sessionId, readSessionMetadata(c));
+        const session = await ensureSession(pb, userId, sessionId, readSessionMetadata(c));
+        if (!session) throw new Error("A valid session id is required");
+        connectedUserId = userId;
+        connectedSessionId = session.sessionId;
+        unregisterSessionConnection = registerSessionConnection(userId, session.sessionId, ws);
+        let calendarEvents: Array<Record<string, any>> = [];
+        let calendarRefreshedAt = 0;
+        let calendarRefresh: Promise<void> | null = null;
+        const refreshCalendarEvents = async () => {
+          if (Date.now() - calendarRefreshedAt < 5 * 60 * 1000) return;
+          if (calendarRefresh) return calendarRefresh;
+
+          calendarRefresh = (async () => {
+            const integrationResult = await listIntegrations(userId);
+            calendarEvents = (await Promise.all(
+              integrationResult.integrations
+                .filter((integration) => integration.type === "caldav")
+                .map((integration) => getUpcomingEvents(
+                  integration.environment,
+                  integration.localData,
+                  (localData) => pb.collection("integrations").update(integration.id, { localData }).then(() => undefined),
+                ).then((events) => events.map((event) => ({ ...event, id: `${integration.id}:${event.id}` }))).catch(() => [])),
+            )).flat().filter((event) => new Date(event.start).getTime() >= new Date().setHours(0, 0, 0, 0));
+            calendarRefreshedAt = Date.now();
+          })().finally(() => {
+            calendarRefresh = null;
+          });
+
+          return calendarRefresh;
+        };
         const sendSnapshot = async () => {
-          const [notificationResult, integrationResult] = await Promise.all([
+          const [notificationResult] = await Promise.all([
             getNotifications(userId),
-            listIntegrations(userId),
+            refreshCalendarEvents(),
           ]);
-          const calendarEvents = (await Promise.all(
-            integrationResult.integrations
-              .filter((integration) => integration.type === "caldav")
-              .map((integration) => getUpcomingEvents(
-                integration.environment,
-                integration.localData,
-                (localData) => pb.collection("integrations").update(integration.id, { localData }).then(() => undefined),
-              ).then((events) => events.map((event) => ({ ...event, id: `${integration.id}:${event.id}` }))).catch(() => [])),
-          )).flat().filter((event) => new Date(event.start).getTime() >= new Date().setHours(0, 0, 0, 0));
           ws.send(JSON.stringify({ type: "activity:snapshot", notifications: notificationResult.items, calendarEvents }));
         };
 
@@ -128,6 +155,12 @@ app.get("/api/v1/activity", upgradeWebSocket((c) => {
     onMessage(event, ws) {
       try {
         const message = JSON.parse(String(event.data));
+        if (connectedUserId && connectedSessionId && handleActivityMessage(
+          connectedUserId,
+          connectedSessionId,
+          ws,
+          message,
+        )) return;
         if (message.type === "activity:subscribe" || message.type === "activity:refresh") {
           void (ws as typeof ws & { data?: { sendSnapshot: () => Promise<void> } }).data?.sendSnapshot();
         }
@@ -138,6 +171,7 @@ app.get("/api/v1/activity", upgradeWebSocket((c) => {
     onClose() {
       if (refreshTimer) clearInterval(refreshTimer);
       unsubscribeActivity?.();
+      unregisterSessionConnection?.();
     },
   };
 }));

apps/backend/src/jobs/index.ts

@@ -5,7 +5,7 @@ import { promisify } from "node:util";
 import { runJob } from "./job-logger";
 import { config } from "../lib/config";
 import { _d } from "../lib/sdk";
-import { runSearchItemsIndexing } from "./search-indexer";
+import { runShortcutsIndexing } from "./shortcuts-indexer";
 import indexStatusMonitoringJobs from "./monitoring/indexer";
 import {
   runStatusMonitoringJobs,
@@ -23,8 +23,8 @@ import { getSuperuserPB } from "../lib/pb/pocketbase";
 const execFileAsync = promisify(execFile);
 const logger = createLogger("Jobs");
 
-async function runSearchIndexerScript() {
-  await runSearchItemsIndexing();
+async function runShortcutsIndexerScript() {
+  await runShortcutsIndexing();
 }
 
 async function runPullIconsScript() {
@@ -34,11 +34,11 @@ async function runPullIconsScript() {
   });
 }
 
-const runSearchItemsJob = (source: string) =>
-  runJob("searchItemsIndexer", runSearchIndexerScript, {
+const runShortcutsJob = (source: string) =>
+  runJob("shortcutsIndexer", runShortcutsIndexerScript, {
     startMessage: `Triggered by ${source}`,
-    successMessage: "Search items indexing completed",
-    errorMessage: "Search items indexing failed",
+    successMessage: "Shortcuts indexing completed",
+    errorMessage: "Shortcuts indexing failed",
   });
 
 const runPullIconsJob = (source: string) =>
@@ -157,9 +157,9 @@ export function registerJobsCron() {
   logger.debug("Dashwise SDK app config", _d.getAppConfig());
 
   void runLegacyUserConfigsMigrationJob("server start");
-  void runSearchItemsJob("server start");
-  Bun.cron(config.SEARCHITEMS_SCHEDULE, async () => {
-    await runSearchItemsJob("cron schedule");
+  void runShortcutsJob("server start");
+  Bun.cron(config.SHORTCUTS_SCHEDULE, async () => {
+    await runShortcutsJob("cron schedule");
   });
 
   if (config.ENABLE_ICONS_REFRESH) {
@@ -199,7 +199,7 @@ export function registerJobsCron() {
 }
 
 export const jobsApi = {
-  runSearchItemsJob,
+  runShortcutsJob,
   runPullIconsJob,
   runMonitoringIndexerJob,
   runMonitoringRunnerJob,