Webhook events and delivery

BLUESTONE_AUTH_BASE="https://idp.test.bluestonepim.com/op/token"
BLUESTONE_API_BASE="https://api.test.bluestonepim.com"
CLIENT_ID="YOUR_CLIENT_ID"
CLIENT_SECRET="YOUR_CLIENT_SECRET"
WEBHOOK_URL="https://example.ngrok-free.app/webhooks/bluestone"
WEBHOOK_SECRET="LongAndSecretPassword"

ACCESS_TOKEN=$(
  curl -sS -X POST "$BLUESTONE_AUTH_BASE" \
    -H "Content-Type: application/x-www-form-urlencoded" \
    --data-urlencode "grant_type=client_credentials" \
    --data-urlencode "client_id=$CLIENT_ID" \
    --data-urlencode "client_secret=$CLIENT_SECRET" |
  jq -r ".access_token"
)

curl -sS -X POST "$BLUESTONE_API_BASE/notification-external/webhooks" \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"active\": true,
    \"url\": \"$WEBHOOK_URL\",
    \"secret\": \"$WEBHOOK_SECRET\"
  }"

WEBHOOK_ID="YOUR_WEBHOOK_ID"

curl -sS -X PUT "$BLUESTONE_API_BASE/notification-external/subscriptions/webhook/$WEBHOOK_ID/events" \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "eventTypes": [
      "PRODUCT_CREATED",
      "PRODUCT_WATCH_ATTRIBUTE_UPDATE_VALUE",
      "PRODUCT_SYNC_DONE"
    ]
  }'

curl -sS "$BLUESTONE_API_BASE/notification-external/subscriptions/webhook/$WEBHOOK_ID/events" \
  -H "Authorization: Bearer $ACCESS_TOKEN"

{
  "eventTypes": ["PRODUCT_CREATED", "PRODUCT_SYNC_DONE"]
}

{
  "timestamp": 1698151920627,
  "events": [
    {
      "changes": {
        "eventType": "PRODUCT_CREATED",
        "entityIds": ["6537bdebb388e74a0482e761"],
        "metadataChanges": [
          {
            "field": "NAME",
            "oldValue": null,
            "newValue": "Product A",
            "context": "en"
          },
          {
            "field": "NUMBER",
            "oldValue": null,
            "newValue": "product-a",
            "context": "en"
          },
          {
            "field": "TYPE",
            "oldValue": null,
            "newValue": "SINGLE",
            "context": "en"
          }
        ]
      }
    }
  ]
}

{
  "timestamp": 1698151295719,
  "events": [
    {
      "changes": {
        "eventType": "PRODUCT_WATCH_ATTRIBUTE_UPDATE_VALUE",
        "entityIds": ["641314b6ee358800012b279b"],
        "attributeChange": {
          "changeType": "UPDATE",
          "attributeType": "text",
          "attributeId": "5da7254de21b84000c6ed075",
          "attributeOldValue": "Red",
          "attributeNewValue": "Blue",
          "context": "en",
          "contextId": null,
          "affectedContextIds": ["en", "no", "pl"]
        }
      }
    }
  ]
}

{
  "timestamp": 1698152131200,
  "events": [
    {
      "changes": {
        "eventType": "PRODUCT_SYNC_DONE",
        "entityIds": [],
        "syncDoneData": {
          "field": "PAPI_SYNC_ID",
          "value": "6537bebdfbde8a0013911d23",
          "context": "en"
        }
      }
    }
  ]
}

{
  "timestamp": 1698225320204,
  "events": [
    {
      "changes": {
        "eventType": "CATEGORY_CREATED",
        "entityIds": ["6538dca57566616af2206608"],
        "name": "Shoes",
        "number": "6538dca57566616af2206608",
        "parentId": "633d67fed6018000014074a1"
      }
    }
  ]
}

curl -sS -X POST "$BLUESTONE_API_BASE/notification-external/messages/webhook/list" \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "webhookId": "YOUR_WEBHOOK_ID",
    "status": "ERROR",
    "page": 0,
    "pageSize": 10
  }'

