feat(@golevelup/nestjs-google-cloud-pubsub): schema validation via validateMessage API, optional peer deps

.changeset/pubsub-schema-validation-refactor.md

@@ -0,0 +1,11 @@
+---
+'@golevelup/nestjs-google-cloud-pubsub': major
+---
+
+Schema validation now sends a serialized stub message to GCP's `validateMessage` API instead of comparing raw schema definitions. This correctly validates compatibility across all schema revisions and respects the configured encoding (Binary/JSON).
+
+`avsc` and `@protobuf-ts/runtime` are now optional peer dependencies — install only what your schema type requires.
+
+`batchManagerOptions.maxWaitTimeInMillis` has been removed from subscription configuration. Wait time is now derived automatically from `maxMessages` using an adaptive formula (50–500ms range).
+
+**Breaking change:** remove `maxWaitTimeInMillis` from any `batchManagerOptions` configuration.

.gitignore

@@ -10,4 +10,6 @@ integration/rabbitmq/dist
 
 # doc ignored directories
 docs/.vitepress/dist
-docs/.vitepress/cache
+docs/.vitepress/cache
+# Claude Code
+.claude/

docs/modules/google-pubsub.md

@@ -5,8 +5,13 @@ A type-safe Google Cloud Pub/Sub integration for NestJS. The module validates to
 ## Installation
 
 ```bash
-npm install @golevelup/nestjs-google-cloud-pubsub
-npm install @google-cloud/pubsub avsc @protobuf-ts/runtime
+npm install @golevelup/nestjs-google-cloud-pubsub @google-cloud/pubsub
+
+# Avro schemas only:
+npm install avsc
+
+# Protocol Buffer schemas only:
+npm install @protobuf-ts/runtime
 ```
 
 ## Quick Start
