feat(payments): add the subscription data model and plan settings

.changeset/subscription-data-model.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(payments): add the subscription data model and plan settings

.knip.json

@@ -17,7 +17,8 @@
             ],
             "ignore": [
                 ".nostr/**",
-                "src/utils/relay-probe/**"
+                "src/utils/relay-probe/**",
+                "src/repositories/user-subscription-repository.ts"
             ],
             "commitlint": false,
             "eslint": false,

CONFIGURATION.md

@@ -250,6 +250,13 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | payments.feeSchedules.admission[].whitelists.event_kinds | List of event kinds to waive admission fee. Use `[min, max]` for ranges. |
 | payments.feeSchedules.admission[].whitelists.pubkeys | List of pubkeys to waive admission fee. |
 | payments.processor                          | Either `zebedee`, `lnbits`, `lnurl`. |
+| payments.subscriptionPlans[].id             | Unique plan name, for example `basic`. `grandfathered` is reserved: users admitted before plans existed are moved onto it and keep access without a paid period. |
+| payments.subscriptionPlans[].amount         | Price of one period in msats. Must be unique across plans, because the amount paid selects the plan. |
+| 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. |
 | 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. |

migrations/20261016_120000_create_user_subscriptions_table.js

@@ -0,0 +1,45 @@
+exports.up = async function (knex) {
+  await knex.schema.createTable('user_subscriptions', (table) => {
+    // One subscription per pubkey, so the pubkey is the key.
+    table.binary('pubkey').primary().references('pubkey').inTable('users').onDelete('CASCADE')
+    table.text('plan_id').notNullable()
+    table.enum('status', ['active', 'renewal_pending', 'past_due', 'expired']).notNullable().defaultTo('active')
+    table.timestamp('current_period_start', { useTz: true }).notNullable()
+    // NULL means the subscription never ends (the grandfathered plan).
+    table.timestamp('current_period_end', { useTz: true }).nullable()
+    table.timestamp('grace_ends_at', { useTz: true }).nullable()
+    table.boolean('cancel_at_period_end').notNullable().defaultTo(false)
+    table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+
+    // The renewal and expiry sweeps filter by status and order by the period or grace end.
+    table.index(['status', 'current_period_end'], 'idx_user_subscriptions_status_period_end')
+    table.index(['status', 'grace_ends_at'], 'idx_user_subscriptions_status_grace_end')
+  })
+
+  await knex.schema.alterTable('invoices', (table) => {
+    table.text('plan_id').nullable()
+    table.timestamp('period_start', { useTz: true }).nullable()
+    table.timestamp('period_end', { useTz: true }).nullable()
+  })
+
+  // Everyone admitted so far keeps access without a paid period: they move onto a plan that never
+  // expires, so turning subscription plans on later cannot lock anyone out.
+  await knex.raw(`
+    INSERT INTO user_subscriptions (pubkey, plan_id, status, current_period_start, current_period_end)
+    SELECT pubkey, 'grandfathered', 'active', COALESCE(tos_accepted_at AT TIME ZONE 'UTC', created_at), NULL
+    FROM users
+    WHERE is_admitted = true
+    ON CONFLICT (pubkey) DO NOTHING
+  `)
+}
+
+exports.down = async function (knex) {
+  await knex.schema.alterTable('invoices', (table) => {
+    table.dropColumn('plan_id')
+    table.dropColumn('period_start')
+    table.dropColumn('period_end')
+  })
+
+  await knex.schema.dropTable('user_subscriptions')
+}

resources/default-settings.yaml

@@ -25,6 +25,14 @@ payments:
             - replace-with-your-pubkey-in-hex
           event_kinds:
             - 9735 # Nip-57 Lightning Zap Receipts
+  # Recurring membership plans. Leave empty to keep the one-time admission fee above.
+  # The amount paid selects the plan, so every plan needs its own amount.
+  # - id: basic
+  #   description: Basic membership, paid every 30 days
+  #   amount: 1000000 # msats per period
+  #   intervalSeconds: 2592000
+  #   graceSeconds: 259200 # access lasts this long after a period ends unrenewed
+  subscriptionPlans: []
   # Applied only when the payments processor does not report an expiry of its own.
   # Without it such invoices can never be retired and stay pending forever.
   # Raise it if your processor issues invoices that stay payable for longer than

src/@types/invoice.ts

