feat(payments): add per-plan event rate limits and retention

.changeset/subscription-tier-limits.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(payments): add per-plan event rate limits and retention

CONFIGURATION.md

@@ -256,8 +256,8 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | payments.subscriptionPlans[].intervalSeconds | Length of one paid period, in seconds. |
 | payments.subscriptionPlans[].graceSeconds   | How long access lasts after a period ends without a renewal, in seconds. `0` ends access straight away. |
 | payments.subscriptionPlans[].description    | Optional text shown to users. |
-| payments.subscriptionPlans[].limits.rateLimits | Optional event rate limits for subscribers on this plan, in the same format as `limits.event.rateLimits`. |
-| payments.subscriptionPlans[].limits.retentionMaxDays | Optional event retention in days for subscribers on this plan. |
+| payments.subscriptionPlans[].limits.rateLimits | Optional event rate limits for subscribers on this plan, in the same format as `limits.event.rateLimits`. They replace the global limits for a subscriber while their subscription gives access, grace period included; an empty list means no limit. Everyone else, and subscribers whose subscription ended, keep the global limits. The plan is cached with admission, so a change of plan takes effect within five minutes. |
+| payments.subscriptionPlans[].limits.retentionMaxDays | Optional event retention in days for subscribers on this plan. It replaces `limits.event.retention.maxDays` for a subscriber while their subscription gives access, so it only applies while that is set. Once the subscription ends, the global window purges their events at the next pass. The kind and pubkey whitelists apply as before. |
 | workers.count                               | Number of workers to spin up to handle incoming connections. |
 |                                             | Spin workers as many CPUs are available when set to zero. Defaults to zero. |
 | limits.message.ipWhitelist                  | List of IPs (IPv4 or IPv6) to ignore rate limits. |

src/@types/repositories.ts

@@ -20,6 +20,8 @@ import { UserSubscription } from './user-subscription'
 
 export interface EventRetentionOptions {
   maxDays?: number
+  /** Retention in days for subscribers on these plans, replacing `maxDays` for them while their access lasts. */
+  planRetention?: { planId: string; maxDays: number }[]
   kindWhitelist?: (EventKinds | EventKindsRange)[]
   pubkeyWhitelist?: Pubkey[]
 }

src/constants/caching.ts

@@ -5,3 +5,9 @@ export enum CacheAdmissionState {
 }
 
 export const admissionCacheKey = (pubkey: string): string => `${pubkey}:is-admitted`
+
+/** Holds `plan:<id>` for a pubkey with an active subscription, or `none`. */
+export const subscriptionPlanCacheKey = (pubkey: string): string => `${pubkey}:subscription-plan`
+
+export const NO_SUBSCRIPTION_PLAN = 'none'
+export const SUBSCRIPTION_PLAN_PREFIX = 'plan:'

src/handlers/event-message-handler.ts

@@ -15,6 +15,7 @@ import {
   verifyNip05Identifier,
 } from '../utils/nip05'
 import { Event, ExpiringEvent } from '../@types/event'
