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 disabled. |
 | 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,8 @@ limits:
       - "10.10.10.1"
       - "::ffff:10.10.10.1"
   event:
+    # Uncomment the next line to enable event retention and set the number of days
+    # retentionDays: 30
     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/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

@@ -3,6 +3,7 @@ import { IRunnable } from '../@types/base'
 
 import { createLogger } from '../factories/logger-factory'
 import { delayMs } from '../utils/misc'
+import { IEventRepository } from '../@types/repositories'
 import { InvoiceStatus } from '../@types/invoice'
 import { IPaymentsService } from '../@types/services'
 import { Settings } from '../@types/settings'
@@ -17,6 +18,7 @@ export class MaintenanceWorker implements IRunnable {
   public constructor(
     private readonly process: NodeJS.Process,
     private readonly paymentsService: IPaymentsService,
+    private readonly eventRepository: IEventRepository,
     private readonly settings: () => Settings,
   ) {
     this.process
@@ -34,6 +36,20 @@ export class MaintenanceWorker implements IRunnable {
   private async onSchedule(): Promise<void> {
     const currentSettings = this.settings()
 
+    const retentionDays = currentSettings.limits?.event?.retentionDays
+
+    if (typeof retentionDays === 'number' && retentionDays > 0) {
+      try {
+        debug('purging deleted and expired 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)
+      }
+    }
+
     if (!path(['payments','enabled'], currentSettings)) {
       return
     }

src/factories/maintenance-worker-factory.ts

@@ -1,7 +1,14 @@
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
 import { createPaymentsService } from './payments-service-factory'
 import { createSettings } from './settings-factory'
+import { EventRepository } from '../repositories/event-repository'
 import { MaintenanceWorker } from '../app/maintenance-worker'
 
 export const maintenanceWorkerFactory = () => {
-  return new MaintenanceWorker(process, createPaymentsService(), createSettings)
+  return new MaintenanceWorker(
+    process,
+    createPaymentsService(),
+    new EventRepository(getMasterDbClient(), getReadReplicaDbClient()),
+    createSettings
+  )
 }

src/repositories/event-repository.ts

@@ -251,4 +251,20 @@ export class EventRepository implements IEventRepository {
         deleted_at: this.masterDbClient.raw('now()'),
       })
   }
+
+  public deleteExpiredAndRetained(retentionDays?: number): Promise<number> {
+    debug('deleting expired and retained events (retentionDays: %d)', retentionDays)
+
+    if (!(typeof retentionDays === 'number' && retentionDays > 0)) {
+      return this.masterDbClient('events').whereRaw('1 = 0').del()
+    }
+
+    const retentionLimit = Math.floor(Date.now() / 1000) - (retentionDays * 86400)
+
+    return this.masterDbClient('events')
+      .where('expires_at', '<', Math.floor(Date.now() / 1000))
+      .orWhereNotNull('deleted_at')
+      .orWhere('event_created_at', '<', retentionLimit)
+      .del()
+  }
 }

test/unit/app/maintenance-worker.spec.ts

@@ -0,0 +1,134 @@
+import * as chai from 'chai'
+import * as sinon from 'sinon'
+import sinonChai from 'sinon-chai'
+
+chai.use(sinonChai)
+
+const { expect } = chai
+
+import { IEventRepository } from '../../../src/@types/repositories'
+import { IPaymentsService } from '../../../src/@types/services'
+import { MaintenanceWorker } from '../../../src/app/maintenance-worker'
+import { Settings } from '../../../src/@types/settings'
+
+describe('MaintenanceWorker', () => {
+  let worker: MaintenanceWorker
+  let sandbox: sinon.SinonSandbox
+  let paymentsService: sinon.SinonStubbedInstance<IPaymentsService>
+  let eventRepository: sinon.SinonStubbedInstance<IEventRepository>
+  let settings: sinon.SinonStub
+  let processMock: any
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox()
+    paymentsService = {
+      getPendingInvoices: sandbox.stub(),
+      getInvoiceFromPaymentsProcessor: sandbox.stub(),
+      updateInvoiceStatus: sandbox.stub(),
+      confirmInvoice: sandbox.stub(),
+      sendInvoiceUpdateNotification: sandbox.stub(),
+    } as any
+    eventRepository = {
+      deleteExpiredAndRetained: sandbox.stub(),
+    } as any
+    settings = sandbox.stub()
+    processMock = {
+      on: sandbox.stub().returnsThis(),
+    }
+
+    worker = new MaintenanceWorker(
+      processMock as any,
+      paymentsService as any,
+      eventRepository as any,
+      settings as any,
+    )
+  })
+
+  afterEach(() => {
+    sandbox.restore()
+  })
+
+  describe('onSchedule', () => {
+    it('purges events and processes invoices', async () => {
+      const currentSettings: Settings = {
+        info: {} as any,
+        network: {} as any,
+        payments: {
+          enabled: true,
+        } as any,
+        limits: {
+          event: {
+            retentionDays: 30,
+          },
+        } as any,
+      }
+      settings.returns(currentSettings)
+      eventRepository.deleteExpiredAndRetained.resolves(10)
+      paymentsService.getPendingInvoices.resolves([])
+
+      await (worker as any).onSchedule()
+
+      expect(eventRepository.deleteExpiredAndRetained).to.have.been.calledOnceWithExactly(30)
+      expect(paymentsService.getPendingInvoices).to.have.been.calledOnce
+    })
+
+    it('purges events even if payments are disabled', async () => {
+      const currentSettings: Settings = {
+        info: {} as any,
+        network: {} as any,
+        payments: {
+          enabled: false,
+        } as any,
+        limits: {
+          event: {
+            retentionDays: 30,
+          },
+        } as any,
+      }
+      settings.returns(currentSettings)
+      eventRepository.deleteExpiredAndRetained.resolves(10)
+
+      await (worker as any).onSchedule()
+
+      expect(eventRepository.deleteExpiredAndRetained).to.have.been.calledOnceWithExactly(30)
+      expect(paymentsService.getPendingInvoices).not.to.have.been.called
+    })
+
+    it('does not purge when retentionDays is not configured', async () => {
+      const currentSettings: Settings = {
+        info: {} as any,
+        network: {} as any,
+        payments: {
+          enabled: false,
+        } as any,
+      }
+      settings.returns(currentSettings)
+
+      await (worker as any).onSchedule()
+
+      expect(eventRepository.deleteExpiredAndRetained).not.to.have.been.called
+    })
+
+    it('handles error during purge when retentionDays is configured', async () => {
+      const currentSettings: Settings = {
+        info: {} as any,
+        network: {} as any,
+        payments: {
+          enabled: false,
+        } as any,
+        limits: {
+          event: {
+            retentionDays: 30,
+          },
+        } as any,
+      }
+      settings.returns(currentSettings)
+      eventRepository.deleteExpiredAndRetained.rejects(new Error('DB Error'))
+
+      // Should not throw
+      await (worker as any).onSchedule()
+
+      expect(eventRepository.deleteExpiredAndRetained).to.have.been.calledOnceWithExactly(30)
+    })
+  })
+})

test/unit/repositories/event-repository.spec.ts

@@ -447,6 +447,29 @@ describe('EventRepository', () => {
     })
   })
 
+  describe('deleteExpiredAndRetained', () => {
+    let clock: sinon.SinonFakeTimers
+    beforeEach(() => {
+      clock = sinon.useFakeTimers(1000000000) // 1970-01-12T13:46:40.000Z
+    })
+
+    afterEach(() => {
+      clock.restore()
+    })
+
+    it('does not delete anything when retentionDays is not set', () => {
+      const query = repository.deleteExpiredAndRetained().toString()
+
+      expect(query).to.equal('delete from "events" where 1 = 0')
+    })
+
+    it('deletes expired, deleted and old events when retentionDays is set', () => {
+      const query = repository.deleteExpiredAndRetained(7).toString()
+
+      expect(query).to.equal('delete from "events" where "expires_at" < 1000000 or "deleted_at" is not null or "event_created_at" < 395200')
+    })
+  })
+
   describe('upsert', () => {
     it('replaces event based on event_pubkey and event_kind', () => {
       const event: Event = {