google pub/sub updates and fixes

src/airhorn.ts

@@ -2,7 +2,11 @@ import { TemplateService } from './template-service.js';
 import { ProviderService } from './provider-service.js';
 import { AirhornProviderType } from './provider-type.js';
 import { type AirhornSubscription } from './subscription.js';
-import { AirhornStore, type AirhornStoreProvider, type CreateAirhornSubscription } from './store.js';
+import { type AirhornNotification } from './notification.js';
+import { AirhornQueue, type AirhornQueueProvider } from './queue.js';
+import {
+	AirhornStore, type CreateAirhornNotification, type AirhornStoreProvider, type CreateAirhornSubscription,
+} from './store.js';
 
 export type AirhornOptions = {
 	TEMPLATE_PATH?: string;
@@ -15,6 +19,7 @@ export type AirhornOptions = {
 	AWS_SNS_REGION?: string;
 	FIREBASE_CERT?: string;
 	STORE_PROVIDER?: AirhornStoreProvider;
+	QUEUE_PROVIDER?: AirhornQueueProvider;
 };
 
 export class Airhorn {
@@ -26,6 +31,7 @@ export class Airhorn {
 	private readonly _templateService = new TemplateService();
 	private readonly _providerService = new ProviderService();
 	private readonly _store?: AirhornStore;
+	private readonly _queue?: AirhornQueue;
 
 	constructor(options?: AirhornOptions) {
 		if (options) {
@@ -37,6 +43,10 @@ export class Airhorn {
 		if (this.options.STORE_PROVIDER) {
 			this._store = new AirhornStore(this.options.STORE_PROVIDER);
 		}
+
+		if (this.options.QUEUE_PROVIDER) {
+			this._queue = new AirhornQueue(this.options.QUEUE_PROVIDER);
+		}
 	}
 
 	public get templates(): TemplateService {
@@ -135,6 +145,16 @@ export class Airhorn {
 
 		throw new Error('Airhorn store not available');
 	}
+
+	public async publishNotification(notification: CreateAirhornNotification): Promise<void> {
+		if (this._queue && this._store) {
+			const updatedNotification = await this._store.createNotification(notification);
+
+			await this._queue.publishNotification(updatedNotification);
+		}
+
+		throw new Error('Airhorn queue and store needed for notifications');
+	}
 }
 
 export { AirhornProviderType } from './provider-type.js';

src/notification.ts

@@ -11,6 +11,7 @@ export enum AirhornNotificationStatus {
 export type AirhornNotification = {
 	id: string;
 	to: string;
+	from: string;
 	subscriptionId: string;
 	externalId?: string;
 	providerType: AirhornProviderType;

src/queue-providers/google-pubsub.ts

@@ -1,4 +1,4 @@
-/* eslint-disable @typescript-eslint/no-empty-function,@typescript-eslint/class-literal-property-style  */
+/* eslint-disable @typescript-eslint/class-literal-property-style  */
 import { PubSub } from '@google-cloud/pubsub';
 import { type AirhornNotification } from '../notification.js';
 import { type AirhornQueueProvider } from '../queue.js';
@@ -55,11 +55,44 @@ export class GooglePubSubQueue implements AirhornQueueProvider {
 		return this._pubsub.topic(this.topicName);
 	}
 
-	async publishNotification(notification: AirhornNotification): Promise<void> {}
+	async publish(notification: AirhornNotification): Promise<void> {
+		await this.createTopic();
+		const topic = await this.getTopic();
+		const data = JSON.stringify(notification);
+		await topic.publishMessage({ data });
+	}
+
+	async subscribe(callback: (notification: AirhornNotification, acknowledge: () => void) => void): Promise<void> {
+		await this.createTopic();
+		const topic = await this.getTopic();
+		const subscriptionName = this._topicName + '-subscription';
+		const subscription = topic.subscription(subscriptionName);
 
-	async acknowledgeNotification(notification: AirhornNotification): Promise<void> {}
+		const [exists] = await subscription.exists();
+		if (!exists) {
+			await topic.createSubscription(subscriptionName);
+		}
 
-	async listenForNotifications(queueName: string, callback: (notification: AirhornNotification) => void): Promise<void> {}
+		subscription.on('message', message => {
+			const notification = JSON.parse(message.data.toString());
+			const acknowledge = () => {
+				message.ack();
+			};
+
+			callback(notification as AirhornNotification, acknowledge);
+		});
+	}
+
+	async clearSubscription(): Promise<void> {
+		const topic = await this.getTopic();
+		const subscriptionName = this._topicName + '-subscription';
+		const subscription = topic.subscription(subscriptionName);
+
+		const [exists] = await subscription.exists();
+		if (exists) {
+			await subscription.delete();
+		}
+	}
 
 	async createTopic(): Promise<void> {
 		if (this._topicCreated) {

src/queue.ts

@@ -3,31 +3,22 @@ import { type AirhornNotification } from './notification.js';
 export type AirhornQueueProvider = {
 	name: string;
 	uri: string;
-	publishNotification(notification: AirhornNotification): Promise<void>;
-	acknowledgeNotification(notification: AirhornNotification): Promise<void>;
-	listenForNotifications(queueName: string, callback: (notification: AirhornNotification) => void): Promise<void>;
-};
-
-export type AirhornQueueOptions = {
-	provider: AirhornQueueProvider;
+	publish(notification: AirhornNotification): Promise<void>;
+	subscribe(callback: (notification: AirhornNotification, acknowledge: () => void) => void): Promise<void>;
 };
 
 export class AirhornQueue {
 	public provider: AirhornQueueProvider;
 
-	constructor(options: AirhornQueueOptions) {
-		this.provider = options.provider;
+	constructor(provider: AirhornQueueProvider) {
+		this.provider = provider;
 	}
 
 	public async publishNotification(notification: AirhornNotification) {
-		await this.provider.publishNotification(notification);
-	}
-
-	public async acknowledgeNotification(notification: AirhornNotification) {
-		await this.provider.acknowledgeNotification(notification);
+		await this.provider.publish(notification);
 	}
 
-	public async listenForNotifications(queueName: string, callback: (notification: AirhornNotification) => void) {
-		await this.provider.listenForNotifications(queueName, callback);
+	public async subscribe(callback: (notification: AirhornNotification, acknowledge: () => void) => void) {
+		await this.provider.subscribe(callback);
 	}
 }

src/store-providers/mongo.ts

@@ -251,6 +251,7 @@ export class MongoStoreProvider implements AirhornStoreProvider {
 		const notification: AirhornNotification = {
 			id: document._id,
 			to: document.to,
+			from: document.from,
 			subscriptionId: document.subscriptionId,
 			externalId: document.externalId,
 			providerType: document.providerType as AirhornProviderType,

test/queue.test.ts

@@ -1,4 +1,5 @@
 /* eslint-disable @typescript-eslint/no-empty-function */
+import { subscribe } from 'node:diagnostics_channel';
 import {describe, test, expect} from 'vitest';
 import {AirhornQueue} from '../src/queue.js';
 import { AirhornNotificationStatus, type AirhornNotification } from '../src/notification.js';
@@ -7,14 +8,14 @@ import { AirhornProviderType } from '../src/provider-type.js';
 const providerMock = {
 	name: 'mock',
 	uri: 'mock://localhost',
-	async publishNotification(notification: AirhornNotification) {},
-	async acknowledgeNotification(notification: AirhornNotification) {},
-	async listenForNotifications(queueName: string, callback: (notification: AirhornNotification) => void) {},
+	async publish(notification: AirhornNotification) {},
+	async subscribe(callback: (notification: AirhornNotification, acknowledge: () => void) => void) {},
 };
 
 const notificationMock: AirhornNotification = {
 	id: '1',
 	to: '1',
+	from: '1',
 	subscriptionId: '1',
 	templateName: '1',
 	providerName: '1',
@@ -27,23 +28,18 @@ const notificationMock: AirhornNotification = {
 
 describe('AirhornQueue', async () => {
 	test('should create a new instance of AirhornQueue', async () => {
-		const queue = new AirhornQueue({ provider: providerMock });
+		const queue = new AirhornQueue(providerMock);
 		expect(queue.provider).toEqual(providerMock);
 		expect(queue.provider.name).toEqual('mock');
 	});
 
 	test('should publish a notification', async () => {
-		const queue = new AirhornQueue({ provider: providerMock });
+		const queue = new AirhornQueue(providerMock);
 		await queue.publishNotification(notificationMock);
 	});
 
-	test('should acknowledge a notification', async () => {
-		const queue = new AirhornQueue({ provider: providerMock });
-		await queue.acknowledgeNotification(notificationMock);
-	});
-
 	test('should listen for notifications', async () => {
-		const queue = new AirhornQueue({ provider: providerMock });
-		await queue.listenForNotifications('queueName', (notification: AirhornNotification) => {});
+		const queue = new AirhornQueue(providerMock);
+		await queue.subscribe((notification, acknowledge) => {});
 	});
 });