+import { UserSubscription } from '../@types/user-subscription'
 import { EventRateLimit, FeeSchedule, Settings } from '../@types/settings'
 import {
   getEventExpiration,
@@ -33,7 +34,13 @@ import {
   isSealEvent,
   isWelcomeRumorEvent,
 } from '../utils/event'
-import { areSubscriptionsEnabled, getAccessEndsAt, hasSubscriptionAccess } from '../utils/subscription'
+import {
+  areSubscriptionsEnabled,
+  findSubscriptionPlan,
+  getAccessEndsAt,
+  getSubscriptionPlans,
+  hasSubscriptionAccess,
+} from '../utils/subscription'
 import { isAuthRequired } from '../utils/nip42'
 import {
   IEventRepository,
@@ -42,7 +49,13 @@ import {
   IUserSubscriptionRepository,
 } from '../@types/repositories'
 import { IEventStrategy, IMessageHandler } from '../@types/message-handlers'
-import { admissionCacheKey, CacheAdmissionState } from '../constants/caching'
+import {
+  admissionCacheKey,
+  CacheAdmissionState,
+  NO_SUBSCRIPTION_PLAN,
+  SUBSCRIPTION_PLAN_PREFIX,
+  subscriptionPlanCacheKey,
+} from '../constants/caching'
 import { createEventCommandResult } from '../telemetry/event-metrics'
 import { createLogger } from '../factories/logger-factory'
 import { Factory } from '../@types/base'
@@ -70,6 +83,8 @@ export class EventMessageHandler implements IMessageHandler {
     private readonly userSubscriptionRepository?: IUserSubscriptionRepository,
   ) {}
 
+  private readonly subscriptionReads = new WeakMap<Event, Promise<UserSubscription | undefined>>()
+
   public async handleMessage(message: IncomingEventMessage): Promise<void> {
     let [, event] = message
 
@@ -392,7 +407,11 @@ export class EventMessageHandler implements IMessageHandler {
       return false
     }
 
-    const { whitelists, rateLimits } = this.settings().limits?.event ?? {}
+    const currentSettings = this.settings()
+    const { whitelists } = currentSettings.limits?.event ?? {}
+    // A subscriber's plan can replace the global limits, which carry on for everyone else.
+    const rateLimits =
+      (await this.getPlanRateLimits(event, currentSettings)) ?? currentSettings.limits?.event?.rateLimits
     if (!rateLimits || !rateLimits.length) {
       return false
     }
@@ -452,6 +471,85 @@ export class EventMessageHandler implements IMessageHandler {
     return false
   }
 
+  /** The rate limits of the plan the sender is subscribed to, if it has its own and the subscription still gives access. */
+  private async getPlanRateLimits(event: Event, settings: Settings): Promise<EventRateLimit[] | undefined> {
+    if (!areSubscriptionsEnabled(settings) || !this.userSubscriptionRepository) {
+      return undefined
+    }
+
+    if (!getSubscriptionPlans(settings).some((plan) => plan.limits?.rateLimits !== undefined)) {
+      return undefined
+    }
+
+    return findSubscriptionPlan(settings, await this.getActivePlanId(event))?.limits?.rateLimits
+  }
+
+  /** Cached like admission, so it is read from the database once per pubkey and not once per event. */
+  private async getActivePlanId(event: Event): Promise<string | undefined> {
+    const pubkey = event.pubkey
+    const cacheKey = subscriptionPlanCacheKey(pubkey)
+
+    try {
+      const cached = await this.cache.getKey(cacheKey)
+      if (cached === NO_SUBSCRIPTION_PLAN) {
+        return undefined
+      }
+      if (typeof cached === 'string' && cached.startsWith(SUBSCRIPTION_PLAN_PREFIX)) {
+        return cached.slice(SUBSCRIPTION_PLAN_PREFIX.length)
+      }
+    } catch (error) {
+      logger('cache error for %s subscription plan: %o', pubkey, error)
+    }
+
+    let subscription
+    try {
+      subscription = await this.readSubscription(event)
+    } catch (error) {
+      // Without the subscription the global limits apply, and nothing is cached so the next event asks again.
+      logger('unable to read the subscription of %s: %o', pubkey, error)
+      return undefined
+    }
+
+    const now = new Date()
+    const planId = subscription && hasSubscriptionAccess(subscription, now) ? subscription.planId : undefined
+    this.cacheSubscriptionPlan(pubkey, planId, subscription ? getAccessEndsAt(subscription) : undefined, now)
+
+    return planId
+  }
+
+  private cacheSubscriptionPlan(pubkey: string, planId: string | undefined, accessEndsAt: Date | undefined, now: Date) {
+    // Only a plan that is being cached ends with access; "no plan" is rechecked after a minute whatever the end was.
+    const ttl = planId ? this.cacheTtlUntil(300, accessEndsAt, now) : 60
+    this.cacheSet(
+      subscriptionPlanCacheKey(pubkey),
+      planId ? `${SUBSCRIPTION_PLAN_PREFIX}${planId}` : NO_SUBSCRIPTION_PLAN,
+      ttl,
+    )
+  }
+
+  /**
+   * The subscription of the sender, read once per message: the rate limits and the admission check both need it,
+   * and on a cold cache they would otherwise each ask the database for the same row.
+   */
+  private readSubscription(event: Event): Promise<UserSubscription | undefined> {
+    let read = this.subscriptionReads.get(event)
+    if (!read) {
+      read = (this.userSubscriptionRepository as IUserSubscriptionRepository).findByPubkey(event.pubkey)
+      this.subscriptionReads.set(event, read)
+    }
+
+    return read
+  }
+
+  /** No longer than access lasts, so a plan's limits end when its period does rather than minutes later. */
+  private cacheTtlUntil(ttl: number, accessEndsAt: Date | undefined, now: Date): number {
+    if (!accessEndsAt) {
+      return ttl
+    }
+
+    return Math.max(1, Math.min(ttl, Math.ceil((accessEndsAt.getTime() - now.getTime()) / 1000)))
+  }
+
   protected async isUserAdmitted(event: Event): Promise<string | undefined> {
     const currentSettings = this.settings()
     const paymentsEnabled = currentSettings.payments?.enabled === true
@@ -534,7 +632,7 @@ export class EventMessageHandler implements IMessageHandler {
 
     let admittedTtl = 300
     if (subscriptionsEnabled && this.userSubscriptionRepository) {
-      const subscription = await this.userSubscriptionRepository.findByPubkey(event.pubkey)
+      const subscription = await this.readSubscription(event)
       const now = new Date()
       if (!hasSubscriptionAccess(subscription, now)) {
         this.cacheSet(cacheKey, CacheAdmissionState.BLOCKED_NOT_ADMITTED, 60)
@@ -543,9 +641,9 @@ export class EventMessageHandler implements IMessageHandler {
 
       // Cache no longer than access lasts, so the block lands when the period ends rather than minutes later.
       const accessEndsAt = subscription ? getAccessEndsAt(subscription) : undefined
-      if (accessEndsAt) {
-        admittedTtl = Math.max(1, Math.min(admittedTtl, Math.ceil((accessEndsAt.getTime() - now.getTime()) / 1000)))
-      }
+      admittedTtl = this.cacheTtlUntil(admittedTtl, accessEndsAt, now)
+      // The plan was just read, so what the rate limits need is cached with it.
+      this.cacheSubscriptionPlan(event.pubkey, subscription?.planId, accessEndsAt, now)
     }
 
     this.cacheSet(cacheKey, CacheAdmissionState.ADMITTED, admittedTtl)

src/repositories/event-repository.ts

@@ -539,6 +539,11 @@ export class EventRepository implements IEventRepository {
     }
 
     const retentionLimit = now - maxDays * SECONDS_PER_DAY
+    // Anything the retention rule purges is older than the latest of the limits, which is what the counts use.
+    const countLimit = Math.max(
+      retentionLimit,
+      ...this.getPlanRetention(options).map((plan) => now - plan.maxDays * SECONDS_PER_DAY),
+    )
 
     logger(
       'deleting expired and retained events (retentionLimit: %d, now: %d, batchSize: %d)',
@@ -553,7 +558,7 @@ export class EventRepository implements IEventRepository {
       .whereIn('event_id', candidates)
       .del(['deleted_at', 'expires_at', 'event_created_at'])
 
-    const getPromise = () => query.then((rows: any) => this.mapToPurgeCounts(rows, now, retentionLimit))
+    const getPromise = () => query.then((rows: any) => this.mapToPurgeCounts(rows, now, countLimit))
 
     return {
       then: <T1, T2>(
@@ -571,10 +576,41 @@ export class EventRepository implements IEventRepository {
     retentionLimit: number,
     options?: EventRetentionOptions,
   ): any {
+    const planRetention = this.getPlanRetention(options)
+    const subscribersOf = (planIds: string[]) =>
+      this.masterDbClient('user_subscriptions')
+        .select('pubkey')
+        .whereIn('plan_id', planIds)
+        .whereNot('status', 'expired')
+        .where((access) =>
+          access
+            .whereRaw('COALESCE(grace_ends_at, current_period_end) IS NULL')
+            .orWhereRaw('COALESCE(grace_ends_at, current_period_end) > ?', [new Date(now * 1000)]),
+        )
+
     return this.masterDbClient('events')
       .select('event_id')
-      .where(function () {
-        this.where('expires_at', '<', now).orWhereNotNull('deleted_at').orWhere('event_created_at', '<', retentionLimit)
+      .where((builder) => {
+        builder.where('expires_at', '<', now).orWhereNotNull('deleted_at')
+
+        if (!planRetention.length) {
+          builder.orWhere('event_created_at', '<', retentionLimit)
+          return
+        }
+
+        // Subscribers on a plan with its own retention are held to it instead of the global one.
+        builder.orWhere((global) =>
+          global
+            .where('event_created_at', '<', retentionLimit)
+            .whereNotIn('event_pubkey', subscribersOf(planRetention.map(({ planId }) => planId))),
+        )
+        for (const { planId, maxDays } of planRetention) {
+          builder.orWhere((plan) =>
+            plan
+              .where('event_created_at', '<', now - maxDays * SECONDS_PER_DAY)
+              .whereIn('event_pubkey', subscribersOf([planId])),
+          )
+        }
       })
       .modify((query) => {
         this.applyRetentionKindWhitelist(query, options?.kindWhitelist)
@@ -586,6 +622,12 @@ export class EventRepository implements IEventRepository {
       .limit(RETENTION_BATCH_SIZE)
   }
 
+  private getPlanRetention(options?: EventRetentionOptions): { planId: string; maxDays: number }[] {
+    return (options?.planRetention ?? []).filter(
+      ({ maxDays }) => typeof maxDays === 'number' && !isNaN(maxDays) && maxDays > 0,
+    )
+  }
+
   private applyRetentionKindWhitelist(query: any, kindWhitelist?: EventRetentionOptions['kindWhitelist']): void {
     const seen = new Set<string>()
     const configuredWhitelist = Array.isArray(kindWhitelist) ? kindWhitelist : []

src/services/maintenance-service.ts

@@ -1,6 +1,7 @@
 import { createLogger } from '../factories/logger-factory'
 import { IEventRepository } from '../@types/repositories'
 import { IMaintenanceService } from '../@types/services'
+import { areSubscriptionsEnabled, getSubscriptionPlans } from '../utils/subscription'
 import { Settings } from '../@types/settings'
 
 const logger = createLogger('maintenance-service')
@@ -22,8 +23,10 @@ export class MaintenanceService implements IMaintenanceService {
 
     try {
       logger('purging deleted, expired and old events')
+      const planRetention = this.getPlanRetention(currentSettings)
       const deletedCounts = await this.eventRepository.deleteExpiredAndRetained({
         maxDays,
+        ...(planRetention.length ? { planRetention } : {}),
         kindWhitelist: retention?.kind?.whitelist,
         pubkeyWhitelist: retention?.pubkey?.whitelist,
       })
@@ -37,4 +40,16 @@ export class MaintenanceService implements IMaintenanceService {
       logger.error('Unable to purge events. Reason:', error)
     }
   }
+
+  private getPlanRetention(settings: Settings): { planId: string; maxDays: number }[] {
+    if (!areSubscriptionsEnabled(settings)) {
+      return []
+    }
+
+    return getSubscriptionPlans(settings).flatMap((plan) =>
+      typeof plan.limits?.retentionMaxDays === 'number'
+        ? [{ planId: plan.id, maxDays: plan.limits.retentionMaxDays }]
+        : [],
+    )
+  }
 }