pangolin_mirror/pangolin
1
0
Fork 0
mirror of https://github.com/fosrl/pangolin.git synced 2026-03-24 03:26:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

First pass at HA

Browse Source
This commit is contained in:
Owen
2026-03-23 11:44:02 -07:00
parent 1366901e24
commit 02033f611f
1 changed files with 355 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

581
server/lib/rebuildClientAssociations.ts
View File

@@ -11,6 +11,7 @@ import {
roleSiteResources, roleSiteResources,
Site, Site,
SiteResource, SiteResource,
siteNetworks,
siteResources, siteResources,
sites, sites,
Transaction, Transaction,
@@ -47,15 +48,23 @@ export async function getClientSiteResourceAccess(
siteResource: SiteResource, siteResource: SiteResource,
trx: Transaction | typeof db = db trx: Transaction | typeof db = db
) { ) {
// get the site // get all sites associated with this siteResource via its network
const [site] = await trx const sitesList = siteResource.networkId
.select() ? await trx
.from(sites) .select()
.where(eq(sites.siteId, siteResource.siteId)) .from(sites)
.limit(1); .innerJoin(
siteNetworks,
eq(siteNetworks.siteId, sites.siteId)
)
.where(eq(siteNetworks.networkId, siteResource.networkId))
.then((rows) => rows.map((row) => row.sites))
: [];
if (!site) { if (sitesList.length === 0) {
throw new Error(`Site with ID ${siteResource.siteId} not found`); logger.warn(
`No sites found for siteResource ${siteResource.siteResourceId} with networkId ${siteResource.networkId}`
);
} }
const roleIds = await trx const roleIds = await trx
@@ -136,7 +145,7 @@ export async function getClientSiteResourceAccess(
const mergedAllClientIds = mergedAllClients.map((c) => c.clientId); const mergedAllClientIds = mergedAllClients.map((c) => c.clientId);
return { return {
site, sitesList,
mergedAllClients, mergedAllClients,
mergedAllClientIds mergedAllClientIds
}; };
@@ -152,40 +161,51 @@ export async function rebuildClientAssociationsFromSiteResource(
subnet: string | null; subnet: string | null;
}[]; }[];
}> { }> {
const siteId = siteResource.siteId; const { sitesList, mergedAllClients, mergedAllClientIds } =
const { site, mergedAllClients, mergedAllClientIds } =
await getClientSiteResourceAccess(siteResource, trx); await getClientSiteResourceAccess(siteResource, trx);
/////////// process the client-siteResource associations /////////// /////////// process the client-siteResource associations ///////////
// get all of the clients associated with other resources on this site // get all of the clients associated with other resources in the same network,
const allUpdatedClientsFromOtherResourcesOnThisSite = await trx // joined through siteNetworks so we know which siteId each client belongs to
.select({ const allUpdatedClientsFromOtherResourcesOnThisSite = siteResource.networkId
clientId: clientSiteResourcesAssociationsCache.clientId ? await trx
}) .select({
.from(clientSiteResourcesAssociationsCache) clientId: clientSiteResourcesAssociationsCache.clientId,
.innerJoin( siteId: siteNetworks.siteId
siteResources, })
eq( .from(clientSiteResourcesAssociationsCache)
clientSiteResourcesAssociationsCache.siteResourceId, .innerJoin(
siteResources.siteResourceId siteResources,
) eq(
) clientSiteResourcesAssociationsCache.siteResourceId,
.where( siteResources.siteResourceId
and( )
eq(siteResources.siteId, siteId), )
ne(siteResources.siteResourceId, siteResource.siteResourceId) .innerJoin(
) siteNetworks,
); eq(siteNetworks.networkId, siteResources.networkId)
)
.where(
and(
eq(siteResources.networkId, siteResource.networkId),
ne(
siteResources.siteResourceId,
siteResource.siteResourceId
)
)
)
: [];
const allClientIdsFromOtherResourcesOnThisSite = Array.from( // Build a per-site map so the loop below can check by siteId rather than
new Set( // across the entire network.
allUpdatedClientsFromOtherResourcesOnThisSite.map( const clientsFromOtherResourcesBySite = new Map<number, Set<number>>();
(row) => row.clientId for (const row of allUpdatedClientsFromOtherResourcesOnThisSite) {
) if (!clientsFromOtherResourcesBySite.has(row.siteId)) {
) clientsFromOtherResourcesBySite.set(row.siteId, new Set());
); }
clientsFromOtherResourcesBySite.get(row.siteId)!.add(row.clientId);
}
const existingClientSiteResources = await trx const existingClientSiteResources = await trx
.select({ .select({
@@ -259,82 +279,90 @@ export async function rebuildClientAssociationsFromSiteResource(
/////////// process the client-site associations /////////// /////////// process the client-site associations ///////////
const existingClientSites = await trx for (const site of sitesList) {
.select({ const siteId = site.siteId;
clientId: clientSitesAssociationsCache.clientId
})
.from(clientSitesAssociationsCache)
.where(eq(clientSitesAssociationsCache.siteId, siteResource.siteId));
const existingClientSiteIds = existingClientSites.map( const existingClientSites = await trx
(row) => row.clientId .select({
); clientId: clientSitesAssociationsCache.clientId
})
.from(clientSitesAssociationsCache)
.where(eq(clientSitesAssociationsCache.siteId, siteId));
// Get full client details for existing clients (needed for sending delete messages) const existingClientSiteIds = existingClientSites.map(
const existingClients = await trx (row) => row.clientId
.select({ );
clientId: clients.clientId,
pubKey: clients.pubKey,
subnet: clients.subnet
})
.from(clients)
.where(inArray(clients.clientId, existingClientSiteIds));
const clientSitesToAdd = mergedAllClientIds.filter( // Get full client details for existing clients (needed for sending delete messages)
(clientId) => const existingClients =
!existingClientSiteIds.includes(clientId) && existingClientSiteIds.length > 0
!allClientIdsFromOtherResourcesOnThisSite.includes(clientId) // dont remove if there is still another connection for another site resource ? await trx
); .select({
clientId: clients.clientId,
pubKey: clients.pubKey,
subnet: clients.subnet
})
.from(clients)
.where(inArray(clients.clientId, existingClientSiteIds))
: [];
const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({ const otherResourceClientIds = clientsFromOtherResourcesBySite.get(siteId) ?? new Set<number>();
clientId,
siteId
}));
if (clientSitesToInsert.length > 0) { const clientSitesToAdd = mergedAllClientIds.filter(
await trx (clientId) =>
.insert(clientSitesAssociationsCache) !existingClientSiteIds.includes(clientId) &&
.values(clientSitesToInsert) !otherResourceClientIds.has(clientId) // dont add if already connected via another site resource
.returning(); );
}
// Now remove any client-site associations that should no longer exist const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({
const clientSitesToRemove = existingClientSiteIds.filter( clientId,
(clientId) => siteId
!mergedAllClientIds.includes(clientId) && }));
!allClientIdsFromOtherResourcesOnThisSite.includes(clientId) // dont remove if there is still another connection for another site resource
);
if (clientSitesToRemove.length > 0) { if (clientSitesToInsert.length > 0) {
await trx await trx
.delete(clientSitesAssociationsCache) .insert(clientSitesAssociationsCache)
.where( .values(clientSitesToInsert)
and( .returning();
eq(clientSitesAssociationsCache.siteId, siteId), }
inArray(
clientSitesAssociationsCache.clientId, // Now remove any client-site associations that should no longer exist
clientSitesToRemove const clientSitesToRemove = existingClientSiteIds.filter(
(clientId) =>
!mergedAllClientIds.includes(clientId) &&
!otherResourceClientIds.has(clientId) // dont remove if there is still another connection for another site resource
);
if (clientSitesToRemove.length > 0) {
await trx
.delete(clientSitesAssociationsCache)
.where(
and(
eq(clientSitesAssociationsCache.siteId, siteId),
inArray(
clientSitesAssociationsCache.clientId,
clientSitesToRemove
)
) )
) );
); }
// Now handle the messages to add/remove peers on both the newt and olm sides
await handleMessagesForSiteClients(
site,
siteId,
mergedAllClients,
existingClients,
clientSitesToAdd,
clientSitesToRemove,
trx
);
} }
/////////// send the messages ///////////
// Now handle the messages to add/remove peers on both the newt and olm sides
await handleMessagesForSiteClients(
site,
siteId,
mergedAllClients,
existingClients,
clientSitesToAdd,
clientSitesToRemove,
trx
);
// Handle subnet proxy target updates for the resource associations // Handle subnet proxy target updates for the resource associations
await handleSubnetProxyTargetUpdates( await handleSubnetProxyTargetUpdates(
siteResource, siteResource,
sitesList,
mergedAllClients, mergedAllClients,
existingResourceClients, existingResourceClients,
clientSiteResourcesToAdd, clientSiteResourcesToAdd,
@@ -623,6 +651,7 @@ export async function updateClientSiteDestinations(
async function handleSubnetProxyTargetUpdates( async function handleSubnetProxyTargetUpdates(
siteResource: SiteResource, siteResource: SiteResource,
sitesList: Site[],
allClients: { allClients: {
clientId: number; clientId: number;
pubKey: string | null; pubKey: string | null;
@@ -637,131 +666,144 @@ async function handleSubnetProxyTargetUpdates(
clientSiteResourcesToRemove: number[], clientSiteResourcesToRemove: number[],
trx: Transaction | typeof db = db trx: Transaction | typeof db = db
): Promise<void> { ): Promise<void> {
// Get the newt for this site const proxyJobs: Promise<any>[] = [];
const [newt] = await trx const olmJobs: Promise<any>[] = [];
.select()
.from(newts)
.where(eq(newts.siteId, siteResource.siteId))
.limit(1);
if (!newt) { for (const siteData of sitesList) {
logger.warn( const siteId = siteData.siteId;
`Newt not found for site ${siteResource.siteId}, skipping subnet proxy target updates`
);
return;
}
const proxyJobs = []; // Get the newt for this site
const olmJobs = []; const [newt] = await trx
// Generate targets for added associations .select()
if (clientSiteResourcesToAdd.length > 0) { .from(newts)
const addedClients = allClients.filter((client) => .where(eq(newts.siteId, siteId))
clientSiteResourcesToAdd.includes(client.clientId) .limit(1);
);
if (addedClients.length > 0) { if (!newt) {
const targetsToAdd = generateSubnetProxyTargets( logger.warn(
siteResource, `Newt not found for site ${siteId}, skipping subnet proxy target updates`
addedClients
); );
continue;
if (targetsToAdd.length > 0) {
logger.info(
`Adding ${targetsToAdd.length} subnet proxy targets for siteResource ${siteResource.siteResourceId}`
);
proxyJobs.push(
addSubnetProxyTargets(
newt.newtId,
targetsToAdd,
newt.version
)
);
}
for (const client of addedClients) {
olmJobs.push(
addPeerData(
client.clientId,
siteResource.siteId,
generateRemoteSubnets([siteResource]),
generateAliasConfig([siteResource])
)
);
}
} }
}
// here we use the existingSiteResource from BEFORE we updated the destination so we dont need to worry about updating destinations here // Generate targets for added associations
if (clientSiteResourcesToAdd.length > 0) {
// Generate targets for removed associations const addedClients = allClients.filter((client) =>
if (clientSiteResourcesToRemove.length > 0) { clientSiteResourcesToAdd.includes(client.clientId)
const removedClients = existingClients.filter((client) =>
clientSiteResourcesToRemove.includes(client.clientId)
);
if (removedClients.length > 0) {
const targetsToRemove = generateSubnetProxyTargets(
siteResource,
removedClients
); );
if (targetsToRemove.length > 0) { if (addedClients.length > 0) {
logger.info( const targetsToAdd = generateSubnetProxyTargets(
`Removing ${targetsToRemove.length} subnet proxy targets for siteResource ${siteResource.siteResourceId}` siteResource,
addedClients
); );
proxyJobs.push(
removeSubnetProxyTargets(
newt.newtId,
targetsToRemove,
newt.version
)
);
}
for (const client of removedClients) { if (targetsToAdd.length > 0) {
// Check if this client still has access to another resource on this site with the same destination logger.info(
const destinationStillInUse = await trx `Adding ${targetsToAdd.length} subnet proxy targets for siteResource ${siteResource.siteResourceId} on site ${siteId}`
.select() );
.from(siteResources) proxyJobs.push(
.innerJoin( addSubnetProxyTargets(
clientSiteResourcesAssociationsCache, newt.newtId,
eq( targetsToAdd,
clientSiteResourcesAssociationsCache.siteResourceId, newt.version
siteResources.siteResourceId
)
)
.where(
and(
eq(
clientSiteResourcesAssociationsCache.clientId,
client.clientId
),
eq(siteResources.siteId, siteResource.siteId),
eq(
siteResources.destination,
siteResource.destination
),
ne(
siteResources.siteResourceId,
siteResource.siteResourceId
)
) )
); );
}
// Only remove remote subnet if no other resource uses the same destination for (const client of addedClients) {
const remoteSubnetsToRemove = olmJobs.push(
destinationStillInUse.length > 0 addPeerData(
? [] client.clientId,
: generateRemoteSubnets([siteResource]); siteId,
generateRemoteSubnets([siteResource]),
generateAliasConfig([siteResource])
)
);
}
}
}
olmJobs.push( // here we use the existingSiteResource from BEFORE we updated the destination so we dont need to worry about updating destinations here
removePeerData(
client.clientId, // Generate targets for removed associations
siteResource.siteId, if (clientSiteResourcesToRemove.length > 0) {
remoteSubnetsToRemove, const removedClients = existingClients.filter((client) =>
generateAliasConfig([siteResource]) clientSiteResourcesToRemove.includes(client.clientId)
) );
if (removedClients.length > 0) {
const targetsToRemove = generateSubnetProxyTargets(
siteResource,
removedClients
); );
if (targetsToRemove.length > 0) {
logger.info(
`Removing ${targetsToRemove.length} subnet proxy targets for siteResource ${siteResource.siteResourceId} on site ${siteId}`
);
proxyJobs.push(
removeSubnetProxyTargets(
newt.newtId,
targetsToRemove,
newt.version
)
);
}
for (const client of removedClients) {
// Check if this client still has access to another resource
// on this specific site with the same destination. We scope
// by siteId (via siteNetworks) rather than networkId because
// removePeerData operates per-site — a resource on a different
// site sharing the same network should not block removal here.
const destinationStillInUse = await trx
.select()
.from(siteResources)
.innerJoin(
clientSiteResourcesAssociationsCache,
eq(
clientSiteResourcesAssociationsCache.siteResourceId,
siteResources.siteResourceId
)
)
.innerJoin(
siteNetworks,
eq(siteNetworks.networkId, siteResources.networkId)
)
.where(
and(
eq(
clientSiteResourcesAssociationsCache.clientId,
client.clientId
),
eq(siteNetworks.siteId, siteId),
eq(
siteResources.destination,
siteResource.destination
),
ne(
siteResources.siteResourceId,
siteResource.siteResourceId
)
)
);
// Only remove remote subnet if no other resource uses the same destination
const remoteSubnetsToRemove =
destinationStillInUse.length > 0
? []
: generateRemoteSubnets([siteResource]);
olmJobs.push(
removePeerData(
client.clientId,
siteId,
remoteSubnetsToRemove,
generateAliasConfig([siteResource])
)
);
}
} }
} }
} }
@@ -868,10 +910,25 @@ export async function rebuildClientAssociationsFromClient(
) )
: []; : [];
// Group by siteId for site-level associations // Group by siteId for site-level associations — look up via siteNetworks since
const newSiteIds = Array.from( // siteResources no longer carries a direct siteId column.
new Set(newSiteResources.map((sr) => sr.siteId)) const networkIds = Array.from(
new Set(
newSiteResources
.map((sr) => sr.networkId)
.filter((id): id is number => id !== null)
)
); );
const newSiteIds =
networkIds.length > 0
? await trx
.select({ siteId: siteNetworks.siteId })
.from(siteNetworks)
.where(inArray(siteNetworks.networkId, networkIds))
.then((rows) =>
Array.from(new Set(rows.map((r) => r.siteId)))
)
: [];
/////////// Process client-siteResource associations /////////// /////////// Process client-siteResource associations ///////////
@@ -1144,13 +1201,45 @@ async function handleMessagesForClientResources(
resourcesToAdd.includes(r.siteResourceId) resourcesToAdd.includes(r.siteResourceId)
); );
// Build (resource, siteId) pairs by looking up siteNetworks for each resource's networkId
const addedNetworkIds = Array.from(
new Set(
addedResources
.map((r) => r.networkId)
.filter((id): id is number => id !== null)
)
);
const addedSiteNetworkRows =
addedNetworkIds.length > 0
? await trx
.select({
networkId: siteNetworks.networkId,
siteId: siteNetworks.siteId
})
.from(siteNetworks)
.where(inArray(siteNetworks.networkId, addedNetworkIds))
: [];
const addedNetworkToSites = new Map<number, number[]>();
for (const row of addedSiteNetworkRows) {
if (!addedNetworkToSites.has(row.networkId)) {
addedNetworkToSites.set(row.networkId, []);
}
addedNetworkToSites.get(row.networkId)!.push(row.siteId);
}
// Group by site for proxy updates // Group by site for proxy updates
const addedBySite = new Map<number, SiteResource[]>(); const addedBySite = new Map<number, SiteResource[]>();
for (const resource of addedResources) { for (const resource of addedResources) {
if (!addedBySite.has(resource.siteId)) { const siteIds =
addedBySite.set(resource.siteId, []); resource.networkId != null
? (addedNetworkToSites.get(resource.networkId) ?? [])
: [];
for (const siteId of siteIds) {
if (!addedBySite.has(siteId)) {
addedBySite.set(siteId, []);
}
addedBySite.get(siteId)!.push(resource);
} }
addedBySite.get(resource.siteId)!.push(resource);
} }
// Add subnet proxy targets for each site // Add subnet proxy targets for each site
@@ -1192,7 +1281,7 @@ async function handleMessagesForClientResources(
olmJobs.push( olmJobs.push(
addPeerData( addPeerData(
client.clientId, client.clientId,
resource.siteId, siteId,
generateRemoteSubnets([resource]), generateRemoteSubnets([resource]),
generateAliasConfig([resource]) generateAliasConfig([resource])
) )
@@ -1204,7 +1293,7 @@ async function handleMessagesForClientResources(
error.message.includes("not found") error.message.includes("not found")
) { ) {
logger.debug( logger.debug(
`Olm data not found for client ${client.clientId} and site ${resource.siteId}, skipping removal` `Olm data not found for client ${client.clientId} and site ${siteId}, skipping addition`
); );
} else { } else {
throw error; throw error;
@@ -1221,13 +1310,45 @@ async function handleMessagesForClientResources(
.from(siteResources) .from(siteResources)
.where(inArray(siteResources.siteResourceId, resourcesToRemove)); .where(inArray(siteResources.siteResourceId, resourcesToRemove));
// Build (resource, siteId) pairs via siteNetworks
const removedNetworkIds = Array.from(
new Set(
removedResources
.map((r) => r.networkId)
.filter((id): id is number => id !== null)
)
);
const removedSiteNetworkRows =
removedNetworkIds.length > 0
? await trx
.select({
networkId: siteNetworks.networkId,
siteId: siteNetworks.siteId
})
.from(siteNetworks)
.where(inArray(siteNetworks.networkId, removedNetworkIds))
: [];
const removedNetworkToSites = new Map<number, number[]>();
for (const row of removedSiteNetworkRows) {
if (!removedNetworkToSites.has(row.networkId)) {
removedNetworkToSites.set(row.networkId, []);
}
removedNetworkToSites.get(row.networkId)!.push(row.siteId);
}
// Group by site for proxy updates // Group by site for proxy updates
const removedBySite = new Map<number, SiteResource[]>(); const removedBySite = new Map<number, SiteResource[]>();
for (const resource of removedResources) { for (const resource of removedResources) {
if (!removedBySite.has(resource.siteId)) { const siteIds =
removedBySite.set(resource.siteId, []); resource.networkId != null
? (removedNetworkToSites.get(resource.networkId) ?? [])
: [];
for (const siteId of siteIds) {
if (!removedBySite.has(siteId)) {
removedBySite.set(siteId, []);
}
removedBySite.get(siteId)!.push(resource);
} }
removedBySite.get(resource.siteId)!.push(resource);
} }
// Remove subnet proxy targets for each site // Remove subnet proxy targets for each site
@@ -1265,7 +1386,11 @@ async function handleMessagesForClientResources(
} }
try { try {
// Check if this client still has access to another resource on this site with the same destination // Check if this client still has access to another resource
// on this specific site with the same destination. We scope
// by siteId (via siteNetworks) rather than networkId because
// removePeerData operates per-site — a resource on a different
// site sharing the same network should not block removal here.
const destinationStillInUse = await trx const destinationStillInUse = await trx
.select() .select()
.from(siteResources) .from(siteResources)
@@ -1276,13 +1401,17 @@ async function handleMessagesForClientResources(
siteResources.siteResourceId siteResources.siteResourceId
) )
) )
.innerJoin(
siteNetworks,
eq(siteNetworks.networkId, siteResources.networkId)
)
.where( .where(
and( and(
eq( eq(
clientSiteResourcesAssociationsCache.clientId, clientSiteResourcesAssociationsCache.clientId,
client.clientId client.clientId
), ),
eq(siteResources.siteId, resource.siteId), eq(siteNetworks.siteId, siteId),
eq( eq(
siteResources.destination, siteResources.destination,
resource.destination resource.destination
@@ -1304,7 +1433,7 @@ async function handleMessagesForClientResources(
olmJobs.push( olmJobs.push(
removePeerData( removePeerData(
client.clientId, client.clientId,
resource.siteId, siteId,
remoteSubnetsToRemove, remoteSubnetsToRemove,
generateAliasConfig([resource]) generateAliasConfig([resource])
) )
@@ -1316,7 +1445,7 @@ async function handleMessagesForClientResources(
error.message.includes("not found") error.message.includes("not found")
) { ) {
logger.debug( logger.debug(
`Olm data not found for client ${client.clientId} and site ${resource.siteId}, skipping removal` `Olm data not found for client ${client.clientId} and site ${siteId}, skipping removal`
); );
} else { } else {
throw error; throw error;