feat: send notifications when account exceeded allowed resources

.env.default

@@ -17,13 +17,16 @@ AWS_REGION=us-east-1
 
 RPC_URL=https://rpc.decentraland.org/mainnet?project=worlds-content-server
 MARKETPLACE_SUBGRAPH_URL=https://api.thegraph.com/subgraphs/name/decentraland/marketplace
+BUILDER_URL=https://decentraland.org/builder
 ALLOW_ENS_DOMAINS=false
 
 SNS_ARN=
 AUTH_SECRET="setup_some_secret_here"
 
 LAMBDAS_URL=https://peer.decentraland.org/lambdas
 CONTENT_URL=https://peer.decentraland.org/content
+NOTIFICATION_SERVICE_URL=
+NOTIFICATION_SERVICE_TOKEN=
 
 ETH_NETWORK=mainnet
 GLOBAL_SCENES_URN=

src/adapters/notifications-service.ts

@@ -0,0 +1,56 @@
+import { AppComponents, Notification, INotificationService } from '../types'
+
+export async function createNotificationsClientComponent({
+  config,
+  fetch,
+  logs
+}: Pick<AppComponents, 'config' | 'fetch' | 'logs'>): Promise<INotificationService> {
+  const notificationServiceUrl = await config.getString('NOTIFICATION_SERVICE_URL')
+  if (!!notificationServiceUrl) {
+    return createHttpNotificationClient({ config, fetch, logs })
+  }
+
+  return createLogNotificationClient({ logs })
+}
+
+async function createHttpNotificationClient({
+  config,
+  fetch,
+  logs
+}: Pick<AppComponents, 'config' | 'fetch' | 'logs'>): Promise<INotificationService> {
+  const logger = logs.getLogger('http-notifications-client')
+  const [notificationServiceUrl, authToken] = await Promise.all([
+    config.getString('NOTIFICATION_SERVICE_URL'),
+    config.getString('NOTIFICATION_SERVICE_TOKEN')
+  ])
+
+  if (!!notificationServiceUrl && !authToken) {
+    throw new Error('Notification service URL provided without a token')
+  }
+  logger.info(`Using notification service at ${notificationServiceUrl}`)
+
+  async function sendNotifications(notifications: Notification[]): Promise<void> {
+    logger.info(`Sending ${notifications.length} notifications`, { notifications: JSON.stringify(notifications) })
+    await fetch.fetch(`${notificationServiceUrl}/notifications`, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        Authorization: `Bearer ${authToken}`
+      },
+      body: JSON.stringify(notifications)
+    })
+  }
+
+  return {
+    sendNotifications
+  }
+}
+
+async function createLogNotificationClient({ logs }: Pick<AppComponents, 'logs'>): Promise<INotificationService> {
+  const logger = logs.getLogger('log-notifications-client')
+  return {
+    async sendNotifications(notifications: Notification[]): Promise<void> {
+      logger.info(`Sending ${notifications.length} notifications`, { notifications: JSON.stringify(notifications) })
+    }
+  }
+}

src/adapters/update-owner-job.ts

