-
Notifications
You must be signed in to change notification settings - Fork 196
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
13 changed files
with
601 additions
and
69 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| import { PrismaClient } from '@prisma/client'; | ||
| import { v4 as uuidv4 } from 'uuid'; | ||
|
|
||
| const prisma = new PrismaClient(); | ||
|
|
||
| async function main() { | ||
| const webhook_endpoint = await prisma.webhook_endpoints.create({ | ||
| data: { | ||
| id_webhook_endpoint: uuidv4(), | ||
| url: 'https://webhook.site/5018c5f3-e582-4f6f-a5ca-9e7389e951e2', | ||
| secret: '12345679', | ||
| active: true, | ||
| created_at: new Date(), | ||
| id_project: '801f9ede-c698-4e66-a7fc-48d19eebaa4f', | ||
| scope: 'crm.contact.synced', | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| main() | ||
| .catch((e) => { | ||
| console.error(e); | ||
| process.exit(1); | ||
| }) | ||
| .finally(async () => { | ||
| await prisma.$disconnect(); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| import { BullModule } from '@nestjs/bull'; | ||
| import { Module } from '@nestjs/common'; | ||
| import { WebhookService } from './webhook.service'; | ||
| import { PrismaService } from '@@core/prisma/prisma.service'; | ||
| import { LoggerService } from '@@core/logger/logger.service'; | ||
| import { WebhookProcessor } from './webhook.processor'; | ||
|
|
||
| @Module({ | ||
| imports: [ | ||
| BullModule.registerQueue({ | ||
| name: 'webhookDelivery', | ||
| }), | ||
| ], | ||
| providers: [WebhookService, PrismaService, LoggerService, WebhookProcessor], | ||
| }) | ||
| export class WebhookModule {} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| import { LoggerService } from '@@core/logger/logger.service'; | ||
| import { PrismaService } from '@@core/prisma/prisma.service'; | ||
| import { OnQueueActive, Process, Processor } from '@nestjs/bull'; | ||
| import axios from 'axios'; | ||
| import { Job } from 'bull'; | ||
| import { v4 as uuidv4 } from 'uuid'; | ||
| import { WebhookService } from './webhook.service'; | ||
|
|
||
| @Processor('webhookDelivery') | ||
| export class WebhookProcessor { | ||
| constructor( | ||
| private logger: LoggerService, | ||
| private prisma: PrismaService, | ||
| private webhookService: WebhookService, | ||
| ) { | ||
| this.logger.setContext(WebhookProcessor.name); | ||
| } | ||
|
|
||
| @OnQueueActive() | ||
| onActive(job: Job) { | ||
| this.logger.log(`Processing job ${job.id} ...`); | ||
| } | ||
|
|
||
| @Process({ concurrency: 5 }) | ||
| async processWebhooks(job: Job) { | ||
| const id_webhook_delivery = job.data.webhook_delivery_id; | ||
|
|
||
| this.logger.log(`Start delivering webhook id ${id_webhook_delivery}...`); | ||
|
|
||
| await this.prisma.webhook_delivery_attempts.update({ | ||
| where: { id_webhook_delivery_attempt: id_webhook_delivery }, | ||
| data: { | ||
| status: 'processed', | ||
| }, | ||
| }); | ||
|
|
||
| // Retrieve the webhook delivery attempt | ||
| const deliveryAttempt = | ||
| await this.prisma.webhook_delivery_attempts.findUnique({ | ||
| where: { id_webhook_delivery_attempt: id_webhook_delivery }, | ||
| include: { | ||
| webhook_endpoints: true, | ||
| webhooks_payloads: true, | ||
| }, | ||
| }); | ||
|
|
||
| // Check if the endpoint is active | ||
| if (deliveryAttempt.webhook_endpoints.active) { | ||
| try { | ||
| // Send the payload to the endpoint URL | ||
| const response = await axios.post( | ||
| deliveryAttempt.webhook_endpoints.url, | ||
| { | ||
| id_event: deliveryAttempt.id_event, | ||
| data: deliveryAttempt.webhooks_payloads.data, | ||
| }, | ||
| ); | ||
|
|
||
| // Populate the webhooks_responses table | ||
| await this.prisma.webhooks_reponses.create({ | ||
| data: { | ||
| id_webhooks_reponse: uuidv4(), | ||
| http_response_data: response.data, | ||
| http_status_code: response.status.toString(), | ||
| }, | ||
| }); | ||
| await this.prisma.webhook_delivery_attempts.update({ | ||
| where: { id_webhook_delivery_attempt: id_webhook_delivery }, | ||
| data: { | ||
| status: 'success', | ||
| }, | ||
| }); | ||
|
|
||
| this.logger.log('Webhook delivered !'); | ||
| } catch (error) { | ||
| // If the POST request fails, set a next retry time and reinsert the job in the queue | ||
| const nextRetry = new Date(); | ||
| nextRetry.setSeconds(nextRetry.getSeconds() + 60); // Retry after 60 seconds | ||
|
|
||
| await this.prisma.webhook_delivery_attempts.update({ | ||
| where: { id_webhook_delivery_attempt: id_webhook_delivery }, | ||
| data: { | ||
| status: 'failed', | ||
| next_retry: nextRetry, | ||
| }, | ||
| }); | ||
|
|
||
| //re-insert the webhook in the queue | ||
| await this.webhookService.handleFailedWebhook(id_webhook_delivery); | ||
|
|
||
| this.logger.log( | ||
| 'Webhook delivery failed. Job reinserted in the queue for retry.', | ||
| ); | ||
| } | ||
| } else { | ||
| this.logger.log('Webhook endpoint is not active. Delivery skipped.'); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| import { Injectable } from '@nestjs/common'; | ||
| import { Queue } from 'bull'; | ||
| import { InjectQueue } from '@nestjs/bull'; | ||
| import { PrismaService } from '@@core/prisma/prisma.service'; | ||
| import { v4 as uuidv4 } from 'uuid'; | ||
| import { LoggerService } from '@@core/logger/logger.service'; | ||
| import { handleServiceError } from '@@core/utils/errors'; | ||
|
|
||
| @Injectable() | ||
| export class WebhookService { | ||
| constructor( | ||
| @InjectQueue('webhookDelivery') private queue: Queue, | ||
| private prisma: PrismaService, | ||
| private logger: LoggerService, | ||
| ) { | ||
| this.logger.setContext(WebhookService.name); | ||
| } | ||
|
|
||
| async handleWebhook( | ||
| data: any, | ||
| eventType: string, | ||
| projectId: string, | ||
| eventId: string, | ||
| ) { | ||
| try { | ||
| //this.logger.log('data any type is ' + data); | ||
| //just create an entry in webhook | ||
| //search if an endpoint webhook exists for such a projectId and such a scope | ||
| const webhook = await this.prisma.webhook_endpoints.findFirst({ | ||
| where: { | ||
| id_project: projectId, | ||
| active: true, | ||
| scope: eventType, //todo | ||
| }, | ||
| }); | ||
| if (!webhook) return; | ||
|
|
||
| const w_payload = await this.prisma.webhooks_payloads.create({ | ||
| data: { | ||
| id_webhooks_payload: uuidv4(), | ||
| data: JSON.stringify(data), | ||
| }, | ||
| }); | ||
|
|
||
| const w_delivery = await this.prisma.webhook_delivery_attempts.create({ | ||
| data: { | ||
| id_webhook_delivery_attempt: uuidv4(), | ||
| id_event: eventId, | ||
| timestamp: new Date(), | ||
| id_webhook_endpoint: webhook.id_webhook_endpoint, | ||
| status: 'queued', // queued | processed | failed | success | ||
| id_webhooks_payload: w_payload.id_webhooks_payload, | ||
| }, | ||
| }); | ||
|
|
||
| // we send the delivery webhook to the queue so it can be processed by our dispatcher worker | ||
| const job = await this.queue.add({ | ||
| webhook_delivery_id: w_delivery.id_webhook_delivery_attempt, | ||
| }); | ||
| } catch (error) { | ||
| handleServiceError(error, this.logger); | ||
| } | ||
| } | ||
|
|
||
| async handleFailedWebhook(failed_id_delivery_webhook: string) { | ||
| try { | ||
| await this.queue.add( | ||
| { | ||
| webhook_delivery_id: failed_id_delivery_webhook, | ||
| }, | ||
| { delay: 60000 }, | ||
| ); | ||
| } catch (error) { | ||
| handleServiceError(error, this.logger); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.