Fix CE not processing alert status

Fixes #2968

server/lib/alerts/events/healthCheckEvents.ts

@@ -1,27 +1,153 @@
-// stub
+import logger from "@server/logger";
+import { processAlerts } from "@server/lib/alerts";
+import {
+    db,
+    statusHistory,
+    targetHealthCheck,
+    targets,
+    resources,
+    Transaction,
+    logsDb
+} from "@server/db";
+import { eq } from "drizzle-orm";
+import { invalidateStatusHistoryCache } from "@server/lib/statusHistory";
+import {
+    fireResourceDegradedAlert,
+    fireResourceHealthyAlert,
+    fireResourceUnhealthyAlert,
+    fireResourceUnknownAlert
+} from "./resourceEvents";
 
+// ---------------------------------------------------------------------------
+// Public API
+// ---------------------------------------------------------------------------
+
+/**
+ * Fire a `health_check_healthy` alert for the given health check.
+ *
+ * Call this after a previously-failing health check has recovered so that any
+ * matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId         - Organisation that owns the health check.
+ * @param healthCheckId - Numeric primary key of the health check.
+ * @param healthCheckName - Human-readable name shown in notifications (optional).
+ * @param extra         - Any additional key/value pairs to include in the payload.
+ */
 export async function fireHealthCheckHealthyAlert(
     orgId: string,
     healthCheckId: number,
-    healthCheckName?: string,
+    healthCheckName?: string | null,
     healthCheckTargetId?: number | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
+    trx: Transaction | typeof db = db
 ): Promise<void> {
-    return;
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "health_check",
+            entityId: healthCheckId,
+            orgId: orgId,
+            status: "healthy",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("health_check", healthCheckId);
+
+        await handleResource(orgId, healthCheckTargetId, send, trx);
+
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "health_check_healthy",
+            orgId,
+            healthCheckId,
+            data: {
+                ...(healthCheckName != null ? { healthCheckName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "health_check_toggle",
+            orgId,
+            healthCheckId,
+            data: {
+                healthCheckId,
+                status: "healthy",
+                ...(healthCheckName != null ? { healthCheckName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireHealthCheckHealthyAlert: unexpected error for healthCheckId ${healthCheckId}`,
+            err
+        );
+    }
 }
 
+/**
+ * Fire a `health_check_unhealthy` alert for the given health check.
+ *
+ * Call this after a health check has been detected as failing so that any
+ * matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId         - Organisation that owns the health check.
+ * @param healthCheckId - Numeric primary key of the health check.
+ * @param healthCheckName - Human-readable name shown in notifications (optional).
+ * @param extra         - Any additional key/value pairs to include in the payload.
+ */
 export async function fireHealthCheckUnhealthyAlert(
     orgId: string,
     healthCheckId: number,
-    healthCheckName?: string,
+    healthCheckName?: string | null,
     healthCheckTargetId?: number | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
+    trx: Transaction | typeof db = db
 ): Promise<void> {
-    return;
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "health_check",
+            entityId: healthCheckId,
+            orgId: orgId,
+            status: "unhealthy",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("health_check", healthCheckId);
+
+        await handleResource(orgId, healthCheckTargetId, send, trx);
+
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "health_check_unhealthy",
+            orgId,
+            healthCheckId,
+            data: {
+                ...(healthCheckName != null ? { healthCheckName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "health_check_toggle",
+            orgId,
+            healthCheckId,
+            data: {
+                healthCheckId,
+                status: "unhealthy",
+                ...(healthCheckName != null ? { healthCheckName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireHealthCheckUnhealthyAlert: unexpected error for healthCheckId ${healthCheckId}`,
+            err
+        );
+    }
 }
 
 export async function fireHealthCheckUnknownAlert(
@@ -31,7 +157,137 @@ export async function fireHealthCheckUnknownAlert(
     healthCheckTargetId?: number | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
+    trx: Transaction | typeof db = db
 ): Promise<void> {
-    return;
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "health_check",
+            entityId: healthCheckId,
+            orgId: orgId,
+            status: "unknown",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("health_check", healthCheckId);
+
+        await handleResource(orgId, healthCheckTargetId, send, trx);
+
+        if (!send) {
+            return;
+        }
+    } catch (err) {
+        logger.error(
+            `fireHealthCheckUnknownAlert: unexpected error for healthCheckId ${healthCheckId}`,
+            err
+        );
+    }
+}
+
+async function handleResource(
+    orgId: string,
+    healthCheckTargetId?: number | null,
+    send: boolean = true,
+    trx: Transaction | typeof db = db
+) {
+    if (!healthCheckTargetId) {
+        return;
+    }
+    // we have targets lets get them
+    const [target] = await trx
+        .select()
+        .from(targets)
+        .where(eq(targets.targetId, healthCheckTargetId))
+        .limit(1);
+
+    if (!target) {
+        return;
+    }
+
+    const [resource] = await trx
+        .select()
+        .from(resources)
+        .where(eq(resources.resourceId, target.resourceId))
+        .limit(1);
+
+    if (!resource) {
+        return;
+    }
+
+    const otherTargets = await trx
+        .select({ hcHealth: targetHealthCheck.hcHealth })
+        .from(targets)
+        .innerJoin(
+            targetHealthCheck,
+            eq(targetHealthCheck.targetId, targets.targetId)
+        )
+        .where(eq(targets.resourceId, resource.resourceId));
+
+    let health = "healthy";
+    const allUnknown = otherTargets.every((t) => t.hcHealth === "unknown");
+    const allHealthy = otherTargets.every((t) => t.hcHealth === "healthy");
+    const allUnhealthy = otherTargets.every((t) => t.hcHealth === "unhealthy");
+
+    if (allUnknown) {
+        logger.debug(
+            `Marking resource ${resource.resourceId} as unknown because all health checks are disabled`
+        );
+        health = "unknown";
+    } else if (allHealthy) {
+        health = "healthy";
+    } else if (allUnhealthy) {
+        logger.debug(
+            `Marking resource ${resource.resourceId} as unhealthy because all targets are unhealthy`
+        );
+        health = "unhealthy";
+    } else {
+        logger.debug(
+            `Marking resource ${resource.resourceId} as degraded because some targets are unhealthy`
+        );
+        health = "degraded";
+    }
+
+    if (health != resource.health) {
+        // it changed
+        await trx
+            .update(resources)
+            .set({ health })
+            .where(eq(resources.resourceId, resource.resourceId));
+
+        if (health === "unknown") {
+            await fireResourceUnknownAlert(
+                orgId,
+                resource.resourceId,
+                resource.name,
+                undefined,
+                send,
+                trx
+            );
+        } else if (health === "unhealthy") {
+            await fireResourceUnhealthyAlert(
+                orgId,
+                resource.resourceId,
+                resource.name,
+                undefined,
+                send,
+                trx
+            );
+        } else if (health === "healthy") {
+            await fireResourceHealthyAlert(
+                orgId,
+                resource.resourceId,
+                resource.name,
+                undefined,
+                send,
+                trx
+            );
+        } else if (health === "degraded") {
+            await fireResourceDegradedAlert(
+                orgId,
+                resource.resourceId,
+                resource.name,
+                undefined,
+                send,
+                trx
+            );
+        }
+    }
 }

server/lib/alerts/events/resourceEvents.ts

@@ -1,26 +1,243 @@
+import logger from "@server/logger";
+import { processAlerts } from "@server/lib/alerts";
+import { db, logsDb, statusHistory, Transaction } from "@server/db";
+import { invalidateStatusHistoryCache } from "@server/lib/statusHistory";
+
+// ---------------------------------------------------------------------------
+// Public API
+// ---------------------------------------------------------------------------
+
+/**
+ * Fire a `resource_healthy` alert for the given resource.
+ *
+ * Call this after a previously-unhealthy resource has recovered so that any
+ * matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId        - Organisation that owns the resource.
+ * @param resourceId   - Numeric primary key of the resource.
+ * @param resourceName - Human-readable name shown in notifications (optional).
+ * @param extra        - Any additional key/value pairs to include in the payload.
+ */
 export async function fireResourceHealthyAlert(
     orgId: string,
     resourceId: number,
     resourceName?: string | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
-): Promise<void> {}
+    trx: Transaction | typeof db = db
+): Promise<void> {
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "resource",
+            entityId: resourceId,
+            orgId: orgId,
+            status: "healthy",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("resource", resourceId);
 
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "resource_healthy",
+            orgId,
+            resourceId,
+            data: {
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "resource_toggle",
+            orgId,
+            resourceId,
+            data: {
+                resourceId,
+                status: "healthy",
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireResourceHealthyAlert: unexpected error for resourceId ${resourceId}`,
+            err
+        );
+    }
+}
+
+/**
+ * Fire a `resource_unhealthy` alert for the given resource.
+ *
+ * Call this after a resource has been detected as unhealthy so that any
+ * matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId        - Organisation that owns the resource.
+ * @param resourceId   - Numeric primary key of the resource.
+ * @param resourceName - Human-readable name shown in notifications (optional).
+ * @param extra        - Any additional key/value pairs to include in the payload.
+ */
 export async function fireResourceUnhealthyAlert(
     orgId: string,
     resourceId: number,
     resourceName?: string | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
-): Promise<void> {}
+    trx: Transaction | typeof db = db
+): Promise<void> {
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "resource",
+            entityId: resourceId,
+            orgId: orgId,
+            status: "unhealthy",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("resource", resourceId);
 
-export async function fireResourceToggleAlert(
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "resource_unhealthy",
+            orgId,
+            resourceId,
+            data: {
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "resource_toggle",
+            orgId,
+            resourceId,
+            data: {
+                resourceId,
+                status: "unhealthy",
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireResourceUnhealthyAlert: unexpected error for resourceId ${resourceId}`,
+            err
+        );
+    }
+}
+
+/**
+ * Fire a `resource_degraded` alert for the given resource.
+ *
+ * Call this after a resource has been detected as degraded so that any
+ * matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId        - Organisation that owns the resource.
+ * @param resourceId   - Numeric primary key of the resource.
+ * @param resourceName - Human-readable name shown in notifications (optional).
+ * @param extra        - Any additional key/value pairs to include in the payload.
+ */
+export async function fireResourceDegradedAlert(
     orgId: string,
     resourceId: number,
     resourceName?: string | null,
     extra?: Record<string, unknown>,
     send: boolean = true,
-    trx?: unknown
-): Promise<void> {}
+    trx: Transaction | typeof db = db
+): Promise<void> {
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "resource",
+            entityId: resourceId,
+            orgId: orgId,
+            status: "degraded",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("resource", resourceId);
+
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "resource_degraded",
+            orgId,
+            resourceId,
+            data: {
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "resource_toggle",
+            orgId,
+            resourceId,
+            data: {
+                resourceId,
+                status: "degraded",
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireResourceDegradedAlert: unexpected error for resourceId ${resourceId}`,
+            err
+        );
+    }
+}
+
+/**
+ * Fire a `resource_unknown` alert for the given resource.
+ *
+ * Call this when all health checks on a resource are disabled so that the
+ * resource status transitions to unknown.
+ *
+ * @param orgId        - Organisation that owns the resource.
+ * @param resourceId   - Numeric primary key of the resource.
+ * @param resourceName - Human-readable name shown in notifications (optional).
+ * @param extra        - Any additional key/value pairs to include in the payload.
+ */
+export async function fireResourceUnknownAlert(
+    orgId: string,
+    resourceId: number,
+    resourceName?: string | null,
+    extra?: Record<string, unknown>,
+    send: boolean = true,
+    trx: Transaction | typeof db = db
+): Promise<void> {
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "resource",
+            entityId: resourceId,
+            orgId: orgId,
+            status: "unknown",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("resource", resourceId);
+
+        if (!send) {
+            return;
+        }
+
+        await processAlerts({
+            eventType: "resource_toggle",
+            orgId,
+            resourceId,
+            data: {
+                resourceId,
+                status: "unknown",
+                ...(resourceName != null ? { resourceName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireResourceUnknownAlert: unexpected error for resourceId ${resourceId}`,
+            err
+        );
+    }
+}

server/lib/alerts/events/siteEvents.ts

@@ -1,21 +1,156 @@
-// stub
+import logger from "@server/logger";
+import { processAlerts } from "@server/lib/alerts";
+import {
+    db,
+    logsDb,
+    statusHistory,
+    targetHealthCheck,
+    Transaction
+} from "@server/db";
+import { invalidateStatusHistoryCache } from "@server/lib/statusHistory";
+import { and, eq, inArray } from "drizzle-orm";
+import { fireHealthCheckUnhealthyAlert } from "./healthCheckEvents";
 
+// ---------------------------------------------------------------------------
+// Public API
+// ---------------------------------------------------------------------------
+
+/**
+ * Fire a `site_online` alert for the given site.
+ *
+ * Call this after the site has been confirmed reachable / connected so that
+ * any matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId    - Organisation that owns the site.
+ * @param siteId   - Numeric primary key of the site.
+ * @param siteName - Human-readable name shown in notifications (optional).
+ * @param extra    - Any additional key/value pairs to include in the payload.
+ */
 export async function fireSiteOnlineAlert(
     orgId: string,
     siteId: number,
     siteName?: string,
     extra?: Record<string, unknown>,
-    trx?: unknown
+    trx: Transaction | typeof db = db
 ): Promise<void> {
-    return;
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "site",
+            entityId: siteId,
+            orgId: orgId,
+            status: "online",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("site", siteId);
+
+        await processAlerts({
+            eventType: "site_online",
+            orgId,
+            siteId,
+            data: {
+                ...(siteName != null ? { siteName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "site_toggle",
+            orgId,
+            siteId,
+            data: {
+                siteId,
+                status: "online",
+                ...(siteName != null ? { siteName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireSiteOnlineAlert: unexpected error for siteId ${siteId}`,
+            err
+        );
+    }
 }
 
+/**
+ * Fire a `site_offline` alert for the given site.
+ *
+ * Call this after the site has been detected as unreachable / disconnected so
+ * that any matching `alertRules` can dispatch their email and webhook actions.
+ *
+ * @param orgId    - Organisation that owns the site.
+ * @param siteId   - Numeric primary key of the site.
+ * @param siteName - Human-readable name shown in notifications (optional).
+ * @param extra    - Any additional key/value pairs to include in the payload.
+ */
 export async function fireSiteOfflineAlert(
     orgId: string,
     siteId: number,
     siteName?: string,
     extra?: Record<string, unknown>,
-    trx?: unknown
+    trx: Transaction | typeof db = db
 ): Promise<void> {
-    return;
-}
+    try {
+        await logsDb.insert(statusHistory).values({
+            entityType: "site",
+            entityId: siteId,
+            orgId: orgId,
+            status: "offline",
+            timestamp: Math.floor(Date.now() / 1000)
+        });
+        await invalidateStatusHistoryCache("site", siteId);
+
+        const unhealthyHealthChecks = await trx
+            .update(targetHealthCheck)
+            .set({ hcHealth: "unhealthy" })
+            .where(
+                and(
+                    eq(targetHealthCheck.orgId, orgId),
+                    eq(targetHealthCheck.siteId, siteId),
+                    eq(targetHealthCheck.hcEnabled, true) // only effect the ones that are enabled
+                )
+            )
+            .returning();
+
+        for (const healthCheck of unhealthyHealthChecks) {
+            logger.info(
+                `Marking health check ${healthCheck.targetHealthCheckId} unhealthy due to site ${siteId} being marked offline`
+            );
+
+            await fireHealthCheckUnhealthyAlert(
+                healthCheck.orgId,
+                healthCheck.targetHealthCheckId,
+                healthCheck.name,
+                healthCheck.targetId, // for the resource if we have one
+                undefined,
+                true,
+                trx
+            );
+        }
+
+        await processAlerts({
+            eventType: "site_offline",
+            orgId,
+            siteId,
+            data: {
+                ...(siteName != null ? { siteName } : {}),
+                ...extra
+            }
+        });
+        await processAlerts({
+            eventType: "site_toggle",
+            orgId,
+            siteId,
+            data: {
+                siteId,
+                status: "offline",
+                ...(siteName != null ? { siteName } : {}),
+                ...extra
+            }
+        });
+    } catch (err) {
+        logger.error(
+            `fireSiteOfflineAlert: unexpected error for siteId ${siteId}`,
+            err
+        );
+    }
+}

server/lib/alerts/index.ts

@@ -1,3 +1,4 @@
 export * from "./events/siteEvents";
 export * from "./events/healthCheckEvents";
 export * from "./events/resourceEvents";
+export * from "./processAlerts";

server/lib/alerts/processAlerts.ts

@@ -0,0 +1,5 @@
+import { AlertContext } from "@server/routers/alertRule/types";
+
+export async function processAlerts(context: AlertContext): Promise<void> {
+    return;
+}

server/lib/billing/getOrgTierData.ts

@@ -1,8 +1,9 @@
 export async function getOrgTierData(
     orgId: string
-): Promise<{ tier: string | null; active: boolean }> {
+): Promise<{ tier: string | null; active: boolean; isTrial: boolean }> {
     const tier = null;
     const active = false;
+    const isTrial = false;
 
-    return { tier, active };
+    return { tier, active, isTrial };
 }

server/lib/blueprints/proxyResources.ts

@@ -34,7 +34,7 @@ import { hashPassword } from "@server/auth/password";
 import { isValidCIDR, isValidIP, isValidUrlGlobPattern } from "../validators";
 import { isValidRegionId } from "@server/db/regions";
 import { isLicensedOrSubscribed } from "#dynamic/lib/isLicencedOrSubscribed";
-import { fireHealthCheckUnknownAlert } from "#dynamic/lib/alerts";
+import { fireHealthCheckUnknownAlert } from "@server/lib/alerts";
 import { tierMatrix } from "../billing/tierMatrix";
 
 export type ProxyResourcesResults = {
@@ -165,7 +165,8 @@ export async function updateProxyResources(
                     hcStatus: healthcheckData?.status,
                     hcHealth: "unknown",
                     hcHealthyThreshold: healthcheckData?.["healthy-threshold"],
-                    hcUnhealthyThreshold: healthcheckData?.["unhealthy-threshold"]
+                    hcUnhealthyThreshold:
+                        healthcheckData?.["unhealthy-threshold"]
                 })
                 .returning();
 
@@ -544,8 +545,10 @@ export async function updateProxyResources(
                                 healthcheckData?.["follow-redirects"],
                             hcMethod: healthcheckData?.method,
                             hcStatus: healthcheckData?.status,
-                            hcHealthyThreshold: healthcheckData?.["healthy-threshold"],
-                            hcUnhealthyThreshold: healthcheckData?.["unhealthy-threshold"]
+                            hcHealthyThreshold:
+                                healthcheckData?.["healthy-threshold"],
+                            hcUnhealthyThreshold:
+                                healthcheckData?.["unhealthy-threshold"]
                         })
                         .where(
                             eq(
@@ -1120,8 +1123,10 @@ function checkIfHealthcheckChanged(
         JSON.stringify(incoming.hcHeaders)
     )
         return true;
-    if (existing.hcHealthyThreshold !== incoming.hcHealthyThreshold) return true;
-    if (existing.hcUnhealthyThreshold !== incoming.hcUnhealthyThreshold) return true;
+    if (existing.hcHealthyThreshold !== incoming.hcHealthyThreshold)
+        return true;
+    if (existing.hcUnhealthyThreshold !== incoming.hcUnhealthyThreshold)
+        return true;
 
     return false;
 }
@@ -1184,7 +1189,11 @@ async function getDomainId(
     orgId: string,
     fullDomain: string,
     trx: Transaction
-): Promise<{ subdomain: string | null; domainId: string; wildcard: boolean } | null> {
+): Promise<{
+    subdomain: string | null;
+    domainId: string;
+    wildcard: boolean;
+} | null> {
     const isWildcardFullDomain = fullDomain.startsWith("*.");
 
     const possibleDomains = await trx