@@ -1,16 +1,20 @@
-import { AppComponents, IRunnable, Whitelist, WorldRecord } from '../types'
+import { AppComponents, BlockedRecord, IRunnable, Notification, TWO_DAYS_IN_MS, Whitelist, WorldRecord } from '../types'
 import SQL from 'sql-template-strings'
 import { CronJob } from 'cron'
 
 type WorldData = Pick<WorldRecord, 'name' | 'owner' | 'size' | 'entity'>
 
 export async function createUpdateOwnerJob(
-  components: Pick<AppComponents, 'config' | 'database' | 'fetch' | 'logs' | 'nameOwnership' | 'walletStats'>
+  components: Pick<
+    AppComponents,
+    'config' | 'database' | 'fetch' | 'logs' | 'nameOwnership' | 'notificationService' | 'walletStats'
+  >
 ): Promise<IRunnable<void>> {
   const { config, fetch, logs } = components
   const logger = logs.getLogger('update-owner-job')
 
   const whitelistUrl = await config.requireString('WHITELIST_URL')
+  const builderUrl = await config.requireString('BUILDER_URL')
 
   function dumpMap(mapName: string, worldWithOwners: ReadonlyMap<string, any>) {
     for (const [key, value] of worldWithOwners) {
@@ -24,17 +28,91 @@ export async function createUpdateOwnerJob(
         VALUES (${wallet.toLowerCase()}, ${new Date()}, ${new Date()})
         ON CONFLICT (wallet)
             DO UPDATE SET updated_at = ${new Date()}
+        RETURNING wallet, created_at, updated_at
     `
-    await components.database.query(sql)
+    const result = await components.database.query<BlockedRecord>(sql)
+    if (result.rowCount > 0) {
+      const { warning, blocked } = result.rows.reduce(
+        (r, o) => {
+          if (o.updated_at.getTime() - o.created_at.getTime() < TWO_DAYS_IN_MS) {
+            r.warning.push(o)
+          } else {
+            r.blocked.push(o)
+          }
+          return r
+        },
+        { warning: [] as BlockedRecord[], blocked: [] as BlockedRecord[] }
+      )
+
+      const notifications: Notification[] = []
+
+      logger.info(
+        `Sending notifications for wallets that are about to be blocked: ${warning.map((r) => r.wallet).join(', ')}`
+      )
+      notifications.push(
+        ...warning.map(
+          (record): Notification => ({
+            type: 'worlds_missing_resources',
+            eventKey: `detected-${record.created_at.toISOString().slice(0, 10)}`,
+            address: record.wallet,
+            metadata: {
+              title: 'Missing Resources',
+              description: 'World access at risk in 48hs. Rectify now to prevent disruption.',
+              url: `${builderUrl}/worlds?tab=dcl`,
+              when: record.created_at.getTime() + TWO_DAYS_IN_MS
+            },
+            timestamp: record.created_at.getTime()
+          })
+        )
+      )
+
+      logger.info(
+        `Sending notifications for wallets that have already been blocked: ${blocked.map((r) => r.wallet).join(', ')}`
+      )
+      notifications.push(
+        ...blocked.map(
+          (record): Notification => ({
+            type: 'worlds_access_restricted',
+            eventKey: `detected-${record.created_at.toISOString().slice(0, 10)}`,
+            address: record.wallet,
+            metadata: {
+              title: 'Worlds restricted',
+              description: 'Access to your Worlds has been restricted due to insufficient resources.',
+              url: `${builderUrl}/worlds?tab=dcl`,
+              when: record.created_at.getTime() + TWO_DAYS_IN_MS
+            },
+            timestamp: record.created_at.getTime() + TWO_DAYS_IN_MS
+          })
+        )
+      )
+      await components.notificationService.sendNotifications(notifications)
+    }
   }
 
   async function clearOldBlockingRecords(startDate: Date) {
     const sql = SQL`
         DELETE
         FROM blocked
         WHERE updated_at < ${startDate}
+        RETURNING wallet, created_at
     `
-    await components.database.query(sql)
+    const result = await components.database.query(sql)
+    if (result.rowCount > 0) {
+      logger.info(`Sending block removal notifications for wallets: ${result.rows.map((row) => row.wallet).join(', ')}`)
+      await components.notificationService.sendNotifications(
+        result.rows.map((record) => ({
+          type: 'worlds_access_restored',
+          eventKey: `detected-${record.created_at.toISOString().slice(0, 10)}`,
+          address: record.wallet,
+          metadata: {
+            title: 'Worlds available',
+            description: 'Access to your Worlds has been restored.',
+            url: `${builderUrl}/worlds?tab=dcl`
+          },
+          timestamp: Date.now()
+        }))
+      )
+    }
   }
 
   async function run() {
@@ -93,15 +171,13 @@ export async function createUpdateOwnerJob(
     }
     dumpMap('worldsByOwner', worldsByOwner)
 
+    const whiteList = await fetch.fetch(whitelistUrl).then(async (data) => (await data.json()) as unknown as Whitelist)
+
     for (const [owner, worlds] of worldsByOwner) {
       if (worlds.length === 0) {
         continue
       }
 
-      const whiteList = await fetch
-        .fetch(whitelistUrl)
-        .then(async (data) => (await data.json()) as unknown as Whitelist)
-
       const walletStats = await components.walletStats.get(owner)
 
       // The size of whitelisted worlds does not count towards the wallet's used space

src/components.ts

@@ -31,6 +31,7 @@ import { createUpdateOwnerJob } from './adapters/update-owner-job'
 import { createSnsClient } from './adapters/sns-client'
 import { createAwsConfig } from './adapters/aws-config'
 import { S3 } from 'aws-sdk'
+import { createNotificationsClientComponent } from './adapters/notifications-service'
 
 // Initialize all the components of the app
 export async function initComponents(): Promise<AppComponents> {
@@ -131,12 +132,15 @@ export async function initComponents(): Promise<AppComponents> {
 
   const migrationExecutor = createMigrationExecutor({ logs, database: database, nameOwnership, storage, worldsManager })
 
+  const notificationService = await createNotificationsClientComponent({ config, fetch, logs })
+
   const updateOwnerJob = await createUpdateOwnerJob({
     config,
     database,
     fetch,
     logs,
     nameOwnership,
+    notificationService,
     walletStats
   })
 
@@ -156,6 +160,7 @@ export async function initComponents(): Promise<AppComponents> {
     nameDenyListChecker,
     nameOwnership,
     namePermissionChecker,
+    notificationService,
     permissionsManager,
     server,
     snsClient,

src/logic/blocked.ts

@@ -1,12 +1,10 @@
-import { WorldMetadata } from '../types'
+import { TWO_DAYS_IN_MS, WorldMetadata } from '../types'
 import { NotAuthorizedError } from '@dcl/platform-server-commons'
 
-const TWO_DAYS = 2 * 24 * 60 * 60 * 1000
-
 export function assertNotBlockedOrWithinInGracePeriod(worldMetadata: WorldMetadata) {
   if (worldMetadata.blockedSince) {
     const now = new Date()
-    if (now.getTime() - worldMetadata.blockedSince.getTime() > TWO_DAYS) {
+    if (now.getTime() - worldMetadata.blockedSince.getTime() > TWO_DAYS_IN_MS) {
       throw new NotAuthorizedError(
         `World "${worldMetadata.runtimeMetadata.name}" has been blocked since ${worldMetadata.blockedSince} as it exceeded its allowed storage space.`
       )

src/types.ts

@@ -165,6 +165,18 @@ export type IPermissionsManager = {
   deleteAddressFromAllowList(worldName: string, permission: Permission, address: string): Promise<void>
 }
 
+export type INotificationService = {
+  sendNotifications(notifications: Notification[]): Promise<void>
+}
+
+export type Notification = {
+  eventKey: string
+  type: string
+  address?: string
+  metadata: object
+  timestamp: number
+}
+
 export enum PermissionType {
   Unrestricted = 'unrestricted',
   SharedSecret = 'shared-secret',
@@ -263,6 +275,7 @@ export type BaseComponents = {
   nameDenyListChecker: INameDenyListChecker
   nameOwnership: INameOwnership
   namePermissionChecker: IWorldNamePermissionChecker
+  notificationService: INotificationService
   permissionsManager: IPermissionsManager
   server: IHttpServerComponent<GlobalContext>
   snsClient: SnsClient
@@ -350,3 +363,7 @@ export type WorldRecord = {
   updated_at: Date
   blocked_since: Date | null
 }
+
+export type BlockedRecord = { wallet: string; created_at: Date; updated_at: Date }
+
+export const TWO_DAYS_IN_MS = 2 * 24 * 60 * 60 * 1000

test/mocks/world-creator.ts

@@ -5,7 +5,6 @@ import { TextDecoder } from 'util'
 import { getIdentity, makeid, storeJson } from '../utils'
 import { Authenticator, AuthIdentity } from '@dcl/crypto'
 import { defaultPermissions } from '../../src/logic/permissions-checker'
-import { hashV1 } from '@dcl/hashing'
 import { bufferToStream } from '@dcl/catalyst-storage'
 
 export function createWorldCreator({