feat(events-backend,events-node): add a parser option to events by content type

Signed-off-by: Rogerio Angeliski <angeliski@hotmail.com>

.changeset/ten-impalas-hope.md

@@ -0,0 +1,6 @@
+---
+'@backstage/plugin-events-backend': minor
+'@backstage/plugin-events-node': minor
+---
+
+add `addHttpPostBodyParser` to events extension to allow body parse customization. This feature will enhance flexibility in handling HTTP POST requests in event-related operations.

plugins/events-backend/README.md

@@ -140,6 +140,48 @@ export const eventsModuleYourFeature = createBackendModule({
 });
 ```
 
+### Request Body Parse
+
+We need to parse the request body before we can validate it. We have some default parsers but you can provide your own when necessary.
+
+```ts
+import { eventsExtensionPoint } from '@backstage/plugin-events-node/alpha';
+
+// [...]
+
+export const eventsModuleYourFeature = createBackendModule({
+  pluginId: 'events',
+  moduleId: 'your-feature',
+  register(env) {
+    // [...]
+    env.registerInit({
+      deps: {
+        // [...]
+        events: eventsExtensionPoint,
+        // [...]
+      },
+      async init({ /* ... */ events /*, ... */ }) {
+        // [...]
+        events.addHttpPostBodyParser({
+          contentType: 'application/x-www-form-urlencoded',
+          parser: async (req, _topic) => {
+            return {
+              bodyParsed: req.body.toString('utf-8'),
+              bodyBuffer: req.body,
+              encoding: 'utf-8',
+            };
+          },
+        });
+      },
+    });
+  },
+});
+```
+
+We have the following default parsers:
+
+- `application/json`
+
 #### Legacy Backend System
 
 ```ts

plugins/events-backend/report.api.md

@@ -11,6 +11,7 @@ import { EventPublisher } from '@backstage/plugin-events-node';
 import { EventsService } from '@backstage/plugin-events-node';
 import { EventSubscriber } from '@backstage/plugin-events-node';
 import express from 'express';
+import { HttpBodyParser } from '@backstage/plugin-events-node';
 import { HttpPostIngressOptions } from '@backstage/plugin-events-node';
 import { Logger } from 'winston';
 import { LoggerService } from '@backstage/backend-plugin-api';
@@ -58,6 +59,9 @@ export class HttpPostIngressEventPublisher {
     ingresses?: {
       [topic: string]: Omit<HttpPostIngressOptions, 'topic'>;
     };
+    bodyParsers?: {
+      [contentType: string]: HttpBodyParser;
+    };
     logger: LoggerService;
   }): HttpPostIngressEventPublisher;
 }

plugins/events-backend/src/service/EventsPlugin.ts

@@ -24,6 +24,7 @@ import {
 } from '@backstage/plugin-events-node/alpha';
 import {
   eventsServiceRef,
+  HttpBodyParserOptions,
   HttpPostIngressOptions,
 } from '@backstage/plugin-events-node';
 import Router from 'express-promise-router';
