Merge pull request #626 from seb-oss/fix/pubsub-subscriber-topic-not-set

fix: pubsub subscriber topic not set

.changeset/quiet-papayas-greet.md

@@ -0,0 +1,5 @@
+---
+"@sebspark/pubsub": patch
+---
+
+Fixed bug that broke topic.initiate, leading to subscription creation failing.

packages/pubsub/src/lib/subscriber.spec.ts

@@ -1,18 +1,14 @@
+import { beforeEach } from 'node:test'
 import { PubSub, type Subscription, type Topic } from '@google-cloud/pubsub'
 import type { Schema } from 'avsc'
-import { type MockedObject, afterAll, describe, expect, it, vi } from 'vitest'
+import { type MockedObject, beforeAll, describe, expect, it, vi } from 'vitest'
 import { createSubscriber } from './subscriber'
 
 type ExampleMessage = {
   messageType: string
   message: string
 }
 
-const message = {
-  messageType: 'type of message',
-  message: 'message data',
-} satisfies ExampleMessage
-
 type ExamplePubsubChannels = {
   example: ExampleMessage
 }
@@ -77,25 +73,62 @@ describe('subscriber', () => {
   const topicName = 'example'
   const subscriptionName = 'example-subscription'
 
-  it('uses an existing subscription if it exists', async () => {
-    const topicMock = new PubSub().topic(topicName) as MockedObject<Topic>
-    const subscriptionMock = topicMock.subscription(
+  let topicMock: MockedObject<Topic>
+  let subscriptionMock: MockedObject<Subscription>
+
+  beforeAll(() => {
+    topicMock = new PubSub().topic(topicName) as MockedObject<Topic>
+    subscriptionMock = topicMock.subscription(
       subscriptionName
     ) as MockedObject<Subscription>
+  })
+
+  beforeEach(() => {})
 
-    subscriptionMock.exists.mockImplementation(() => [true])
+  describe('subscribe', () => {
+    it('uses an existing subscription if it exists', async () => {
+      const subscriber = createSubscriber<ExamplePubsubChannels>({
+        projectId: 'test',
+      })
 
-    const subscriber = createSubscriber<ExamplePubsubChannels>({
-      projectId: 'test',
+      topicMock.createSubscription.mockClear()
+
+      await subscriber.topic('example').subscribe('existing-subscription', {
+        onMessage: () => Promise.resolve(),
+      })
+
+      expect(topicMock.subscription).toHaveBeenCalled()
+      expect(topicMock.createSubscription.mock.calls.length).toBe(0)
     })
+  })
+
+  describe('initiate', () => {
+    it('does not create a subscription if it exists', async () => {
+      subscriptionMock.exists.mockImplementationOnce(() => [true])
 
-    topicMock.createSubscription.mockClear()
+      const subscriber = createSubscriber<ExamplePubsubChannels>({
+        projectId: 'test',
+      })
 
-    await subscriber.topic('example').subscribe('existing-subscription', {
-      onMessage: () => Promise.resolve(),
+      topicMock.createSubscription.mockClear()
+
+      await subscriber.topic('example').initiate('existing-subscription')
+
+      expect(topicMock.createSubscription.mock.calls.length).toBe(0)
     })
 
-    expect(topicMock.subscription).toHaveBeenCalled()
-    expect(topicMock.createSubscription.mock.calls.length).toBe(0)
+    it('creates a subscription if it does not exist', async () => {
+      subscriptionMock.exists.mockImplementationOnce(() => [false])
+
+      const subscriber = createSubscriber<ExamplePubsubChannels>({
+        projectId: 'test',
+      })
+
+      topicMock.createSubscription.mockClear()
+
+      await subscriber.topic('example').initiate('example-subscription')
+
+      expect(topicMock.createSubscription).toHaveBeenCalled()
+    })
   })
 })

packages/pubsub/src/lib/subscriber.ts

@@ -7,7 +7,7 @@ import {
   type Topic,
 } from '@google-cloud/pubsub'
 
-const makeSureSubacriptionExists = async (
+const makeSureSubscriptionExists = async (
   topic: Topic,
   name: string,
   options?: PubSubOptions
@@ -68,16 +68,13 @@ export const createSubscriber = <T extends Record<string, unknown>>(
 
   const typedClient: SubscriptionClient<T> = {
     topic: (name) => {
-      let _topic: Topic
+      const _topic: Topic = client.topic(name as string)
 
       return {
         initiate: async (subscriptionName, options) => {
-          await makeSureSubacriptionExists(_topic, subscriptionName, options)
+          await makeSureSubscriptionExists(_topic, subscriptionName, options)
         },
         subscribe: async (subscriptionName, callbacks, options) => {
-          if (!_topic) {
-            _topic = client.topic(name as string)
-          }
           const subscription = _topic.subscription(subscriptionName)
           subscription.on('message', async (msg) => {
             const data = JSON.parse(msg.data.toString('utf8'))