setup extension config

extension.yaml

@@ -22,7 +22,14 @@ sourceUrl: https://github.com/teaglebuilt/firebase-auth-outbox-extension
 
 # Specify whether a paid-tier billing plan is required to use your extension.
 # Learn more in the docs: https://firebase.google.com/docs/extensions/reference/extension-yaml#billing-required-field
-billingRequired: true
+billingRequired: false
+
+author:
+  authorName: Dillan Teagle
+  email: dillan.teagle.va@gmail.com
+  url: https://github.com/teaglebuilt
+contributors:
+  - authorName: Dillan Teagle
 
 # In an `apis` field, list any Google APIs (like Cloud Translation, BigQuery, etc.)
 # required for your extension to operate.
@@ -74,6 +81,12 @@ resources:
 # Learn more in the docs:
 # https://firebase.google.com/docs/extensions/reference/extension-yaml#params-field
 params:
+  - param: BROKER
+    label: Broker URL
+    description: "url of broker example: https://kafka.domain-name.com:9092"
+  - param: TOPIC
+    label: Name of Topic
+    description: Name of the topic you will be publishing messages to
   - param: FIELDS_TO_INCLUDED
     label: Fields to include
     description: >-
@@ -98,8 +111,7 @@ params:
         value: disabled
       - label: creationTime
         value: creationTime
-    default: email,displayName,photoURL
-    required: true
+
 
   - param: LOCATION
     label: Cloud Functions location

functions/src/clients/kafka/client.ts

@@ -0,0 +1,59 @@
+import * as KafkaJS from 'kafkajs';
+
+export interface MicroserviceMessage<BodyType> {
+  messageSchema: any;
+  getMessageName(): string;
+  fullyQualifiedName(): string;
+}
+
+
+export class KafkaClient {
+  kafka: KafkaJS.Kafka;
+  producer: KafkaJS.Producer;
+  protected isReady = false;
+
+  /**
+   * Instantiates a new Kafka client
+   *
+   * @param brokers - List of brokers to connect to
+   * @param kafkaConfig - Kafka client configuration
+   */
+  constructor(kafkaConfig: KafkaJS.KafkaConfig) {
+    this.kafka = new KafkaJS.Kafka({
+      clientId: 'firebase-auth-producer',
+      ...kafkaConfig,
+    });
+    this.producer = this.kafka.producer(kafkaConfig)
+  }
+
+  async init() {
+    if(this.ready())
+      await this.producer.connect();
+  }
+
+  /**
+   * Returns true if the Producer is ready to produce messages
+   */
+  public ready(): boolean {
+    return this.isReady;
+  }
+
+  async disconnect() {
+    await this.producer.disconnect();
+  }
+
+  async send<T>(
+    topicName: string,
+    payload: T,
+  ): Promise<KafkaJS.RecordMetadata[]> {
+    const result = await this.producer.send({
+      topic: topicName,
+      messages: [
+        {
+          value: JSON.stringify(payload),
+        },
+      ],
+    });
+    return result;
+  }
+}

functions/src/config.ts

@@ -1,11 +1,15 @@
 interface Configuration {
+  broker: string;
+  topic: string;
   fieldsToInclude: string[];
   location: string;
 }
 
 const config: Configuration = {
   fieldsToInclude: process.env.FIELDS_TO_INCLUDE!.split(','),
   location: process.env.FUNCTIONS_LOCATION!,
+  broker: process.env.KAFKA_BROKER!,
+  topic: process.env.KAFKA_TOPIC!
 };
 
 export default config;

functions/src/index.ts

@@ -12,8 +12,12 @@ import { UserRecord, getAuth } from "firebase-admin/auth";
 import * as functions from "firebase-functions";
 import config from './config'
 import { AuthUserRecord } from "firebase-functions/lib/common/providers/identity";
+import { KafkaClient } from "./clients/kafka/client";
 
 initializeApp();
+const kafka = new KafkaClient({
+  brokers: [config.broker],
+})
 
 const getUserAuthFields = (user: UserRecord) => {
   const fields: any = {};
@@ -33,23 +37,22 @@ export const produceUserCreatedEvent = functions.auth
   .user()
   .onCreate(async (user) => {
     const data = getUserAuthFields(user);
-
-    // TODO: raise event to kafka
+    console.log(data)
+    await kafka.send(config.topic, data)
   });
 
 export const produceUserDeletedEvent = functions.auth
   .user()
   .onDelete(async (user) => {
     const data = getUserAuthFields(user);
-
-    // TODO: raise event to kafka
+    console.log(data)
+    await kafka.send(config.topic, data)
   });
 
 
 export const produceUserSignInEvent = functions.auth
   .user()
   .beforeSignIn(async (auth: AuthUserRecord) => {
     console.log(auth)
-
-    // TODO: raise event to kafka
+    await kafka.send(config.topic, auth)
   });