diff --git a/.changeset/subscription-tier-limits.md b/.changeset/subscription-tier-limits.md new file mode 100644 index 000000000..97badc62f --- /dev/null +++ b/.changeset/subscription-tier-limits.md @@ -0,0 +1,5 @@ +--- +"nostream": minor +--- + +feat(payments): add per-plan event rate limits and retention diff --git a/CONFIGURATION.md b/CONFIGURATION.md index e5888025d..80df043b4 100644 --- a/CONFIGURATION.md +++ b/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. | diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index 987044ca0..e64c2873e 100644 --- a/src/@types/repositories.ts +++ b/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[] } diff --git a/src/constants/caching.ts b/src/constants/caching.ts index 751614472..5bfd49105 100644 --- a/src/constants/caching.ts +++ b/src/constants/caching.ts @@ -5,3 +5,9 @@ export enum CacheAdmissionState { } export const admissionCacheKey = (pubkey: string): string => `${pubkey}:is-admitted` + +/** Holds `plan:` 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:' diff --git a/src/handlers/event-message-handler.ts b/src/handlers/event-message-handler.ts index ba362f990..e51e02c51 100644 --- a/src/handlers/event-message-handler.ts +++ b/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>() + public async handleMessage(message: IncomingEventMessage): Promise { 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 { + 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 { + 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 { + 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 { 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) diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index f95de9445..fe201d89f 100644 --- a/src/repositories/event-repository.ts +++ b/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: ( @@ -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() const configuredWhitelist = Array.isArray(kindWhitelist) ? kindWhitelist : [] diff --git a/src/services/maintenance-service.ts b/src/services/maintenance-service.ts index f0fa03d65..5f5949471 100644 --- a/src/services/maintenance-service.ts +++ b/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 }] + : [], + ) + } } diff --git a/test/unit/handlers/event-message-handler.spec.ts b/test/unit/handlers/event-message-handler.spec.ts index 22e072bc8..aa222e46e 100644 --- a/test/unit/handlers/event-message-handler.spec.ts +++ b/test/unit/handlers/event-message-handler.spec.ts @@ -1346,6 +1346,307 @@ describe('EventMessageHandler', () => { expect(actualResult).to.be.true expect(rateLimiterHitStub).to.have.been.calledOnce }) + + describe('subscription plans', () => { + const DAY = 86_400_000 + const PUBKEY = 'f'.repeat(64) + let subscriptionFindByPubkeyStub: SinonStub + let cacheStub: { getKey: SinonStub; setKey: SinonStub } + let subscriber: Event + + const subscription = (overrides: Partial = {}): UserSubscription => ({ + pubkey: PUBKEY, + planId: 'premium', + status: SubscriptionStatus.ACTIVE, + currentPeriodStart: new Date(Date.now() - 10 * DAY), + currentPeriodEnd: new Date(Date.now() + 20 * DAY), + graceEndsAt: new Date(Date.now() + 23 * DAY), + cancelAtPeriodEnd: false, + createdAt: new Date(), + updatedAt: new Date(), + ...overrides, + }) + + beforeEach(() => { + subscriber = { ...event, pubkey: PUBKEY, kind: 1 } + eventLimits.rateLimits = [{ period: 60000, rate: 1 }] + settings.payments = { + enabled: true, + feeSchedules: { admission: [] }, + subscriptionPlans: [ + { + id: 'premium', + amount: 5000000n, + intervalSeconds: 2_592_000, + graceSeconds: 259_200, + limits: { rateLimits: [{ period: 60000, rate: 100 }] }, + }, + { id: 'basic', amount: 1000000n, intervalSeconds: 2_592_000, graceSeconds: 259_200 }, + ], + } as any + rateLimiterHitStub.resolves(false) + subscriptionFindByPubkeyStub = sandbox.stub().resolves(subscription()) + cacheStub = { getKey: sandbox.stub().resolves(null), setKey: sandbox.stub().resolves(true) } + handler = new EventMessageHandler( + webSocket, + () => null, + {} as any, + userRepository, + () => settings, + {} as any, + cacheStub as any, + () => ({ hit: rateLimiterHitStub }), + {} as any, + { findByPubkey: subscriptionFindByPubkeyStub } as any, + ) + }) + + it('holds a subscriber to the limits of their plan instead of the global ones, keyed the same way', async () => { + await expect((handler as any).isRateLimited(subscriber)).to.eventually.equal(false) + + expect(rateLimiterHitStub).to.have.been.calledOnceWithExactly(`${PUBKEY}:events:60000`, 1, { + period: 60000, + rate: 100, + }) + }) + + it('holds everyone else to the global limits', async () => { + subscriptionFindByPubkeyStub.resolves(undefined) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub).to.have.been.calledOnceWithExactly(`${PUBKEY}:events:60000`, 1, { + period: 60000, + rate: 1, + }) + }) + + it('limits a subscriber who tops the plan limit', async () => { + rateLimiterHitStub.resolves(true) + + await expect((handler as any).isRateLimited(subscriber)).to.eventually.equal(60000) + }) + + it('keeps the global limits for a plan that has none of its own', async () => { + subscriptionFindByPubkeyStub.resolves(subscription({ planId: 'basic' })) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('lets a plan with an empty list of limits send without limit', async () => { + settings.payments.subscriptionPlans[0].limits.rateLimits = [] + + await expect((handler as any).isRateLimited(subscriber)).to.eventually.equal(false) + + expect(rateLimiterHitStub).not.to.have.been.called + }) + + it('replaces the whole global list, including limits for kinds the plan does not mention', async () => { + eventLimits.rateLimits = [ + { period: 60000, rate: 1, kinds: [1] }, + { period: 3600000, rate: 5 }, + ] + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub).to.have.been.calledOnce + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 100 }) + }) + + it('applies the plan limits for the kinds the plan limits name, as the global ones do', async () => { + settings.payments.subscriptionPlans[0].limits.rateLimits = [{ period: 60000, rate: 100, kinds: [7] }] + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub).not.to.have.been.called + }) + + it('falls back to the global limits once the subscription expired', async () => { + subscriptionFindByPubkeyStub.resolves(subscription({ status: SubscriptionStatus.EXPIRED })) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('falls back to the global limits when the period and grace have both passed, before the sweep ran', async () => { + subscriptionFindByPubkeyStub.resolves( + subscription({ + currentPeriodEnd: new Date(Date.now() - 5 * DAY), + graceEndsAt: new Date(Date.now() - DAY), + }), + ) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('keeps the plan limits through the grace period', async () => { + subscriptionFindByPubkeyStub.resolves( + subscription({ + status: SubscriptionStatus.PAST_DUE, + currentPeriodEnd: new Date(Date.now() - DAY), + graceEndsAt: new Date(Date.now() + 2 * DAY), + }), + ) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 100 }) + }) + + it('falls back to the global limits when the plan was removed from the settings', async () => { + subscriptionFindByPubkeyStub.resolves(subscription({ planId: 'retired' })) + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('still lets a whitelisted pubkey send without limit', async () => { + ;(eventLimits as any).whitelists = { pubkeys: [PUBKEY] } + + await expect((handler as any).isRateLimited(subscriber)).to.eventually.equal(false) + + expect(rateLimiterHitStub).not.to.have.been.called + }) + + describe('when subscriptions are not in use', () => { + it('does not look at subscriptions while payments are off', async () => { + settings.payments.enabled = false + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).not.to.have.been.called + expect(cacheStub.getKey).not.to.have.been.called + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('does not look at subscriptions when no plan is configured', async () => { + settings.payments.subscriptionPlans = [] + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).not.to.have.been.called + expect(cacheStub.getKey).not.to.have.been.called + }) + + it('does not look at subscriptions when no plan has limits of its own', async () => { + delete settings.payments.subscriptionPlans[0].limits + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).not.to.have.been.called + expect(cacheStub.getKey).not.to.have.been.called + }) + }) + + describe('caching', () => { + it('reads the plan from the database once and caches it, no longer than the usual 300s', async () => { + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).to.have.been.calledOnceWithExactly(PUBKEY) + expect(cacheStub.setKey).to.have.been.calledOnceWithExactly(`${PUBKEY}:subscription-plan`, 'plan:premium', 300) + }) + + it('uses the cached plan without reading the database', async () => { + cacheStub.getKey.withArgs(`${PUBKEY}:subscription-plan`).resolves('plan:premium') + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).not.to.have.been.called + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 100 }) + }) + + it('uses a cached "no plan" without reading the database', async () => { + cacheStub.getKey.withArgs(`${PUBKEY}:subscription-plan`).resolves('none') + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).not.to.have.been.called + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + }) + + it('keeps a plan whose id is "none" apart from "no plan"', async () => { + settings.payments.subscriptionPlans[0].id = 'none' + cacheStub.getKey.withArgs(`${PUBKEY}:subscription-plan`).resolves('plan:none') + + await (handler as any).isRateLimited(subscriber) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 100 }) + }) + + it('caches "no plan" for 60s', async () => { + subscriptionFindByPubkeyStub.resolves(undefined) + + await (handler as any).isRateLimited(subscriber) + + expect(cacheStub.setKey).to.have.been.calledOnceWithExactly(`${PUBKEY}:subscription-plan`, 'none', 60) + }) + + it('caches "no plan" for a full 60s after the subscription expired, not just until its past end', async () => { + subscriptionFindByPubkeyStub.resolves( + subscription({ status: SubscriptionStatus.EXPIRED, currentPeriodEnd: new Date(Date.now() - 5 * DAY) }), + ) + + await (handler as any).isRateLimited(subscriber) + + expect(cacheStub.setKey).to.have.been.calledOnceWithExactly(`${PUBKEY}:subscription-plan`, 'none', 60) + }) + + it('reads the subscription once for a message that is rate limited and then checked for admission', async () => { + sandbox.stub(EventMessageHandler.prototype, 'getRelayPublicKey' as any).returns('a'.repeat(64)) + userRepository.findByPubkey = sandbox.stub().resolves({ isAdmitted: true, balance: 0n }) + + await (handler as any).isRateLimited(subscriber) + await expect((handler as any).isUserAdmitted(subscriber)).to.eventually.be.undefined + + expect(subscriptionFindByPubkeyStub).to.have.been.calledOnce + }) + + it('reads the subscription again for the next message of the same sender', async () => { + sandbox.stub(EventMessageHandler.prototype, 'getRelayPublicKey' as any).returns('a'.repeat(64)) + userRepository.findByPubkey = sandbox.stub().resolves({ isAdmitted: true, balance: 0n }) + + await (handler as any).isRateLimited(subscriber) + await (handler as any).isRateLimited({ ...subscriber, id: 'e'.repeat(64) }) + + expect(subscriptionFindByPubkeyStub).to.have.been.calledTwice + }) + + it('caches the plan no longer than access lasts', async () => { + subscriptionFindByPubkeyStub.resolves( + subscription({ currentPeriodEnd: new Date(Date.now() + 20_000), graceEndsAt: new Date(Date.now() + 40_000) }), + ) + + await (handler as any).isRateLimited(subscriber) + + const [, , ttl] = cacheStub.setKey.firstCall.args + expect(ttl).to.be.within(39, 41) + }) + + it('reads the database when the cache cannot be read', async () => { + cacheStub.getKey.rejects(new Error('cache unavailable')) + + await (handler as any).isRateLimited(subscriber) + + expect(subscriptionFindByPubkeyStub).to.have.been.calledOnce + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 100 }) + }) + + it('falls back to the global limits when the subscription cannot be read, and caches nothing', async () => { + subscriptionFindByPubkeyStub.rejects(new Error('db unavailable')) + + await expect((handler as any).isRateLimited(subscriber)).to.eventually.equal(false) + + expect(rateLimiterHitStub.firstCall.args[2]).to.deep.equal({ period: 60000, rate: 1 }) + expect(cacheStub.setKey).not.to.have.been.called + }) + }) + }) }) describe('isUserAdmitted', () => { @@ -1756,6 +2057,22 @@ describe('EventMessageHandler', () => { expect(ttl).to.be.within(39, 41) }) + it('caches the plan with the admission, so the rate limits need no second read', async () => { + subscriptionFindByPubkeyStub.resolves(subscription()) + + await (handler as any).isUserAdmitted(event) + + expect(cacheStub.setKey).to.have.been.calledWith(`${event.pubkey}:subscription-plan`, 'plan:basic', 300) + }) + + it('caches that a user admitted without a subscription has no plan', async () => { + subscriptionFindByPubkeyStub.resolves(undefined) + + await (handler as any).isUserAdmitted(event) + + expect(cacheStub.setKey).to.have.been.calledWith(`${event.pubkey}:subscription-plan`, 'none', 60) + }) + it('caches admission for the usual 300s when access lasts longer', async () => { subscriptionFindByPubkeyStub.resolves(subscription()) diff --git a/test/unit/repositories/event-repository.spec.ts b/test/unit/repositories/event-repository.spec.ts index 0e89d0934..8cba85b66 100644 --- a/test/unit/repositories/event-repository.spec.ts +++ b/test/unit/repositories/event-repository.spec.ts @@ -1039,6 +1039,115 @@ describe('EventRepository', () => { ) }) + describe('with plan retention', () => { + beforeEach(() => { + sandbox.useFakeTimers(new Date('2023-11-14T22:13:20.000Z')) + }) + + // knex prints dates in the local time zone of the machine, so the moment is masked. + const purge = (options: Parameters[0]) => + repository + .deleteExpiredAndRetained(options) + .toString() + .replace(/'\d{4}-\d{2}-\d{2}[ T][^']*'/g, '') + + it('purges as before when no plan has its own retention', () => { + expect(purge({ maxDays: 7, planRetention: [] })).to.equal(purge({ maxDays: 7 })) + }) + + it('holds the subscribers of a plan to its retention and everyone else to the global one', () => { + const query = purge({ maxDays: 7, planRetention: [{ planId: 'premium', maxDays: 90 }] }) + + // 1700000000 is now; 7 days back is 1699395200, 90 days back is 1692224000 + const active = + 'not "status" = \'expired\' and (COALESCE(grace_ends_at, current_period_end) IS NULL or COALESCE(grace_ends_at, current_period_end) > )' + expect(query).to.equal( + 'delete from "events" where "event_id" in (select "event_id" from "events" where ("expires_at" < 1700000000 or "deleted_at" is not null' + + ` or ("event_created_at" < 1699395200 and "event_pubkey" not in (select "pubkey" from "user_subscriptions" where "plan_id" in ('premium') and ${active}))` + + ` or ("event_created_at" < 1692224000 and "event_pubkey" in (select "pubkey" from "user_subscriptions" where "plan_id" in ('premium') and ${active})))` + + ' and not ("event_kind" = 62) limit 1000) returning "deleted_at", "expires_at", "event_created_at"', + ) + }) + + it('keeps the whitelists in force for subscribers too', () => { + const query = purge({ + maxDays: 7, + kindWhitelist: [1], + pubkeyWhitelist: ['001122'], + planRetention: [{ planId: 'premium', maxDays: 90 }], + }) + + expect(query).to.contain('and not ("event_kind" = 1 or "event_kind" = 62) and "event_pubkey" not in (X\'001122\') limit 1000') + }) + + it('gives each plan its own window, excluding the subscribers of all of them from the global one', () => { + const query = purge({ + maxDays: 7, + planRetention: [ + { planId: 'basic', maxDays: 30 }, + { planId: 'vip', maxDays: 365 }, + ], + }) + + expect(query).to.contain(`"event_created_at" < 1699395200 and "event_pubkey" not in (select "pubkey" from "user_subscriptions" where "plan_id" in ('basic', 'vip')`) + expect(query).to.contain(`"event_created_at" < ${1700000000 - 30 * 86400} and "event_pubkey" in (select "pubkey" from "user_subscriptions" where "plan_id" in ('basic')`) + expect(query).to.contain(`"event_created_at" < ${1700000000 - 365 * 86400} and "event_pubkey" in (select "pubkey" from "user_subscriptions" where "plan_id" in ('vip')`) + }) + + it('applies a plan retention that is shorter than the global one', () => { + expect(purge({ maxDays: 90, planRetention: [{ planId: 'free', maxDays: 7 }] })).to.contain( + `"event_created_at" < 1699395200 and "event_pubkey" in (select "pubkey" from "user_subscriptions" where "plan_id" in ('free')`, + ) + }) + + it('only counts subscriptions whose access has not ended, so an expired subscriber falls back to the global window', () => { + const query = purge({ maxDays: 7, planRetention: [{ planId: 'premium', maxDays: 90 }] }) + + expect(query).to.contain('not "status" = \'expired\'') + expect(query).to.contain('COALESCE(grace_ends_at, current_period_end) IS NULL') + expect(query).to.contain('COALESCE(grace_ends_at, current_period_end) > ') + }) + + it('ignores plan retentions that are not positive numbers', () => { + const options = { maxDays: 7 } + + expect( + purge({ + ...options, + planRetention: [ + { planId: 'zero', maxDays: 0 }, + { planId: 'negative', maxDays: -5 }, + { planId: 'nan', maxDays: Number.NaN }, + ], + }), + ).to.equal(purge(options)) + }) + + it('does not purge anything when the global retention is off, whatever the plans say', async () => { + const result = await repository.deleteExpiredAndRetained({ + maxDays: 0, + planRetention: [{ planId: 'premium', maxDays: 90 }], + }) + + expect(result).to.deep.equal({ deleted: 0, expired: 0, retained: 0 }) + }) + + it('counts a row purged by a longer or shorter plan window as retained', () => { + const now = 1700000000 + const result = (repository as any).mapToPurgeCounts( + [ + // purged by a 7 day plan window although it is newer than the 90 day global one + { deleted_at: null, expires_at: null, event_created_at: now - 20 * 86400 }, + { deleted_at: null, expires_at: null, event_created_at: now - 120 * 86400 }, + ], + now, + now - 7 * 86400, + ) + + expect(result.retained).to.equal(2) + }) + }) + it('maps purged rows to deleted, expired, and retained counts', () => { const result = (repository as any).mapToPurgeCounts( [ diff --git a/test/unit/services/maintenance-service.spec.ts b/test/unit/services/maintenance-service.spec.ts index 33b9319ea..fdf3f0621 100644 --- a/test/unit/services/maintenance-service.spec.ts +++ b/test/unit/services/maintenance-service.spec.ts @@ -111,5 +111,77 @@ describe('MaintenanceService', () => { expect(eventRepository.deleteExpiredAndRetained).to.have.been.calledOnce }) + describe('plan retention', () => { + const plan = (id: string, limits?: unknown) => ({ + id, + amount: 1000000n, + intervalSeconds: 2592000, + graceSeconds: 259200, + ...(limits ? { limits } : {}), + }) + + const withPlans = (plans: unknown[], payments: Record = {}) => { + settings.returns({ + payments: { enabled: true, subscriptionPlans: plans, ...payments }, + limits: { event: { retention: { maxDays: 30 } } }, + } as any) + eventRepository.deleteExpiredAndRetained.resolves({ deleted: 0, expired: 0, retained: 0 }) + } + + it('hands the retention of each plan that has one to the purge', async () => { + withPlans([ + plan('basic', { retentionMaxDays: 90 }), + plan('free'), + plan('vip', { retentionMaxDays: 365, rateLimits: [] }), + ]) + + await service.clearOldEvents() + + expect(eventRepository.deleteExpiredAndRetained).to.have.been.calledOnceWithExactly({ + maxDays: 30, + planRetention: [ + { planId: 'basic', maxDays: 90 }, + { planId: 'vip', maxDays: 365 }, + ], + kindWhitelist: undefined, + pubkeyWhitelist: undefined, + }) + }) + + it('purges as before when no plan has a retention', async () => { + withPlans([plan('basic'), plan('vip', { rateLimits: [] })]) + + await service.clearOldEvents() + + expect(eventRepository.deleteExpiredAndRetained.firstCall.args[0]).to.not.have.property('planRetention') + }) + + it('ignores plan retentions while payments are off', async () => { + withPlans([plan('basic', { retentionMaxDays: 90 })], { enabled: false }) + + await service.clearOldEvents() + + expect(eventRepository.deleteExpiredAndRetained.firstCall.args[0]).to.not.have.property('planRetention') + }) + + it('ignores plan retentions when no plan is configured', async () => { + withPlans([]) + + await service.clearOldEvents() + + expect(eventRepository.deleteExpiredAndRetained.firstCall.args[0]).to.not.have.property('planRetention') + }) + + it('does not purge when the global retention is off, however long the plans keep events', async () => { + settings.returns({ + payments: { enabled: true, subscriptionPlans: [plan('basic', { retentionMaxDays: 90 })] }, + limits: { event: { retention: { maxDays: 0 } } }, + } as any) + + await service.clearOldEvents() + + expect(eventRepository.deleteExpiredAndRetained).not.to.have.been.called + }) + }) }) })