mirror of
https://github.com/fosrl/pangolin.git
synced 2026-09-03 17:59:29 +02:00
Compare commits
5 Commits
1.17.0
...
eb771ceda4
| Author | SHA1 | Date | |
|---|---|---|---|
| eb771ceda4 | |||
| 1efd2af44b | |||
| 466f137590 | |||
| 28ef5238c9 | |||
| d948d2ec33 |
@@ -0,0 +1 @@
|
|||||||
|
* @oschwartz10612 @miloschwartz
|
||||||
@@ -230,7 +230,7 @@ export const siteResources = pgTable("siteResources", {
|
|||||||
.references(() => orgs.orgId, { onDelete: "cascade" }),
|
.references(() => orgs.orgId, { onDelete: "cascade" }),
|
||||||
niceId: varchar("niceId").notNull(),
|
niceId: varchar("niceId").notNull(),
|
||||||
name: varchar("name").notNull(),
|
name: varchar("name").notNull(),
|
||||||
mode: varchar("mode").$type<"host" | "cidr">().notNull(), // "host" | "cidr" | "port"
|
mode: varchar("mode").$type<"host" | "cidr" | "http" | "https">().notNull(), // "host" | "cidr" | "http" | "https"
|
||||||
protocol: varchar("protocol"), // only for port mode
|
protocol: varchar("protocol"), // only for port mode
|
||||||
proxyPort: integer("proxyPort"), // only for port mode
|
proxyPort: integer("proxyPort"), // only for port mode
|
||||||
destinationPort: integer("destinationPort"), // only for port mode
|
destinationPort: integer("destinationPort"), // only for port mode
|
||||||
|
|||||||
@@ -258,7 +258,7 @@ export const siteResources = sqliteTable("siteResources", {
|
|||||||
.references(() => orgs.orgId, { onDelete: "cascade" }),
|
.references(() => orgs.orgId, { onDelete: "cascade" }),
|
||||||
niceId: text("niceId").notNull(),
|
niceId: text("niceId").notNull(),
|
||||||
name: text("name").notNull(),
|
name: text("name").notNull(),
|
||||||
mode: text("mode").$type<"host" | "cidr">().notNull(), // "host" | "cidr" | "port"
|
mode: text("mode").$type<"host" | "cidr" | "http" | "https">().notNull(), // "host" | "cidr" | "http" | "https"
|
||||||
protocol: text("protocol"), // only for port mode
|
protocol: text("protocol"), // only for port mode
|
||||||
proxyPort: integer("proxyPort"), // only for port mode
|
proxyPort: integer("proxyPort"), // only for port mode
|
||||||
destinationPort: integer("destinationPort"), // only for port mode
|
destinationPort: integer("destinationPort"), // only for port mode
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ import { TraefikConfigManager } from "@server/lib/traefik/TraefikConfigManager";
|
|||||||
import { initCleanup } from "#dynamic/cleanup";
|
import { initCleanup } from "#dynamic/cleanup";
|
||||||
import license from "#dynamic/license/license";
|
import license from "#dynamic/license/license";
|
||||||
import { initLogCleanupInterval } from "@server/lib/cleanupLogs";
|
import { initLogCleanupInterval } from "@server/lib/cleanupLogs";
|
||||||
|
import { initAcmeCertSync } from "#dynamic/lib/acmeCertSync";
|
||||||
import { fetchServerIp } from "@server/lib/serverIpService";
|
import { fetchServerIp } from "@server/lib/serverIpService";
|
||||||
|
|
||||||
async function startServers() {
|
async function startServers() {
|
||||||
@@ -39,6 +40,7 @@ async function startServers() {
|
|||||||
initTelemetryClient();
|
initTelemetryClient();
|
||||||
|
|
||||||
initLogCleanupInterval();
|
initLogCleanupInterval();
|
||||||
|
initAcmeCertSync();
|
||||||
|
|
||||||
// Start all servers
|
// Start all servers
|
||||||
const apiServer = createApiServer();
|
const apiServer = createApiServer();
|
||||||
|
|||||||
@@ -0,0 +1,3 @@
|
|||||||
|
export function initAcmeCertSync(): void {
|
||||||
|
// stub
|
||||||
|
}
|
||||||
@@ -84,6 +84,7 @@ export async function updateClientResources(
|
|||||||
siteId: site.siteId,
|
siteId: site.siteId,
|
||||||
mode: resourceData.mode,
|
mode: resourceData.mode,
|
||||||
destination: resourceData.destination,
|
destination: resourceData.destination,
|
||||||
|
destinationPort: resourceData["destination-port"],
|
||||||
enabled: true, // hardcoded for now
|
enabled: true, // hardcoded for now
|
||||||
// enabled: resourceData.enabled ?? true,
|
// enabled: resourceData.enabled ?? true,
|
||||||
alias: resourceData.alias || null,
|
alias: resourceData.alias || null,
|
||||||
@@ -223,6 +224,7 @@ export async function updateClientResources(
|
|||||||
name: resourceData.name || resourceNiceId,
|
name: resourceData.name || resourceNiceId,
|
||||||
mode: resourceData.mode,
|
mode: resourceData.mode,
|
||||||
destination: resourceData.destination,
|
destination: resourceData.destination,
|
||||||
|
destinationPort: resourceData["destination-port"],
|
||||||
enabled: true, // hardcoded for now
|
enabled: true, // hardcoded for now
|
||||||
// enabled: resourceData.enabled ?? true,
|
// enabled: resourceData.enabled ?? true,
|
||||||
alias: resourceData.alias || null,
|
alias: resourceData.alias || null,
|
||||||
|
|||||||
@@ -325,11 +325,11 @@ export function isTargetsOnlyResource(resource: any): boolean {
|
|||||||
export const ClientResourceSchema = z
|
export const ClientResourceSchema = z
|
||||||
.object({
|
.object({
|
||||||
name: z.string().min(1).max(255),
|
name: z.string().min(1).max(255),
|
||||||
mode: z.enum(["host", "cidr"]),
|
mode: z.enum(["host", "cidr", "http", "https"]),
|
||||||
site: z.string(),
|
site: z.string(),
|
||||||
// protocol: z.enum(["tcp", "udp"]).optional(),
|
// protocol: z.enum(["tcp", "udp"]).optional(),
|
||||||
// proxyPort: z.int().positive().optional(),
|
// proxyPort: z.int().positive().optional(),
|
||||||
// destinationPort: z.int().positive().optional(),
|
"destination-port": z.int().positive().optional(),
|
||||||
destination: z.string().min(1),
|
destination: z.string().min(1),
|
||||||
// enabled: z.boolean().default(true),
|
// enabled: z.boolean().default(true),
|
||||||
"tcp-ports": portRangeStringSchema.optional().default("*"),
|
"tcp-ports": portRangeStringSchema.optional().default("*"),
|
||||||
|
|||||||
@@ -0,0 +1,277 @@
|
|||||||
|
/*
|
||||||
|
* This file is part of a proprietary work.
|
||||||
|
*
|
||||||
|
* Copyright (c) 2025 Fossorial, Inc.
|
||||||
|
* All rights reserved.
|
||||||
|
*
|
||||||
|
* This file is licensed under the Fossorial Commercial License.
|
||||||
|
* You may not use this file except in compliance with the License.
|
||||||
|
* Unauthorized use, copying, modification, or distribution is strictly prohibited.
|
||||||
|
*
|
||||||
|
* This file is not licensed under the AGPLv3.
|
||||||
|
*/
|
||||||
|
|
||||||
|
import fs from "fs";
|
||||||
|
import crypto from "crypto";
|
||||||
|
import { certificates, domains, db } from "@server/db";
|
||||||
|
import { and, eq } from "drizzle-orm";
|
||||||
|
import { encryptData, decryptData } from "@server/lib/encryption";
|
||||||
|
import logger from "@server/logger";
|
||||||
|
import config from "#private/lib/config";
|
||||||
|
|
||||||
|
interface AcmeCert {
|
||||||
|
domain: { main: string; sans?: string[] };
|
||||||
|
certificate: string;
|
||||||
|
key: string;
|
||||||
|
Store: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface AcmeJson {
|
||||||
|
[resolver: string]: {
|
||||||
|
Certificates: AcmeCert[];
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function getEncryptionKey(): Buffer {
|
||||||
|
const keyHex = config.getRawPrivateConfig().server.encryption_key;
|
||||||
|
if (!keyHex) {
|
||||||
|
throw new Error("acmeCertSync: encryption key is not configured");
|
||||||
|
}
|
||||||
|
return Buffer.from(keyHex, "hex");
|
||||||
|
}
|
||||||
|
|
||||||
|
async function findDomainId(certDomain: string): Promise<string | null> {
|
||||||
|
// Strip wildcard prefix before lookup (*.example.com -> example.com)
|
||||||
|
const lookupDomain = certDomain.startsWith("*.")
|
||||||
|
? certDomain.slice(2)
|
||||||
|
: certDomain;
|
||||||
|
|
||||||
|
// 1. Exact baseDomain match (any domain type)
|
||||||
|
const exactMatch = await db
|
||||||
|
.select({ domainId: domains.domainId })
|
||||||
|
.from(domains)
|
||||||
|
.where(eq(domains.baseDomain, lookupDomain))
|
||||||
|
.limit(1);
|
||||||
|
|
||||||
|
if (exactMatch.length > 0) {
|
||||||
|
return exactMatch[0].domainId;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Walk up the domain hierarchy looking for a wildcard-type domain whose
|
||||||
|
// baseDomain is a suffix of the cert domain. e.g. cert "sub.example.com"
|
||||||
|
// matches a wildcard domain with baseDomain "example.com".
|
||||||
|
const parts = lookupDomain.split(".");
|
||||||
|
for (let i = 1; i < parts.length; i++) {
|
||||||
|
const candidate = parts.slice(i).join(".");
|
||||||
|
if (!candidate) continue;
|
||||||
|
|
||||||
|
const wildcardMatch = await db
|
||||||
|
.select({ domainId: domains.domainId })
|
||||||
|
.from(domains)
|
||||||
|
.where(
|
||||||
|
and(
|
||||||
|
eq(domains.baseDomain, candidate),
|
||||||
|
eq(domains.type, "wildcard")
|
||||||
|
)
|
||||||
|
)
|
||||||
|
.limit(1);
|
||||||
|
|
||||||
|
if (wildcardMatch.length > 0) {
|
||||||
|
return wildcardMatch[0].domainId;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function extractFirstCert(pemBundle: string): string | null {
|
||||||
|
const match = pemBundle.match(
|
||||||
|
/-----BEGIN CERTIFICATE-----[\s\S]+?-----END CERTIFICATE-----/
|
||||||
|
);
|
||||||
|
return match ? match[0] : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
async function syncAcmeCerts(
|
||||||
|
acmeJsonPath: string,
|
||||||
|
resolver: string
|
||||||
|
): Promise<void> {
|
||||||
|
let raw: string;
|
||||||
|
try {
|
||||||
|
raw = fs.readFileSync(acmeJsonPath, "utf8");
|
||||||
|
} catch (err) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: could not read ${acmeJsonPath}: ${err}`
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let acmeJson: AcmeJson;
|
||||||
|
try {
|
||||||
|
acmeJson = JSON.parse(raw);
|
||||||
|
} catch (err) {
|
||||||
|
logger.debug(`acmeCertSync: could not parse acme.json: ${err}`);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const resolverData = acmeJson[resolver];
|
||||||
|
if (!resolverData || !Array.isArray(resolverData.Certificates)) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: no certificates found for resolver "${resolver}"`
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const encryptionKey = getEncryptionKey();
|
||||||
|
|
||||||
|
for (const cert of resolverData.Certificates) {
|
||||||
|
const domain = cert.domain?.main;
|
||||||
|
|
||||||
|
if (!domain) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: skipping cert with missing domain`
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!cert.certificate || !cert.key) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: skipping cert for ${domain} - empty certificate or key field`
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
const certPem = Buffer.from(cert.certificate, "base64").toString(
|
||||||
|
"utf8"
|
||||||
|
);
|
||||||
|
const keyPem = Buffer.from(cert.key, "base64").toString("utf8");
|
||||||
|
|
||||||
|
if (!certPem.trim() || !keyPem.trim()) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: skipping cert for ${domain} - blank PEM after base64 decode`
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check if cert already exists in DB
|
||||||
|
const existing = await db
|
||||||
|
.select()
|
||||||
|
.from(certificates)
|
||||||
|
.where(eq(certificates.domain, domain))
|
||||||
|
.limit(1);
|
||||||
|
|
||||||
|
if (existing.length > 0 && existing[0].certFile) {
|
||||||
|
try {
|
||||||
|
const storedCertPem = decryptData(
|
||||||
|
existing[0].certFile,
|
||||||
|
encryptionKey
|
||||||
|
);
|
||||||
|
if (storedCertPem === certPem) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: cert for ${domain} is unchanged, skipping`
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
// Decryption failure means we should proceed with the update
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: could not decrypt stored cert for ${domain}, will update: ${err}`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse cert expiry from the first cert in the PEM bundle
|
||||||
|
let expiresAt: number | null = null;
|
||||||
|
const firstCertPem = extractFirstCert(certPem);
|
||||||
|
if (firstCertPem) {
|
||||||
|
try {
|
||||||
|
const x509 = new crypto.X509Certificate(firstCertPem);
|
||||||
|
expiresAt = Math.floor(
|
||||||
|
new Date(x509.validTo).getTime() / 1000
|
||||||
|
);
|
||||||
|
} catch (err) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: could not parse cert expiry for ${domain}: ${err}`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const wildcard = domain.startsWith("*.");
|
||||||
|
const encryptedCert = encryptData(certPem, encryptionKey);
|
||||||
|
const encryptedKey = encryptData(keyPem, encryptionKey);
|
||||||
|
const now = Math.floor(Date.now() / 1000);
|
||||||
|
|
||||||
|
const domainId = await findDomainId(domain);
|
||||||
|
if (domainId) {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: resolved domainId "${domainId}" for cert domain "${domain}"`
|
||||||
|
);
|
||||||
|
} else {
|
||||||
|
logger.debug(
|
||||||
|
`acmeCertSync: no matching domain record found for cert domain "${domain}"`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (existing.length > 0) {
|
||||||
|
await db
|
||||||
|
.update(certificates)
|
||||||
|
.set({
|
||||||
|
certFile: encryptedCert,
|
||||||
|
keyFile: encryptedKey,
|
||||||
|
status: "valid",
|
||||||
|
expiresAt,
|
||||||
|
updatedAt: now,
|
||||||
|
wildcard,
|
||||||
|
...(domainId !== null && { domainId })
|
||||||
|
})
|
||||||
|
.where(eq(certificates.domain, domain));
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
`acmeCertSync: updated certificate for ${domain} (expires ${expiresAt ? new Date(expiresAt * 1000).toISOString() : "unknown"})`
|
||||||
|
);
|
||||||
|
} else {
|
||||||
|
await db.insert(certificates).values({
|
||||||
|
domain,
|
||||||
|
domainId,
|
||||||
|
certFile: encryptedCert,
|
||||||
|
keyFile: encryptedKey,
|
||||||
|
status: "valid",
|
||||||
|
expiresAt,
|
||||||
|
createdAt: now,
|
||||||
|
updatedAt: now,
|
||||||
|
wildcard
|
||||||
|
});
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
`acmeCertSync: inserted new certificate for ${domain} (expires ${expiresAt ? new Date(expiresAt * 1000).toISOString() : "unknown"})`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export function initAcmeCertSync(): void {
|
||||||
|
const privateConfig = config.getRawPrivateConfig();
|
||||||
|
|
||||||
|
if (!privateConfig.flags?.enable_acme_cert_sync) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const acmeJsonPath =
|
||||||
|
privateConfig.acme?.acme_json_path ?? "config/letsencrypt/acme.json";
|
||||||
|
const resolver = privateConfig.acme?.resolver ?? "letsencrypt";
|
||||||
|
const intervalMs = privateConfig.acme?.sync_interval_ms ?? 5000;
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
`acmeCertSync: starting ACME cert sync from "${acmeJsonPath}" using resolver "${resolver}" every ${intervalMs}ms`
|
||||||
|
);
|
||||||
|
|
||||||
|
// Run immediately on init, then on the configured interval
|
||||||
|
syncAcmeCerts(acmeJsonPath, resolver).catch((err) => {
|
||||||
|
logger.error(`acmeCertSync: error during initial sync: ${err}`);
|
||||||
|
});
|
||||||
|
|
||||||
|
setInterval(() => {
|
||||||
|
syncAcmeCerts(acmeJsonPath, resolver).catch((err) => {
|
||||||
|
logger.error(`acmeCertSync: error during sync: ${err}`);
|
||||||
|
});
|
||||||
|
}, intervalMs);
|
||||||
|
}
|
||||||
@@ -95,10 +95,21 @@ export const privateConfigSchema = z.object({
|
|||||||
.object({
|
.object({
|
||||||
enable_redis: z.boolean().optional().default(false),
|
enable_redis: z.boolean().optional().default(false),
|
||||||
use_pangolin_dns: z.boolean().optional().default(false),
|
use_pangolin_dns: z.boolean().optional().default(false),
|
||||||
use_org_only_idp: z.boolean().optional()
|
use_org_only_idp: z.boolean().optional(),
|
||||||
|
enable_acme_cert_sync: z.boolean().optional().default(false)
|
||||||
})
|
})
|
||||||
.optional()
|
.optional()
|
||||||
.prefault({}),
|
.prefault({}),
|
||||||
|
acme: z
|
||||||
|
.object({
|
||||||
|
acme_json_path: z
|
||||||
|
.string()
|
||||||
|
.optional()
|
||||||
|
.default("config/letsencrypt/acme.json"),
|
||||||
|
resolver: z.string().optional().default("letsencrypt"),
|
||||||
|
sync_interval_ms: z.number().optional().default(5000)
|
||||||
|
})
|
||||||
|
.optional(),
|
||||||
branding: z
|
branding: z
|
||||||
.object({
|
.object({
|
||||||
app_name: z.string().optional(),
|
app_name: z.string().optional(),
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import { db } from "@server/db";
|
import { db } from "@server/db";
|
||||||
import { sites, clients, olms } from "@server/db";
|
import { sites, clients, olms } from "@server/db";
|
||||||
import { eq, inArray } from "drizzle-orm";
|
import { inArray } from "drizzle-orm";
|
||||||
import logger from "@server/logger";
|
import logger from "@server/logger";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -21,7 +21,7 @@ import logger from "@server/logger";
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
const FLUSH_INTERVAL_MS = 10_000; // Flush every 10 seconds
|
const FLUSH_INTERVAL_MS = 10_000; // Flush every 10 seconds
|
||||||
const MAX_RETRIES = 2;
|
const MAX_RETRIES = 5;
|
||||||
const BASE_DELAY_MS = 50;
|
const BASE_DELAY_MS = 50;
|
||||||
|
|
||||||
// ── Site (newt) pings ──────────────────────────────────────────────────
|
// ── Site (newt) pings ──────────────────────────────────────────────────
|
||||||
@@ -36,6 +36,14 @@ const pendingOlmArchiveResets: Set<string> = new Set();
|
|||||||
|
|
||||||
let flushTimer: NodeJS.Timeout | null = null;
|
let flushTimer: NodeJS.Timeout | null = null;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Guard that prevents two flush cycles from running concurrently.
|
||||||
|
* setInterval does not await async callbacks, so without this a slow flush
|
||||||
|
* (e.g. due to DB latency) would overlap with the next scheduled cycle and
|
||||||
|
* the two concurrent bulk UPDATEs would deadlock each other.
|
||||||
|
*/
|
||||||
|
let isFlushing = false;
|
||||||
|
|
||||||
// ── Public API ─────────────────────────────────────────────────────────
|
// ── Public API ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -72,6 +80,12 @@ export function recordClientPing(
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Flush all accumulated site pings to the database.
|
* Flush all accumulated site pings to the database.
|
||||||
|
*
|
||||||
|
* Each batch of up to BATCH_SIZE rows is written with a **single** UPDATE
|
||||||
|
* statement. We use the maximum timestamp across the batch so that `lastPing`
|
||||||
|
* reflects the most recent ping seen for any site in the group. This avoids
|
||||||
|
* the multi-statement transaction that previously created additional
|
||||||
|
* row-lock ordering hazards.
|
||||||
*/
|
*/
|
||||||
async function flushSitePingsToDb(): Promise<void> {
|
async function flushSitePingsToDb(): Promise<void> {
|
||||||
if (pendingSitePings.size === 0) {
|
if (pendingSitePings.size === 0) {
|
||||||
@@ -83,55 +97,35 @@ async function flushSitePingsToDb(): Promise<void> {
|
|||||||
const pingsToFlush = new Map(pendingSitePings);
|
const pingsToFlush = new Map(pendingSitePings);
|
||||||
pendingSitePings.clear();
|
pendingSitePings.clear();
|
||||||
|
|
||||||
// Sort by siteId for consistent lock ordering (prevents deadlocks)
|
const entries = Array.from(pingsToFlush.entries());
|
||||||
const sortedEntries = Array.from(pingsToFlush.entries()).sort(
|
|
||||||
([a], [b]) => a - b
|
|
||||||
);
|
|
||||||
|
|
||||||
const BATCH_SIZE = 50;
|
const BATCH_SIZE = 50;
|
||||||
for (let i = 0; i < sortedEntries.length; i += BATCH_SIZE) {
|
for (let i = 0; i < entries.length; i += BATCH_SIZE) {
|
||||||
const batch = sortedEntries.slice(i, i + BATCH_SIZE);
|
const batch = entries.slice(i, i + BATCH_SIZE);
|
||||||
|
|
||||||
|
// Use the latest timestamp in the batch so that `lastPing` always
|
||||||
|
// moves forward. Using a single timestamp for the whole batch means
|
||||||
|
// we only ever need one UPDATE statement (no transaction).
|
||||||
|
const maxTimestamp = Math.max(...batch.map(([, ts]) => ts));
|
||||||
|
const siteIds = batch.map(([id]) => id);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await withRetry(async () => {
|
await withRetry(async () => {
|
||||||
// Group by timestamp for efficient bulk updates
|
await db
|
||||||
const byTimestamp = new Map<number, number[]>();
|
.update(sites)
|
||||||
for (const [siteId, timestamp] of batch) {
|
.set({
|
||||||
const group = byTimestamp.get(timestamp) || [];
|
online: true,
|
||||||
group.push(siteId);
|
lastPing: maxTimestamp
|
||||||
byTimestamp.set(timestamp, group);
|
})
|
||||||
}
|
.where(inArray(sites.siteId, siteIds));
|
||||||
|
|
||||||
if (byTimestamp.size === 1) {
|
|
||||||
const [timestamp, siteIds] = Array.from(
|
|
||||||
byTimestamp.entries()
|
|
||||||
)[0];
|
|
||||||
await db
|
|
||||||
.update(sites)
|
|
||||||
.set({
|
|
||||||
online: true,
|
|
||||||
lastPing: timestamp
|
|
||||||
})
|
|
||||||
.where(inArray(sites.siteId, siteIds));
|
|
||||||
} else {
|
|
||||||
await db.transaction(async (tx) => {
|
|
||||||
for (const [timestamp, siteIds] of byTimestamp) {
|
|
||||||
await tx
|
|
||||||
.update(sites)
|
|
||||||
.set({
|
|
||||||
online: true,
|
|
||||||
lastPing: timestamp
|
|
||||||
})
|
|
||||||
.where(inArray(sites.siteId, siteIds));
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}, "flushSitePingsToDb");
|
}, "flushSitePingsToDb");
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.error(
|
logger.error(
|
||||||
`Failed to flush site ping batch (${batch.length} sites), re-queuing for next cycle`,
|
`Failed to flush site ping batch (${batch.length} sites), re-queuing for next cycle`,
|
||||||
{ error }
|
{ error }
|
||||||
);
|
);
|
||||||
|
// Re-queue only if the preserved timestamp is newer than any
|
||||||
|
// update that may have landed since we snapshotted.
|
||||||
for (const [siteId, timestamp] of batch) {
|
for (const [siteId, timestamp] of batch) {
|
||||||
const existing = pendingSitePings.get(siteId);
|
const existing = pendingSitePings.get(siteId);
|
||||||
if (!existing || existing < timestamp) {
|
if (!existing || existing < timestamp) {
|
||||||
@@ -144,6 +138,8 @@ async function flushSitePingsToDb(): Promise<void> {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Flush all accumulated client (OLM) pings to the database.
|
* Flush all accumulated client (OLM) pings to the database.
|
||||||
|
*
|
||||||
|
* Same single-UPDATE-per-batch approach as `flushSitePingsToDb`.
|
||||||
*/
|
*/
|
||||||
async function flushClientPingsToDb(): Promise<void> {
|
async function flushClientPingsToDb(): Promise<void> {
|
||||||
if (pendingClientPings.size === 0 && pendingOlmArchiveResets.size === 0) {
|
if (pendingClientPings.size === 0 && pendingOlmArchiveResets.size === 0) {
|
||||||
@@ -159,51 +155,25 @@ async function flushClientPingsToDb(): Promise<void> {
|
|||||||
|
|
||||||
// ── Flush client pings ─────────────────────────────────────────────
|
// ── Flush client pings ─────────────────────────────────────────────
|
||||||
if (pingsToFlush.size > 0) {
|
if (pingsToFlush.size > 0) {
|
||||||
const sortedEntries = Array.from(pingsToFlush.entries()).sort(
|
const entries = Array.from(pingsToFlush.entries());
|
||||||
([a], [b]) => a - b
|
|
||||||
);
|
|
||||||
|
|
||||||
const BATCH_SIZE = 50;
|
const BATCH_SIZE = 50;
|
||||||
for (let i = 0; i < sortedEntries.length; i += BATCH_SIZE) {
|
for (let i = 0; i < entries.length; i += BATCH_SIZE) {
|
||||||
const batch = sortedEntries.slice(i, i + BATCH_SIZE);
|
const batch = entries.slice(i, i + BATCH_SIZE);
|
||||||
|
|
||||||
|
const maxTimestamp = Math.max(...batch.map(([, ts]) => ts));
|
||||||
|
const clientIds = batch.map(([id]) => id);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await withRetry(async () => {
|
await withRetry(async () => {
|
||||||
const byTimestamp = new Map<number, number[]>();
|
await db
|
||||||
for (const [clientId, timestamp] of batch) {
|
.update(clients)
|
||||||
const group = byTimestamp.get(timestamp) || [];
|
.set({
|
||||||
group.push(clientId);
|
lastPing: maxTimestamp,
|
||||||
byTimestamp.set(timestamp, group);
|
online: true,
|
||||||
}
|
archived: false
|
||||||
|
})
|
||||||
if (byTimestamp.size === 1) {
|
.where(inArray(clients.clientId, clientIds));
|
||||||
const [timestamp, clientIds] = Array.from(
|
|
||||||
byTimestamp.entries()
|
|
||||||
)[0];
|
|
||||||
await db
|
|
||||||
.update(clients)
|
|
||||||
.set({
|
|
||||||
lastPing: timestamp,
|
|
||||||
online: true,
|
|
||||||
archived: false
|
|
||||||
})
|
|
||||||
.where(inArray(clients.clientId, clientIds));
|
|
||||||
} else {
|
|
||||||
await db.transaction(async (tx) => {
|
|
||||||
for (const [timestamp, clientIds] of byTimestamp) {
|
|
||||||
await tx
|
|
||||||
.update(clients)
|
|
||||||
.set({
|
|
||||||
lastPing: timestamp,
|
|
||||||
online: true,
|
|
||||||
archived: false
|
|
||||||
})
|
|
||||||
.where(
|
|
||||||
inArray(clients.clientId, clientIds)
|
|
||||||
);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}, "flushClientPingsToDb");
|
}, "flushClientPingsToDb");
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.error(
|
logger.error(
|
||||||
@@ -260,7 +230,12 @@ export async function flushPingsToDb(): Promise<void> {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Simple retry wrapper with exponential backoff for transient errors
|
* Simple retry wrapper with exponential backoff for transient errors
|
||||||
* (connection timeouts, unexpected disconnects).
|
* (deadlocks, connection timeouts, unexpected disconnects).
|
||||||
|
*
|
||||||
|
* PostgreSQL deadlocks (40P01) are always safe to retry: the database
|
||||||
|
* guarantees exactly one winner per deadlock pair, so the loser just needs
|
||||||
|
* to try again. MAX_RETRIES is intentionally higher than typical connection
|
||||||
|
* retry budgets to give deadlock victims enough chances to succeed.
|
||||||
*/
|
*/
|
||||||
async function withRetry<T>(
|
async function withRetry<T>(
|
||||||
operation: () => Promise<T>,
|
operation: () => Promise<T>,
|
||||||
@@ -277,7 +252,8 @@ async function withRetry<T>(
|
|||||||
const jitter = Math.random() * baseDelay;
|
const jitter = Math.random() * baseDelay;
|
||||||
const delay = baseDelay + jitter;
|
const delay = baseDelay + jitter;
|
||||||
logger.warn(
|
logger.warn(
|
||||||
`Transient DB error in ${context}, retrying attempt ${attempt}/${MAX_RETRIES} after ${delay.toFixed(0)}ms`
|
`Transient DB error in ${context}, retrying attempt ${attempt}/${MAX_RETRIES} after ${delay.toFixed(0)}ms`,
|
||||||
|
{ code: error?.code ?? error?.cause?.code }
|
||||||
);
|
);
|
||||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||||
continue;
|
continue;
|
||||||
@@ -288,14 +264,14 @@ async function withRetry<T>(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Detect transient connection errors that are safe to retry.
|
* Detect transient errors that are safe to retry.
|
||||||
*/
|
*/
|
||||||
function isTransientError(error: any): boolean {
|
function isTransientError(error: any): boolean {
|
||||||
if (!error) return false;
|
if (!error) return false;
|
||||||
|
|
||||||
const message = (error.message || "").toLowerCase();
|
const message = (error.message || "").toLowerCase();
|
||||||
const causeMessage = (error.cause?.message || "").toLowerCase();
|
const causeMessage = (error.cause?.message || "").toLowerCase();
|
||||||
const code = error.code || "";
|
const code = error.code || error.cause?.code || "";
|
||||||
|
|
||||||
// Connection timeout / terminated
|
// Connection timeout / terminated
|
||||||
if (
|
if (
|
||||||
@@ -308,12 +284,17 @@ function isTransientError(error: any): boolean {
|
|||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
// PostgreSQL deadlock
|
// PostgreSQL deadlock detected — always safe to retry (one winner guaranteed)
|
||||||
if (code === "40P01" || message.includes("deadlock")) {
|
if (code === "40P01" || message.includes("deadlock")) {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
// ECONNRESET, ECONNREFUSED, EPIPE
|
// PostgreSQL serialization failure
|
||||||
|
if (code === "40001") {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
// ECONNRESET, ECONNREFUSED, EPIPE, ETIMEDOUT
|
||||||
if (
|
if (
|
||||||
code === "ECONNRESET" ||
|
code === "ECONNRESET" ||
|
||||||
code === "ECONNREFUSED" ||
|
code === "ECONNREFUSED" ||
|
||||||
@@ -337,12 +318,26 @@ export function startPingAccumulator(): void {
|
|||||||
}
|
}
|
||||||
|
|
||||||
flushTimer = setInterval(async () => {
|
flushTimer = setInterval(async () => {
|
||||||
|
// Skip this tick if the previous flush is still in progress.
|
||||||
|
// setInterval does not await async callbacks, so without this guard
|
||||||
|
// two flush cycles can run concurrently and deadlock each other on
|
||||||
|
// overlapping bulk UPDATE statements.
|
||||||
|
if (isFlushing) {
|
||||||
|
logger.debug(
|
||||||
|
"Ping accumulator: previous flush still in progress, skipping cycle"
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
isFlushing = true;
|
||||||
try {
|
try {
|
||||||
await flushPingsToDb();
|
await flushPingsToDb();
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.error("Unhandled error in ping accumulator flush", {
|
logger.error("Unhandled error in ping accumulator flush", {
|
||||||
error
|
error
|
||||||
});
|
});
|
||||||
|
} finally {
|
||||||
|
isFlushing = false;
|
||||||
}
|
}
|
||||||
}, FLUSH_INTERVAL_MS);
|
}, FLUSH_INTERVAL_MS);
|
||||||
|
|
||||||
@@ -364,7 +359,22 @@ export async function stopPingAccumulator(): Promise<void> {
|
|||||||
flushTimer = null;
|
flushTimer = null;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Final flush to persist any remaining pings
|
// Final flush to persist any remaining pings.
|
||||||
|
// Wait for any in-progress flush to finish first so we don't race.
|
||||||
|
if (isFlushing) {
|
||||||
|
logger.debug(
|
||||||
|
"Ping accumulator: waiting for in-progress flush before stopping…"
|
||||||
|
);
|
||||||
|
await new Promise<void>((resolve) => {
|
||||||
|
const poll = setInterval(() => {
|
||||||
|
if (!isFlushing) {
|
||||||
|
clearInterval(poll);
|
||||||
|
resolve();
|
||||||
|
}
|
||||||
|
}, 50);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await flushPingsToDb();
|
await flushPingsToDb();
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -379,4 +389,4 @@ export async function stopPingAccumulator(): Promise<void> {
|
|||||||
*/
|
*/
|
||||||
export function getPendingPingCount(): number {
|
export function getPendingPingCount(): number {
|
||||||
return pendingSitePings.size + pendingClientPings.size;
|
return pendingSitePings.size + pendingClientPings.size;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -36,11 +36,11 @@ const createSiteResourceParamsSchema = z.strictObject({
|
|||||||
const createSiteResourceSchema = z
|
const createSiteResourceSchema = z
|
||||||
.strictObject({
|
.strictObject({
|
||||||
name: z.string().min(1).max(255),
|
name: z.string().min(1).max(255),
|
||||||
mode: z.enum(["host", "cidr", "port"]),
|
mode: z.enum(["host", "cidr", "port", "http", "https"]),
|
||||||
siteId: z.int(),
|
siteId: z.int(),
|
||||||
// protocol: z.enum(["tcp", "udp"]).optional(),
|
// protocol: z.enum(["tcp", "udp"]).optional(),
|
||||||
// proxyPort: z.int().positive().optional(),
|
// proxyPort: z.int().positive().optional(),
|
||||||
// destinationPort: z.int().positive().optional(),
|
destinationPort: z.int().positive().optional(),
|
||||||
destination: z.string().min(1),
|
destination: z.string().min(1),
|
||||||
enabled: z.boolean().default(true),
|
enabled: z.boolean().default(true),
|
||||||
alias: z
|
alias: z
|
||||||
@@ -163,7 +163,7 @@ export async function createSiteResource(
|
|||||||
mode,
|
mode,
|
||||||
// protocol,
|
// protocol,
|
||||||
// proxyPort,
|
// proxyPort,
|
||||||
// destinationPort,
|
destinationPort,
|
||||||
destination,
|
destination,
|
||||||
enabled,
|
enabled,
|
||||||
alias,
|
alias,
|
||||||
@@ -295,6 +295,7 @@ export async function createSiteResource(
|
|||||||
name,
|
name,
|
||||||
mode: mode as "host" | "cidr",
|
mode: mode as "host" | "cidr",
|
||||||
destination,
|
destination,
|
||||||
|
destinationPort,
|
||||||
enabled,
|
enabled,
|
||||||
alias,
|
alias,
|
||||||
aliasAddress,
|
aliasAddress,
|
||||||
|
|||||||
@@ -51,10 +51,10 @@ const updateSiteResourceSchema = z
|
|||||||
)
|
)
|
||||||
.optional(),
|
.optional(),
|
||||||
// mode: z.enum(["host", "cidr", "port"]).optional(),
|
// mode: z.enum(["host", "cidr", "port"]).optional(),
|
||||||
mode: z.enum(["host", "cidr"]).optional(),
|
mode: z.enum(["host", "cidr", "http", "https"]).optional(),
|
||||||
// protocol: z.enum(["tcp", "udp"]).nullish(),
|
// protocol: z.enum(["tcp", "udp"]).nullish(),
|
||||||
// proxyPort: z.int().positive().nullish(),
|
// proxyPort: z.int().positive().nullish(),
|
||||||
// destinationPort: z.int().positive().nullish(),
|
destinationPort: z.int().positive().nullish(),
|
||||||
destination: z.string().min(1).optional(),
|
destination: z.string().min(1).optional(),
|
||||||
enabled: z.boolean().optional(),
|
enabled: z.boolean().optional(),
|
||||||
alias: z
|
alias: z
|
||||||
@@ -176,6 +176,7 @@ export async function updateSiteResource(
|
|||||||
niceId,
|
niceId,
|
||||||
mode,
|
mode,
|
||||||
destination,
|
destination,
|
||||||
|
destinationPort,
|
||||||
alias,
|
alias,
|
||||||
enabled,
|
enabled,
|
||||||
userIds,
|
userIds,
|
||||||
@@ -347,6 +348,7 @@ export async function updateSiteResource(
|
|||||||
niceId,
|
niceId,
|
||||||
mode,
|
mode,
|
||||||
destination,
|
destination,
|
||||||
|
destinationPort,
|
||||||
enabled,
|
enabled,
|
||||||
alias: alias && alias.trim() ? alias : null,
|
alias: alias && alias.trim() ? alias : null,
|
||||||
tcpPortRangeString,
|
tcpPortRangeString,
|
||||||
@@ -450,6 +452,7 @@ export async function updateSiteResource(
|
|||||||
siteId: siteId,
|
siteId: siteId,
|
||||||
mode: mode,
|
mode: mode,
|
||||||
destination: destination,
|
destination: destination,
|
||||||
|
destinationPort: destinationPort,
|
||||||
enabled: enabled,
|
enabled: enabled,
|
||||||
alias: alias && alias.trim() ? alias : null,
|
alias: alias && alias.trim() ? alias : null,
|
||||||
tcpPortRangeString: tcpPortRangeString,
|
tcpPortRangeString: tcpPortRangeString,
|
||||||
|
|||||||
@@ -235,7 +235,9 @@ export default async function migration() {
|
|||||||
for (const row of existingUserInviteRoles) {
|
for (const row of existingUserInviteRoles) {
|
||||||
await db.execute(sql`
|
await db.execute(sql`
|
||||||
INSERT INTO "userInviteRoles" ("inviteId", "roleId")
|
INSERT INTO "userInviteRoles" ("inviteId", "roleId")
|
||||||
VALUES (${row.inviteId}, ${row.roleId})
|
SELECT ${row.inviteId}, ${row.roleId}
|
||||||
|
WHERE EXISTS (SELECT 1 FROM "userInvites" WHERE "inviteId" = ${row.inviteId})
|
||||||
|
AND EXISTS (SELECT 1 FROM "roles" WHERE "roleId" = ${row.roleId})
|
||||||
ON CONFLICT DO NOTHING
|
ON CONFLICT DO NOTHING
|
||||||
`);
|
`);
|
||||||
}
|
}
|
||||||
@@ -258,7 +260,10 @@ export default async function migration() {
|
|||||||
for (const row of existingUserOrgRoles) {
|
for (const row of existingUserOrgRoles) {
|
||||||
await db.execute(sql`
|
await db.execute(sql`
|
||||||
INSERT INTO "userOrgRoles" ("userId", "orgId", "roleId")
|
INSERT INTO "userOrgRoles" ("userId", "orgId", "roleId")
|
||||||
VALUES (${row.userId}, ${row.orgId}, ${row.roleId})
|
SELECT ${row.userId}, ${row.orgId}, ${row.roleId}
|
||||||
|
WHERE EXISTS (SELECT 1 FROM "user" WHERE "id" = ${row.userId})
|
||||||
|
AND EXISTS (SELECT 1 FROM "orgs" WHERE "orgId" = ${row.orgId})
|
||||||
|
AND EXISTS (SELECT 1 FROM "roles" WHERE "roleId" = ${row.roleId})
|
||||||
ON CONFLICT DO NOTHING
|
ON CONFLICT DO NOTHING
|
||||||
`);
|
`);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -145,7 +145,7 @@ export default async function migration() {
|
|||||||
).run();
|
).run();
|
||||||
|
|
||||||
db.prepare(
|
db.prepare(
|
||||||
`INSERT INTO '__new_userOrgs'("userId", "orgId", "isOwner", "autoProvisioned", "pamUsername") SELECT "userId", "orgId", "isOwner", "autoProvisioned", "pamUsername" FROM 'userOrgs';`
|
`INSERT INTO '__new_userOrgs'("userId", "orgId", "isOwner", "autoProvisioned", "pamUsername") SELECT "userId", "orgId", "isOwner", "autoProvisioned", "pamUsername" FROM 'userOrgs' WHERE EXISTS (SELECT 1 FROM 'user' WHERE id = userOrgs.userId) AND EXISTS (SELECT 1 FROM 'orgs' WHERE orgId = userOrgs.orgId);`
|
||||||
).run();
|
).run();
|
||||||
db.prepare(`DROP TABLE 'userOrgs';`).run();
|
db.prepare(`DROP TABLE 'userOrgs';`).run();
|
||||||
db.prepare(
|
db.prepare(
|
||||||
@@ -246,12 +246,15 @@ export default async function migration() {
|
|||||||
// Re-insert the preserved invite role assignments into the new userInviteRoles table
|
// Re-insert the preserved invite role assignments into the new userInviteRoles table
|
||||||
if (existingUserInviteRoles.length > 0) {
|
if (existingUserInviteRoles.length > 0) {
|
||||||
const insertUserInviteRole = db.prepare(
|
const insertUserInviteRole = db.prepare(
|
||||||
`INSERT OR IGNORE INTO 'userInviteRoles' ("inviteId", "roleId") VALUES (?, ?)`
|
`INSERT OR IGNORE INTO 'userInviteRoles' ("inviteId", "roleId")
|
||||||
|
SELECT ?, ?
|
||||||
|
WHERE EXISTS (SELECT 1 FROM 'userInvites' WHERE inviteId = ?)
|
||||||
|
AND EXISTS (SELECT 1 FROM 'roles' WHERE roleId = ?)`
|
||||||
);
|
);
|
||||||
|
|
||||||
const insertAll = db.transaction(() => {
|
const insertAll = db.transaction(() => {
|
||||||
for (const row of existingUserInviteRoles) {
|
for (const row of existingUserInviteRoles) {
|
||||||
insertUserInviteRole.run(row.inviteId, row.roleId);
|
insertUserInviteRole.run(row.inviteId, row.roleId, row.inviteId, row.roleId);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -265,12 +268,16 @@ export default async function migration() {
|
|||||||
// Re-insert the preserved role assignments into the new userOrgRoles table
|
// Re-insert the preserved role assignments into the new userOrgRoles table
|
||||||
if (existingUserOrgRoles.length > 0) {
|
if (existingUserOrgRoles.length > 0) {
|
||||||
const insertUserOrgRole = db.prepare(
|
const insertUserOrgRole = db.prepare(
|
||||||
`INSERT OR IGNORE INTO 'userOrgRoles' ("userId", "orgId", "roleId") VALUES (?, ?, ?)`
|
`INSERT OR IGNORE INTO 'userOrgRoles' ("userId", "orgId", "roleId")
|
||||||
|
SELECT ?, ?, ?
|
||||||
|
WHERE EXISTS (SELECT 1 FROM 'user' WHERE id = ?)
|
||||||
|
AND EXISTS (SELECT 1 FROM 'orgs' WHERE orgId = ?)
|
||||||
|
AND EXISTS (SELECT 1 FROM 'roles' WHERE roleId = ?)`
|
||||||
);
|
);
|
||||||
|
|
||||||
const insertAll = db.transaction(() => {
|
const insertAll = db.transaction(() => {
|
||||||
for (const row of existingUserOrgRoles) {
|
for (const row of existingUserOrgRoles) {
|
||||||
insertUserOrgRole.run(row.userId, row.orgId, row.roleId);
|
insertUserOrgRole.run(row.userId, row.orgId, row.roleId, row.userId, row.orgId, row.roleId);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user