Repository navigation
Expand file tree
/
Copy pathsnapshot.js
More file actions
106 lines (87 loc) · 3.74 KB
/
Copy pathsnapshot.js
File metadata and controls
106 lines (87 loc) · 3.74 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
const config = require('../config');
async function getLatestSnapshot(client, accountId, upToSequenceNo) {
const { rows } = await client.query(
`select sequence_no, balance_cents from account_snapshots
where account_id = $1 and sequence_no <= $2
order by sequence_no desc
limit 1`,
[accountId, upToSequenceNo]
);
if (rows.length === 0) {
return null;
}
return { sequenceNo: Number(rows[0].sequence_no), balanceCents: Number(rows[0].balance_cents) };
}
// Re-derives an account's true balance up to a given sequence number, independently of any snapshots, and returns the result.
async function sumEntriesInRange(client, accountId, fromSequenceNoExclusive, toSequenceNoInclusive) {
const { rows } = await client.query(
`select coalesce(sum(case when direction = 'debit' then -amount_cents else amount_cents end), 0)::bigint as delta
from event_accounts ea
join events e on e.id = ea.event_id
where ea.account_id = $1 and e.sequence_no > $2 and e.sequence_no <= $3`,
[accountId, fromSequenceNoExclusive, toSequenceNoInclusive]
);
return Number(rows[0].delta);
}
/**
* Snapshots when THIS account's own event count since its last snapshot
* crosses the interval - not the raw gap in the global `events.sequence_no`.
* That global sequence is shared by every account in the system, so a
* gap-based trigger would fire on every single touch for a lightly-used
* account sitting in an otherwise busy ledger.
*
* Runs inside the caller's transaction: if the transfer that triggered this
* rolls back, the snapshot insert rolls back with it - never an orphan.
*/
async function maybeCreateSnapshot(client, accountId, currentSequenceNo) {
const latest = await getLatestSnapshot(client, accountId, currentSequenceNo);
const lastSnapshotSeq = latest ? latest.sequenceNo : 0;
const { rows } = await client.query(
`select
count(*)::int as event_count,
coalesce(sum(case when direction = 'debit' then -amount_cents else amount_cents end), 0)::bigint as delta
from event_accounts ea
join events e on e.id = ea.event_id
where ea.account_id = $1 and e.sequence_no > $2 and e.sequence_no <= $3`,
[accountId, lastSnapshotSeq, currentSequenceNo]
);
const eventCount = Number(rows[0].event_count);
if (eventCount < config.snapshotIntervalEvents) {
return null;
}
const balanceCents = (latest ? latest.balanceCents : 0) + Number(rows[0].delta);
await client.query(
`insert into account_snapshots (account_id, sequence_no, balance_cents)
values ($1, $2, $3)
on conflict (account_id, sequence_no) do nothing`,
[accountId, currentSequenceNo, balanceCents]
);
return { sequenceNo: currentSequenceNo, balanceCents };
}
/**
* Re-derives an account's true balance up to its own latest snapshot's
* checkpoint, independently of the snapshot, and reports whether they agree.
* Used only by reconciliation - never by the fast replay path, which is the
* whole point (see reconciliation.js).
*/
async function verifyLatestSnapshot(client, accountId) {
const { rows } = await client.query(
`select sequence_no, balance_cents from account_snapshots
where account_id = $1
order by sequence_no desc
limit 1`,
[accountId]
);
if (rows.length === 0) {
return null;
}
const snapshot = { sequenceNo: Number(rows[0].sequence_no), balanceCents: Number(rows[0].balance_cents) };
const trueBalanceCents = await sumEntriesInRange(client, accountId, 0, snapshot.sequenceNo);
return {
sequenceNo: snapshot.sequenceNo,
balanceCents: snapshot.balanceCents,
trueBalanceCents,
matches: trueBalanceCents === snapshot.balanceCents,
};
}
module.exports = { getLatestSnapshot, sumEntriesInRange, maybeCreateSnapshot, verifyLatestSnapshot };