calvin.erfmann/pangolin
1
0
Fork 0
mirror of https://github.com/fosrl/pangolin.git synced 2026-06-06 07:38:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use transactions

Browse Source
This commit is contained in:
Owen
2026-04-22 18:13:02 -07:00
parent dcbd22b4ad
commit 245755a140
10 changed files with 147 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
server/lib/alerts/events/healthCheckEvents.ts
View File

@@ -4,7 +4,9 @@ export async function fireHealthCheckHealthyAlert(
orgId: string, orgId: string,
healthCheckId: number, healthCheckId: number,
healthCheckName?: string, healthCheckName?: string,
extra?: Record<string, unknown> healthCheckTargetId?: number | null,
extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> { ): Promise<void> {
return; return;
} }
@@ -13,7 +15,9 @@ export async function fireHealthCheckUnhealthyAlert(
orgId: string, orgId: string,
healthCheckId: number, healthCheckId: number,
healthCheckName?: string, healthCheckName?: string,
extra?: Record<string, unknown> healthCheckTargetId?: number | null,
extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> { ): Promise<void> {
return; return;
} }

11
server/lib/alerts/events/resourceEvents.ts
View File

@@ -2,19 +2,22 @@ export async function fireResourceHealthyAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> {} ): Promise<void> {}
export async function fireResourceUnhealthyAlert( export async function fireResourceUnhealthyAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> {} ): Promise<void> {}
export async function fireResourceToggleAlert( export async function fireResourceToggleAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
): Promise<void> {} trx?: unknown
): Promise<void> {}

8
server/lib/alerts/events/siteEvents.ts
View File

@@ -4,7 +4,8 @@ export async function fireSiteOnlineAlert(
orgId: string, orgId: string,
siteId: number, siteId: number,
siteName?: string, siteName?: string,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> { ): Promise<void> {
return; return;
} }
@@ -13,7 +14,8 @@ export async function fireSiteOfflineAlert(
orgId: string, orgId: string,
siteId: number, siteId: number,
siteName?: string, siteName?: string,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx?: unknown
): Promise<void> { ): Promise<void> {
return; return;
} }

35
server/private/lib/alerts/events/healthCheckEvents.ts
View File