@@ -26,6 +26,10 @@ export interface Invoice {
   updatedAt: Date
   createdAt: Date
   verifyURL?: string
+  /** Subscription plan this invoice pays for, when it is not an admission fee. */
+  planId?: string | null
+  periodStart?: Date | null
+  periodEnd?: Date | null
 }
 
 export interface LnurlInvoice extends Invoice {
@@ -46,4 +50,7 @@ export interface DBInvoice {
   updated_at: Date
   created_at: Date
   verify_url: string
+  plan_id?: string | null
+  period_start?: Date | null
+  period_end?: Date | null
 }

src/@types/repositories.ts

@@ -16,6 +16,7 @@ import { Report } from './report'
 import { EventKindsRange } from './settings'
 import { SubscriptionFilter } from './subscription'
 import { User } from './user'
+import { UserSubscription } from './user-subscription'
 
 export interface EventRetentionOptions {
   maxDays?: number
@@ -59,6 +60,17 @@ export interface IInvoiceRepository {
   findPendingInvoices(offset?: number, limit?: number, client?: DatabaseClient): Promise<Invoice[]>
 }
 
+export interface IUserSubscriptionRepository {
+  findByPubkey(pubkey: Pubkey, client?: DatabaseClient): Promise<UserSubscription | undefined>
+  upsert(subscription: Omit<UserSubscription, 'createdAt' | 'updatedAt'>, client?: DatabaseClient): Promise<number>
+  /** Active subscriptions whose period ends by `before` and that are not set to cancel. */
+  findDueForRenewal(before: Date, limit?: number, client?: DatabaseClient): Promise<UserSubscription[]>
+  /** Active or renewal-pending subscriptions whose period has ended by `now`. */
+  findPeriodEnded(now: Date, limit?: number, client?: DatabaseClient): Promise<UserSubscription[]>
+  /** Past-due subscriptions whose grace period has ended by `now`. */
+  findGraceEnded(now: Date, limit?: number, client?: DatabaseClient): Promise<UserSubscription[]>
+}
+
 export interface IUserRepository {
   findByPubkey(pubkey: Pubkey, client?: DatabaseClient): Promise<User | undefined>
   upsert(user: Partial<User>, client?: DatabaseClient): Promise<number>

src/@types/settings.ts

@@ -224,10 +224,31 @@ export interface FeeSchedules {
   publication: FeeSchedule[]
 }
 
+export interface SubscriptionPlanLimits {
+  /** Replaces limits.event.rateLimits for subscribers on this plan. */
+  rateLimits?: EventRateLimit[]
+  /** Replaces limits.event.retention.maxDays for subscribers on this plan. */
+  retentionMaxDays?: number
+}
+
+export interface SubscriptionPlan {
+  /** Unique, and never `grandfathered`, which is reserved. */
+  id: string
+  description?: string
+  /** Price of one period in msats. Unique across plans, since the amount paid selects the plan. */
+  amount: bigint
+  intervalSeconds: number
+  /** How long access lasts after a period ends without a renewal. */
+  graceSeconds: number
+  limits?: SubscriptionPlanLimits
+}
+
 export interface Payments {
   enabled: boolean
   processor: keyof PaymentsProcessors
   feeSchedules: FeeSchedules
+  /** Recurring membership plans. Empty keeps the one-time admission fee. */
+  subscriptionPlans?: SubscriptionPlan[]
   /** Fallback when the processor reports no expiry. A reported one always wins. */
   invoiceExpirySeconds?: number
 }

src/@types/user-subscription.ts

@@ -0,0 +1,36 @@
+import { Pubkey } from './base'
+
+export enum SubscriptionStatus {
+  ACTIVE = 'active',
+  RENEWAL_PENDING = 'renewal_pending',
+  PAST_DUE = 'past_due',
+  EXPIRED = 'expired',
+}
+
+/** Plan id of the subscriptions created for users admitted before subscription plans existed. */
+export const GRANDFATHERED_PLAN_ID = 'grandfathered'
+
+export interface UserSubscription {
+  pubkey: Pubkey
+  planId: string
+  status: SubscriptionStatus
+  currentPeriodStart: Date
+  /** Null when the subscription never ends. */
+  currentPeriodEnd: Date | null
+  graceEndsAt: Date | null
+  cancelAtPeriodEnd: boolean
+  createdAt: Date
+  updatedAt: Date
+}
+
+export interface DBUserSubscription {
+  pubkey: Buffer
+  plan_id: string
+  status: SubscriptionStatus
+  current_period_start: Date
+  current_period_end: Date | null
+  grace_ends_at: Date | null
+  cancel_at_period_end: boolean
+  created_at: Date
+  updated_at: Date
+}

src/repositories/invoice-repository.ts

@@ -92,6 +92,9 @@ export class InvoiceRepository implements IInvoiceRepository {
       updated_at: always(new Date()),
       created_at: prop('createdAt'),
       verify_url: prop('verifyURL'),
+      plan_id: prop('planId'),
+      period_start: prop('periodStart'),
+      period_end: prop('periodEnd'),
     })(invoice)
 
     logger('row: %o', row)
@@ -110,6 +113,9 @@ export class InvoiceRepository implements IInvoiceRepository {
           'expires_at',
           'created_at',
           'verify_url',
+          'plan_id',
+          'period_start',
+          'period_end',
         ])(row),
       )
 

