/** * 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, } 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 { 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 { return this.executeSync('incremental', triggeredBy); } private async executeSync(syncType: 'full' | 'incremental', triggeredBy: string): Promise { 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 }> = [ { 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() }, ]; 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 WHERE id = $5`, [status, completedAt, totalRecords, errors.length > 0 ? errors.join('; ') : null, 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 { 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 { 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 { 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 { 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 { 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 { 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; } }