Moving around processors to separate audit logs out of central event handling.

packages/backend-core/src/events/events.ts

@@ -1,38 +1,12 @@
-import {
-  AuditLogFn,
-  Event,
-  IdentityType,
-  AuditedEventFriendlyName,
-  AuditLogQueueEvent,
-} from "@budibase/types"
+import { Event, AuditedEventFriendlyName } from "@budibase/types"
 import { processors } from "./processors"
 import identification from "./identification"
-import { getAppId } from "../context"
 import * as backfill from "./backfill"
-import { createQueue, JobQueue } from "../queue"
-import BullQueue from "bull"
 
 export function isAudited(event: Event) {
   return !!AuditedEventFriendlyName[event]
 }
 
-let auditLogsEnabled = false
-let auditLogQueue: BullQueue.Queue<AuditLogQueueEvent>
-
-export const configure = (fn: AuditLogFn) => {
-  auditLogsEnabled = true
-  const writeAuditLogs = fn
-  auditLogQueue = createQueue<AuditLogQueueEvent>(JobQueue.AUDIT_LOG)
-  return auditLogQueue.process(async job => {
-    await writeAuditLogs(job.data.event, job.data.properties, {
-      userId: job.data.opts.userId,
-      timestamp: job.data.opts.timestamp,
-      appId: job.data.opts.appId,
-      hostInfo: job.data.opts.hostInfo,
-    })
-  })
-}
-
 export const publishEvent = async (
   event: Event,
   properties: any,
@@ -45,22 +19,6 @@ export const publishEvent = async (
   // no backfill - send the event and exit
   if (!backfilling) {
     await processors.processEvent(event, identity, properties, timestamp)
-    if (auditLogsEnabled && isAudited(event)) {
-      // only audit log actual events, don't include backfills
-      const userId =
-        identity.type === IdentityType.USER ? identity.id : undefined
-      // add to the event queue, rather than just writing immediately
-      await auditLogQueue.add({
-        event,
-        properties,
-        opts: {
-          userId,
-          timestamp,
-          appId: getAppId(),
-          hostInfo: identity.hostInfo,
-        },
-      })
-    }
     return
   }
 

packages/backend-core/src/events/index.ts

@@ -3,7 +3,7 @@ export * as processors from "./processors"
 export * as analytics from "./analytics"
 export { default as identification } from "./identification"
 export * as backfillCache from "./backfill"
-export { configure, isAudited } from "./events"
+export { isAudited } from "./events"
 
 import { processors } from "./processors"
 

packages/backend-core/src/events/processors/AuditLogsProcessor.ts

@@ -0,0 +1,71 @@
+import {
+  Event,
+  Identity,
+  Group,
+  IdentityType,
+  AuditLogQueueEvent,
+  AuditLogFn,
+} from "@budibase/types"
+import { EventProcessor } from "./types"
+import { getAppId } from "../../context"
+import { isAudited } from "../events"
+import BullQueue from "bull"
+import { createQueue, JobQueue } from "../../queue"
+
+export default class AuditLogsProcessor implements EventProcessor {
+  static auditLogsEnabled = false
+  static auditLogQueue: BullQueue.Queue<AuditLogQueueEvent>
+
+  // can't use constructor as need to return promise
+  static init(fn: AuditLogFn) {
+    AuditLogsProcessor.auditLogsEnabled = true
+    const writeAuditLogs = fn
+    AuditLogsProcessor.auditLogQueue = createQueue<AuditLogQueueEvent>(
+      JobQueue.AUDIT_LOG
+    )
+    return AuditLogsProcessor.auditLogQueue.process(async job => {
+      await writeAuditLogs(job.data.event, job.data.properties, {
+        userId: job.data.opts.userId,
+        timestamp: job.data.opts.timestamp,
+        appId: job.data.opts.appId,
+        hostInfo: job.data.opts.hostInfo,
+      })
+    })
+  }
+
+  async processEvent(
+    event: Event,
+    identity: Identity,
+    properties: any,
+    timestamp?: string
+  ): Promise<void> {
+    if (AuditLogsProcessor.auditLogsEnabled && isAudited(event)) {
+      // only audit log actual events, don't include backfills
+      const userId =
+        identity.type === IdentityType.USER ? identity.id : undefined
+      // add to the event queue, rather than just writing immediately
+      await AuditLogsProcessor.auditLogQueue.add({
+        event,
+        properties,
+        opts: {
+          userId,
+          timestamp,
+          appId: getAppId(),
+          hostInfo: identity.hostInfo,
+        },
+      })
+    }
+  }
+
+  async identify(identity: Identity, timestamp?: string | number) {
+    // no-op
+  }
+
+  async identifyGroup(group: Group, timestamp?: string | number) {
+    // no-op
+  }
+
+  shutdown(): void {
+    AuditLogsProcessor.auditLogQueue.close()
+  }
+}

packages/backend-core/src/events/processors/index.ts

@@ -1,8 +1,19 @@
 import AnalyticsProcessor from "./AnalyticsProcessor"
 import LoggingProcessor from "./LoggingProcessor"
+import AuditLogsProcessor from "./AuditLogsProcessor"
 import Processors from "./Processors"
+import { AuditLogFn } from "@budibase/types"
 
 export const analyticsProcessor = new AnalyticsProcessor()
 const loggingProcessor = new LoggingProcessor()
+const auditLogsProcessor = new AuditLogsProcessor()
 
-export const processors = new Processors([analyticsProcessor, loggingProcessor])
+export function init(auditingFn: AuditLogFn) {
+  return AuditLogsProcessor.init(auditingFn)
+}
+
+export const processors = new Processors([
+  analyticsProcessor,
+  loggingProcessor,
+  auditLogsProcessor,
+])

packages/server/src/startup.ts

@@ -127,7 +127,7 @@ export async function startup(app?: any, server?: any) {
   let queuePromises = []
   // configure events to use the pro audit log write
   // can't integrate directly into backend-core due to cyclic issues
-  queuePromises.push(events.configure(proSdk.auditLogs.write))
+  queuePromises.push(events.processors.init(proSdk.auditLogs.write))
   queuePromises.push(automations.init())
   queuePromises.push(initPro())
   if (app) {

packages/types/src/sdk/auditLogs.ts

@@ -1,4 +1,5 @@
 import { Event, HostInfo } from "./events"
+import { AuditLogDoc } from "../documents"
 
 export type AuditWriteOpts = {
   appId?: string
@@ -11,7 +12,7 @@ export type AuditLogFn = (
   event: Event,
   metadata: any,
   opts: AuditWriteOpts
-) => Promise<void>
+) => Promise<AuditLogDoc | undefined>
 
 export type AuditLogQueueEvent = {
   event: Event

packages/worker/src/index.ts

@@ -14,7 +14,7 @@ import Application from "koa"
 import { bootstrap } from "global-agent"
 import * as db from "./db"
 import { auth, logging, events, middleware } from "@budibase/backend-core"
-import { sdk } from "@budibase/pro"
+import { sdk as proSdk, sdk } from "@budibase/pro"
 db.init()
 import Koa from "koa"
 import koaBody from "koa-body"
@@ -30,7 +30,7 @@ import destroyable from "server-destroy"
 
 // configure events to use the pro audit log write
 // can't integrate directly into backend-core due to cyclic issues
-events.configure(sdk.auditLogs.write)
+events.processors.init(proSdk.auditLogs.write)
 
 if (env.ENABLE_SSO_MAINTENANCE_MODE) {
   console.warn(