diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml
index 12dc90b75..1d7ed2c34 100755
--- a/resources/default-settings.yaml
+++ b/resources/default-settings.yaml
@@ -462,6 +462,10 @@ admin:
admission.invoice.created: true
admission.invoice.paid: true
admission.invoice.failed: true
+ peering.invoice.created: true
+ peering.invoice.paid: true
+ peering.invoice.failed: true
+ peering.access.lapsed: true
settings.changed: true
relay.restarted: false
probe.failed: true
diff --git a/src/@types/invoice.ts b/src/@types/invoice.ts
index 6ca842785..f6cd05133 100644
--- a/src/@types/invoice.ts
+++ b/src/@types/invoice.ts
@@ -12,6 +12,12 @@ export enum InvoiceStatus {
EXPIRED = 'expired',
}
+export enum InvoicePurpose {
+ ADMISSION = 'admission',
+ SUBSCRIPTION = 'subscription',
+ PEERING = 'peering',
+}
+
export interface Invoice {
id: string
pubkey: Pubkey
@@ -26,10 +32,16 @@ export interface Invoice {
updatedAt: Date
createdAt: Date
verifyURL?: string
+ /** What this invoice pays for. Older rows are admission or subscription. */
+ purpose?: InvoicePurpose | null
/** Subscription plan this invoice pays for, when it is not an admission fee. */
planId?: string | null
periodStart?: Date | null
periodEnd?: Date | null
+ /** Peering term snapshot used to issue this invoice. */
+ termAmountMsats?: bigint | null
+ /** Peering period length in seconds for this invoice. */
+ termPeriodSeconds?: number | null
}
export interface LnurlInvoice extends Invoice {
@@ -50,7 +62,10 @@ export interface DBInvoice {
updated_at: Date
created_at: Date
verify_url: string
+ purpose?: InvoicePurpose | null
plan_id?: string | null
period_start?: Date | null
period_end?: Date | null
+ term_amount_msats?: string | null
+ term_period_seconds?: number | null
}
diff --git a/src/@types/operator-notifications.ts b/src/@types/operator-notifications.ts
index 880dfe4f2..31395b476 100644
--- a/src/@types/operator-notifications.ts
+++ b/src/@types/operator-notifications.ts
@@ -2,6 +2,10 @@ export enum OperatorNotificationEventType {
ADMISSION_INVOICE_CREATED = 'admission.invoice.created',
ADMISSION_INVOICE_PAID = 'admission.invoice.paid',
ADMISSION_INVOICE_FAILED = 'admission.invoice.failed',
+ PEERING_INVOICE_CREATED = 'peering.invoice.created',
+ PEERING_INVOICE_PAID = 'peering.invoice.paid',
+ PEERING_INVOICE_FAILED = 'peering.invoice.failed',
+ PEERING_ACCESS_LAPSED = 'peering.access.lapsed',
SETTINGS_CHANGED = 'settings.changed',
RELAY_RESTARTED = 'relay.restarted',
PROBE_FAILED = 'probe.failed',
@@ -26,6 +30,10 @@ export interface OperatorNotificationEventToggles {
'admission.invoice.created'?: boolean
'admission.invoice.paid'?: boolean
'admission.invoice.failed'?: boolean
+ 'peering.invoice.created'?: boolean
+ 'peering.invoice.paid'?: boolean
+ 'peering.invoice.failed'?: boolean
+ 'peering.access.lapsed'?: boolean
'settings.changed'?: boolean
'relay.restarted'?: boolean
'probe.failed'?: boolean
diff --git a/src/@types/peering.ts b/src/@types/peering.ts
index fecb40cf2..f257df599 100644
--- a/src/@types/peering.ts
+++ b/src/@types/peering.ts
@@ -1,3 +1,4 @@
+import { Pubkey } from './base'
import { PeeringScope, PeeringTerm } from './settings'
/** One scope advertised by a peer in its NIP-11 `peering` section. */
@@ -14,6 +15,68 @@ export interface PeeringAdvertisement {
export type PeeringSyncStatus = 'idle' | 'connecting' | 'reconciling' | 'live' | 'error' | 'stopped'
+export enum PeeringPaymentDirection {
+ SELL = 'sell',
+ BUY = 'buy',
+}
+
+export enum PeeringPaymentStatus {
+ UNPAID = 'unpaid',
+ INVOICED = 'invoiced',
+ ACTIVE = 'active',
+ RENEWAL_PENDING = 'renewal_pending',
+ LAPSED = 'lapsed',
+}
+
+/** Billing state for one peer relationship in one direction. */
+export interface PeeringPayment {
+ peerPubkey: Pubkey
+ direction: PeeringPaymentDirection
+ status: PeeringPaymentStatus
+ currentInvoiceId?: string | null
+ renewalInvoiceId?: string | null
+ periodStart?: Date | null
+ periodEnd?: Date | null
+ termAmountMsats?: bigint | null
+ termPeriodSeconds?: number | null
+ lastSettledInvoiceId?: string | null
+ lastSettledAt?: Date | null
+ lastCheckedAt?: Date | null
+ lastError?: string | null
+ createdAt: Date
+ updatedAt: Date
+}
+
+export interface DBPeeringPayment {
+ peer_pubkey: Buffer
+ direction: PeeringPaymentDirection
+ status: PeeringPaymentStatus
+ current_invoice_id: string | null
+ renewal_invoice_id: string | null
+ period_start: Date | null
+ period_end: Date | null
+ term_amount_msats: string | null
+ term_period_seconds: number | null
+ last_settled_invoice_id: string | null
+ last_settled_at: Date | null
+ last_checked_at: Date | null
+ last_error: string | null
+ created_at: Date
+ updated_at: Date
+}
+
+export interface PeeringAccess {
+ allowed: boolean
+ status: PeeringPaymentStatus | 'free' | 'unconfigured'
+ periodEnd?: Date | null
+ reason?: string
+}
+
+export interface PeeringTermSnapshot {
+ amountMsats: bigint
+ periodSeconds: number
+}
+
/**
* Per-peer sync state. Written by the peering-sync worker and read by the admin
* console; `lagSeconds` is how old the newest event we hold in a current scope
diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts
index 3da242738..1a03d5086 100644
--- a/src/@types/repositories.ts
+++ b/src/@types/repositories.ts
@@ -12,6 +12,7 @@ import {
NotificationDeliveryStatus,
OperatorNotificationChannelType,
} from './operator-notifications'
+import { PeeringPayment } from './peering'
import { Report } from './report'
import { EventKindsRange } from './settings'
import { SubscriptionFilter } from './subscription'
@@ -72,7 +73,28 @@ export interface IInvoiceRepository {
upsert(invoice: Partial
, client?: DatabaseClient): Promise
updateStatus(invoice: Pick, client?: DatabaseClient): Promise
confirmInvoice(invoiceId: string, amountReceived: bigint, confirmedAt: Date, client?: DatabaseClient): Promise
+ updatePeriod(invoiceId: string, periodStart: Date, periodEnd: Date, client?: DatabaseClient): Promise
findPendingInvoices(offset?: number, limit?: number, client?: DatabaseClient): Promise
+ /**
+ * Completed peering invoices whose settlement never committed, oldest first.
+ * `confirmed_at` stays null when the settlement transaction rolls back, so the
+ * maintenance worker can retry them instead of stranding a paid peer.
+ */
+ findUnconfirmedPeeringInvoices(limit?: number, offset?: number, client?: DatabaseClient): Promise
+}
+
+export interface IPeeringPaymentRepository {
+ findByPeer(pubkey: Pubkey, client?: DatabaseClient): Promise
+ /** Inserts an unpaid row when one does not exist, so the following lock always finds it. */
+ ensureRow(pubkey: Pubkey, client: DatabaseClient): Promise
+ /** SELECT ... FOR UPDATE; must run inside a transaction. */
+ lockByPeer(pubkey: Pubkey, client: DatabaseClient): Promise
+ save(payment: PeeringPayment, client?: DatabaseClient): Promise
+ list(): Promise
+ /** Active or renewal-pending relationships whose period ends by `before`. */
+ findDueForRenewal(before: Date, limit: number, offset: number): Promise
+ /** Active or renewal-pending relationships whose period has ended by `now`. */
+ findPeriodEnded(now: Date, limit: number, offset: number): Promise
}
export interface IUserSubscriptionRepository {
diff --git a/src/@types/services.ts b/src/@types/services.ts
index 21a04c177..de9366223 100644
--- a/src/@types/services.ts
+++ b/src/@types/services.ts
@@ -1,4 +1,4 @@
-import { Pubkey } from './base'
+import { DatabaseTransaction, Pubkey } from './base'
import { Invoice } from './invoice'
import { NotificationOutboxPayload } from './notification-outbox'
@@ -37,7 +37,21 @@ export interface IPaymentsService {
sendInvoiceUpdateNotification(invoice: Invoice): Promise
/** Tells the subscriber, with a kind 402 event, that their renewal invoice can be paid. */
sendRenewalNotification(invoice: Invoice): Promise
+ /** Creates a peering invoice outside the admission/user flow. */
+ createPeeringInvoice(
+ pubkey: Pubkey,
+ amountMsats: bigint,
+ description: string,
+ termPeriodSeconds: number,
+ client?: DatabaseTransaction,
+ ): Promise
+ /** Publishes a peering invoice as a signed, p-tagged kind 402 event. */
+ sendPeeringInvoiceNotification(invoice: Invoice): Promise
+ /** Publishes only when no unexpired kind 402 event for this invoice exists. */
+ ensurePeeringInvoiceNotification(invoice: Invoice): Promise
getPendingInvoices(offset?: number): Promise
+ /** Completed peering invoices whose settlement did not commit, for retry. */
+ getUnconfirmedPeeringInvoices(limit?: number, offset?: number): Promise
}
export interface ISubscriptionService {
diff --git a/src/app/maintenance-worker.ts b/src/app/maintenance-worker.ts
index b3d9e1f42..e066b3a46 100644
--- a/src/app/maintenance-worker.ts
+++ b/src/app/maintenance-worker.ts
@@ -94,6 +94,11 @@ export class MaintenanceWorker implements IRunnable {
* ten invoices that never resolve starve everything behind them.
*/
private pendingInvoiceOffset = 0
+ /**
+ * Where the next retry scan starts. A page of invoices that keep failing must
+ * not starve the ones behind them.
+ */
+ private retryPeeringInvoiceOffset = 0
public constructor(
private readonly process: NodeJS.Process,
@@ -187,12 +192,45 @@ export class MaintenanceWorker implements IRunnable {
return
}
- const invoices = await this.paymentsService.getPendingInvoices(this.pendingInvoiceOffset)
- logger('found %d pending invoices from offset %d', invoices.length, this.pendingInvoiceOffset)
-
- // A short page means we reached the end, so start over next pass.
+ // A paid peering invoice is marked completed before settlement runs. If that
+ // settlement rolled back, confirmed_at is still null and the invoice is no
+ // longer in the pending page. Pick those up first so a paid peer is never
+ // stranded between "completed" and "settled".
+ const retryOffset = this.retryPeeringInvoiceOffset
+ const retryablePeering = await this.paymentsService.getUnconfirmedPeeringInvoices(
+ PENDING_INVOICE_PAGE_SIZE,
+ retryOffset,
+ )
+ const pending = await this.paymentsService.getPendingInvoices(this.pendingInvoiceOffset)
+ logger(
+ 'found %d pending invoice(s) from offset %d and %d unconfirmed peering invoice(s) from offset %d',
+ pending.length,
+ this.pendingInvoiceOffset,
+ retryablePeering.length,
+ retryOffset,
+ )
+
+ // A short page means we reached the end, so start over next pass. A full
+ // page advances the retry cursor so the same failing invoices cannot starve
+ // the ones behind them.
this.pendingInvoiceOffset =
- invoices.length < PENDING_INVOICE_PAGE_SIZE ? 0 : this.pendingInvoiceOffset + PENDING_INVOICE_PAGE_SIZE
+ pending.length < PENDING_INVOICE_PAGE_SIZE ? 0 : this.pendingInvoiceOffset + PENDING_INVOICE_PAGE_SIZE
+ this.retryPeeringInvoiceOffset =
+ retryablePeering.length < PENDING_INVOICE_PAGE_SIZE ? 0 : retryOffset + PENDING_INVOICE_PAGE_SIZE
+
+ // A retryable peering invoice is completed, not pending, so the two lists do
+ // not normally overlap. Deduplicate anyway so a future query change cannot
+ // settle the same invoice twice in one pass.
+ const seen = new Set()
+ const invoices = [...retryablePeering, ...pending].filter((invoice) => {
+ if (seen.has(invoice.id)) {
+ return false
+ }
+
+ seen.add(invoice.id)
+ return true
+ })
+
const delay = () => delayMs(100 + Math.floor(Math.random() * 10))
let successful = 0
@@ -211,16 +249,25 @@ export class MaintenanceWorker implements IRunnable {
await this.paymentsService.updateInvoiceStatus({ id, status })
- if (
- invoice.status !== updatedInvoice.status &&
- updatedInvoice.status == InvoiceStatus.COMPLETED &&
- updatedInvoice.confirmedAt
- ) {
+ const needsConfirmation =
+ updatedInvoice.status === InvoiceStatus.COMPLETED &&
+ updatedInvoice.confirmedAt &&
+ (invoice.status !== updatedInvoice.status || invoice.confirmedAt == null)
+
+ if (needsConfirmation) {
logger('confirming invoice %s & notifying %s', invoice.id, invoice.pubkey)
+ // The processor may report a real paid amount for a peering invoice
+ // (underpayment or overpayment). Preserve it; fall back to the
+ // requested amount only when the processor does not report one.
+ const amountPaid =
+ typeof updatedInvoice.amountPaid === 'bigint'
+ ? updatedInvoice.amountPaid
+ : invoice.amountRequested
+
const update = pipe(
mergeDeepLeft(updatedInvoice),
- mergeDeepLeft({ amountPaid: invoice.amountRequested }),
+ mergeDeepLeft({ amountPaid }),
)(invoice)
await Promise.all([
@@ -232,7 +279,9 @@ export class MaintenanceWorker implements IRunnable {
}
successful++
} catch (error) {
- if (isNotFoundError(error) && isExpiredInvoice(invoice)) {
+ // Only a pending invoice has no settled period to protect. A completed
+ // invoice may be a peering settlement that rolled back; never expire it.
+ if (invoice.status === InvoiceStatus.PENDING && isNotFoundError(error) && isExpiredInvoice(invoice)) {
logger('marking expired invoice %s after payment processor returned 404', invoice.id)
await this.paymentsService.updateInvoiceStatus({
id: invoice.id,
diff --git a/src/constants/base.ts b/src/constants/base.ts
index 02e1706fe..d82f1fc0e 100644
--- a/src/constants/base.ts
+++ b/src/constants/base.ts
@@ -94,6 +94,8 @@ export enum EventTags {
Deduplication = 'd',
Expiration = 'expiration',
Invoice = 'bolt11',
+ /** NIP-12 topic tag, also used to mark peering invoice notifications. */
+ Tag = 't',
// NIP-03: target event kind on an OpenTimestamps attestation
Kind = 'k',
// NIP-25: Reactions
diff --git a/src/controllers/callbacks/lnbits-callback-controller.ts b/src/controllers/callbacks/lnbits-callback-controller.ts
index e8b93fca8..8fda6eb8b 100644
--- a/src/controllers/callbacks/lnbits-callback-controller.ts
+++ b/src/controllers/callbacks/lnbits-callback-controller.ts
@@ -94,12 +94,16 @@ export class LNbitsCallbackController implements IController {
return
}
- if (storedInvoice.status === InvoiceStatus.COMPLETED) {
+ // A completed invoice without a confirmation is a settlement that rolled
+ // back. Let the redelivery retry it instead of answering 409 forever.
+ if (storedInvoice.status === InvoiceStatus.COMPLETED && storedInvoice.confirmedAt) {
response.status(409).setHeader('content-type', 'text/plain; charset=utf8').send('Invoice is already marked paid')
return
}
- invoice.amountPaid = invoice.amountRequested
+ if (typeof invoice.amountPaid !== 'bigint') {
+ invoice.amountPaid = invoice.amountRequested
+ }
try {
await this.paymentsService.confirmInvoice(invoice as Invoice)
diff --git a/src/controllers/callbacks/nodeless-callback-controller.ts b/src/controllers/callbacks/nodeless-callback-controller.ts
index 7aca3a923..73d7e06df 100644
--- a/src/controllers/callbacks/nodeless-callback-controller.ts
+++ b/src/controllers/callbacks/nodeless-callback-controller.ts
@@ -91,8 +91,12 @@ export class NodelessCallbackController implements IController {
return
}
- invoice.amountPaid = invoice.amountRequested
- updatedInvoice.amountPaid = invoice.amountRequested
+ if (typeof invoice.amountPaid !== 'bigint') {
+ invoice.amountPaid = invoice.amountRequested
+ }
+ if (typeof updatedInvoice.amountPaid !== 'bigint') {
+ updatedInvoice.amountPaid = invoice.amountPaid
+ }
try {
await this.paymentsService.confirmInvoice(invoice)
diff --git a/src/controllers/callbacks/opennode-callback-controller.ts b/src/controllers/callbacks/opennode-callback-controller.ts
index cc349c95d..71bbb2062 100644
--- a/src/controllers/callbacks/opennode-callback-controller.ts
+++ b/src/controllers/callbacks/opennode-callback-controller.ts
@@ -78,6 +78,9 @@ export class OpenNodeCallbackController implements IController {
const statusMap: Record = {
expired: InvoiceStatus.EXPIRED,
+ // A refund only expires the invoice. Removing the paid peering period is a
+ // lifecycle policy and is deliberately not automatic here; the peer row is
+ // left untouched so an operator can decide whether to revoke access.
refunded: InvoiceStatus.EXPIRED,
unpaid: InvoiceStatus.PENDING,
processing: InvoiceStatus.PENDING,
@@ -112,7 +115,9 @@ export class OpenNodeCallbackController implements IController {
if (!updatedInvoice.confirmedAt) {
updatedInvoice.confirmedAt = new Date()
}
- updatedInvoice.amountPaid = updatedInvoice.amountRequested
+ if (typeof updatedInvoice.amountPaid !== 'bigint') {
+ updatedInvoice.amountPaid = updatedInvoice.amountRequested
+ }
try {
await this.paymentsService.confirmInvoice({
diff --git a/src/controllers/callbacks/zebedee-callback-controller.ts b/src/controllers/callbacks/zebedee-callback-controller.ts
index e4c2b792a..003325dcd 100644
--- a/src/controllers/callbacks/zebedee-callback-controller.ts
+++ b/src/controllers/callbacks/zebedee-callback-controller.ts
@@ -56,9 +56,13 @@ export class ZebedeeCallbackController implements IController {
return
}
- invoice.amountPaid = invoice.amountRequested
+ if (typeof invoice.amountPaid !== 'bigint') {
+ invoice.amountPaid = invoice.amountRequested
+ }
invoice.status = updatedInvoice.status
- updatedInvoice.amountPaid = invoice.amountRequested
+ if (typeof updatedInvoice.amountPaid !== 'bigint') {
+ updatedInvoice.amountPaid = invoice.amountPaid
+ }
try {
await this.paymentsService.confirmInvoice({
@@ -66,7 +70,7 @@ export class ZebedeeCallbackController implements IController {
pubkey: invoice.pubkey,
status: invoice.status,
confirmedAt: invoice.confirmedAt,
- amountPaid: invoice.amountRequested,
+ amountPaid: invoice.amountPaid,
})
await this.paymentsService.sendInvoiceUpdateNotification(updatedInvoice)
} catch (error) {
diff --git a/src/factories/payments-service-factory.ts b/src/factories/payments-service-factory.ts
index 07e3bcf59..b9ef8e580 100644
--- a/src/factories/payments-service-factory.ts
+++ b/src/factories/payments-service-factory.ts
@@ -4,6 +4,7 @@ import { createPaymentsProcessor } from './payments-processor-factory'
import { createSettings } from './settings-factory'
import { InvoiceRepository } from '../repositories/invoice-repository'
import { PaymentsService } from '../services/payments-service'
+import { PeeringPaymentRepository } from '../repositories/peering-payment-repository'
import { UserSubscriptionRepository } from '../repositories/user-subscription-repository'
import { UserRepository } from '../repositories/user-repository'
import { NotificationOutboxRepository } from '../repositories/notification-outbox-repository'
@@ -26,5 +27,6 @@ export const createPaymentsService = () => {
createSettings,
notificationOutboxRepository,
new UserSubscriptionRepository(dbClient),
+ new PeeringPaymentRepository(dbClient),
)
}
diff --git a/src/repositories/invoice-repository.ts b/src/repositories/invoice-repository.ts
index 36734916e..cd7428ab2 100644
--- a/src/repositories/invoice-repository.ts
+++ b/src/repositories/invoice-repository.ts
@@ -1,6 +1,6 @@
-import { always, applySpec, head, ifElse, is, map, omit, pipe, prop, propSatisfies, toString } from 'ramda'
+import { always, applySpec, head, ifElse, is, isNil, map, omit, pipe, prop, propSatisfies, toString } from 'ramda'
-import { DBInvoice, Invoice, InvoiceStatus } from '../@types/invoice'
+import { DBInvoice, Invoice, InvoicePurpose, InvoiceStatus } from '../@types/invoice'
import { fromDBInvoice, toBuffer } from '../utils/transform'
import { createLogger } from '../factories/logger-factory'
import { DatabaseClient } from '../@types/base'
@@ -29,6 +29,21 @@ export class InvoiceRepository implements IInvoiceRepository {
}
}
+ public async updatePeriod(
+ invoiceId: string,
+ periodStart: Date,
+ periodEnd: Date,
+ client: DatabaseClient = this.dbClient,
+ ): Promise {
+ await client('invoices')
+ .update({
+ period_start: periodStart,
+ period_end: periodEnd,
+ updated_at: new Date(),
+ })
+ .where('id', invoiceId)
+ }
+
public async findById(id: string, client: DatabaseClient = this.dbClient): Promise {
const [dbInvoice] = await client('invoices').where('id', id).select()
@@ -53,6 +68,29 @@ export class InvoiceRepository implements IInvoiceRepository {
return dbInvoices.map(fromDBInvoice)
}
+ public async findUnconfirmedPeeringInvoices(
+ limit = 10,
+ offset = 0,
+ client: DatabaseClient = this.dbClient,
+ ): Promise {
+ // The status is committed before settlement runs, so a rolled-back
+ // settlement leaves a completed invoice with no confirmed_at. The worker
+ // must be able to find those and try the settlement again. Offset advances
+ // the scan past invoices that keep failing so later ones are not starved.
+ const dbInvoices = await client('invoices')
+ .where({
+ purpose: InvoicePurpose.PEERING,
+ status: InvoiceStatus.COMPLETED,
+ })
+ .whereNull('confirmed_at')
+ .orderBy('created_at', 'asc')
+ .offset(offset)
+ .limit(limit)
+ .select()
+
+ return dbInvoices.map(fromDBInvoice)
+ }
+
public updateStatus(invoice: Invoice, client: DatabaseClient = this.dbClient): Promise {
logger('updating invoice status: %o', invoice)
@@ -92,9 +130,16 @@ export class InvoiceRepository implements IInvoiceRepository {
updated_at: always(new Date()),
created_at: prop('createdAt'),
verify_url: prop('verifyURL'),
+ purpose: prop('purpose'),
plan_id: prop('planId'),
period_start: prop('periodStart'),
period_end: prop('periodEnd'),
+ term_amount_msats: ifElse(
+ propSatisfies((value) => isNil(value), 'termAmountMsats'),
+ always(null),
+ pipe(prop('termAmountMsats') as () => bigint, toString),
+ ),
+ term_period_seconds: prop('termPeriodSeconds'),
})(invoice)
logger('row: %o', row)
@@ -113,9 +158,12 @@ export class InvoiceRepository implements IInvoiceRepository {
'expires_at',
'created_at',
'verify_url',
+ 'purpose',
'plan_id',
'period_start',
'period_end',
+ 'term_amount_msats',
+ 'term_period_seconds',
])(row),
)
diff --git a/src/repositories/peering-payment-repository.ts b/src/repositories/peering-payment-repository.ts
new file mode 100644
index 000000000..708cf6342
--- /dev/null
+++ b/src/repositories/peering-payment-repository.ts
@@ -0,0 +1,120 @@
+import { DatabaseClient, Pubkey } from '../@types/base'
+import { DBPeeringPayment, PeeringPayment, PeeringPaymentDirection, PeeringPaymentStatus } from '../@types/peering'
+import { IPeeringPaymentRepository } from '../@types/repositories'
+import { createLogger } from '../factories/logger-factory'
+import { fromDBPeeringPayment, toBuffer } from '../utils/transform'
+
+const logger = createLogger('peering-payment-repository')
+
+const toRow = (payment: PeeringPayment): Omit => ({
+ peer_pubkey: toBuffer(payment.peerPubkey),
+ direction: payment.direction,
+ status: payment.status,
+ current_invoice_id: payment.currentInvoiceId ?? null,
+ renewal_invoice_id: payment.renewalInvoiceId ?? null,
+ period_start: payment.periodStart ?? null,
+ period_end: payment.periodEnd ?? null,
+ term_amount_msats:
+ typeof payment.termAmountMsats === 'bigint' && payment.termAmountMsats >= 0n
+ ? payment.termAmountMsats.toString()
+ : null,
+ term_period_seconds: payment.termPeriodSeconds ?? null,
+ last_settled_invoice_id: payment.lastSettledInvoiceId ?? null,
+ last_settled_at: payment.lastSettledAt ?? null,
+ last_checked_at: payment.lastCheckedAt ?? null,
+ last_error: payment.lastError ?? null,
+})
+
+export class PeeringPaymentRepository implements IPeeringPaymentRepository {
+ public constructor(private readonly dbClient: DatabaseClient) {}
+
+ public async findByPeer(pubkey: Pubkey, client: DatabaseClient = this.dbClient): Promise {
+ const row = await client('peering_payments')
+ .where({
+ peer_pubkey: toBuffer(pubkey),
+ direction: PeeringPaymentDirection.SELL,
+ })
+ .first()
+
+ return row ? fromDBPeeringPayment(row) : undefined
+ }
+
+ public async ensureRow(pubkey: Pubkey, client: DatabaseClient): Promise {
+ await client('peering_payments')
+ .insert({
+ peer_pubkey: toBuffer(pubkey),
+ direction: PeeringPaymentDirection.SELL,
+ status: PeeringPaymentStatus.UNPAID,
+ })
+ .onConflict(['peer_pubkey', 'direction'])
+ .ignore()
+ }
+
+ public async lockByPeer(pubkey: Pubkey, client: DatabaseClient): Promise {
+ const row = await client('peering_payments')
+ .where({
+ peer_pubkey: toBuffer(pubkey),
+ direction: PeeringPaymentDirection.SELL,
+ })
+ .forUpdate()
+ .first()
+
+ return row ? fromDBPeeringPayment(row) : undefined
+ }
+
+ public async save(payment: PeeringPayment, client: DatabaseClient = this.dbClient): Promise {
+ const row = toRow(payment)
+ const now = new Date()
+
+ await client('peering_payments')
+ .update({
+ ...row,
+ updated_at: now,
+ })
+ .where({
+ peer_pubkey: row.peer_pubkey,
+ direction: row.direction,
+ })
+
+ return {
+ ...payment,
+ updatedAt: now,
+ }
+ }
+
+ public async list(): Promise {
+ const rows = await this.dbClient('peering_payments')
+ .where('direction', PeeringPaymentDirection.SELL)
+ .orderBy('peer_pubkey', 'asc')
+
+ return rows.map(fromDBPeeringPayment)
+ }
+
+ public async findDueForRenewal(before: Date, limit: number, offset: number): Promise {
+ const rows = await this.dbClient('peering_payments')
+ .where('direction', PeeringPaymentDirection.SELL)
+ .whereIn('status', [PeeringPaymentStatus.ACTIVE, PeeringPaymentStatus.RENEWAL_PENDING])
+ .whereNotNull('period_end')
+ .where('period_end', '<=', before)
+ .orderBy('period_end', 'asc')
+ .offset(offset)
+ .limit(limit)
+
+ return rows.map(fromDBPeeringPayment)
+ }
+
+ public async findPeriodEnded(now: Date, limit: number, offset: number): Promise {
+ const rows = await this.dbClient('peering_payments')
+ .where('direction', PeeringPaymentDirection.SELL)
+ .whereIn('status', [PeeringPaymentStatus.ACTIVE, PeeringPaymentStatus.RENEWAL_PENDING])
+ .whereNotNull('period_end')
+ .where('period_end', '<=', now)
+ .orderBy('period_end', 'asc')
+ .offset(offset)
+ .limit(limit)
+
+ logger('found %d peering period end(s)', rows.length)
+
+ return rows.map(fromDBPeeringPayment)
+ }
+}
diff --git a/src/schemas/admin-notifications-schema.ts b/src/schemas/admin-notifications-schema.ts
index bba837296..7f5a0e26c 100644
--- a/src/schemas/admin-notifications-schema.ts
+++ b/src/schemas/admin-notifications-schema.ts
@@ -21,6 +21,10 @@ export const adminNotificationsPatchBodySchema = z
'admission.invoice.created': z.boolean().optional(),
'admission.invoice.paid': z.boolean().optional(),
'admission.invoice.failed': z.boolean().optional(),
+ 'peering.invoice.created': z.boolean().optional(),
+ 'peering.invoice.paid': z.boolean().optional(),
+ 'peering.invoice.failed': z.boolean().optional(),
+ 'peering.access.lapsed': z.boolean().optional(),
'settings.changed': z.boolean().optional(),
'relay.restarted': z.boolean().optional(),
'probe.failed': z.boolean().optional(),
diff --git a/src/services/payments-service.ts b/src/services/payments-service.ts
index 36f6ec3f3..cbc9a2190 100644
--- a/src/services/payments-service.ts
+++ b/src/services/payments-service.ts
@@ -1,25 +1,32 @@
import { andThen, otherwise, pipe } from 'ramda'
import { broadcastEvent, getPublicKey, getRelayPrivateKey, identifyEvent, signEvent } from '../utils/event'
-import { DatabaseClient, Pubkey } from '../@types/base'
+import { DatabaseClient, DatabaseTransaction, Pubkey, Tag } from '../@types/base'
import { FeeSchedule, Settings } from '../@types/settings'
import {
IEventRepository,
IInvoiceRepository,
INotificationOutboxRepository,
+ IPeeringPaymentRepository,
IUserRepository,
IUserSubscriptionRepository,
} from '../@types/repositories'
import { OperatorNotificationEventType } from '../@types/operator-notifications'
-import { Invoice, InvoiceStatus, InvoiceUnit } from '../@types/invoice'
+import { Invoice, InvoicePurpose, InvoiceStatus, InvoiceUnit } from '../@types/invoice'
import { Event, ExpiringEvent, UnidentifiedEvent } from '../@types/event'
import { EventExpirationTimeMetadataKey, EventKinds, EventTags } from '../constants/base'
import { createLogger } from '../factories/logger-factory'
import { IPaymentsProcessor } from '../@types/clients'
import { IPaymentsService } from '../@types/services'
+import { PeeringPaymentStatus } from '../@types/peering'
import { Transaction } from '../database/transaction'
-import { applyPayment, areSubscriptionsEnabled, findSubscriptionPlan } from '../utils/subscription'
+import { applyPayment, findSubscriptionPlan } from '../utils/subscription'
import { resolveInvoiceExpiry } from '../utils/invoice'
+import {
+ InvalidPeeringTermError,
+ PEERING_INVOICE_TAG,
+ peeringPeriodEnd,
+} from '../utils/peering-billing'
const logger = createLogger('payments-service')
@@ -36,6 +43,7 @@ export class PaymentsService implements IPaymentsService {
private readonly settings: () => Settings,
private readonly notificationOutboxRepository: INotificationOutboxRepository,
private readonly userSubscriptionRepository?: IUserSubscriptionRepository,
+ private readonly peeringPaymentRepository?: IPeeringPaymentRepository,
) {}
public async getPendingInvoices(offset = 0): Promise {
@@ -49,6 +57,20 @@ export class PaymentsService implements IPaymentsService {
}
}
+ public async getUnconfirmedPeeringInvoices(
+ limit = PENDING_INVOICE_PAGE_SIZE,
+ offset = 0,
+ ): Promise {
+ logger('get unconfirmed peering invoices from offset %d', offset)
+ try {
+ return await this.invoiceRepository.findUnconfirmedPeeringInvoices(limit, offset)
+ } catch (error) {
+ logger.error('Unable to get unconfirmed peering invoices.', error)
+
+ throw error
+ }
+ }
+
public async getInvoiceFromPaymentsProcessor(invoice: Invoice | string): Promise> {
try {
return await this.paymentsProcessor.getInvoice(
@@ -96,6 +118,7 @@ export class PaymentsService implements IPaymentsService {
updatedAt: date,
createdAt: date,
verifyURL: invoiceResponse.verifyURL,
+ purpose: planId ? InvoicePurpose.SUBSCRIPTION : InvoicePurpose.ADMISSION,
planId,
},
transaction.transaction,
@@ -126,6 +149,7 @@ export class PaymentsService implements IPaymentsService {
updatedAt: date,
createdAt: invoiceResponse.createdAt,
verifyURL: invoiceResponse.verifyURL,
+ purpose: planId ? InvoicePurpose.SUBSCRIPTION : InvoicePurpose.ADMISSION,
planId,
}
} catch (error) {
@@ -136,6 +160,105 @@ export class PaymentsService implements IPaymentsService {
}
}
+ /**
+ * Creates a peering invoice without touching the user table or the admission
+ * notifications. The caller may already hold a transaction (the billing
+ * service locks the peer row before asking the processor), in which case this
+ * method joins it instead of opening a nested one.
+ */
+ public async createPeeringInvoice(
+ pubkey: Pubkey,
+ amountMsats: bigint,
+ description: string,
+ termPeriodSeconds: number,
+ client?: DatabaseTransaction,
+ ): Promise {
+ logger('create peering invoice for %s for %s msats', pubkey, amountMsats.toString())
+
+ // The boundary repeats the term validator's checks: a caller that skips
+ // resolvePaidPeeringTerm must not persist a term settlement will reject.
+ if (amountMsats <= 0n || amountMsats % 1000n !== 0n) {
+ throw new InvalidPeeringTermError(
+ 'peering invoice amount must be a positive whole number of satoshis',
+ )
+ }
+
+ if (amountMsats / 1000n > BigInt(Number.MAX_SAFE_INTEGER)) {
+ throw new InvalidPeeringTermError('peering invoice amount is too large for the payment processors')
+ }
+
+ if (!Number.isSafeInteger(termPeriodSeconds) || termPeriodSeconds <= 0) {
+ throw new InvalidPeeringTermError('peering invoice period must be a positive number of seconds')
+ }
+
+ const transaction = client ? undefined : new Transaction(this.dbClient)
+ if (transaction) {
+ await transaction.begin()
+ }
+ const db = client ?? transaction!.transaction
+
+ try {
+ const invoiceResponse = await this.paymentsProcessor.createInvoice({
+ amount: amountMsats,
+ description,
+ requestId: pubkey,
+ })
+
+ const date = new Date()
+ const expiresAt = resolveInvoiceExpiry(
+ invoiceResponse.expiresAt,
+ date,
+ this.settings()?.payments?.invoiceExpirySeconds,
+ )
+
+ const invoice: Invoice = {
+ id: invoiceResponse.id,
+ pubkey,
+ bolt11: invoiceResponse.bolt11,
+ amountRequested: invoiceResponse.amountRequested,
+ description: invoiceResponse.description,
+ unit: invoiceResponse.unit,
+ status: invoiceResponse.status,
+ expiresAt,
+ updatedAt: date,
+ createdAt: date,
+ verifyURL: invoiceResponse.verifyURL,
+ purpose: InvoicePurpose.PEERING,
+ planId: null,
+ termAmountMsats: amountMsats,
+ termPeriodSeconds,
+ }
+
+ await this.invoiceRepository.upsert(invoice, db)
+
+ await this.notificationOutboxRepository.enqueue(
+ OperatorNotificationEventType.PEERING_INVOICE_CREATED,
+ {
+ invoiceId: invoice.id,
+ pubkey,
+ amountRequested: invoice.amountRequested.toString(),
+ unit: invoice.unit,
+ termAmountMsats: amountMsats.toString(),
+ termPeriodSeconds,
+ },
+ db,
+ )
+
+ if (transaction) {
+ await transaction.commit()
+ }
+
+ return invoice
+ } catch (error) {
+ if (transaction) {
+ await transaction.rollback()
+ }
+ logger.error('Unable to create peering invoice:', error)
+
+ throw error
+ }
+ }
+
public async updateInvoice(invoice: Partial): Promise {
logger('update invoice %s: %o', invoice.id, invoice)
try {
@@ -179,12 +302,11 @@ export class PaymentsService implements IPaymentsService {
await transaction.begin()
const currentSettings = this.settings()
- const subscriptionsEnabled = areSubscriptionsEnabled(currentSettings)
- // The callbacks only pass what their processor reports, so the plan, the unit and whether the
- // invoice was already confirmed come from the stored invoice, read before it is confirmed.
- const stored = subscriptionsEnabled
- ? await this.invoiceRepository.findById(invoice.id, transaction.transaction)
- : undefined
+ // The callbacks only pass what their processor reports, so the purpose, plan, unit and
+ // whether the invoice was already confirmed come from the stored invoice. This read must
+ // happen before confirmation and must not be gated on subscriptions: peering invoices
+ // need it even when subscription plans are off.
+ const stored = await this.invoiceRepository.findById(invoice.id, transaction.transaction)
await this.invoiceRepository.confirmInvoice(
invoice.id,
@@ -202,6 +324,41 @@ export class PaymentsService implements IPaymentsService {
amountPaidMsat *= 1000n * 100000000n
}
+ if (stored?.purpose === InvoicePurpose.PEERING) {
+ // The stored row is read inside this transaction. Once an invoice is confirmed,
+ // a redelivered older callback must not buy another period even if the peer has
+ // since settled a newer invoice.
+ if (stored.confirmedAt) {
+ await transaction.commit()
+ return
+ }
+
+ const result = await this.settlePeeringInvoice(invoice, stored, amountPaidMsat, transaction.transaction)
+
+ if (result.duplicate) {
+ await transaction.commit()
+ return
+ }
+
+ await this.notificationOutboxRepository.enqueue(
+ result.activated
+ ? OperatorNotificationEventType.PEERING_INVOICE_PAID
+ : OperatorNotificationEventType.PEERING_INVOICE_FAILED,
+ {
+ invoiceId: invoice.id,
+ pubkey: invoice.pubkey,
+ amountPaid: invoice.amountPaid.toString(),
+ unit,
+ confirmedAt: invoice.confirmedAt.toISOString(),
+ reason: result.reason,
+ },
+ transaction.transaction,
+ )
+
+ await transaction.commit()
+ return
+ }
+
const isApplicableFee = (feeSchedule: FeeSchedule) =>
feeSchedule.enabled && !feeSchedule.whitelists?.pubkeys?.includes(invoice.pubkey)
const admissionFeeSchedules = currentSettings.payments?.feeSchedules?.admission ?? []
@@ -255,6 +412,131 @@ export class PaymentsService implements IPaymentsService {
}
}
+ private async settlePeeringInvoice(
+ invoice: Invoice,
+ stored: Invoice,
+ amountPaidMsat: bigint,
+ client: DatabaseTransaction,
+ ): Promise<{ activated: boolean; duplicate?: boolean; reason?: string }> {
+ if (!this.peeringPaymentRepository) {
+ throw new Error('Peering payment repository is not configured')
+ }
+
+ const termAmountMsats = stored.termAmountMsats ?? 0n
+ const termPeriodSeconds = stored.termPeriodSeconds ?? 0
+
+ if (termAmountMsats <= 0n || termPeriodSeconds <= 0) {
+ throw new Error(`Peering invoice ${invoice.id} has no term snapshot`)
+ }
+
+ const payment = await this.peeringPaymentRepository.lockByPeer(invoice.pubkey, client)
+ if (!payment) {
+ throw new Error(`Peering payment row is missing for ${invoice.pubkey}`)
+ }
+
+ if (payment.lastSettledInvoiceId === invoice.id) {
+ // A duplicate callback or a worker poll after the webhook: the period was already granted.
+ return { activated: true, duplicate: true }
+ }
+
+ // The invoice's confirmedAt is the receipt timestamp. It is not reliable as a
+ // payment time: LNbits, for example, fills it from the invoice creation time.
+ // The access window starts when this relay actually grants it.
+ const settledAt = new Date()
+
+ if (amountPaidMsat < termAmountMsats) {
+ logger(
+ 'peering invoice %s paid %s msats, below term %s msats',
+ invoice.id,
+ amountPaidMsat.toString(),
+ termAmountMsats.toString(),
+ )
+
+ await this.peeringPaymentRepository.save(
+ {
+ ...payment,
+ lastCheckedAt: settledAt,
+ lastError: `underpaid: received ${amountPaidMsat.toString()} of ${termAmountMsats.toString()} msats`,
+ },
+ client,
+ )
+
+ return { activated: false, reason: 'underpaid' }
+ }
+
+ // Overpayments are not credited to the peer row. The invoice's amount_paid
+ // is the receipt; an operator can refund the surplus out of band.
+ // Paying early extends from the current period end; paying after a lapse starts now.
+ const stillRunning = payment.periodEnd instanceof Date && payment.periodEnd.getTime() > settledAt.getTime()
+ const periodStart = stillRunning ? (payment.periodEnd as Date) : settledAt
+ const periodEnd = peeringPeriodEnd(periodStart, termPeriodSeconds)
+
+ await this.peeringPaymentRepository.save(
+ {
+ ...payment,
+ status: PeeringPaymentStatus.ACTIVE,
+ currentInvoiceId: null,
+ renewalInvoiceId: null,
+ periodStart,
+ periodEnd,
+ lastSettledInvoiceId: invoice.id,
+ lastSettledAt: settledAt,
+ lastCheckedAt: settledAt,
+ lastError: null,
+ },
+ client,
+ )
+
+ // The invoice is the ledger entry for the period it bought; the peer row is the live window.
+ await this.invoiceRepository.updatePeriod(invoice.id, periodStart, periodEnd, client)
+
+ return { activated: true }
+ }
+
+ /**
+ * A peer may have missed the original broadcast. Republish only when the
+ * expiring kind 402 event is not already stored, so a re-request does not
+ * create one event per poll.
+ */
+ public async ensurePeeringInvoiceNotification(invoice: Invoice): Promise {
+ const existing = await this.eventRepository.findByFilters(
+ [
+ {
+ kinds: [EventKinds.INVOICE_UPDATE],
+ '#i': [invoice.id],
+ limit: 1,
+ },
+ ],
+ { excludeDeleted: true },
+ )
+
+ if (existing.length > 0) {
+ return
+ }
+
+ await this.sendPeeringInvoiceNotification(invoice)
+ }
+
+ /** Publishes the peering invoice as a signed, p-tagged kind 402 event. */
+ public async sendPeeringInvoiceNotification(invoice: Invoice): Promise {
+ logger('peering invoice notification %s: %o', invoice.id, invoice)
+
+ const periodSeconds = invoice.termPeriodSeconds ?? 0
+ const days = periodSeconds > 0 ? Math.max(1, Math.round(periodSeconds / 86_400)) : 0
+ const content = `Paid peering invoice: ${invoice.amountRequested.toString()} ${invoice.unit}${
+ days ? ` for ${days} day(s)` : ''
+ }`
+
+ const expiration = invoice.expiresAt ?? new Date(Date.now() + 24 * 60 * 60 * 1000)
+
+ const extraTags: Tag[] = [
+ [EventTags.Tag, PEERING_INVOICE_TAG],
+ [EventTags.Index, invoice.id],
+ ]
+
+ await this.publishInvoiceNotification(invoice, content, expiration, extraTags)
+ }
+
public async sendInvoiceUpdateNotification(invoice: Invoice): Promise {
logger('invoice updated notification %s: %o', invoice.id, invoice)
@@ -293,7 +575,12 @@ export class PaymentsService implements IPaymentsService {
await this.publishInvoiceNotification(invoice, `Subscription renewal due: ${amount.toString()} ${unit}`, expiration)
}
- private async publishInvoiceNotification(invoice: Invoice, content: string, expiration: Date): Promise {
+ private async publishInvoiceNotification(
+ invoice: Invoice,
+ content: string,
+ expiration: Date,
+ extraTags: Tag[] = [],
+ ): Promise {
const {
info: { relay_url: relayUrl },
} = this.settings()
@@ -310,6 +597,7 @@ export class PaymentsService implements IPaymentsService {
content,
tags: [
[EventTags.Pubkey, invoice.pubkey],
+ ...extraTags,
[EventTags.Invoice, invoice.bolt11],
[EventTags.Expiration, Math.floor(expiration.getTime() / 1000).toString()],
],
diff --git a/src/utils/peering-billing.ts b/src/utils/peering-billing.ts
new file mode 100644
index 000000000..9ce024d5b
--- /dev/null
+++ b/src/utils/peering-billing.ts
@@ -0,0 +1,143 @@
+import { PeeringTermSnapshot } from '../@types/peering'
+import { Settings } from '../@types/settings'
+import { SubscriptionFilter } from '../@types/subscription'
+import { EventKinds } from '../constants/base'
+
+/** Topic tag carried by every peering invoice notification. */
+export const PEERING_INVOICE_TAG = 'peering'
+
+/** How long before a paid period ends the relay offers its renewal. */
+export const PEERING_RENEWAL_LEAD_SECONDS = 3 * 24 * 60 * 60
+
+/** A peer that just let an invoice expire cannot immediately mint another one. */
+export const PEERING_REISSUE_COOLDOWN_SECONDS = 15
+
+/** Do not ask the processor about the same unpaid peer more often than this. */
+export const PEERING_RECHECK_COOLDOWN_SECONDS = 30
+
+const UNIT_MULTIPLIER: Record = {
+ msats: 1n,
+ sats: 1000n,
+ btc: 100_000_000_000n,
+}
+
+export class InvalidPeeringTermError extends Error {
+ public constructor(message: string) {
+ super(message)
+ this.name = 'InvalidPeeringTermError'
+ }
+}
+
+export class PeeringInvoiceCooldownError extends Error {
+ public constructor(public readonly retryAfterSeconds: number) {
+ super(`peering invoice reissue is cooling down for ${retryAfterSeconds}s`)
+ this.name = 'PeeringInvoiceCooldownError'
+ }
+}
+
+const toMsats = (amount: number, unit: string): bigint | undefined => {
+ const multiplier = UNIT_MULTIPLIER[unit.trim().toLowerCase()]
+ if (!multiplier) {
+ return undefined
+ }
+
+ return BigInt(amount) * multiplier
+}
+
+/**
+ * The paid term this relay sells. An empty list, or only zero-amount terms, means
+ * settlement-free peering and returns undefined. A configured paid term that cannot
+ * be charged safely throws instead of silently granting free access.
+ */
+export const resolvePaidPeeringTerm = (settings: Settings): PeeringTermSnapshot | undefined => {
+ const terms = settings.peering?.terms ?? []
+
+ for (const term of terms) {
+ if (typeof term !== 'object' || term === null) {
+ throw new InvalidPeeringTermError('peering term must be an object')
+ }
+
+ if (typeof term.amount !== 'number' || !Number.isSafeInteger(term.amount)) {
+ throw new InvalidPeeringTermError('peering term amount must be a safe integer')
+ }
+
+ // Zero is the explicit settlement-free term. Every other malformed amount must
+ // fail closed: returning undefined would grant free peering by accident.
+ if (term.amount === 0) {
+ continue
+ }
+
+ if (term.amount < 0) {
+ throw new InvalidPeeringTermError('peering term amount must not be negative')
+ }
+
+ const amountMsats = toMsats(term.amount, term.unit)
+ if (amountMsats === undefined) {
+ throw new InvalidPeeringTermError(`unsupported peering term unit "${term.unit}"; use msats, sats or btc`)
+ }
+
+ if (amountMsats <= 0n) {
+ throw new InvalidPeeringTermError('peering term amount must be positive')
+ }
+
+ // The processor adapters take msats but several truncate to whole sats. A
+ // sub-satoshi term would silently undercharge, so refuse it at the boundary.
+ if (amountMsats % 1000n !== 0n) {
+ throw new InvalidPeeringTermError('peering term amount must be a whole number of satoshis')
+ }
+
+ // Processor adapters convert msats to sats with Number(). Reject a term that
+ // would cross Number.MAX_SAFE_INTEGER sats, or the invoice would be created
+ // for a rounded amount that no longer matches the stored term.
+ if (amountMsats / 1000n > BigInt(Number.MAX_SAFE_INTEGER)) {
+ throw new InvalidPeeringTermError('peering term amount is too large for the payment processors')
+ }
+
+ if (!(typeof term.period === 'number' && Number.isSafeInteger(term.period) && term.period > 0)) {
+ throw new InvalidPeeringTermError('peering term period must be a positive number of seconds')
+ }
+
+ return {
+ amountMsats,
+ periodSeconds: term.period,
+ }
+ }
+
+ return undefined
+}
+
+export const isPeeringInvoiceFilter = (filter: SubscriptionFilter, peerPubkey: string): boolean => {
+ const kinds = filter.kinds
+ const pubkeys = filter['#p']
+ const topics = filter['#t']
+
+ return (
+ Array.isArray(kinds) &&
+ kinds.length === 1 &&
+ kinds[0] === EventKinds.INVOICE_UPDATE &&
+ Array.isArray(pubkeys) &&
+ pubkeys.includes(peerPubkey) &&
+ Array.isArray(topics) &&
+ topics.includes(PEERING_INVOICE_TAG)
+ )
+}
+
+export const isPeeringInvoiceRequest = (filters: SubscriptionFilter[], peerPubkey: string): boolean =>
+ filters.length > 0 && filters.every((filter) => isPeeringInvoiceFilter(filter, peerPubkey))
+
+export const peeringPeriodEnd = (start: Date, periodSeconds: number): Date =>
+ new Date(start.getTime() + periodSeconds * 1000)
+
+export const peeringInvoiceDescription = (
+ settings: Settings,
+ peerPubkey: string,
+ term: PeeringTermSnapshot,
+): string => {
+ const days = Math.max(1, Math.round(term.periodSeconds / 86_400))
+ return `${settings.info.name} paid peering for ${peerPubkey.slice(0, 12)}…: ${term.amountMsats.toString()} msats for ${days} day(s)`
+}
+
+export const isPeeringAccessActive = (status: string, periodEnd: Date | null | undefined, now: Date): boolean =>
+ (status === 'active' || status === 'renewal_pending') &&
+ periodEnd instanceof Date &&
+ periodEnd.getTime() > now.getTime()
diff --git a/src/utils/transform.ts b/src/utils/transform.ts
index a9021037d..2f083a927 100644
--- a/src/utils/transform.ts
+++ b/src/utils/transform.ts
@@ -16,6 +16,7 @@ import {
} from 'ramda'
import { Invoice, InvoiceStatus, InvoiceUnit } from '../@types/invoice'
+import { DBPeeringPayment, PeeringPayment } from '../@types/peering'
import { User } from '../@types/user'
const BECH32_ALPHABET = 'qpzry9x8gf2tvdw0s3jn54khce6mua7l'
@@ -132,9 +133,38 @@ export const fromDBInvoice = applySpec({
updatedAt: prop('updated_at'),
createdAt: prop('created_at'),
verifyURL: prop('verify_url'),
+ purpose: prop('purpose'),
planId: prop('plan_id'),
periodStart: prop('period_start'),
periodEnd: prop('period_end'),
+ termAmountMsats: ifElse(
+ propSatisfies(isNil, 'term_amount_msats'),
+ always(undefined),
+ pipe(prop('term_amount_msats') as () => string, toBigInt),
+ ),
+ termPeriodSeconds: prop('term_period_seconds'),
+})
+
+export const fromDBPeeringPayment = applySpec({
+ peerPubkey: pipe(prop('peer_pubkey') as () => Buffer, fromBuffer),
+ direction: prop('direction'),
+ status: prop('status'),
+ currentInvoiceId: prop('current_invoice_id'),
+ renewalInvoiceId: prop('renewal_invoice_id'),
+ periodStart: prop('period_start'),
+ periodEnd: prop('period_end'),
+ termAmountMsats: ifElse(
+ propSatisfies(isNil, 'term_amount_msats'),
+ always(undefined),
+ pipe(prop('term_amount_msats') as () => string, toBigInt),
+ ),
+ termPeriodSeconds: prop('term_period_seconds'),
+ lastSettledInvoiceId: prop('last_settled_invoice_id'),
+ lastSettledAt: prop('last_settled_at'),
+ lastCheckedAt: prop('last_checked_at'),
+ lastError: prop('last_error'),
+ createdAt: prop('created_at'),
+ updatedAt: prop('updated_at'),
})
export const fromDBUser = applySpec({
diff --git a/test/integration/features/peering/paid-peering-ledger.feature b/test/integration/features/peering/paid-peering-ledger.feature
new file mode 100644
index 000000000..b62534f60
--- /dev/null
+++ b/test/integration/features/peering/paid-peering-ledger.feature
@@ -0,0 +1,30 @@
+@paid-peering-ledger
+Feature: Paid peering billing ledger
+
+ The ledger records a peering invoice, grants exactly one paid period, and keeps
+ the payment separate from the user admission and balance flow.
+
+ Scenario: A paid peering invoice activates exactly one term
+ When the peer pays a 21000 msat peering invoice
+ Then the peer has an active peering period for 60 seconds
+ And the peering payment did not credit a user balance
+ When the same peering invoice is paid again
+ Then the peer period did not change
+
+ Scenario: An underpaid peering invoice is recorded without activating access
+ When the peer underpays a 21000 msat peering invoice with 1000 msats
+ Then the underpayment is recorded without activating the peer
+
+ Scenario: A completed but unconfirmed peering invoice is retryable
+ When a peering invoice is marked completed without a confirmation
+ Then the unconfirmed peering invoice is returned for retry
+ When the retryable peering invoice is paid
+ Then the unconfirmed peering invoice is no longer returned for retry
+
+ Scenario: Terms, invoice requests, notification switches and creation boundaries
+ When I parse the paid peering terms
+ Then only all-invoice peering requests are exempt
+ And the peering notification switches are accepted
+ And the peering invoice creation boundary rejects invalid terms
+ When the peering invoice notification is published
+ Then the invoice notification is published once with peering tags
diff --git a/test/integration/features/peering/paid-peering-ledger.feature.ts b/test/integration/features/peering/paid-peering-ledger.feature.ts
new file mode 100644
index 000000000..6318e0d88
--- /dev/null
+++ b/test/integration/features/peering/paid-peering-ledger.feature.ts
@@ -0,0 +1,501 @@
+import { After, Before, Then, When, World } from '@cucumber/cucumber'
+import { expect } from 'chai'
+import { randomBytes, randomUUID } from 'node:crypto'
+import Sinon from 'sinon'
+
+import { adminNotificationsPatchBodySchema } from '../../../../src/schemas/admin-notifications-schema'
+import { getMasterDbClient } from '../../../../src/database/client'
+import { InvoiceStatus, InvoiceUnit } from '../../../../src/@types/invoice'
+import { InvoiceRepository } from '../../../../src/repositories/invoice-repository'
+import { PeeringPaymentRepository } from '../../../../src/repositories/peering-payment-repository'
+import { PaymentsService } from '../../../../src/services/payments-service'
+import { SettingsStatic } from '../../../../src/utils/settings'
+import * as eventUtils from '../../../../src/utils/event'
+import {
+ InvalidPeeringTermError,
+ isPeeringAccessActive,
+ isPeeringInvoiceRequest,
+ peeringInvoiceDescription,
+ peeringPeriodEnd,
+ resolvePaidPeeringTerm,
+} from '../../../../src/utils/peering-billing'
+
+const TERM_MSATS = 21000n
+const TERM_SECONDS = 60
+
+const randomPeer = () => randomBytes(32).toString('hex')
+
+const rejects = async (promise: Promise): Promise => {
+ try {
+ await promise
+ return false
+ } catch {
+ return true
+ }
+}
+
+const createService = (world: World>): void => {
+ const db = getMasterDbClient()
+ const invoiceRepository = new InvoiceRepository(db)
+ const peeringPaymentRepository = new PeeringPaymentRepository(db)
+
+ const eventRepository = {
+ findByFilters: async () => world.parameters.peeringEvents.map(() => ({ event_id: Buffer.alloc(32) })),
+ create: async (event: unknown) => {
+ world.parameters.peeringEvents.push(event)
+ world.parameters.peeringEventCreates = (world.parameters.peeringEventCreates ?? 0) + 1
+ return 1
+ },
+ } as any
+
+ const userRepository = {
+ upsert: async () => {
+ throw new Error('peering must not upsert a user')
+ },
+ admitUser: async () => {
+ throw new Error('peering must not admit a user')
+ },
+ getBalanceByPubkey: async () => 0n,
+ } as any
+
+ const notificationOutboxRepository = {
+ enqueue: async () => ({}),
+ } as any
+
+ const paymentsProcessor = {
+ createInvoice: async (request: any) => {
+ world.parameters.peeringInvoiceCreates = (world.parameters.peeringInvoiceCreates ?? 0) + 1
+ return {
+ id: `peering-${randomUUID()}`,
+ pubkey: request.requestId,
+ bolt11: `lnbc${Math.max(1, Number(request.amount / 1000n))}n1peering`,
+ amountRequested: request.amount,
+ description: request.description,
+ unit: InvoiceUnit.MSATS,
+ status: InvoiceStatus.PENDING,
+ expiresAt: new Date(Date.now() + 900_000),
+ createdAt: new Date(),
+ }
+ },
+ getInvoice: async () => {
+ throw new Error('processor lookup is not used by the ledger test')
+ },
+ } as any
+
+ world.parameters.peeringDb = db
+ world.parameters.peeringInvoiceRepository = invoiceRepository
+ world.parameters.peeringPaymentRepository = peeringPaymentRepository
+ world.parameters.peeringService = new PaymentsService(
+ db,
+ paymentsProcessor,
+ userRepository,
+ invoiceRepository,
+ eventRepository,
+ () => SettingsStatic._settings as any,
+ notificationOutboxRepository,
+ undefined,
+ peeringPaymentRepository,
+ )
+}
+
+const createInvoice = async (
+ world: World>,
+ peer: string,
+ amountMsats = TERM_MSATS,
+ periodSeconds = TERM_SECONDS,
+) => {
+ const service = world.parameters.peeringService as PaymentsService
+ const repository = world.parameters.peeringPaymentRepository as PeeringPaymentRepository
+ const db = world.parameters.peeringDb
+
+ const invoice = await service.createPeeringInvoice(
+ peer,
+ amountMsats,
+ 'paid peering integration test',
+ periodSeconds,
+ )
+
+ world.parameters.peeringInvoiceIds.push(invoice.id)
+ if (!world.parameters.peeringPeerPubkeys.includes(peer)) {
+ world.parameters.peeringPeerPubkeys.push(peer)
+ }
+
+ if (!world.parameters.peeringUserPubkeys.includes(peer)) {
+ await db('users').insert({ pubkey: Buffer.from(peer, 'hex') }).onConflict('pubkey').ignore()
+ world.parameters.peeringUserPubkeys.push(peer)
+ }
+
+ await repository.ensureRow(peer, db)
+
+ return invoice
+}
+
+const payInvoice = async (
+ world: World>,
+ invoice: any,
+ amountPaid: bigint = invoice.amountRequested,
+) => {
+ const service = world.parameters.peeringService as PaymentsService
+ await service.confirmInvoice({
+ ...invoice,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: new Date(),
+ amountPaid,
+ })
+}
+
+const peerRow = (world: World>, peer: string) =>
+ world.parameters.peeringDb('peering_payments').where('peer_pubkey', Buffer.from(peer, 'hex')).first()
+
+Before({ tags: '@paid-peering-ledger' }, function (this: World>) {
+ this.parameters.peeringPreviousSettings = SettingsStatic._settings
+ this.parameters.peeringPeerPubkeys = []
+ this.parameters.peeringInvoiceIds = []
+ this.parameters.peeringUserPubkeys = []
+ this.parameters.peeringEvents = []
+ this.parameters.peeringEventCreates = 0
+ this.parameters.peeringInvoiceCreates = 0
+ this.parameters.peeringSandbox = Sinon.createSandbox()
+ this.parameters.peeringSandbox.stub(eventUtils, 'broadcastEvent').resolves()
+
+ const settings = SettingsStatic._settings as any
+ SettingsStatic._settings = {
+ ...settings,
+ peering: {
+ ...(settings?.peering ?? {}),
+ enabled: true,
+ terms: [{ amount: Number(TERM_MSATS), unit: 'msats', period: TERM_SECONDS }],
+ },
+ } as any
+
+ createService(this)
+})
+
+After({ tags: '@paid-peering-ledger' }, async function (this: World>) {
+ const db = this.parameters.peeringDb
+ const peers: string[] = this.parameters.peeringPeerPubkeys ?? []
+ const invoices: string[] = this.parameters.peeringInvoiceIds ?? []
+ const users: string[] = this.parameters.peeringUserPubkeys ?? []
+
+ try {
+ if (db && peers.length > 0) {
+ await db('peering_payments')
+ .whereIn(
+ 'peer_pubkey',
+ peers.map((peer) => Buffer.from(peer, 'hex')),
+ )
+ .delete()
+ }
+ if (db && invoices.length > 0) {
+ await db('invoices').whereIn('id', invoices).delete()
+ }
+ if (db && users.length > 0) {
+ await db('users')
+ .whereIn(
+ 'pubkey',
+ users.map((peer) => Buffer.from(peer, 'hex')),
+ )
+ .delete()
+ }
+ } finally {
+ this.parameters.peeringSandbox?.restore()
+ if (this.parameters.peeringPreviousSettings) {
+ SettingsStatic._settings = this.parameters.peeringPreviousSettings
+ }
+
+ this.parameters.peeringDb = undefined
+ this.parameters.peeringInvoiceRepository = undefined
+ this.parameters.peeringPaymentRepository = undefined
+ this.parameters.peeringService = undefined
+ this.parameters.peeringPreviousSettings = undefined
+ this.parameters.peeringSandbox = undefined
+ this.parameters.peeringEvents = []
+ this.parameters.peeringPeerPubkeys = []
+ this.parameters.peeringInvoiceIds = []
+ this.parameters.peeringUserPubkeys = []
+ this.parameters.peeringInvoice = undefined
+ this.parameters.peeringPeer = undefined
+ this.parameters.peeringFirstPeriodEnd = undefined
+ this.parameters.peeringUnderpaidInvoice = undefined
+ this.parameters.peeringUnderpaidPeer = undefined
+ this.parameters.peeringRetryInvoice = undefined
+ this.parameters.peeringRetryPeer = undefined
+ this.parameters.peeringNotificationInvoice = undefined
+ this.parameters.peeringNotificationPeer = undefined
+ }
+})
+
+When('the peer pays a {int} msat peering invoice', async function (this: World>, amount: number) {
+ const peer = randomPeer()
+ const invoice = await createInvoice(this, peer, BigInt(amount))
+
+ await payInvoice(this, invoice)
+
+ this.parameters.peeringPeer = peer
+ this.parameters.peeringInvoice = invoice
+})
+
+Then(
+ 'the peer has an active peering period for {int} seconds',
+ async function (this: World>, seconds: number) {
+ const peer = this.parameters.peeringPeer as string
+ const invoice = this.parameters.peeringInvoice as any
+ const row = await peerRow(this, peer)
+
+ expect(row).to.exist
+ expect(row.status).to.equal('active')
+ expect(row.last_settled_invoice_id).to.equal(invoice.id)
+ expect(row.period_start).to.be.instanceOf(Date)
+ expect(row.period_end).to.be.instanceOf(Date)
+ expect(Math.round((row.period_end.getTime() - row.period_start.getTime()) / 1000)).to.equal(seconds)
+ expect(row.period_end.getTime()).to.be.greaterThan(Date.now())
+
+ const invoiceRow = await this.parameters.peeringDb('invoices').where('id', invoice.id).first()
+ expect(invoiceRow.confirmed_at).to.be.instanceOf(Date)
+ expect(invoiceRow.period_start).to.be.instanceOf(Date)
+ expect(invoiceRow.period_end).to.be.instanceOf(Date)
+
+ const repository = this.parameters.peeringPaymentRepository as PeeringPaymentRepository
+ const found = await repository.findByPeer(peer)
+ expect(found?.status).to.equal('active')
+
+ const listed = await repository.list()
+ expect(listed.some((payment) => payment.peerPubkey === peer)).to.equal(true)
+
+ const before = new Date(Date.now() + 2 * seconds * 1000)
+ const due = await repository.findDueForRenewal(before, 10, 0)
+ expect(due.some((payment) => payment.peerPubkey === peer)).to.equal(true)
+
+ const ended = await repository.findPeriodEnded(before, 10, 0)
+ expect(ended.some((payment) => payment.peerPubkey === peer)).to.equal(true)
+ },
+)
+
+Then('the peering payment did not credit a user balance', async function (this: World>) {
+ const peer = this.parameters.peeringPeer as string
+ const user = await this.parameters.peeringDb('users').where('pubkey', Buffer.from(peer, 'hex')).first()
+
+ expect(user).to.exist
+ expect(String(user.balance)).to.equal('0')
+ expect(user.is_admitted).to.equal(false)
+})
+
+When('the same peering invoice is paid again', async function (this: World>) {
+ const peer = this.parameters.peeringPeer as string
+ const row = await peerRow(this, peer)
+ this.parameters.peeringFirstPeriodEnd = row.period_end
+
+ await payInvoice(this, this.parameters.peeringInvoice)
+})
+
+Then('the peer period did not change', async function (this: World>) {
+ const peer = this.parameters.peeringPeer as string
+ const row = await peerRow(this, peer)
+
+ expect(row.period_end.getTime()).to.equal(this.parameters.peeringFirstPeriodEnd.getTime())
+ expect(row.last_settled_invoice_id).to.equal(this.parameters.peeringInvoice.id)
+})
+
+When(
+ 'the peer underpays a {int} msat peering invoice with {int} msats',
+ async function (this: World>, amount: number, paid: number) {
+ const peer = randomPeer()
+ const invoice = await createInvoice(this, peer, BigInt(amount))
+
+ await payInvoice(this, invoice, BigInt(paid))
+
+ this.parameters.peeringUnderpaidPeer = peer
+ this.parameters.peeringUnderpaidInvoice = invoice
+ },
+)
+
+Then(
+ 'the underpayment is recorded without activating the peer',
+ async function (this: World>) {
+ const peer = this.parameters.peeringUnderpaidPeer as string
+ const invoice = this.parameters.peeringUnderpaidInvoice as any
+ const row = await peerRow(this, peer)
+
+ expect(row).to.exist
+ expect(row.status).to.equal('unpaid')
+ expect(row.period_end).to.equal(null)
+ expect(row.last_settled_invoice_id).to.equal(null)
+ expect(row.last_error).to.include('underpaid')
+
+ const invoiceRow = await this.parameters.peeringDb('invoices').where('id', invoice.id).first()
+ expect(invoiceRow.confirmed_at).to.be.instanceOf(Date)
+ expect(invoiceRow.period_end).to.equal(null)
+ },
+)
+
+When(
+ 'a peering invoice is marked completed without a confirmation',
+ async function (this: World>) {
+ const peer = randomPeer()
+ const invoice = await createInvoice(this, peer)
+
+ await this.parameters.peeringInvoiceRepository.updateStatus({
+ id: invoice.id,
+ status: InvoiceStatus.COMPLETED,
+ })
+
+ this.parameters.peeringRetryPeer = peer
+ this.parameters.peeringRetryInvoice = invoice
+ },
+)
+
+Then('the unconfirmed peering invoice is returned for retry', async function (this: World>) {
+ const service = this.parameters.peeringService as PaymentsService
+ const invoice = this.parameters.peeringRetryInvoice as any
+ const retryable = await service.getUnconfirmedPeeringInvoices(100, 0)
+
+ expect(retryable.some((candidate) => candidate.id === invoice.id)).to.equal(true)
+})
+
+When('the retryable peering invoice is paid', async function (this: World>) {
+ await payInvoice(this, this.parameters.peeringRetryInvoice)
+})
+
+Then(
+ 'the unconfirmed peering invoice is no longer returned for retry',
+ async function (this: World>) {
+ const service = this.parameters.peeringService as PaymentsService
+ const invoice = this.parameters.peeringRetryInvoice as any
+ const retryable = await service.getUnconfirmedPeeringInvoices(100, 0)
+
+ expect(retryable.some((candidate) => candidate.id === invoice.id)).to.equal(false)
+ },
+)
+
+When('I parse the paid peering terms', function () {
+ const base = SettingsStatic._settings as any
+ const withTerms = (terms: unknown[]) => ({
+ ...base,
+ peering: {
+ ...(base?.peering ?? {}),
+ terms,
+ },
+ })
+
+ const valid = [
+ { amount: 21000, unit: 'msats', period: 60, amountMsats: 21000n },
+ { amount: 21, unit: 'sats', period: 60, amountMsats: 21000n },
+ { amount: 1, unit: 'btc', period: 60, amountMsats: 100_000_000_000n },
+ ]
+
+ for (const term of valid) {
+ expect(resolvePaidPeeringTerm(withTerms([term]))).to.deep.equal({
+ amountMsats: term.amountMsats,
+ periodSeconds: term.period,
+ })
+ }
+
+ expect(resolvePaidPeeringTerm(withTerms([{ amount: 0, unit: 'msats', period: 60 }]))).to.be.undefined
+
+ const invalid = [
+ { amount: -1, unit: 'msats', period: 60 },
+ { amount: 1500, unit: 'msats', period: 60 },
+ { amount: 1, unit: 'usd', period: 60 },
+ { amount: 100_000_000, unit: 'btc', period: 60 },
+ { amount: 21000, unit: 'msats', period: 0 },
+ ]
+
+ for (const term of invalid) {
+ expect(() => resolvePaidPeeringTerm(withTerms([term]))).to.throw(InvalidPeeringTermError)
+ }
+
+ const start = new Date('2026-01-01T00:00:00.000Z')
+ expect(peeringPeriodEnd(start, 60).getTime() - start.getTime()).to.equal(60_000)
+ expect(
+ peeringInvoiceDescription(base, randomPeer(), { amountMsats: 21000n, periodSeconds: 86_400 }),
+ ).to.include('21000 msats')
+
+ const now = new Date()
+ expect(isPeeringAccessActive('active', new Date(now.getTime() + 60_000), now)).to.equal(true)
+ expect(isPeeringAccessActive('active', new Date(now.getTime() - 60_000), now)).to.equal(false)
+ expect(isPeeringAccessActive('lapsed', new Date(now.getTime() + 60_000), now)).to.equal(false)
+ expect(isPeeringAccessActive('renewal_pending', new Date(now.getTime() + 60_000), now)).to.equal(true)
+})
+
+Then('only all-invoice peering requests are exempt', function () {
+ const peer = randomPeer()
+ const invoiceFilter = { kinds: [402], '#p': [peer], '#t': ['peering'] }
+
+ expect(isPeeringInvoiceRequest([invoiceFilter], peer)).to.equal(true)
+ expect(isPeeringInvoiceRequest([invoiceFilter, { kinds: [1] }], peer)).to.equal(false)
+ expect(isPeeringInvoiceRequest([], peer)).to.equal(false)
+ expect(isPeeringInvoiceRequest([{ ...invoiceFilter, '#p': [randomPeer()] }], peer)).to.equal(false)
+ expect(isPeeringInvoiceRequest([{ kinds: [402], '#t': ['peering'] }], peer)).to.equal(false)
+ expect(isPeeringInvoiceRequest([{ kinds: [402, 1], '#p': [peer], '#t': ['peering'] }], peer)).to.equal(false)
+})
+
+Then('the peering notification switches are accepted', function () {
+ const accepted = adminNotificationsPatchBodySchema.safeParse({
+ events: {
+ 'peering.invoice.created': false,
+ 'peering.invoice.paid': false,
+ 'peering.invoice.failed': false,
+ 'peering.access.lapsed': false,
+ },
+ })
+ expect(accepted.success).to.equal(true)
+
+ const rejected = adminNotificationsPatchBodySchema.safeParse({
+ events: { 'peering.unknown': false },
+ })
+ expect(rejected.success).to.equal(false)
+})
+
+Then(
+ 'the peering invoice creation boundary rejects invalid terms',
+ async function (this: World>) {
+ const service = this.parameters.peeringService as PaymentsService
+ const peer = randomPeer()
+ const before = this.parameters.peeringInvoiceCreates
+ const invalid: [bigint, number][] = [
+ [0n, 60],
+ [1500n, 60],
+ [-1000n, 60],
+ [10n ** 19n, 60],
+ [21000n, 0],
+ ]
+
+ for (const [amount, period] of invalid) {
+ expect(await rejects(service.createPeeringInvoice(peer, amount, 'invalid', period))).to.equal(true)
+ }
+
+ expect(this.parameters.peeringInvoiceCreates).to.equal(before)
+ },
+)
+
+When(
+ 'the peering invoice notification is published',
+ async function (this: World>) {
+ const peer = randomPeer()
+ const invoice = await createInvoice(this, peer)
+ const service = this.parameters.peeringService as PaymentsService
+
+ await service.ensurePeeringInvoiceNotification(invoice)
+ await service.ensurePeeringInvoiceNotification(invoice)
+
+ this.parameters.peeringNotificationPeer = peer
+ this.parameters.peeringNotificationInvoice = invoice
+ },
+)
+
+Then(
+ 'the invoice notification is published once with peering tags',
+ function (this: World>) {
+ expect(this.parameters.peeringEventCreates).to.equal(1)
+
+ const event = this.parameters.peeringEvents[0] as any
+ const tags = event.tags as string[][]
+
+ expect(event.kind).to.equal(402)
+ expect(tags).to.deep.include(['p', this.parameters.peeringNotificationPeer])
+ expect(tags).to.deep.include(['t', 'peering'])
+ expect(tags).to.deep.include(['i', this.parameters.peeringNotificationInvoice.id])
+ expect(tags).to.deep.include(['bolt11', this.parameters.peeringNotificationInvoice.bolt11])
+ expect(tags.some(([name]) => name === 'expiration')).to.equal(true)
+ },
+)
diff --git a/test/unit/app/maintenance-worker.spec.ts b/test/unit/app/maintenance-worker.spec.ts
index dc3defe1c..cbd03e1d6 100644
--- a/test/unit/app/maintenance-worker.spec.ts
+++ b/test/unit/app/maintenance-worker.spec.ts
@@ -5,11 +5,12 @@ import chaiAsPromised from 'chai-as-promised'
import Sinon from 'sinon'
import sinonChai from 'sinon-chai'
-import { InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
+import { InvoicePurpose, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
import { Nip05Verification } from '../../../src/@types/nip05'
import { IMaintenanceService, INotificationOutboxService, IPaymentsService } from '../../../src/@types/services'
import { Settings } from '../../../src/@types/settings'
import { applyReverificationOutcome, MaintenanceWorker } from '../../../src/app/maintenance-worker'
+import { PENDING_INVOICE_PAGE_SIZE } from '../../../src/services/payments-service'
import * as metricsTelemetry from '../../../src/telemetry/metrics'
import * as misc from '../../../src/utils/misc'
import * as nip05Utils from '../../../src/utils/nip05'
@@ -84,6 +85,7 @@ describe('MaintenanceWorker', () => {
paymentsService = {
getPendingInvoices: sandbox.stub().resolves([]),
+ getUnconfirmedPeeringInvoices: sandbox.stub().resolves([]),
getInvoiceFromPaymentsProcessor: sandbox.stub(),
updateInvoiceStatus: sandbox.stub().resolves(),
confirmInvoice: sandbox.stub().resolves(),
@@ -489,6 +491,147 @@ describe('MaintenanceWorker', () => {
expect(confirmArg.amountPaid).to.equal(pendingInvoice.amountRequested)
})
+ it('confirms a completed peering invoice that was never settled', async () => {
+ const confirmedAt = new Date()
+ const completedUnconfirmed = {
+ ...pendingInvoice,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ }
+ settingsState.payments = { enabled: true } as any
+ paymentsService.getUnconfirmedPeeringInvoices.resolves([completedUnconfirmed])
+ paymentsService.getPendingInvoices.resolves([])
+ paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+ id: 'inv-1',
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt,
+ })
+
+ await (worker as any).onSchedule()
+
+ expect(paymentsService.getUnconfirmedPeeringInvoices).to.have.been.calledOnceWithExactly(
+ PENDING_INVOICE_PAGE_SIZE,
+ 0,
+ )
+ expect(paymentsService.updateInvoiceStatus).to.have.been.calledOnceWithExactly({
+ id: 'inv-1',
+ status: InvoiceStatus.COMPLETED,
+ })
+ expect(paymentsService.confirmInvoice).to.have.been.calledOnce
+
+ const [confirmArg] = paymentsService.confirmInvoice.firstCall.args
+ expect(confirmArg).to.include({
+ id: 'inv-1',
+ status: InvoiceStatus.COMPLETED,
+ purpose: InvoicePurpose.PEERING,
+ })
+ })
+
+ it('uses the paid amount reported by the processor', async () => {
+ const confirmedAt = new Date()
+ const completedUnconfirmed = {
+ ...pendingInvoice,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ amountRequested: 1000n,
+ }
+ settingsState.payments = { enabled: true } as any
+ paymentsService.getUnconfirmedPeeringInvoices.resolves([completedUnconfirmed])
+ paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+ id: 'inv-1',
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt,
+ amountPaid: 1500n,
+ })
+
+ await (worker as any).onSchedule()
+
+ const [confirmArg] = paymentsService.confirmInvoice.firstCall.args
+ expect(confirmArg.amountPaid).to.equal(1500n)
+ })
+
+ it('retries a failed peering settlement on the next pass', async () => {
+ const confirmedAt = new Date()
+ const completedUnconfirmed = {
+ ...pendingInvoice,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ }
+ settingsState.payments = { enabled: true } as any
+ paymentsService.getUnconfirmedPeeringInvoices.resolves([completedUnconfirmed])
+ paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+ id: 'inv-1',
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt,
+ })
+ paymentsService.confirmInvoice.onFirstCall().rejects(new Error('settlement failed'))
+ paymentsService.confirmInvoice.onSecondCall().resolves()
+
+ await (worker as any).onSchedule()
+ await (worker as any).onSchedule()
+
+ expect(paymentsService.confirmInvoice).to.have.been.calledTwice
+ })
+
+ it('advances the retry cursor past a full page of failing peering invoices', async () => {
+ settingsState.payments = { enabled: true } as any
+ const fullPage = Array.from({ length: PENDING_INVOICE_PAGE_SIZE }, (_, i) => ({
+ ...pendingInvoice,
+ id: `peering-${i}`,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ }))
+ paymentsService.getUnconfirmedPeeringInvoices.resolves(fullPage)
+ paymentsService.getPendingInvoices.resolves([])
+ paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+ id: 'peering-0',
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: new Date(),
+ })
+ paymentsService.confirmInvoice.rejects(new Error('settlement failed'))
+
+ await (worker as any).onSchedule()
+ await (worker as any).onSchedule()
+
+ expect(paymentsService.getUnconfirmedPeeringInvoices.getCall(0).args[1]).to.equal(0)
+ expect(paymentsService.getUnconfirmedPeeringInvoices.getCall(1).args[1]).to.equal(
+ PENDING_INVOICE_PAGE_SIZE,
+ )
+ })
+
+ it('resets the retry cursor after a short page', async () => {
+ settingsState.payments = { enabled: true } as any
+ const fullPage = Array.from({ length: PENDING_INVOICE_PAGE_SIZE }, (_, i) => ({
+ ...pendingInvoice,
+ id: `peering-${i}`,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ }))
+ paymentsService.getUnconfirmedPeeringInvoices.onFirstCall().resolves(fullPage)
+ paymentsService.getUnconfirmedPeeringInvoices.onSecondCall().resolves([fullPage[0]])
+ paymentsService.getUnconfirmedPeeringInvoices.onThirdCall().resolves([])
+ paymentsService.getPendingInvoices.resolves([])
+ paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+ id: 'peering-0',
+ status: InvoiceStatus.PENDING,
+ })
+
+ await (worker as any).onSchedule()
+ await (worker as any).onSchedule()
+ await (worker as any).onSchedule()
+
+ expect(paymentsService.getUnconfirmedPeeringInvoices.getCall(0).args[1]).to.equal(0)
+ expect(paymentsService.getUnconfirmedPeeringInvoices.getCall(1).args[1]).to.equal(
+ PENDING_INVOICE_PAGE_SIZE,
+ )
+ expect(paymentsService.getUnconfirmedPeeringInvoices.getCall(2).args[1]).to.equal(0)
+ })
+
it('continues processing remaining invoices when one throws', async () => {
const secondInvoice = { ...pendingInvoice, id: 'inv-2' }
settingsState.payments = { enabled: true } as any
@@ -581,6 +724,23 @@ describe('MaintenanceWorker', () => {
})
})
+ it('does not expire a completed peering invoice when the processor returns 404', async () => {
+ const completedUnconfirmed = {
+ ...pendingInvoice,
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: null,
+ purpose: InvoicePurpose.PEERING,
+ expiresAt: new Date(Date.now() - 60000),
+ }
+ settingsState.payments = { enabled: true } as any
+ paymentsService.getUnconfirmedPeeringInvoices.resolves([completedUnconfirmed])
+ paymentsService.getInvoiceFromPaymentsProcessor.rejects({ response: { status: 404 } })
+
+ await (worker as any).onSchedule()
+
+ expect(paymentsService.updateInvoiceStatus).not.to.have.been.called
+ })
+
it('keeps an expired pending invoice pending when the processor lookup fails without 404', async () => {
const expiredInvoice = {
...pendingInvoice,
diff --git a/test/unit/controllers/admin/patch-notifications-controller.spec.ts b/test/unit/controllers/admin/patch-notifications-controller.spec.ts
index 9698bcfae..df34f311a 100644
--- a/test/unit/controllers/admin/patch-notifications-controller.spec.ts
+++ b/test/unit/controllers/admin/patch-notifications-controller.spec.ts
@@ -130,6 +130,28 @@ describe('PatchAdminNotificationsController', () => {
expect(response.send.firstCall.args[0].ok).to.equal(true)
})
+ it('accepts the peering notification switches', async () => {
+ const controller = new PatchAdminNotificationsController(outboxRepository as any)
+ await controller.handleRequest(
+ {
+ body: {
+ events: {
+ 'peering.invoice.paid': false,
+ 'peering.access.lapsed': false,
+ },
+ },
+ ip: '127.0.0.1',
+ } as any,
+ response as any,
+ )
+
+ expect(response.status).to.have.been.calledWith(200)
+
+ const saved = (settingsConfig.saveSettings as Sinon.SinonStub).firstCall.args[0]
+ expect(saved.admin.notifications.events['peering.invoice.paid']).to.equal(false)
+ expect(saved.admin.notifications.events['peering.access.lapsed']).to.equal(false)
+ })
+
it('still returns 200 when outbox enqueue fails', async () => {
outboxRepository.enqueue.rejects(new Error('db unavailable'))
const controller = new PatchAdminNotificationsController(outboxRepository as any)
diff --git a/test/unit/controllers/callbacks/lnbits-callback-controller.spec.ts b/test/unit/controllers/callbacks/lnbits-callback-controller.spec.ts
index 62e0d4014..ca6e3b2c5 100644
--- a/test/unit/controllers/callbacks/lnbits-callback-controller.spec.ts
+++ b/test/unit/controllers/callbacks/lnbits-callback-controller.spec.ts
@@ -260,6 +260,28 @@ describe('LNbitsCallbackController', () => {
expect(res.send).to.have.been.calledWith('Invoice is already marked paid')
})
+ it('retries a completed invoice that has no confirmation yet', async () => {
+ const invoiceRepository = {
+ findById: sinon.stub().resolves(makeInvoice({ status: InvoiceStatus.COMPLETED, confirmedAt: null })),
+ }
+ const paymentsService = {
+ getInvoiceFromPaymentsProcessor: sinon
+ .stub()
+ .resolves(makeInvoice({ status: InvoiceStatus.COMPLETED, confirmedAt: new Date() })),
+ updateInvoice: sinon.stub().resolves(),
+ confirmInvoice: sinon.stub().resolves(),
+ sendInvoiceUpdateNotification: sinon.stub().resolves(),
+ }
+ const { controller } = makeController({ paymentsService, invoiceRepository })
+ const res = makeRes()
+
+ await controller.handleRequest(makeReq(), res)
+
+ expect(res.status).to.have.been.calledWith(200)
+ expect(res.send).to.have.been.calledWith('OK')
+ expect(paymentsService.confirmInvoice).to.have.been.calledOnce
+ })
+
it('confirms and notifies when invoice transitions to completed', async () => {
const paymentsService = {
getInvoiceFromPaymentsProcessor: sinon.stub().resolves(makeInvoice({ status: InvoiceStatus.COMPLETED })),
diff --git a/test/unit/repositories/invoice-repository.spec.ts b/test/unit/repositories/invoice-repository.spec.ts
index 281abd08c..ac92b1420 100644
--- a/test/unit/repositories/invoice-repository.spec.ts
+++ b/test/unit/repositories/invoice-repository.spec.ts
@@ -5,7 +5,7 @@ import sinonChai from 'sinon-chai'
import chaiAsPromised from 'chai-as-promised'
import { DatabaseClient } from '../../../src/@types/base'
-import { Invoice, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
+import { Invoice, InvoicePurpose, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
import { IInvoiceRepository } from '../../../src/@types/repositories'
import { InvoiceRepository } from '../../../src/repositories/invoice-repository'
@@ -218,6 +218,57 @@ describe('InvoiceRepository', () => {
})
})
+ describe('.findUnconfirmedPeeringInvoices', () => {
+ function makeClient(results: any[]) {
+ const selectStub = sandbox.stub().resolves(results)
+ const limitStub = sandbox.stub().returns({ select: selectStub })
+ const offsetStub = sandbox.stub().returns({ limit: limitStub })
+ const orderByStub = sandbox.stub().returns({ offset: offsetStub })
+ const whereNullStub = sandbox.stub().returns({ orderBy: orderByStub })
+ const whereStub = sandbox.stub().returns({ whereNull: whereNullStub })
+ return {
+ client: sandbox.stub().returns({ where: whereStub }) as unknown as DatabaseClient,
+ whereStub,
+ whereNullStub,
+ orderByStub,
+ offsetStub,
+ limitStub,
+ }
+ }
+
+ it('returns completed peering invoices with no confirmation', async () => {
+ const peeringRow = {
+ ...dbInvoiceRow,
+ purpose: InvoicePurpose.PEERING,
+ status: InvoiceStatus.COMPLETED,
+ confirmed_at: null,
+ }
+ const { client } = makeClient([peeringRow])
+
+ const result = await repository.findUnconfirmedPeeringInvoices(10, 0, client)
+
+ expect(result).to.have.length(1)
+ expect(result[0].id).to.equal('test-invoice-id')
+ expect(result[0].purpose).to.equal(InvoicePurpose.PEERING)
+ expect(result[0].confirmedAt).to.equal(null)
+ })
+
+ it('filters by peering purpose, completed status and a null confirmation', async () => {
+ const { client, whereStub, whereNullStub, orderByStub, offsetStub, limitStub } = makeClient([])
+
+ await repository.findUnconfirmedPeeringInvoices(25, 5, client)
+
+ expect(whereStub).to.have.been.calledWith({
+ purpose: InvoicePurpose.PEERING,
+ status: InvoiceStatus.COMPLETED,
+ })
+ expect(whereNullStub).to.have.been.calledWith('confirmed_at')
+ expect(orderByStub).to.have.been.calledWith('created_at', 'asc')
+ expect(offsetStub).to.have.been.calledWith(5)
+ expect(limitStub).to.have.been.calledWith(25)
+ })
+ })
+
describe('.updateStatus', () => {
it('returns an object with then, catch, and toString', () => {
const result = repository.updateStatus(testInvoice)
diff --git a/test/unit/repositories/peering-payment-repository.spec.ts b/test/unit/repositories/peering-payment-repository.spec.ts
new file mode 100644
index 000000000..7c13d3a8d
--- /dev/null
+++ b/test/unit/repositories/peering-payment-repository.spec.ts
@@ -0,0 +1,116 @@
+import { expect } from 'chai'
+
+import { PeeringPaymentDirection, PeeringPaymentStatus } from '../../../src/@types/peering'
+import { PeeringPaymentRepository } from '../../../src/repositories/peering-payment-repository'
+import { createRecordingKnex } from './recording-knex'
+
+describe('PeeringPaymentRepository', () => {
+ const pubkey = 'a'.repeat(64)
+ const now = new Date('2026-11-09T12:00:00.000Z')
+
+ const row = {
+ peer_pubkey: Buffer.from(pubkey, 'hex'),
+ direction: PeeringPaymentDirection.SELL,
+ status: PeeringPaymentStatus.ACTIVE,
+ current_invoice_id: 'inv-1',
+ renewal_invoice_id: null,
+ period_start: now,
+ period_end: new Date(now.getTime() + 86_400_000),
+ term_amount_msats: '21000',
+ term_period_seconds: 86_400,
+ last_settled_invoice_id: 'inv-0',
+ last_settled_at: now,
+ last_checked_at: now,
+ last_error: null,
+ created_at: now,
+ updated_at: now,
+ }
+
+ it('maps a stored row to the domain type', async () => {
+ const recording = createRecordingKnex()
+ const repository = new PeeringPaymentRepository(recording.db)
+ recording.respondWith([row])
+
+ const result = await repository.findByPeer(pubkey)
+
+ expect(result).to.include({
+ peerPubkey: pubkey,
+ direction: PeeringPaymentDirection.SELL,
+ status: PeeringPaymentStatus.ACTIVE,
+ currentInvoiceId: 'inv-1',
+ termPeriodSeconds: 86_400,
+ })
+ expect(result?.termAmountMsats).to.equal(21000n)
+ expect(recording.queries[0].sql).to.include('"peering_payments"')
+ expect(recording.queries[0].sql).to.include('"peer_pubkey"')
+ })
+
+ it('creates a missing row without overwriting an existing one', async () => {
+ const recording = createRecordingKnex()
+ const repository = new PeeringPaymentRepository(recording.db)
+ recording.respondWith(0)
+
+ await repository.ensureRow(pubkey, recording.db)
+
+ const sql = recording.queries[0].sql
+ expect(sql).to.include('insert into "peering_payments"')
+ expect(sql).to.include('on conflict')
+ expect(sql).to.include('do nothing')
+ })
+
+ it('locks the peer row for update', async () => {
+ const recording = createRecordingKnex()
+ const repository = new PeeringPaymentRepository(recording.db)
+ recording.respondWith([row])
+
+ await repository.lockByPeer(pubkey, recording.db)
+
+ expect(recording.queries[0].sql).to.include('for update')
+ })
+
+ it('saves the billing state', async () => {
+ const recording = createRecordingKnex()
+ const repository = new PeeringPaymentRepository(recording.db)
+ recording.respondWith(1)
+
+ const saved = await repository.save(
+ {
+ peerPubkey: pubkey,
+ direction: PeeringPaymentDirection.SELL,
+ status: PeeringPaymentStatus.LAPSED,
+ currentInvoiceId: null,
+ renewalInvoiceId: null,
+ periodStart: now,
+ periodEnd: now,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 86_400,
+ lastSettledInvoiceId: null,
+ lastSettledAt: null,
+ lastCheckedAt: now,
+ lastError: null,
+ createdAt: now,
+ updatedAt: now,
+ },
+ recording.db,
+ )
+
+ const sql = recording.queries[0].sql
+ expect(sql).to.include('update "peering_payments"')
+ expect(sql).to.include('"status"')
+ expect(saved.status).to.equal(PeeringPaymentStatus.LAPSED)
+ })
+
+ it('finds periods due for renewal and periods that ended', async () => {
+ const recording = createRecordingKnex()
+ const repository = new PeeringPaymentRepository(recording.db)
+ recording.respondWith([row], [row])
+
+ await repository.findDueForRenewal(now, 10, 0)
+ await repository.findPeriodEnded(now, 10, 0)
+
+ expect(recording.queries[0].sql).to.include('"status" in')
+ expect(recording.queries[0].sql).to.include('"period_end" <=')
+ expect(recording.queries[1].sql).to.include('"status" in')
+ expect(recording.queries[1].sql).to.include('"period_end" <=')
+ })
+})
diff --git a/test/unit/services/payments-service.spec.ts b/test/unit/services/payments-service.spec.ts
index c8a5c8135..576fc653a 100644
--- a/test/unit/services/payments-service.spec.ts
+++ b/test/unit/services/payments-service.spec.ts
@@ -7,9 +7,9 @@ chai.use(sinonChai)
chai.use(chaiAsPromised)
import * as eventUtils from '../../../src/utils/event'
-import { Invoice, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
+import { Invoice, InvoicePurpose, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice'
import { OperatorNotificationEventType } from '../../../src/@types/operator-notifications'
-import { PaymentsService } from '../../../src/services/payments-service'
+import { PENDING_INVOICE_PAGE_SIZE, PaymentsService } from '../../../src/services/payments-service'
const { expect } = chai
@@ -63,10 +63,13 @@ describe('PaymentsService', () => {
}
invoiceRepository = {
+ findById: sandbox.stub().resolves(undefined),
findPendingInvoices: sandbox.stub(),
+ findUnconfirmedPeeringInvoices: sandbox.stub().resolves([]),
upsert: sandbox.stub().resolves(),
updateStatus: sandbox.stub(),
confirmInvoice: sandbox.stub().resolves(),
+ updatePeriod: sandbox.stub().resolves(),
}
eventRepository = {
@@ -138,6 +141,38 @@ describe('PaymentsService', () => {
})
+ describe('getUnconfirmedPeeringInvoices', () => {
+ it('returns completed peering invoices that never settled', async () => {
+ const invoices = [stubInvoice({ confirmedAt: null })]
+ invoiceRepository.findUnconfirmedPeeringInvoices.resolves(invoices)
+
+ const result = await service.getUnconfirmedPeeringInvoices()
+
+ expect(result).to.deep.equal(invoices)
+ expect(invoiceRepository.findUnconfirmedPeeringInvoices).to.have.been.calledOnceWithExactly(
+ PENDING_INVOICE_PAGE_SIZE,
+ 0,
+ )
+ })
+
+ it('passes the retry offset through', async () => {
+ invoiceRepository.findUnconfirmedPeeringInvoices.resolves([stubInvoice()])
+
+ await service.getUnconfirmedPeeringInvoices(PENDING_INVOICE_PAGE_SIZE, 20)
+
+ expect(invoiceRepository.findUnconfirmedPeeringInvoices).to.have.been.calledOnceWithExactly(
+ PENDING_INVOICE_PAGE_SIZE,
+ 20,
+ )
+ })
+
+ it('re-throws repository errors', async () => {
+ invoiceRepository.findUnconfirmedPeeringInvoices.rejects(new Error('db error'))
+
+ await expect(service.getUnconfirmedPeeringInvoices()).to.be.rejectedWith('db error')
+ })
+ })
+
describe('getInvoiceFromPaymentsProcessor', () => {
it('passes a string invoice ID directly to the payments processor', async () => {
const partial = { id: 'inv', status: InvoiceStatus.PENDING }
@@ -702,12 +737,12 @@ describe('PaymentsService', () => {
expect(userRepository.admitUser).to.have.been.calledOnce
})
- it('does not read the invoice or subscriptions while no plan is configured', async () => {
+ it('reads the invoice for its purpose even when no plan is configured', async () => {
settings.returns(plansSettings([]))
await service.confirmInvoice(makeCompletedInvoice({ amountPaid: 2000n }))
- expect(invoiceRepository.findById).not.to.have.been.called
+ expect(invoiceRepository.findById).to.have.been.calledOnce
expect(userSubscriptionRepository.findByPubkey).not.to.have.been.called
expect(userRepository.admitUser).to.have.been.calledOnce
})
@@ -847,4 +882,429 @@ describe('PaymentsService', () => {
expect(eventRepository.create).not.to.have.been.called
})
})
+
+ describe('peering billing', () => {
+ let peeringPaymentRepository: any
+ let peeringService: PaymentsService
+
+ const peeringRow = (overrides: Record = {}) => ({
+ peerPubkey: 'pubkey1234',
+ direction: 'sell',
+ status: 'unpaid',
+ currentInvoiceId: null,
+ renewalInvoiceId: null,
+ periodStart: null,
+ periodEnd: null,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ lastSettledInvoiceId: null,
+ lastSettledAt: null,
+ lastCheckedAt: null,
+ lastError: null,
+ createdAt: new Date(),
+ updatedAt: new Date(),
+ ...overrides,
+ })
+
+ const makeCompletedInvoice = (overrides: Partial = {}): Invoice =>
+ stubInvoice({
+ status: InvoiceStatus.COMPLETED,
+ confirmedAt: new Date('2026-11-09T12:00:00.000Z'),
+ amountPaid: 21000n,
+ unit: InvoiceUnit.MSATS,
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ ...overrides,
+ })
+
+ beforeEach(() => {
+ eventRepository.findByFilters = sandbox.stub().resolves([])
+ peeringPaymentRepository = {
+ lockByPeer: sandbox.stub().resolves(peeringRow()),
+ save: sandbox.stub().callsFake(async (value: unknown) => value),
+ }
+
+ peeringService = new PaymentsService(
+ dbClient,
+ paymentsProcessor,
+ userRepository,
+ invoiceRepository,
+ eventRepository,
+ settings,
+ notificationOutboxRepository,
+ undefined,
+ peeringPaymentRepository,
+ )
+ })
+
+ it('creates a peering invoice without upserting a user', async () => {
+ settings.returns({ payments: { invoiceExpirySeconds: 3600 } })
+ paymentsProcessor.createInvoice.resolves({
+ id: 'peering-inv',
+ bolt11: 'lnbc...',
+ amountRequested: 21000n,
+ description: 'peering',
+ unit: InvoiceUnit.MSATS,
+ status: InvoiceStatus.PENDING,
+ expiresAt: new Date('2099-01-01'),
+ createdAt: new Date(),
+ })
+
+ const result = await peeringService.createPeeringInvoice('peer', 21000n, 'peering', 2592000)
+
+ expect(userRepository.upsert).not.to.have.been.called
+ expect(paymentsProcessor.createInvoice).to.have.been.calledWith({
+ amount: 21000n,
+ description: 'peering',
+ requestId: 'peer',
+ })
+ expect(invoiceRepository.upsert.firstCall.args[0]).to.include({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ planId: null,
+ })
+ expect(notificationOutboxRepository.enqueue).to.have.been.calledWith(
+ OperatorNotificationEventType.PEERING_INVOICE_CREATED,
+ Sinon.match({ invoiceId: 'peering-inv' }),
+ mockTrx,
+ )
+ expect(result.id).to.equal('peering-inv')
+ })
+
+ it('rejects a zero-amount peering invoice before calling the processor', async () => {
+ await expect(peeringService.createPeeringInvoice('peer', 0n, 'peering', 2592000)).to.be.rejectedWith(
+ 'positive whole number of satoshis',
+ )
+ expect(paymentsProcessor.createInvoice).not.to.have.been.called
+ })
+
+ it('rejects a sub-satoshi peering invoice before calling the processor', async () => {
+ await expect(peeringService.createPeeringInvoice('peer', 1500n, 'peering', 2592000)).to.be.rejectedWith(
+ 'positive whole number of satoshis',
+ )
+ expect(paymentsProcessor.createInvoice).not.to.have.been.called
+ })
+
+ it('rejects a peering invoice too large for the processor conversion', async () => {
+ await expect(
+ peeringService.createPeeringInvoice('peer', 10n ** 19n, 'peering', 2592000),
+ ).to.be.rejectedWith('too large for the payment processors')
+ expect(paymentsProcessor.createInvoice).not.to.have.been.called
+ })
+
+ it('rejects a negative peering invoice before calling the processor', async () => {
+ await expect(peeringService.createPeeringInvoice('peer', -1000n, 'peering', 2592000)).to.be.rejectedWith(
+ 'positive whole number of satoshis',
+ )
+ expect(paymentsProcessor.createInvoice).not.to.have.been.called
+ })
+
+ it('rejects a peering invoice with no period before calling the processor', async () => {
+ await expect(peeringService.createPeeringInvoice('peer', 21000n, 'peering', 0)).to.be.rejectedWith(
+ 'positive number of seconds',
+ )
+ expect(paymentsProcessor.createInvoice).not.to.have.been.called
+ })
+
+ it('applies a paid period and clears the open invoice pointers', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ unit: InvoiceUnit.MSATS,
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow({ currentInvoiceId: 'invoice-id' }))
+
+ await peeringService.confirmInvoice(makeCompletedInvoice())
+
+ expect(peeringPaymentRepository.save).to.have.been.calledWith(
+ Sinon.match({
+ status: 'active',
+ currentInvoiceId: null,
+ renewalInvoiceId: null,
+ lastSettledInvoiceId: 'invoice-id',
+ }),
+ mockTrx,
+ )
+ expect(invoiceRepository.updatePeriod).to.have.been.calledWith(
+ 'invoice-id',
+ Sinon.match.instanceOf(Date),
+ Sinon.match.instanceOf(Date),
+ mockTrx,
+ )
+ expect(userRepository.admitUser).not.to.have.been.called
+ expect(notificationOutboxRepository.enqueue).to.have.been.calledWith(
+ OperatorNotificationEventType.PEERING_INVOICE_PAID,
+ Sinon.match({ invoiceId: 'invoice-id' }),
+ mockTrx,
+ )
+ expect(mockTrx.commit).to.have.been.calledOnce
+ })
+
+ it('grants exactly one term for an overpayment and keeps the full amount on the receipt', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ unit: InvoiceUnit.MSATS,
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow({ currentInvoiceId: 'invoice-id' }))
+
+ await peeringService.confirmInvoice(makeCompletedInvoice({ amountPaid: 42000n }))
+
+ expect(invoiceRepository.confirmInvoice).to.have.been.calledWith(
+ 'invoice-id',
+ 42000n,
+ Sinon.match.instanceOf(Date),
+ mockTrx,
+ )
+ expect(peeringPaymentRepository.save).to.have.been.calledWith(
+ Sinon.match({
+ status: 'active',
+ lastSettledInvoiceId: 'invoice-id',
+ }),
+ mockTrx,
+ )
+ expect(invoiceRepository.updatePeriod).to.have.been.calledOnce
+
+ const [, periodStart, periodEnd] = invoiceRepository.updatePeriod.firstCall.args
+ expect((periodEnd as Date).getTime() - (periodStart as Date).getTime()).to.equal(2592000 * 1000)
+ })
+
+ it('does not grant a second period for a duplicate callback', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow({ lastSettledInvoiceId: 'invoice-id' }))
+
+ await peeringService.confirmInvoice(makeCompletedInvoice())
+
+ expect(peeringPaymentRepository.save).not.to.have.been.called
+ expect(invoiceRepository.updatePeriod).not.to.have.been.called
+ expect(notificationOutboxRepository.enqueue).not.to.have.been.called
+ expect(mockTrx.commit).to.have.been.calledOnce
+ })
+
+ it('settles the same invoice only once across two callbacks', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ let stored = stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ unit: InvoiceUnit.MSATS,
+ })
+ invoiceRepository.findById.callsFake(async () => stored)
+ invoiceRepository.confirmInvoice.callsFake(async () => {
+ stored = { ...stored, confirmedAt: new Date() }
+ })
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow())
+
+ await peeringService.confirmInvoice(makeCompletedInvoice())
+ await peeringService.confirmInvoice(makeCompletedInvoice())
+
+ expect(peeringPaymentRepository.lockByPeer).to.have.been.calledOnce
+ expect(peeringPaymentRepository.save).to.have.been.calledOnce
+ expect(invoiceRepository.updatePeriod).to.have.been.calledOnce
+ })
+
+ it('records an underpayment without activating access', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ }),
+ )
+
+ await peeringService.confirmInvoice(makeCompletedInvoice({ amountPaid: 1000n }))
+
+ expect(peeringPaymentRepository.save).to.have.been.calledWith(
+ Sinon.match({ lastError: Sinon.match('underpaid') }),
+ mockTrx,
+ )
+ expect(invoiceRepository.updatePeriod).not.to.have.been.called
+ expect(notificationOutboxRepository.enqueue).to.have.been.calledWith(
+ OperatorNotificationEventType.PEERING_INVOICE_FAILED,
+ Sinon.match({ reason: 'underpaid' }),
+ mockTrx,
+ )
+ })
+
+ it('publishes peering invoice events with the peering and invoice-id tags', async () => {
+ settings.returns({ info: { relay_url: 'wss://relay.example.com' } })
+ const invoice = stubInvoice({
+ id: 'peering-inv',
+ purpose: InvoicePurpose.PEERING,
+ termPeriodSeconds: 2592000,
+ })
+
+ await peeringService.sendPeeringInvoiceNotification(invoice)
+
+ const [unsignedEvent] = (eventUtils.identifyEvent as Sinon.SinonStub).firstCall.args
+ expect(unsignedEvent.tags).to.deep.include.members([
+ ['p', invoice.pubkey],
+ ['t', 'peering'],
+ ['bolt11', invoice.bolt11],
+ ['i', invoice.id],
+ ])
+ expect(unsignedEvent.content).to.include('Paid peering invoice')
+ })
+
+ it('does not republish an invoice that already has a stored notification', async () => {
+ settings.returns({ info: { relay_url: 'wss://relay.example.com' } })
+ const invoice = stubInvoice({ id: 'peering-inv', purpose: InvoicePurpose.PEERING })
+ eventRepository.findByFilters.resolves([{ event_id: Buffer.alloc(32) }])
+
+ await peeringService.ensurePeeringInvoiceNotification(invoice)
+
+ expect(eventUtils.identifyEvent).not.to.have.been.called
+ expect(eventRepository.findByFilters).to.have.been.calledWith(
+ [Sinon.match({ kinds: [402], '#i': ['peering-inv'], limit: 1 })],
+ { excludeDeleted: true },
+ )
+ })
+
+ it('publishes when no stored notification exists', async () => {
+ settings.returns({ info: { relay_url: 'wss://relay.example.com' } })
+ const invoice = stubInvoice({ id: 'peering-inv', purpose: InvoicePurpose.PEERING })
+
+ await peeringService.ensurePeeringInvoiceNotification(invoice)
+
+ expect(eventUtils.identifyEvent).to.have.been.calledOnce
+ })
+
+ it('does not grant another period when an older confirmed invoice callback is redelivered', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ id: 'older-invoice',
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ confirmedAt: new Date('2026-10-01T00:00:00.000Z'),
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow({ lastSettledInvoiceId: 'newer-invoice' }))
+
+ await peeringService.confirmInvoice(makeCompletedInvoice({ id: 'older-invoice' }))
+
+ expect(peeringPaymentRepository.lockByPeer).not.to.have.been.called
+ expect(peeringPaymentRepository.save).not.to.have.been.called
+ expect(invoiceRepository.updatePeriod).not.to.have.been.called
+ expect(notificationOutboxRepository.enqueue).not.to.have.been.called
+ expect(mockTrx.commit).to.have.been.calledOnce
+ })
+
+ it('starts a first period at settlement time, not the receipt timestamp', async () => {
+ const clock = Sinon.useFakeTimers({ now: new Date('2026-11-14T12:00:00.000Z') })
+ const receiptTime = new Date('2026-11-14T11:58:00.000Z')
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 60,
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(peeringRow({ periodEnd: null }))
+
+ try {
+ await peeringService.confirmInvoice(makeCompletedInvoice({ confirmedAt: receiptTime }))
+
+ const [, periodStart, periodEnd] = invoiceRepository.updatePeriod.firstCall.args
+ expect(periodStart).to.deep.equal(new Date('2026-11-14T12:00:00.000Z'))
+ expect((periodEnd as Date).getTime()).to.equal(
+ (periodStart as Date).getTime() + 60 * 1000,
+ )
+ } finally {
+ clock.restore()
+ }
+ })
+
+ it('extends a running period from its existing end', async () => {
+ const existingPeriodEnd = new Date(Date.now() + 5 * 86_400_000)
+ const confirmedAt = new Date(Date.now() - 86_400_000)
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ }),
+ )
+ peeringPaymentRepository.lockByPeer.resolves(
+ peeringRow({
+ status: 'active',
+ currentInvoiceId: 'invoice-id',
+ periodEnd: existingPeriodEnd,
+ }),
+ )
+
+ await peeringService.confirmInvoice(makeCompletedInvoice({ confirmedAt }))
+
+ const [, periodStart, periodEnd] = invoiceRepository.updatePeriod.firstCall.args
+ expect(periodStart).to.equal(existingPeriodEnd)
+ expect((periodEnd as Date).getTime()).to.equal(existingPeriodEnd.getTime() + 2592000 * 1000)
+ })
+
+ it('rolls back when saving the peer billing row fails', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ }),
+ )
+ peeringPaymentRepository.save.rejects(new Error('peer save failed'))
+
+ await expect(peeringService.confirmInvoice(makeCompletedInvoice())).to.be.rejectedWith('peer save failed')
+
+ expect(mockTrx.rollback).to.have.been.calledOnce
+ expect(invoiceRepository.updatePeriod).not.to.have.been.called
+ })
+
+ it('rolls back when updating the invoice period fails', async () => {
+ settings.returns({ payments: { enabled: true, feeSchedules: { admission: [] } } })
+ invoiceRepository.findById.resolves(
+ stubInvoice({
+ purpose: InvoicePurpose.PEERING,
+ termAmountMsats: 21000n,
+ termPeriodSeconds: 2592000,
+ }),
+ )
+ invoiceRepository.updatePeriod.rejects(new Error('period update failed'))
+
+ await expect(peeringService.confirmInvoice(makeCompletedInvoice())).to.be.rejectedWith('period update failed')
+
+ expect(mockTrx.rollback).to.have.been.calledOnce
+ })
+
+ it('rolls back a peering invoice when the processor fails', async () => {
+ settings.returns({ payments: { invoiceExpirySeconds: 3600 } })
+ paymentsProcessor.createInvoice.rejects(new Error('processor failed'))
+
+ await expect(peeringService.createPeeringInvoice('peer', 21000n, 'peering', 2592000)).to.be.rejectedWith(
+ 'processor failed',
+ )
+
+ expect(mockTrx.rollback).to.have.been.calledOnce
+ expect(mockTrx.commit).not.to.have.been.called
+ })
+ })
+
})
diff --git a/test/unit/utils/peering-billing.spec.ts b/test/unit/utils/peering-billing.spec.ts
new file mode 100644
index 000000000..233e7d300
--- /dev/null
+++ b/test/unit/utils/peering-billing.spec.ts
@@ -0,0 +1,84 @@
+import { expect } from 'chai'
+
+import { Settings } from '../../../src/@types/settings'
+import {
+ InvalidPeeringTermError,
+ isPeeringInvoiceRequest,
+ resolvePaidPeeringTerm,
+} from '../../../src/utils/peering-billing'
+
+const withTerms = (terms: unknown[]): Settings =>
+ ({
+ peering: {
+ enabled: true,
+ terms,
+ },
+ }) as Settings
+
+describe('resolvePaidPeeringTerm', () => {
+ it('returns undefined when there are no terms or only zero-amount terms', () => {
+ expect(resolvePaidPeeringTerm(withTerms([]))).to.be.undefined
+ expect(resolvePaidPeeringTerm(withTerms([{ amount: 0, unit: 'msats', period: 2592000 }]))).to.be.undefined
+ })
+
+ it('normalizes a valid msats term', () => {
+ expect(resolvePaidPeeringTerm(withTerms([{ amount: 21000, unit: 'msats', period: 2592000 }]))).to.deep.equal({
+ amountMsats: 21000n,
+ periodSeconds: 2592000,
+ })
+ })
+
+ it('normalizes sats and btc terms', () => {
+ expect(resolvePaidPeeringTerm(withTerms([{ amount: 21, unit: 'sats', period: 60 }]))).to.deep.equal({
+ amountMsats: 21000n,
+ periodSeconds: 60,
+ })
+ expect(resolvePaidPeeringTerm(withTerms([{ amount: 1, unit: 'btc', period: 60 }]))).to.deep.equal({
+ amountMsats: 100_000_000_000n,
+ periodSeconds: 60,
+ })
+ })
+
+ it('fails closed for malformed amounts instead of granting free peering', () => {
+ for (const amount of ['21000', 21.5, -1, Number.NaN, Number.POSITIVE_INFINITY]) {
+ expect(() => resolvePaidPeeringTerm(withTerms([{ amount, unit: 'msats', period: 60 }]))).to.throw(
+ InvalidPeeringTermError,
+ )
+ }
+ })
+
+ it('rejects an unsupported unit', () => {
+ expect(() => resolvePaidPeeringTerm(withTerms([{ amount: 1, unit: 'usd', period: 60 }]))).to.throw(
+ InvalidPeeringTermError,
+ )
+ })
+
+ it('rejects a sub-satoshi msat term', () => {
+ expect(() => resolvePaidPeeringTerm(withTerms([{ amount: 1500, unit: 'msats', period: 60 }]))).to.throw(
+ InvalidPeeringTermError,
+ )
+ })
+
+ it('rejects an amount whose satoshi value exceeds Number.MAX_SAFE_INTEGER', () => {
+ // 100,000,000 BTC = 10^16 sats, above Number.MAX_SAFE_INTEGER. The processor
+ // adapters convert msats to sats with Number(), so this must fail closed.
+ expect(() =>
+ resolvePaidPeeringTerm(withTerms([{ amount: 100_000_000, unit: 'btc', period: 60 }])),
+ ).to.throw(InvalidPeeringTermError)
+ })
+
+ it('rejects an invalid period', () => {
+ expect(() => resolvePaidPeeringTerm(withTerms([{ amount: 1000, unit: 'msats', period: 0 }]))).to.throw(
+ InvalidPeeringTermError,
+ )
+ })
+
+ it('exempts only requests where every filter is an invoice filter', () => {
+ const peer = 'a'.repeat(64)
+ const invoice = { kinds: [402], '#p': [peer], '#t': ['peering'] }
+
+ expect(isPeeringInvoiceRequest([invoice], peer)).to.be.true
+ expect(isPeeringInvoiceRequest([invoice, { kinds: [1] }], peer)).to.be.false
+ expect(isPeeringInvoiceRequest([], peer)).to.be.false
+ })
+})