feat(admin): Send notification by topic

.env

@@ -58,7 +58,7 @@ export LND2_TYPE=offchain
 export LND1_NAME=lnd1
 export LND2_NAME=lnd2
 
-export MONGODB_CON=mongodb://${DOCKER_HOST_IP}:27017/galoy
+export MONGODB_CON=mongodb://localhost:27017/galoy
 
 export REDIS_0_DNS=${DOCKER_HOST_IP}
 export REDIS_0_PORT=6378

dev/bin/subscribe-device-tokens-to-broadcast.js

@@ -0,0 +1,104 @@
+#!/usr/bin/env node
+/**
+ * One-off script: subscribes all device tokens stored in deviceTopics to their
+ * respective FCM topics via the Firebase Admin SDK.
+ *
+ * Run AFTER the migration `20260317125624-subscribe-device-tokens-to-broadcast` has
+ * been applied to populate the deviceTopics field.
+ *
+ * Required env vars:
+ *   MONGODB_CON                     e.g. mongodb://localhost/galoy
+ *   GOOGLE_APPLICATION_CREDENTIALS  path to Firebase service account JSON
+ *   FCM_TOPIC_PREFIX                (optional) e.g. "test" → topic becomes "test-broadcast"
+ *                                   omit on prod → topic is "broadcast"
+ */
+
+const { MongoClient } = require("mongodb")
+const admin = require("firebase-admin")
+
+const BATCH_SIZE = 1000
+
+const MONGODB_CON = process.env.MONGODB_CON
+if (!MONGODB_CON) {
+  console.error("Error: MONGODB_CON environment variable is required")
+  process.exit(1)
+}
+
+if (!process.env.GOOGLE_APPLICATION_CREDENTIALS) {
+  console.error("Error: GOOGLE_APPLICATION_CREDENTIALS environment variable is required")
+  process.exit(1)
+}
+
+admin.initializeApp({ credential: admin.credential.applicationDefault() })
+const messaging = admin.messaging()
+
+async function subscribeInBatches(tokens, topic) {
+  let successCount = 0
+  let failureCount = 0
+
+  for (let i = 0; i < tokens.length; i += BATCH_SIZE) {
+    const batch = tokens.slice(i, i + BATCH_SIZE)
+    const batchNum = Math.floor(i / BATCH_SIZE) + 1
+    console.log(
+      `Subscribing batch ${batchNum} (tokens ${i + 1}–${i + batch.length}) to topic "${topic}"`,
+    )
+
+    const response = await messaging.subscribeToTopic(batch, topic)
+    successCount += response.successCount
+    failureCount += response.failureCount
+
+    if (response.errors.length > 0) {
+      response.errors.forEach(({ index, error }) => {
+        console.warn(`  Token[${index}] failed: ${error.message}`)
+      })
+    }
+  }
+
+  return { successCount, failureCount }
+}
+
+async function main() {
+  const client = new MongoClient(MONGODB_CON)
+
+  try {
+    await client.connect()
+    const db = client.db()
+
+    const users = await db
+      .collection("users")
+      .find(
+        { deviceTopics: { $exists: true } },
+        { projection: { _id: 0, deviceTopics: 1 } },
+      )
+      .toArray()
+
+    if (users.length === 0) {
+      console.log("No users with deviceTopics found — run the migration first")
+      return
+    }
+
+    // Group tokens by topic
+    const tokensByTopic = {}
+    for (const user of users) {
+      for (const [token, topics] of Object.entries(user.deviceTopics)) {
+        for (const topic of topics) {
+          if (!tokensByTopic[topic]) tokensByTopic[topic] = []
+          tokensByTopic[topic].push(token)
+        }
+      }
+    }
+
+    for (const [topic, tokens] of Object.entries(tokensByTopic)) {
+      console.log(`\nSubscribing ${tokens.length} tokens to topic "${topic}"`)
+      const { successCount, failureCount } = await subscribeInBatches(tokens, topic)
+      console.log(`Done — success: ${successCount}, failures: ${failureCount}`)
+    }
+  } finally {
+    await client.close()
+  }
+}
+
+main().catch((err) => {
+  console.error(err)
+  process.exit(1)
+})

dev/bruno/Flash GraphQL API/admin/notification-topics.bru

@@ -0,0 +1,21 @@
+meta {
+  name: notification-topics
+  type: graphql
+  seq: 3
+}
+
+post {
+  url: {{admin_url}}
+  body: graphql
+  auth: bearer
+}
+
+auth:bearer {
+  token: {{admin_token}}
+}
+
+body:graphql {
+  query {
+    notificationTopics
+  }
+}

