586 lines
24 KiB
TypeScript
586 lines
24 KiB
TypeScript
// R-P5 (P5) §P2c — Apps-Engine domain service: DataConnection lifecycle, sync
|
|
// runs, and health monitoring. NO React/UI imports; routes/worker call these and
|
|
// build View-Models from the rows.
|
|
//
|
|
// Tenant isolation: every query is `where: { tenantId }` AND runs inside
|
|
// withTenantRLS() (RLS-ready; functional no-op until apps-engine-isolation.sql is
|
|
// activated at the P5 cutover). Credentials are stored ENVELOPE-ENCRYPTED via the
|
|
// per-tenant DEK (R-P5 Phase 1) — never plaintext, never selected into list VMs.
|
|
// Connect/sync/health route through the type-safe adapter registry (DoD #1/#2/#4).
|
|
|
|
import { CatalogError, foundationErrors, notify, withIdempotency, type AdapterResult, type AdapterSyncOutcome } from "@hivegrid/foundation";
|
|
import { createAuditEvent } from "@hivegrid/db/audit-events.server";
|
|
import { db } from "~/utils/db.server";
|
|
import { decryptTenantField, encryptTenantField } from "@hivegrid/db/crypto/tenant-envelope.server";
|
|
import { withTenantRLS, type TenantRlsClient } from "~/utils/db/.server/tenantRls.server";
|
|
import { resolveAdapter } from "@hivegrid/integrations";
|
|
|
|
const APPS_AUDIT_MODULE = "integrations";
|
|
|
|
/** Domain error carrying an i18n message key (M002 Klartext). */
|
|
export class AppsEngineServiceError extends Error {
|
|
readonly messageKey: string;
|
|
constructor(messageKey: string) {
|
|
super(messageKey);
|
|
this.name = "AppsEngineServiceError";
|
|
this.messageKey = messageKey;
|
|
}
|
|
}
|
|
|
|
export type ConnectionStatus = "pending" | "active" | "degraded" | "broken" | "revoked";
|
|
export type SyncKind = "manual" | "scheduled" | "webhook";
|
|
export type SyncStatus = "pending" | "running" | "completed" | "partial" | "failed" | "cancelled";
|
|
|
|
/** A connection as shown in a list/detail VM — NEVER includes credentials. */
|
|
export type DataConnectionRow = {
|
|
id: string;
|
|
tenantId: string;
|
|
appId: string;
|
|
label: string;
|
|
connectionStatus: ConnectionStatus;
|
|
lastHealthCheck: Date | null;
|
|
healthMessageKey: string | null;
|
|
tokenExpiresAt: Date | null;
|
|
createdByUserId: string | null;
|
|
isDemoData: boolean;
|
|
createdAt: Date;
|
|
archivedAt: Date | null;
|
|
};
|
|
|
|
const CONNECTION_SELECT = {
|
|
id: true,
|
|
tenantId: true,
|
|
appId: true,
|
|
label: true,
|
|
connectionStatus: true,
|
|
lastHealthCheck: true,
|
|
healthMessageKey: true,
|
|
tokenExpiresAt: true,
|
|
createdByUserId: true,
|
|
isDemoData: true,
|
|
createdAt: true,
|
|
archivedAt: true,
|
|
} as const;
|
|
|
|
// --- Pure helpers (unit-tested, no DB) ------------------------------------
|
|
|
|
/** Aggregate an adapter sync outcome into AppSyncRun counters + a terminal status. */
|
|
export function aggregateSyncOutcome(outcome: AdapterSyncOutcome): {
|
|
itemsTotal: number;
|
|
itemsSucceeded: number;
|
|
itemsFailed: number;
|
|
status: "completed" | "partial" | "failed";
|
|
summaryKey: string;
|
|
} {
|
|
const itemsTotal = outcome.items.length;
|
|
const itemsFailed = outcome.items.filter((item) => item.status === "failed").length;
|
|
const itemsSucceeded = outcome.items.filter((item) => item.status === "synced").length;
|
|
const status = itemsFailed === 0 ? "completed" : itemsSucceeded === 0 ? "failed" : "partial";
|
|
return { itemsTotal, itemsSucceeded, itemsFailed, status, summaryKey: outcome.summaryKey };
|
|
}
|
|
|
|
// --- Credential crypto (envelope, per-tenant DEK) -------------------------
|
|
|
|
async function encryptCredentials(tenantId: string, credentials: Record<string, unknown>): Promise<string> {
|
|
return encryptTenantField(tenantId, JSON.stringify(credentials));
|
|
}
|
|
|
|
async function decryptCredentials(
|
|
tenantId: string,
|
|
ciphertext: string,
|
|
client?: TenantRlsClient,
|
|
): Promise<Record<string, unknown>> {
|
|
return JSON.parse(
|
|
await decryptTenantField(tenantId, ciphertext, client ? { client: client as never } : undefined),
|
|
) as Record<string, unknown>;
|
|
}
|
|
|
|
/**
|
|
* Execute provider work with one tenant-scoped DataConnection's decrypted
|
|
* credentials. The callback is the only intended plaintext boundary and must
|
|
* never return or persist the credential object. The database transaction ends
|
|
* before external provider I/O starts.
|
|
*/
|
|
export async function runWithConnectionCredentials<T>(
|
|
input: { tenantId: string; connectionId: string; expectedAppId: string },
|
|
callback: (credentials: Readonly<Record<string, unknown>>, connection: { id: string; appId: string; label: string }) => Promise<T>,
|
|
): Promise<T> {
|
|
const loaded = await withTenantRLS(input.tenantId, async (tx) => {
|
|
const connection = await tx.dataConnection.findFirst({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId, archivedAt: null },
|
|
select: { id: true, appId: true, label: true, connectionStatus: true, credentialsCiphertext: true },
|
|
});
|
|
if (!connection || connection.appId !== input.expectedAppId) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
if ((connection.connectionStatus !== "active" && connection.connectionStatus !== "degraded") || !connection.credentialsCiphertext) {
|
|
throw new AppsEngineServiceError("integrations.error.connectionUnavailable");
|
|
}
|
|
|
|
return {
|
|
connection: { id: connection.id, appId: connection.appId, label: connection.label },
|
|
credentials: Object.freeze(await decryptCredentials(input.tenantId, connection.credentialsCiphertext, tx)),
|
|
};
|
|
});
|
|
return callback(loaded.credentials, loaded.connection);
|
|
}
|
|
|
|
// --- Connection lifecycle (DoD #1/#6/#7) ----------------------------------
|
|
|
|
/** List a tenant's connections (newest first). Credentials are never selected. */
|
|
export async function listConnections(tenantId: string, opts?: { appId?: string; includeArchived?: boolean }): Promise<DataConnectionRow[]> {
|
|
const rows = await withTenantRLS(tenantId, (tx) =>
|
|
tx.dataConnection.findMany({
|
|
where: {
|
|
tenantId,
|
|
...(opts?.appId ? { appId: opts.appId } : {}),
|
|
...(opts?.includeArchived ? {} : { archivedAt: null }),
|
|
},
|
|
select: CONNECTION_SELECT,
|
|
orderBy: { createdAt: "desc" },
|
|
}),
|
|
);
|
|
return rows as DataConnectionRow[];
|
|
}
|
|
|
|
export async function getConnectionOrThrow(tenantId: string, connectionId: string, expectedAppId: string): Promise<DataConnectionRow> {
|
|
const row = await withTenantRLS(tenantId, (tx) =>
|
|
tx.dataConnection.findFirst({ where: { id: connectionId, tenantId, appId: expectedAppId }, select: CONNECTION_SELECT }),
|
|
);
|
|
if (!row) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId });
|
|
}
|
|
return row as DataConnectionRow;
|
|
}
|
|
|
|
export type CreateConnectionInput = {
|
|
tenantId: string;
|
|
appId: string;
|
|
label: string;
|
|
credentials: Record<string, unknown>;
|
|
createdByUserId: string;
|
|
isDemoData?: boolean;
|
|
};
|
|
|
|
export type ActivateOAuthConnectionInput = {
|
|
tenantId: string;
|
|
connectionId: string;
|
|
appId: string;
|
|
credentials: Record<string, unknown>;
|
|
tokenExpiresAt: Date | null;
|
|
actorUserId: string;
|
|
/** DataConnection.updatedAt captured before redirecting to the provider. */
|
|
expectedUpdatedAt: Date;
|
|
};
|
|
|
|
/**
|
|
* Connect a new DataConnection: validate the credentials via the app's adapter,
|
|
* then store them envelope-encrypted with status `active`. Multiple connections
|
|
* per (tenant, app) are allowed (DoD #7). Validation failure surfaces the
|
|
* adapter's Klartext message (M002) without persisting.
|
|
*/
|
|
export async function createConnection(input: CreateConnectionInput): Promise<{ id: string }> {
|
|
const label = input.label.trim();
|
|
if (!label) {
|
|
throw new AppsEngineServiceError("integrations.error.labelRequired");
|
|
}
|
|
const adapter = resolveAdapter(input.appId);
|
|
if (!adapter) {
|
|
throw new AppsEngineServiceError("integrations.error.unknownApp");
|
|
}
|
|
const result = await adapter.validateConnection(input.credentials);
|
|
if (!result.ok) {
|
|
throw new AppsEngineServiceError(result.messageKey);
|
|
}
|
|
const ciphertext = await encryptCredentials(input.tenantId, input.credentials);
|
|
const created = await withTenantRLS(input.tenantId, (tx) =>
|
|
tx.dataConnection.create({
|
|
data: {
|
|
tenantId: input.tenantId,
|
|
appId: input.appId,
|
|
label,
|
|
credentialsCiphertext: ciphertext,
|
|
connectionStatus: "active",
|
|
lastHealthCheck: new Date(),
|
|
healthMessageKey: result.messageKey,
|
|
createdByUserId: input.createdByUserId,
|
|
isDemoData: input.isDemoData ?? false,
|
|
},
|
|
select: { id: true },
|
|
}),
|
|
);
|
|
await audit(input.tenantId, input.createdByUserId, "INTEGRATION_CONNECT", created.id, { appId: input.appId });
|
|
return created;
|
|
}
|
|
|
|
/**
|
|
* Activate a previously reserved OAuth connection after the provider callback.
|
|
*
|
|
* Provider validation deliberately finishes before the database transaction so
|
|
* a slow or unavailable OAuth/resource endpoint never holds RLS database locks.
|
|
* Only the validated credential bundle crosses the envelope-encryption boundary.
|
|
*/
|
|
export async function activateOAuthConnection(input: ActivateOAuthConnectionInput): Promise<void> {
|
|
const adapter = resolveAdapter(input.appId);
|
|
if (!adapter || adapter.connectionType !== "oauth2") {
|
|
throw new AppsEngineServiceError("integrations.error.unknownApp");
|
|
}
|
|
|
|
const result = await adapter.validateConnection(input.credentials);
|
|
if (!result.ok) {
|
|
throw new AppsEngineServiceError(result.messageKey);
|
|
}
|
|
|
|
const ciphertext = await encryptCredentials(input.tenantId, input.credentials);
|
|
const updated = await withTenantRLS(input.tenantId, (tx) =>
|
|
tx.dataConnection.updateMany({
|
|
where: {
|
|
id: input.connectionId,
|
|
tenantId: input.tenantId,
|
|
appId: input.appId,
|
|
archivedAt: null,
|
|
updatedAt: input.expectedUpdatedAt,
|
|
},
|
|
data: {
|
|
credentialsCiphertext: ciphertext,
|
|
tokenExpiresAt: input.tokenExpiresAt,
|
|
connectionStatus: result.health,
|
|
lastHealthCheck: new Date(),
|
|
healthMessageKey: result.messageKey,
|
|
},
|
|
}),
|
|
);
|
|
if (updated.count !== 1) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
|
|
await audit(input.tenantId, input.actorUserId, "INTEGRATION_OAUTH_CONNECT", input.connectionId, {
|
|
appId: input.appId,
|
|
});
|
|
}
|
|
|
|
/** Reconnect an existing connection (same id, no history loss): re-validate +
|
|
* re-encrypt new credentials, status back to `active`. */
|
|
export async function reconnectConnection(input: {
|
|
tenantId: string;
|
|
connectionId: string;
|
|
expectedAppId: string;
|
|
credentials: Record<string, unknown>;
|
|
actorUserId: string;
|
|
}): Promise<void> {
|
|
await withTenantRLS(input.tenantId, async (tx) => {
|
|
const existing = await tx.dataConnection.findFirst({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId },
|
|
select: { id: true, appId: true },
|
|
});
|
|
if (!existing) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
const adapter = resolveAdapter(existing.appId);
|
|
if (!adapter) {
|
|
throw new AppsEngineServiceError("integrations.error.unknownApp");
|
|
}
|
|
const result = await adapter.validateConnection(input.credentials);
|
|
if (!result.ok) {
|
|
throw new AppsEngineServiceError(result.messageKey);
|
|
}
|
|
const ciphertext = await encryptCredentials(input.tenantId, input.credentials);
|
|
await tx.dataConnection.updateMany({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId },
|
|
data: { credentialsCiphertext: ciphertext, connectionStatus: "active", lastHealthCheck: new Date(), healthMessageKey: result.messageKey, archivedAt: null },
|
|
});
|
|
});
|
|
await audit(input.tenantId, input.actorUserId, "INTEGRATION_RECONNECT", input.connectionId, {});
|
|
}
|
|
|
|
/** Disconnect: set status `revoked` (NOT hard-delete → sync history preserved). */
|
|
export async function disconnectConnection(input: { tenantId: string; connectionId: string; expectedAppId: string; actorUserId: string }): Promise<void> {
|
|
await withTenantRLS(input.tenantId, async (tx) => {
|
|
const existing = await tx.dataConnection.findFirst({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId },
|
|
select: { id: true },
|
|
});
|
|
if (!existing) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
// Keep the Klartext consistent with the revoked badge (A.1): without this
|
|
// the card keeps showing the stale "connection active" health message.
|
|
await tx.dataConnection.updateMany({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId },
|
|
data: {
|
|
connectionStatus: "revoked",
|
|
credentialsCiphertext: null,
|
|
tokenExpiresAt: null,
|
|
healthMessageKey: "integrations.health.notConnected",
|
|
lastHealthCheck: new Date(),
|
|
},
|
|
});
|
|
});
|
|
await audit(input.tenantId, input.actorUserId, "INTEGRATION_DISCONNECT", input.connectionId, {});
|
|
}
|
|
|
|
// --- Sync (DoD #3/#8) -----------------------------------------------------
|
|
|
|
export type AppSyncRunRow = {
|
|
id: string;
|
|
tenantId: string;
|
|
dataConnectionId: string;
|
|
kind: SyncKind;
|
|
status: SyncStatus;
|
|
itemsTotal: number;
|
|
itemsSucceeded: number;
|
|
itemsFailed: number;
|
|
summaryText: string | null;
|
|
startedAt: Date | null;
|
|
completedAt: Date | null;
|
|
createdAt: Date;
|
|
};
|
|
|
|
/**
|
|
* Run a sync for a connection NOW (manual re-sync, DoD #8) or from the worker
|
|
* (scheduled/webhook). Resolves the adapter, decrypts credentials on demand,
|
|
* runs the adapter sync, and persists the AppSyncRun + items + a summary log.
|
|
* A broken connection short-circuits with a failed run.
|
|
*/
|
|
export async function runSyncForConnection(input: {
|
|
tenantId: string;
|
|
connectionId: string;
|
|
expectedAppId: string;
|
|
kind?: SyncKind;
|
|
createdByUserId?: string | null;
|
|
}): Promise<{ runId: string; status: SyncStatus }> {
|
|
return withTenantRLS(input.tenantId, async (tx) => {
|
|
const connection = await tx.dataConnection.findFirst({
|
|
where: { id: input.connectionId, tenantId: input.tenantId, appId: input.expectedAppId },
|
|
select: { id: true, appId: true, connectionStatus: true, credentialsCiphertext: true },
|
|
});
|
|
if (!connection) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
const run = await tx.appSyncRun.create({
|
|
data: { tenantId: input.tenantId, dataConnectionId: input.connectionId, kind: input.kind ?? "manual", status: "running", startedAt: new Date(), createdByUserId: input.createdByUserId ?? null },
|
|
select: { id: true },
|
|
});
|
|
|
|
const adapter = resolveAdapter(connection.appId);
|
|
if (!adapter || connection.connectionStatus === "revoked" || !connection.credentialsCiphertext) {
|
|
await tx.appSyncRun.updateMany({ where: { id: run.id, tenantId: input.tenantId }, data: { status: "failed", completedAt: new Date(), summaryText: "integrations.sync.summary.notConnected" } });
|
|
await tx.appSyncRunLog.create({ data: { tenantId: input.tenantId, syncRunId: run.id, severity: "error", messageText: "integrations.sync.log.notConnected" } });
|
|
return { runId: run.id, status: "failed" as SyncStatus };
|
|
}
|
|
|
|
let outcome: AdapterSyncOutcome;
|
|
try {
|
|
const credentials = await decryptCredentials(input.tenantId, connection.credentialsCiphertext);
|
|
outcome = await adapter.sync(credentials);
|
|
} catch {
|
|
await tx.appSyncRun.updateMany({ where: { id: run.id, tenantId: input.tenantId }, data: { status: "failed", completedAt: new Date(), summaryText: "integrations.sync.summary.adapterError" } });
|
|
await tx.appSyncRunLog.create({ data: { tenantId: input.tenantId, syncRunId: run.id, severity: "error", messageText: "integrations.sync.log.adapterError" } });
|
|
return { runId: run.id, status: "failed" as SyncStatus };
|
|
}
|
|
|
|
const agg = aggregateSyncOutcome(outcome);
|
|
if (outcome.items.length > 0) {
|
|
await tx.appSyncItem.createMany({
|
|
data: outcome.items.map((item) => ({
|
|
tenantId: input.tenantId,
|
|
syncRunId: run.id,
|
|
entityType: item.entityType,
|
|
entityId: item.entityId ?? null,
|
|
externalId: item.externalId,
|
|
operation: item.operation,
|
|
status: item.status,
|
|
errorText: item.errorText ?? null,
|
|
})),
|
|
});
|
|
}
|
|
await tx.appSyncRun.updateMany({
|
|
where: { id: run.id, tenantId: input.tenantId },
|
|
data: { status: agg.status, itemsTotal: agg.itemsTotal, itemsSucceeded: agg.itemsSucceeded, itemsFailed: agg.itemsFailed, summaryText: agg.summaryKey, completedAt: new Date() },
|
|
});
|
|
await tx.appSyncRunLog.create({ data: { tenantId: input.tenantId, syncRunId: run.id, severity: agg.status === "failed" ? "error" : "success", messageText: agg.summaryKey } });
|
|
return { runId: run.id, status: agg.status };
|
|
});
|
|
}
|
|
|
|
export async function listSyncRuns(tenantId: string, connectionId: string, expectedAppId: string, limit = 50): Promise<AppSyncRunRow[]> {
|
|
const rows = await withTenantRLS(tenantId, (tx) =>
|
|
tx.appSyncRun.findMany({
|
|
where: { tenantId, dataConnectionId: connectionId, dataConnection: { tenantId, appId: expectedAppId } },
|
|
orderBy: { createdAt: "desc" },
|
|
take: Math.min(Math.max(limit, 1), 100),
|
|
}),
|
|
);
|
|
return rows as AppSyncRunRow[];
|
|
}
|
|
|
|
export type AppSyncItemRowData = {
|
|
id: string;
|
|
entityType: string;
|
|
entityId: string | null;
|
|
externalId: string;
|
|
operation: "created" | "updated" | "deleted" | "skipped";
|
|
status: "pending" | "synced" | "failed" | "skipped";
|
|
errorText: string | null;
|
|
createdAt: Date;
|
|
};
|
|
|
|
export type AppSyncRunLogRowData = {
|
|
id: string;
|
|
severity: "info" | "success" | "warning" | "error";
|
|
messageText: string;
|
|
createdAt: Date;
|
|
};
|
|
|
|
export async function getSyncRunWithItems(
|
|
tenantId: string,
|
|
runId: string,
|
|
connectionId: string,
|
|
expectedAppId: string,
|
|
): Promise<{ run: AppSyncRunRow; items: AppSyncItemRowData[]; logs: AppSyncRunLogRowData[] }> {
|
|
return withTenantRLS(tenantId, async (tx) => {
|
|
const run = await tx.appSyncRun.findFirst({
|
|
where: { id: runId, tenantId, dataConnectionId: connectionId, dataConnection: { tenantId, appId: expectedAppId } },
|
|
});
|
|
if (!run) {
|
|
throw new CatalogError(foundationErrors.notFound, { runId });
|
|
}
|
|
const [items, logs] = await Promise.all([
|
|
tx.appSyncItem.findMany({
|
|
where: { tenantId, syncRunId: runId },
|
|
select: { id: true, entityType: true, entityId: true, externalId: true, operation: true, status: true, errorText: true, createdAt: true },
|
|
orderBy: { createdAt: "asc" },
|
|
}),
|
|
tx.appSyncRunLog.findMany({
|
|
where: { tenantId, syncRunId: runId },
|
|
select: { id: true, severity: true, messageText: true, createdAt: true },
|
|
orderBy: { createdAt: "asc" },
|
|
}),
|
|
]);
|
|
return { run: run as AppSyncRunRow, items: items as AppSyncItemRowData[], logs: logs as AppSyncRunLogRowData[] };
|
|
});
|
|
}
|
|
|
|
// --- Webhook-triggered sync with storm protection (DoD #8) ----------------
|
|
|
|
/** Dedup window for inbound webhooks: identical events within this window collapse
|
|
* to a single sync (no storm). */
|
|
const WEBHOOK_DEDUP_WINDOW_MS = 60_000;
|
|
|
|
/**
|
|
* Trigger a sync from an inbound webhook, GUARDED by withIdempotency (M007) so a
|
|
* webhook storm (the same event delivered many times) runs the sync ONCE. The
|
|
* concrete per-app webhook ingest routes (P14/P15/P16) call this with a stable
|
|
* `webhookEventKey` (provider event id / signature). `replayed = true` means the
|
|
* call was deduped (a storm hit), not a fresh sync.
|
|
*/
|
|
export async function triggerWebhookSync(input: {
|
|
tenantId: string;
|
|
connectionId: string;
|
|
expectedAppId: string;
|
|
webhookEventKey: string;
|
|
}): Promise<{ runId: string | null; status: SyncStatus | "replayed"; replayed: boolean }> {
|
|
const result = await withIdempotency(
|
|
{
|
|
tenantId: input.tenantId,
|
|
scope: `integrations-webhook:${input.tenantId}:${input.connectionId}`,
|
|
key: input.webhookEventKey,
|
|
fingerprint: { connectionId: input.connectionId },
|
|
expiresAt: new Date(Date.now() + WEBHOOK_DEDUP_WINDOW_MS),
|
|
},
|
|
// db has the IdempotencyKey model; the structural cast bridges Prisma's broader
|
|
// delegate signatures to the narrow IdempotencyClient seam (M007).
|
|
db as unknown as Parameters<typeof withIdempotency>[1],
|
|
async () => {
|
|
const sync = await runSyncForConnection({
|
|
tenantId: input.tenantId,
|
|
connectionId: input.connectionId,
|
|
expectedAppId: input.expectedAppId,
|
|
kind: "webhook",
|
|
});
|
|
return { status: 200, body: { runId: sync.runId, status: sync.status } };
|
|
},
|
|
);
|
|
const body = result.body as { runId?: string; status?: SyncStatus } | undefined;
|
|
return { runId: body?.runId ?? null, status: result.replayed ? "replayed" : (body?.status ?? "pending"), replayed: result.replayed };
|
|
}
|
|
|
|
// --- Health monitoring (DoD #4) -------------------------------------------
|
|
|
|
/**
|
|
* Probe a connection's health via its adapter and persist the new status. On a
|
|
* degraded/broken transition, notify the connection owner (who holds
|
|
* integrations.connect) so they can act.
|
|
*/
|
|
export async function checkConnectionHealth(input: {
|
|
tenantId: string;
|
|
connectionId: string;
|
|
notificationDeeplink: string;
|
|
}): Promise<{ status: ConnectionStatus; messageKey: string }> {
|
|
const outcome = await withTenantRLS(input.tenantId, async (tx) => {
|
|
const connection = await tx.dataConnection.findFirst({ where: { id: input.connectionId, tenantId: input.tenantId }, select: { id: true, appId: true, connectionStatus: true, credentialsCiphertext: true, createdByUserId: true } });
|
|
if (!connection) {
|
|
throw new CatalogError(foundationErrors.notFound, { connectionId: input.connectionId });
|
|
}
|
|
const adapter = resolveAdapter(connection.appId);
|
|
let result: AdapterResult;
|
|
if (!adapter || connection.connectionStatus === "revoked" || !connection.credentialsCiphertext) {
|
|
result = { ok: false, health: "broken", messageKey: "integrations.health.notConnected" };
|
|
} else {
|
|
try {
|
|
const credentials = await decryptCredentials(input.tenantId, connection.credentialsCiphertext);
|
|
result = await adapter.healthCheck(credentials);
|
|
} catch {
|
|
result = { ok: false, health: "broken", messageKey: "integrations.health.probeFailed" };
|
|
}
|
|
}
|
|
const status: ConnectionStatus = connection.connectionStatus === "revoked" ? "revoked" : result.health;
|
|
await tx.dataConnection.updateMany({ where: { id: input.connectionId, tenantId: input.tenantId }, data: { connectionStatus: status, lastHealthCheck: new Date(), healthMessageKey: result.messageKey } });
|
|
return { status, messageKey: result.messageKey, ownerUserId: connection.createdByUserId, degradedOrBroken: status === "degraded" || status === "broken" };
|
|
});
|
|
|
|
if (outcome.degradedOrBroken && outcome.ownerUserId) {
|
|
// Notify the owner through the M011 notify() contract on the request-scoped
|
|
// db client (RLS context is bound by the surrounding request). Critical
|
|
// category: a broken connection halts this tenant's sync, so the alert is
|
|
// delivered regardless of preferences; the sourceKey dedupes repeated probes
|
|
// into one delivery. A notify failure must not fail the probe.
|
|
await notify(
|
|
{
|
|
tenantId: input.tenantId,
|
|
userId: outcome.ownerUserId,
|
|
channel: "IN_APP",
|
|
category: "critical",
|
|
messageKey: "integrations.notify.connectionUnhealthy",
|
|
sourceKey: `integrations:connection-unhealthy:${input.connectionId}`,
|
|
payloadMetadata: {
|
|
source: APPS_AUDIT_MODULE,
|
|
connectionId: input.connectionId,
|
|
health: outcome.status,
|
|
deeplink: input.notificationDeeplink,
|
|
},
|
|
},
|
|
db as unknown as Parameters<typeof notify>[1]
|
|
).catch(() => undefined);
|
|
}
|
|
return { status: outcome.status, messageKey: outcome.messageKey };
|
|
}
|
|
|
|
// --- Audit helper ---------------------------------------------------------
|
|
|
|
async function audit(tenantId: string, actorId: string | null, action: string, targetId: string, metadata: Record<string, unknown>): Promise<void> {
|
|
await createAuditEvent({
|
|
tenantId,
|
|
actorId,
|
|
actorType: "user",
|
|
action,
|
|
targetType: "DataConnection",
|
|
targetId,
|
|
sourceModule: APPS_AUDIT_MODULE,
|
|
outcome: "success",
|
|
metadata: metadata as never,
|
|
}).catch(() => undefined);
|
|
}
|
|
|
|
export type { TenantRlsClient };
|