| … | … | @@ -107,8 +107,10 @@ export async function startApBridge (config, { log = defaultLog, bootstrap = nul |
| 107 | 107 | // ⚠ Assigned below, once PerUser exists. A null resolver means a queued entry with a signerId is |
| 108 | 108 | // HELD rather than sent under the wrong identity — the safe direction. |
| 109 | 109 | let signerByKeyId = null |
| 110 | // Kept, so metrics() can read pending and dead entries; the deliverer exposes only counts. |
| 111 | const deliveryQueue = new DeliveryQueue(join(dataDir, 'queue.json')) |
| 110 | 112 | const outbound = createQueuedDeliverer({ |
| 111 | | queue: new DeliveryQueue(join(dataDir, 'queue.json')), |
| 113 | queue: deliveryQueue, |
| 112 | 114 | deliver: sendNow, |
| 113 | 115 | resolveSigner: (keyId) => (signerByKeyId ? signerByKeyId(keyId) : null), |
| 114 | 116 | log |
| … | … | @@ -324,9 +326,12 @@ export async function startApBridge (config, { log = defaultLog, bootstrap = nul |
| 324 | 326 | ? config.users |
| 325 | 327 | : (registrations ? [{ username: config.username, primary: true, favelaSocket: config.favelaSocket || null }] : null) |
| 326 | 328 | |
| 329 | // Hoisted out of the block below so metrics() can read them; null when there is no user registry. |
| 330 | let users = null |
| 331 | let perUser = null |
| 327 | 332 | if (userEntries) { |
| 328 | | const users = new Users(userEntries, config.host) |
| 329 | | const perUser = new PerUser({ |
| 333 | users = new Users(userEntries, config.host) |
| 334 | perUser = new PerUser({ |
| 330 | 335 | users, |
| 331 | 336 | dataDir, |
| 332 | 337 | // ⛔ The overlay's Files, so a held user's avatar resolves from THEIR bytes rather than being |
| … | … | @@ -632,6 +637,50 @@ export async function startApBridge (config, { log = defaultLog, bootstrap = nul |
| 632 | 637 | } |
| 633 | 638 | } |
| 634 | 639 | |
| 640 | /** |
| 641 | * What the gateway's panel and CLI show for this role — read from what the bridge already keeps on |
| 642 | * disk and in memory. Nothing here talks to the network. |
| 643 | * |
| 644 | * `delivered` is the publisher's delivered-set size (rkeys sent at least once), read from the same |
| 645 | * state file the publisher writes. `failing` lists pending entries that have failed at least once, |
| 646 | * newest error first, with the inbox HOST only — an operator needs to see which server is refusing, |
| 647 | * not the activity body. |
| 648 | */ |
| 649 | const metrics = async () => { |
| 650 | const deliveredFor = (u) => { |
| 651 | const file = u.primary ? 'publisher.json' : 'publisher-' + u.username + '.json' |
| 652 | try { |
| 653 | const d = JSON.parse(fs.readFileSync(join(dataDir, file), 'utf8')).delivered |
| 654 | return Array.isArray(d) ? d.length : Object.keys(d || {}).length |
| 655 | } catch { return 0 } |
| 656 | } |
| 657 | const list = users ? users.all() : [{ username: config.username, primary: true }] |
| 658 | const userRows = [] |
| 659 | for (const u of list) { |
| 660 | const store = u.primary || !perUser ? followers : perUser.followers(u) |
| 661 | userRows.push({ username: u.username, primary: !!u.primary, held: !!u.peer, followers: await store.count(), delivered: deliveredFor(u) }) |
| 662 | } |
| 663 | const pending = await deliveryQueue.list() |
| 664 | const dead = await deliveryQueue.dead() |
| 665 | const hostOf = (inbox) => { try { return new URL(inbox).host } catch { return String(inbox || '') } } |
| 666 | const failing = pending.filter((e) => e.attempts > 0) |
| 667 | .sort((a, b) => (b.nextAt || 0) - (a.nextAt || 0)) |
| 668 | .slice(0, 10) |
| 669 | .map((e) => ({ host: hostOf(e.inbox), attempts: e.attempts, lastError: e.lastError, nextAt: e.nextAt })) |
| 670 | const newest = await inbound.list({ limit: 1 }) |
| 671 | return { |
| 672 | users: userRows, |
| 673 | queue: { |
| 674 | pending: pending.length, |
| 675 | dead: dead.length, |
| 676 | oldestPendingAt: pending.length ? Math.min(...pending.map((e) => e.createdAt || Date.now())) : null, |
| 677 | failing |
| 678 | }, |
| 679 | inbound: { count: await inbound.count(), lastReceivedAt: newest.length ? (newest[0].receivedAt || null) : null }, |
| 680 | registrations: registrations ? (await registrations.list()).length : 0 |
| 681 | } |
| 682 | } |
| 683 | |
| 635 | 684 | const closeServer = (s) => new Promise((resolve) => s.close(() => resolve())) |
| 636 | 685 | |
| 637 | 686 | return { |
| … | … | @@ -639,6 +688,7 @@ export async function startApBridge (config, { log = defaultLog, bootstrap = nul |
| 639 | 688 | adminPort, |
| 640 | 689 | bridgeKey: open.reg ? open.reg.key : null, |
| 641 | 690 | ...callbacks, |
| 691 | metrics, |
| 642 | 692 | handlers: bridgeHandlers({ name: 'ap-bridge', version, ...callbacks }), |
| 643 | 693 | // The stop sequence. The delivery drain first — never the queue itself: anything still owed is on |
| 644 | 694 | // disk and is picked up by the next boot, which is the difference that module was built to make. |