feat: added heartbeat and fixed up gateway page

This commit is contained in:
Sheen Capadngan
2025-09-02 21:51:29 +08:00
parent 879361fd7a
commit d5dcaf0d59
13 changed files with 269 additions and 66 deletions

View File

@@ -120,6 +120,8 @@ export async function up(knex: Knex): Promise<void> {
t.foreign("proxyId").references("id").inTable(TableName.Proxy).onDelete("SET NULL");
t.string("name").notNullable().unique();
t.dateTime("heartbeat");
});
await createOnUpdateTrigger(knex, TableName.GatewayV2);

View File

@@ -14,7 +14,8 @@ export const GatewaysV2Schema = z.object({
orgId: z.string().uuid(),
identityId: z.string().uuid(),
proxyId: z.string().uuid().nullable().optional(),
name: z.string()
name: z.string(),
heartbeat: z.date().nullable().optional()
});
export type TGatewaysV2 = z.infer<typeof GatewaysV2Schema>;

View File

@@ -1,5 +1,6 @@
import z from "zod";
import { writeLimit } from "@app/server/config/rateLimiter";
import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
import { AuthMode } from "@app/services/auth/auth-type";
@@ -29,6 +30,29 @@ export const registerGatewayV2Router = async (server: FastifyZodProvider) => {
}
});
server.route({
method: "POST",
url: "/heartbeat",
config: {
rateLimit: writeLimit
},
schema: {
response: {
200: z.object({
message: z.string()
})
}
},
onRequest: verifyAuth([AuthMode.IDENTITY_ACCESS_TOKEN]),
handler: async (req) => {
await server.services.gatewayV2.heartbeat({
orgPermission: req.permission
});
return { message: "Successfully triggered heartbeat" };
}
});
server.route({
method: "GET",
url: "/",
@@ -46,4 +70,28 @@ export const registerGatewayV2Router = async (server: FastifyZodProvider) => {
return gateways;
}
});
server.route({
method: "DELETE",
url: "/:id",
config: {
rateLimit: writeLimit
},
schema: {
params: z.object({
id: z.string()
}),
response: {
200: z.any()
}
},
onRequest: verifyAuth([AuthMode.IDENTITY_ACCESS_TOKEN, AuthMode.JWT]),
handler: async (req) => {
const gateway = await server.services.gatewayV2.deleteGatewayById({
orgPermission: req.permission,
id: req.params.id
});
return { gateway };
}
});
};

View File

@@ -1,11 +1,43 @@
import { TDbClient } from "@app/db";
import { TableName } from "@app/db/schemas";
import { ormify } from "@app/lib/knex";
import { GatewaysV2Schema, TableName, TGatewaysV2 } from "@app/db/schemas";
import { DatabaseError } from "@app/lib/errors";
import { buildFindFilter, ormify, selectAllTableCols, TFindFilter, TFindOpt } from "@app/lib/knex";
export type TGatewayV2DALFactory = ReturnType<typeof gatewayV2DalFactory>;
export const gatewayV2DalFactory = (db: TDbClient) => {
const orm = ormify(db, TableName.GatewayV2);
return orm;
const find = async (filter: TFindFilter<TGatewaysV2>, { offset, limit, sort, tx }: TFindOpt<TGatewaysV2> = {}) => {
try {
const query = (tx || db)(TableName.GatewayV2)
// eslint-disable-next-line @typescript-eslint/no-misused-promises
.where(buildFindFilter(filter, TableName.GatewayV2))
.join(TableName.Identity, `${TableName.Identity}.id`, `${TableName.GatewayV2}.identityId`)
.join(
TableName.IdentityOrgMembership,
`${TableName.IdentityOrgMembership}.identityId`,
`${TableName.GatewayV2}.identityId`
)
.select(selectAllTableCols(TableName.GatewayV2))
.select(db.ref("name").withSchema(TableName.Identity).as("identityName"));
if (limit) void query.limit(limit);
if (offset) void query.offset(offset);
if (sort) {
void query.orderBy(sort.map(([column, order, nulls]) => ({ column: column as string, order, nulls })));
}
const docs = await query;
return docs.map((el) => ({
...GatewaysV2Schema.parse(el),
identity: { id: el.identityId, name: el.identityName }
}));
} catch (error) {
throw new DatabaseError({ error, name: `${TableName.GatewayV2}: Find` });
}
};
return { ...orm, find };
};