@@ -31,7 +32,8 @@ import { HttpPostIngressEventPublisher } from './http';
 import { createEventBusRouter } from './hub';
 
 class EventsExtensionPointImpl implements EventsExtensionPoint {
-  #httpPostIngresses: HttpPostIngressOptions[] = [];
+  readonly #httpPostIngresses: HttpPostIngressOptions[] = [];
+  readonly #httpBodyParsers: HttpBodyParserOptions[] = [];
 
   setEventBroker(_: any): void {
     throw new Error(
@@ -55,9 +57,17 @@ class EventsExtensionPointImpl implements EventsExtensionPoint {
     this.#httpPostIngresses.push(options);
   }
 
+  addHttpPostBodyParser(options: HttpBodyParserOptions): void {
+    this.#httpBodyParsers.push(options);
+  }
+
   get httpPostIngresses() {
     return this.#httpPostIngresses;
   }
+
+  get httpBodyParsers() {
+    return this.#httpBodyParsers;
+  }
 }
 
 /**
@@ -99,10 +109,18 @@ export const eventsPlugin = createBackendPlugin({
           ]),
         );
 
+        const bodyParsers = Object.fromEntries(
+          extensionPoint.httpBodyParsers.map(option => [
+            option.contentType,
+            option.parser,
+          ]),
+        );
+
         const http = HttpPostIngressEventPublisher.fromConfig({
           config,
           events,
           ingresses,
+          bodyParsers,
           logger,
         });
         const eventsRouter = Router();

plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts

@@ -22,6 +22,7 @@ import request from 'supertest';
 import { HttpPostIngressEventPublisher } from './HttpPostIngressEventPublisher';
 import { mockServices } from '@backstage/backend-test-utils';
 import { MiddlewareFactory } from '@backstage/backend-defaults/rootHttpRouter';
+import { HttpBodyParser } from '@backstage/plugin-events-node';
 
 const middleware = MiddlewareFactory.create({
   logger: mockServices.logger.mock(),
@@ -207,7 +208,8 @@ describe('HttpPostIngressEventPublisher', () => {
     expect(response.body).toEqual(
       expect.objectContaining({
         error: {
-          message: 'Unsupported media type: text/plain',
+          message:
+            'Unsupported media type: text/plain. You need to provide a custom body parser for this media type using events extension.',
           name: 'UnsupportedMediaTypeError',
           statusCode: 415,
         },
@@ -217,6 +219,89 @@ describe('HttpPostIngressEventPublisher', () => {
     );
   });
 
+  it('with a text/plain body parser implementation', async () => {
+    const config = new ConfigReader({
+      events: {
+        http: {
+          topics: ['testA'],
+        },
+      },
+    });
+
+    const router = Router();
+    const app = express().use(router);
+    const events = new TestEventsService();
+
+    const bodyParser: HttpBodyParser = async (req, _topic) => {
+      return {
+        bodyParsed: req.body.toString('utf-8'),
+        bodyBuffer: req.body,
+        encoding: 'utf-8',
+      };
+    };
+
+    const publisher = HttpPostIngressEventPublisher.fromConfig({
+      config,
+      events,
+      bodyParsers: {
+        'text/plain': bodyParser,
+      },
+      logger,
+    });
+    publisher.bind(router);
+    router.use(middleware.error());
+
+    const response = await request(app)
+      .post('/http/testA')
+      .type('text/plain')
+      .timeout(1000)
+      .send('Textual information');
+    expect(response.status).toBe(202);
+  });
+
+  it('with a custom application/json body parser implementation', async () => {
+    const config = new ConfigReader({
+      events: {
+        http: {
+          topics: ['testA'],
+        },
+      },
+    });
+
+    const router = Router();
+    const app = express().use(router);
+    const events = new TestEventsService();
+    const customParse = jest.fn();
+
+    const bodyParser: HttpBodyParser = async (req, _topic) => {
+      customParse();
+      return {
+        bodyParsed: JSON.parse(req.body.toString('utf-8')),
+        bodyBuffer: req.body,
+        encoding: 'utf-8',
+      };
+    };
+
+    const publisher = HttpPostIngressEventPublisher.fromConfig({
+      config,
+      events,
+      bodyParsers: {
+        'application/json': bodyParser,
+      },
+      logger,
+    });
+    publisher.bind(router);
+    router.use(middleware.error());
+
+    const response = await request(app)
+      .post('/http/testA')
+      .type('application/json')
+      .timeout(1000)
+      .send(JSON.stringify({ testA: 'data' }));
+    expect(response.status).toBe(202);
+    expect(customParse).toHaveBeenCalled();
+  });
+
   it('with validator', async () => {
     const config = new ConfigReader({
       events: {

plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts

@@ -16,35 +16,18 @@
 
 import { LoggerService } from '@backstage/backend-plugin-api';
 import { Config } from '@backstage/config';
-import { CustomErrorBase } from '@backstage/errors';
 import {
   EventsService,
+  HttpBodyParser,
   HttpPostIngressOptions,
   RequestValidator,
 } from '@backstage/plugin-events-node';
 import contentType from 'content-type';
 import express from 'express';
 import Router from 'express-promise-router';
+import { defaultHttpBodyParsers } from './body-parser';
 import { RequestValidationContextImpl } from './validation';
-
-class UnsupportedCharsetError extends CustomErrorBase {
-  name = 'UnsupportedCharsetError' as const;
-  statusCode = 415 as const;
-
-  constructor(charset: string) {
-    super(`Unsupported charset: ${charset}`);
-  }
-}
-
-class UnsupportedMediaTypeError extends CustomErrorBase {
-  name = 'UnsupportedMediaTypeError' as const;
-  statusCode = 415 as const;
-
-  constructor(mediaType?: string) {
-    super(`Unsupported media type: ${mediaType ?? 'unknown'}`);
-  }
-}
-
+import { UnsupportedMediaTypeError } from './errors';
 /**
  * Publishes events received from their origin (e.g., webhook events from an SCM system)
  * via HTTP POST endpoint and passes the request body as event payload to the registered subscribers.
@@ -57,6 +40,7 @@ export class HttpPostIngressEventPublisher {
     config: Config;
     events: EventsService;
     ingresses?: { [topic: string]: Omit<HttpPostIngressOptions, 'topic'> };
+    bodyParsers?: { [contentType: string]: HttpBodyParser };
     logger: LoggerService;
   }): HttpPostIngressEventPublisher {
     const topics =
@@ -71,7 +55,14 @@ export class HttpPostIngressEventPublisher {
       }
     });
 
-    return new HttpPostIngressEventPublisher(env.events, env.logger, ingresses);
+    const parsers = { ...defaultHttpBodyParsers, ...env.bodyParsers };
+
+    return new HttpPostIngressEventPublisher(
+      env.events,
+      env.logger,
+      ingresses,
+      parsers,
+    );
   }
 
   private constructor(
@@ -80,6 +71,9 @@ export class HttpPostIngressEventPublisher {
     private readonly ingresses: {
       [topic: string]: Omit<HttpPostIngressOptions, 'topic'>;
     },
+    private readonly bodyParsers: {
+      [contentType: string]: HttpBodyParser;
+    },
   ) {}
 
   bind(router: express.Router): void {
@@ -108,32 +102,17 @@ export class HttpPostIngressEventPublisher {
     const logger = this.logger;
 
     router.post(path, async (request, response) => {
-      const requestBody = request.body;
-      if (!Buffer.isBuffer(requestBody)) {
-        throw new Error(
-          `Failed to retrieve raw body from incoming event for topic ${topic}; not a buffer: ${typeof requestBody}`,
-        );
+      const requestContentType = contentType.parse(request);
+      const bodyParser = this.bodyParsers[requestContentType.type];
+
+      if (!bodyParser) {
+        throw new UnsupportedMediaTypeError(requestContentType.type);
       }
 
-      const bodyBuffer: Buffer = requestBody;
-      const parsedContentType = contentType.parse(request);
-      if (
-        !parsedContentType.type ||
-        parsedContentType.type !== 'application/json'
-      ) {
-        throw new UnsupportedMediaTypeError(parsedContentType.type);
-      }
-
-      const encoding = parsedContentType.parameters.charset ?? 'utf-8';
-      if (!Buffer.isEncoding(encoding)) {
-        throw new UnsupportedCharsetError(encoding);
-      }
-
-      const bodyString = bodyBuffer.toString(encoding);
-      const bodyParsed =
-        parsedContentType.type === 'application/json'
-          ? JSON.parse(bodyString)
-          : bodyString;
+      const { bodyParsed, bodyBuffer, encoding } = await bodyParser(
+        request,
+        topic,
+      );
 
       if (validator) {
         const requestDetails = {

plugins/events-backend/src/service/http/body-parser/HttpApplicationJsonBodyParser.ts

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { HttpBodyParser } from '@backstage/plugin-events-node';
+import contentType from 'content-type';
+import { UnsupportedCharsetError } from '../errors';
+
+export const HttpApplicationJsonBodyParser: HttpBodyParser = async (
+  request,
+  topic,
+) => {
+  const requestBody = request.body;
+  if (!Buffer.isBuffer(requestBody)) {
+    throw new Error(
+      `Failed to retrieve raw body from incoming event for topic ${topic}; not a buffer: ${typeof requestBody}`,
+    );
+  }
+
+  const bodyBuffer: Buffer = requestBody;
+  const parsedContentType = contentType.parse(request);
+
+  const encoding = parsedContentType.parameters.charset ?? 'utf-8';
+  if (!Buffer.isEncoding(encoding)) {
+    throw new UnsupportedCharsetError(encoding);
+  }
+
+  const bodyString = bodyBuffer.toString(encoding);
+  const bodyParsed =
+    parsedContentType.type === 'application/json'
+      ? JSON.parse(bodyString)
+      : bodyString;
+  return { bodyParsed, bodyBuffer, encoding };
+};

plugins/events-backend/src/service/http/body-parser/index.ts

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { HttpBodyParser } from '@backstage/plugin-events-node';
+import { HttpApplicationJsonBodyParser } from './HttpApplicationJsonBodyParser';
+
+export const defaultHttpBodyParsers: { [contentType: string]: HttpBodyParser } =
+  {
+    'application/json': HttpApplicationJsonBodyParser,
+  };

plugins/events-backend/src/service/http/errors.ts

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { CustomErrorBase } from '@backstage/errors';
+
+export class UnsupportedCharsetError extends CustomErrorBase {
+  name = 'UnsupportedCharsetError' as const;
+  statusCode = 415 as const;
+
+  constructor(charset: string) {
+    super(`Unsupported charset: ${charset}`);
+  }
+}
+
+export class UnsupportedMediaTypeError extends CustomErrorBase {
+  name = 'UnsupportedMediaTypeError' as const;
+  statusCode = 415 as const;
+
+  constructor(mediaType?: string) {
+    super(
+      `Unsupported media type: ${
+        mediaType ?? 'unknown'
+      }. You need to provide a custom body parser for this media type using events extension.`,
+    );
+  }
+}

plugins/events-node/package.json

@@ -55,7 +55,9 @@
     "@backstage/backend-plugin-api": "workspace:^",
     "@backstage/errors": "workspace:^",
     "@backstage/types": "workspace:^",
+    "@types/express": "^4.17.6",
     "cross-fetch": "^4.0.0",
+    "express": "^4.17.1",
     "uri-template": "^2.0.0"
   },
   "devDependencies": {

plugins/events-node/report-alpha.api.md

@@ -7,10 +7,13 @@ import { EventBroker } from '@backstage/plugin-events-node';
 import { EventPublisher } from '@backstage/plugin-events-node';
 import { EventSubscriber } from '@backstage/plugin-events-node';
 import { ExtensionPoint } from '@backstage/backend-plugin-api';
+import { HttpBodyParserOptions } from '@backstage/plugin-events-node';
 import { HttpPostIngressOptions } from '@backstage/plugin-events-node';
 
 // @alpha (undocumented)
 export interface EventsExtensionPoint {
+  // (undocumented)
+  addHttpPostBodyParser(options: HttpBodyParserOptions): void;
   // (undocumented)
   addHttpPostIngress(options: HttpPostIngressOptions): void;
   // @deprecated (undocumented)

plugins/events-node/report.api.md

@@ -9,6 +9,7 @@ import { AuthService } from '@backstage/backend-plugin-api';
 import { DiscoveryService } from '@backstage/backend-plugin-api';
 import { LifecycleService } from '@backstage/backend-plugin-api';
 import { LoggerService } from '@backstage/backend-plugin-api';
+import { Request as Request_2 } from 'express';
 import { RootConfigService } from '@backstage/backend-plugin-api';
 import { ServiceFactory } from '@backstage/backend-plugin-api';
 import { ServiceRef } from '@backstage/backend-plugin-api';
@@ -111,6 +112,27 @@ export interface EventSubscriber {
   supportsEventTopics(): string[];
 }
 
+// @public (undocumented)
+export type HttpBodyParsed = {
+  bodyParsed: any;
+  bodyBuffer: Buffer;
+  encoding: string;
+};
+
+// @public (undocumented)
+export type HttpBodyParser = (
+  request: Request_2,
+  topic: string,
+) => Promise<HttpBodyParsed>;
+
+// @public (undocumented)
+export interface HttpBodyParserOptions {
+  // (undocumented)
+  contentType: string;
+  // (undocumented)
+  parser: HttpBodyParser;
+}
+
 // @public (undocumented)
 export interface HttpPostIngressOptions {
   // (undocumented)

plugins/events-node/src/api/http/HttpBodyParserOptions.ts

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { HttpBodyParser } from './body-parser';
+
+/**
+ * @public
+ */
+export interface HttpBodyParserOptions {
+  contentType: string;
+  parser: HttpBodyParser;
+}

plugins/events-node/src/api/http/body-parser/HttpBodyParser.ts

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { Request } from 'express';
+/**
+ * @public
+ */
+export type HttpBodyParsed = {
+  bodyParsed: any;
+  bodyBuffer: Buffer;
+  encoding: string;
+};
+
+/**
+ * @public
+ */
+export type HttpBodyParser = (
+  request: Request,
+  topic: string,
+) => Promise<HttpBodyParsed>;

plugins/events-node/src/api/http/body-parser/index.ts

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+export * from './HttpBodyParser';

plugins/events-node/src/api/http/index.ts

@@ -15,4 +15,6 @@
  */
 
 export type { HttpPostIngressOptions } from './HttpPostIngressOptions';
+export type { HttpBodyParserOptions } from './HttpBodyParserOptions';
 export * from './validation';
+export * from './body-parser';

plugins/events-node/src/extensions.ts

@@ -19,6 +19,7 @@ import {
   EventBroker,
   EventPublisher,
   EventSubscriber,
+  HttpBodyParserOptions,
   HttpPostIngressOptions,
 } from '@backstage/plugin-events-node';
 
@@ -46,6 +47,8 @@ export interface EventsExtensionPoint {
   ): void;
 
   addHttpPostIngress(options: HttpPostIngressOptions): void;
+
+  addHttpPostBodyParser(options: HttpBodyParserOptions): void;
 }
 
 /**

yarn.lock

@@ -6727,7 +6727,9 @@ __metadata:
     "@backstage/cli": "workspace:^"
     "@backstage/errors": "workspace:^"
     "@backstage/types": "workspace:^"
+    "@types/express": ^4.17.6
     cross-fetch: ^4.0.0
+    express: ^4.17.1
     msw: ^1.0.0
     uri-template: ^2.0.0
   languageName: unknown