feat: add opt-in event retention purge (#359)

Author: Justxd22

CONFIGURATION.md

@@ -109,6 +109,7 @@ Running `nostream` for the first time creates the settings file in `<project_roo
 | limits.event.rateLimits[].rate              | Maximum number of events during period. |
 | limits.event.whitelists.pubkeys             | List of public keys to ignore rate limits. |
 | limits.event.whitelists.ipAddresses         | List of IPs (IPv4 or IPv6) to ignore rate limits. |
+| limits.event.retentionDays                  | Number of days to retain events (based on created_at). Older events are purged. Defaults to -1 (disabled). Any non-positive number disables the feature. |
 | limits.client.subscription.maxSubscriptions | Maximum number of subscriptions per connected client. Defaults to 10. Disabled when set to zero. |
 | limits.client.subscription.maxFilters       | Maximum number of filters per subscription. Defaults to 10. Disabled when set to zero. |
 | limits.message.rateLimits[].period          | Rate limit period in milliseconds. |

resources/default-settings.yaml

@@ -76,6 +76,7 @@ limits:
       - "10.10.10.1"
       - "::ffff:10.10.10.1"
   event:
+    retentionDays: -1
     eventId:
       minLeadingZeroBits: 0
     kind:

src/@types/repositories.ts

@@ -17,6 +17,7 @@ export interface IEventRepository {
   upsert(event: Event): Promise<number>
   findByFilters(filters: SubscriptionFilter[]): IQueryResult<DBEvent[]>
   deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise<number>
+  deleteExpiredAndRetained(retentionDays?: number): Promise<number>
 }
 
 export interface IInvoiceRepository {

src/@types/services.ts

@@ -1,6 +1,10 @@
 import { Invoice } from './invoice'
 import { Pubkey } from './base'
 
+export interface IMaintenanceService {
+  clearOldEvents(): Promise<void>
+}
+
 export interface IPaymentsService {
   getInvoiceFromPaymentsProcessor(invoice: string | Invoice): Promise<Partial<Invoice>>
   createInvoice(

src/@types/settings.ts

@@ -77,6 +77,7 @@ export interface EventLimits {
   content?: ContentLimits | ContentLimits[]
   rateLimits?: EventRateLimit[]
   whitelists?: EventWhitelists
+  retentionDays?: number
 }
 
 export interface ClientSubscriptionLimits {

src/app/maintenance-worker.ts

@@ -1,10 +1,10 @@
+import { IMaintenanceService, IPaymentsService } from '../@types/services'
 import { mergeDeepLeft, path, pipe } from 'ramda'
 import { IRunnable } from '../@types/base'
 
 import { createLogger } from '../factories/logger-factory'
 import { delayMs } from '../utils/misc'
 import { InvoiceStatus } from '../@types/invoice'
-import { IPaymentsService } from '../@types/services'
 import { Settings } from '../@types/settings'
 
 const UPDATE_INVOICE_INTERVAL = 60000
@@ -17,6 +17,7 @@ export class MaintenanceWorker implements IRunnable {
   public constructor(
     private readonly process: NodeJS.Process,
     private readonly paymentsService: IPaymentsService,
+    private readonly maintenanceService: IMaintenanceService,
     private readonly settings: () => Settings,
   ) {
     this.process
@@ -34,6 +35,8 @@ export class MaintenanceWorker implements IRunnable {
   private async onSchedule(): Promise<void> {
     const currentSettings = this.settings()
 
+    await this.maintenanceService.clearOldEvents()
+
     if (!path(['payments','enabled'], currentSettings)) {
       return
     }

src/factories/maintenance-service-factory.ts

@@ -0,0 +1,11 @@
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createSettings } from './settings-factory'
+import { EventRepository } from '../repositories/event-repository'
+import { MaintenanceService } from '../services/maintenance-service'
+
+export const createMaintenanceService = () => {
+  return new MaintenanceService(
+    new EventRepository(getMasterDbClient(), getReadReplicaDbClient()),
+    createSettings
+  )
+}

src/factories/maintenance-worker-factory.ts

@@ -1,7 +1,13 @@
+import { createMaintenanceService } from './maintenance-service-factory'
 import { createPaymentsService } from './payments-service-factory'
 import { createSettings } from './settings-factory'
 import { MaintenanceWorker } from '../app/maintenance-worker'
 
 export const maintenanceWorkerFactory = () => {
-  return new MaintenanceWorker(process, createPaymentsService(), createSettings)
+  return new MaintenanceWorker(
+    process,
+    createPaymentsService(),
+    createMaintenanceService(),
+    createSettings
+  )
 }

src/repositories/event-repository.ts

@@ -251,4 +251,27 @@ export class EventRepository implements IEventRepository {
         deleted_at: this.masterDbClient.raw('now()'),
       })
   }
+
+  public deleteExpiredAndRetained(retentionDays?: number): Promise<number> {
+    const now = Math.floor(Date.now() / 1000)
+
+    if (typeof retentionDays !== 'number' || isNaN(retentionDays) || retentionDays <= 0) {
+      debug('skipping purge: retentionDays is not a positive number')
+      return Promise.resolve(0)
+    }
+
+    const retentionLimit = now - (retentionDays * 86400)
+
+    debug('deleting expired and retained events (retentionLimit: %d, now: %d)', retentionLimit, now)
+
+    return this.masterDbClient('events')
+      .where(function () {
+        this.where(function () {
+          this.whereNotNull('expires_at').andWhere('expires_at', '<', now)
+        })
+        .orWhereNotNull('deleted_at')
+        .orWhere('event_created_at', '<', retentionLimit)
+      })
+      .del()
+  }
 }

src/services/maintenance-service.ts

@@ -0,0 +1,32 @@
+import { createLogger } from '../factories/logger-factory'
+import { IEventRepository } from '../@types/repositories'
+import { IMaintenanceService } from '../@types/services'
+import { Settings } from '../@types/settings'
+
+const debug = createLogger('maintenance-service')
+
+export class MaintenanceService implements IMaintenanceService {
+  public constructor(
+    private readonly eventRepository: IEventRepository,
+    private readonly settings: () => Settings,
+  ) {}
+
+  public async clearOldEvents(): Promise<void> {
+    const currentSettings = this.settings()
+    const retentionDays = currentSettings.limits?.event?.retentionDays
+
+    if (typeof retentionDays !== 'number' || isNaN(retentionDays) || retentionDays <= 0) {
+      return
+    }
+
+    try {
+      debug('purging deleted, expired and old events')
+      const deletedCount = await this.eventRepository.deleteExpiredAndRetained(retentionDays)
+      if (deletedCount > 0) {
+        console.info(`[Maintenance] Deleted ${deletedCount} expired and retained events.`)
+      }
+    } catch (error) {
+      console.error('Unable to purge events. Reason:', error)
+    }
+  }
+}