import crypto from "node:crypto";
import express from "express";

const app = express();
const WEBHOOK_SECRET = process.env.BLUESTONE_WEBHOOK_SECRET ?? "";
const processedEventKeys = new Set<string>();

app.use("/webhooks/bluestone", express.raw({ type: "application/json" }));

function normalizeSignature(signature: string): string {
  return signature.startsWith("sha256=") ? signature.slice("sha256=".length) : signature;
}

function verifySignature(rawBody: Buffer, signatureHeader: string, secret: string): boolean {
  if (!signatureHeader || !secret) {
    return false;
  }

  const expected = crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  const actual = normalizeSignature(signatureHeader);
  const expectedBuffer = Buffer.from(expected, "hex");
  const actualBuffer = Buffer.from(actual, "hex");

  return (
    expectedBuffer.length === actualBuffer.length &&
    crypto.timingSafeEqual(expectedBuffer, actualBuffer)
  );
}

app.post("/webhooks/bluestone", (req, res) => {
  const rawBody = req.body as Buffer;
  const signature = req.header("x-bs-signature") ?? "";

  if (!verifySignature(rawBody, signature, WEBHOOK_SECRET)) {
    res.sendStatus(401);
    return;
  }

  const payload = JSON.parse(rawBody.toString("utf8")) as {
    timestamp: number;
    events: Array<{ changes: { eventType: string; entityIds?: string[] } }>;
  };

  payload.events.forEach((event, index) => {
    const eventKey = crypto
      .createHash("sha256")
      .update(`${payload.timestamp}:${index}:${JSON.stringify(event.changes)}`)
      .digest("hex");

    if (processedEventKeys.has(eventKey)) {
      return;
    }

    processedEventKeys.add(eventKey);

    console.log("Received Bluestone event", {
      eventType: event.changes.eventType,
      entityIds: event.changes.entityIds ?? [],
      timestamp: payload.timestamp,
    });

    // Queue work here, then fetch current entity state from the Management API if needed.
  });

  res.sendStatus(204);
});

app.listen(3000, () => {
  console.log("Listening on http://localhost:3000/webhooks/bluestone");
});

import hashlib
import hmac
import json
import os
from typing import Annotated

from fastapi import FastAPI, Header, HTTPException, Request, Response

app = FastAPI()
WEBHOOK_SECRET = os.environ["BLUESTONE_WEBHOOK_SECRET"].encode("utf-8")
processed_event_keys: set[str] = set()


def normalize_signature(signature: str) -> str:
    return signature.removeprefix("sha256=")


def verify_signature(raw_body: bytes, signature_header: str) -> bool:
    expected = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    actual = normalize_signature(signature_header)
    return hmac.compare_digest(expected, actual)


@app.post("/webhooks/bluestone")
async def bluestone_webhook(
    request: Request,
    x_bs_signature: Annotated[str | None, Header(alias="x-bs-signature")] = None,
) -> Response:
    raw_body = await request.body()

    if not x_bs_signature or not verify_signature(raw_body, x_bs_signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = json.loads(raw_body)
    timestamp = payload["timestamp"]

    for index, event in enumerate(payload.get("events", [])):
        changes = event["changes"]
        event_key = hashlib.sha256(
            f"{timestamp}:{index}:{json.dumps(changes, sort_keys=True)}".encode("utf-8")
        ).hexdigest()

        if event_key in processed_event_keys:
            continue

        processed_event_keys.add(event_key)

        print(
            "Received Bluestone event",
            {
                "eventType": changes["eventType"],
                "entityIds": changes.get("entityIds", []),
                "timestamp": timestamp,
            },
        )

        # Queue work here, then fetch current entity state from the Management API if needed.

    return Response(status_code=204)

curl -sS -X POST "$BLUESTONE_API_BASE/notification-external/messages/webhook" \
  -G \
  --data-urlencode "url=$WEBHOOK_URL" \
  --data-urlencode "secret=$WEBHOOK_SECRET" \
  -H "Authorization: Bearer $ACCESS_TOKEN"