wulf-pulse/lib/services/veeam-sync-service.ts

503 lines
24 KiB
TypeScript

/**
* Veeam Sync Service
* Orchestrates syncing VSPC data to PostgreSQL
*/
import postgresClient from './postgres-client';
import { VeeamClient } from './veeam-client';
import { getVeeamClient } from './veeam-factory';
import {
VspcOrganization,
VspcBackupServer,
VspcBackupJob,
VspcBackupAgentJob,
VspcProtectedWorkload,
VspcRepository,
VspcBackupAgent,
VspcAlarm,
} from '@/lib/types/veeam';
import { VeeamComplianceService } from './veeam-compliance-service';
export interface VeeamSyncResult {
syncId: string;
syncType: 'full' | 'incremental';
status: 'completed' | 'failed';
startedAt: Date;
completedAt: Date;
duration: number;
entities: VeeamEntitySyncResult[];
errors: string[];
}
export interface VeeamEntitySyncResult {
entity: string;
success: boolean;
recordsUpserted: number;
duration: number;
error?: string;
}
export class VeeamSyncService {
private client: VeeamClient;
private isSyncing = false;
constructor(client?: VeeamClient) {
this.client = client || getVeeamClient();
}
isSyncInProgress(): boolean {
return this.isSyncing;
}
/**
* Full sync — fetches all entities and upserts into PostgreSQL
*/
async fullSync(triggeredBy: string = 'system'): Promise<VeeamSyncResult> {
return this.executeSync('full', triggeredBy);
}
/**
* Incremental sync — same as full for now since VSPC API doesn't support
* filtering by last-modified. We upsert all records so unchanged rows are no-ops.
*/
async incrementalSync(triggeredBy: string = 'system'): Promise<VeeamSyncResult> {
return this.executeSync('incremental', triggeredBy);
}
private async executeSync(syncType: 'full' | 'incremental', triggeredBy: string): Promise<VeeamSyncResult> {
if (this.isSyncing) {
throw new Error('A Veeam sync operation is already in progress');
}
this.isSyncing = true;
const syncId = `veeam-${Date.now()}`;
const startTime = new Date();
const entityResults: VeeamEntitySyncResult[] = [];
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`,
['veeam', syncType, 'started', startTime, triggeredBy]
);
historyId = histResult.rows[0].id;
} catch (e) {
console.warn('[VEEAM-SYNC] Could not create sync history record:', e);
}
console.log(`[VEEAM-SYNC] Starting ${syncType} sync (${syncId})`);
try {
// Sync in dependency order
const steps: Array<{ name: string; fn: () => Promise<number> }> = [
{ name: 'organizations', fn: () => this.syncOrganizations() },
{ name: 'backup_servers', fn: () => this.syncBackupServers() },
{ name: 'repositories', fn: () => this.syncRepositories() },
{ name: 'backup_jobs', fn: () => this.syncBackupJobs() },
{ name: 'backup_agent_jobs', fn: () => this.syncBackupAgentJobs() },
{ name: 'protected_workloads', fn: () => this.syncProtectedWorkloads() },
{ name: 'backup_agents', fn: () => this.syncBackupAgents() },
{ name: 'alarms', fn: () => this.syncAlarms() },
];
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(`[VEEAM-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(`[VEEAM-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(`[VEEAM-SYNC] Sync ${status} in ${duration}ms — ${totalRecords} total records`);
// Run compliance computation after successful sync
if (status === 'completed') {
try {
const complianceService = new VeeamComplianceService();
await complianceService.computeCompliance();
} catch (compError) {
console.error('[VEEAM-SYNC] Compliance computation failed:', compError);
}
}
// Update sync history
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('[VEEAM-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('[VEEAM-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;
}
}
// ============================================================================
// Entity Sync Methods
// ============================================================================
private async syncOrganizations(): Promise<number> {
const orgs = await this.client.getOrganizations();
if (orgs.length === 0) return 0;
let count = 0;
for (const org of orgs) {
const parsed = org.companyId ? parseInt(org.companyId, 10) : null;
const companyId = parsed && !isNaN(parsed) ? parsed : null;
// Verify company exists in Autotask if companyId is set
let matchedCompanyId = companyId;
if (companyId) {
const check = await postgresClient.query(
'SELECT id FROM companies WHERE id = $1', [companyId]
);
if (check.rows.length === 0) {
matchedCompanyId = null;
}
}
await postgresClient.query(
`INSERT INTO veeam_organizations (instance_uid, name, type, company_id, tax_id, email, phone, country, state, city, street, zip_code, website, notes, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
name=EXCLUDED.name, type=EXCLUDED.type, company_id=EXCLUDED.company_id,
tax_id=EXCLUDED.tax_id, email=EXCLUDED.email, phone=EXCLUDED.phone,
country=EXCLUDED.country, state=EXCLUDED.state, city=EXCLUDED.city,
street=EXCLUDED.street, zip_code=EXCLUDED.zip_code, website=EXCLUDED.website,
notes=EXCLUDED.notes, synced_at=NOW(), updated_at=NOW()`,
[
org.instanceUid, org.name, org.type, matchedCompanyId,
org.taxId || null, org.email || null, org.phone || null,
org.countryName || null, org.regionName || null, org.city || null,
org.street || null, org.zipCode || null, org.website || null, org.notes || null,
]
);
count++;
}
return count;
}
private async syncBackupServers(): Promise<number> {
const servers = await this.client.getBackupServers();
if (servers.length === 0) return 0;
// Build set of known org UIDs for FK safety
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const s of servers) {
const orgUid = orgUids.has(s.organizationUid) ? s.organizationUid : null;
await postgresClient.query(
`INSERT INTO veeam_backup_servers (instance_uid, name, organization_uid, version, display_version, status, backup_server_role_type, installation_uid, location_uid, management_agent_uid, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
name=EXCLUDED.name, organization_uid=EXCLUDED.organization_uid,
version=EXCLUDED.version, display_version=EXCLUDED.display_version,
status=EXCLUDED.status, backup_server_role_type=EXCLUDED.backup_server_role_type,
installation_uid=EXCLUDED.installation_uid, location_uid=EXCLUDED.location_uid,
management_agent_uid=EXCLUDED.management_agent_uid, synced_at=NOW(), updated_at=NOW()`,
[
s.instanceUid, s.name, orgUid, s.version, s.displayVersion,
s.status, s.backupServerRoleType, s.installationUid, s.locationUid, s.managementAgentUid,
]
);
count++;
}
return count;
}
private async syncBackupJobs(): Promise<number> {
const jobs = await this.client.getBackupJobs();
if (jobs.length === 0) return 0;
// Build sets for FK safety
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
const knownServers = await postgresClient.query('SELECT instance_uid FROM veeam_backup_servers');
const serverUids = new Set(knownServers.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const j of jobs) {
const orgUid = orgUids.has(j.organizationUid) ? j.organizationUid : null;
const serverUid = serverUids.has(j.backupServerUid) ? j.backupServerUid : null;
await postgresClient.query(
`INSERT INTO veeam_backup_jobs (instance_uid, unique_uid, name, description, backup_server_uid, organization_uid, status, type, last_run, last_end_time, last_duration, processing_rate, avg_duration, transferred_data, backup_chain_size, bottleneck, is_enabled, schedule_type, failure_message, target_type, destination, retention_limit, retention_limit_type, is_gfs_option_enabled, 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,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
unique_uid=EXCLUDED.unique_uid, name=EXCLUDED.name, description=EXCLUDED.description,
backup_server_uid=EXCLUDED.backup_server_uid, organization_uid=EXCLUDED.organization_uid,
status=EXCLUDED.status, type=EXCLUDED.type, last_run=EXCLUDED.last_run,
last_end_time=EXCLUDED.last_end_time, last_duration=EXCLUDED.last_duration,
processing_rate=EXCLUDED.processing_rate, avg_duration=EXCLUDED.avg_duration,
transferred_data=EXCLUDED.transferred_data, backup_chain_size=EXCLUDED.backup_chain_size,
bottleneck=EXCLUDED.bottleneck, is_enabled=EXCLUDED.is_enabled,
schedule_type=EXCLUDED.schedule_type, failure_message=EXCLUDED.failure_message,
target_type=EXCLUDED.target_type, destination=EXCLUDED.destination,
retention_limit=EXCLUDED.retention_limit, retention_limit_type=EXCLUDED.retention_limit_type,
is_gfs_option_enabled=EXCLUDED.is_gfs_option_enabled, synced_at=NOW(), updated_at=NOW()`,
[
j.instanceUid, j.uniqueUid, j.name, j.description || null,
serverUid, orgUid, j.status, j.type,
j.lastRun || null, j.lastEndTime || null, j.lastDuration, j.processingRate,
j.avgDuration, j.transferredData, j.backupChainSize, j.bottleneck,
j.isEnabled, j.scheduleType, j.failureMessage || null,
j.targetType, j.destination, j.retentionLimit, j.retentionLimitType,
j.isGfsOptionEnabled,
]
);
count++;
}
return count;
}
private async syncBackupAgentJobs(): Promise<number> {
const jobs = await this.client.getBackupAgentJobs();
if (jobs.length === 0) return 0;
// Build set for FK safety
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const j of jobs) {
const orgUid = orgUids.has(j.organizationUid) ? j.organizationUid : null;
await postgresClient.query(
`INSERT INTO veeam_backup_agent_jobs (instance_uid, original_uid, backup_agent_uid, organization_uid, name, description, config_uid, system_type, backup_policy_uid, backup_policy_name, backup_policy_assign_status, backup_policy_failure_message, status, operation_mode, destination, restore_points, last_run, last_end_time, last_duration, next_run, avg_duration, backup_mode, target_type, is_enabled, schedule_type, failure_message, backed_up_size, free_space, 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,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
original_uid=EXCLUDED.original_uid, backup_agent_uid=EXCLUDED.backup_agent_uid,
organization_uid=EXCLUDED.organization_uid, name=EXCLUDED.name, description=EXCLUDED.description,
config_uid=EXCLUDED.config_uid, system_type=EXCLUDED.system_type,
backup_policy_uid=EXCLUDED.backup_policy_uid, backup_policy_name=EXCLUDED.backup_policy_name,
backup_policy_assign_status=EXCLUDED.backup_policy_assign_status,
backup_policy_failure_message=EXCLUDED.backup_policy_failure_message,
status=EXCLUDED.status, operation_mode=EXCLUDED.operation_mode,
destination=EXCLUDED.destination, restore_points=EXCLUDED.restore_points,
last_run=EXCLUDED.last_run, last_end_time=EXCLUDED.last_end_time,
last_duration=EXCLUDED.last_duration, next_run=EXCLUDED.next_run,
avg_duration=EXCLUDED.avg_duration, backup_mode=EXCLUDED.backup_mode,
target_type=EXCLUDED.target_type, is_enabled=EXCLUDED.is_enabled,
schedule_type=EXCLUDED.schedule_type, failure_message=EXCLUDED.failure_message,
backed_up_size=EXCLUDED.backed_up_size, free_space=EXCLUDED.free_space,
synced_at=NOW(), updated_at=NOW()`,
[
j.instanceUid, j.originalUid, j.backupAgentUid, orgUid,
j.name || j.backupPolicyName || 'Unknown', j.description || null, j.configUid, j.systemType,
j.backupPolicyUid, j.backupPolicyName, j.backupPolicyAssignStatus,
j.backupPolicyFailureMessage || null, j.status, j.operationMode,
j.destination, j.restorePoints, j.lastRun || null, j.lastEndTime || null,
j.lastDuration, j.nextRun || null, j.avgDuration, j.backupMode,
j.targetType, j.isEnabled, j.scheduleType, j.failureMessage || null,
j.backedUpSize, j.freeSpace,
]
);
count++;
}
return count;
}
private async syncProtectedWorkloads(): Promise<number> {
const workloads = await this.client.getProtectedWorkloads();
if (workloads.length === 0) return 0;
// Build sets for FK safety
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
const knownServers = await postgresClient.query('SELECT instance_uid FROM veeam_backup_servers');
const serverUids = new Set(knownServers.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const w of workloads) {
const orgUid = orgUids.has(w.organizationUid) ? w.organizationUid : null;
const serverUid = serverUids.has(w.backupServerUid) ? w.backupServerUid : null;
await postgresClient.query(
`INSERT INTO veeam_protected_workloads (instance_uid, name, backup_server_uid, organization_uid, job_uid, ip_addresses, provisioned_source_size, used_source_size, total_restore_point_size, latest_restore_point_size, restore_points, latest_restore_point_date, malware_state, immutable, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
name=EXCLUDED.name, backup_server_uid=EXCLUDED.backup_server_uid,
organization_uid=EXCLUDED.organization_uid, job_uid=EXCLUDED.job_uid,
ip_addresses=EXCLUDED.ip_addresses, provisioned_source_size=EXCLUDED.provisioned_source_size,
used_source_size=EXCLUDED.used_source_size, total_restore_point_size=EXCLUDED.total_restore_point_size,
latest_restore_point_size=EXCLUDED.latest_restore_point_size, restore_points=EXCLUDED.restore_points,
latest_restore_point_date=EXCLUDED.latest_restore_point_date, malware_state=EXCLUDED.malware_state,
immutable=EXCLUDED.immutable, synced_at=NOW(), updated_at=NOW()`,
[
w.instanceUid, w.name, serverUid, orgUid,
w.jobUid, w.ipAddresses ? JSON.stringify(w.ipAddresses) : null,
w.provisionedSourceSize, w.usedSourceSize, w.totalRestorePointSize,
w.latestRestorePointSize, w.restorePoints, w.latestRestorePointDate || null,
w.malwareState, w.immutable,
]
);
count++;
}
return count;
}
private async syncRepositories(): Promise<number> {
const repos = await this.client.getRepositories();
if (repos.length === 0) return 0;
// Build set for FK safety
const knownServers = await postgresClient.query('SELECT instance_uid FROM veeam_backup_servers');
const serverUids = new Set(knownServers.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const r of repos) {
const serverUid = serverUids.has(r.backupServerUid) ? r.backupServerUid : null;
await postgresClient.query(
`INSERT INTO veeam_repositories (instance_uid, name, backup_server_uid, capacity_bytes, free_space_bytes, used_space_bytes, repository_type, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
name=EXCLUDED.name, backup_server_uid=EXCLUDED.backup_server_uid,
capacity_bytes=EXCLUDED.capacity_bytes, free_space_bytes=EXCLUDED.free_space_bytes,
used_space_bytes=EXCLUDED.used_space_bytes, repository_type=EXCLUDED.repository_type,
synced_at=NOW(), updated_at=NOW()`,
[
r.instanceUid, r.name, serverUid,
r.capacityBytes || null, r.freeSpaceBytes || null,
r.usedSpaceBytes || null, r.repositoryType || null,
]
);
count++;
}
return count;
}
private async syncBackupAgents(): Promise<number> {
const agents = await this.client.getBackupAgents();
if (agents.length === 0) return 0;
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const a of agents) {
const orgUid = orgUids.has(a.organizationUid) ? a.organizationUid : null;
await postgresClient.query(
`INSERT INTO veeam_backup_agents (instance_uid, organization_uid, site_uid, management_agent_uid,
name, agent_platform, status, management_agent_status, operation_mode, gui_mode,
platform, version, version_status, management_mode, installation_type, activation_time,
total_jobs_count, running_jobs_count, success_jobs_count, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
organization_uid=EXCLUDED.organization_uid, site_uid=EXCLUDED.site_uid,
management_agent_uid=EXCLUDED.management_agent_uid, name=EXCLUDED.name,
agent_platform=EXCLUDED.agent_platform, status=EXCLUDED.status,
management_agent_status=EXCLUDED.management_agent_status, operation_mode=EXCLUDED.operation_mode,
gui_mode=EXCLUDED.gui_mode, platform=EXCLUDED.platform, version=EXCLUDED.version,
version_status=EXCLUDED.version_status, management_mode=EXCLUDED.management_mode,
installation_type=EXCLUDED.installation_type, activation_time=EXCLUDED.activation_time,
total_jobs_count=EXCLUDED.total_jobs_count, running_jobs_count=EXCLUDED.running_jobs_count,
success_jobs_count=EXCLUDED.success_jobs_count, synced_at=NOW(), updated_at=NOW()`,
[
a.instanceUid, orgUid, a.siteUid || null, a.managementAgentUid || null,
a.name || a.instanceUid, a.agentPlatform || null, a.status || null, a.managementAgentStatus || null,
a.operationMode || null, a.guiMode || null, a.platform || null,
a.version || null, a.versionStatus || null, a.managementMode || null,
a.installationType || null, a.activationTime ? new Date(a.activationTime) : null,
a.totalJobsCount ?? 0, a.runningJobsCount ?? 0, a.successJobsCount ?? 0,
]
);
count++;
}
return count;
}
private async syncAlarms(): Promise<number> {
const alarms = await this.client.getActiveAlarms();
if (alarms.length === 0) return 0;
const knownOrgs = await postgresClient.query('SELECT instance_uid FROM veeam_organizations');
const orgUids = new Set(knownOrgs.rows.map((r: any) => r.instance_uid));
let count = 0;
for (const alarm of alarms) {
const orgUid = alarm.object?.organizationUid && orgUids.has(alarm.object.organizationUid)
? alarm.object.organizationUid : null;
const lastStatus = alarm.lastActivation?.status || null;
const resolved = lastStatus === 'Resolved';
await postgresClient.query(
`INSERT INTO veeam_alarms (instance_uid, alarm_template_uid, repeat_count,
object_uid, object_type, object_name, object_computer_name,
organization_uid, location_uid, management_agent_uid,
last_activation_uid, last_activation_time, last_activation_status,
last_activation_message, last_activation_remark, area, resolved, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,NOW())
ON CONFLICT (instance_uid) DO UPDATE SET
alarm_template_uid=EXCLUDED.alarm_template_uid, repeat_count=EXCLUDED.repeat_count,
object_uid=EXCLUDED.object_uid, object_type=EXCLUDED.object_type,
object_name=EXCLUDED.object_name, object_computer_name=EXCLUDED.object_computer_name,
organization_uid=EXCLUDED.organization_uid, location_uid=EXCLUDED.location_uid,
management_agent_uid=EXCLUDED.management_agent_uid,
last_activation_uid=EXCLUDED.last_activation_uid,
last_activation_time=EXCLUDED.last_activation_time,
last_activation_status=EXCLUDED.last_activation_status,
last_activation_message=EXCLUDED.last_activation_message,
last_activation_remark=EXCLUDED.last_activation_remark,
area=EXCLUDED.area, resolved=EXCLUDED.resolved,
synced_at=NOW(), updated_at=NOW()`,
[
alarm.instanceUid, alarm.alarmTemplateUid || null, alarm.repeatCount ?? 0,
alarm.object?.objectUid || alarm.object?.instanceUid || null,
alarm.object?.type || null,
alarm.object?.objectName || null,
alarm.object?.computerName || null,
orgUid,
alarm.object?.locationUid || null,
alarm.object?.managementAgentUid || null,
alarm.lastActivation?.instanceUid || null,
alarm.lastActivation?.time ? new Date(alarm.lastActivation.time) : null,
lastStatus,
alarm.lastActivation?.message || null,
alarm.lastActivation?.remark || null,
alarm.area || null,
resolved,
]
);
count++;
}
return count;
}
}