lorentz
62c529fb91
- Add getValidCompanyIds() helper mirroring getValidResourceIds() - Add a new TICKETS validation block that nullifies ticket.company_id when the referenced company is not present (is_deleted=false) in the Pulse mirror, instead of letting tickets_company_id_fkey roll back the bulkUpsert transaction - Block runs after the existing recordsWithoutCompany filter and before the existing resource-FK nullification block (correct ordering) - Belt-and-suspenders on top of Task 1: covers hard-deleted-in-Autotask companies that Task 1's widening still won't fetch
1478 lines
58 KiB
TypeScript
1478 lines
58 KiB
TypeScript
/**
|
|
* 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,
|
|
buildCompaniesFilter,
|
|
buildContractsFilter,
|
|
buildContractServicesFilter,
|
|
buildProjectsFilter,
|
|
buildProjectPhasesFilter,
|
|
buildTimeEntriesFilter,
|
|
buildBillingItemsFilter,
|
|
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<number>;
|
|
private cachedValidContactIds?: Set<number>;
|
|
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<EntitySyncStats> {
|
|
// Route picklist entities to their specific sync methods
|
|
if (entity === EntityType.STATUSES) {
|
|
return await this.syncStatuses(isIncremental);
|
|
}
|
|
if (entity === EntityType.WORK_TYPES) {
|
|
return await this.syncWorkTypes(isIncremental);
|
|
}
|
|
if (entity === EntityType.ISSUE_TYPES) {
|
|
return await this.syncIssueTypes(isIncremental);
|
|
}
|
|
if (entity === EntityType.SUB_ISSUE_TYPES) {
|
|
return await this.syncSubIssueTypes(isIncremental);
|
|
}
|
|
if (entity === EntityType.QUEUES) {
|
|
return await this.syncQueues(isIncremental);
|
|
}
|
|
if (entity === EntityType.PRIORITIES) {
|
|
return await this.syncPriorities(isIncremental);
|
|
}
|
|
if (entity === EntityType.TICKET_CATEGORIES) {
|
|
return await this.syncTicketCategories(isIncremental);
|
|
}
|
|
if (entity === EntityType.COMPANY_CATEGORIES) {
|
|
return await this.syncCompanyCategories(isIncremental);
|
|
}
|
|
if (entity === EntityType.COMPANY_TYPES) {
|
|
return await this.syncCompanyTypes(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
|
|
// Note: Companies and Resources don't support date-based filtering in Autotask API
|
|
const supportsIncremental = entity !== EntityType.COMPANIES && entity !== EntityType.RESOURCES && entity !== EntityType.COMPANY_TEAMS;
|
|
|
|
if (isIncremental && supportsIncremental) {
|
|
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}`);
|
|
}
|
|
}
|
|
|
|
// For full sync OR entities that don't support incremental, build filters
|
|
if (!isIncremental || (isIncremental && !supportsIncremental)) {
|
|
if (isIncremental && !supportsIncremental) {
|
|
entityLogger.info(`${entity} does not support incremental sync, performing full sync instead`);
|
|
}
|
|
|
|
const filters: Array<{ field: string; op: string; value: any }> = [];
|
|
|
|
// Special handling for entities that require filters
|
|
if (entity === EntityType.COMPANIES) {
|
|
filters.push(...buildCompaniesFilter());
|
|
entityLogger.info('Full sync of all companies (active + inactive)');
|
|
} else if (entity === EntityType.CONTRACTS) {
|
|
filters.push(...buildContractsFilter());
|
|
entityLogger.info('Full sync with status filter for active contracts');
|
|
} else if (entity === EntityType.CONTRACT_SERVICES) {
|
|
filters.push(...buildContractServicesFilter());
|
|
entityLogger.info('Full sync of all contract services');
|
|
} else if (entity === EntityType.PROJECTS) {
|
|
filters.push(...buildProjectsFilter());
|
|
entityLogger.info('Full sync with status filter for non-completed projects');
|
|
} else if (entity === EntityType.PROJECT_PHASES) {
|
|
filters.push(...buildProjectPhasesFilter());
|
|
entityLogger.info('Full sync of all project phases');
|
|
} else if (entity === EntityType.TIME_ENTRIES) {
|
|
filters.push(...buildTimeEntriesFilter(yearsBack));
|
|
entityLogger.info(`Full sync with dateWorked filter for last ${yearsBack} years`);
|
|
} else if (entity === EntityType.BILLING_ITEMS) {
|
|
filters.push(...buildBillingItemsFilter(yearsBack));
|
|
entityLogger.info(`Full sync with itemDate 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;
|
|
}
|
|
}
|
|
|
|
// Track whether any filters were applied - if so, skip soft deletes
|
|
const hasAppliedFilters = params.filter && params.filter.length > 0;
|
|
|
|
// 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<string, any>[];
|
|
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 company_id foreign keys for tickets (defensive — Task 1 widening
|
|
// the Companies filter should make this a near-zero count, but catches the
|
|
// truly hard-deleted Autotask company case).
|
|
if (entity === EntityType.TICKETS) {
|
|
const validCompanyIds = await this.getValidCompanyIds();
|
|
let nullifiedCompanyCount = 0;
|
|
const nullifiedCompanySamples: number[] = [];
|
|
mappedRecords = mappedRecords.map(ticket => {
|
|
if (ticket.company_id && !validCompanyIds.has(ticket.company_id)) {
|
|
if (nullifiedCompanySamples.length < 5) {
|
|
nullifiedCompanySamples.push(ticket.id);
|
|
}
|
|
ticket.company_id = null;
|
|
nullifiedCompanyCount++;
|
|
}
|
|
return ticket;
|
|
});
|
|
if (nullifiedCompanyCount > 0) {
|
|
entityLogger.warn('Nullified ticket company_id for companies missing from mirror', {
|
|
nullifiedCount: nullifiedCompanyCount,
|
|
sampleTicketIds: nullifiedCompanySamples,
|
|
});
|
|
}
|
|
}
|
|
|
|
// 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
|
|
let skippedResourceCount = 0;
|
|
mappedRecords = mappedRecords.map(ticket => {
|
|
// Set to null — bulkUpsert uses COALESCE for these columns so existing DB value is preserved
|
|
if (ticket.assigned_resource_id && !validResourceIds.has(ticket.assigned_resource_id)) {
|
|
entityLogger.debug(`Unknown assigned_resource_id, will preserve existing DB value via COALESCE`, {
|
|
ticketId: ticket.id,
|
|
unknownResourceId: ticket.assigned_resource_id,
|
|
});
|
|
ticket.assigned_resource_id = null;
|
|
skippedResourceCount++;
|
|
}
|
|
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;
|
|
});
|
|
|
|
if (skippedResourceCount > 0) {
|
|
entityLogger.warn('Skipped unknown resource references (existing DB values preserved via COALESCE)', { skippedResourceCount });
|
|
}
|
|
}
|
|
|
|
// 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 });
|
|
}
|
|
}
|
|
|
|
// Validate project and resource foreign keys for tasks
|
|
if (entity === EntityType.TASKS) {
|
|
const initialCount = mappedRecords.length;
|
|
|
|
// Get all valid project IDs and resource IDs from database
|
|
const validProjectIds = await this.getValidProjectIds();
|
|
const validResourceIds = await this.getValidResourceIds();
|
|
|
|
// Filter tasks with invalid foreign key references
|
|
mappedRecords = mappedRecords.map(task => {
|
|
// Set invalid project IDs to null
|
|
if (task.project_id && !validProjectIds.has(task.project_id)) {
|
|
entityLogger.debug(`Invalid project_id, setting to null`, {
|
|
taskId: task.id,
|
|
invalidProjectId: task.project_id,
|
|
});
|
|
task.project_id = null;
|
|
}
|
|
|
|
// Set invalid resource IDs to null
|
|
if (task.assigned_resource_id && !validResourceIds.has(task.assigned_resource_id)) {
|
|
entityLogger.debug(`Invalid assigned_resource_id, setting to null`, {
|
|
taskId: task.id,
|
|
invalidResourceId: task.assigned_resource_id,
|
|
});
|
|
task.assigned_resource_id = null;
|
|
}
|
|
if (task.creator_resource_id && !validResourceIds.has(task.creator_resource_id)) {
|
|
task.creator_resource_id = null;
|
|
}
|
|
if (task.completed_by_resource_id && !validResourceIds.has(task.completed_by_resource_id)) {
|
|
task.completed_by_resource_id = null;
|
|
}
|
|
if (task.last_activity_resource_id && !validResourceIds.has(task.last_activity_resource_id)) {
|
|
task.last_activity_resource_id = null;
|
|
}
|
|
|
|
return task;
|
|
});
|
|
|
|
const nullifiedProjectCount = initialCount - mappedRecords.filter(t => t.project_id).length;
|
|
const nullifiedResourceCount = initialCount - mappedRecords.filter(t => t.assigned_resource_id).length;
|
|
if (nullifiedProjectCount > 0 || nullifiedResourceCount > 0) {
|
|
entityLogger.warn('Nullified invalid foreign key references', {
|
|
nullifiedProjectCount,
|
|
nullifiedResourceCount
|
|
});
|
|
}
|
|
}
|
|
|
|
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 });
|
|
|
|
// Post-sync enrichment: populate service_name from autotask_services lookup
|
|
if (entity === EntityType.CONTRACT_SERVICES) {
|
|
try {
|
|
const enrichResult = await postgresClient.query(`
|
|
UPDATE contract_services cs
|
|
SET service_name = s.name
|
|
FROM autotask_services s
|
|
WHERE cs.service_id = s.id AND cs.service_name IS NULL AND s.name IS NOT NULL
|
|
`);
|
|
entityLogger.info('Enriched contract_services with service names', { rowCount: enrichResult.rowCount });
|
|
} catch (err) {
|
|
entityLogger.warn('service_name enrichment skipped (autotask_services may not be synced yet)');
|
|
}
|
|
}
|
|
|
|
// Post-sync backfill: after syncing phases, populate tasks.project_id via phase → project join.
|
|
// The Autotask Tasks bulk query does not return projectID — it must be resolved through phaseID.
|
|
if (entity === EntityType.PROJECT_PHASES) {
|
|
try {
|
|
const backfillResult = await postgresClient.query(
|
|
`UPDATE tasks
|
|
SET project_id = pp.project_id
|
|
FROM project_phases pp
|
|
JOIN projects p ON p.id = pp.project_id
|
|
WHERE tasks.phase_id = pp.id
|
|
AND tasks.project_id IS DISTINCT FROM pp.project_id
|
|
AND pp.project_id IS NOT NULL`
|
|
);
|
|
const updated = (backfillResult as any).rowCount ?? 0;
|
|
entityLogger.info('Backfilled tasks.project_id via phase → project join', { updatedTaskCount: updated });
|
|
} catch (err) {
|
|
entityLogger.warn('Task project_id backfill failed', { error: String(err) });
|
|
}
|
|
}
|
|
|
|
// After contacts sync, backfill primary_contact_id and billing_contact_id on companies
|
|
if (entity === EntityType.CONTACTS) {
|
|
try {
|
|
const primaryResult = await postgresClient.query(
|
|
`UPDATE companies c
|
|
SET primary_contact_id = (
|
|
SELECT id FROM contacts
|
|
WHERE company_id = c.id AND primary_contact = true AND is_deleted = false
|
|
ORDER BY id LIMIT 1
|
|
)
|
|
WHERE EXISTS (
|
|
SELECT 1 FROM contacts
|
|
WHERE company_id = c.id AND primary_contact = true AND is_deleted = false
|
|
)`
|
|
);
|
|
entityLogger.info('Backfilled companies.primary_contact_id', { updatedCount: (primaryResult as any).rowCount ?? 0 });
|
|
|
|
const billingResult = await postgresClient.query(
|
|
`UPDATE companies c
|
|
SET billing_contact_id = (
|
|
SELECT id FROM contacts
|
|
WHERE company_id = c.id AND billing_contact = true AND is_deleted = false
|
|
ORDER BY id LIMIT 1
|
|
)
|
|
WHERE EXISTS (
|
|
SELECT 1 FROM contacts
|
|
WHERE company_id = c.id AND billing_contact = true AND is_deleted = false
|
|
)`
|
|
);
|
|
entityLogger.info('Backfilled companies.billing_contact_id', { updatedCount: (billingResult as any).rowCount ?? 0 });
|
|
} catch (err) {
|
|
entityLogger.warn('Contact FK backfill on companies failed', { error: String(err) });
|
|
}
|
|
}
|
|
|
|
// For full sync, soft delete records not in the fetched set
|
|
// IMPORTANT: Skip soft deletes if ANY filters were applied (date, status, active, etc.)
|
|
// because we cannot know what records exist outside the filter criteria.
|
|
// EXCEPTION: TICKETS — we do a separate ID-only fetch from Autotask (no filters) to detect
|
|
// deletions even when date filters were applied to the main sync.
|
|
let deletedCount = 0;
|
|
|
|
if (!isIncremental && !hasAppliedFilters) {
|
|
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);
|
|
}
|
|
} else if (!isIncremental && hasAppliedFilters && entity === EntityType.TICKETS) {
|
|
entityLogger.phase(SyncPhase.DELETING, 'Fetching all Autotask ticket IDs for deletion diff');
|
|
syncProgressTracker.updateProgress(trackingId, { phase: 'deleting' });
|
|
|
|
try {
|
|
// Fetch all ticket IDs from Autotask with no filters for deletion diff
|
|
const allAutotaskTickets = await this.autotaskClient.queryEntityPaginated(
|
|
'Tickets',
|
|
{},
|
|
500
|
|
);
|
|
const liveIds = new Set(allAutotaskTickets.map((t: any) => String(t.id)));
|
|
|
|
// Find Pulse ticket IDs not present in Autotask → these were deleted
|
|
const pulseResult = await postgresClient.query(
|
|
`SELECT id FROM tickets WHERE is_deleted = false`
|
|
);
|
|
const toDelete = pulseResult.rows
|
|
.map((r: { id: number }) => r.id)
|
|
.filter((id: number) => !liveIds.has(String(id)));
|
|
|
|
if (toDelete.length > 0) {
|
|
const result = await postgresClient.query(
|
|
`UPDATE tickets
|
|
SET is_deleted = true, deleted_at = CURRENT_TIMESTAMP
|
|
WHERE id = ANY($1::int[])`,
|
|
[toDelete]
|
|
);
|
|
deletedCount = result.rowCount || 0;
|
|
entityLogger.info('Soft deleted tickets missing from Autotask', { deletedCount });
|
|
} else {
|
|
entityLogger.info('No deleted tickets detected');
|
|
}
|
|
} catch (error) {
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
entityLogger.warn('Ticket deletion diff failed, continuing sync', {}, err);
|
|
}
|
|
} else if (!isIncremental && hasAppliedFilters) {
|
|
entityLogger.info('Skipping soft-delete because filters were applied (would delete records outside filter criteria)');
|
|
}
|
|
|
|
// 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<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.COMPANIES, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Tickets
|
|
*/
|
|
async syncTickets(isIncremental: boolean = false, yearsBack?: number): Promise<EntitySyncStats> {
|
|
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<EntitySyncStats> {
|
|
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 });
|
|
}
|
|
|
|
// Null out unknown resource references — bulkUpsert uses COALESCE so existing DB value is preserved
|
|
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<Set<number>> {
|
|
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 => Number(row.id)));
|
|
}
|
|
|
|
/**
|
|
* Get all valid company IDs from the database
|
|
* Used to validate foreign key references before insert
|
|
* @returns Set of valid company IDs
|
|
*/
|
|
private async getValidCompanyIds(): Promise<Set<number>> {
|
|
const query = 'SELECT id FROM companies WHERE is_deleted = false';
|
|
const result = await postgresClient.query<{ id: number }>(query);
|
|
return new Set(result.rows.map(row => Number(row.id)));
|
|
}
|
|
|
|
/**
|
|
* Used to validate foreign key references before insert
|
|
* @returns Set of valid contact IDs
|
|
*/
|
|
private async getValidContactIds(): Promise<Set<number>> {
|
|
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 => Number(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();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get all valid project IDs from the database
|
|
* Used to validate foreign key references before insert
|
|
* @returns Set of valid project IDs
|
|
*/
|
|
private async getValidProjectIds(): Promise<Set<number>> {
|
|
try {
|
|
const query = 'SELECT id FROM projects WHERE is_deleted = false';
|
|
const result = await postgresClient.query<{ id: number }>(query);
|
|
return new Set(result.rows.map(row => Number(row.id)));
|
|
} catch (error) {
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
this.logger.error('Failed to fetch valid project IDs', {}, err);
|
|
// Return empty set on error - will cause all project 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<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.TASKS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Projects
|
|
*/
|
|
async syncProjects(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.PROJECTS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Project Phases, then backfill project_id on tasks that reference those phases.
|
|
* Autotask's Tasks bulk query does not return projectID — it must be resolved via phaseID.
|
|
*/
|
|
async syncProjectPhases(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const stats = await this.syncEntity(EntityType.PROJECT_PHASES, isIncremental);
|
|
|
|
// Backfill tasks.project_id using the newly synced phases
|
|
try {
|
|
const result = await postgresClient.query<{ rowCount: number }>(
|
|
`UPDATE tasks
|
|
SET project_id = pp.project_id
|
|
FROM project_phases pp
|
|
WHERE tasks.phase_id = pp.id
|
|
AND tasks.project_id IS DISTINCT FROM pp.project_id
|
|
AND pp.project_id IS NOT NULL`
|
|
);
|
|
const updated = (result as any).rowCount ?? 0;
|
|
if (updated > 0) {
|
|
this.logger.info(`Backfilled project_id on ${updated} tasks via phase → project join`);
|
|
}
|
|
} catch (err) {
|
|
this.logger.warn('Task project_id backfill failed', { error: String(err) });
|
|
}
|
|
|
|
return stats;
|
|
}
|
|
|
|
/**
|
|
* Sync Resources (Users)
|
|
*/
|
|
async syncResources(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.RESOURCES, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Configuration Items
|
|
*/
|
|
async syncConfigurationItems(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.CONFIGURATION_ITEMS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Contacts
|
|
*/
|
|
async syncContacts(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.CONTACTS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Company Teams (TAMs, CSMs, co-managed resources assigned to a company)
|
|
*/
|
|
async syncCompanyTeams(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.COMPANY_TEAMS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Contracts
|
|
*/
|
|
async syncContracts(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.CONTRACTS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Billing Items
|
|
*/
|
|
async syncBillingItems(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.BILLING_ITEMS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Statuses (Picklist from Ticket field)
|
|
*/
|
|
async syncStatuses(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.STATUSES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
// Get status picklist values from Tickets entity
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('Tickets', 'status');
|
|
|
|
// 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 });
|
|
|
|
// Get existing values to calculate added vs updated
|
|
const tableName = getTableName(EntityType.STATUSES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
// Upsert to database
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = {
|
|
recordsAdded,
|
|
recordsUpdated,
|
|
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 Issue Types (Picklist from Ticket field)
|
|
*/
|
|
async syncIssueTypes(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
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 });
|
|
|
|
// Get existing values to calculate added vs updated
|
|
const tableName = getTableName(EntityType.ISSUE_TYPES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
// Upsert to database
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = {
|
|
recordsAdded,
|
|
recordsUpdated,
|
|
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<EntitySyncStats> {
|
|
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 });
|
|
|
|
// Get existing values to calculate added vs updated
|
|
const tableName = getTableName(EntityType.SUB_ISSUE_TYPES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map(r => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
// Upsert to database
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = {
|
|
recordsAdded,
|
|
recordsUpdated,
|
|
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 Queues (Picklist from Ticket field)
|
|
*/
|
|
async syncQueues(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.QUEUES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('Tickets', 'queueID');
|
|
|
|
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 });
|
|
|
|
const tableName = getTableName(EntityType.QUEUES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = { recordsAdded, recordsUpdated, 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 Priorities (Picklist from Ticket field)
|
|
*/
|
|
async syncPriorities(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.PRIORITIES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('Tickets', 'priority');
|
|
|
|
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 });
|
|
|
|
const tableName = getTableName(EntityType.PRIORITIES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = { recordsAdded, recordsUpdated, 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 Ticket Categories (Picklist from Ticket field)
|
|
*/
|
|
async syncTicketCategories(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.TICKET_CATEGORIES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('Tickets', 'ticketCategory');
|
|
|
|
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 });
|
|
|
|
const tableName = getTableName(EntityType.TICKET_CATEGORIES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = { recordsAdded, recordsUpdated, 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 Company Categories (queried from CompanyCategories entity — id, name, isActive)
|
|
*/
|
|
async syncCompanyCategories(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const syncLogger = this.logger.child({ entityType: EntityType.COMPANY_CATEGORIES, syncType: 'entity' });
|
|
const syncStartTime = syncLogger.start('Company Categories sync');
|
|
|
|
try {
|
|
const apiRecords = await this.autotaskClient.queryEntityPaginated(
|
|
'CompanyCategories',
|
|
{ filter: [{ field: 'id', op: 'gt', value: 0 }] },
|
|
500
|
|
);
|
|
|
|
const records = apiRecords.map((r: any) => ({
|
|
value: r.id,
|
|
label: r.name || r.nickname || String(r.id),
|
|
is_active: r.isActive !== false,
|
|
sort_order: r.id,
|
|
synced_at: new Date(),
|
|
}));
|
|
|
|
syncLogger.info('Fetched company categories', { recordCount: records.length });
|
|
|
|
const tableName = getTableName(EntityType.COMPANY_CATEGORIES);
|
|
const existingResult = await postgresClient.query<{ value: number }>(`SELECT value FROM ${tableName}`);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = { recordsAdded, recordsUpdated, recordsDeleted: 0 };
|
|
syncLogger.complete('Company Categories sync', syncStartTime, stats);
|
|
return stats;
|
|
} catch (error) {
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
syncLogger.fail('Company Categories sync', syncStartTime, err);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sync Company Types (Picklist from Companies field)
|
|
*/
|
|
async syncCompanyTypes(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.COMPANY_TYPES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('Companies', 'companyType');
|
|
|
|
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 });
|
|
|
|
const tableName = getTableName(EntityType.COMPANY_TYPES);
|
|
const existingResult = await postgresClient.query<{ value: number }>(`SELECT value FROM ${tableName}`);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = { recordsAdded, recordsUpdated, recordsDeleted: 0 };
|
|
picklistLogger.complete('Picklist sync', syncStartTime, stats);
|
|
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 from TimeEntry field)
|
|
*/
|
|
async syncWorkTypes(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
const picklistLogger = this.logger.child({ entityType: EntityType.WORK_TYPES, syncType: 'picklist' });
|
|
const syncStartTime = picklistLogger.start('Picklist sync');
|
|
|
|
try {
|
|
// Get work type picklist values from TimeEntries entity
|
|
const picklistValues = await this.autotaskClient.getPicklistValues('TimeEntries', 'workType');
|
|
|
|
// 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 });
|
|
|
|
// Get existing values to calculate added vs updated
|
|
const tableName = getTableName(EntityType.WORK_TYPES);
|
|
const existingQuery = `SELECT value FROM ${tableName}`;
|
|
const existingResult = await postgresClient.query<{ value: number }>(existingQuery);
|
|
const existingValues = new Set(existingResult.rows.map((r: any) => r.value));
|
|
|
|
const recordsAdded = records.filter((r: any) => !existingValues.has(r.value)).length;
|
|
const recordsUpdated = records.filter((r: any) => existingValues.has(r.value)).length;
|
|
|
|
// Upsert to database
|
|
await postgresClient.bulkUpsert(tableName, records, ['value']);
|
|
|
|
const stats: EntitySyncStats = {
|
|
recordsAdded,
|
|
recordsUpdated,
|
|
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 Time Entries
|
|
*/
|
|
async syncTimeEntries(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.TIME_ENTRIES, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Ticket Notes. Webhooks are the primary path; this exists so missed
|
|
* events (webhook outages, replays) get reconciled by the scheduled sync.
|
|
*/
|
|
async syncTicketNotes(
|
|
isIncremental: boolean = false,
|
|
yearsBack: number = 2
|
|
): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.TICKET_NOTES, isIncremental, yearsBack);
|
|
}
|
|
|
|
/**
|
|
* Sync Tag Groups
|
|
*/
|
|
async syncTagGroups(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.TAG_GROUPS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync Tags
|
|
*/
|
|
async syncTags(isIncremental: boolean = false): Promise<EntitySyncStats> {
|
|
return await this.syncEntity(EntityType.TAGS, isIncremental);
|
|
}
|
|
|
|
/**
|
|
* Sync TicketTagAssociations from Autotask API into the ticket_tags junction table.
|
|
* Autotask exposes TicketTagAssociations as a bulk-queryable entity with fields:
|
|
* id, tagID, ticketID
|
|
*/
|
|
async syncTicketTagAssociations(): Promise<EntitySyncStats> {
|
|
const tagLogger = this.logger.child({ entityType: 'ticket_tag_associations', syncType: 'full' });
|
|
const syncStartTime = tagLogger.start('TicketTagAssociations sync');
|
|
|
|
try {
|
|
// Fetch all TicketTagAssociations from Autotask
|
|
tagLogger.info('Fetching TicketTagAssociations from Autotask');
|
|
const associations = await this.autotaskClient.queryEntityPaginated<{
|
|
id: number;
|
|
tagID: number;
|
|
ticketID: number;
|
|
}>('TicketTagAssociations', { filter: [{ op: 'exist', field: 'id' }] }, 500);
|
|
|
|
tagLogger.info('Fetched TicketTagAssociations', { count: associations.length });
|
|
|
|
if (associations.length === 0) {
|
|
tagLogger.info('No ticket tag associations found');
|
|
tagLogger.complete('TicketTagAssociations sync', syncStartTime, { recordsAdded: 0 });
|
|
return { recordsAdded: 0, recordsUpdated: 0, recordsDeleted: 0 };
|
|
}
|
|
|
|
// Validate tag IDs exist in our DB to avoid FK violations on autotask_tags
|
|
// (ticket_id has no FK constraint since tagged tickets may be outside sync window)
|
|
const validTagResult = await postgresClient.query<{ id: number }>(
|
|
'SELECT id FROM autotask_tags'
|
|
);
|
|
const validTagIds = new Set(validTagResult.rows.map(r => Number(r.id)));
|
|
|
|
const validAssociations = associations.filter(a => validTagIds.has(Number(a.tagID)));
|
|
|
|
tagLogger.info('Validated associations', {
|
|
total: associations.length,
|
|
valid: validAssociations.length,
|
|
skippedInvalidTag: associations.length - validAssociations.length,
|
|
});
|
|
|
|
// Replace all ticket_tags: truncate and re-insert for a clean full sync
|
|
await postgresClient.query('DELETE FROM ticket_tags');
|
|
|
|
const CHUNK = 500;
|
|
let inserted = 0;
|
|
for (let i = 0; i < validAssociations.length; i += CHUNK) {
|
|
const chunk = validAssociations.slice(i, i + CHUNK);
|
|
const values = chunk
|
|
.map((_, idx) => `($${idx * 2 + 1}::bigint, $${idx * 2 + 2}::bigint, NOW())`)
|
|
.join(', ');
|
|
const params = chunk.flatMap(a => [a.ticketID, a.tagID]);
|
|
await postgresClient.query(
|
|
`INSERT INTO ticket_tags (ticket_id, tag_id, synced_at)
|
|
VALUES ${values}
|
|
ON CONFLICT (ticket_id, tag_id) DO UPDATE SET synced_at = NOW()`,
|
|
params
|
|
);
|
|
inserted += chunk.length;
|
|
}
|
|
|
|
const stats: EntitySyncStats = {
|
|
recordsAdded: inserted,
|
|
recordsUpdated: 0,
|
|
recordsDeleted: 0,
|
|
};
|
|
|
|
tagLogger.complete('TicketTagAssociations sync', syncStartTime, stats);
|
|
return stats;
|
|
} catch (error) {
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
tagLogger.fail('TicketTagAssociations sync', syncStartTime, err);
|
|
throw error;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Create entity sync service instance
|
|
* @param autotaskClient Autotask client instance
|
|
* @returns EntitySyncService instance
|
|
*/
|
|
export function createEntitySyncService(autotaskClient: AutotaskClient): EntitySyncService {
|
|
return new EntitySyncService(autotaskClient);
|
|
}
|