From ed28d34945f4d4a3c789a274ff7abb16d13039c5 Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 24 Sep 2026 11:56:10 -0400 Subject: [PATCH] Support batched relay,unrelay,local,unlocal messages --- server/private/lib/dns/server.ts | 45 ++++-- server/private/lib/readConfigFile.ts | 10 ++ server/routers/newt/peers.ts | 34 +++++ server/routers/olm/batchUtils.ts | 55 ++++++++ server/routers/olm/handleOlmLocalMessage.ts | 88 +++++++++--- server/routers/olm/handleOlmRelayMessage.ts | 133 ++++++++++++++---- server/routers/olm/handleOlmUnLocalMessage.ts | 120 ++++++++++++---- server/routers/olm/handleOlmUnRelayMessage.ts | 122 ++++++++++++---- 8 files changed, 492 insertions(+), 115 deletions(-) create mode 100644 server/routers/olm/batchUtils.ts diff --git a/server/private/lib/dns/server.ts b/server/private/lib/dns/server.ts index 10c46c358..4198f3d2e 100644 --- a/server/private/lib/dns/server.ts +++ b/server/private/lib/dns/server.ts @@ -384,7 +384,8 @@ export class AuthoritativeDNSServer { // Get records from cache or database const records = await this.getResourceRecordsByFullDomain( queryName, - queryType + queryType, + baseDomain! ); if (records.length > 0) { @@ -475,15 +476,30 @@ export class AuthoritativeDNSServer { const labels = queryName.replace(/\.$/, "").split("."); - // Fast path: O(1) in-memory Set lookup — no DB or network I/O - if (this.allDomains.size > 0) { - for (let i = 0; i < labels.length; i++) { - const candidate = labels.slice(i).join("."); - if (this.allDomains.has(candidate)) { - this.authoritativeDomainCache.set(cacheKey, candidate); - return candidate; - } + // Infrastructure zone cuts (e.g. this deployment's own nameserver + // zone, cname/site extension zones) that this server is + // authoritative for regardless of the customer `domains` table. + // Static config, so always available - checked alongside allDomains + // below rather than gating a separate fast path on it. + const configuredZones = (config.getRawConfig().dns?.zones ?? []).map( + (z) => z.toLowerCase() + ); + + // Fast path: O(1) in-memory Set/array lookup — no DB or network I/O + for (let i = 0; i < labels.length; i++) { + const candidate = labels.slice(i).join("."); + if ( + this.allDomains.has(candidate) || + configuredZones.includes(candidate) + ) { + this.authoritativeDomainCache.set(cacheKey, candidate); + return candidate; } + } + + // Once the in-memory customer-domain set has loaded, the checks + // above are conclusive - no DB fallback needed. + if (this.allDomains.size > 0) { this.authoritativeDomainCache.set(cacheKey, null); return null; } @@ -582,7 +598,8 @@ export class AuthoritativeDNSServer { private async getResourceRecordsByFullDomain( name: string, - queryType: dns.RecordType + queryType: dns.RecordType, + baseDomain: string ): Promise { const cacheKey = `resourceRecords:${name}:${queryType}`; @@ -630,8 +647,14 @@ export class AuthoritativeDNSServer { if (resourceRows.length === 0) { // Resource doesn't exist at all — also pre-populate the domainExists // cache so the subsequent domainExists() call hits memory, not the DB. + // Skip that pre-population when `name` is the zone apex itself: the + // apex always exists (it owns the SOA) even with no resource row, and + // caching `false` here would make domainExists() return that stale + // false instead of ever reaching its own apex check. logger.debug(`No resource found for domain: ${name}`); - this.cache.set(`exists:${name}`, false, 60); + if (name.replace(/\.$/, "") !== baseDomain) { + this.cache.set(`exists:${name}`, false, 60); + } this.cache.set(cacheKey, [], 60); return []; } diff --git a/server/private/lib/readConfigFile.ts b/server/private/lib/readConfigFile.ts index db6999724..a78a16d3b 100644 --- a/server/private/lib/readConfigFile.ts +++ b/server/private/lib/readConfigFile.ts @@ -109,6 +109,16 @@ export const privateConfigSchema = z .array(z.string()) .optional() .default([]), + // Zone names (zone cuts) this server is authoritative for + // beyond the customer domains stored in the `domains` table - + // e.g. a self-hosted deployment's own nameserver zone or + // cname/site extension zones. Any query landing on one of + // these names, or an empty non-terminal beneath one, that + // doesn't match a more specific record gets NOERROR/NODATA + + // SOA (not NXDOMAIN), so QNAME-minimising resolvers (RFC + // 9156) don't treat the ancestor as proof nothing exists + // below it (RFC 8020). Left for the user to populate. + zones: z.array(z.string()).optional().default([]), rate_limit: z .object({ enabled: z.boolean().optional().default(true), diff --git a/server/routers/newt/peers.ts b/server/routers/newt/peers.ts index bd11f03d0..cb69d7480 100644 --- a/server/routers/newt/peers.ts +++ b/server/routers/newt/peers.ts @@ -127,6 +127,40 @@ export async function deletePeersBatch( logger.info(`Deleted ${peers.length} peer(s) from newts (batch)`); } +export async function updatePeersBatch( + peers: { + siteId: number; + publicKey: string; + newtId: string; + peer: { + allowedIps?: string[]; + endpoint?: string; + }; + }[] +) { + if (peers.length === 0) { + return; + } + + await sendToClientsBatch( + peers.map((peer) => ({ + clientId: peer.newtId, + message: { + type: "newt/wg/peer/update", + data: { + publicKey: peer.publicKey, + ...peer.peer + } + }, + options: { incrementConfigVersion: true } + })) + ).catch((error) => { + logger.warn(`Error sending batched newt peer updates:`, error); + }); + + logger.info(`Updated ${peers.length} peer(s) on newts (batch)`); +} + export async function updatePeer( siteId: number, publicKey: string, diff --git a/server/routers/olm/batchUtils.ts b/server/routers/olm/batchUtils.ts new file mode 100644 index 000000000..681008e35 --- /dev/null +++ b/server/routers/olm/batchUtils.ts @@ -0,0 +1,55 @@ +// Shared parsing for the olm relay/unrelay/local/unlocal websocket messages, which accept +// either a single siteId or a batched siteIds array (with a parallel chainIds array), so +// olm clients with many sites can coalesce decisions into one message instead of sending +// one message per site. Older olm clients that only ever send the singular form are handled +// identically to a batch of one, and the reply mirrors whichever form the request used. + +import { db, newts } from "@server/db"; +import { inArray } from "drizzle-orm"; + +// Resolves the newtId for each of the given siteIds in a single query, for batching the +// resulting newt/wg/peer/update pushes instead of looking each one up individually. +export async function resolveNewtIdsBySite( + siteIds: number[] +): Promise> { + if (siteIds.length === 0) { + return new Map(); + } + + const rows = await db + .select({ siteId: newts.siteId, newtId: newts.newtId }) + .from(newts) + .where(inArray(newts.siteId, siteIds)); + + const map = new Map(); + for (const row of rows) { + if (row.siteId != null) { + map.set(row.siteId, row.newtId); + } + } + return map; +} + +export type SiteChainBatch = { + siteIds: number[]; + chainIds: (string | undefined)[]; + isBatch: boolean; +}; + +export function parseSiteChainBatch(data: any): SiteChainBatch { + if (Array.isArray(data?.siteIds)) { + const siteIds: number[] = data.siteIds; + const chainIds: (string | undefined)[] = + Array.isArray(data.chainIds) && + data.chainIds.length === siteIds.length + ? data.chainIds + : siteIds.map(() => data.chainId); + return { siteIds, chainIds, isBatch: true }; + } + + return { + siteIds: data?.siteId !== undefined ? [data.siteId] : [], + chainIds: [data?.chainId], + isBatch: false + }; +} diff --git a/server/routers/olm/handleOlmLocalMessage.ts b/server/routers/olm/handleOlmLocalMessage.ts index 83ef8cbf1..ca80fea06 100644 --- a/server/routers/olm/handleOlmLocalMessage.ts +++ b/server/routers/olm/handleOlmLocalMessage.ts @@ -1,9 +1,10 @@ import { db, sites } from "@server/db"; import { MessageHandler } from "@server/routers/ws"; import { clients, Olm } from "@server/db"; -import { and, eq } from "drizzle-orm"; -import { updatePeer as newtUpdatePeer } from "../newt/peers"; +import { eq, inArray } from "drizzle-orm"; +import { updatePeersBatch } from "../newt/peers"; import logger from "@server/logger"; +import { parseSiteChainBatch, resolveNewtIdsBySite } from "./batchUtils"; export const handleOlmLocalMessage: MessageHandler = async (context) => { const { message, client: c, sendToClient } = context; @@ -40,33 +41,82 @@ export const handleOlmLocalMessage: MessageHandler = async (context) => { return; } - const { siteId, chainId } = message.data; + const { siteIds, chainIds, isBatch } = parseSiteChainBatch(message.data); - // Get the site - const [site] = await db - .select() - .from(sites) - .where(eq(sites.siteId, siteId)) - .limit(1); - - if (!site || !site.exitNodeId) { - logger.warn("Site not found or has no exit node"); + if (siteIds.length === 0) { + logger.warn("Local message has no siteId(s)"); return; } - // update the peer on the newt - await newtUpdatePeer(siteId, client.pubKey, { - endpoint: "" // this removes the endpoint so the newt knows to accept local + // Get the sites + const siteRows = await db + .select() + .from(sites) + .where(inArray(sites.siteId, siteIds)); + const sitesById = new Map(siteRows.map((s) => [s.siteId, s])); + + const valid: { siteId: number; chainId?: string }[] = []; + + for (let i = 0; i < siteIds.length; i++) { + const siteId = siteIds[i]; + + const site = sitesById.get(siteId); + if (!site || !site.exitNodeId) { + logger.warn(`Site ${siteId} not found or has no exit node`); + continue; + } + + valid.push({ siteId, chainId: chainIds[i] }); + } + + if (valid.length === 0) { + return; + } + + // Only ack sites we can actually tell their newt to accept local + const newtIdBySiteId = await resolveNewtIdsBySite(valid.map((v) => v.siteId)); + const pushable = valid.filter((v) => { + if (!newtIdBySiteId.has(v.siteId)) { + logger.warn(`Newt not found for site ${v.siteId}`); + return false; + } + return true; }); + if (pushable.length === 0) { + return; + } + + // update the peer on each newt to accept local + await updatePeersBatch( + pushable.map((v) => ({ + siteId: v.siteId, + publicKey: client.pubKey!, + newtId: newtIdBySiteId.get(v.siteId)!, + peer: { endpoint: "" } // this removes the endpoint so the newt knows to accept local + })) + ); + // Just ack the message, we don't keep sending it + if (isBatch) { + return { + message: { + type: "olm/wg/peer/local", + data: { + siteIds: pushable.map((v) => v.siteId), + chainIds: pushable.map((v) => v.chainId) + } + }, + broadcast: false, + excludeSender: false + }; + } + + const single = pushable[0]; return { message: { type: "olm/wg/peer/local", - data: { - siteId: siteId, - chainId - } + data: { siteId: single.siteId, chainId: single.chainId } }, broadcast: false, excludeSender: false diff --git a/server/routers/olm/handleOlmRelayMessage.ts b/server/routers/olm/handleOlmRelayMessage.ts index 406ed7bbb..1cd27912b 100644 --- a/server/routers/olm/handleOlmRelayMessage.ts +++ b/server/routers/olm/handleOlmRelayMessage.ts @@ -1,10 +1,11 @@ import { db, exitNodes, sites } from "@server/db"; import { MessageHandler } from "@server/routers/ws"; import { clients, clientSitesAssociationsCache, Olm } from "@server/db"; -import { and, eq } from "drizzle-orm"; -import { updatePeer as newtUpdatePeer } from "../newt/peers"; +import { and, eq, inArray } from "drizzle-orm"; +import { updatePeersBatch } from "../newt/peers"; import logger from "@server/logger"; import config from "@server/lib/config"; +import { parseSiteChainBatch, resolveNewtIdsBySite } from "./batchUtils"; export const handleOlmRelayMessage: MessageHandler = async (context) => { const { message, client: c, sendToClient } = context; @@ -41,29 +42,66 @@ export const handleOlmRelayMessage: MessageHandler = async (context) => { return; } - const { siteId, chainId } = message.data; + const { siteIds, chainIds, isBatch } = parseSiteChainBatch(message.data); - // Get the site - const [site] = await db - .select() - .from(sites) - .where(eq(sites.siteId, siteId)) - .limit(1); - - if (!site || !site.exitNodeId) { - logger.warn("Site not found or has no exit node"); + if (siteIds.length === 0) { + logger.warn("Relay message has no siteId(s)"); return; } - // get the site's exit node - const [exitNode] = await db + // Get the sites + const siteRows = await db .select() - .from(exitNodes) - .where(eq(exitNodes.exitNodeId, site.exitNodeId)) - .limit(1); + .from(sites) + .where(inArray(sites.siteId, siteIds)); + const sitesById = new Map(siteRows.map((s) => [s.siteId, s])); - if (!exitNode) { - logger.warn("Exit node not found for site"); + const exitNodeIds = [ + ...new Set( + siteRows + .map((s) => s.exitNodeId) + .filter((id): id is number => id != null) + ) + ]; + + // Get the sites' exit nodes + const exitNodeRows = exitNodeIds.length + ? await db + .select() + .from(exitNodes) + .where(inArray(exitNodes.exitNodeId, exitNodeIds)) + : []; + const exitNodesById = new Map(exitNodeRows.map((e) => [e.exitNodeId, e])); + + const valid: { + siteId: number; + chainId?: string; + relayEndpoint: string; + }[] = []; + + for (let i = 0; i < siteIds.length; i++) { + const siteId = siteIds[i]; + + const site = sitesById.get(siteId); + if (!site || !site.exitNodeId) { + logger.warn(`Site ${siteId} not found or has no exit node`); + continue; + } + + const exitNode = exitNodesById.get(site.exitNodeId); + if (!exitNode) { + logger.warn(`Exit node not found for site ${siteId}`); + continue; + } + + valid.push({ + siteId, + chainId: chainIds[i], + relayEndpoint: exitNode.endpoint + }); + } + + if (valid.length === 0) { return; } @@ -75,23 +113,64 @@ export const handleOlmRelayMessage: MessageHandler = async (context) => { .where( and( eq(clientSitesAssociationsCache.clientId, olm.clientId), - eq(clientSitesAssociationsCache.siteId, siteId) + inArray( + clientSitesAssociationsCache.siteId, + valid.map((v) => v.siteId) + ) ) ); - // update the peer on the newt - await newtUpdatePeer(siteId, client.pubKey, { - endpoint: "" // this removes the endpoint so the newt knows to relay + // Only ack sites we can actually tell their newt to relay for + const newtIdBySiteId = await resolveNewtIdsBySite(valid.map((v) => v.siteId)); + const pushable = valid.filter((v) => { + if (!newtIdBySiteId.has(v.siteId)) { + logger.warn(`Newt not found for site ${v.siteId}`); + return false; + } + return true; }); + if (pushable.length === 0) { + return; + } + + // update the peer on each newt so it knows to relay + await updatePeersBatch( + pushable.map((v) => ({ + siteId: v.siteId, + publicKey: client.pubKey!, + newtId: newtIdBySiteId.get(v.siteId)!, + peer: { endpoint: "" } // this removes the endpoint so the newt knows to relay + })) + ); + + const relayPort = config.getRawConfig().gerbil.clients_start_port; + + if (isBatch) { + return { + message: { + type: "olm/wg/peer/relay", + data: { + siteIds: pushable.map((v) => v.siteId), + relayEndpoints: pushable.map((v) => v.relayEndpoint), + relayPort, + chainIds: pushable.map((v) => v.chainId) + } + }, + broadcast: false, + excludeSender: false + }; + } + + const single = pushable[0]; return { message: { type: "olm/wg/peer/relay", data: { - siteId: siteId, - relayEndpoint: exitNode.endpoint, - relayPort: config.getRawConfig().gerbil.clients_start_port, - chainId + siteId: single.siteId, + relayEndpoint: single.relayEndpoint, + relayPort, + chainId: single.chainId } }, broadcast: false, diff --git a/server/routers/olm/handleOlmUnLocalMessage.ts b/server/routers/olm/handleOlmUnLocalMessage.ts index 55d312815..f98aaf436 100644 --- a/server/routers/olm/handleOlmUnLocalMessage.ts +++ b/server/routers/olm/handleOlmUnLocalMessage.ts @@ -1,9 +1,10 @@ -import { db, exitNodes, sites } from "@server/db"; +import { db, sites } from "@server/db"; import { MessageHandler } from "@server/routers/ws"; import { clients, clientSitesAssociationsCache, Olm } from "@server/db"; -import { and, eq } from "drizzle-orm"; -import { updatePeer as newtUpdatePeer } from "../newt/peers"; +import { and, eq, inArray } from "drizzle-orm"; +import { updatePeersBatch } from "../newt/peers"; import logger from "@server/logger"; +import { parseSiteChainBatch, resolveNewtIdsBySite } from "./batchUtils"; export const handleOlmUnLocalMessage: MessageHandler = async (context) => { const { message, client: c, sendToClient } = context; @@ -40,54 +41,111 @@ export const handleOlmUnLocalMessage: MessageHandler = async (context) => { return; } - const { siteId, chainId } = message.data; + const { siteIds, chainIds, isBatch } = parseSiteChainBatch(message.data); - // Get the site - const [site] = await db - .select() - .from(sites) - .where(eq(sites.siteId, siteId)) - .limit(1); - - if (!site) { - logger.warn("Site not found or has no exit node"); + if (siteIds.length === 0) { + logger.warn("Unlocal message has no siteId(s)"); return; } - const [clientSiteAssociation] = await db + // Get the sites + const siteRows = await db + .select() + .from(sites) + .where(inArray(sites.siteId, siteIds)); + const sitesById = new Map(siteRows.map((s) => [s.siteId, s])); + + const assocRows = await db .select() .from(clientSitesAssociationsCache) .where( and( eq(clientSitesAssociationsCache.clientId, olm.clientId), - eq(clientSitesAssociationsCache.siteId, siteId) + inArray(clientSitesAssociationsCache.siteId, siteIds) ) ); + const assocBySiteId = new Map(assocRows.map((a) => [a.siteId, a])); - if (!clientSiteAssociation) { - logger.warn("Client-Site association not found"); + const valid: { siteId: number; chainId?: string; endpoint: string }[] = []; + + for (let i = 0; i < siteIds.length; i++) { + const siteId = siteIds[i]; + + const site = sitesById.get(siteId); + if (!site) { + logger.warn(`Site ${siteId} not found or has no exit node`); + continue; + } + + const clientSiteAssociation = assocBySiteId.get(siteId); + if (!clientSiteAssociation) { + logger.warn(`Client-Site association not found for site ${siteId}`); + continue; + } + + if (!clientSiteAssociation.endpoint) { + logger.warn( + `Client-Site association has no endpoint, cannot unrelay site ${siteId}` + ); + continue; + } + + valid.push({ + siteId, + chainId: chainIds[i], + endpoint: clientSiteAssociation.isRelayed + ? "" + : clientSiteAssociation.endpoint // this is the endpoint of the client to connect directly to the newt + }); + } + + if (valid.length === 0) { return; } - if (!clientSiteAssociation.endpoint) { - logger.warn("Client-Site association has no endpoint, cannot unrelay"); - return; - } - - // update the peer on the newt - await newtUpdatePeer(siteId, client.pubKey, { - endpoint: clientSiteAssociation.isRelayed - ? "" - : clientSiteAssociation.endpoint // this is the endpoint of the client to connect directly to the newt + // Only ack sites we can actually push to their newt + const newtIdBySiteId = await resolveNewtIdsBySite(valid.map((v) => v.siteId)); + const pushable = valid.filter((v) => { + if (!newtIdBySiteId.has(v.siteId)) { + logger.warn(`Newt not found for site ${v.siteId}`); + return false; + } + return true; }); + if (pushable.length === 0) { + return; + } + + // update the peer on each newt + await updatePeersBatch( + pushable.map((v) => ({ + siteId: v.siteId, + publicKey: client.pubKey!, + newtId: newtIdBySiteId.get(v.siteId)!, + peer: { endpoint: v.endpoint } + })) + ); + + if (isBatch) { + return { + message: { + type: "olm/wg/peer/unlocal", + data: { + siteIds: pushable.map((v) => v.siteId), + chainIds: pushable.map((v) => v.chainId) + } + }, + broadcast: false, + excludeSender: false + }; + } + + const single = pushable[0]; return { message: { type: "olm/wg/peer/unlocal", - data: { - siteId: siteId, - chainId - } + data: { siteId: single.siteId, chainId: single.chainId } }, broadcast: false, excludeSender: false diff --git a/server/routers/olm/handleOlmUnRelayMessage.ts b/server/routers/olm/handleOlmUnRelayMessage.ts index 3f73a5834..a6e89852e 100644 --- a/server/routers/olm/handleOlmUnRelayMessage.ts +++ b/server/routers/olm/handleOlmUnRelayMessage.ts @@ -1,9 +1,10 @@ import { db, exitNodes, sites } from "@server/db"; import { MessageHandler } from "@server/routers/ws"; import { clients, clientSitesAssociationsCache, Olm } from "@server/db"; -import { and, eq } from "drizzle-orm"; -import { updatePeer as newtUpdatePeer } from "../newt/peers"; +import { and, eq, inArray } from "drizzle-orm"; +import { updatePeersBatch } from "../newt/peers"; import logger from "@server/logger"; +import { parseSiteChainBatch, resolveNewtIdsBySite } from "./batchUtils"; export const handleOlmUnRelayMessage: MessageHandler = async (context) => { const { message, client: c, sendToClient } = context; @@ -40,21 +41,21 @@ export const handleOlmUnRelayMessage: MessageHandler = async (context) => { return; } - const { siteId, chainId } = message.data; + const { siteIds, chainIds, isBatch } = parseSiteChainBatch(message.data); - // Get the site - const [site] = await db - .select() - .from(sites) - .where(eq(sites.siteId, siteId)) - .limit(1); - - if (!site) { - logger.warn("Site not found or has no exit node"); + if (siteIds.length === 0) { + logger.warn("Unrelay message has no siteId(s)"); return; } - const [clientSiteAssociation] = await db + // Get the sites + const siteRows = await db + .select() + .from(sites) + .where(inArray(sites.siteId, siteIds)); + const sitesById = new Map(siteRows.map((s) => [s.siteId, s])); + + const assocRows = await db .update(clientSitesAssociationsCache) .set({ isRelayed: false @@ -62,33 +63,100 @@ export const handleOlmUnRelayMessage: MessageHandler = async (context) => { .where( and( eq(clientSitesAssociationsCache.clientId, olm.clientId), - eq(clientSitesAssociationsCache.siteId, siteId) + inArray(clientSitesAssociationsCache.siteId, siteIds) ) ) .returning(); + const assocBySiteId = new Map(assocRows.map((a) => [a.siteId, a])); - if (!clientSiteAssociation) { - logger.warn("Client-Site association not found"); + const valid: { + siteId: number; + chainId?: string; + endpoint: string; + clientEndpoint: string; + }[] = []; + + for (let i = 0; i < siteIds.length; i++) { + const siteId = siteIds[i]; + + const site = sitesById.get(siteId); + if (!site) { + logger.warn(`Site ${siteId} not found or has no exit node`); + continue; + } + + const clientSiteAssociation = assocBySiteId.get(siteId); + if (!clientSiteAssociation) { + logger.warn(`Client-Site association not found for site ${siteId}`); + continue; + } + + if (!clientSiteAssociation.endpoint) { + logger.warn( + `Client-Site association has no endpoint, cannot unrelay site ${siteId}` + ); + continue; + } + + valid.push({ + siteId, + chainId: chainIds[i], + endpoint: site.endpoint ?? "", + clientEndpoint: clientSiteAssociation.endpoint + }); + } + + if (valid.length === 0) { return; } - if (!clientSiteAssociation.endpoint) { - logger.warn("Client-Site association has no endpoint, cannot unrelay"); - return; - } - - // update the peer on the newt - await newtUpdatePeer(siteId, client.pubKey, { - endpoint: clientSiteAssociation.endpoint // this is the endpoint of the client to connect directly to the newt + // Only ack sites we can actually tell their newt to connect directly + const newtIdBySiteId = await resolveNewtIdsBySite(valid.map((v) => v.siteId)); + const pushable = valid.filter((v) => { + if (!newtIdBySiteId.has(v.siteId)) { + logger.warn(`Newt not found for site ${v.siteId}`); + return false; + } + return true; }); + if (pushable.length === 0) { + return; + } + + // update the peer on each newt to connect directly to the client + await updatePeersBatch( + pushable.map((v) => ({ + siteId: v.siteId, + publicKey: client.pubKey!, + newtId: newtIdBySiteId.get(v.siteId)!, + peer: { endpoint: v.clientEndpoint } // this is the endpoint of the client to connect directly to the newt + })) + ); + + if (isBatch) { + return { + message: { + type: "olm/wg/peer/unrelay", + data: { + siteIds: pushable.map((v) => v.siteId), + endpoints: pushable.map((v) => v.endpoint), + chainIds: pushable.map((v) => v.chainId) + } + }, + broadcast: false, + excludeSender: false + }; + } + + const single = pushable[0]; return { message: { type: "olm/wg/peer/unrelay", data: { - siteId: siteId, - endpoint: site.endpoint, - chainId + siteId: single.siteId, + endpoint: single.endpoint, + chainId: single.chainId } }, broadcast: false,