135 lines
5.7 KiB
TypeScript
135 lines
5.7 KiB
TypeScript
// R-P5 (P5) §P2d — shared per-tenant maintenance for DataConnections, used by the
|
|
// worker crons (token refresh + health sweep). Lives here (not apps/web) because
|
|
// the worker only depends on @hivegrid/db + @hivegrid/integrations, not apps/web.
|
|
//
|
|
// Both functions take an active Prisma client/tx (the caller wraps them in the
|
|
// worker's withTenantRLS so the tenant GUC is set) and operate ONLY on that
|
|
// tenant's rows. Credentials are decrypted/re-encrypted via the per-tenant DEK
|
|
// envelope (Phase 1), passing the same client so the crypto store stays in the
|
|
// tenant scope.
|
|
|
|
import { decryptTenantField, encryptTenantField } from "@hivegrid/db/crypto/tenant-envelope.server";
|
|
import { createNotificationDelivery } from "@hivegrid/db/notifications.server";
|
|
import { resolveAdapter } from "./registry.js";
|
|
|
|
/** Minimal structural client shape (satisfied by the full client or a tx). */
|
|
type ConnectionRow = {
|
|
id: string;
|
|
appId: string;
|
|
credentialsCiphertext: string | null;
|
|
connectionStatus: string;
|
|
createdByUserId: string | null;
|
|
};
|
|
|
|
type WorkerDataConnectionClient = {
|
|
dataConnection: {
|
|
findMany(args: {
|
|
where: Record<string, unknown>;
|
|
select: { id: true; appId: true; credentialsCiphertext: true; connectionStatus: true; createdByUserId: true };
|
|
}): Promise<ConnectionRow[]>;
|
|
updateMany(args: { where: { id: string; tenantId: string }; data: Record<string, unknown> }): Promise<{ count: number }>;
|
|
};
|
|
};
|
|
|
|
const CONNECTION_SELECT = { id: true, appId: true, credentialsCiphertext: true, connectionStatus: true, createdByUserId: true } as const;
|
|
|
|
async function notifyOwner(tenantId: string, ownerUserId: string | null, connectionId: string, health: string): Promise<void> {
|
|
if (!ownerUserId) {
|
|
return;
|
|
}
|
|
await createNotificationDelivery({
|
|
tenantId,
|
|
userId: ownerUserId,
|
|
channel: "in-app",
|
|
category: "system",
|
|
messageKey: "integrations.notify.connectionUnhealthy",
|
|
status: "PENDING",
|
|
payloadMetadata: { source: "integrations", connectionId, health },
|
|
}).catch(() => undefined);
|
|
}
|
|
|
|
/**
|
|
* Refresh OAuth tokens that are within `windowMs` of expiry for one tenant. Only
|
|
* adapters that implement refreshToken() participate; api_key/rest_api connections
|
|
* are skipped. A refresh failure marks the connection broken + notifies the owner.
|
|
* Returns the number of connections refreshed.
|
|
*/
|
|
export async function refreshExpiringTokensForTenant(
|
|
client: WorkerDataConnectionClient,
|
|
tenantId: string,
|
|
opts?: { now?: Date; windowMs?: number },
|
|
): Promise<{ refreshed: number; failed: number }> {
|
|
const now = opts?.now ?? new Date();
|
|
const horizon = new Date(now.getTime() + (opts?.windowMs ?? 10 * 60_000));
|
|
const connections = await client.dataConnection.findMany({
|
|
where: { tenantId, archivedAt: null, connectionStatus: "active", tokenExpiresAt: { not: null, lte: horizon } },
|
|
select: CONNECTION_SELECT,
|
|
});
|
|
|
|
let refreshed = 0;
|
|
let failed = 0;
|
|
for (const connection of connections) {
|
|
const adapter = resolveAdapter(connection.appId);
|
|
if (!adapter?.refreshToken || !connection.credentialsCiphertext) {
|
|
continue;
|
|
}
|
|
try {
|
|
const creds = JSON.parse(await decryptTenantField(tenantId, connection.credentialsCiphertext, { client: client as never })) as Record<string, unknown>;
|
|
const result = await adapter.refreshToken(creds);
|
|
const ciphertext = await encryptTenantField(tenantId, JSON.stringify(result.creds), { client: client as never });
|
|
await client.dataConnection.updateMany({
|
|
where: { id: connection.id, tenantId },
|
|
data: { credentialsCiphertext: ciphertext, tokenExpiresAt: result.expiresAt, connectionStatus: "active", lastHealthCheck: now },
|
|
});
|
|
refreshed += 1;
|
|
} catch {
|
|
failed += 1;
|
|
await client.dataConnection.updateMany({ where: { id: connection.id, tenantId }, data: { connectionStatus: "broken", healthMessageKey: "integrations.health.refreshFailed", lastHealthCheck: now } });
|
|
await notifyOwner(tenantId, connection.createdByUserId, connection.id, "broken");
|
|
}
|
|
}
|
|
return { refreshed, failed };
|
|
}
|
|
|
|
/**
|
|
* Probe every active connection's health for one tenant, persisting the new
|
|
* status. Degraded/broken transitions notify the owner. Returns counts.
|
|
*/
|
|
export async function sweepConnectionHealthForTenant(
|
|
client: WorkerDataConnectionClient,
|
|
tenantId: string,
|
|
opts?: { now?: Date },
|
|
): Promise<{ checked: number; unhealthy: number }> {
|
|
const now = opts?.now ?? new Date();
|
|
const connections = await client.dataConnection.findMany({
|
|
where: { tenantId, archivedAt: null, connectionStatus: { in: ["active", "degraded"] } },
|
|
select: CONNECTION_SELECT,
|
|
});
|
|
|
|
let checked = 0;
|
|
let unhealthy = 0;
|
|
for (const connection of connections) {
|
|
const adapter = resolveAdapter(connection.appId);
|
|
let health: "active" | "degraded" | "broken" = "broken";
|
|
let messageKey = "integrations.health.notConnected";
|
|
if (adapter && connection.credentialsCiphertext) {
|
|
try {
|
|
const creds = JSON.parse(await decryptTenantField(tenantId, connection.credentialsCiphertext, { client: client as never })) as Record<string, unknown>;
|
|
const result = await adapter.healthCheck(creds);
|
|
health = result.health;
|
|
messageKey = result.messageKey;
|
|
} catch {
|
|
health = "broken";
|
|
messageKey = "integrations.health.probeFailed";
|
|
}
|
|
}
|
|
await client.dataConnection.updateMany({ where: { id: connection.id, tenantId }, data: { connectionStatus: health, lastHealthCheck: now, healthMessageKey: messageKey } });
|
|
checked += 1;
|
|
if (health !== "active") {
|
|
unhealthy += 1;
|
|
await notifyOwner(tenantId, connection.createdByUserId, connection.id, health);
|
|
}
|
|
}
|
|
return { checked, unhealthy };
|
|
}
|