src/repositories/user-subscription-repository.ts

@@ -0,0 +1,118 @@
+import { DatabaseClient, Pubkey } from '../@types/base'
+import { DBUserSubscription, SubscriptionStatus, UserSubscription } from '../@types/user-subscription'
+import { fromBuffer, toBuffer } from '../utils/transform'
+import { createLogger } from '../factories/logger-factory'
+import { IUserSubscriptionRepository } from '../@types/repositories'
+
+const logger = createLogger('user-subscription-repository')
+
+const DEFAULT_PAGE_SIZE = 10
+
+const fromDBUserSubscription = (row: DBUserSubscription): UserSubscription => ({
+  pubkey: fromBuffer(row.pubkey),
+  planId: row.plan_id,
+  status: row.status,
+  currentPeriodStart: row.current_period_start,
+  currentPeriodEnd: row.current_period_end,
+  graceEndsAt: row.grace_ends_at,
+  cancelAtPeriodEnd: row.cancel_at_period_end,
+  createdAt: row.created_at,
+  updatedAt: row.updated_at,
+})
+
+export class UserSubscriptionRepository implements IUserSubscriptionRepository {
+  public constructor(private readonly dbClient: DatabaseClient) {}
+
+  public async findByPubkey(
+    pubkey: Pubkey,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<UserSubscription | undefined> {
+    const [row] = await client<DBUserSubscription>('user_subscriptions').where('pubkey', toBuffer(pubkey)).select()
+
+    return row ? fromDBUserSubscription(row) : undefined
+  }
+
+  public async upsert(
+    subscription: Omit<UserSubscription, 'createdAt' | 'updatedAt'>,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<number> {
+    logger('upserting subscription for %s: %s', subscription.pubkey, subscription.planId)
+
+    const now = new Date()
+    const row: DBUserSubscription = {
+      pubkey: toBuffer(subscription.pubkey),
+      plan_id: subscription.planId,
+      status: subscription.status,
+      current_period_start: subscription.currentPeriodStart,
+      current_period_end: subscription.currentPeriodEnd,
+      grace_ends_at: subscription.graceEndsAt,
+      cancel_at_period_end: subscription.cancelAtPeriodEnd,
+      created_at: now,
+      updated_at: now,
+    }
+
+    const result = await client<DBUserSubscription>('user_subscriptions')
+      .insert(row)
+      .onConflict('pubkey')
+      .merge([
+        'plan_id',
+        'status',
+        'current_period_start',
+        'current_period_end',
+        'grace_ends_at',
+        'cancel_at_period_end',
+        'updated_at',
+      ])
+
+    return (result as unknown as { rowCount?: number }).rowCount ?? 0
+  }
+
+  public async findDueForRenewal(
+    before: Date,
+    limit = DEFAULT_PAGE_SIZE,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<UserSubscription[]> {
+    const rows = await client<DBUserSubscription>('user_subscriptions')
+      .where('status', SubscriptionStatus.ACTIVE)
+      .where('cancel_at_period_end', false)
+      .whereNotNull('current_period_end')
+      .where('current_period_end', '<=', before)
+      .orderBy('current_period_end', 'asc')
+      .limit(limit)
+      .select()
+
+    return rows.map(fromDBUserSubscription)
+  }
+
+  public async findPeriodEnded(
+    now: Date,
+    limit = DEFAULT_PAGE_SIZE,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<UserSubscription[]> {
+    const rows = await client<DBUserSubscription>('user_subscriptions')
+      .whereIn('status', [SubscriptionStatus.ACTIVE, SubscriptionStatus.RENEWAL_PENDING])
+      .whereNotNull('current_period_end')
+      .where('current_period_end', '<=', now)
+      .orderBy('current_period_end', 'asc')
+      .limit(limit)
+      .select()
+
+    return rows.map(fromDBUserSubscription)
+  }
+
+  public async findGraceEnded(
+    now: Date,
+    limit = DEFAULT_PAGE_SIZE,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<UserSubscription[]> {
+    const rows = await client<DBUserSubscription>('user_subscriptions')
+      .where('status', SubscriptionStatus.PAST_DUE)
+      .whereNotNull('grace_ends_at')
+      .where('grace_ends_at', '<=', now)
+      .orderBy('grace_ends_at', 'asc')
+      .limit(limit)
+      .select()
+
+    return rows.map(fromDBUserSubscription)
+  }
+}