feat(backend): periodically clean up expired rows

interledger · Jul 22, 2022 · 4c684b9 · 4c684b9
1 parent 8343f66
commit 4c684b9
Show file tree

Hide file tree

Showing 2 changed files with 90 additions and 1 deletion.
diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts
@@ -5,7 +5,7 @@ import { EventEmitter } from 'events'
 import { ParsedUrlQuery } from 'querystring'
 
 import { IocContract } from '@adonisjs/fold'
-import Knex from 'knex'
+import Knex, { QueryInterface } from 'knex'
 import Koa, { DefaultState } from 'koa'
 import bodyParser from 'koa-bodyparser'
 import { Logger } from 'pino'
@@ -100,6 +100,28 @@ export type ListContext = Context<
   AppRequest<'accountId', never, PageQueryParams>
 >
 
+export interface DatabaseCleanupRule {
+  /**
+   * the name of the column containing the starting time from which the age will be computed
+   * ex: `createdAt` or `updatedAt`
+   */
+  absoluteStartTimeColumnName: string
+  /**
+   * the column which will be used to either set or offset the computed age
+   * if not provided, rows are considered expired when the difference between the current time
+   * and the time specified in `absoluteStartTimeColumnName` is greater than or equal to `minLapseTimeMillis`
+   */
+  lapseTime?: {
+    columnName: string
+    absolute?: boolean
+  }
+  /**
+   * the minimum number of milliseconds since expiration before rows of this table will be
+   * considered safe to delete during clean up
+   */
+  minLapseTimeMillis: number
+}
+
 type ContextType<T> = T extends (
   ctx: infer Context
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -150,6 +172,9 @@ export class App {
   private config!: IAppConfig
   private outgoingPaymentTimer!: NodeJS.Timer
   private deactivateInvoiceTimer!: NodeJS.Timer
+  private databaseCleanupRules!: {
+    [tableName: string]: DatabaseCleanupRule | undefined
+  }
 
   public constructor(private container: IocContract<AppServices>) {}
 
@@ -169,6 +194,16 @@ export class App {
     this.koa.context.closeEmitter = await this.container.use('closeEmitter')
     this.messageProducer = await this.container.use('messageProducer')
     this.publicRouter = new Router()
+    this.databaseCleanupRules = {
+      accessTokens: {
+        absoluteStartTimeColumnName: 'updatedAt',
+        lapseTime: {
+          columnName: 'expiresIn',
+          absolute: false
+        },
+        minLapseTimeMillis: this.config.accessTokenCleanupMinAge
+      }
+    }
 
     this.koa.use(
       async (
@@ -205,6 +240,9 @@ export class App {
       for (let i = 0; i < this.config.webhookWorkers; i++) {
         process.nextTick(() => this.processWebhook())
       }
+      for (let i = 0; i < this.config.databaseCleanupWorkers; i++) {
+        process.nextTick(() => this.processDatabaseCleanup())
+      }
     }
   }
 
@@ -478,4 +516,52 @@ export class App {
           ).unref()
       })
   }
+
+  private isDatabaseRowExpired<TRecord, TResult>(
+    now: number,
+    row: QueryInterface<TRecord, TResult>,
+    rule: DatabaseCleanupRule
+  ): boolean {
+    // get the base time used to compute the row's age
+    let absoluteLapseTime = row.column[rule.absoluteStartTimeColumnName]
+
+    if (rule.lapseTime) {
+      const timeParameter =
+        row.column[rule.lapseTime.columnName] || rule.minLapseTimeMillis
+      if (rule.lapseTime.absolute) {
+        absoluteLapseTime = timeParameter
+      } else {
+        // offset the working base time by a time span stored in another column
+        absoluteLapseTime += timeParameter
+      }
+    } else {
+      absoluteLapseTime =
+        row.column[rule.absoluteStartTimeColumnName] + rule.minLapseTimeMillis
+    }
+
+    return now - absoluteLapseTime >= rule.minLapseTimeMillis
+  }
+
+  private async processDatabaseCleanup(): Promise<void> {
+    const knex = await this.container.use('knex')
+
+    const tableNames = Object.keys(this.databaseCleanupRules)
+    for (const tableName of tableNames) {
+      const rule = this.databaseCleanupRules[tableName]
+      if (rule) {
+        const now = Date.now()
+
+        try {
+          await knex(tableName)
+            .where((row) => this.isDatabaseRowExpired(now, row, rule))
+            .delete()
+        } catch (err) {
+          this.logger.warn(
+            { error: err.message, tableName },
+            'processDatabaseCleanup error'
+          )
+        }
+      }
+    }
+  }
 }
diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts
@@ -89,6 +89,7 @@ export const Config = {
   webhookWorkerIdle: envInt('WEBHOOK_WORKER_IDLE', 200), // milliseconds
   webhookUrl: envString('WEBHOOK_URL', 'http://127.0.0.1:4001/webhook'),
   webhookTimeout: envInt('WEBHOOK_TIMEOUT', 2000), // milliseconds
+  databaseCleanupWorkers: envInt('DATABASE_CLEANUP_WORKERS', 1),
 
   withdrawalThrottleDelay:
     process.env.WITHDRAWAL_THROTTLE_DELAY == null
@@ -107,6 +108,8 @@ export const Config = {
     'https://raw.githubusercontent.com/interledger/open-payments/ab840c8ff904a4b8c45d94ac23f5518a79a67686/auth-server-open-api-spec.yaml'
   ),
 
+  accessTokenCleanupMinAge: 1000 * 60 * 60 * 24 * 30, // 30 days
+
   /** Frontend **/
   frontendUrl: envString('FRONTEND_URL', 'http://localhost:3000')
 }