Add Dashwise Activity Layer

apps/backend/openapi.yaml

@@ -6,6 +6,7 @@ info:
 servers:
   - url: /api/v1
 tags:
+  - name: activities
   - name: app
   - name: auth
   - name: config
@@ -64,6 +65,83 @@ paths:
                   - userSignupDisabled
                 additionalProperties: false
   # END GENERATED SYSTEM CONTRACTS
+  /activities:
+    get:
+      tags:
+        - activities
+      summary: List the authenticated user's activities
+      parameters:
+        - name: page
+          in: query
+          schema: { type: integer, minimum: 1 }
+        - name: perPage
+          in: query
+          schema: { type: integer, minimum: 1, maximum: 100 }
+        - name: source
+          in: query
+          schema: { type: string, maxLength: 120 }
+        - name: type
+          in: query
+          schema: { type: string, maxLength: 120 }
+        - name: severity
+          in: query
+          schema: { type: string, enum: [info, success, warning, error] }
+        - name: from
+          in: query
+          schema: { type: string, format: date-time }
+        - name: to
+          in: query
+          schema: { type: string, format: date-time }
+      responses:
+        "200":
+          description: Activity page
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ActivityPage"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+  /activities/{id}:
+    get:
+      tags:
+        - activities
+      summary: Get an activity owned by the authenticated user
+      parameters:
+        - name: id
+          in: path
+          required: true
+          schema: { type: string }
+      responses:
+        "200":
+          description: Activity
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [activity]
+                properties:
+                  activity:
+                    $ref: "#/components/schemas/Activity"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+        "404":
+          $ref: "#/components/responses/JsonNotFound"
+    delete:
+      tags:
+        - activities
+      summary: Delete an activity owned by the authenticated user
+      parameters:
+        - name: id
+          in: path
+          required: true
+          schema: { type: string }
+      responses:
+        "200":
+          $ref: "#/components/responses/JsonOk"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+        "404":
+          $ref: "#/components/responses/JsonNotFound"
   /auth/callback:
     get:
       tags:
@@ -219,6 +297,57 @@ paths:
       responses:
         "200":
           $ref: "#/components/responses/JsonOk"
+  /integrations/{integrationId}/activities:
+    post:
+      tags:
+        - activities
+      summary: Publish an activity for an integration owned by the authenticated user
+      parameters:
+        - name: integrationId
+          in: path
+          required: true
+          schema: { type: string }
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [type, title]
+              properties:
+                type: { type: string, maxLength: 120 }
+                title: { type: string, maxLength: 200 }
+                description: { type: string, maxLength: 2000 }
+                severity: { type: string, enum: [info, success, warning, error] }
+                eventId: { type: string, maxLength: 200 }
+                idempotencyKey: { type: string, maxLength: 256 }
+                occurredAt: { type: string, format: date-time }
+                metadata: { type: object, maxProperties: 128, additionalProperties: true }
+                action:
+                  type: object
+                  required: [label, target]
+                  properties:
+                    label: { type: string, maxLength: 80 }
+                    target: { type: string, maxLength: 2048, description: Safe internal application path only. }
+      responses:
+        "201":
+          description: Activity published
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/PublishActivityResult"
+        "200":
+          description: Duplicate idempotent submission
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/PublishActivityResult"
+        "400":
+          $ref: "#/components/responses/JsonBadRequest"
+        "401":
+          $ref: "#/components/responses/JsonUnauthorized"
+        "404":
+          $ref: "#/components/responses/JsonNotFound"
   /integrations/test-endpoint:
     post:
       tags:
@@ -1473,6 +1602,47 @@ components:
       schema:
         type: string
   schemas:
+    Activity:
+      type: object
+      required: [id, ownerId, source, type, title, severity, occurredAt, createdAt]
+      properties:
+        id: { type: string }
+        ownerId: { type: string }
+        source: { type: string, maxLength: 120 }
+        sourceId: { type: string, maxLength: 200 }
+        eventId: { type: string, maxLength: 200 }
+        type: { type: string, maxLength: 120 }
+        title: { type: string, maxLength: 200 }
+        description: { type: string, maxLength: 2000 }
+        severity: { type: string, enum: [info, success, warning, error] }
+        occurredAt: { type: string, format: date-time }
+        createdAt: { type: string, format: date-time }
+        metadata: { type: object, additionalProperties: true }
+        action:
+          type: object
+          required: [label, target]
+          properties:
+            label: { type: string, maxLength: 80 }
+            target: { type: string, maxLength: 2048, description: Safe internal application path only. }
+    ActivityPage:
+      type: object
+      required: [page, perPage, totalItems, totalPages, items]
+      properties:
+        page: { type: integer }
+        perPage: { type: integer }
+        totalItems: { type: integer }
+        totalPages: { type: integer }
+        items:
+          type: array
+          items:
+            $ref: "#/components/schemas/Activity"
+    PublishActivityResult:
+      type: object
+      required: [activity, duplicate]
+      properties:
+        activity:
+          $ref: "#/components/schemas/Activity"
+        duplicate: { type: boolean }
     GenericObject:
       type: object
       additionalProperties: true

