ledger-repo.js
7.32 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
import { randomUUID } from 'node:crypto'
const LEDGER = 'billing_ledger'
export function createLedgerRepo(db, runner = db) {
const query = runner
const transactionBound = query !== db
async function inTransaction(work) {
if (transactionBound) return work(query)
return db.transaction(work)
}
async function getBalance(user_id, trx = query) {
const row = await trx(LEDGER).where({ user_id }).sum({ balance: 'amount' }).first()
return Number(row.balance)
}
async function applyDelta({ user_id, delta, require_sufficient, ledger_row }) {
assertIdempotencyKey(ledger_row.idempotency_key)
return inTransaction(async trx => {
const user = await trx('users').where({ id: user_id }).select('id').forUpdate().first()
if (user === undefined) throw new Error(`Cannot write billing_ledger: user_id=${user_id} does not exist`)
const balance_before = await getBalance(user_id, trx)
const balance_after = balance_before + delta
if (require_sufficient && balance_after < 0) {
return { kind: 'insufficient_balance', balance_before }
}
const id = randomUUID()
const row = { id, user_id, amount: delta, balance_after, created_at: new Date(), ...ledger_row }
try {
await trx(LEDGER).insert(row)
return { kind: 'applied', ledger_id: id, balance_after }
} catch (error) {
if (error.code !== 'ER_DUP_ENTRY') throw error
const existing = await trx(LEDGER).where({ idempotency_key: ledger_row.idempotency_key }).first()
assertSameLedger(existing, row, ledger_row.idempotency_key)
return { kind: 'replayed', ledger_id: existing.id, balance_after: Number(existing.balance_after) }
}
})
}
async function reverseDelta({ user_id, total_points, ledger_row }) {
assertIdempotencyKey(ledger_row.idempotency_key)
return inTransaction(async trx => {
const user = await trx('users').where({ id: user_id }).select('id').forUpdate().first()
if (user === undefined) throw new Error(`Cannot write billing_ledger: user_id=${user_id} does not exist`)
const balance_before = await getBalance(user_id, trx)
const points_reversed = Math.min(balance_before, total_points)
const reversal_shortfall_points = total_points - points_reversed
if (points_reversed === 0) {
return {
kind: 'not_applied',
ledger_id: null,
balance_after: balance_before,
points_reversed,
reversal_shortfall_points,
}
}
const id = randomUUID()
const row = {
id,
user_id,
amount: -points_reversed,
balance_after: balance_before - points_reversed,
created_at: new Date(),
...ledger_row,
}
try {
await trx(LEDGER).insert(row)
return { kind: 'applied', ledger_id: id, balance_after: row.balance_after, points_reversed, reversal_shortfall_points }
} catch (error) {
if (error.code !== 'ER_DUP_ENTRY') throw error
const existing = await trx(LEDGER).where({ idempotency_key: ledger_row.idempotency_key }).first()
assertSameLedger(existing, row, ledger_row.idempotency_key)
return {
kind: 'replayed',
ledger_id: existing.id,
balance_after: Number(existing.balance_after),
points_reversed: -Number(existing.amount),
reversal_shortfall_points: total_points + Number(existing.amount),
}
}
})
}
async function findLedgerByIdempotencyKey(idempotency_key) {
if (idempotency_key === null) return null
const row = await query(LEDGER).where({ idempotency_key }).first()
if (row === undefined) return null
return ledgerFromRow(row)
}
async function listLedger(user_id, filter) {
let rows = query(LEDGER).where({ user_id })
if (filter.type !== undefined) rows = rows.where({ type: filter.type })
if (filter.since !== undefined) rows = rows.where('created_at', '>=', filter.since)
if (filter.until !== undefined) rows = rows.where('created_at', '<=', filter.until)
const limit = filter.limit === undefined ? 20 : filter.limit
const result = await rows.orderBy('created_at', 'desc').orderBy('id', 'desc').limit(limit + 1)
const items = result.slice(0, limit).map(ledgerFromRow)
const last = items.at(-1)
return {
items,
next_cursor: result.length > limit ? `${last.created_at.toISOString()}|${last.id}` : null,
}
}
return { getBalance, applyDelta, reverseDelta, findLedgerByIdempotencyKey, listLedger }
}
function assertSameLedger(existing, row, idempotency_key) {
if (existing === undefined) {
throw new Error(`billing_ledger idempotency conflict: key=${idempotency_key}, field=existing_row`)
}
const existingLedger = ledgerFromRow(existing)
for (const field of ['user_id', 'amount', 'type', 'source', 'reason', 'ref_type', 'ref_id', 'credit_kind', 'note', 'failure_code']) {
if (existingLedger[field] !== row[field]) {
throw new Error(`billing_ledger idempotency conflict: key=${idempotency_key}, field=${field}`)
}
}
}
function assertIdempotencyKey(idempotency_key) {
if (typeof idempotency_key !== 'string' || idempotency_key === '') {
throw new Error('billing_ledger requires a stable idempotency_key')
}
}
function ledgerFromRow(row) {
const id = requiredString(row.id, 'billing_ledger.id')
const type = requiredString(row.type, 'billing_ledger.type', id)
const amount = requiredInteger(row.amount, 'billing_ledger.amount', id)
const credit_kind = nullableString(row.credit_kind, 'billing_ledger.credit_kind', id)
switch (type) {
case 'grant':
if (amount <= 0 || credit_kind === null) throw new Error(`Invalid billing_ledger grant, id=${id}`)
break
case 'deduct':
if (amount >= 0 || credit_kind !== null) throw new Error(`Invalid billing_ledger deduct, id=${id}`)
break
default:
throw new Error(`Unhandled billing_ledger.type: ${type}, id: ${id}`)
}
return {
id,
user_id: requiredString(row.user_id, 'billing_ledger.user_id', id),
type,
amount,
source: requiredString(row.source, 'billing_ledger.source', id),
credit_kind,
reason: requiredString(row.reason, 'billing_ledger.reason', id),
ref_type: nullableString(row.ref_type, 'billing_ledger.ref_type', id),
ref_id: nullableString(row.ref_id, 'billing_ledger.ref_id', id),
note: nullableString(row.note, 'billing_ledger.note', id),
failure_code: nullableString(row.failure_code, 'billing_ledger.failure_code', id),
balance_after: requiredInteger(row.balance_after, 'billing_ledger.balance_after', id),
idempotency_key: nullableString(row.idempotency_key, 'billing_ledger.idempotency_key', id),
created_at: requiredDate(row.created_at, 'billing_ledger.created_at', id),
}
}
function requiredString(value, field, id = 'unknown') {
if (typeof value !== 'string' || value === '') throw new Error(`Invalid ${field}, id=${id}`)
return value
}
function nullableString(value, field, id) {
if (value === null) return null
return requiredString(value, field, id)
}
function requiredInteger(value, field, id) {
const number = Number(value)
if (!Number.isSafeInteger(number)) throw new Error(`Invalid ${field}, id=${id}`)
return number
}
function requiredDate(value, field, id) {
const date = new Date(value)
if (Number.isNaN(date.getTime())) throw new Error(`Invalid ${field}, id=${id}`)
return date
}