View File

@@ -1,9 +1,13 @@
import net from "node:net";
import * as x509 from "@peculiar/x509";
import { TProxies } from "@app/db/schemas";
import { PgSqlLock } from "@app/keystore/keystore";
import { crypto } from "@app/lib/crypto";
import { BadRequestError, NotFoundError } from "@app/lib/errors";
import { GatewayProxyProtocol } from "@app/lib/gateway/types";
import { withGatewayV2Proxy } from "@app/lib/gateway-v2/gateway-v2";
import { OrgServiceActor } from "@app/lib/types";
import { ActorType } from "@app/services/auth/auth-type";
import { constructPemChainFromCerts } from "@app/services/certificate/certificate-fns";
@@ -494,9 +498,115 @@ export const gatewayV2ServiceFactory = ({
};
};
const heartbeat = async ({ orgPermission }: { orgPermission: OrgServiceActor }) => {
const gateway = await gatewayV2DAL.findOne({
orgId: orgPermission.orgId,
identityId: orgPermission.id
});
if (!gateway) {
throw new NotFoundError({ message: `Gateway for identity ${orgPermission.id} not found.` });
}
const gatewayV2ConnectionDetails = await getPlatformConnectionDetailsByGatewayId({
gatewayId: gateway.id,
targetHost: "health-check",
targetPort: 443
});
if (!gatewayV2ConnectionDetails) {
throw new NotFoundError({ message: `Gateway connection details for gateway ${gateway.id} not found.` });
}
const isGatewayReachable = await withGatewayV2Proxy(
async (port) => {
return new Promise<boolean>((resolve, reject) => {
const socket = new net.Socket();
let responseReceived = false;
let isResolved = false;
// Set socket timeout
socket.setTimeout(10000);
const cleanup = () => {
if (!socket.destroyed) {
socket.destroy();
}
};
socket.on("data", (data: Buffer) => {
const response = data.toString().trim();
if (response === "PONG" && !isResolved) {
isResolved = true;
responseReceived = true;
cleanup();
resolve(true);
}
});
socket.on("error", (err: Error) => {
if (!isResolved) {
isResolved = true;
cleanup();
reject(new Error(`TCP connection error: ${err.message}`));
}
});
socket.on("timeout", () => {
if (!isResolved) {
isResolved = true;
cleanup();
reject(new Error("TCP connection timeout"));
}
});
socket.on("close", () => {
if (!isResolved && !responseReceived) {
isResolved = true;
cleanup();
reject(new Error("Connection closed without receiving PONG"));
}
});
socket.connect(port, "localhost");
});
},
{
protocol: GatewayProxyProtocol.Ping,
proxyIp: gatewayV2ConnectionDetails.proxyIp,
gateway: gatewayV2ConnectionDetails.gateway,
proxy: gatewayV2ConnectionDetails.proxy
}
);
if (!isGatewayReachable) {
throw new BadRequestError({ message: `Gateway ${gateway.id} is not reachable` });
}
await gatewayV2DAL.updateById(gateway.id, { heartbeat: new Date() });
};
const deleteGatewayById = async ({ orgPermission, id }: { orgPermission: OrgServiceActor; id: string }) => {
// const { permission } = await permissionService.getOrgPermission(
// orgPermission.type,
// orgPermission.id,
// orgPermission.orgId,
// orgPermission.authMethod,
// orgPermission.orgId
// );
// ForbiddenError.from(permission).throwUnlessCan(
// OrgPermissionGatewayActions.DeleteGateways,
// OrgPermissionSubjects.Gateway
// );
return gatewayV2DAL.deleteById(id);
};
return {
listGateways,
registerGateway,
getPlatformConnectionDetailsByGatewayId
getPlatformConnectionDetailsByGatewayId,
deleteGatewayById,
heartbeat
};
};

View File

