432 lines
15 KiB
TypeScript
432 lines
15 KiB
TypeScript
/**
|
|
* Webhook Service
|
|
* Handles incoming webhooks from Autotask for real-time updates
|
|
*/
|
|
|
|
import { createHmac } from 'crypto';
|
|
import { postgresClient } from './postgres-client';
|
|
import { AutotaskWebhookPayload, WebhookProcessingResult, WebhookLog, WebhookEventType, WebhookEntityType } from '../types/webhook';
|
|
import { EntityType } from '../types/sync';
|
|
import { mapAutotaskToDatabase } from '../utils/entity-mapper';
|
|
import { getTableName, getAutotaskEntityName } from '../utils/sync-helpers';
|
|
import { AutotaskClient } from './autotask-client';
|
|
import { workflowEngine } from './workflow-engine';
|
|
import { WorkflowEvent, TicketData } from '../types/workflow';
|
|
|
|
export class WebhookService {
|
|
private _autotaskClient: AutotaskClient | null = null;
|
|
|
|
private getAutotaskClient(): AutotaskClient {
|
|
if (!this._autotaskClient) {
|
|
this._autotaskClient = new AutotaskClient({
|
|
apiUrl: process.env.AUTOTASK_API_URL || '',
|
|
username: process.env.AUTOTASK_USERNAME || '',
|
|
password: process.env.AUTOTASK_SECRET || '',
|
|
apiIntegrationCode: process.env.AUTOTASK_API_INTEGRATION_CODE || '',
|
|
});
|
|
}
|
|
return this._autotaskClient;
|
|
}
|
|
|
|
/**
|
|
* Verify webhook signature.
|
|
* Autotask sends: x-hook-signature: sha1=<base64>
|
|
* Computed as HMAC-SHA1 of the raw body using the secret key.
|
|
*/
|
|
verifySignature(rawBody: string, signatureHeader: string | null): boolean {
|
|
const secret = process.env.AUTOTASK_WEBHOOK_SECRET;
|
|
if (!secret) {
|
|
console.warn('[WEBHOOK] No AUTOTASK_WEBHOOK_SECRET configured, skipping signature verification');
|
|
return true;
|
|
}
|
|
if (!signatureHeader) {
|
|
console.warn('[WEBHOOK] No x-hook-signature header in webhook request');
|
|
return false;
|
|
}
|
|
|
|
// Header format: "sha1=<base64>"
|
|
const signature = signatureHeader.startsWith('sha1=')
|
|
? signatureHeader.slice(5)
|
|
: signatureHeader;
|
|
|
|
const computed = createHmac('sha1', secret).update(rawBody).digest('base64');
|
|
const match = computed === signature;
|
|
if (!match) {
|
|
console.warn(`[WEBHOOK] Signature mismatch: expected=${computed}, received=${signature}`);
|
|
}
|
|
return match;
|
|
}
|
|
/**
|
|
* Process an incoming webhook from Autotask
|
|
*/
|
|
async processWebhook(payload: AutotaskWebhookPayload, sourceIp?: string, userAgent?: string): Promise<WebhookProcessingResult> {
|
|
const startTime = Date.now();
|
|
|
|
try {
|
|
// Log the webhook event
|
|
await this.logWebhookEvent(payload, 'pending', sourceIp, userAgent);
|
|
|
|
console.log(`[WEBHOOK] Processing ${payload.eventType} event for ${payload.entityType} #${payload.entityId}`);
|
|
|
|
// Check if webhook is configured and active
|
|
const isActive = await this.isWebhookActive(payload.entityType);
|
|
if (!isActive) {
|
|
console.log(`[WEBHOOK] Webhook disabled for ${payload.entityType}, skipping`);
|
|
await this.updateWebhookLog(payload.eventId, 'processed', 'Webhook disabled for entity type');
|
|
return {
|
|
success: true,
|
|
eventId: payload.eventId,
|
|
entityType: payload.entityType,
|
|
entityId: payload.entityId,
|
|
action: 'skipped',
|
|
processingTime: Date.now() - startTime,
|
|
};
|
|
}
|
|
|
|
let action: 'created' | 'updated' | 'deleted' | 'skipped';
|
|
|
|
// Handle different event types
|
|
switch (payload.eventType) {
|
|
case WebhookEventType.CREATE:
|
|
case WebhookEventType.UPDATE:
|
|
action = await this.handleCreateOrUpdate(payload);
|
|
break;
|
|
|
|
case WebhookEventType.DELETE:
|
|
action = await this.handleDelete(payload);
|
|
break;
|
|
|
|
default:
|
|
throw new Error(`Unknown event type: ${payload.eventType}`);
|
|
}
|
|
|
|
const processingTime = Date.now() - startTime;
|
|
|
|
// Update webhook log as processed
|
|
await this.updateWebhookLog(payload.eventId, 'processed', undefined, processingTime);
|
|
|
|
console.log(`[WEBHOOK] Successfully processed ${payload.eventType} for ${payload.entityType} #${payload.entityId} in ${processingTime}ms`);
|
|
|
|
// Trigger workflow engine for new tickets (fire-and-forget)
|
|
if (payload.entityType === WebhookEntityType.TICKETS && payload.eventType === WebhookEventType.CREATE) {
|
|
this.triggerWorkflowEngine(payload).catch(err =>
|
|
console.error('[WEBHOOK] Workflow engine error:', err)
|
|
);
|
|
}
|
|
|
|
return {
|
|
success: true,
|
|
eventId: payload.eventId,
|
|
entityType: payload.entityType,
|
|
entityId: payload.entityId,
|
|
action,
|
|
processingTime,
|
|
};
|
|
|
|
} catch (error) {
|
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
const processingTime = Date.now() - startTime;
|
|
|
|
console.error(`[WEBHOOK] Failed to process webhook:`, errorMessage);
|
|
|
|
// Update webhook log as failed
|
|
await this.updateWebhookLog(payload.eventId, 'failed', errorMessage, processingTime);
|
|
|
|
return {
|
|
success: false,
|
|
eventId: payload.eventId,
|
|
entityType: payload.entityType,
|
|
entityId: payload.entityId,
|
|
action: 'skipped',
|
|
error: errorMessage,
|
|
processingTime,
|
|
};
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle create or update events
|
|
*/
|
|
private async handleCreateOrUpdate(payload: AutotaskWebhookPayload): Promise<'created' | 'updated'> {
|
|
// If webhook includes full entity data, use it directly
|
|
if (payload.entity && Object.keys(payload.entity).length > 2) {
|
|
await this.upsertEntity(payload.entityType, payload.entity);
|
|
return payload.eventType === WebhookEventType.CREATE ? 'created' : 'updated';
|
|
}
|
|
|
|
// Fetch the full entity from Autotask API
|
|
const internalEntityType = this.mapWebhookEntityType(payload.entityType);
|
|
const autotaskEntityName = getAutotaskEntityName(internalEntityType);
|
|
|
|
console.log(`[WEBHOOK] Fetching ${autotaskEntityName} #${payload.entityId} from Autotask API`);
|
|
|
|
const client = this.getAutotaskClient();
|
|
const entity = await client.getEntityById(autotaskEntityName, payload.entityId);
|
|
|
|
if (!entity) {
|
|
console.warn(`[WEBHOOK] Entity ${autotaskEntityName} #${payload.entityId} not found in Autotask API`);
|
|
return payload.eventType === WebhookEventType.CREATE ? 'created' : 'updated';
|
|
}
|
|
|
|
await this.upsertEntity(payload.entityType, entity as Record<string, any>);
|
|
return payload.eventType === WebhookEventType.CREATE ? 'created' : 'updated';
|
|
}
|
|
|
|
/**
|
|
* Handle delete events
|
|
*/
|
|
private async handleDelete(payload: AutotaskWebhookPayload): Promise<'deleted'> {
|
|
const tableName = this.getTableNameFromWebhookEntity(payload.entityType);
|
|
|
|
// Soft delete the entity
|
|
const query = `
|
|
UPDATE ${tableName}
|
|
SET is_deleted = true, deleted_at = NOW()
|
|
WHERE id = $1
|
|
`;
|
|
|
|
await postgresClient.query(query, [payload.entityId]);
|
|
|
|
console.log(`[WEBHOOK] Soft deleted ${payload.entityType} #${payload.entityId}`);
|
|
|
|
return 'deleted';
|
|
}
|
|
|
|
/**
|
|
* Upsert entity data to database
|
|
*/
|
|
private async upsertEntity(entityType: WebhookEntityType, entityData: Record<string, any>): Promise<void> {
|
|
// Map webhook entity type to internal EntityType
|
|
const internalEntityType = this.mapWebhookEntityType(entityType);
|
|
|
|
// Map Autotask data to PostgreSQL schema
|
|
const mappedData = mapAutotaskToDatabase(internalEntityType, entityData);
|
|
|
|
if (!mappedData) {
|
|
throw new Error(`Failed to map entity data for ${entityType}`);
|
|
}
|
|
|
|
// Get table name
|
|
const tableName = getTableName(internalEntityType);
|
|
|
|
// Build upsert query
|
|
const keys = Object.keys(mappedData);
|
|
const values = Object.values(mappedData);
|
|
const placeholders = keys.map((_, i) => `$${i + 1}`).join(', ');
|
|
const updateClause = keys
|
|
.filter(k => k !== 'id')
|
|
.map(k => `${k} = EXCLUDED.${k}`)
|
|
.join(', ');
|
|
|
|
const query = `
|
|
INSERT INTO ${tableName} (${keys.join(', ')})
|
|
VALUES (${placeholders})
|
|
ON CONFLICT (id)
|
|
DO UPDATE SET ${updateClause}, updated_at = NOW()
|
|
`;
|
|
|
|
await postgresClient.query(query, values);
|
|
|
|
console.log(`[WEBHOOK] Upserted ${entityType} #${mappedData.id}`);
|
|
}
|
|
|
|
/**
|
|
* Log webhook event to database
|
|
*/
|
|
private async logWebhookEvent(payload: AutotaskWebhookPayload, status: 'pending' | 'processed' | 'failed', sourceIp?: string, userAgent?: string): Promise<void> {
|
|
const query = `
|
|
INSERT INTO webhook_logs (event_id, entity_type, entity_id, event_type, status, source_ip, user_agent, payload)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT (event_id) DO NOTHING
|
|
`;
|
|
|
|
await postgresClient.query(query, [
|
|
payload.eventId,
|
|
payload.entityType,
|
|
payload.entityId,
|
|
payload.eventType,
|
|
status,
|
|
sourceIp || null,
|
|
userAgent || null,
|
|
JSON.stringify(payload),
|
|
]);
|
|
}
|
|
|
|
/**
|
|
* Update webhook log status
|
|
*/
|
|
private async updateWebhookLog(
|
|
eventId: string,
|
|
status: 'processed' | 'failed',
|
|
errorMessage?: string,
|
|
processingTime?: number
|
|
): Promise<void> {
|
|
const query = `
|
|
UPDATE webhook_logs
|
|
SET status = $1,
|
|
error_message = $2,
|
|
processed_at = NOW(),
|
|
processing_time_ms = $3
|
|
WHERE event_id = $4
|
|
`;
|
|
|
|
await postgresClient.query(query, [status, errorMessage || null, processingTime || null, eventId]);
|
|
}
|
|
|
|
/**
|
|
* Check if webhook is active for entity type
|
|
*/
|
|
private async isWebhookActive(entityType: WebhookEntityType): Promise<boolean> {
|
|
const query = `
|
|
SELECT is_active
|
|
FROM webhook_configs
|
|
WHERE entity_type = $1
|
|
`;
|
|
|
|
const result = await postgresClient.query<{ is_active: boolean }>(query, [entityType]);
|
|
|
|
if (result.rows.length === 0) {
|
|
return false; // No config = disabled
|
|
}
|
|
|
|
return result.rows[0].is_active;
|
|
}
|
|
|
|
/**
|
|
* Get recent webhook logs
|
|
*/
|
|
async getWebhookLogs(limit: number = 100, entityType?: string): Promise<WebhookLog[]> {
|
|
let query = `
|
|
SELECT *
|
|
FROM webhook_logs
|
|
`;
|
|
|
|
const params: any[] = [];
|
|
|
|
if (entityType) {
|
|
query += ` WHERE entity_type = $1`;
|
|
params.push(entityType);
|
|
}
|
|
|
|
query += ` ORDER BY received_at DESC LIMIT $${params.length + 1}`;
|
|
params.push(limit);
|
|
|
|
const result = await postgresClient.query<WebhookLog>(query, params);
|
|
return result.rows;
|
|
}
|
|
|
|
/**
|
|
* Get webhook statistics
|
|
*/
|
|
async getWebhookStats(hours: number = 24): Promise<{
|
|
total: number;
|
|
processed: number;
|
|
failed: number;
|
|
pending: number;
|
|
byEntityType: Record<string, number>;
|
|
}> {
|
|
const query = `
|
|
SELECT
|
|
(SELECT COUNT(*) FROM webhook_logs WHERE received_at >= NOW() - INTERVAL '${hours} hours') as total,
|
|
(SELECT COUNT(*) FROM webhook_logs WHERE received_at >= NOW() - INTERVAL '${hours} hours' AND status = 'processed') as processed,
|
|
(SELECT COUNT(*) FROM webhook_logs WHERE received_at >= NOW() - INTERVAL '${hours} hours' AND status = 'failed') as failed,
|
|
(SELECT COUNT(*) FROM webhook_logs WHERE received_at >= NOW() - INTERVAL '${hours} hours' AND status = 'pending') as pending,
|
|
(SELECT COALESCE(jsonb_object_agg(entity_type, entity_count), '{}'::jsonb) FROM (
|
|
SELECT entity_type, COUNT(*) as entity_count
|
|
FROM webhook_logs
|
|
WHERE received_at >= NOW() - INTERVAL '${hours} hours'
|
|
GROUP BY entity_type
|
|
) ec) as by_entity_type
|
|
`;
|
|
|
|
const result = await postgresClient.query(query);
|
|
|
|
if (result.rows.length === 0) {
|
|
return {
|
|
total: 0,
|
|
processed: 0,
|
|
failed: 0,
|
|
pending: 0,
|
|
byEntityType: {},
|
|
};
|
|
}
|
|
|
|
const row = result.rows[0];
|
|
return {
|
|
total: parseInt(row.total) || 0,
|
|
processed: parseInt(row.processed) || 0,
|
|
failed: parseInt(row.failed) || 0,
|
|
pending: parseInt(row.pending) || 0,
|
|
byEntityType: row.by_entity_type || {},
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Map webhook entity type to internal EntityType
|
|
*/
|
|
private mapWebhookEntityType(webhookType: WebhookEntityType): EntityType {
|
|
const mapping: Record<WebhookEntityType, EntityType> = {
|
|
[WebhookEntityType.COMPANIES]: EntityType.COMPANIES,
|
|
[WebhookEntityType.TICKETS]: EntityType.TICKETS,
|
|
[WebhookEntityType.TICKET_NOTES]: EntityType.TICKET_NOTES,
|
|
[WebhookEntityType.TASKS]: EntityType.TASKS,
|
|
[WebhookEntityType.PROJECTS]: EntityType.PROJECTS,
|
|
[WebhookEntityType.TIME_ENTRIES]: EntityType.TIME_ENTRIES,
|
|
[WebhookEntityType.CONTACTS]: EntityType.CONTACTS,
|
|
[WebhookEntityType.CONTRACTS]: EntityType.CONTRACTS,
|
|
[WebhookEntityType.CONFIGURATION_ITEMS]: EntityType.CONFIGURATION_ITEMS,
|
|
};
|
|
|
|
return mapping[webhookType];
|
|
}
|
|
|
|
/**
|
|
* Get table name from webhook entity type
|
|
*/
|
|
private getTableNameFromWebhookEntity(webhookType: WebhookEntityType): string {
|
|
const entityType = this.mapWebhookEntityType(webhookType);
|
|
return getTableName(entityType);
|
|
}
|
|
|
|
/**
|
|
* Trigger the workflow engine for a new ticket.
|
|
* Runs asynchronously — does not block webhook response.
|
|
*/
|
|
private async triggerWorkflowEngine(payload: AutotaskWebhookPayload): Promise<void> {
|
|
const event: WorkflowEvent = {
|
|
trigger_event: 'ticket.created',
|
|
entity_type: 'ticket',
|
|
entity_id: payload.entityId,
|
|
ticket_number: payload.fields?.ticketNumber || undefined,
|
|
};
|
|
|
|
// If the webhook payload includes the full entity, build TicketData from it
|
|
if (payload.entity) {
|
|
event.ticket_data = {
|
|
id: payload.entityId,
|
|
ticket_number: payload.entity.ticketNumber || null,
|
|
title: payload.entity.title || '',
|
|
description: payload.entity.description || null,
|
|
ticket_category: payload.entity.ticketCategory || null,
|
|
ticket_type: payload.entity.ticketType || null,
|
|
priority: payload.entity.priority || null,
|
|
queue_id: payload.entity.queueID || null,
|
|
issue_type: payload.entity.issueType || null,
|
|
sub_issue_type: payload.entity.subIssueType || null,
|
|
company_id: payload.entity.companyID || 0,
|
|
contact_id: payload.entity.contactID || null,
|
|
assigned_resource_id: payload.entity.assignedResourceID || null,
|
|
creator_resource_id: payload.entity.creatorResourceID || null,
|
|
person_id: payload.personId || null,
|
|
status: payload.entity.status || null,
|
|
source: payload.entity.source || null,
|
|
};
|
|
}
|
|
|
|
console.log(`[WEBHOOK] Triggering workflow engine for ticket ${payload.entityId}`);
|
|
await workflowEngine.process(event);
|
|
}
|
|
}
|
|
|
|
// Export singleton instance
|
|
export const webhookService = new WebhookService();
|