dev/bruno/Flash GraphQL API/admin/send-notification.bru

@@ -0,0 +1,36 @@
+meta {
+  name: send-notification
+  type: graphql
+  seq: 2
+}
+
+post {
+  url: {{admin_url}}
+  body: graphql
+  auth: bearer
+}
+
+auth:bearer {
+  token: {{admin_token}}
+}
+
+body:graphql {
+  mutation SendNotification($input: SendNotificationInput!) {
+    sendNotification(input: $input) {
+      errors {
+        message
+      }
+      success
+    }
+  }
+}
+
+body:graphql:vars {
+  {
+    "input": {
+      "topic": "brh28-ATTENTION",
+      "title": "Hello",
+      "body": "This is a notification"
+    }
+  }
+}

dev/bruno/Flash GraphQL API/environments/flash-test.bru

@@ -0,0 +1,14 @@
+vars {
+  main_protocol: https
+  main_domain: api.test.flashapp.me
+  main_port: 4002
+  ~admin_url: http://localhost:4001/graphql
+  ~admin_token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOiJhZG1pbiIsInJvbGVzIjpbIkFjY291bnRzIE1hbmFnZXIiXX0.UOmQR2K6RdS1FVvQbjvSQfoQ-VsTC6Y7x2YAXZImdsA
+  currency: BTC
+  ~phone: +1301
+  ~code: 000000
+  token: 
+  walletId: 
+  walletIdUsd: c593736e-5a58-42e4-93fa-dc895856c1f1
+  graphqlUrl: https://api.test.flashapp.me/graphql
+}

dev/bruno/Flash GraphQL API/notoken/folder.bru

@@ -1,6 +1,11 @@
 meta {
   name: notoken
-  seq: 1
+}
+
+headers {
+  Accept: */*
+  Connection: keep-alive
+  Accept-Encoding: gzip, deflate, br
 }
 
 auth {

dev/bruno/Flash GraphQL API/notoken/queries/supportedBanks.bru

@@ -0,0 +1,24 @@
+meta {
+  name: supportedBanks
+  type: graphql
+  seq: 8
+}
+
+post {
+  url: {{graphqlUrl}}
+  body: graphql
+  auth: inherit
+}
+
+body:graphql {
+  query {
+      supportedBanks {
+        name            
+      }            
+    } 
+}
+
+settings {
+  encodeUrl: true
+  timeout: 0
+}

dev/config/base-config.yaml

@@ -51,3 +51,11 @@ frappe:
 
 sendgrid:
   apiKey: "<replace>"
+
+# FCM topic names for push notifications.
+# Replace "dev" with a unique identifier to avoid accidentally sending to other environments.
+notificationTopics:
+  - "dev-EMERGENCY"
+  - "dev-ATTENTION"
+  - "dev-INFO"
+  - "dev-MARKETING"

src/app/admin/index.ts

@@ -1,6 +1,6 @@
 export * from "./update-user-phone"
-export * from "./send-admin-push-notification"
-export * from "./send-broadcast-notification"
+// export * from "./send-admin-push-notification"
+// export * from "./send-broadcast-notification"
 
 import { checkedToAccountUuid, checkedToUsername } from "@domain/accounts"
 import { IdentityRepository } from "@services/kratos"

src/config/schema.ts

@@ -673,6 +673,12 @@ export const configSchema = {
       type: "object",
       required: ["apiKey"],
     },
+    notificationTopics: {
+      type: "array",
+      items: { type: "string" },
+      uniqueItems: true,
+      default: [],
+    },
   },
   required: [
     "lightningAddressDomain",

src/config/schema.types.d.ts

@@ -190,6 +190,7 @@ type YamlSchema = {
   }
   sendgrid: SendGridConfig
   frappe: FrappeConfig
+  notificationTopics: string[]
 }
 
 type FrappeCredentials = {

src/config/yaml.ts

@@ -269,6 +269,8 @@ export const getTestAccounts = (config = yamlConfig): TestAccount[] =>
 
 export const getCronConfig = (config = yamlConfig): CronConfig => config.cronConfig
 
+export const getNotificationTopics = (config = yamlConfig): string[] => config.notificationTopics
+
 export const getCaptcha = (config = yamlConfig): CaptchaConfig => config.captcha
 
 export const getRewardsConfig = (): RewardsConfig => {