@@ -53,20 +58,9 @@ export const topics = [
       type: SchemaTypes.Avro,
     },
     subscriptions: [
-      /**
-       * High Density Configuration (Designed for up to 100 subscriptions per instance):
-       *
-       * 1. Flow Control:
-       *    - Limits each subscription to 10MB or 500 messages.
-       *    - Math: 100 subs * 10MB = 1GB Raw Buffer (~2GB Real RAM Usage).
-       *
-       * 2. Batch Manager:
-       *    - Aggregates up to 125 messages.
-       *    - Flushes every 200ms even if 125 messages weren't aggregated.
-       */
       {
         name: 'order.created.subscription.order-processor-service',
-        batchManagerOptions: { maxMessages: 125, maxWaitTimeMilliseconds: 200 },
+        batchManagerOptions: { maxMessages: 125 },
         options: {
           flowControl: {
             allowExcessMessages: false,
@@ -85,8 +79,6 @@ export const topics = [
         PaymentProcessedProtocolBufferSchema as MessageType<PaymentProcessedProtocolBufferSchema>,
       encoding: Encodings.Binary,
       name: 'payment.processed.schema',
-      protoPath:
-        '/Users/Desktop/google-cloud-pubsub/proto/payment-processed.proto',
       type: SchemaTypes.ProtocolBuffer,
     },
     subscriptions: [
@@ -197,8 +189,6 @@ export class AppService {
 
 - Custom error hooks / metric emitters.
 - Push subscription support.
-- Split optional dependencies so Avro (`avsc`) and Protocol Buffer runtimes are only required when those schema types are used.
 - Publisher.ready() helper that awaits PubsubClient initialization.
 - Manual ack/nack.
-- Iterate over schema revisions page by page instead of fetching them all at once.
 - Support for custom Avro serialization options (e.g. `wrapUnions`, `logicalTypes`) with adaptive type inference.

integration/google-cloud-pubsub/package.json

@@ -1,5 +1,5 @@
 {
-  "name": "google-cloud-pubsub-integration",
+  "name": "@integration/google-cloud-pubsub",
   "version": "0.0.1",
   "description": "Google cloud pubsub integration project",
   "private": true,

integration/google-cloud-pubsub/src/google-cloud-pubsub.config.ts

@@ -3,7 +3,6 @@ import {
   PubsubTopicConfiguration,
 } from '@golevelup/nestjs-google-cloud-pubsub';
 import { MessageType } from '@protobuf-ts/runtime';
-import * as path from 'node:path';
 
 import { Level5ProtocolBuffer } from '../proto/level5';
 
@@ -38,14 +37,7 @@ export const topics = [
     subscriptions: [
       {
         name: 'order.created.subscription.order-processor-service',
-        batchManagerOptions: { maxMessages: 15, maxWaitTimeMilliseconds: 200 },
-        options: {
-          flowControl: {
-            allowExcessMessages: false,
-            maxBytes: 10 * 1024 * 1024,
-            maxMessages: 500,
-          },
-        },
+        batchManagerOptions: { maxMessages: 15 },
       },
       { name: 'order.created.subscription.analytic-service' },
     ],
@@ -60,17 +52,11 @@ export const topics = [
       definition: Level5ProtocolBuffer as MessageType<Level5ProtocolBuffer>,
       encoding: 'BINARY',
       name: 'payment.processed.schema',
-      protoPath: path.join(process.cwd(), 'proto/level5.proto'),
       type: 'PROTOCOL_BUFFER',
     },
     subscriptions: [
       { name: 'payment.processed.payment-processor-service' },
-      {
-        name: 'payment.processed.analytic-service',
-        options: {
-          flowControl: { maxMessages: 100 },
-        },
-      },
+      { name: 'payment.processed.analytic-service' },
     ],
   },
   {

integration/google-cloud-pubsub/src/recreate-infrastructure.ts

@@ -52,7 +52,7 @@ async function bootstrap() {
         definition = JSON.stringify(topicConfiguration.schema.definition);
       } else {
         definition = await readFile(
-          path.resolve(topicConfiguration.schema.protoPath),
+          path.resolve(process.cwd(), 'proto/level5.proto'),
           'utf-8',
         );
       }

packages/google-cloud-pubsub/e2e/pubsub-adaptive-flow-control.e2e-spec.ts

@@ -0,0 +1,211 @@
+import { Subscription } from '@google-cloud/pubsub';
+import { PubsubClient } from '../src/client/pubsub.client';
+import { PubsubSubscriptionContainer } from '../src/client/pubsub-subscription.container';
+import { PubsubTopicContainer } from '../src/client/pubsub-topic.container';
+import { ResourceState } from '../src/client/resource-manager';
+
+const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+function blockEventLoop(durationMs: number) {
+  const end = Date.now() + durationMs;
+  while (Date.now() < end) {}
+}
+
+function createMockSubscription() {
+  const closeFn = jest.fn();
+  const openFn = jest.fn();
+
+  return {
+    close: closeFn.mockResolvedValue(undefined),
+    open: openFn,
+    removeAllListeners: jest.fn().mockReturnThis(),
+    on: jest.fn().mockReturnThis(),
+    name: 'mock-subscription',
+  } as unknown as Subscription & { close: jest.Mock; open: jest.Mock };
+}
+
+function injectSubscriptionContainer(
+  client: PubsubClient,
+  name: string,
+  subscription: Subscription,
+) {
+  const containers = (client as any).subscriptionContainers as Map<
+    string,
+    PubsubSubscriptionContainer
+  >;
+
+  const container = new PubsubSubscriptionContainer(
+    subscription,
+    { name } as any,
+    {} as PubsubTopicContainer,
+  );
+
+  containers.set(name, container);
+
+  return container;
+}
+
+describe.skip('Adaptive Flow Control — Subscription Pause/Resume', () => {
+  jest.setTimeout(30000);
+
+  let client: PubsubClient;
+
+  afterEach(async () => {
+    await client?.close().catch(() => {});
+  });
+
+  it('should pause subscriptions on Critical and resume on recovery.', async () => {
+    client = new PubsubClient({ adaptiveFlowControl: true });
+
+    const mockSub = createMockSubscription();
+    injectSubscriptionContainer(client, 'sub-1', mockSub);
+
+    // Trigger initialize to register the resource manager listener.
+    await client.initialize([]);
+
+    await delay(1500);
+
+    for (let i = 0; i < 5; i++) {
+      blockEventLoop(300);
+      await delay(100);
+    }
+
+    await delay(3000);
+
+    expect(mockSub.close).toHaveBeenCalled();
+
+    await delay(5000);
+
+    expect(mockSub.open).toHaveBeenCalled();
+  });
+
+  it('should only pause on Critical, not on Pressure.', async () => {
+    client = new PubsubClient({ adaptiveFlowControl: true });
+
+    const mockSub = createMockSubscription();
+    injectSubscriptionContainer(client, 'sub-1', mockSub);
+
+    await client.initialize([]);
+
+    const resourceManager = (client as any).resourceManager;
+
+    resourceManager.emit('stateChanged', {
+      previousState: ResourceState.Healthy,
+      newState: ResourceState.Pressure,
+      metrics: {} as any,
+    });
+
+    expect(mockSub.close).not.toHaveBeenCalled();
+    expect(mockSub.open).not.toHaveBeenCalled();
+
+    resourceManager.emit('stateChanged', {
+      previousState: ResourceState.Pressure,
+      newState: ResourceState.Critical,
+      metrics: {} as any,
+    });
+
+    expect(mockSub.close).toHaveBeenCalledTimes(1);
+
+    resourceManager.emit('stateChanged', {
+      previousState: ResourceState.Critical,
+      newState: ResourceState.Healthy,
+      metrics: {} as any,
+    });
+
+    expect(mockSub.open).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not re-open subscriptions if close() is called during Critical.', async () => {
+    client = new PubsubClient({ adaptiveFlowControl: true });
+
+    const mockSub = createMockSubscription();
+    injectSubscriptionContainer(client, 'sub-1', mockSub);
+
+    await client.initialize([]);
+
+    const resourceManager = (client as any).resourceManager;
+
+    // Simulate entering Critical state.
+    resourceManager.emit('stateChanged', {
+      previousState: ResourceState.Healthy,
+      newState: ResourceState.Critical,
+      metrics: {} as any,
+    });
+
+    expect(mockSub.close).toHaveBeenCalled();
+
+    // Shut down while in Critical state — flag is reset, resourceManager stopped.
+    await client.close();
+
+    // Simulate recovery event after close (should be no-op since flag was reset).
+    resourceManager.emit('stateChanged', {
+      previousState: ResourceState.Critical,
+      newState: ResourceState.Healthy,
+      metrics: {} as any,
+    });
+
+    expect(mockSub.open).not.toHaveBeenCalled();
+  });
+
+  it('should pause and resume multiple subscriptions.', async () => {
+    client = new PubsubClient({ adaptiveFlowControl: true });
+
+    const mockSub1 = createMockSubscription();
+    const mockSub2 = createMockSubscription();
+    const mockSub3 = createMockSubscription();
+
+    injectSubscriptionContainer(client, 'sub-1', mockSub1);
+    injectSubscriptionContainer(client, 'sub-2', mockSub2);
+    injectSubscriptionContainer(client, 'sub-3', mockSub3);
+
+    await client.initialize([]);
+
+    await delay(1500);
+
+    for (let i = 0; i < 5; i++) {
+      blockEventLoop(300);
+      await delay(100);
+    }
+
+    await delay(3000);
+
+    expect(mockSub1.close).toHaveBeenCalled();
+    expect(mockSub2.close).toHaveBeenCalled();
+    expect(mockSub3.close).toHaveBeenCalled();
+
+    await delay(5000);
+
+    expect(mockSub1.open).toHaveBeenCalled();
+    expect(mockSub2.open).toHaveBeenCalled();
+    expect(mockSub3.open).toHaveBeenCalled();
+  });
+
+  it('should handle rapid Critical transitions without errors.', async () => {
+    client = new PubsubClient({ adaptiveFlowControl: true });
+
+    const mockSub = createMockSubscription();
+    injectSubscriptionContainer(client, 'sub-1', mockSub);
+
+    await client.initialize([]);
+
+    const resourceManager = (client as any).resourceManager;
+
+    // Simulate rapid Critical in/out transitions directly.
+    for (let i = 0; i < 10; i++) {
+      resourceManager.emit('stateChanged', {
+        previousState: ResourceState.Healthy,
+        newState: ResourceState.Critical,
+        metrics: {} as any,
+      });
+
+      resourceManager.emit('stateChanged', {
+        previousState: ResourceState.Critical,
+        newState: ResourceState.Healthy,
+        metrics: {} as any,
+      });
+    }
+
+    expect(mockSub.close).toHaveBeenCalledTimes(10);
+    expect(mockSub.open).toHaveBeenCalledTimes(10);
+  });
+});

packages/google-cloud-pubsub/e2e/pubsub-client-attach-batch-handler.e2e-spec.ts

@@ -31,7 +31,6 @@ describe.skip('PubsubClient.attachBatchHandler()', () => {
           name: subscriptionName,
           batchManagerOptions: {
             maxMessages: 5,
-            maxWaitTimeMilliseconds: 10000,
           },
         },
       ],
@@ -72,13 +71,13 @@ describe.skip('PubsubClient.attachBatchHandler()', () => {
     expect(processedMessagesCount).toBe(5);
     expect(endTime - startTime).toBeLessThan(5000);
 
-    await pubsubClient.publish(topicName, { data: Buffer.from('6') });
+    await pubsubClient.close();
+
+    await pubsub.topic(topicName).publishMessage({ data: Buffer.from('6') });
 
     await new Promise((resolve) => setTimeout(resolve, 1500));
 
     expect(processedMessagesCount).toBe(5);
-
-    await pubsubClient.close();
   });
 
   it('flush by timer: should flush quickly when maxMessages limit is not reached.', async () => {
@@ -92,7 +91,6 @@ describe.skip('PubsubClient.attachBatchHandler()', () => {
           name: subscriptionName,
           batchManagerOptions: {
             maxMessages: 100,
-            maxWaitTimeMilliseconds: 1000,
           },
         },
       ],
@@ -129,7 +127,7 @@ describe.skip('PubsubClient.attachBatchHandler()', () => {
 
     expect(processedMessagesCount).toBe(2);
 
-    expect(duration).toBeGreaterThanOrEqual(1000);
+    expect(duration).toBeGreaterThanOrEqual(300);
     expect(duration).toBeLessThan(4000);
 
     await pubsubClient.close();
@@ -146,7 +144,6 @@ describe.skip('PubsubClient.attachBatchHandler()', () => {
           name: subscriptionName,
           batchManagerOptions: {
             maxMessages: 100,
-            maxWaitTimeMilliseconds: 5000,
           },
         },
       ],