diff --git a/platforms/ecurrency/api/src/scripts/sync-ledger.ts b/platforms/ecurrency/api/src/scripts/sync-ledger.ts index 57a85422e..e1a6fdebe 100644 --- a/platforms/ecurrency/api/src/scripts/sync-ledger.ts +++ b/platforms/ecurrency/api/src/scripts/sync-ledger.ts @@ -5,14 +5,16 @@ import { config } from "dotenv"; config({ path: path.resolve(__dirname, "../../../../../.env") }); import { AppDataSource } from "../database/data-source"; +import { Currency } from "../database/entities/Currency"; import { Group } from "../database/entities/Group"; import { AccountType, Ledger } from "../database/entities/Ledger"; import { User } from "../database/entities/User"; import { adapter } from "../web3adapter"; // Checks that every eCurrency ledger row is in the eVault of each party to it -// (sender and receiver, the same rule the live subscriber uses) and, with -// --apply, writes whatever is missing and re-checks it. +// (sender and receiver, the same rule the live subscriber uses; a mint, burn +// or "Account initialized" has none, so it goes to its holder's vault) and, +// with --apply, writes whatever is missing and re-checks it. // // The live sync is fire-and-forget: a row is pushed 3s after insert and again // after its hash is saved, failures are only logged, and a restart inside that @@ -21,18 +23,30 @@ import { adapter } from "../web3adapter"; // actually returns over GraphQL (not the mapping DB, which can claim a sync // that never landed). // +// Each currency's own record must exist too (in its creator's eVault, as the +// adapter writes it), or the ledger entries point at an internal id nothing +// else can resolve. A copy that carries the wrong currency id is outdated. +// +// It also syncs each holder's Account records (one per holder and currency, +// the shape scripts/backfill-accounts.ts writes). The live adapter never +// writes or updates these, so they go missing for new holders and their +// balance goes stale after every transaction. +// // Usage (from platforms/ecurrency/api, with the production .env): // pnpm sync:ledger check only, writes nothing // pnpm sync:ledger --apply write missing/outdated copies, then verify // pnpm sync:ledger --only=@ limit to one eVault // pnpm sync:ledger --settle=300 ignore rows newer than 300s (default 120) // -// Exit code 0 means every row is verified present in every party's eVault. +// Exit code 0 means every row is verified present in every party's eVault and +// every holder's Account record exists with the current balance. // One vault failing never stops the others. eNames the Registry has no eVault // for (deleted or never provisioned) are listed as skipped, not as failures, // since there is nowhere to write to. const LEDGER_ONTOLOGY = "550e8400-e29b-41d4-a716-446655440006"; +const ACCOUNT_ONTOLOGY = "6fda64db-fd14-4fa2-bd38-77d2e5e6136d"; +const CURRENCY_ONTOLOGY = "550e8400-e29b-41d4-a716-446655440008"; const REGISTRY_URL = process.env.PUBLIC_REGISTRY_URL ?? "https://registry.w3ds.metastate.foundation"; // The platform the live adapter certifies as (see web3adapter/watchers/subscriber.ts). @@ -157,7 +171,7 @@ interface VaultRecord { parsed: Record; } -async function listLedger(ename: string): Promise { +async function list(ename: string, ontology: string): Promise { const out: VaultRecord[] = []; let after: string | null = null; do { @@ -174,7 +188,7 @@ async function listLedger(ename: string): Promise { pageInfo { hasNextPage endCursor } } }`, - { first: 100, after, filter: { ontologyId: LEDGER_ONTOLOGY } }, + { first: 100, after, filter: { ontologyId: ontology } }, ); out.push(...data.metaEnvelopes.edges.map((e) => e.node)); after = data.metaEnvelopes.pageInfo.hasNextPage @@ -185,34 +199,37 @@ async function listLedger(ename: string): Promise { } // Same mutations and legacy acl as the live adapter (web3-adapter EVaultClient). -async function storeLedger(ename: string, payload: object): Promise { +async function store(ename: string, ontology: string, payload: object): Promise { const data: { storeMetaEnvelope: { metaEnvelope: { id: string } } } = await graphql( ename, `mutation ($input: MetaEnvelopeInput!) { storeMetaEnvelope(input: $input) { metaEnvelope { id } } }`, - { input: { ontology: LEDGER_ONTOLOGY, payload, acl: ["*"] } }, + { input: { ontology, payload, acl: ["*"] } }, ); return data.storeMetaEnvelope.metaEnvelope.id; } -async function updateLedger(ename: string, id: string, payload: object): Promise { +async function update(ename: string, id: string, ontology: string, payload: object): Promise { await graphql( ename, `mutation ($id: String!, $input: MetaEnvelopeInput!) { updateMetaEnvelopeById(id: $id, input: $input) { metaEnvelope { id } } }`, - { id, input: { ontology: LEDGER_ONTOLOGY, payload, acl: ["*"] } }, + { id, input: { ontology, payload, acl: ["*"] } }, ); } // ─── Expected state ────────────────────────────────────────────────────────── +/** eCurrency currency id → its record's global id, filled once currencies are synced. */ +const currencyGlobalIds = new Map(); +const expectedCurrencyId = (row: Ledger) => currencyGlobalIds.get(row.currencyId) ?? row.currencyId; + /** The payload the live adapter writes for a row (ledger.mapping.json). */ -async function expectedPayload(row: Ledger): Promise> { - const currencyGlobalId = await adapter.mappingDb.getGlobalId(row.currencyId); +function expectedPayload(row: Ledger): Record { return { - currencyId: currencyGlobalId ?? row.currencyId, + currencyId: expectedCurrencyId(row), accountId: row.accountId, accountType: row.accountType, amount: row.amount, @@ -252,17 +269,143 @@ function isCopyOf(record: Record, row: Ledger): boolean { ); } -/** Complete: carries the row's hash (rows from before hashing have none to carry). */ +/** + * Complete: carries the row's hash (rows from before hashing have none to + * carry) and points at the currency's record, not eCurrency's internal id. + */ const isComplete = (record: Record, row: Ledger) => - !row.hash || record.hash === row.hash; + (!row.hash || record.hash === row.hash) && record.currencyId === expectedCurrencyId(row); type Status = "synced" | "outdated" | "missing"; +/** Synced only when every copy is complete: apps may read a stale duplicate. */ function classify(records: VaultRecord[], row: Ledger) { const copies = records.filter((r) => isCopyOf(r.parsed, row)); - const complete = copies.find((r) => isComplete(r.parsed, row)); - const status: Status = complete ? "synced" : copies.length ? "outdated" : "missing"; - return { status, copies, best: complete ?? copies[0] ?? null }; + const stale = copies.filter((r) => !isComplete(r.parsed, row)); + const status: Status = !copies.length ? "missing" : stale.length ? "outdated" : "synced"; + const best = copies.find((r) => isComplete(r.parsed, row)) ?? copies[0] ?? null; + return { status, copies, stale, best }; +} + +// ─── Currencies ────────────────────────────────────────────────────────────── + +/** The payload the live adapter writes for a currency (currency.mapping.json). */ +const currencyPayload = (c: Currency) => ({ + name: c.name, + description: c.description, + ename: c.ename, + groupId: c.groupId, + allowNegative: c.allowNegative, + maxNegativeBalance: c.maxNegativeBalance, + allowNegativeGroupOnly: c.allowNegativeGroupOnly, + createdBy: c.createdBy, + createdAt: c.createdAt.toISOString(), + updatedAt: c.updatedAt.toISOString(), +}); + +interface CurrencyReport { + synced: number; + missing: number; + created: number; + /** Mapped, but the record is not in the creator's eVault (may be in another admin's). */ + elsewhere: string[]; + failed: string[]; +} + +/** + * Every currency's record in its creator's eVault. Missing ones are written + * (with --apply) and mapped, so ledger entries can carry the record's id. + */ +async function syncCurrencies( + currencies: Currency[], + enameOf: Map, +): Promise { + const report: CurrencyReport = { synced: 0, missing: 0, created: 0, elsewhere: [], failed: [] }; + for (const c of currencies) { + const label = `${c.name} (${c.ename})`; + const owner = enameOf.get(`user:${c.createdBy}`); + const mapped = await adapter.mappingDb.getGlobalId(c.id); + if (!owner) { + if (mapped) currencyGlobalIds.set(c.id, mapped); + report.failed.push(`${label}: creator has no eName`); + continue; + } + try { + const find = (records: VaultRecord[]) => + records.find( + (r) => + r.id === mapped || + (typeof r.parsed.ename === "string" && + normalizeEname(r.parsed.ename) === normalizeEname(c.ename)), + ); + const found = find(await list(owner, CURRENCY_ONTOLOGY)); + if (found) { + report.synced++; + currencyGlobalIds.set(c.id, found.id); + if (APPLY && found.id !== mapped) await rememberId(c.id, found.id); + continue; + } + if (mapped) { + currencyGlobalIds.set(c.id, mapped); + report.elsewhere.push(`${label}: mapped to ${mapped}, not in ${owner}`); + continue; + } + report.missing++; + if (!APPLY) { + report.failed.push(`${label}: no record in any eVault`); + continue; + } + const id = await store(owner, CURRENCY_ONTOLOGY, currencyPayload(c)); + if (!find(await list(owner, CURRENCY_ONTOLOGY))) { + report.failed.push(`${label}: written to ${owner} but not there on re-read`); + continue; + } + await rememberId(c.id, id); + currencyGlobalIds.set(c.id, id); + report.created++; + console.log(`[SYNC] currency ${label}: created ${id} in ${owner}`); + } catch (error) { + if (mapped) currencyGlobalIds.set(c.id, mapped); + report.failed.push(`${label}: ${(error as Error).message}`); + } + } + return report; +} + +// ─── Accounts ──────────────────────────────────────────────────────────────── + +/** One holder's account in one currency, as backfill-accounts.ts writes it. */ +interface ExpectedAccount { + key: string; + payload: { + accountId: string; + accountEname: string; + accountType: AccountType; + currencyEname: string; + currencyName: string; + balance: number; + createdAt: string; + }; +} + +// Currency eNames are unique, so a holder's account is (accountId, currencyEname). +const isAccountCopy = (record: Record, a: ExpectedAccount) => + record.accountId === a.payload.accountId && + typeof record.currencyEname === "string" && + normalizeEname(record.currencyEname) === a.payload.currencyEname; + +const isAccountCurrent = (record: Record, a: ExpectedAccount) => + Number(record.balance) === a.payload.balance && + record.accountType === a.payload.accountType && + record.currencyName === a.payload.currencyName && + typeof record.accountEname === "string" && + normalizeEname(record.accountEname) === a.payload.accountEname; + +function classifyAccount(records: VaultRecord[], a: ExpectedAccount) { + const copies = records.filter((r) => isAccountCopy(r.parsed, a)); + const current = copies.find((r) => isAccountCurrent(r.parsed, a)); + const status: Status = current ? "synced" : copies.length ? "outdated" : "missing"; + return { status, copies, best: current ?? copies[0] ?? null }; } // ─── Run ───────────────────────────────────────────────────────────────────── @@ -277,12 +420,26 @@ interface VaultReport { created: number; updated: number; stillUnsynced: string[]; + accounts: { + expected: number; + synced: number; + outdated: number; + missing: number; + duplicates: number; + created: number; + updated: number; + stillUnsynced: string[]; + }; error?: string; /** No eVault exists for this eName, so there is nowhere to write; not a failure. */ skipped?: string; } -async function syncVault(ename: string, rows: Ledger[]): Promise { +async function syncVault( + ename: string, + rows: Ledger[], + accounts: ExpectedAccount[], +): Promise { const report: VaultReport = { ename, expected: rows.length, @@ -293,28 +450,38 @@ async function syncVault(ename: string, rows: Ledger[]): Promise { created: 0, updated: 0, stillUnsynced: [], + accounts: { + expected: accounts.length, + synced: 0, + outdated: 0, + missing: 0, + duplicates: 0, + created: 0, + updated: 0, + stillUnsynced: [], + }, }; try { - const records = await listLedger(ename); - const work: { row: Ledger; status: Status; target: string | null }[] = []; + const records = rows.length ? await list(ename, LEDGER_ONTOLOGY) : []; + const work: { row: Ledger; stale: string[] }[] = []; for (const row of rows) { - const { status, copies, best } = classify(records, row); + const { status, copies, stale, best } = classify(records, row); report[status]++; if (copies.length > 1) report.duplicates += copies.length - 1; - if (status !== "synced") work.push({ row, status, target: best?.id ?? null }); - else if (APPLY && best) await remember(row, ename, best.id); + if (status !== "synced") work.push({ row, stale: stale.map((r) => r.id) }); + if (APPLY && best && status !== "missing") await remember(row, ename, best.id); } if (APPLY) { - for (const { row, status, target } of work) { - const payload = await expectedPayload(row); + for (const { row, stale } of work) { + const payload = expectedPayload(row); try { - if (status === "outdated" && target) { - await updateLedger(ename, target, payload); - await remember(row, ename, target); - report.updated++; + if (stale.length) { + // Every stale copy, duplicates included, so none is left behind. + for (const id of stale) await update(ename, id, LEDGER_ONTOLOGY, payload); + report.updated += stale.length; } else { - const id = await storeLedger(ename, payload); + const id = await store(ename, LEDGER_ONTOLOGY, payload); await remember(row, ename, id); report.created++; } @@ -324,24 +491,66 @@ async function syncVault(ename: string, rows: Ledger[]): Promise { } // Verify against the vault itself, not against what was sent. - const after = await listLedger(ename); + const after = rows.length ? await list(ename, LEDGER_ONTOLOGY) : []; report.stillUnsynced = rows .filter((row) => classify(after, row).status !== "synced") .map((row) => row.id); } else { report.stillUnsynced = work.map((w) => w.row.id); } + + await syncAccounts(ename, accounts, report.accounts); } catch (error) { if (error instanceof NoEVaultError) report.skipped = error.message; else report.error = (error as Error).message; report.stillUnsynced = rows.map((r) => r.id); + report.accounts.stillUnsynced = accounts.map((a) => a.key); } return report; } +async function syncAccounts( + ename: string, + accounts: ExpectedAccount[], + report: VaultReport["accounts"], +): Promise { + if (!accounts.length) return; + const records = await list(ename, ACCOUNT_ONTOLOGY); + const work: { account: ExpectedAccount; target: string | null }[] = []; + for (const account of accounts) { + const { status, copies, best } = classifyAccount(records, account); + report[status]++; + if (copies.length > 1) report.duplicates += copies.length - 1; + if (status !== "synced") work.push({ account, target: best?.id ?? null }); + } + if (!APPLY) { + report.stillUnsynced = work.map((w) => w.account.key); + return; + } + for (const { account, target } of work) { + try { + if (target) { + await update(ename, target, ACCOUNT_ONTOLOGY, account.payload); + report.updated++; + } else { + await store(ename, ACCOUNT_ONTOLOGY, account.payload); + report.created++; + } + } catch (error) { + console.error(`[SYNC] ${ename} account ${account.key}: ${(error as Error).message}`); + } + } + const after = await list(ename, ACCOUNT_ONTOLOGY); + report.stillUnsynced = accounts + .filter((a) => classifyAccount(after, a).status !== "synced") + .map((a) => a.key); +} + /** Point the live adapter's mapping at the copy that is really in the vault. */ -async function remember(row: Ledger, ename: string, globalId: string): Promise { - const localId = `${row.id}::${ename}`; +const remember = (row: Ledger, ename: string, globalId: string) => + rememberId(`${row.id}::${ename}`, globalId); + +async function rememberId(localId: string, globalId: string): Promise { const current = await adapter.mappingDb.getGlobalId(localId); if (current === globalId) return; try { @@ -355,7 +564,7 @@ async function remember(row: Ledger, ename: string, globalId: string): Promise row.createdAt <= cutoff); - // Parties per row, exactly as the subscriber resolves them. + // Parties per row, as the subscriber resolves them (plus holders of one-sided rows). const byVault = new Map(); let noParties = 0; const unnamed = new Set(); @@ -388,20 +597,68 @@ async function main() { if (ename) parties.add(ename); else unnamed.add(`${type}:${id}`); } - if (parties.size === 0) noParties++; + // Mints, burns and "Account initialized" have no parties: the live sync + // never wrote them anywhere, so they go to the holder's own vault. + if (parties.size === 0) { + const holder = enameOf.get(`${row.accountType}:${row.accountId}`); + if (holder) parties.add(holder); + else noParties++; + } for (const ename of parties) { if (ONLY && ename !== ONLY) continue; byVault.set(ename, [...(byVault.get(ename) ?? []), row]); } } + const currencyList = await AppDataSource.getRepository(Currency).find(); + const currencies = new Map(currencyList.map((c) => [c.id, c])); + const currencyReport = await syncCurrencies(currencyList, enameOf); + console.log( + `[SYNC] ${currencyList.length} currencies: ${currencyReport.synced} synced, ${currencyReport.missing} missing` + + (APPLY ? `, ${currencyReport.created} created` : ""), + ); + + // One Account per holder and currency. Its balance is the sum of the + // chain, which wins over any row's stored balance should they disagree. + // Every row counts here, including ones with no parties ("Account initialized"). + const accountsByVault = new Map>(); + for (const row of rows) { + const ename = enameOf.get(`${row.accountType}:${row.accountId}`); + const currency = currencies.get(row.currencyId); + if (!ename || !currency?.ename) continue; + if (ONLY && ename !== ONLY) continue; + const accounts = accountsByVault.get(ename) ?? new Map(); + accountsByVault.set(ename, accounts); + const key = `${row.accountType}:${row.accountId}|${normalizeEname(currency.ename)}`; + const existing = accounts.get(key); + const balance = Math.round(((existing?.payload.balance ?? 0) + Number(row.amount)) * 100) / 100; + accounts.set(key, { + key, + payload: { + accountId: row.accountId, + accountEname: ename, + accountType: row.accountType, + currencyEname: normalizeEname(currency.ename), + currencyName: currency.name, + balance, + createdAt: existing?.payload.createdAt ?? row.createdAt.toISOString(), + }, + }); + } + const accountCount = [...accountsByVault.values()].reduce((s, m) => s + m.size, 0); + const vaultNames = new Set([...byVault.keys(), ...accountsByVault.keys()]); + console.log( - `[SYNC] ${rows.length} ledger rows, ${byVault.size} eVaults to check` + - (noParties ? `, ${noParties} rows with no parties (not synced by design, e.g. "Account initialized")` : "") + + `[SYNC] ${rows.length} ledger rows, ${accountCount} accounts, ${vaultNames.size} eVaults to check` + + (noParties ? `, ${noParties} rows with no parties and a holder with no eName (cannot be synced)` : "") + (unnamed.size ? `, ${unnamed.size} accounts with no eName (cannot be synced)` : ""), ); - const vaults = [...byVault.entries()]; + const vaults = [...vaultNames].map((ename): [string, Ledger[], ExpectedAccount[]] => [ + ename, + byVault.get(ename) ?? [], + [...(accountsByVault.get(ename)?.values() ?? [])], + ]); const reports: VaultReport[] = []; const inFlight = new Set(); const heartbeat = setInterval(() => { @@ -413,24 +670,29 @@ async function main() { await Promise.all( Array.from({ length: VAULT_CONCURRENCY }, async () => { while (next < vaults.length) { - const [ename, vaultRows] = vaults[next++]; + const [ename, vaultRows, vaultAccounts] = vaults[next++]; inFlight.add(ename); - const r = await syncVault(ename, vaultRows); + const r = await syncVault(ename, vaultRows, vaultAccounts); inFlight.delete(ename); reports.push(r); if (r.skipped) { - console.log(`[SYNC] ${ename}: ${r.expected} rows · SKIPPED, ${r.skipped}`); + console.log( + `[SYNC] ${ename}: ${r.expected} rows, ${r.accounts.expected} accounts · SKIPPED, ${r.skipped}`, + ); continue; } + const a = r.accounts; const verdict = r.error ? `ERROR ${r.error}` - : r.stillUnsynced.length - ? `✗ ${r.stillUnsynced.length} unsynced` + : r.stillUnsynced.length || a.stillUnsynced.length + ? `✗ ${r.stillUnsynced.length} rows, ${a.stillUnsynced.length} accounts unsynced` : "✓ verified"; console.log( `[SYNC] ${ename}: ${r.expected} rows · ${r.synced} synced · ${r.missing} missing · ${r.outdated} outdated` + (r.duplicates ? ` · ${r.duplicates} duplicate copies` : "") + (APPLY ? ` → created ${r.created}, updated ${r.updated}` : "") + + ` | ${a.expected} accounts · ${a.synced} synced · ${a.missing} missing · ${a.outdated} stale` + + (APPLY ? ` → created ${a.created}, updated ${a.updated}` : "") + ` · ${verdict}`, ); } @@ -438,28 +700,51 @@ async function main() { ); clearInterval(heartbeat); - const sum = (k: "expected" | "synced" | "missing" | "outdated" | "duplicates" | "created" | "updated") => - checked.reduce((s, r) => s + r[k], 0); + type Counter = "expected" | "synced" | "missing" | "outdated" | "duplicates" | "created" | "updated"; + const sum = (k: Counter) => checked.reduce((s, r) => s + r[k], 0); + const sumAccounts = (k: Counter) => checked.reduce((s, r) => s + r.accounts[k], 0); const skipped = reports.filter((r) => r.skipped); const checked = reports.filter((r) => !r.skipped); - const failed = checked.filter((r) => r.error || r.stillUnsynced.length); + const failed = checked.filter( + (r) => r.error || r.stillUnsynced.length || r.accounts.stillUnsynced.length, + ); console.log("[SYNC] =========================================="); + console.log(`[SYNC] Currencies : ${currencyReport.synced} synced, ${currencyReport.missing} missing` + + (APPLY ? `, ${currencyReport.created} created` : "")); + for (const line of currencyReport.failed) console.log(`[SYNC] ✗ ${line}`); + for (const line of currencyReport.elsewhere) console.log(`[SYNC] ? ${line}`); console.log(`[SYNC] Row copies expected : ${sum("expected")}`); console.log(`[SYNC] Already synced : ${sum("synced")}`); console.log(`[SYNC] Missing : ${sum("missing")}`); - console.log(`[SYNC] Outdated (no hash) : ${sum("outdated")}`); + console.log(`[SYNC] Outdated : ${sum("outdated")} (no hash, or the wrong currency id)`); console.log(`[SYNC] Duplicate copies : ${sum("duplicates")} (reported, never deleted)`); if (APPLY) { console.log(`[SYNC] Created : ${sum("created")}`); console.log(`[SYNC] Updated in place : ${sum("updated")}`); } + console.log(`[SYNC] Accounts expected : ${sumAccounts("expected")}`); + console.log(`[SYNC] Already synced : ${sumAccounts("synced")}`); + console.log(`[SYNC] Missing : ${sumAccounts("missing")}`); + console.log(`[SYNC] Stale : ${sumAccounts("outdated")} (wrong balance or details)`); + console.log(`[SYNC] Duplicate copies : ${sumAccounts("duplicates")} (reported, never deleted)`); + if (APPLY) { + console.log(`[SYNC] Created : ${sumAccounts("created")}`); + console.log(`[SYNC] Updated in place : ${sumAccounts("updated")}`); + } console.log( `[SYNC] Vaults fully in sync : ${checked.length - failed.length}/${checked.length}` + (APPLY ? " (re-read from each eVault after writing)" : ""), ); for (const r of failed) { console.log( - `[SYNC] ${r.ename}: ${r.error ?? `${r.stillUnsynced.length} rows unsynced: ${r.stillUnsynced.slice(0, 5).join(", ")}${r.stillUnsynced.length > 5 ? ", …" : ""}`}`, + `[SYNC] ${r.ename}: ${r.error ?? [ + r.stillUnsynced.length && + `${r.stillUnsynced.length} rows unsynced: ${r.stillUnsynced.slice(0, 5).join(", ")}${r.stillUnsynced.length > 5 ? ", …" : ""}`, + r.accounts.stillUnsynced.length && + `${r.accounts.stillUnsynced.length} accounts unsynced: ${r.accounts.stillUnsynced.slice(0, 5).join(", ")}`, + ] + .filter(Boolean) + .join("; ")}`, ); } if (skipped.length) { @@ -471,7 +756,7 @@ async function main() { console.log("[SYNC] =========================================="); await AppDataSource.destroy(); - return failed.length === 0; + return failed.length === 0 && currencyReport.failed.length === 0; } main()