fix(api-sync): various fixes and unit test coverage

packages/api-sync/source/defaults.test.ts

@@ -0,0 +1,86 @@
+import { describe } from "@mainsail/test-runner";
+
+import { defaults } from "./defaults.js";
+
+const ENV_KEYS = [
+	"MAINSAIL_API_SYNC_ENABLED",
+	"MAINSAIL_API_SYNC_INTERVAL",
+	"MAINSAIL_API_SYNC_TOKEN_CACHE_SIZE",
+	"MAINSAIL_API_SYNC_TOKEN_WHITELIST_SYNC_INTERVAL",
+	"MAINSAIL_API_SYNC_TOKEN_WHITELIST_REMOTE_URL",
+	"MAINSAIL_API_SYNC_RESTORE_BLOCKS_BATCH_SIZE",
+];
+
+// The module reads the environment at import time, so a fresh copy is imported
+// with a cache-busting query once the environment has been prepared.
+const importFresh = async (): Promise<typeof defaults> => {
+	const { defaults: freshDefaults } = await import(`./defaults.js?bust=${Math.random()}`);
+	return freshDefaults;
+};
+
+describe<{
+	previousEnv: Record<string, string | undefined>;
+}>("defaults", ({ it, beforeEach, afterEach, assert }) => {
+	beforeEach((context) => {
+		context.previousEnv = {};
+		for (const key of ENV_KEYS) {
+			context.previousEnv[key] = process.env[key];
+			delete process.env[key];
+		}
+	});
+
+	afterEach(({ previousEnv }) => {
+		for (const key of ENV_KEYS) {
+			if (previousEnv[key] === undefined) {
+				delete process.env[key];
+			} else {
+				process.env[key] = previousEnv[key];
+			}
+		}
+	});
+
+	it("uses the documented defaults when no environment variables are set", async () => {
+		const fresh = await importFresh();
+
+		assert.false(fresh.enabled);
+		assert.equal(fresh.restore.blocks.batchSize, 1000);
+		assert.equal(fresh.syncInterval, 8000);
+		assert.equal(fresh.tokenCacheSize, 256);
+		assert.equal(fresh.tokenWhitelistRefreshInterval, 60_000);
+		assert.equal(fresh.tokenWhitelistRemoteUrl, "");
+	});
+
+	it("honors the environment variable overrides", async () => {
+		process.env.MAINSAIL_API_SYNC_ENABLED = "true";
+		process.env.MAINSAIL_API_SYNC_INTERVAL = "5000";
+		process.env.MAINSAIL_API_SYNC_TOKEN_CACHE_SIZE = "16";
+		process.env.MAINSAIL_API_SYNC_TOKEN_WHITELIST_SYNC_INTERVAL = "30000";
+		process.env.MAINSAIL_API_SYNC_TOKEN_WHITELIST_REMOTE_URL = "https://tokens.example.org/whitelist.json";
+		process.env.MAINSAIL_API_SYNC_RESTORE_BLOCKS_BATCH_SIZE = "50";
+
+		const fresh = await importFresh();
+
+		assert.true(fresh.enabled);
+		assert.equal(fresh.restore.blocks.batchSize, "50");
+		assert.equal(fresh.syncInterval, "5000");
+		assert.equal(fresh.tokenCacheSize, "16");
+		assert.equal(fresh.tokenWhitelistRefreshInterval, "30000");
+		assert.equal(fresh.tokenWhitelistRemoteUrl, "https://tokens.example.org/whitelist.json");
+	});
+
+	it("treats '1' as enabled", async () => {
+		process.env.MAINSAIL_API_SYNC_ENABLED = "1";
+
+		const fresh = await importFresh();
+
+		assert.true(fresh.enabled);
+	});
+
+	it("treats any other value as disabled", async () => {
+		process.env.MAINSAIL_API_SYNC_ENABLED = "yes";
+
+		const fresh = await importFresh();
+
+		assert.false(fresh.enabled);
+	});
+});

packages/api-sync/source/listeners.test.ts

