swap from postgres to elastic for analytics

package.json

@@ -48,6 +48,7 @@
     "typescript": "^4.5.5"
   },
   "dependencies": {
+    "@elastic/elasticsearch": "^7.17.0",
     "@sentry/integrations": "^6.16.1",
     "@sentry/node": "^6.16.1",
     "@thesharks/jagtag-js": "^2.0.0",

src/database/migrations/exec/1644249527-remove_analytics.ts

@@ -0,0 +1,19 @@
+import driver from '../../driver'
+
+export async function up (db: typeof driver): Promise<void> {
+  await db`DROP TABLE IF EXISTS analytics;`
+}
+
+export async function down (db: typeof driver): Promise<void> {
+  await db`
+    CREATE TABLE IF NOT EXISTS analytics (
+      id UUID NOT NULL PRIMARY KEY,
+      timestamp TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW(),
+      type INTEGER NOT NULL,
+      guild_id BIGINT,
+      user_id BIGINT NOT NULL,
+      data JSONB NOT NULL,
+      name TEXT NOT NULL
+    );
+  `
+}

src/interactions/base.ts

@@ -1,31 +1,23 @@
-import { randomUUID } from 'crypto'
 import { Constants, Interaction, Structures } from 'detritus-client'
-import db from '../database/driver'
+import { add, flush } from '../utils/elastic'
 import { translate } from '../utils/i18n'
 import { error } from '../utils/logger'
 const { ApplicationCommandTypes, ApplicationCommandOptionTypes, MessageFlags, Permissions } = Constants
 
 export class BaseInteractionCommand<ParsedArgsFinished = Interaction.ParsedArgs> extends Interaction.InteractionCommand<ParsedArgsFinished> {
   async onSuccess (context: Interaction.InteractionContext, args: ParsedArgsFinished): Promise<void> {
-    await db`
-      INSERT INTO analytics (
-        id,
-        timestamp,
-        type,
-        guild_id,
-        user_id,
-        data,
-        name
-      ) VALUES (
-        ${randomUUID()},
-        NOW(),
-        ${context.interaction.type},
-        ${context.guildId ?? 0},
-        ${context.user.id},
-        ${JSON.stringify(args)},
-        ${context.command.name}
-      );
-    `
+    add({
+      type: 'command',
+      guildId: context.guildId,
+      inDm: context.inDm,
+      userId: context.user.id,
+      data: {
+        args,
+        interaction_id: context.interactionId
+      },
+      name: context.command.name
+    })
+    await flush()
   }
 
   async onDmBlocked (context: Interaction.InteractionContext): Promise<unknown> {
@@ -84,15 +76,6 @@ export class BaseInteractionCommand<ParsedArgsFinished = Interaction.ParsedArgs>
       flags: MessageFlags.EPHEMERAL
     })
   }
-
-  async safeReply (context: Interaction.InteractionContext, message: string | Structures.InteractionEditOrRespond): Promise<any> {
-    return await context.editOrRespond({
-      ...((typeof message === 'string') ? { content: message } : message),
-      allowedMentions: {
-        parse: []
-      }
-    })
-  }
 }
 
 export class BaseCommandOption<ParsedArgsFinished = Interaction.ParsedArgs> extends Interaction.InteractionCommandOption<ParsedArgsFinished> {

src/jobs/elastic-flush.client.ts

@@ -0,0 +1,13 @@
+import { ScheduledJob } from './base'
+import { flush } from '../utils/elastic'
+
+const job = new ScheduledJob('analytics-prune')
+  // every hour
+  .setInterval(60 * 60 * 1000)
+  .setExec(async () => {
+    await flush()
+  })
+
+if (process.env.ELASTIC_URL !== undefined) job.start()
+
+export default job

src/utils/elastic.ts

@@ -0,0 +1,60 @@
+import { Client } from '@elastic/elasticsearch'
+import { BulkStats } from '@elastic/elasticsearch/lib/Helpers'
+import { format } from 'date-fns'
+import discord from '../structures/client'
+import { debug, trace, warn } from './logger'
+
+const storage = new Set()
+
+export function add (data: any): number {
+  data['@timestamp'] = new Date().toISOString()
+  storage.add(data)
+  return storage.size
+}
+
+export function drop (): number {
+  storage.clear()
+  return storage.size
+}
+
+export async function * drain (): AsyncGenerator<any> {
+  for (const data of storage) {
+    storage.delete(data)
+    yield data
+  }
+}
+
+export async function flush (): Promise<BulkStats> {
+  if (process.env.ELASTIC_URL === undefined) throw new Error('Trying to flush to Elasticsearch, but no URL is set.')
+  debug('Flushing elasticsearch', 'Elastic')
+  const client = new Client({
+    node: process.env.ELASTIC_URL,
+    // auth can be either be a b64 encoded api key, or basic auth incorporated into the URL
+    ...(process.env.ELASTIC_API_KEY !== undefined
+      ? {
+          auth: {
+            apiKey: process.env.ELASTIC_API_KEY
+          }
+        }
+      : {})
+  })
+  const result = await client.helpers.bulk({
+    datasource: drain(),
+    onDocument (doc) {
+      return {
+        create: {
+          _index: `wildbeast-${discord.client.applicationId}-${format(new Date(), 'yyyy-MM-dd')}`
+        }
+      }
+    }
+  })
+  debug(`Flushed ${result.successful} analytics documents to Elastic, ${result.failed} failed`, 'Elastic')
+  trace(result, 'Elastic')
+  if (result.failed > 0) {
+    warn(`Failed to flush ${result.failed} analytics documents to Elastic`, 'Elastic')
+  }
+  // Persistent connection to elastic isn't necessary, close it
+  // we recreate the client on the next flush anyway
+  await client.close()
+  return result
+}