319 lines
14 KiB
TypeScript
319 lines
14 KiB
TypeScript
/**
|
|
* Datto RMM Sync Service
|
|
* Syncs sites, devices, and alerts from Datto RMM API to PostgreSQL
|
|
*/
|
|
|
|
import postgresClient from './postgres-client';
|
|
import { DattoRMMClient } from './datto-rmm-client';
|
|
import { DattoRMMSite, DattoRMMDevice, DattoRMMAlert } from '@/lib/types/datto-rmm';
|
|
|
|
export interface DattoRMMSyncResult {
|
|
syncId: string;
|
|
syncType: 'full' | 'incremental';
|
|
status: 'completed' | 'failed';
|
|
startedAt: Date;
|
|
completedAt: Date;
|
|
duration: number;
|
|
entities: DattoRMMEntitySyncResult[];
|
|
errors: string[];
|
|
}
|
|
|
|
export interface DattoRMMEntitySyncResult {
|
|
entity: string;
|
|
success: boolean;
|
|
recordsUpserted: number;
|
|
duration: number;
|
|
error?: string;
|
|
}
|
|
|
|
export class DattoRMMSyncService {
|
|
private client: DattoRMMClient;
|
|
private isSyncing = false;
|
|
|
|
constructor(client?: DattoRMMClient) {
|
|
if (client) {
|
|
this.client = client;
|
|
} else {
|
|
this.client = new DattoRMMClient({
|
|
apiUrl: process.env.DATTO_RMM_API_URL || 'https://concord-api.centrastage.net',
|
|
apiKey: process.env.DATTO_RMM_API_KEY || '',
|
|
apiSecret: process.env.DATTO_RMM_API_SECRET || '',
|
|
});
|
|
}
|
|
}
|
|
|
|
isSyncInProgress(): boolean {
|
|
return this.isSyncing;
|
|
}
|
|
|
|
async fullSync(triggeredBy = 'system'): Promise<DattoRMMSyncResult> {
|
|
return this.executeSync('full', triggeredBy);
|
|
}
|
|
|
|
async incrementalSync(triggeredBy = 'system'): Promise<DattoRMMSyncResult> {
|
|
return this.executeSync('incremental', triggeredBy);
|
|
}
|
|
|
|
private async executeSync(syncType: 'full' | 'incremental', triggeredBy: string): Promise<DattoRMMSyncResult> {
|
|
if (this.isSyncing) {
|
|
throw new Error('A Datto RMM sync is already in progress');
|
|
}
|
|
|
|
this.isSyncing = true;
|
|
const syncId = `datto-rmm-${Date.now()}`;
|
|
const startTime = new Date();
|
|
const entityResults: DattoRMMEntitySyncResult[] = [];
|
|
const errors: string[] = [];
|
|
|
|
// Create sync history record
|
|
let historyId: number | null = null;
|
|
try {
|
|
const histResult = await postgresClient.query<{ id: number }>(
|
|
`INSERT INTO sync_history (entity_type, sync_type, status, started_at, records_added, records_updated, records_deleted, triggered_by)
|
|
VALUES ($1, $2, $3, $4, 0, 0, 0, $5) RETURNING id`,
|
|
['datto_rmm', syncType, 'started', startTime, triggeredBy]
|
|
);
|
|
historyId = histResult.rows[0].id;
|
|
} catch (e) {
|
|
console.warn('[DATTO-RMM-SYNC] Could not create sync history record:', e);
|
|
}
|
|
|
|
console.log(`[DATTO-RMM-SYNC] Starting ${syncType} sync (${syncId})`);
|
|
|
|
try {
|
|
const steps: Array<{ name: string; fn: () => Promise<number> }> = [
|
|
{ name: 'sites', fn: () => this.syncSites() },
|
|
{ name: 'devices', fn: () => this.syncDevices() },
|
|
{ name: 'open_alerts', fn: () => this.syncOpenAlerts() },
|
|
{ name: 'resolved_alerts', fn: () => this.syncResolvedAlerts() },
|
|
];
|
|
|
|
for (const step of steps) {
|
|
const stepStart = Date.now();
|
|
try {
|
|
const count = await step.fn();
|
|
const duration = Date.now() - stepStart;
|
|
entityResults.push({ entity: step.name, success: true, recordsUpserted: count, duration });
|
|
console.log(`[DATTO-RMM-SYNC] ${step.name}: ${count} records in ${duration}ms`);
|
|
} catch (error) {
|
|
const duration = Date.now() - stepStart;
|
|
const msg = error instanceof Error ? error.message : String(error);
|
|
errors.push(`${step.name}: ${msg}`);
|
|
entityResults.push({ entity: step.name, success: false, recordsUpserted: 0, duration, error: msg });
|
|
console.error(`[DATTO-RMM-SYNC] ${step.name} failed:`, msg);
|
|
}
|
|
}
|
|
|
|
const completedAt = new Date();
|
|
const duration = completedAt.getTime() - startTime.getTime();
|
|
const status = errors.length === 0 ? 'completed' : 'failed';
|
|
const totalRecords = entityResults.reduce((sum, r) => sum + r.recordsUpserted, 0);
|
|
|
|
console.log(`[DATTO-RMM-SYNC] Sync ${status} in ${duration}ms — ${totalRecords} total records`);
|
|
|
|
if (historyId) {
|
|
try {
|
|
await postgresClient.query(
|
|
`UPDATE sync_history SET status = $1, completed_at = $2, records_added = $3, error_message = $4, entity_details = $5 WHERE id = $6`,
|
|
[status, completedAt, totalRecords, errors.length > 0 ? errors.join('; ') : null, JSON.stringify(entityResults), historyId]
|
|
);
|
|
} catch (e) {
|
|
console.warn('[DATTO-RMM-SYNC] Could not update sync history:', e);
|
|
}
|
|
}
|
|
|
|
return { syncId, syncType, status, startedAt: startTime, completedAt, duration, entities: entityResults, errors };
|
|
} catch (error) {
|
|
const completedAt = new Date();
|
|
const msg = error instanceof Error ? error.message : String(error);
|
|
console.error('[DATTO-RMM-SYNC] Sync failed catastrophically:', msg);
|
|
|
|
if (historyId) {
|
|
try {
|
|
await postgresClient.query(
|
|
`UPDATE sync_history SET status = 'failed', completed_at = $1, error_message = $2 WHERE id = $3`,
|
|
[completedAt, msg, historyId]
|
|
);
|
|
} catch (e) { /* ignore */ }
|
|
}
|
|
|
|
return {
|
|
syncId, syncType, status: 'failed', startedAt: startTime, completedAt,
|
|
duration: completedAt.getTime() - startTime.getTime(), entities: entityResults, errors: [msg],
|
|
};
|
|
} finally {
|
|
this.isSyncing = false;
|
|
}
|
|
}
|
|
|
|
// ── Sites ─────────────────────────────────────────────────────────────────────
|
|
private async syncSites(): Promise<number> {
|
|
const sites = await this.client.getAllSites();
|
|
if (sites.length === 0) return 0;
|
|
|
|
// Build set of known company IDs for FK safety
|
|
const knownCompanies = await postgresClient.query('SELECT id FROM companies');
|
|
const companyIds = new Set(knownCompanies.rows.map((r: any) => r.id));
|
|
|
|
let count = 0;
|
|
for (const s of sites) {
|
|
const atCompanyId = s.autotaskCompanyId ? parseInt(s.autotaskCompanyId, 10) : null;
|
|
const matchedCompanyId = atCompanyId && !isNaN(atCompanyId) && atCompanyId > 0 && companyIds.has(atCompanyId)
|
|
? atCompanyId : null;
|
|
|
|
await postgresClient.query(
|
|
`INSERT INTO datto_rmm_sites (id, uid, account_uid, name, description, notes, on_demand,
|
|
autotask_company_id, autotask_company_name,
|
|
number_of_devices, number_of_online_devices, number_of_offline_devices,
|
|
portal_url, synced_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NOW())
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
uid=EXCLUDED.uid, account_uid=EXCLUDED.account_uid, name=EXCLUDED.name,
|
|
description=EXCLUDED.description, notes=EXCLUDED.notes, on_demand=EXCLUDED.on_demand,
|
|
autotask_company_id=EXCLUDED.autotask_company_id, autotask_company_name=EXCLUDED.autotask_company_name,
|
|
number_of_devices=EXCLUDED.number_of_devices, number_of_online_devices=EXCLUDED.number_of_online_devices,
|
|
number_of_offline_devices=EXCLUDED.number_of_offline_devices,
|
|
portal_url=EXCLUDED.portal_url, synced_at=NOW(), updated_at=NOW()`,
|
|
[
|
|
s.id, s.uid, s.accountUid || null, s.name, s.description || null, s.notes || null, s.onDemand,
|
|
matchedCompanyId, s.autotaskCompanyName || null,
|
|
s.devicesStatus?.numberOfDevices ?? 0,
|
|
s.devicesStatus?.numberOfOnlineDevices ?? 0,
|
|
s.devicesStatus?.numberOfOfflineDevices ?? 0,
|
|
s.portalUrl || null,
|
|
]
|
|
);
|
|
count++;
|
|
}
|
|
return count;
|
|
}
|
|
|
|
// ── Devices ───────────────────────────────────────────────────────────────────
|
|
private async syncDevices(): Promise<number> {
|
|
const devices = await this.client.getAllDevices();
|
|
if (devices.length === 0) return 0;
|
|
|
|
// Build set of known site IDs for FK safety
|
|
const knownSites = await postgresClient.query('SELECT id FROM datto_rmm_sites');
|
|
const siteIds = new Set(knownSites.rows.map((r: any) => r.id));
|
|
|
|
let count = 0;
|
|
for (const d of devices) {
|
|
const siteId = siteIds.has(d.siteId) ? d.siteId : null;
|
|
|
|
await postgresClient.query(
|
|
`INSERT INTO datto_rmm_devices (id, uid, site_id, site_uid, site_name, hostname, description,
|
|
device_type_category, device_type, operating_system, domain,
|
|
int_ip_address, ext_ip_address, last_logged_in_user,
|
|
last_seen, last_reboot, last_audit_date, creation_date,
|
|
online, suspended, deleted, reboot_required, a64_bit,
|
|
cag_version, display_version,
|
|
antivirus_product, antivirus_status,
|
|
patch_status, patches_approved_pending, patches_not_approved, patches_installed,
|
|
software_status, portal_url, web_remote_url, warranty_date,
|
|
snmp_enabled, device_class, network_probe, udf, synced_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35,$36,$37,$38,$39,NOW())
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
uid=EXCLUDED.uid, site_id=EXCLUDED.site_id, site_uid=EXCLUDED.site_uid, site_name=EXCLUDED.site_name,
|
|
hostname=EXCLUDED.hostname, description=EXCLUDED.description,
|
|
device_type_category=EXCLUDED.device_type_category, device_type=EXCLUDED.device_type,
|
|
operating_system=EXCLUDED.operating_system, domain=EXCLUDED.domain,
|
|
int_ip_address=EXCLUDED.int_ip_address, ext_ip_address=EXCLUDED.ext_ip_address,
|
|
last_logged_in_user=EXCLUDED.last_logged_in_user,
|
|
last_seen=EXCLUDED.last_seen, last_reboot=EXCLUDED.last_reboot,
|
|
last_audit_date=EXCLUDED.last_audit_date, creation_date=EXCLUDED.creation_date,
|
|
online=EXCLUDED.online, suspended=EXCLUDED.suspended, deleted=EXCLUDED.deleted,
|
|
reboot_required=EXCLUDED.reboot_required, a64_bit=EXCLUDED.a64_bit,
|
|
cag_version=EXCLUDED.cag_version, display_version=EXCLUDED.display_version,
|
|
antivirus_product=EXCLUDED.antivirus_product, antivirus_status=EXCLUDED.antivirus_status,
|
|
patch_status=EXCLUDED.patch_status, patches_approved_pending=EXCLUDED.patches_approved_pending,
|
|
patches_not_approved=EXCLUDED.patches_not_approved, patches_installed=EXCLUDED.patches_installed,
|
|
software_status=EXCLUDED.software_status, portal_url=EXCLUDED.portal_url,
|
|
web_remote_url=EXCLUDED.web_remote_url, warranty_date=EXCLUDED.warranty_date,
|
|
snmp_enabled=EXCLUDED.snmp_enabled, device_class=EXCLUDED.device_class,
|
|
network_probe=EXCLUDED.network_probe, udf=EXCLUDED.udf,
|
|
synced_at=NOW(), updated_at=NOW()`,
|
|
[
|
|
d.id, d.uid, siteId, d.siteUid, d.siteName, d.hostname, d.description || null,
|
|
d.deviceType?.category || null, d.deviceType?.type || null,
|
|
d.operatingSystem || null, d.domain || null,
|
|
d.intIpAddress || null, d.extIpAddress || null, d.lastLoggedInUser || null,
|
|
d.lastSeen ? new Date(d.lastSeen) : null,
|
|
d.lastReboot ? new Date(d.lastReboot) : null,
|
|
d.lastAuditDate ? new Date(d.lastAuditDate) : null,
|
|
d.creationDate ? new Date(d.creationDate) : null,
|
|
d.online, d.suspended, d.deleted, d.rebootRequired ?? false, d.a64Bit ?? true,
|
|
d.cagVersion || null, d.displayVersion || null,
|
|
d.antivirus?.antivirusProduct || null, d.antivirus?.antivirusStatus || null,
|
|
d.patchManagement?.patchStatus || null,
|
|
d.patchManagement?.patchesApprovedPending ?? 0,
|
|
d.patchManagement?.patchesNotApproved ?? 0,
|
|
d.patchManagement?.patchesInstalled ?? 0,
|
|
d.softwareStatus || null, d.portalUrl || null, d.webRemoteUrl || null,
|
|
d.warrantyDate ? new Date(d.warrantyDate) : null,
|
|
d.snmpEnabled ?? false, d.deviceClass || null, (d as any).networkProbe ?? false,
|
|
d.udf ? JSON.stringify(d.udf) : null,
|
|
]
|
|
);
|
|
count++;
|
|
}
|
|
return count;
|
|
}
|
|
|
|
// ── Alerts ────────────────────────────────────────────────────────────────────
|
|
private async upsertAlerts(alerts: DattoRMMAlert[]): Promise<number> {
|
|
let count = 0;
|
|
for (const a of alerts) {
|
|
await postgresClient.query(
|
|
`INSERT INTO datto_rmm_alerts (alert_uid, device_uid, device_name, site_uid, site_name,
|
|
priority, alert_context, alert_monitor_info, diagnostics,
|
|
resolved, resolved_by, resolved_on, muted, ticket_number,
|
|
autoresolve_mins, response_actions, timestamp, synced_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,NOW())
|
|
ON CONFLICT (alert_uid) DO UPDATE SET
|
|
device_uid=EXCLUDED.device_uid, device_name=EXCLUDED.device_name,
|
|
site_uid=EXCLUDED.site_uid, site_name=EXCLUDED.site_name,
|
|
priority=EXCLUDED.priority, alert_context=EXCLUDED.alert_context,
|
|
alert_monitor_info=EXCLUDED.alert_monitor_info, diagnostics=EXCLUDED.diagnostics,
|
|
resolved=EXCLUDED.resolved, resolved_by=EXCLUDED.resolved_by,
|
|
resolved_on=EXCLUDED.resolved_on, muted=EXCLUDED.muted,
|
|
ticket_number=EXCLUDED.ticket_number, autoresolve_mins=EXCLUDED.autoresolve_mins,
|
|
response_actions=EXCLUDED.response_actions,
|
|
synced_at=NOW(), updated_at=NOW()`,
|
|
[
|
|
a.alertUid,
|
|
a.alertSourceInfo?.deviceUid || null,
|
|
a.alertSourceInfo?.deviceName || null,
|
|
a.alertSourceInfo?.siteUid || null,
|
|
a.alertSourceInfo?.siteName || null,
|
|
a.priority || null,
|
|
a.alertContext ? JSON.stringify(a.alertContext) : null,
|
|
a.alertMonitorInfo ? JSON.stringify(a.alertMonitorInfo) : null,
|
|
a.diagnostics || null,
|
|
a.resolved,
|
|
a.resolvedBy || null,
|
|
a.resolvedOn ? new Date(a.resolvedOn) : null,
|
|
a.muted,
|
|
a.ticketNumber || null,
|
|
a.autoresolveMins ?? null,
|
|
a.responseActions ? JSON.stringify(a.responseActions) : null,
|
|
new Date(a.timestamp),
|
|
]
|
|
);
|
|
count++;
|
|
}
|
|
return count;
|
|
}
|
|
|
|
private async syncOpenAlerts(): Promise<number> {
|
|
const alerts = await this.client.getAllOpenAlerts();
|
|
if (alerts.length === 0) return 0;
|
|
return this.upsertAlerts(alerts);
|
|
}
|
|
|
|
private async syncResolvedAlerts(): Promise<number> {
|
|
const alerts = await this.client.getRecentResolvedAlerts(4);
|
|
if (alerts.length === 0) return 0;
|
|
return this.upsertAlerts(alerts);
|
|
}
|
|
}
|