Spaces:
Runtime error
Runtime error
| import { | |
| type IntegrationDefinitionV2, | |
| type IntegrationAccount, | |
| } from "@core/database"; | |
| import { IntegrationEventType } from "@redplanethq/sdk"; | |
| import { prisma } from "~/db.server"; | |
| import { logger } from "./logger.service"; | |
| import { runIntegrationTrigger } from "./integration.server"; | |
| export type EventHeaders = Record<string, string | string[]>; | |
| export type EventBody = Record<string, any>; | |
| export class WebhookService { | |
| async handleEvents( | |
| sourceName: string, | |
| integrationAccountId: string | undefined, | |
| eventHeaders: EventHeaders, | |
| eventBody: EventBody, | |
| ): Promise<{ challenge?: string; status: string }> { | |
| logger.log(`Received webhook ${sourceName}`, { | |
| where: "WebhookService.handleEvents", | |
| }); | |
| let integrationAccount: | |
| | (IntegrationAccount & { | |
| integrationDefinition: IntegrationDefinitionV2; | |
| }) | |
| | null = null; | |
| if (!integrationAccountId) { | |
| // Find integration account by identifying the webhook account | |
| const integrationDefinition = | |
| await prisma.integrationDefinitionV2.findFirst({ | |
| where: { slug: sourceName, deleted: null }, | |
| }); | |
| if (integrationDefinition) { | |
| try { | |
| const identifyResponse = await runIntegrationTrigger( | |
| integrationDefinition, | |
| { | |
| event: IntegrationEventType.IDENTIFY, | |
| eventBody: { | |
| eventHeaders, | |
| event: { ...eventBody }, | |
| }, | |
| }, | |
| ); | |
| let accountId: string | undefined; | |
| // Handle new CLI message format response | |
| if (identifyResponse?.success && identifyResponse?.result) { | |
| // Check if there are identifiers in the response | |
| if ( | |
| identifyResponse.result.identifiers && | |
| identifyResponse.result.identifiers.length > 0 | |
| ) { | |
| accountId = identifyResponse.result.identifiers[0].id; | |
| } else if ( | |
| identifyResponse.result.activities && | |
| identifyResponse.result.activities.length > 0 | |
| ) { | |
| // Sometimes the account ID might be in activities data | |
| const firstActivity = identifyResponse.result.activities[0]; | |
| accountId = firstActivity.accountId || firstActivity.id; | |
| } else { | |
| // Check raw output for backward compatibility | |
| accountId = identifyResponse.rawOutput?.trim(); | |
| } | |
| } else if (identifyResponse?.error) { | |
| logger.warn("Integration IDENTIFY command failed", { | |
| error: identifyResponse.error, | |
| sourceName, | |
| }); | |
| } else { | |
| // Handle legacy response format for backward compatibility | |
| if ( | |
| identifyResponse?.message?.startsWith("The event payload type is") | |
| ) { | |
| accountId = undefined; | |
| } else { | |
| accountId = identifyResponse; | |
| } | |
| } | |
| if (accountId) { | |
| integrationAccount = await prisma.integrationAccount.findFirst({ | |
| where: { accountId }, | |
| include: { integrationDefinition: true }, | |
| }); | |
| logger.info("Found integration account for webhook", { | |
| accountId, | |
| integrationAccountId: integrationAccount?.id, | |
| sourceName, | |
| }); | |
| } else { | |
| logger.warn("No account ID found from IDENTIFY command", { | |
| sourceName, | |
| response: identifyResponse, | |
| }); | |
| } | |
| } catch (error) { | |
| logger.error("Failed to identify integration account", { | |
| error, | |
| sourceName, | |
| }); | |
| } | |
| } | |
| } else { | |
| integrationAccount = await prisma.integrationAccount.findUnique({ | |
| where: { id: integrationAccountId }, | |
| include: { integrationDefinition: true }, | |
| }); | |
| } | |
| if (integrationAccount) { | |
| try { | |
| logger.info(`Processing webhook for ${sourceName}`, { | |
| integrationAccountId: integrationAccount.id, | |
| integrationSlug: integrationAccount.integrationDefinition.slug, | |
| }); | |
| const processResponse = await runIntegrationTrigger( | |
| integrationAccount.integrationDefinition, | |
| { | |
| event: IntegrationEventType.PROCESS, | |
| eventBody: { | |
| eventHeaders, | |
| eventData: { ...eventBody }, | |
| }, | |
| }, | |
| integrationAccount.integratedById, | |
| integrationAccount.workspaceId, | |
| integrationAccount, | |
| ); | |
| if (processResponse?.success) { | |
| logger.log(`Successfully processed webhook for ${sourceName}`, { | |
| integrationAccountId: integrationAccount.id, | |
| activitiesCreated: processResponse.result?.activities?.length || 0, | |
| messagesProcessed: processResponse.messages?.length || 0, | |
| }); | |
| } else { | |
| logger.warn(`Webhook processing had issues for ${sourceName}`, { | |
| integrationAccountId: integrationAccount.id, | |
| error: processResponse?.error, | |
| success: processResponse?.success, | |
| }); | |
| } | |
| } catch (error) { | |
| logger.error(`Failed to process webhook for ${sourceName}`, { | |
| error, | |
| integrationAccountId: integrationAccount.id, | |
| }); | |
| } | |
| } else { | |
| logger.log( | |
| `Could not find integration account for webhook ${sourceName}`, | |
| { | |
| where: "WebhookService.handleEvents", | |
| }, | |
| ); | |
| } | |
| return { status: "acknowledged" }; | |
| } | |
| } | |
| export const webhookService = new WebhookService(); | |