feat(federation): backfill snowflake-named replicated-user stubs
Heals existing legacy stubs (created when resolveOrCreateReplicatedUser used homeUserId@domain) by asking the peer for the canonical username via lookupRemoteUserByHomeId and rewriting the local row. Idempotent and collision-safe. Wired into onPeerActivated (per-origin) so future peer flaps re-attempt for stubs whose home was unreachable on a prior pass, and into a one-shot pass at startupBootstrapSync for all currently-active peers (not just first-time lastSyncedAt=0 peers).
This commit is contained in:
@@ -52,6 +52,20 @@ export async function onPeerActivated(
|
||||
resetOutboxBackoff(peerId);
|
||||
await syncPeerMutationLog(peerId, reason);
|
||||
await fanoutOutboundSubscribers(peerId);
|
||||
|
||||
// Look up the peer's origin once for the post-sync invariants.
|
||||
const peerRow = getDb()
|
||||
.select({ origin: schema.federationPeers.origin })
|
||||
.from(schema.federationPeers)
|
||||
.where(eq(schema.federationPeers.id, peerId))
|
||||
.get();
|
||||
if (peerRow?.origin) {
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
await backfillStubUsernamesForPeer(peerRow.origin).catch((e) => {
|
||||
console.warn(`[onPeerActivated] backfillStubUsernamesForPeer(${peerRow.origin}) failed`, e);
|
||||
});
|
||||
}
|
||||
|
||||
const { connectionManager } = await import('../ws/handler.js');
|
||||
connectionManager.sendToAdmins({ type: 'federation_peers_changed' as const });
|
||||
} catch (err) {
|
||||
@@ -274,15 +288,31 @@ export async function startupBootstrapSync(): Promise<void> {
|
||||
if (!isFederationRelayEnabled()) return;
|
||||
|
||||
const db = getDb();
|
||||
const peers = db.select().from(schema.federationPeers)
|
||||
const firstTimePeers = db.select().from(schema.federationPeers)
|
||||
.where(and(
|
||||
eq(schema.federationPeers.status, 'active'),
|
||||
eq(schema.federationPeers.lastSyncedAt, 0),
|
||||
)).all();
|
||||
|
||||
for (const peer of peers) {
|
||||
for (const peer of firstTimePeers) {
|
||||
await onPeerActivated(peer.id, 'startup_bootstrap');
|
||||
}
|
||||
|
||||
// One-shot stub-username backfill for ALL currently-active peers (including
|
||||
// those with lastSyncedAt > 0). Heals legacy snowflake-named stubs created
|
||||
// before resolveOrCreateReplicatedUser used the realname scheme. Idempotent;
|
||||
// skips stubs already migrated. Non-blocking — failures retry on next
|
||||
// onPeerActivated for that origin.
|
||||
const allActivePeers = db.select({ origin: schema.federationPeers.origin })
|
||||
.from(schema.federationPeers)
|
||||
.where(eq(schema.federationPeers.status, 'active'))
|
||||
.all();
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
for (const peer of allActivePeers) {
|
||||
backfillStubUsernamesForPeer(peer.origin).catch((err) => {
|
||||
console.warn(`[startup] stub-backfill ${peer.origin} failed`, err);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export type PeerDeactivationReason =
|
||||
|
||||
@@ -0,0 +1,161 @@
|
||||
import { describe, it, expect, beforeEach, vi } from 'vitest';
|
||||
import Database from 'better-sqlite3';
|
||||
import { drizzle } from 'drizzle-orm/better-sqlite3';
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import * as schema from '../db/schema.js';
|
||||
import { eq } from 'drizzle-orm';
|
||||
|
||||
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
||||
type TestDb = ReturnType<typeof drizzle<typeof schema>>;
|
||||
let sqlite: Database.Database;
|
||||
let testDb: TestDb;
|
||||
|
||||
const lookupCalls: Array<{ peerOrigin: string; homeUserId: string }> = [];
|
||||
const lookupResponses = new Map<string, unknown>();
|
||||
|
||||
vi.mock('../db/index.js', () => ({
|
||||
getDb: () => testDb,
|
||||
getRawDb: () => sqlite,
|
||||
schema,
|
||||
}));
|
||||
|
||||
vi.mock('./federationLookup.js', () => ({
|
||||
lookupRemoteUserByHomeId: vi.fn(async (peerOrigin: string, homeUserId: string) => {
|
||||
lookupCalls.push({ peerOrigin, homeUserId });
|
||||
const r = lookupResponses.get(homeUserId);
|
||||
if (!r) return { ok: false, reason: 'not_found' };
|
||||
return r;
|
||||
}),
|
||||
}));
|
||||
|
||||
function applyMigrations(db: Database.Database): void {
|
||||
const migrationsDir = path.resolve(__dirname, '../../drizzle');
|
||||
const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
|
||||
for (const f of files) {
|
||||
const sql = fs.readFileSync(path.join(migrationsDir, f), 'utf8');
|
||||
for (const stmt of sql.split(/-->\s*statement-breakpoint/)) {
|
||||
const clean = stmt.trim();
|
||||
if (clean) db.exec(clean);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
sqlite = new Database(':memory:');
|
||||
testDb = drizzle(sqlite, { schema });
|
||||
applyMigrations(sqlite);
|
||||
lookupCalls.length = 0;
|
||||
lookupResponses.clear();
|
||||
// Active peer
|
||||
testDb.insert(schema.federationPeers).values({
|
||||
id: 'peer-orbit',
|
||||
origin: 'https://orbit.ddns.net',
|
||||
hmacSecret: 'a'.repeat(64),
|
||||
status: 'active',
|
||||
createdAt: Date.now(),
|
||||
lastSeenAt: Date.now(),
|
||||
}).run();
|
||||
});
|
||||
|
||||
describe('backfillStubUsernamesForPeer', () => {
|
||||
it('rewrites snowflake-style username to realname when lookup succeeds', async () => {
|
||||
// Seed legacy stub: username = `${homeUserId}@${domain}` (the old scheme)
|
||||
testDb.insert(schema.users).values({
|
||||
id: 'stub-1',
|
||||
username: 'home-1@orbit.ddns.net',
|
||||
displayName: null,
|
||||
passwordHash: '!federation-replicated',
|
||||
status: 'offline',
|
||||
isAdmin: 0,
|
||||
homeInstance: 'orbit.ddns.net',
|
||||
homeUserId: 'home-1',
|
||||
createdAt: Date.now(),
|
||||
}).run();
|
||||
lookupResponses.set('home-1', {
|
||||
ok: true,
|
||||
homeUserId: 'home-1',
|
||||
username: 'pbtest3',
|
||||
profile: { displayName: null, avatar: null, avatarColor: null, banner: null, bio: null },
|
||||
});
|
||||
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
await backfillStubUsernamesForPeer('https://orbit.ddns.net');
|
||||
|
||||
expect(lookupCalls).toEqual([{ peerOrigin: 'https://orbit.ddns.net', homeUserId: 'home-1' }]);
|
||||
const row = testDb.select().from(schema.users).where(eq(schema.users.id, 'stub-1')).get();
|
||||
expect(row!.username).toBe('pbtest3@orbit.ddns.net');
|
||||
expect(row!.displayName).toBe('pbtest3'); // displayName ?? username fallback fills in real handle
|
||||
});
|
||||
|
||||
it('skips stubs whose username is already human-readable (no lookup triggered)', async () => {
|
||||
testDb.insert(schema.users).values({
|
||||
id: 'stub-1',
|
||||
username: 'pbtest3@orbit.ddns.net', // already migrated
|
||||
displayName: 'pbtest3',
|
||||
passwordHash: '!federation-replicated',
|
||||
status: 'offline',
|
||||
isAdmin: 0,
|
||||
homeInstance: 'orbit.ddns.net',
|
||||
homeUserId: 'home-1',
|
||||
createdAt: Date.now(),
|
||||
}).run();
|
||||
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
await backfillStubUsernamesForPeer('https://orbit.ddns.net');
|
||||
|
||||
expect(lookupCalls).toEqual([]); // no lookup triggered
|
||||
});
|
||||
|
||||
it('leaves stub untouched on lookup miss (tombstoned home user)', async () => {
|
||||
testDb.insert(schema.users).values({
|
||||
id: 'stub-1',
|
||||
username: 'unknown-id@orbit.ddns.net',
|
||||
displayName: null,
|
||||
passwordHash: '!federation-replicated',
|
||||
status: 'offline',
|
||||
isAdmin: 0,
|
||||
homeInstance: 'orbit.ddns.net',
|
||||
homeUserId: 'unknown-id',
|
||||
createdAt: Date.now(),
|
||||
}).run();
|
||||
// No lookupResponses entry → mock returns { ok: false, reason: 'not_found' }
|
||||
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
await backfillStubUsernamesForPeer('https://orbit.ddns.net');
|
||||
|
||||
const row = testDb.select().from(schema.users).where(eq(schema.users.id, 'stub-1')).get();
|
||||
expect(row!.username).toBe('unknown-id@orbit.ddns.net'); // unchanged
|
||||
expect(row!.displayName).toBeNull(); // unchanged
|
||||
});
|
||||
|
||||
it('is a no-op when peer is not active', async () => {
|
||||
testDb.update(schema.federationPeers)
|
||||
.set({ status: 'unreachable' })
|
||||
.where(eq(schema.federationPeers.id, 'peer-orbit'))
|
||||
.run();
|
||||
testDb.insert(schema.users).values({
|
||||
id: 'stub-1',
|
||||
username: 'home-1@orbit.ddns.net',
|
||||
displayName: null,
|
||||
passwordHash: '!federation-replicated',
|
||||
status: 'offline',
|
||||
isAdmin: 0,
|
||||
homeInstance: 'orbit.ddns.net',
|
||||
homeUserId: 'home-1',
|
||||
createdAt: Date.now(),
|
||||
}).run();
|
||||
lookupResponses.set('home-1', {
|
||||
ok: true,
|
||||
homeUserId: 'home-1',
|
||||
username: 'pbtest3',
|
||||
profile: { displayName: null, avatar: null, avatarColor: null, banner: null, bio: null },
|
||||
});
|
||||
|
||||
const { backfillStubUsernamesForPeer } = await import('./federationStubBackfill.js');
|
||||
await backfillStubUsernamesForPeer('https://orbit.ddns.net');
|
||||
|
||||
expect(lookupCalls).toEqual([]); // gated on peer status
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,89 @@
|
||||
import { and, eq, like } from 'drizzle-orm';
|
||||
import { getDb, schema } from '../db/index.js';
|
||||
import { lookupRemoteUserByHomeId } from './federationLookup.js';
|
||||
import { extractDomain } from '../routes/federation.js';
|
||||
|
||||
/**
|
||||
* For each replicated stub on this instance whose username still matches the
|
||||
* legacy `<homeUserId>@<domain>` pattern AND whose home_instance equals the
|
||||
* given peer's domain, ask the peer for the canonical username via
|
||||
* lookupRemoteUserByHomeId and rewrite the stub.
|
||||
*
|
||||
* Idempotent — stubs already migrated (username does not start with their
|
||||
* homeUserId) are skipped without a network call.
|
||||
*
|
||||
* Gated on peer.status='active' — the lookup endpoint requires the requesting
|
||||
* peer to be active on the receiving side. We additionally check our local
|
||||
* peer row here so we don't waste outbound RTTs on peers we know aren't ready.
|
||||
*
|
||||
* Called from onPeerActivated (per-origin, gated on peer status='active') and
|
||||
* from a one-shot startup pass for any peer already active at boot.
|
||||
*/
|
||||
export async function backfillStubUsernamesForPeer(peerOrigin: string): Promise<void> {
|
||||
const db = getDb();
|
||||
const peerDomain = extractDomain(peerOrigin);
|
||||
|
||||
const peer = db
|
||||
.select({ status: schema.federationPeers.status })
|
||||
.from(schema.federationPeers)
|
||||
.where(eq(schema.federationPeers.origin, peerOrigin))
|
||||
.get();
|
||||
if (!peer || peer.status !== 'active') return;
|
||||
|
||||
// Coarse SQL prefilter: stubs from this peer whose username ends with @peerDomain.
|
||||
// We then narrow in JS to the legacy `<homeUserId>@<domain>` shape because Drizzle
|
||||
// can't express that comparison portably.
|
||||
const candidates = db
|
||||
.select()
|
||||
.from(schema.users)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.users.homeInstance, peerDomain),
|
||||
eq(schema.users.isDeleted, 0),
|
||||
like(schema.users.username, '%@' + peerDomain),
|
||||
),
|
||||
)
|
||||
.all();
|
||||
|
||||
for (const stub of candidates) {
|
||||
if (!stub.homeUserId) continue;
|
||||
const expectedLegacy = `${stub.homeUserId}@${peerDomain}`.toLowerCase();
|
||||
if (stub.username !== expectedLegacy) continue; // already migrated or non-legacy shape
|
||||
|
||||
const result = await lookupRemoteUserByHomeId(peerOrigin, stub.homeUserId);
|
||||
if (!result.ok) {
|
||||
// not_found / unreachable / rate_limited — leave untouched.
|
||||
// Will retry on next onPeerActivated for this origin.
|
||||
continue;
|
||||
}
|
||||
|
||||
const newUsername = `${result.username}@${peerDomain}`.toLowerCase();
|
||||
if (newUsername === stub.username) continue; // already correct
|
||||
|
||||
// Collision check: another row at the target username (rare under the new
|
||||
// scheme but possible from prior partial replication).
|
||||
const collision = db
|
||||
.select({ id: schema.users.id })
|
||||
.from(schema.users)
|
||||
.where(eq(schema.users.username, newUsername))
|
||||
.get();
|
||||
if (collision && collision.id !== stub.id) {
|
||||
console.warn(`[stub-backfill] username collision on ${newUsername} — leaving stub ${stub.id} as ${stub.username}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Fill displayName from result.profile if the stub has none, mirroring the
|
||||
// displayName ?? username fallback applied at hydrate / profile_update time.
|
||||
const updates: { username: string; displayName?: string } = { username: newUsername };
|
||||
if (!stub.displayName) {
|
||||
updates.displayName = result.profile.displayName ?? result.username;
|
||||
}
|
||||
|
||||
db.update(schema.users)
|
||||
.set(updates)
|
||||
.where(eq(schema.users.id, stub.id))
|
||||
.run();
|
||||
|
||||
console.log(`[stub-backfill] rewrote stub ${stub.id}: ${stub.username} → ${newUsername}`);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user