@@ -0,0 +1,130 @@
+import { Identifiers as ApiDatabaseIdentifiers } from "@mainsail/api-database";
+import { Events, Identifiers } from "@mainsail/constants";
+import { Application } from "@mainsail/kernel";
+import { describe } from "@mainsail/test-runner";
+
+import { Listeners } from "./listeners.js";
+
+const makeFakeRepo = (primaryColumn: string, tableName: string) => ({
+	clear: () => {},
+	delete: () => {},
+	metadata: {
+		primaryColumns: [{ propertyName: primaryColumn }],
+		tableNameWithoutPrefix: tableName,
+	},
+	upsert: () => {},
+});
+
+describe<{
+	app: Application;
+	listeners: Listeners;
+	events: { listen: any; forget: any };
+	apiNodeRepo: ReturnType<typeof makeFakeRepo>;
+	contractRepo: ReturnType<typeof makeFakeRepo>;
+	peerRepo: ReturnType<typeof makeFakeRepo>;
+	pluginRepo: ReturnType<typeof makeFakeRepo>;
+	peerFactoryArgs: unknown[];
+}>("Listeners", ({ it, beforeEach, assert, spy, clock }) => {
+	beforeEach((context) => {
+		context.events = { forget: () => {}, listen: () => {} };
+		context.apiNodeRepo = makeFakeRepo("url", "api_nodes");
+		context.contractRepo = makeFakeRepo("name", "contracts");
+		context.peerRepo = makeFakeRepo("ip", "peers");
+		context.pluginRepo = makeFakeRepo("name", "plugins");
+
+		context.app = new Application();
+		context.app
+			.bind(Identifiers.ServiceProvider.Configuration)
+			.toConstantValue({ getRequired: () => 1000 })
+			.whenTagged("plugin", "api-sync");
+		context.app.bind(ApiDatabaseIdentifiers.DataSource).toConstantValue({});
+		context.app.bind(Identifiers.Services.EventDispatcher.Service).toConstantValue(context.events);
+		context.app.bind(Identifiers.ApiSync.Logger).toConstantValue({ debug: () => {}, error: () => {} });
+		context.app.bind(ApiDatabaseIdentifiers.ApiNodeRepositoryFactory).toConstantValue(() => context.apiNodeRepo);
+		context.app.bind(ApiDatabaseIdentifiers.ContractRepositoryFactory).toConstantValue(() => context.contractRepo);
+		context.peerFactoryArgs = [];
+		context.app.bind(ApiDatabaseIdentifiers.PeerRepositoryFactory).toConstantValue((dataSource: unknown) => {
+			context.peerFactoryArgs.push(dataSource);
+			return context.peerRepo;
+		});
+		context.app.bind(ApiDatabaseIdentifiers.PluginRepositoryFactory).toConstantValue(() => context.pluginRepo);
+		context.app.rebind(Identifiers.ServiceProvider.Repository).toConstantValue({
+			get: () => ({ config: () => ({ all: () => ({}) }) }),
+		});
+
+		context.listeners = context.app.resolve(Listeners);
+	});
+
+	it("register: registers every concrete listener on the event dispatcher", async ({ listeners, events }) => {
+		const listen = spy(events, "listen");
+
+		await listeners.register();
+
+		// ApiNodes (2) + DeployerContracts (1) + Peers (3) + Plugins (1)
+		listen.calledTimes(7);
+		listen.calledWith(Events.ApiNodeEvent.Added);
+		listen.calledWith(Events.DeployerEvent.ContractCreated);
+		listen.calledWith(Events.PeerEvent.Updated);
+		listen.calledWith(Events.KernelEvent.ServiceProviderBooted);
+	});
+
+	it("bootstrap: boots each registered listener (truncating its table)", async ({
+		listeners,
+		apiNodeRepo,
+		contractRepo,
+		peerRepo,
+		pluginRepo,
+	}) => {
+		const clk = clock();
+		const clears = [apiNodeRepo, contractRepo, peerRepo, pluginRepo].map((repo) => spy(repo, "clear"));
+
+		await listeners.register();
+		await listeners.bootstrap();
+		await clk.nextAsync();
+
+		for (const clear of clears) {
+			clear.calledOnce();
+		}
+
+		await listeners.dispose();
+	});
+
+	it("flush: flushes each listener through the given entity manager", async (context) => {
+		const { listeners, events, peerRepo } = context;
+		const captured: Record<string, unknown> = {};
+		events.listen = (name: string, listener: unknown) => {
+			captured[name] = listener;
+		};
+
+		await listeners.register();
+
+		// Feed one peer event through the captured Peers listener, then flush through the aggregate.
+		const peers = captured[Events.PeerEvent.Added] as any;
+		await peers.handle({
+			data: { header: { blockNumber: 1 }, ip: "127.0.0.1", latency: 1, plugins: {}, port: 4002, ports: {} },
+			name: Events.PeerEvent.Added,
+		});
+
+		const upsert = spy(peerRepo, "upsert");
+		const entityManager = { marker: "entity-manager" };
+
+		await listeners.flush(entityManager as any);
+
+		upsert.calledOnce();
+		// The repository is created against the flushed entity manager, not the base data source.
+		assert.equal(context.peerFactoryArgs, [entityManager]);
+	});
+
+	it("dispose: forgets all events and clears the listener list", async ({ listeners, events }) => {
+		const forget = spy(events, "forget");
+
+		await listeners.register();
+		await listeners.dispose();
+
+		forget.calledTimes(7);
+
+		// A second dispose is a no-op because the listener list was cleared.
+		await listeners.dispose();
+		forget.calledTimes(7);
+	});
+});