wulf-pulse/lib/services/webhook-service.ts

/**
 * Webhook Service
 * Handles incoming webhooks from Autotask for real-time updates
 */

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 } from '../utils/sync-helpers';

export class WebhookService {
  /**
   * 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`);
      
      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
    if (payload.entity) {
      await this.upsertEntity(payload.entityType, payload.entity);
      return payload.eventType === WebhookEventType.CREATE ? 'created' : 'updated';
    }
    
    // Otherwise, fetch the entity from Autotask API
    // Note: This requires the AutotaskClient to fetch individual entities
    // For now, we'll log and skip - can be enhanced later
    console.warn(`[WEBHOOK] Entity data not included in webhook, skipping upsert for ${payload.entityType} #${payload.entityId}`);
    return '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
        COUNT(*) as total,
        COUNT(*) FILTER (WHERE status = 'processed') as processed,
        COUNT(*) FILTER (WHERE status = 'failed') as failed,
        COUNT(*) FILTER (WHERE status = 'pending') as pending,
        jsonb_object_agg(entity_type, entity_count) as by_entity_type
      FROM (
        SELECT
          entity_type,
          COUNT(*) as entity_count
        FROM webhook_logs
        WHERE received_at >= NOW() - INTERVAL '${hours} hours'
        GROUP BY entity_type
      ) entity_counts,
      webhook_logs
      WHERE received_at >= NOW() - INTERVAL '${hours} hours'
    `;
    
    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.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);
  }
}

// Export singleton instance
export const webhookService = new WebhookService();