@@ -9,11 +9,6 @@ import { BadRequestError } from "../errors";
import { GatewayProxyProtocol } from "../gateway/types";
import { logger } from "../logger";
/*
TODOs:
- Add heartbeat tracking to gateway connection
*/
interface IGatewayProxyServer {
server: net.Server;
port: number;
@@ -58,7 +53,9 @@ const createProxyConnection = async ({
});
socket.on("close", (hadError: boolean) => {
logger.error(`TLS connection closed${hadError ? " with error" : ""}`);
if (hadError) {
logger.error("TLS connection closed with error");
}
});
socket.on("timeout", () => {
@@ -175,6 +172,8 @@ const setupProxyServer = async ({
command += "\n";
} else if (protocol === GatewayProxyProtocol.Tcp) {
command += `FORWARD-TCP\n`;
} else if (protocol === GatewayProxyProtocol.Ping) {
command += `PING\n`;
} else {
throw new BadRequestError({
message: `Invalid protocol: ${protocol as string}`
@@ -222,7 +221,6 @@ const setupProxyServer = async ({
return;
}
console.log(`Gateway proxy started on port ${address.port}`);
resolve({
server,
port: address.port,
@@ -230,7 +228,7 @@ const setupProxyServer = async ({
try {
server.close();
} catch (err) {
console.debug("Error closing server:", err);
logger.debug("Error closing server:", err instanceof Error ? err.message : String(err));
}
},
getProxyError: () => proxyErrorMsg.join(",")

View File

@@ -6,7 +6,8 @@ export type TGatewayTlsOptions = { ca: string; cert: string; key: string };
export enum GatewayProxyProtocol {
Http = "http",
Tcp = "tcp"
Tcp = "tcp",
Ping = "ping"
}
export enum GatewayHttpProxyActions {

View File

@@ -1 +1 @@
export { gatewaysV2QueryKeys } from "./queries";
export * from "./mutations";

View File

@@ -0,0 +1,17 @@
import { useMutation, useQueryClient } from "@tanstack/react-query";
import { apiRequest } from "@app/config/request";
import { gatewaysQueryKeys } from "../gateways/queries";
export const useDeleteGatewayV2ById = () => {
const queryClient = useQueryClient();
return useMutation({
mutationFn: (id: string) => {
return apiRequest.delete(`/api/v2/gateways/${id}`);
},
onSuccess: () => {
queryClient.invalidateQueries(gatewaysQueryKeys.list());
}
});
};

View File

@@ -1,18 +0,0 @@
import { queryOptions } from "@tanstack/react-query";
import { apiRequest } from "@app/config/request";
import { TGatewayV2 } from "./types";
export const gatewaysV2QueryKeys = {
allKey: () => ["gateways-v2"],
listKey: () => [...gatewaysV2QueryKeys.allKey(), "list"],
list: () =>
queryOptions({
queryKey: gatewaysV2QueryKeys.listKey(),
queryFn: async () => {
const { data } = await apiRequest.get<{ gateways: TGatewayV2[] }>("/api/v2/gateways");
return data.gateways;
}
})
};

View File

@@ -4,6 +4,7 @@ export type TGatewayV2 = {
name: string;
createdAt: string;
updatedAt: string;
heartbeat: string;
identity: {
name: string;
id: string;

View File

@@ -15,7 +15,16 @@ export const gatewaysQueryKeys = {
const { data } = await apiRequest.get<{ gateways: TGateway[] }>("/api/v1/gateways");
const { data: dataV2 } = await apiRequest.get<TGatewayV2[]>("/api/v2/gateways");
return [...data.gateways, ...dataV2];
return [
...data.gateways.map((g) => ({
...g,
isV1: true
})),
...dataV2.map((g) => ({
...g,
isV1: false
}))
];
}
})
};

View File

@@ -14,7 +14,7 @@ import {
} from "@fortawesome/free-solid-svg-icons";
import { FontAwesomeIcon } from "@fortawesome/react-fontawesome";
import { useQuery } from "@tanstack/react-query";
import { format, formatRelative } from "date-fns";
import { formatRelative } from "date-fns";
import { createNotification } from "@app/components/notifications";
import { OrgPermissionCan } from "@app/components/permissions";
@@ -48,13 +48,14 @@ import {
import { withPermission } from "@app/hoc";
import { usePopUp } from "@app/hooks";
import { gatewaysQueryKeys, useDeleteGatewayById } from "@app/hooks/api/gateways";
import { useDeleteGatewayV2ById } from "@app/hooks/api/gateways-v2";
import { EditGatewayDetailsModal } from "./components/EditGatewayDetailsModal";
export const GatewayListPage = withPermission(
() => {
const [search, setSearch] = useState("");
const { data: gateways, isPending: isGatewayLoading } = useQuery(gatewaysQueryKeys.list());
const { data: gateways, isPending: isGatewaysLoading } = useQuery(gatewaysQueryKeys.list());
const { popUp, handlePopUpOpen, handlePopUpToggle } = usePopUp([
"deleteGateway",
@@ -62,16 +63,20 @@ export const GatewayListPage = withPermission(
] as const);
const deleteGatewayById = useDeleteGatewayById();
const deleteGatewayV2ById = useDeleteGatewayV2ById();
const handleDeleteGateway = async () => {
await deleteGatewayById.mutateAsync((popUp.deleteGateway.data as { id: string }).id, {
onSuccess: () => {
handlePopUpToggle("deleteGateway");
createNotification({
type: "success",
text: "Successfully delete gateway"
});
}
const data = popUp.deleteGateway.data as { id: string; isV1: boolean };
if (data.isV1) {
await deleteGatewayById.mutateAsync(data.id);
} else {
await deleteGatewayV2ById.mutateAsync(data.id);
}
handlePopUpToggle("deleteGateway");
createNotification({
type: "success",
text: "Successfully deleted gateway"
});
};
@@ -127,7 +132,6 @@ export const GatewayListPage = withPermission(
<THead>
<Tr>
<Th className="w-1/3">Name</Th>
<Th>Cert Issued At</Th>
<Th>Identity</Th>
<Th>
Health Check
@@ -143,13 +147,12 @@ export const GatewayListPage = withPermission(
</Tr>
</THead>
<TBody>
{isGatewayLoading && (
{isGatewaysLoading && (
<TableSkeleton innerKey="gateway-table" columns={4} key="gateway-table" />
)}
{filteredGateway?.map((el) => (
<Tr key={el.id}>
<Td>{el.name}</Td>
<Td>{format(new Date(el.issuedAt), "yyyy-MM-dd hh:mm:ss aaa")}</Td>
<Td>{el.identity.name}</Td>
<Td>
{el.heartbeat
@@ -176,20 +179,22 @@ export const GatewayListPage = withPermission(
>
Copy ID
</DropdownMenuItem>
<OrgPermissionCan
I={OrgGatewayPermissionActions.EditGateways}
a={OrgPermissionSubjects.Gateway}
>
{(isAllowed: boolean) => (
<DropdownMenuItem
isDisabled={!isAllowed}
icon={<FontAwesomeIcon icon={faEdit} />}
onClick={() => handlePopUpOpen("editDetails", el)}
>
Edit Details
</DropdownMenuItem>
)}
</OrgPermissionCan>
{el.isV1 && (
<OrgPermissionCan
I={OrgGatewayPermissionActions.EditGateways}
a={OrgPermissionSubjects.Gateway}
>
{(isAllowed: boolean) => (
<DropdownMenuItem
isDisabled={!isAllowed}
icon={<FontAwesomeIcon icon={faEdit} />}
onClick={() => handlePopUpOpen("editDetails", el)}
>
Edit Details
</DropdownMenuItem>
)}
</OrgPermissionCan>
)}
<OrgPermissionCan
I={OrgPermissionAppConnectionActions.Delete}
a={OrgPermissionSubjects.AppConnections}
@@ -224,7 +229,7 @@ export const GatewayListPage = withPermission(
/>
</ModalContent>
</Modal>
{!isGatewayLoading && !filteredGateway?.length && (
{!isGatewaysLoading && !filteredGateway?.length && (
<EmptyState
title={
gateways?.length
@@ -251,8 +256,5 @@ export const GatewayListPage = withPermission(
</div>
);
},
{
action: OrgPermissionAppConnectionActions.Read,
subject: OrgPermissionSubjects.AppConnections
}
{ action: OrgGatewayPermissionActions.ListGateways, subject: OrgPermissionSubjects.Gateway }
);