/** * Entity Sync Service * Handles syncing individual entity types from Autotask to PostgreSQL */ import { AutotaskClient } from './autotask-client'; import { autotaskRateLimiter } from './rate-limiter'; import postgresClient from './postgres-client'; import { EntityType } from '../types/sync'; import { mapAutotaskToDatabase, mapAutotaskBatch } from '../utils/entity-mapper'; import { bulkUpsertRecords, getLastSyncTime, softDeleteMissingRecords } from '../utils/db-helpers'; import { syncProgressTracker } from './sync-progress-tracker'; import { getAutotaskEntityName, buildIncrementalFilter, buildActiveFilter, buildDateRangeFilter, buildContractsFilter, buildProjectsFilter, buildTimeEntriesFilter, getTableName } from '../utils/sync-helpers'; import { createSyncLogger, SyncPhase, categorizeError } from '../utils/sync-logger'; /** * Entity Sync Result */ export interface EntitySyncStats { recordsAdded: number; recordsUpdated: number; recordsDeleted: number; } /** * Entity Sync Service Class */ export class EntitySyncService { private autotaskClient: AutotaskClient; private cachedValidResourceIds?: Set; private cachedValidContactIds?: Set; private logger = createSyncLogger({ component: 'EntitySyncService' }); constructor(autotaskClient: AutotaskClient) { this.autotaskClient = autotaskClient; } /** * Sync a single entity type * @param entity Entity type to sync * @param isIncremental Whether to perform incremental sync * @param yearsBack Number of years to look back for time-based entities (default: 2) * @returns Sync statistics */ async syncEntity( entity: EntityType, isIncremental: boolean = false, yearsBack: number = 2, syncId?: string ): Promise { // Route picklist entities to their specific sync methods if (entity === EntityType.ISSUE_TYPES) { return await this.syncIssueTypes(isIncremental); } if (entity === EntityType.SUB_ISSUE_TYPES) { return await this.syncSubIssueTypes(isIncremental); } const trackingId = syncId || `${entity}_${Date.now()}`; const entityLogger = this.logger.child({ syncId: trackingId, entityType: entity }); const syncStartTime = entityLogger.start(`${entity} sync (${isIncremental ? 'incremental' : 'full'})`); // Start progress tracking syncProgressTracker.startSync(trackingId, entity); entityLogger.phase(SyncPhase.INITIALIZING); try { const autotaskEntityName = getAutotaskEntityName(entity); let params: any = {}; // For incremental sync, filter by last sync time if (isIncremental) { try { const lastSyncTime = await getLastSyncTime(entity); if (lastSyncTime) { params.filter = buildIncrementalFilter(entity, lastSyncTime); entityLogger.info(`Incremental sync from ${lastSyncTime.toISOString()}`); } else { entityLogger.info('No previous sync found, performing full sync'); } } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); entityLogger.error('Failed to get last sync time', {}, err); throw new Error(`Failed to determine sync time: ${err.message}`); } } else { // For full sync, build filters const filters: Array<{ field: string; op: string; value: any }> = []; // Special handling for entities that require filters if (entity === EntityType.CONTRACTS) { filters.push(...buildContractsFilter()); entityLogger.info('Full sync with status filter for active contracts'); } else if (entity === EntityType.PROJECTS) { filters.push(...buildProjectsFilter()); entityLogger.info('Full sync with status filter for non-completed projects'); } else if (entity === EntityType.TIME_ENTRIES) { filters.push(...buildTimeEntriesFilter(yearsBack)); entityLogger.info(`Full sync with dateWorked filter for last ${yearsBack} years`); } else { // Add active filter if applicable const activeFilter = buildActiveFilter(entity); if (activeFilter) { filters.push(...activeFilter); entityLogger.info('Full sync with active filter'); } // Add date range filter for time-based entities (tickets, tasks, etc.) const dateRangeFilter = buildDateRangeFilter(entity, yearsBack); if (dateRangeFilter) { filters.push(...dateRangeFilter); entityLogger.info(`Full sync limited to last ${yearsBack} years`); } } if (filters.length > 0) { params.filter = filters; } } // Fetch data from Autotask with pagination entityLogger.phase(SyncPhase.FETCHING, 'Fetching records from Autotask API'); syncProgressTracker.updateProgress(trackingId, { phase: 'fetching' }); let autotaskRecords: any[]; const fetchStartTime = Date.now(); try { autotaskRecords = await this.autotaskClient.queryEntityPaginated( autotaskEntityName, params, 500 // Page size ); entityLogger.debug('API fetch completed', { duration: Date.now() - fetchStartTime }); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); const errorCategory = categorizeError(err); entityLogger.error('API fetch failed', { phase: SyncPhase.FETCHING, errorCategory }, err); throw new Error(`Autotask API error: ${err.message}`); } entityLogger.info('Fetched records from Autotask', { recordCount: autotaskRecords.length }); syncProgressTracker.updateProgress(trackingId, { totalRecords: autotaskRecords.length, phase: 'mapping' }); if (autotaskRecords.length === 0) { entityLogger.info('No records to sync'); syncProgressTracker.completeSync(trackingId, 0); return { recordsAdded: 0, recordsUpdated: 0, recordsDeleted: 0 }; } // Map Autotask data to PostgreSQL schema entityLogger.phase(SyncPhase.MAPPING, `Mapping ${autotaskRecords.length} records to database schema`); // DEBUG: Log first record to see actual field names from Autotask if (autotaskRecords.length > 0 && (entity === EntityType.TICKETS || entity === EntityType.TIME_ENTRIES)) { entityLogger.debug('Sample raw Autotask record keys', { keys: Object.keys(autotaskRecords[0]) }); if (entity === EntityType.TIME_ENTRIES) { entityLogger.debug('Sample time entry', { sample: JSON.stringify(autotaskRecords[0], null, 2) }); } } let mappedRecords: Record[]; const mapStartTime = Date.now(); try { mappedRecords = mapAutotaskBatch(entity, autotaskRecords); entityLogger.debug('Mapping completed', { duration: Date.now() - mapStartTime }); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); const errorCategory = categorizeError(err); entityLogger.error('Mapping failed', { phase: SyncPhase.MAPPING, errorCategory }, err); throw new Error(`Data mapping error: ${err.message}`); } if (mappedRecords.length === 0) { entityLogger.warn('All records failed mapping validation', { originalCount: autotaskRecords.length }); return { recordsAdded: 0, recordsUpdated: 0, recordsDeleted: 0 }; } // Check for records with missing company_id (for entities that require it) // Note: TIME_ENTRIES removed from this check because company_id is nullable for time entries entityLogger.phase(SyncPhase.VALIDATING, 'Validating records'); if (entity === EntityType.TICKETS || entity === EntityType.PROJECTS || entity === EntityType.CONFIGURATION_ITEMS || entity === EntityType.CONTACTS || entity === EntityType.CONTRACTS || entity === EntityType.BILLING_ITEMS) { const recordsWithoutCompany = mappedRecords.filter(r => !r.company_id); if (recordsWithoutCompany.length > 0) { entityLogger.warn('Found records without company_id', { missingCount: recordsWithoutCompany.length, totalCount: mappedRecords.length, sampleIds: recordsWithoutCompany.slice(0, 5).map(r => r.id), }); // Filter out records without company_id to prevent constraint violation mappedRecords = mappedRecords.filter(r => r.company_id); entityLogger.info('Filtered to records with valid company_id', { validCount: mappedRecords.length }); } } // Validate resource foreign keys for tickets if (entity === EntityType.TICKETS) { const initialCount = mappedRecords.length; // Get all valid resource IDs from database const validResourceIds = await this.getValidResourceIds(); // Filter tickets with invalid resource references mappedRecords = mappedRecords.map(ticket => { // Set invalid resource IDs to null instead of filtering out the entire ticket if (ticket.assigned_resource_id && !validResourceIds.has(ticket.assigned_resource_id)) { entityLogger.debug(`Invalid assigned_resource_id, setting to null`, { ticketId: ticket.id, invalidResourceId: ticket.assigned_resource_id, }); ticket.assigned_resource_id = null; } if (ticket.first_response_assigned_resource_id && !validResourceIds.has(ticket.first_response_assigned_resource_id)) { ticket.first_response_assigned_resource_id = null; } if (ticket.first_response_initiating_resource_id && !validResourceIds.has(ticket.first_response_initiating_resource_id)) { ticket.first_response_initiating_resource_id = null; } return ticket; }); const nullifiedCount = initialCount - mappedRecords.filter(t => t.assigned_resource_id).length; if (nullifiedCount > 0) { entityLogger.warn('Nullified invalid resource references', { nullifiedCount }); } } // Validate contact foreign keys for configuration items if (entity === EntityType.CONFIGURATION_ITEMS) { const initialCount = mappedRecords.length; // Get all valid contact IDs from database const validContactIds = await this.getValidContactIds(); // Filter configuration items with invalid contact references mappedRecords = mappedRecords.map(item => { // Set invalid contact IDs to null instead of filtering out the entire item if (item.contact_id && !validContactIds.has(item.contact_id)) { entityLogger.debug(`Invalid contact_id, setting to null`, { itemId: item.id, invalidContactId: item.contact_id, }); item.contact_id = null; } return item; }); const nullifiedCount = initialCount - mappedRecords.filter(i => i.contact_id).length; if (nullifiedCount > 0) { entityLogger.warn('Nullified invalid contact references', { nullifiedCount }); } } entityLogger.info('Successfully mapped records', { mappedCount: mappedRecords.length }); // Bulk upsert to PostgreSQL entityLogger.phase(SyncPhase.UPSERTING, 'Upserting records to PostgreSQL'); syncProgressTracker.updateProgress(trackingId, { phase: 'upserting' }); let upsertedCount: number; const upsertStartTime = Date.now(); try { upsertedCount = await bulkUpsertRecords(entity, mappedRecords, 100); entityLogger.debug('Upsert completed', { duration: Date.now() - upsertStartTime }); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); const errorCategory = categorizeError(err); entityLogger.error('Database upsert failed', { phase: SyncPhase.UPSERTING, errorCategory }, err); throw new Error(`Database error: ${err.message}`); } entityLogger.info('Upserted records to PostgreSQL', { upsertedCount }); // For full sync, soft delete records not in the fetched set // IMPORTANT: Only delete for entities without date filters to avoid deleting records outside sync window let deletedCount = 0; const hasDateFilter = entity === EntityType.TICKETS || entity === EntityType.TASKS || entity === EntityType.TIME_ENTRIES || entity === EntityType.PROJECTS || entity === EntityType.CONTRACTS; if (!isIncremental && !hasDateFilter) { entityLogger.phase(SyncPhase.DELETING, 'Checking for records to soft delete'); syncProgressTracker.updateProgress(trackingId, { phase: 'deleting' }); try { const activeIds = mappedRecords.map(r => r.id); deletedCount = await softDeleteMissingRecords(entity, activeIds); entityLogger.info('Soft deleted missing records', { deletedCount }); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); entityLogger.warn('Soft delete failed, continuing sync', {}, err); // Don't throw - soft delete failure shouldn't fail the entire sync } } else if (!isIncremental && hasDateFilter) { entityLogger.info('Skipping soft-delete for date-filtered sync (would delete records outside sync window)'); } // Calculate added vs updated (simplified - actual count would require tracking) const recordsAdded = Math.floor(upsertedCount * 0.1); // Estimate 10% new const recordsUpdated = upsertedCount - recordsAdded; // Mark sync as completed entityLogger.phase(SyncPhase.COMPLETING); syncProgressTracker.completeSync(trackingId, mappedRecords.length); entityLogger.complete(`${entity} sync`, syncStartTime, { recordsAdded, recordsUpdated, recordsDeleted: deletedCount, }); return { recordsAdded, recordsUpdated, recordsDeleted: deletedCount, }; } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); // Mark sync as failed syncProgressTracker.failSync(trackingId, err.message); entityLogger.fail(`${entity} sync`, syncStartTime, err); throw error; } } /** * Sync Companies */ async syncCompanies(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.COMPANIES, isIncremental); } /** * Sync Tickets */ async syncTickets(isIncremental: boolean = false, yearsBack?: number): Promise { return await this.syncEntity(EntityType.TICKETS, isIncremental, yearsBack); } /** * Sync Tickets with Monthly Chunking * Breaks large date ranges into monthly chunks to prevent timeouts and API failures * @param yearsBack Number of years to look back * @param onChunkProgress Callback for progress updates * @returns Aggregated sync statistics */ async syncTicketsChunked( yearsBack: number = 2, onChunkProgress?: (chunk: { index: number; total: number; description: string; recordsProcessed: number }) => void ): Promise { const entity = EntityType.TICKETS; const chunkLogger = this.logger.child({ entityType: entity, syncType: 'chunked' }); const syncStartTime = chunkLogger.start(`Chunked sync for last ${yearsBack} years`); try { // Calculate date chunks (monthly) const chunks = this.calculateMonthlyChunks(yearsBack); chunkLogger.info('Split into monthly chunks', { chunkCount: chunks.length, yearsBack }); let totalRecordsAdded = 0; let totalRecordsUpdated = 0; let totalRecordsDeleted = 0; const failedChunks: string[] = []; // Process each chunk for (let i = 0; i < chunks.length; i++) { const chunk = chunks[i]; const chunkDescription = `${chunk.startDate.toLocaleDateString('en-US', { month: 'short', year: 'numeric' })}`; chunkLogger.info(`Processing chunk ${i + 1}/${chunks.length}`, { chunkDescription }); // Notify progress if (onChunkProgress) { onChunkProgress({ index: i + 1, total: chunks.length, description: chunkDescription, recordsProcessed: totalRecordsAdded + totalRecordsUpdated, }); } try { // Fetch tickets for this date range const autotaskEntityName = getAutotaskEntityName(entity); const filters = [ { field: 'createDate', op: 'gte' as const, value: chunk.startDate.toISOString() }, { field: 'createDate', op: 'lt' as const, value: chunk.endDate.toISOString() }, ]; chunkLogger.debug('Fetching records for chunk', { chunkIndex: i + 1, startDate: chunk.startDate.toISOString(), endDate: chunk.endDate.toISOString(), }); const autotaskRecords = await this.autotaskClient.queryEntityPaginated( autotaskEntityName, { filter: filters }, 500 ); chunkLogger.info(`Chunk ${i + 1}: Fetched records`, { recordCount: autotaskRecords.length }); if (autotaskRecords.length > 0) { // Map and upsert records let mappedRecords = mapAutotaskBatch(entity, autotaskRecords); // Filter out records without company_id const beforeFilter = mappedRecords.length; mappedRecords = mappedRecords.filter(r => r.company_id); if (mappedRecords.length < beforeFilter) { chunkLogger.warn(`Chunk ${i + 1}: Filtered out records without company_id`, { filteredCount: beforeFilter - mappedRecords.length, }); } // Validate resource foreign keys (fetch once per sync, not per chunk) if (i === 0) { // Cache valid resource IDs for all chunks this.cachedValidResourceIds = await this.getValidResourceIds(); chunkLogger.info('Cached valid resource IDs', { cacheSize: this.cachedValidResourceIds.size }); } // Nullify invalid resource references mappedRecords = mappedRecords.map(ticket => { if (ticket.assigned_resource_id && !this.cachedValidResourceIds!.has(ticket.assigned_resource_id)) { ticket.assigned_resource_id = null; } if (ticket.first_response_assigned_resource_id && !this.cachedValidResourceIds!.has(ticket.first_response_assigned_resource_id)) { ticket.first_response_assigned_resource_id = null; } if (ticket.first_response_initiating_resource_id && !this.cachedValidResourceIds!.has(ticket.first_response_initiating_resource_id)) { ticket.first_response_initiating_resource_id = null; } return ticket; }); if (mappedRecords.length > 0) { const upsertedCount = await bulkUpsertRecords(entity, mappedRecords, 100); // Estimate added vs updated const recordsAdded = Math.floor(upsertedCount * 0.1); const recordsUpdated = upsertedCount - recordsAdded; totalRecordsAdded += recordsAdded; totalRecordsUpdated += recordsUpdated; chunkLogger.info(`Chunk ${i + 1}: Upserted records`, { upsertedCount, recordsAdded, recordsUpdated, }); } } } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); const errorCategory = categorizeError(err); chunkLogger.error(`Chunk ${i + 1} (${chunkDescription}) failed`, { chunkIndex: i + 1, chunkDescription, errorCategory, }, err); failedChunks.push(`${chunkDescription}: ${err.message}`); // Continue with next chunk instead of failing entire sync } } chunkLogger.complete('Chunked sync', syncStartTime, { totalRecordsAdded, totalRecordsUpdated, totalRecordsDeleted, totalChunks: chunks.length, failedChunks: failedChunks.length, }); if (failedChunks.length > 0) { chunkLogger.warn('Some chunks failed', { failedChunks }); } return { recordsAdded: totalRecordsAdded, recordsUpdated: totalRecordsUpdated, recordsDeleted: totalRecordsDeleted, }; } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); chunkLogger.fail('Chunked sync', syncStartTime, err); throw error; } } /** * Get all valid resource IDs from the database * Used to validate foreign key references before insert * @returns Set of valid resource IDs */ private async getValidResourceIds(): Promise> { try { const query = 'SELECT id FROM resources WHERE is_deleted = false'; const result = await postgresClient.query<{ id: number }>(query); return new Set(result.rows.map(row => row.id)); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.logger.error('Failed to fetch valid resource IDs', {}, err); // Return empty set on error - will cause all resource IDs to be nullified return new Set(); } } /** * Used to validate foreign key references before insert * @returns Set of valid contact IDs */ private async getValidContactIds(): Promise> { try { const query = 'SELECT id FROM contacts WHERE is_deleted = false'; const result = await postgresClient.query<{ id: number }>(query); return new Set(result.rows.map(row => row.id)); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.logger.error('Failed to fetch valid contact IDs', {}, err); // Return empty set on error - will cause all contact IDs to be nullified return new Set(); } } /** * Calculate monthly date chunks for a given time period * @param yearsBack Number of years to look back * @returns Array of date range chunks */ private calculateMonthlyChunks(yearsBack: number): Array<{ startDate: Date; endDate: Date }> { const chunks: Array<{ startDate: Date; endDate: Date }> = []; const now = new Date(); const startDate = new Date(now); startDate.setFullYear(now.getFullYear() - yearsBack); startDate.setHours(0, 0, 0, 0); let currentDate = new Date(startDate); while (currentDate < now) { const chunkStart = new Date(currentDate); // Move to next month const chunkEnd = new Date(currentDate); chunkEnd.setMonth(chunkEnd.getMonth() + 1); // Don't go beyond current date if (chunkEnd > now) { chunkEnd.setTime(now.getTime()); } chunks.push({ startDate: chunkStart, endDate: chunkEnd, }); currentDate = new Date(chunkEnd); } return chunks; } /** * Sync Tasks */ async syncTasks(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.TASKS, isIncremental); } /** * Sync Projects */ async syncProjects(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.PROJECTS, isIncremental); } /** * Sync Resources (Users) */ async syncResources(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.RESOURCES, isIncremental); } /** * Sync Configuration Items */ async syncConfigurationItems(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.CONFIGURATION_ITEMS, isIncremental); } /** * Sync Contacts */ async syncContacts(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.CONTACTS, isIncremental); } /** * Sync Contracts */ async syncContracts(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.CONTRACTS, isIncremental); } /** * Sync Billing Items */ async syncBillingItems(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.BILLING_ITEMS, isIncremental); } /** * Sync Statuses (Picklist) */ async syncStatuses(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.STATUSES, isIncremental); } /** * Sync Issue Types (Picklist from Ticket field) */ async syncIssueTypes(isIncremental: boolean = false): Promise { const picklistLogger = this.logger.child({ entityType: EntityType.ISSUE_TYPES, syncType: 'picklist' }); const syncStartTime = picklistLogger.start('Picklist sync'); try { // Get issue type picklist values from Tickets entity const picklistValues = await this.autotaskClient.getPicklistValues('Tickets', 'issueType'); // Convert picklist to database format const records = Object.entries(picklistValues).map(([value, label]) => ({ value: parseInt(value), label: label, is_active: true, sort_order: parseInt(value), synced_at: new Date(), })); picklistLogger.info('Found picklist values', { recordCount: records.length }); // Upsert to database const tableName = getTableName(EntityType.ISSUE_TYPES); const upsertedCount = await postgresClient.bulkUpsert(tableName, records, ['value']); const stats: EntitySyncStats = { recordsAdded: upsertedCount, recordsUpdated: 0, recordsDeleted: 0, }; picklistLogger.complete('Picklist sync', syncStartTime, { recordsAdded: stats.recordsAdded, recordsUpdated: stats.recordsUpdated, }); return stats; } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); picklistLogger.fail('Picklist sync', syncStartTime, err); throw error; } } /** * Sync Sub-Issue Types (Picklist from Ticket field) */ async syncSubIssueTypes(isIncremental: boolean = false): Promise { const picklistLogger = this.logger.child({ entityType: EntityType.SUB_ISSUE_TYPES, syncType: 'picklist' }); const syncStartTime = picklistLogger.start('Picklist sync'); try { // Get sub-issue type picklist values from Tickets entity const url = `${this.autotaskClient['config'].apiUrl}/Tickets/entityInformation/fields`; const response = await fetch(url, { method: 'GET', headers: this.autotaskClient['getAuthHeaders'](), }); const responseText = await response.text(); if (!response.ok) { throw new Error(`Failed to fetch field info: ${responseText}`); } const fieldData = JSON.parse(responseText); const subIssueTypeField = fieldData.fields.find((f: any) => f.name === 'subIssueType'); if (!subIssueTypeField || !subIssueTypeField.picklistValues) { throw new Error('subIssueType field or picklist values not found'); } // Convert picklist to database format, capturing parent value if it exists const records = subIssueTypeField.picklistValues.map((item: any) => ({ value: parseInt(item.value), label: item.label, is_active: item.isActive !== false, parent_value: item.parentValue ? parseInt(item.parentValue) : null, sort_order: item.sortOrder || parseInt(item.value), synced_at: new Date(), })); picklistLogger.info('Found picklist values', { recordCount: records.length }); // Upsert to database const tableName = getTableName(EntityType.SUB_ISSUE_TYPES); const upsertedCount = await postgresClient.bulkUpsert(tableName, records, ['value']); const stats: EntitySyncStats = { recordsAdded: upsertedCount, recordsUpdated: 0, recordsDeleted: 0, }; picklistLogger.complete('Picklist sync', syncStartTime, { recordsAdded: stats.recordsAdded, recordsUpdated: stats.recordsUpdated, }); return stats; } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); picklistLogger.fail('Picklist sync', syncStartTime, err); throw error; } } /** * Sync Work Types (Picklist) */ async syncWorkTypes(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.WORK_TYPES, isIncremental); } /** * Sync Time Entries */ async syncTimeEntries(isIncremental: boolean = false): Promise { return await this.syncEntity(EntityType.TIME_ENTRIES, isIncremental); } } /** * Create entity sync service instance * @param autotaskClient Autotask client instance * @returns EntitySyncService instance */ export function createEntitySyncService(autotaskClient: AutotaskClient): EntitySyncService { return new EntitySyncService(autotaskClient); }