apps/backend/src/index.ts

@@ -20,6 +20,7 @@ import { getMonitoringSshHostById, getMonitoringSshHostCredentials, getSystemAge
 import { getNotifications } from "./lib/data/notifications/items";
 import { listIntegrations } from "./lib/data/integrations";
 import { getUpcomingEvents } from "./lib/calendar";
+import { listActivities } from "./lib/data/activities";
 import { systemAgentClient } from "./lib/systemAgent";
 import { requireAuth } from "./routes/shared";
 import authRoute from "./routes/auth.route";
@@ -154,11 +155,17 @@ app.get("/api/v1/activity",  upgradeWebSocket((c) => {
           return calendarRefresh;
         };
         const sendSnapshot = async () => {
-          const [notificationResult] = await Promise.all([
+          const [notificationResult, activityResult] = await Promise.all([
             getNotifications(userId),
+            listActivities(userId, { page: 1, perPage: 20 }),
             refreshCalendarEvents(),
           ]);
-          ws.send(JSON.stringify({ type: "activity:snapshot", notifications: notificationResult.items, calendarEvents }));
+          ws.send(JSON.stringify({
+            type: "activity:snapshot",
+            notifications: notificationResult.items,
+            calendarEvents,
+            activities: activityResult.items,
+          }));
         };
 
         unsubscribeActivity = subscribeActivity(userId, sendSnapshot);

apps/backend/src/jobs/index.ts

@@ -19,6 +19,7 @@ import { processQueuedNotifications } from "./notifications/forwarder";
 import { createLogger } from "../lib/logger";
 import { migrateLegacyPageConfig } from "../lib/data/config";
 import { getSuperuserPB } from "../lib/pb/pocketbase";
+import { deleteExpiredActivities } from "../lib/data/activities";
 
 const execFileAsync = promisify(execFile);
 const logger = createLogger("Jobs");
@@ -111,6 +112,13 @@ const runNotificationForwarderJob = (source: string) =>
     errorMessage: "Notification forwarder failed",
   });
 
+const runActivityRetentionJob = (source: string) =>
+  runJob("activityRetention", () => deleteExpiredActivities(config.ACTIVITY_RETENTION_DAYS), {
+    startMessage: `Triggered by ${source}`,
+    successMessage: "Activity retention cleanup completed",
+    errorMessage: "Activity retention cleanup failed",
+  });
+
 const runLegacyUserConfigsMigration = async () => {
   const pb = await getSuperuserPB();
   const users = await pb.collection("users").getFullList<{ id: string }>(500, {
@@ -198,6 +206,11 @@ export function registerJobsCron() {
   Bun.cron(config.NOTIFICATION_FORWARDER_SCHEDULE, async () => {
     await runNotificationForwarderJob("cron schedule");
   });
+
+  void runActivityRetentionJob("server start");
+  Bun.cron("0 5 * * *", async () => {
+    await runActivityRetentionJob("cron schedule");
+  });
 }
 
 export const jobsApi = {
@@ -210,5 +223,6 @@ export const jobsApi = {
   runDefaultIntegrationsJob,
   runNewsFeedBuilderJob,
   runNotificationForwarderJob,
+  runActivityRetentionJob,
   runLegacyUserConfigsMigrationJob,
 };

apps/backend/src/jobs/monitoring/runner.ts

@@ -7,6 +7,7 @@ import {
     updateMonitoringJob,
 } from "../../lib/data/superuser";
 import { createNotificationByTopicId } from "../../lib/data/notifications/publish";
+import { publishActivity } from "../../lib/data/activities";
 import { createLogger } from "../../lib/logger";
 import { getLinkIdFromSource, getLinkSource, parseConfigObject, type StatusCheckMethod } from "./shared";
 
@@ -46,6 +47,33 @@ type AvgLatencyInfo = {
 
 const logger = createLogger("Monitoring");
 
+async function publishMonitorTransitionActivity(
+    job: Record<string, any>,
+    previousStatus: string,
+    status: string,
+    occurredAt: string,
+) {
+    const ownerId = String(job.userId || "").trim();
+    const monitorId = String(job.id || "").trim();
+    if (!ownerId || !monitorId) return;
+
+    const recovered = status === "healthy";
+    try {
+        await publishActivity(ownerId, "monitoring", {
+            type: "monitor.availability_changed",
+            title: recovered ? "A monitored service recovered" : "A monitored service is unavailable",
+            description: recovered ? "A monitored service is responding again." : "A monitored service stopped responding.",
+            severity: recovered ? "success" : "warning",
+            occurredAt,
+            eventId: `${monitorId}:${previousStatus}:${status}:${occurredAt}`,
+            metadata: { monitorId, previousStatus, status },
+        }, monitorId);
+    } catch {
+        // Activity persistence is best-effort and must not interrupt monitoring.
+        logger.warn("Could not publish monitor activity", { jobId: monitorId });
+    }
+}
+
 export async function runStatusMonitoringJobs(): Promise<{
     processed: number;
     skipped: number;
@@ -179,6 +207,8 @@ export async function runStatusMonitoringJobsWithOptions(options?: {
                     ...latencyUpdate.payload,
                 });
 
+                await publishMonitorTransitionActivity(job, currentStatus, newStatus, pingTimestamp);
+
                 if (job.notifyOnStatusChange && job.notifyTopicId) {
                     try {
                         const content = `Monitor "${job.title || job.endpoint || job.source || 'Unnamed'}" changed from ${currentStatus} to ${newStatus}`;
@@ -229,11 +259,14 @@ export async function runStatusMonitoringJobsWithOptions(options?: {
                         },
                     ];
 
+                    const occurredAt = new Date().toISOString();
                     await updateMonitoringJob(job.id, {
                         status: 'unhealthy',
                         pings: updatedPings,
                     });
 
+                    await publishMonitorTransitionActivity(job, currentStatus, 'unhealthy', occurredAt);
+
                     if (job.notifyOnStatusChange && job.notifyTopicId) {
                         try {
                             const content = `Monitor "${job.title || job.endpoint || job.source || 'Unnamed'}" changed from ${currentStatus} to unhealthy (fetch error)`;

apps/backend/src/jobs/news/feed-builder.ts

@@ -23,6 +23,7 @@ import {
 } from "../../lib/cache/feed-items";
 import { config } from "../../lib/config";
 import { createLogger } from "../../lib/logger";
+import { publishActivity } from "../../lib/data/activities";
 import { getFeedItems } from "./helper";
 import type { NewsFeedItem } from "@dashwise/types/sdk";
 
@@ -43,6 +44,32 @@ type BuilderOptions = { userId?: string; feedIds?: string[] };
 
 const logger = createLogger("NewsFeedBuilder");
 
+async function publishNewsFeedTransitionActivity(
+  subscription: NewsSubscription,
+  transition: "unavailable" | "recovered",
+) {
+  const ownerId = String(subscription.userId || "").trim();
+  const subscriptionId = String(subscription.id || "").trim();
+  if (!ownerId || !subscriptionId) return;
+
+  const occurredAt = new Date().toISOString();
+  const recovered = transition === "recovered";
+  try {
+    await publishActivity(ownerId, "news", {
+      type: recovered ? "news.feed_recovered" : "news.feed_unavailable",
+      title: recovered ? "A news feed recovered" : "A news feed is unavailable",
+      description: recovered ? "A news feed is updating again." : "A news feed could not be updated.",
+      severity: recovered ? "success" : "warning",
+      occurredAt,
+      eventId: `${subscriptionId}:${transition}:${occurredAt}`,
+      metadata: { subscriptionId },
+    }, subscriptionId);
+  } catch {
+    // Activity persistence is best-effort and must not interrupt feed updates.
+    logger.warn("Could not publish news feed activity", { subscriptionId });
+  }
+}
+
 function itemTime(item: Record<string, unknown>) {
   const value = item.pubDate;
   const time = value instanceof Date ? value.getTime() : new Date(String(value || "")).getTime();
@@ -94,6 +121,7 @@ async function fetchAndCacheSubscription(subscription: NewsSubscription, result:
   const id = String(subscription.id || "");
   const feedUrl = String(subscription.url || subscription.feedUrl || "");
   if (!id || !feedUrl) return false;
+  const wasFailing = Boolean(String(subscription.fetchErrors || "").trim());
 
   try {
     const raw = await getFeedItems({
@@ -106,13 +134,15 @@ async function fetchAndCacheSubscription(subscription: NewsSubscription, result:
     });
     const articles = deduplicateSubscriptionArticles(raw as unknown as Record<string, unknown>[], subscription);
     await writeSubscriptionArticles(id, articles);
-    await updateNewsSubscription(id, { fetchErrors: "" });
+    const updated = await updateNewsSubscription(id, { fetchErrors: "" });
+    if (updated && wasFailing) await publishNewsFeedTransitionActivity(subscription, "recovered");
     return true;
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
     result.errors++;
     result.details.push({ subscriptionId: id, action: "feed_fetch_error", error: message });
-    await updateNewsSubscription(id, { fetchErrors: message }).catch(() => undefined);
+    const updated = await updateNewsSubscription(id, { fetchErrors: message }).catch(() => null);
+    if (updated && !wasFailing) await publishNewsFeedTransitionActivity(subscription, "unavailable");
     logger.error(`Error fetching feed "${subscription.title || feedUrl}": ${message}`);
     // A failed fetch deliberately leaves the previous sorted-set cache intact.
     return false;