@@ -18,7 +18,8 @@ import {
statusHistory, statusHistory,
targetHealthCheck, targetHealthCheck,
targets, targets,
resources resources,
Transaction
} from "@server/db"; } from "@server/db";
import { eq } from "drizzle-orm"; import { eq } from "drizzle-orm";
import { import {
@@ -46,10 +47,11 @@ export async function fireHealthCheckHealthyAlert(
healthCheckId: number, healthCheckId: number,
healthCheckName?: string | null, healthCheckName?: string | null,
healthCheckTargetId?: number | null, healthCheckTargetId?: number | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "health_check", entityType: "health_check",
entityId: healthCheckId, entityId: healthCheckId,
orgId: orgId, orgId: orgId,
@@ -57,7 +59,7 @@ export async function fireHealthCheckHealthyAlert(
timestamp: Math.floor(Date.now() / 1000) timestamp: Math.floor(Date.now() / 1000)
}); });
await handleResource(orgId, healthCheckTargetId); await handleResource(orgId, healthCheckTargetId, trx);
await processAlerts({ await processAlerts({
eventType: "health_check_healthy", eventType: "health_check_healthy",
@@ -102,10 +104,11 @@ export async function fireHealthCheckUnhealthyAlert(
healthCheckId: number, healthCheckId: number,
healthCheckName?: string | null, healthCheckName?: string | null,
healthCheckTargetId?: number | null, healthCheckTargetId?: number | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "health_check", entityType: "health_check",
entityId: healthCheckId, entityId: healthCheckId,
orgId: orgId, orgId: orgId,
@@ -113,7 +116,7 @@ export async function fireHealthCheckUnhealthyAlert(
timestamp: Math.floor(Date.now() / 1000) timestamp: Math.floor(Date.now() / 1000)
}); });
await handleResource(orgId, healthCheckTargetId); await handleResource(orgId, healthCheckTargetId, trx);
await processAlerts({ await processAlerts({
eventType: "health_check_unhealthy", eventType: "health_check_unhealthy",
@@ -142,12 +145,12 @@ export async function fireHealthCheckUnhealthyAlert(
} }
} }
async function handleResource(orgId: string, healthCheckTargetId?: number | null) { async function handleResource(orgId: string, healthCheckTargetId?: number | null, trx: Transaction | typeof db = db) {
if (!healthCheckTargetId) { if (!healthCheckTargetId) {
return; return;
} }
// we have resources lets get them // we have resources lets get them
const [target] = await db const [target] = await trx
.select() .select()
.from(targets) .from(targets)
.where(eq(targets.targetId, healthCheckTargetId)) .where(eq(targets.targetId, healthCheckTargetId))
@@ -156,7 +159,7 @@ async function handleResource(orgId: string, healthCheckTargetId?: number | null
if (!target) { if (!target) {
return; return;
} }
const [resource] = await db const [resource] = await trx
.select() .select()
.from(resources) .from(resources)
.where(eq(resources.resourceId, target.resourceId)) .where(eq(resources.resourceId, target.resourceId))
@@ -165,7 +168,7 @@ async function handleResource(orgId: string, healthCheckTargetId?: number | null
if (!resource) { if (!resource) {
return; return;
} }
const otherTargets = await db const otherTargets = await trx
.select({ hcHealth: targetHealthCheck.hcHealth }) .select({ hcHealth: targetHealthCheck.hcHealth })
.from(targets) .from(targets)
.where(eq(targets.resourceId, resource.resourceId)); .where(eq(targets.resourceId, resource.resourceId));
@@ -181,7 +184,7 @@ async function handleResource(orgId: string, healthCheckTargetId?: number | null
if (health != resource.health) { if (health != resource.health) {
// it changed // it changed
await db await trx
.update(resources) .update(resources)
.set({ health }) .set({ health })
.where(eq(resources.resourceId, resource.resourceId)); .where(eq(resources.resourceId, resource.resourceId));
@@ -190,13 +193,17 @@ async function handleResource(orgId: string, healthCheckTargetId?: number | null
await fireResourceUnhealthyAlert( await fireResourceUnhealthyAlert(
orgId, orgId,
resource.resourceId, resource.resourceId,
resource.name resource.name,
undefined,
trx
); );
} else if (health === "healthy") { } else if (health === "healthy") {
await fireResourceHealthyAlert( await fireResourceHealthyAlert(
orgId, orgId,
resource.resourceId, resource.resourceId,
resource.name resource.name,
undefined,
trx
); );
} }
} }

15
server/private/lib/alerts/events/resourceEvents.ts
View File

@@ -13,7 +13,7 @@
import logger from "@server/logger"; import logger from "@server/logger";
import { processAlerts } from "../processAlerts"; import { processAlerts } from "../processAlerts";
import { db, statusHistory } from "@server/db"; import { db, statusHistory, Transaction } from "@server/db";
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Public API // Public API
@@ -34,10 +34,11 @@ export async function fireResourceHealthyAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "resource", entityType: "resource",
entityId: resourceId, entityId: resourceId,
orgId: orgId, orgId: orgId,
@@ -87,10 +88,11 @@ export async function fireResourceUnhealthyAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "resource", entityType: "resource",
entityId: resourceId, entityId: resourceId,
orgId: orgId, orgId: orgId,
@@ -140,7 +142,8 @@ export async function fireResourceToggleAlert(
orgId: string, orgId: string,
resourceId: number, resourceId: number,
resourceName?: string | null, resourceName?: string | null,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await processAlerts({ await processAlerts({

19
server/private/lib/alerts/events/siteEvents.ts
View File

@@ -13,7 +13,7 @@
import logger from "@server/logger"; import logger from "@server/logger";
import { processAlerts } from "../processAlerts"; import { processAlerts } from "../processAlerts";
import { db, sites, statusHistory, targetHealthCheck } from "@server/db"; import { db, sites, statusHistory, targetHealthCheck, Transaction } from "@server/db";
import { and, eq, inArray } from "drizzle-orm"; import { and, eq, inArray } from "drizzle-orm";
import { fireHealthCheckUnhealthyAlert } from "./healthCheckEvents"; import { fireHealthCheckUnhealthyAlert } from "./healthCheckEvents";
@@ -36,10 +36,11 @@ export async function fireSiteOnlineAlert(
orgId: string, orgId: string,
siteId: number, siteId: number,
siteName?: string, siteName?: string,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "site", entityType: "site",
entityId: siteId, entityId: siteId,
orgId: orgId, orgId: orgId,
@@ -89,10 +90,11 @@ export async function fireSiteOfflineAlert(
orgId: string, orgId: string,
siteId: number, siteId: number,
siteName?: string, siteName?: string,
extra?: Record<string, unknown> extra?: Record<string, unknown>,
trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
try { try {
await db.insert(statusHistory).values({ await trx.insert(statusHistory).values({
entityType: "site", entityType: "site",
entityId: siteId, entityId: siteId,
orgId: orgId, orgId: orgId,
@@ -100,7 +102,7 @@ export async function fireSiteOfflineAlert(
timestamp: Math.floor(Date.now() / 1000) timestamp: Math.floor(Date.now() / 1000)
}); });
const unhealthyHealthChecks = await db const unhealthyHealthChecks = await trx
.update(targetHealthCheck) .update(targetHealthCheck)
.set({ hcHealth: "unhealthy" }) .set({ hcHealth: "unhealthy" })
.where( .where(
@@ -119,7 +121,10 @@ export async function fireSiteOfflineAlert(
await fireHealthCheckUnhealthyAlert( await fireHealthCheckUnhealthyAlert(
healthCheck.orgId, healthCheck.orgId,
healthCheck.targetHealthCheckId, healthCheck.targetHealthCheckId,
healthCheck.name healthCheck.name,
undefined,
undefined,
trx
); );
} }

23
server/routers/newt/handleNewtDisconnectingMessage.ts
View File

@@ -2,10 +2,7 @@ import { MessageHandler } from "@server/routers/ws";
import { import {
db, db,
Newt, Newt,
sites, sites
statusHistory,
targetHealthCheck,
targets
} from "@server/db"; } from "@server/db";
import { eq } from "drizzle-orm"; import { eq } from "drizzle-orm";
import logger from "@server/logger"; import logger from "@server/logger";
@@ -32,15 +29,17 @@ export const handleNewtDisconnectingMessage: MessageHandler = async (
try { try {
// Update the client's last ping timestamp // Update the client's last ping timestamp
const [site] = await db await db.transaction(async (trx) => {
.update(sites) const [site] = await trx
.set({ .update(sites)
online: false .set({
}) online: false
.where(eq(sites.siteId, newt.siteId)) })
.returning(); .where(eq(sites.siteId, newt.siteId!))
.returning();
await fireSiteOfflineAlert(site.orgId, site.siteId, site.name); await fireSiteOfflineAlert(site.orgId, site.siteId, site.name, undefined, trx);
});
} catch (error) { } catch (error) {
logger.error("Error handling disconnecting message", { error }); logger.error("Error handling disconnecting message", { error });
} }

66
server/routers/newt/offlineChecker.ts
View File

@@ -77,16 +77,20 @@ export const startNewtOfflineChecker = (): void => {
`Marking site ${staleSite.siteId} offline: newt ${staleSite.newtId} has no recent ping and no active WebSocket connection` `Marking site ${staleSite.siteId} offline: newt ${staleSite.newtId} has no recent ping and no active WebSocket connection`
); );
await db await db.transaction(async (trx) => {
.update(sites) await trx
.set({ online: false }) .update(sites)
.where(eq(sites.siteId, staleSite.siteId)); .set({ online: false })
.where(eq(sites.siteId, staleSite.siteId));
await fireSiteOfflineAlert( await fireSiteOfflineAlert(
staleSite.orgId, staleSite.orgId,
staleSite.siteId, staleSite.siteId,
staleSite.name staleSite.name,
); undefined,
trx
);
});
} }
// this part only effects self hosted. Its not efficient but we dont expect people to have very many wireguard sites // this part only effects self hosted. Its not efficient but we dont expect people to have very many wireguard sites
@@ -123,16 +127,20 @@ export const startNewtOfflineChecker = (): void => {
`Marking wireguard site ${site.siteId} offline: no bandwidth update in over ${OFFLINE_THRESHOLD_BANDWIDTH_MS / 60000} minutes` `Marking wireguard site ${site.siteId} offline: no bandwidth update in over ${OFFLINE_THRESHOLD_BANDWIDTH_MS / 60000} minutes`
); );
await db await db.transaction(async (trx) => {
.update(sites) await trx
.set({ online: false }) .update(sites)
.where(eq(sites.siteId, site.siteId)); .set({ online: false })
.where(eq(sites.siteId, site.siteId));
await fireSiteOfflineAlert( await fireSiteOfflineAlert(
site.orgId, site.orgId,
site.siteId, site.siteId,
site.name site.name,
); undefined,
trx
);
});
} else if ( } else if (
lastBandwidthUpdate >= wireguardOfflineThreshold && lastBandwidthUpdate >= wireguardOfflineThreshold &&
!site.online !site.online
@@ -141,16 +149,20 @@ export const startNewtOfflineChecker = (): void => {
`Marking wireguard site ${site.siteId} online: recent bandwidth update` `Marking wireguard site ${site.siteId} online: recent bandwidth update`
); );
await db await db.transaction(async (trx) => {
.update(sites) await trx
.set({ online: true }) .update(sites)
.where(eq(sites.siteId, site.siteId)); .set({ online: true })
.where(eq(sites.siteId, site.siteId));
await fireSiteOnlineAlert( await fireSiteOnlineAlert(
site.orgId, site.orgId,
site.siteId, site.siteId,
site.name site.name,
); undefined,
trx
);
});
} }
} }
} catch (error) { } catch (error) {

4
server/routers/newt/pingAccumulator.ts
View File

@@ -147,7 +147,9 @@ async function flushSitePingsToDb(): Promise<void> {
}, "flushSitePingsToDb"); }, "flushSitePingsToDb");
for (const site of newlyOnlineSites) { for (const site of newlyOnlineSites) {
await fireSiteOnlineAlert(site.orgId, site.siteId, site.name); await db.transaction(async (trx) => {
await fireSiteOnlineAlert(site.orgId, site.siteId, site.name, undefined, trx);
});
} }
} catch (error) { } catch (error) {
logger.error( logger.error(

63
server/routers/target/handleHealthcheckStatusMessage.ts
View File

@@ -14,10 +14,7 @@ import {
fireHealthCheckHealthyAlert, fireHealthCheckHealthyAlert,
fireHealthCheckUnhealthyAlert fireHealthCheckUnhealthyAlert
} from "#dynamic/lib/alerts"; } from "#dynamic/lib/alerts";
import {
fireResourceHealthyAlert,
fireResourceUnhealthyAlert
} from "#dynamic/lib/alerts";
interface TargetHealthStatus { interface TargetHealthStatus {
status: string; status: string;
@@ -125,33 +122,39 @@ export const handleHealthcheckStatusMessage: MessageHandler = async (
continue; continue;
} }
// Update the target's health status in the database // Update the target's health status in the database and fire alert in a transaction
await db await db.transaction(async (trx) => {
.update(targetHealthCheck) await trx
.set({ .update(targetHealthCheck)
hcHealth: healthStatus.status as .set({
| "unknown" hcHealth: healthStatus.status as
| "healthy" | "unknown"
| "unhealthy" | "healthy"
}) | "unhealthy"
.where(eq(targetHealthCheck.targetHealthCheckId, targetCheck.targetHealthCheckId)); })
.where(eq(targetHealthCheck.targetHealthCheckId, targetCheck.targetHealthCheckId));
// because we are checking above if there was a change we can fire the alert here because it changed // because we are checking above if there was a change we can fire the alert here because it changed
if (healthStatus.status === "unhealthy") { if (healthStatus.status === "unhealthy") {
await fireHealthCheckUnhealthyAlert( await fireHealthCheckUnhealthyAlert(
targetCheck.orgId, targetCheck.orgId,
targetCheck.targetHealthCheckId, targetCheck.targetHealthCheckId,
targetCheck.name ?? undefined, targetCheck.name ?? undefined,
targetCheck.targetId targetCheck.targetId,
); undefined,
} else if (healthStatus.status === "healthy") { trx
await fireHealthCheckHealthyAlert( );
targetCheck.orgId, } else if (healthStatus.status === "healthy") {
targetCheck.targetHealthCheckId, await fireHealthCheckHealthyAlert(
targetCheck.name ?? undefined, targetCheck.orgId,
targetCheck.targetId targetCheck.targetHealthCheckId,
); targetCheck.name ?? undefined,
} targetCheck.targetId,
undefined,
trx
);
}
});
logger.debug( logger.debug(
`Updated health status for target ${targetId} to ${healthStatus.status}` `Updated health status for target ${targetId} to ${healthStatus.status}`