webhooks

import { Queue } from "bullmq";
import { createHash } from "crypto";

export const connection = { host: "127.0.0.1", port: 6379 };

export interface ChargePayload {

module Webhooks
  class StripeController < ApplicationController
    skip_before_action :verify_authenticity_token

    def create
      payload = request.body.read

import hashlib
import hmac
import time


class SignatureError(Exception):

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';

async function bootstrap() {
  // rawBody: true preserves the exact bytes Stripe signed
  const app = await NestFactory.create(AppModule, { rawBody: true });

const crypto = require('crypto');

function parseSignatureHeader(header) {
  const parts = {};
  for (const segment of String(header || '').split(',')) {
    const [key, value] = segment.split('=');

import datetime as dt

from sqlalchemy import Column, DateTime, Integer, String, UniqueConstraint
from sqlalchemy.orm import declarative_base

Base = declarative_base()

from datetime import datetime
from app.extensions import db


class IdempotencyKey(db.Model):
    __tablename__ = "idempotency_keys"

from django.db import models
from django.utils import timezone


class WebhookEvent(models.Model):
    class Status(models.TextChoices):

<?php

namespace App\Controller;

use App\Message\ProcessStripeEvent;
use App\Webhook\StripeSignatureVerifier;

class CreateWebhookEvents < ActiveRecord::Migration[7.1]
  def change
    create_table :webhook_events do |t|
      t.string :event_id, null: false
      t.string :source, null: false, default: "stripe"
      t.string :event_type, null: false

class SyncContactJob < ApplicationJob
  queue_as :external

  BACKOFF = ->(executions) do
    (2**executions) + rand(0.0..1.0) # exponential + jitter, in seconds
  end

const express = require('express');
const webhookRouter = require('./webhookRouter');
const apiRouter = require('./apiRouter');

const app = express();