Added new integration tests for event ingestion endpoint

server.ts

@@ -1,19 +1,13 @@
 import './src/utils/instrumentation';
 import {fileURLToPath} from 'url';
+import {initializeApp} from './src/initializeApp';
 
 const isMainModule = process.argv[1] === fileURLToPath(import.meta.url);
 const port: number = parseInt(process.env.PORT || '3000', 10);
 const isWorkerMode = process.env.WORKER_MODE === 'true';
 
-// Load the appropriate app once
-let app;
-if (isWorkerMode) {
-    const workerModule = await import('./src/worker-app');
-    app = workerModule.default;
-} else {
-    const appModule = await import('./src/app');
-    app = appModule.default;
-}
+// Create an instance of the appropriate app
+const app = await initializeApp({isWorkerMode});
 
 // Start the server if this file is run directly
 if (isMainModule) {
@@ -35,5 +29,4 @@ if (isMainModule) {
     start();
 }
 
-// Export the app
-export default app;
+export default app;

src/initializeApp.ts

@@ -0,0 +1,5 @@
+export async function initializeApp({isWorkerMode}: {isWorkerMode: boolean}) {
+    const appModulePath = isWorkerMode ? './src/worker-app' : './src/service-app';
+    const appModule = await import(appModulePath);
+    return appModule.default();
+}

src/service-app.ts

@@ -0,0 +1,46 @@
+// Main module file
+import fastify from 'fastify';
+import {TypeBoxTypeProvider} from '@fastify/type-provider-typebox';
+import loggingPlugin from './plugins/logging';
+import corsPlugin from './plugins/cors';
+import proxyPlugin from './plugins/proxy';
+import {getLoggerConfig} from './utils/logger';
+import {errorHandler} from './utils/error-handler';
+import v1Routes from './routes/v1';
+import replyFrom from '@fastify/reply-from';
+
+function createApp() {
+    const app = fastify({
+        logger: getLoggerConfig(),
+        disableRequestLogging: true,
+        trustProxy: process.env.TRUST_PROXY !== 'false'
+    }).withTypeProvider<TypeBoxTypeProvider>();
+
+    // Register global validation error handler
+    app.setErrorHandler(errorHandler());
+
+    // Register reply-from plugin
+    app.register(replyFrom);
+
+    // Register CORS plugin
+    app.register(corsPlugin);
+
+    // Register logging plugin
+    app.register(loggingPlugin);
+
+    // Register proxy plugin
+    app.register(proxyPlugin);
+
+    // Register v1 routes
+    app.register(v1Routes, {prefix: '/api/v1'});
+
+    // Routes
+    app.get('/', async () => {
+        return 'Hello World - Github Actions Deployment Test';
+    });
+
+    return app;
+}
+
+export default createApp;
+

src/services/events/publisher.ts

@@ -11,8 +11,8 @@ class EventPublisher {
     private static instance: EventPublisher;
     private pubsub: PubSub;
 
-    private constructor() {
-        this.pubsub = new PubSub({
+    private constructor(pubsub?: PubSub) {
+        this.pubsub = pubsub || new PubSub({
             projectId: process.env.GOOGLE_CLOUD_PROJECT
         });
     }
@@ -24,6 +24,10 @@ class EventPublisher {
         return EventPublisher.instance;
     }
 
+    static resetInstance(pubsub?: PubSub): void {
+        EventPublisher.instance = new EventPublisher(pubsub);
+    }
+
     async publishEvent({topic, payload, logger}: PublishEventOptions): Promise<string> {
         try {
             const message = {
@@ -32,7 +36,7 @@ class EventPublisher {
             };
 
             const messageId = await this.pubsub.topic(topic).publishMessage(message);
-            
+
             logger.info({
                 messageId,
                 topic,
@@ -55,4 +59,7 @@ class EventPublisher {
 export const publishEvent = async ({topic, payload, logger}: PublishEventOptions): Promise<string> => {
     const publisher = EventPublisher.getInstance();
     return publisher.publishEvent({topic, payload, logger});
-};
+};
+
+// Export for testing purposes
+export {EventPublisher};

src/worker-app.ts

@@ -4,25 +4,29 @@ import loggingPlugin from './plugins/logging';
 import workerPlugin from './plugins/worker-plugin';
 import {getLoggerConfig} from './utils/logger';
 
-const app = fastify({
-    logger: getLoggerConfig(),
-    disableRequestLogging: true,
-    trustProxy: process.env.TRUST_PROXY !== 'false'
-});
+function createApp() {
+    const app = fastify({
+        logger: getLoggerConfig(),
+        disableRequestLogging: true,
+        trustProxy: process.env.TRUST_PROXY !== 'false'
+    });
+
+    // Register logging plugin for consistent log formatting
+    app.register(loggingPlugin);
+
+    // Register worker plugin for heartbeat logging
+    app.register(workerPlugin);
+
+    // Health endpoints for Cloud Run deployment
+    app.get('/', async () => {
+        return {status: 'worker-healthy'};
+    });
+
+    app.get('/health', async () => {
+        return {status: 'worker-healthy'};
+    });
 
-// Register logging plugin for consistent log formatting
-app.register(loggingPlugin);
+    return app;
+}
 
-// Register worker plugin for heartbeat logging
-app.register(workerPlugin);
-
-// Health endpoints for Cloud Run deployment
-app.get('/', async () => {
-    return {status: 'worker-healthy'};
-});
-
-app.get('/health', async () => {
-    return {status: 'worker-healthy'};
-});
-
-export default app;
+export default createApp;

test/integration/routes/v1/page_hit.test.ts → test/integration/api/v1/page_hit.test.ts

@@ -1,17 +1,16 @@
 import {describe, it, expect, beforeAll} from 'vitest';
 import {FastifyInstance} from 'fastify';
+import {initializeApp} from '../../../../src/initializeApp';
 
 describe('/api/v1/page_hit', () => {
-    let fastify: FastifyInstance;
+    let app: FastifyInstance;
 
     beforeAll(async function () {
-        const appModule = await import('../../../../src/app');
-        fastify = appModule.default;
-        await fastify.ready();
+        app = await initializeApp({isWorkerMode: false});
     });
 
     it('should return 200', async function () {
-        const response = await fastify.inject({
+        const response = await app.inject({
             method: 'GET',
             url: '/api/v1/page_hit'
         });

test/integration/api/web_analytics/batch-mode.test.ts

@@ -0,0 +1,88 @@
+import {describe, expect, it, beforeEach, beforeAll, vi} from 'vitest';
+import {FastifyInstance, FastifyRequest, FastifyReply} from 'fastify';
+import {expectResponse} from '../../../utils/assertions';
+import {createPubSubSpy} from '../../../utils/pubsub-spy';
+import defaultValidRequestQuery from '../../../utils/fixtures/defaultValidRequestQuery.json';
+import defaultValidRequestHeaders from '../../../utils/fixtures/defaultValidRequestHeaders.json';
+import defaultValidRequestBody from '../../../utils/fixtures/defaultValidRequestBody.json';
+import headersWithoutSiteUuid from '../../../utils/fixtures/headersWithoutSiteUuid.json';
+import {initializeApp} from '../../../../src/initializeApp';
+import {EventPublisher} from '../../../../src/services/events/publisher';
+
+const preHandlerStub = async (_request: FastifyRequest, reply: FastifyReply) => {
+    reply.code(202);
+};
+
+describe('POST /tb/web_analytics', () => {
+    let app: FastifyInstance;
+
+    describe('Batch Mode - Publishing to pub/sub', function () {
+        let pubSubSpy: ReturnType<typeof createPubSubSpy>;
+        const pageHitsRawTopic: string = 'page-hits-raw';
+
+        beforeAll(async function () {
+            vi.stubEnv('PUBSUB_TOPIC_PAGE_HITS_RAW', pageHitsRawTopic);
+        });
+        beforeEach(async function () {
+            app = await initializeApp({isWorkerMode: false});
+            app.addHook('preHandler', preHandlerStub);
+            pubSubSpy = createPubSubSpy();
+            EventPublisher.resetInstance(pubSubSpy.mockPubSub as any);
+        });
+
+        it('should transform the request body and publish to pub/sub', async function () {
+            await app.inject({
+                method: 'POST',
+                url: '/tb/web_analytics',
+                query: defaultValidRequestQuery,
+                headers: defaultValidRequestHeaders,
+                body: defaultValidRequestBody
+            });
+
+            pubSubSpy.expectPublishedMessageToTopic(pageHitsRawTopic).withMessageData({
+                timestamp: expect.any(String),
+                action: 'page_hit',
+                version: '1',
+                site_uuid: defaultValidRequestHeaders['x-site-uuid'],
+                payload: {
+                    event_id: defaultValidRequestBody.payload.event_id,
+                    href: 'https://www.example.com/',
+                    pathname: '/',
+                    member_uuid: 'undefined',
+                    member_status: 'undefined',
+                    post_uuid: 'undefined',
+                    post_type: 'null',
+                    parsedReferrer: {
+                        medium: '',
+                        source: '',
+                        url: ''
+                    },
+                    locale: 'en-US',
+                    location: 'US',
+                    referrer: null
+                },
+                meta: {
+                    ip: expect.any(String),
+                    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'
+                }
+            });
+        });
+
+        it('should not publish a message if the request fails validation', async function () {
+            const response = await app.inject({
+                method: 'POST',
+                url: '/tb/web_analytics',
+                query: defaultValidRequestQuery,
+                headers: headersWithoutSiteUuid,
+                body: defaultValidRequestBody
+            });
+            expectResponse({
+                response, 
+                statusCode: 400, 
+                errorType: 'Bad Request',
+                message: 'headers must have required property \'x-site-uuid\''
+            });
+            pubSubSpy.expectNoMessagesPublished();
+        });
+    });
+});