refactor(api): tidy the api and shared packages (#2733)

fluxer_api/pkgs/kv_client/src/IKVProvider.ts

@@ -28,6 +28,11 @@ export interface IKVSubscription {
 	removeAllListeners(event?: 'message' | 'error'): void;
 }
 
+export interface KVPurgeBatchResult {
+	urls: Array<string>;
+	tokensConsumed: number;
+}
+
 export interface KVRateLimitResult {
 	allowed: boolean;
 	limit: number;
@@ -98,10 +103,7 @@ export interface IKVProvider {
 		maxTokens: number,
 		refillRate: number,
 		refillIntervalMs: number,
-	): Promise<{
-		urls: Array<string>;
-		tokensConsumed: number;
-	}>;
+	): Promise<KVPurgeBatchResult>;
 	pipeline(): IKVPipeline;
 	multi(): IKVPipeline;
 	isClustered(): boolean;

fluxer_api/pkgs/kv_client/src/KVClient.ts

@@ -1,14 +1,21 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {createHash} from 'node:crypto';
-import type {IKVPipeline, IKVProvider, IKVSubscription, KVRateLimitResult} from '@pkgs/kv_client/src/IKVProvider';
+import type {
+	IKVPipeline,
+	IKVProvider,
+	IKVSubscription,
+	KVPurgeBatchResult,
+	KVRateLimitResult,
+} from '@pkgs/kv_client/src/IKVProvider';
 import {
 	type IKVLogger,
 	type KVClientConfig,
 	type ResolvedKVClientConfig,
 	resolveKVClientConfig,
 } from '@pkgs/kv_client/src/KVClientConfig';
-import {KVClientError, KVClientErrorCode} from '@pkgs/kv_client/src/KVClientError';
+import {createInvalidResponseError, KVClientError, KVClientErrorCode} from '@pkgs/kv_client/src/KVClientError';
+import {resolveKVClusterConnection} from '@pkgs/kv_client/src/KVClusterConnection';
 import {
 	createStringEntriesFromPairs,
 	createZSetMembersFromScorePairs,
@@ -21,6 +28,30 @@ import {KVPipeline} from '@pkgs/kv_client/src/KVPipeline';
 import {KVSubscription} from '@pkgs/kv_client/src/KVSubscription';
 import Redis, {Cluster} from 'ioredis';
 
+const MAX_DATE_TIMESTAMP = 8640000000000000;
+
+const DECODE_BUCKET_STATE_SCRIPT = `
+local function decodeBucketState(rawState, amountField, timestampField)
+	local ok, state = pcall(cjson.decode, rawState)
+	if not ok or type(state) ~= 'table' then
+		error('invalid stored bucket state')
+	end
+	local amount = state[amountField]
+	if type(amount) ~= 'number' or amount ~= amount or amount < 0 or amount == math.huge then
+		error('invalid stored bucket amount: ' .. amountField)
+	end
+	local timestamp = state[timestampField]
+	if type(timestamp) ~= 'number'
+		or timestamp ~= timestamp
+		or math.abs(timestamp) > ${MAX_DATE_TIMESTAMP}
+		or timestamp ~= math.floor(timestamp)
+	then
+		error('invalid stored bucket timestamp: ' .. timestampField)
+	end
+	return amount, timestamp
+end
+`;
+
 const RELEASE_LOCK_SCRIPT = `
 if redis.call('GET', KEYS[1]) == ARGV[1] then
 	return redis.call('DEL', KEYS[1])
@@ -42,6 +73,7 @@ end
 return 0
 `;
 const TRY_CONSUME_TOKENS_SCRIPT = `
+${DECODE_BUCKET_STATE_SCRIPT}
 local key = KEYS[1]
 local now = tonumber(ARGV[1])
 local requested = tonumber(ARGV[2])
@@ -54,11 +86,7 @@ local tokens = maxTokens
 local lastRefill = now
 
 if data then
-	local ok, bucket = pcall(cjson.decode, data)
-	if ok and bucket then
-		tokens = tonumber(bucket.tokens) or maxTokens
-		lastRefill = tonumber(bucket.lastRefill) or now
-	end
+	tokens, lastRefill = decodeBucketState(data, 'tokens', 'lastRefill')
 end
 
 local elapsed = now - lastRefill
@@ -84,6 +112,7 @@ redis.call('SET', key, cjson.encode({tokens = tokens, lastRefill = lastRefill}),
 return consumed
 `;
 const CHECK_LEAKY_BUCKET_LIMIT_SCRIPT = `
+${DECODE_BUCKET_STATE_SCRIPT}
 local key = KEYS[1]
 local nowMs = tonumber(ARGV[1])
 local limit = tonumber(ARGV[2])
@@ -107,11 +136,7 @@ local updatedAt = nowMs
 
 local rawState = redis.call('GET', key)
 if rawState then
-	local ok, state = pcall(cjson.decode, rawState)
-	if ok and state then
-		level = tonumber(state.level) or 0
-		updatedAt = tonumber(state.updatedAt) or nowMs
-	end
+	level, updatedAt = decodeBucketState(rawState, 'level', 'updatedAt')
 end
 
 local elapsed = nowMs - updatedAt
@@ -181,6 +206,7 @@ redis.call('SET', KEYS[2], ARGV[2])
 return 1
 `;
 const DEQUEUE_PURGE_BATCH_SCRIPT = `
+${DECODE_BUCKET_STATE_SCRIPT}
 local queueKey = KEYS[1]
 local bucketKey = KEYS[2]
 local maxItems = tonumber(ARGV[1])
@@ -191,18 +217,14 @@ local refillIntervalMs = tonumber(ARGV[5])
 
 local queueSize = redis.call('SCARD', queueKey)
 if queueSize == 0 then
-	return cjson.encode({urls = {}, tokens = 0})
+	return '{"urls":[],"tokens":0}'
 end
 
 local tokens = maxTokens
 local lastRefill = now
 local data = redis.call('GET', bucketKey)
 if data then
-	local ok, bucket = pcall(cjson.decode, data)
-	if ok and bucket then
-		tokens = tonumber(bucket.tokens) or maxTokens
-		lastRefill = tonumber(bucket.lastRefill) or now
-	end
+	tokens, lastRefill = decodeBucketState(data, 'tokens', 'lastRefill')
 end
 
 local elapsed = now - lastRefill
@@ -215,7 +237,7 @@ end
 local toPop = math.min(maxItems, math.floor(tokens), queueSize)
 if toPop <= 0 then
 	redis.call('SET', bucketKey, cjson.encode({tokens = tokens, lastRefill = lastRefill}), 'EX', 3600)
-	return cjson.encode({urls = {}, tokens = 0})
+	return '{"urls":[],"tokens":0}'
 end
 
 local urls = redis.call('SPOP', queueKey, toPop)
@@ -252,17 +274,13 @@ return 1
 
 const SCRIPT_SHA_CACHE = new Map<string, string>();
 
-interface ScriptPurgeBatchResult {
-	urls: Array<string>;
-	tokens: number;
-}
-
 export class KVClient implements IKVProvider {
 	private readonly client: Redis | Cluster;
 	private readonly config: ResolvedKVClientConfig;
 	private readonly logger: IKVLogger;
 	private readonly url: string;
 	private readonly timeoutMs: number;
+	private closed = false;
 
 	constructor(config: KVClientConfig | string) {
 		const resolvedConfig = resolveKVClientConfig(config);
@@ -283,13 +301,13 @@ export class KVClient implements IKVProvider {
 	}
 
 	private createClusterClient(clusterConfig: ResolvedKVClientConfig): Cluster {
-		const nodes =
-			clusterConfig.clusterNodes.length > 0 ? clusterConfig.clusterNodes : parseClusterNodesFromUrl(clusterConfig.url);
+		const {nodes, redisOptions} = resolveKVClusterConnection(clusterConfig.url, clusterConfig.clusterNodes);
 		const natMap = clusterConfig.clusterNatMap;
 		const hasNatMap = Object.keys(natMap).length > 0;
 		return new Cluster(nodes, {
 			clusterRetryStrategy: createRetryStrategy(),
 			redisOptions: {
+				...redisOptions,
 				connectTimeout: clusterConfig.timeoutMs,
 				commandTimeout: clusterConfig.timeoutMs,
 				maxRetriesPerRequest: 1,
@@ -299,11 +317,19 @@ export class KVClient implements IKVProvider {
 		});
 	}
 
+	close(): void {
+		if (this.closed) {
+			return;
+		}
+		this.closed = true;
+		this.client.disconnect(false);
+	}
+
 	async health(): Promise<boolean> {
 		try {
 			return (await this.execute('health', async () => this.client.ping())) === 'PONG';
 		} catch (error) {
-			this.logger.debug({url: this.url, error}, 'KV health check failed');
+			this.logger.debug({error}, 'KV health check failed');
 			return false;
 		}
 	}
@@ -314,45 +340,38 @@ export class KVClient implements IKVProvider {
 
 	async set(key: string, value: string, ...args: Array<string | number>): Promise<string | null> {
 		const options = parseSetArguments(args);
-		if (options.useNx) {
-			if (options.ttlSeconds !== undefined) {
-				const ttlSeconds = options.ttlSeconds;
-				return await this.execute('set', async () => {
-					const result = await this.client.call('SET', key, value, 'EX', ttlSeconds, 'NX');
-					return normalizeStringOrNull(result);
-				});
-			}
-			return await this.execute('set', async () => {
-				const result = await this.client.call('SET', key, value, 'NX');
-				return normalizeStringOrNull(result);
-			});
-		}
+		const modifiers: Array<string | number> = [];
 		if (options.ttlSeconds !== undefined) {
-			const ttlSeconds = options.ttlSeconds;
-			return await this.execute('set', async () => {
-				const result = await this.client.call('SET', key, value, 'EX', ttlSeconds);
-				return normalizeStringOrNull(result);
-			});
+			modifiers.push('EX', options.ttlSeconds);
 		}
-		return await this.execute('set', async () => this.client.set(key, value));
+		if (options.useNx) {
+			modifiers.push('NX');
+		}
+		return await this.execute('set', async () => {
+			const result =
+				modifiers.length === 0
+					? await this.client.set(key, value)
+					: await this.client.call('SET', key, value, ...modifiers);
+			return parseSetReply(result, 'set', options.useNx);
+		});
 	}
 
 	async setex(key: string, ttlSeconds: number, value: string): Promise<void> {
 		await this.execute('setex', async () => {
-			await this.client.setex(key, ttlSeconds, value);
+			parseSetReply(await this.client.setex(key, ttlSeconds, value), 'setex');
 		});
 	}
 
 	async setnx(key: string, value: string, ttlSeconds?: number): Promise<boolean> {
 		if (ttlSeconds !== undefined) {
-			const ttlSecondsValue = ttlSeconds;
 			const result = await this.execute('setnx', async () => {
-				const commandResult = await this.client.call('SET', key, value, 'EX', ttlSecondsValue, 'NX');
-				return normalizeStringOrNull(commandResult);
+				const commandResult = await this.client.call('SET', key, value, 'EX', ttlSeconds, 'NX');
+				return parseSetReply(commandResult, 'setnx', true);
 			});
 			return result === 'OK';
 		}
-		return (await this.execute('setnx', async () => this.client.setnx(key, value))) === 1;
+		const result = await this.execute('setnx', async () => this.client.setnx(key, value));
+		return parseIntegerDecision(result, 'setnx');
 	}
 
 	async mget(...keys: Array<string>): Promise<Array<string | null>> {
@@ -360,15 +379,18 @@ export class KVClient implements IKVProvider {
 			return [];
 		}
 		return await this.execute('mget', async () => {
-			const values = new Array<string | null>(keys.length).fill(null);
+			const values = new Array<string | null>(keys.length);
 			const batches = this.splitBySlot(
 				keys.map((key, index) => ({key, index})),
 				(entry) => entry.key,
 			);
 			await runSlotBatches(batches, async (batch) => {
 				const batchValues = await this.client.mget(...batch.map((entry) => entry.key));
+				if (!Array.isArray(batchValues) || batchValues.length !== batch.length) {
+					throw createInvalidResponseError('mget', `${batch.length} values`);
+				}
 				for (const [position, entry] of batch.entries()) {
-					values[entry.index] = batchValues[position] ?? null;
+					values[entry.index] = parseNullableStringReply(batchValues[position], 'mget');
 				}
 			});
 			return values;
@@ -383,7 +405,8 @@ export class KVClient implements IKVProvider {
 		await this.execute('mset', async () => {
 			const batches = this.splitBySlot(entries, (entry) => entry.key);
 			await runSlotBatches(batches, async (batch) => {
-				await this.client.mset(...batch.flatMap((entry) => [entry.key, entry.value]));
+				const result = await this.client.mset(...batch.flatMap((entry) => [entry.key, entry.value]));
+				parseSetReply(result, 'mset');
 			});
 		});
 	}
@@ -421,14 +444,14 @@ export class KVClient implements IKVProvider {
 	async getex(key: string, ttlSeconds: number): Promise<string | null> {
 		return await this.execute('getex', async () => {
 			const result = await this.client.call('GETEX', key, 'EX', ttlSeconds);
-			return normalizeStringOrNull(result);
+			return parseNullableStringReply(result, 'getex');
 		});
 	}
 
 	async getdel(key: string): Promise<string | null> {
 		return await this.execute('getdel', async () => {
 			const result = await this.client.call('GETDEL', key);
-			return normalizeStringOrNull(result);
+			return parseNullableStringReply(result, 'getdel');
 		});
 	}
 
@@ -464,10 +487,7 @@ export class KVClient implements IKVProvider {
 		}
 		return await this.execute('spop', async () => {
 			const result = await this.client.spop(key, count);
-			if (result === null) {
-				return [];
-			}
-			return Array.isArray(result) ? result : [result];
+			return parsePoppedValues(result, 'spop', count);
 		});
 	}
 
@@ -522,16 +542,9 @@ export class KVClient implements IKVProvider {
 		}
 		return await this.execute('lpop', async () => {
 			if (count !== undefined) {
-				const result = await this.client.call('LPOP', key, count);
-				if (result === null) {
-					return [];
-				}
-				if (Array.isArray(result)) {
-					return result.map((entry) => String(entry));
-				}
-				return [String(result)];
+				return parsePoppedValues(await this.client.lpop(key, count), 'lpop', count);
 			}
-			const single = await this.client.lpop(key);
+			const single = parseNullableStringReply(await this.client.lpop(key), 'lpop');
 			return single === null ? [] : [single];
 		});
 	}
@@ -564,6 +577,7 @@ export class KVClient implements IKVProvider {
 	}
 
 	duplicate(): IKVSubscription {
+		this.assertOpen();
 		return new KVSubscription({
 			url: this.url,
 			mode: this.config.mode,
@@ -576,19 +590,19 @@ export class KVClient implements IKVProvider {
 	async acquireLock(key: string, token: string, ttlSeconds: number): Promise<boolean> {
 		const result = await this.execute('acquireLock', async () => {
 			const commandResult = await this.client.call('SET', key, token, 'EX', ttlSeconds, 'NX');
-			return normalizeStringOrNull(commandResult);
+			return parseSetReply(commandResult, 'acquireLock', true);
 		});
 		return result === 'OK';
 	}
 
 	async releaseLock(key: string, token: string): Promise<boolean> {
 		const result = await this.executeScript('releaseLock', RELEASE_LOCK_SCRIPT, 1, key, token);
-		return Number(result) === 1;
+		return parseIntegerDecision(result, 'releaseLock');
 	}
 
 	async extendLock(key: string, token: string, ttlSeconds: number): Promise<boolean> {
 		const result = await this.executeScript('extendLock', EXTEND_LOCK_SCRIPT, 1, key, token, ttlSeconds);
-		return Number(result) === 1;
+		return parseIntegerDecision(result, 'extendLock');
 	}
 
 	async renewSnowflakeNode(key: string, instanceId: string, ttlSeconds: number): Promise<boolean> {
@@ -600,20 +614,21 @@ export class KVClient implements IKVProvider {
 			instanceId,
 			ttlSeconds,
 		);
-		return Number(result) === 1;
+		return parseIntegerDecision(result, 'renewSnowflakeNode');
 	}
 
 	async checkLeakyBucketLimit(key: string, limit: number, windowMs: number, cost: number): Promise<KVRateLimitResult> {
-		const now = Date.now();
-		return await this.executeRateLimitScript(
+		const result = await this.executeJsonScript(
 			'checkLeakyBucketLimit',
 			CHECK_LEAKY_BUCKET_LIMIT_SCRIPT,
+			1,
 			key,
-			now,
+			Date.now(),
 			limit,
 			windowMs,
 			cost,
 		);
+		return parseRateLimitResult(result);
 	}
 
 	async tryConsumeTokens(
@@ -635,11 +650,14 @@ export class KVClient implements IKVProvider {
 			refillRate,
 			refillIntervalMs,
 		);
-		return Number(result);
+		if (!isNonNegativeSafeInteger(result) || result > requested) {
+			throw createInvalidResponseError('tryConsumeTokens', 'an integer token count within the requested amount');
+		}
+		return result;
 	}
 
 	async scheduleBulkDeletion(queueKey: string, secondaryKey: string, score: number, value: string): Promise<void> {
-		await this.executeScript(
+		const result = await this.executeScript(
 			'scheduleBulkDeletion',
 			SCHEDULE_BULK_DELETION_SCRIPT,
 			2,
@@ -648,6 +666,7 @@ export class KVClient implements IKVProvider {
 			score,
 			value,
 		);
+		if (result !== 1) throw createInvalidResponseError('scheduleBulkDeletion', 'the acknowledgement 1');
 	}
 
 	async claimBulkDeletion(queueKey: string, member: string, maxScore: number, leaseScore: number): Promise<boolean> {
@@ -660,7 +679,7 @@ export class KVClient implements IKVProvider {
 			maxScore,
 			leaseScore,
 		);
-		return Number(result) === 1;
+		return parseIntegerDecision(result, 'claimBulkDeletion');
 	}
 
 	async removeBulkDeletion(queueKey: string, secondaryKey: string, member = ''): Promise<boolean> {
@@ -672,7 +691,7 @@ export class KVClient implements IKVProvider {
 			secondaryKey,
 			member,
 		);
-		return Number(result) === 1;
+		return parseIntegerDecision(result, 'removeBulkDeletion');
 	}
 
 	async dequeuePurgeBatch(
@@ -682,12 +701,9 @@ export class KVClient implements IKVProvider {
 		maxTokens: number,
 		refillRate: number,
 		refillIntervalMs: number,
-	): Promise<{
-		urls: Array<string>;
-		tokensConsumed: number;
-	}> {
+	): Promise<KVPurgeBatchResult> {
 		const now = Date.now();
-		const parsed = await this.executeJsonScript<ScriptPurgeBatchResult>(
+		const result = await this.executeJsonScript(
 			'dequeuePurgeBatch',
 			DEQUEUE_PURGE_BATCH_SCRIPT,
 			2,
@@ -699,7 +715,7 @@ export class KVClient implements IKVProvider {
 			refillRate,
 			refillIntervalMs,
 		);
-		return {urls: parsed.urls, tokensConsumed: parsed.tokens};
+		return parsePurgeBatchResult(result, maxItems);
 	}
 
 	async evalScript(
@@ -738,7 +754,10 @@ export class KVClient implements IKVProvider {
 
 	pipeline(): IKVPipeline {
 		return new KVPipeline({
-			createCommander: () => this.client.pipeline(),
+			createCommander: () => {
+				this.assertOpen();
+				return this.client.pipeline();
+			},
 			normalizeError: (command, error) => this.normalizeError(command, error),
 			mode: 'pipeline',
 		});
@@ -746,7 +765,10 @@ export class KVClient implements IKVProvider {
 
 	multi(): IKVPipeline {
 		return new KVPipeline({
-			createCommander: () => this.client.multi(),
+			createCommander: () => {
+				this.assertOpen();
+				return this.client.multi();
+			},
 			normalizeError: (command, error) => this.normalizeError(command, error),
 			mode: 'multi',
 		});
@@ -754,12 +776,19 @@ export class KVClient implements IKVProvider {
 
 	private async execute<T>(command: string, fn: () => Promise<T>): Promise<T> {
 		try {
+			this.assertOpen();
 			return await fn();
 		} catch (error) {
 			throw this.normalizeError(command, error);
 		}
 	}
 
+	private assertOpen(): void {
+		if (this.closed) {
+			throw new Error('KV client is closed');
+		}
+	}
+
 	private async executeScript(
 		command: string,
 		script: string,
@@ -780,43 +809,21 @@ export class KVClient implements IKVProvider {
 		}
 	}
 
-	private async executeJsonScript<T>(
+	private async executeJsonScript(
 		command: string,
 		script: string,
 		keyCount: number,
 		...args: Array<string | number>
-	): Promise<T> {
+	): Promise<unknown> {
 		const result = await this.executeScript(command, script, keyCount, ...args);
+		if (typeof result !== 'string') {
+			throw createInvalidResponseError(command, 'a JSON string');
+		}
 		try {
-			return JSON.parse(String(result)) as T;
-		} catch (error) {
-			throw new KVClientError({
-				code: KVClientErrorCode.INVALID_RESPONSE,
-				message: `KV request returned invalid JSON (${command}): ${getErrorMessage(error)}`,
-			});
+			return JSON.parse(result);
+		} catch {
+			throw createInvalidResponseError(command, 'valid JSON');
 		}
-	}
-
-	private async executeRateLimitScript(
-		command: string,
-		script: string,
-		key: string,
-		nowMs: number,
-		limit: number,
-		windowMs: number,
-		cost: number,
-	): Promise<KVRateLimitResult> {
-		const parsed = await this.executeJsonScript<KVRateLimitResult>(
-			command,
-			script,
-			1,
-			key,
-			nowMs,
-			limit,
-			windowMs,
-			cost,
-		);
-		return normalizeRateLimitResult(parsed);
 	}
 
 	private normalizeError(command: string, error: unknown): KVClientError {
@@ -836,18 +843,6 @@ export class KVClient implements IKVProvider {
 	}
 }
 
-function parseClusterNodesFromUrl(url: string): Array<{
-	host: string;
-	port: number;
-}> {
-	try {
-		const parsed = new URL(url);
-		return [{host: parsed.hostname, port: Number.parseInt(parsed.port || '6379', 10)}];
-	} catch {
-		return [{host: '127.0.0.1', port: 6379}];
-	}
-}
-
 function createRetryStrategy(): (times: number) => number {
 	return (times: number) => {
 		const backoffMs = Math.min(times * 100, 2000);
@@ -855,22 +850,77 @@ function createRetryStrategy(): (times: number) => number {
 	};
 }
 
-function normalizeStringOrNull(value: unknown): string | null {
-	if (value === null || value === undefined) {
-		return null;
-	}
-	return String(value);
+function parseNullableStringReply(value: unknown, command: string): string | null {
+	if (value === null || typeof value === 'string') return value;
+	throw createInvalidResponseError(command, 'a string or null');
 }
 
-function normalizeRateLimitResult(result: KVRateLimitResult): KVRateLimitResult {
-	return {
-		allowed: Boolean(result.allowed),
-		limit: Number(result.limit),
-		remaining: Number(result.remaining),
-		resetAfterMs: Number(result.resetAfterMs),
-		resetAtMs: Number(result.resetAtMs),
-		retryAfterMs: Number(result.retryAfterMs),
-	};
+function parseSetReply(value: unknown, command: string, conditional = false): 'OK' | null {
+	if (value === 'OK' || (conditional && value === null)) return value;
+	throw createInvalidResponseError(command, conditional ? 'OK or null' : 'OK');
+}
+
+function parsePoppedValues(value: unknown, command: string, count: number): Array<string> {
+	if (value === null) return [];
+	if (!Array.isArray(value) || value.length > count) {
+		throw createInvalidResponseError(command, 'a string array within the requested count or null');
+	}
+	for (const entry of value) {
+		if (typeof entry !== 'string') throw createInvalidResponseError(command, 'string array entries');
+	}
+	return value;
+}
+
+function parseIntegerDecision(value: unknown, command: string): boolean {
+	if (value === 0 || value === 1) return value === 1;
+	throw createInvalidResponseError(command, 'the integer 0 or 1');
+}
+
+function parseRateLimitResult(value: unknown): KVRateLimitResult {
+	const command = 'checkLeakyBucketLimit';
+	if (!isJsonObject(value)) throw createInvalidResponseError(command, 'a rate-limit result object');
+	const {allowed, limit, remaining, resetAfterMs, resetAtMs, retryAfterMs} = value;
+	if (
+		typeof allowed !== 'boolean' ||
+		!isNonNegativeSafeInteger(limit) ||
+		limit === 0 ||
+		!isNonNegativeSafeInteger(remaining) ||
+		remaining > limit ||
+		!isNonNegativeSafeInteger(resetAfterMs) ||
+		typeof resetAtMs !== 'number' ||
+		!Number.isSafeInteger(resetAtMs) ||
+		Math.abs(resetAtMs) > MAX_DATE_TIMESTAMP ||
+		!isNonNegativeSafeInteger(retryAfterMs) ||
+		retryAfterMs > resetAfterMs ||
+		(allowed ? retryAfterMs !== 0 : remaining !== 0 || retryAfterMs === 0)
+	) {
+		throw createInvalidResponseError(command, 'a boolean decision and valid integer rate-limit counts and times');
+	}
+	return {allowed, limit, remaining, resetAfterMs, resetAtMs, retryAfterMs};
+}
+
+function parsePurgeBatchResult(value: unknown, maxItems: number): KVPurgeBatchResult {
+	const command = 'dequeuePurgeBatch';
+	if (!isJsonObject(value)) throw createInvalidResponseError(command, 'a purge batch object');
+	const {urls, tokens} = value;
+	if (
+		!Array.isArray(urls) ||
+		!urls.every((url): url is string => typeof url === 'string') ||
+		!isNonNegativeSafeInteger(tokens) ||
+		tokens !== urls.length ||
+		urls.length > maxItems
+	) {
+		throw createInvalidResponseError(command, 'a bounded string array and matching token count');
+	}
+	return {urls, tokensConsumed: tokens};
+}
+
+function isJsonObject(value: unknown): value is Record<string, unknown> {
+	return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
+function isNonNegativeSafeInteger(value: unknown): value is number {
+	return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0;
 }
 
 function getScriptSha(script: string): string {

fluxer_api/pkgs/kv_client/src/KVClientConfig.ts

@@ -38,23 +38,14 @@ const noopLogger: IKVLogger = {
 };
 
 export function resolveKVClientConfig(config: KVClientConfig | string): ResolvedKVClientConfig {
-	if (typeof config === 'string') {
-		return {
-			url: normalizeUrl(config),
-			mode: 'standalone' as const,
-			clusterNodes: [],
-			clusterNatMap: {},
-			timeoutMs: DEFAULT_KV_TIMEOUT_MS,
-			logger: noopLogger,
-		};
-	}
+	const options: KVClientConfig = typeof config === 'string' ? {url: config} : config;
 	return {
-		url: normalizeUrl(config.url),
-		mode: config.mode ?? 'standalone',
-		clusterNodes: config.clusterNodes ?? [],
-		clusterNatMap: config.clusterNatMap ?? {},
-		timeoutMs: config.timeoutMs ?? DEFAULT_KV_TIMEOUT_MS,
-		logger: config.logger ?? noopLogger,
+		url: normalizeUrl(options.url),
+		mode: options.mode ?? 'standalone',
+		clusterNodes: options.clusterNodes ?? [],
+		clusterNatMap: options.clusterNatMap ?? {},
+		timeoutMs: options.timeoutMs ?? DEFAULT_KV_TIMEOUT_MS,
+		logger: options.logger ?? noopLogger,
 	};
 }
 
@@ -63,5 +54,9 @@ function normalizeUrl(url: string): string {
 	if (trimmed.length === 0) {
 		throw new Error('KV client URL must not be empty');
 	}
+	const query = new URLSearchParams(trimmed.match(/^[^?#]*(\?[^#]*)/)?.[1]);
+	if (query.has('stringNumbers')) {
+		throw new Error('KV client URLs do not support stringNumbers; numeric replies are required');
+	}
 	return trimmed;
 }

fluxer_api/pkgs/kv_client/src/KVClientError.ts

@@ -26,3 +26,10 @@ export class KVClientError extends FluxerError {
 		this.name = 'KVClientError';
 	}
 }
+
+export function createInvalidResponseError(command: string, expected: string): KVClientError {
+	return new KVClientError({
+		code: KVClientErrorCode.INVALID_RESPONSE,
+		message: `KV request returned an invalid response (${command}): expected ${expected}`,
+	});
+}

fluxer_api/pkgs/kv_client/src/KVClusterConnection.ts

@@ -0,0 +1,130 @@
+import {domainToASCII} from 'node:url';
+import type {KVClusterNode} from '@pkgs/kv_client/src/KVClientConfig';
+import type {RedisOptions} from 'ioredis';
+
+interface KVClusterConnection {
+	nodes: Array<KVClusterNode>;
+	redisOptions: RedisOptions;
+}
+
+export function resolveKVClusterConnection(url: string, nodes: ReadonlyArray<KVClusterNode>): KVClusterConnection {
+	const normalizedUrl = url.trim();
+	for (let index = 0; index < normalizedUrl.length; index++) {
+		const code = normalizedUrl.charCodeAt(index);
+		if (code <= 0x1f || code === 0x7f) {
+			throw new Error('KV cluster URL must not contain control characters');
+		}
+	}
+	const parsed = parseClusterUrl(normalizedUrl);
+	const authorityStart = normalizedUrl.indexOf('://');
+	if (authorityStart === -1) {
+		throw new Error('Invalid KV cluster URL');
+	}
+	const authorityAndPath = normalizedUrl.slice(authorityStart + 3).split(/[?#]/, 1)[0];
+	const pathStart = authorityAndPath.indexOf('/');
+	const authority = pathStart === -1 ? authorityAndPath : authorityAndPath.slice(0, pathStart);
+	const path = pathStart === -1 ? '' : authorityAndPath.slice(pathStart + 1);
+	const queryNames = new Set<string>();
+	for (const name of parsed.searchParams.keys()) {
+		if (name !== 'username' && name !== 'password' && name !== 'db') {
+			throw new Error('Unsupported KV cluster URL option');
+		}
+		if (queryNames.has(name)) {
+			throw new Error('Duplicate KV cluster URL option');
+		}
+		queryNames.add(name);
+	}
+	if (path.length > 0) {
+		validateClusterDatabase(path);
+	}
+	const database = parsed.searchParams.get('db');
+	if (database !== null) {
+		validateClusterDatabase(database);
+	}
+	const redisOptions = resolveClusterAuthentication(authority, parsed.searchParams);
+	if (parsed.protocol === 'rediss:') {
+		redisOptions.tls = {};
+	}
+	const host = resolveClusterHost(authority, parsed);
+	const resolvedNodes = nodes.length > 0 ? [...nodes] : [{host, port: Number(parsed.port || '6379')}];
+	for (const node of resolvedNodes) {
+		if (node.host.trim().length === 0) {
+			throw new Error('KV cluster node must include a host');
+		}
+		if (!Number.isInteger(node.port) || node.port < 1 || node.port > 65535) {
+			throw new Error('KV cluster node port must be an integer between 1 and 65535');
+		}
+	}
+	return {
+		nodes: resolvedNodes,
+		redisOptions,
+	};
+}
+
+function parseClusterUrl(url: string): URL {
+	let parsed: URL;
+	try {
+		parsed = new URL(url);
+	} catch {
+		throw new Error('Invalid KV cluster URL');
+	}
+	if (parsed.protocol !== 'redis:' && parsed.protocol !== 'rediss:') {
+		throw new Error('KV cluster URL must use redis:// or rediss://');
+	}
+	if (!parsed.hostname) {
+		throw new Error('KV cluster URL must include a host');
+	}
+	return parsed;
+}
+
+function resolveClusterHost(authority: string, parsed: URL): string {
+	if (parsed.hostname.startsWith('[')) {
+		return parsed.hostname.slice(1, -1);
+	}
+	const hostAndPort = authority.slice(authority.lastIndexOf('@') + 1);
+	const portStart = hostAndPort.lastIndexOf(':');
+	const hostname = portStart === -1 ? hostAndPort : hostAndPort.slice(0, portStart);
+	if (hostname.includes('%')) {
+		throw new Error('KV cluster URL must not contain a percent-encoded hostname');
+	}
+	const host = domainToASCII(hostname.toLowerCase());
+	if (!host) {
+		throw new Error('Invalid KV cluster URL hostname');
+	}
+	return host;
+}
+
+function validateClusterDatabase(database: string): void {
+	if (!/^0+$/.test(database)) {
+		throw new Error('KV cluster URL must select database 0');
+	}
+}
+
+function resolveClusterAuthentication(authority: string, query: URLSearchParams): RedisOptions {
+	const authEnd = authority.lastIndexOf('@');
+	let auth = '';
+	if (authEnd !== -1) {
+		try {
+			auth = decodeURIComponent(authority.slice(0, authEnd));
+		} catch {
+			throw new Error('Invalid KV cluster URL credentials');
+		}
+	}
+	if (auth) {
+		const separator = auth.indexOf(':');
+		return {
+			username: separator === -1 ? auth : auth.slice(0, separator),
+			password: separator === -1 ? '' : auth.slice(separator + 1),
+		};
+	}
+	const options: RedisOptions = {};
+	const username = query.get('username');
+	const password = query.get('password');
+	if (username !== null) {
+		options.username = username;
+	}
+	if (password !== null) {
+		options.password = password;
+	}
+	return options;
+}

fluxer_api/pkgs/kv_client/src/KVCommandArguments.ts

@@ -83,7 +83,7 @@ export function createZSetMembersFromScorePairs(scoreMembers: Array<number | str
 			throw createInvalidArgumentError('zadd member must be a string');
 		}
 		members.push({
-			score: parseFiniteNumber(rawScore, 'zadd score'),
+			score: parseNumberArgument(rawScore, 'zadd score'),
 			value: rawMember,
 		});
 	}
@@ -136,14 +136,6 @@ function parseNumberArgument(value: string | number | undefined, label: string):
 	throw createInvalidArgumentError(`${label} must be a finite number`);
 }
 
-function parseFiniteNumber(value: string | number, label: string): number {
-	const parsed = parseNumberArgument(value, label);
-	if (!Number.isFinite(parsed)) {
-		throw createInvalidArgumentError(`${label} must be finite`);
-	}
-	return parsed;
-}
-
 function parseNonNegativeInteger(value: string | number | undefined, label: string): number {
 	const parsed = parseNumberArgument(value, label);
 	if (!Number.isInteger(parsed) || parsed < 0) {

fluxer_api/pkgs/kv_client/src/KVHashSlots.ts

@@ -16,10 +16,10 @@ function extractHashTag(key: string): string {
 }
 
 export function computeHashSlot(key: string): number {
-	const hashed = extractHashTag(key);
+	const hashed = Buffer.from(extractHashTag(key));
 	let crc = 0;
-	for (let index = 0; index < hashed.length; index += 1) {
-		crc ^= (hashed.charCodeAt(index) & 0xff) << 8;
+	for (const byte of hashed) {
+		crc ^= byte << 8;
 		for (let bit = 0; bit < 8; bit += 1) {
 			crc = (crc & 0x8000) === 0 ? (crc << 1) & 0xffff : ((crc << 1) ^ 0x1021) & 0xffff;
 		}
@@ -52,17 +52,17 @@ export function splitIntoSlotBatches<T>(
 }
 
 export async function runSlotBatches<T>(batches: ReadonlyArray<T>, run: (batch: T) => Promise<void>): Promise<void> {
-	if (batches.length <= MAX_CONCURRENT_SLOT_BATCHES) {
-		await Promise.all(batches.map(async (batch) => await run(batch)));
-		return;
-	}
 	let nextIndex = 0;
-	const workers = Array.from({length: MAX_CONCURRENT_SLOT_BATCHES}, async () => {
+	const workers = Array.from({length: Math.min(batches.length, MAX_CONCURRENT_SLOT_BATCHES)}, async () => {
 		while (nextIndex < batches.length) {
 			const batch = batches[nextIndex];
 			nextIndex += 1;
 			await run(batch);
 		}
 	});
-	await Promise.all(workers);
+	for (const result of await Promise.allSettled(workers)) {
+		if (result.status === 'rejected') {
+			throw result.reason;
+		}
+	}
 }

fluxer_api/pkgs/kv_client/src/KVPipeline.ts

@@ -1,10 +1,12 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import type {IKVPipeline} from '@pkgs/kv_client/src/IKVProvider';
+import {createInvalidResponseError} from '@pkgs/kv_client/src/KVClientError';
 import {createStringEntriesFromPairs} from '@pkgs/kv_client/src/KVCommandArguments';
 import type {ChainableCommander} from 'ioredis';
 
 type PipelineExecResult = [Error | null, unknown];
+type PipelineCommandReply = [Error] | PipelineExecResult;
 
 interface KVPipelineOptions {
 	createCommander: () => ChainableCommander;
@@ -88,25 +90,27 @@ export class KVPipeline implements IKVPipeline {
 	async exec(): Promise<Array<PipelineExecResult>> {
 		const command = `${this.mode}.exec`;
 		try {
-			const rawResults = (await this.commander.exec()) as Array<PipelineExecResult> | null;
+			const commander = this.commander;
+			const expectedResults = commander.length - (this.mode === 'multi' ? 1 : 0);
 			this.commander = this.createCommander();
-			if (rawResults === null) {
-				return [];
+			const rawResults = await commander.exec();
+			if (!Array.isArray(rawResults) || rawResults.length !== expectedResults) {
+				throw createInvalidResponseError(command, `${expectedResults} command results`);
 			}
-			return rawResults.map((result: PipelineExecResult) => {
-				const [error, value] = result;
-				return [error ? normalizePipelineError(error) : null, value] as PipelineExecResult;
+			return Array.from(rawResults, (result: unknown): PipelineExecResult => {
+				if (!isPipelineCommandReply(result)) {
+					throw createInvalidResponseError(command, 'an error/value pair for every command');
+				}
+				return [result[0], result[1]];
 			});
 		} catch (error) {
-			this.commander = this.createCommander();
 			throw this.normalizeError(command, error);
 		}
 	}
 }
 
-function normalizePipelineError(error: unknown): Error {
-	if (error instanceof Error) {
-		return error;
-	}
-	return new Error(String(error));
+function isPipelineCommandReply(value: unknown): value is PipelineCommandReply {
+	if (!Array.isArray(value)) return false;
+	if (value[0] instanceof Error) return value.length === 1 || value.length === 2;
+	return value.length === 2 && value[0] === null && value[1] !== undefined;
 }

fluxer_api/pkgs/kv_client/src/KVSubscription.ts

@@ -2,7 +2,8 @@
 
 import type {IKVSubscription} from '@pkgs/kv_client/src/IKVProvider';
 import type {IKVLogger, KVClientMode, KVClusterNode} from '@pkgs/kv_client/src/KVClientConfig';
-import Redis from 'ioredis';
+import {resolveKVClusterConnection} from '@pkgs/kv_client/src/KVClusterConnection';
+import Redis, {type RedisOptions} from 'ioredis';
 
 interface KVSubscriptionConfig {
 	url: string;
@@ -12,16 +13,23 @@ interface KVSubscriptionConfig {
 	logger: IKVLogger;
 }
 
+interface KVSubscriptionConnect {
+	completion: Promise<void>;
+	controller: AbortController;
+}
+
 export class KVSubscription implements IKVSubscription {
 	private readonly url: string;
 	private readonly mode: KVClientMode;
 	private readonly clusterNodes: Array<KVClusterNode>;
 	private readonly timeoutMs: number;
 	private readonly logger: IKVLogger;
-	private readonly channels: Set<string> = new Set();
+	private readonly desiredChannels = new Set<string>();
 	private readonly messageCallbacks: Set<(channel: string, message: string) => void> = new Set();
 	private readonly errorCallbacks: Set<(error: Error) => void> = new Set();
 	private client: Redis | null = null;
+	private connecting: KVSubscriptionConnect | null = null;
+	private closing: Promise<void> | null = null;
 
 	constructor(config: KVSubscriptionConfig) {
 		this.url = config.url;
@@ -32,31 +40,125 @@ export class KVSubscription implements IKVSubscription {
 	}
 
 	async connect(): Promise<void> {
-		if (this.client !== null) {
+		this.assertNotClosing();
+		if (this.connecting !== null) {
+			return await this.connecting.completion;
+		}
+		if (this.client?.status === 'ready') {
 			return;
 		}
-		const connectionUrl = this.resolveSubscriptionUrl();
-		const client = new Redis(connectionUrl, {
+		if (this.client === null || this.client.status === 'end') {
+			this.client = this.createClient();
+		}
+		const controller = new AbortController();
+		const connecting: KVSubscriptionConnect = {
+			completion: this.connectClient(this.client, controller.signal),
+			controller,
+		};
+		this.connecting = connecting;
+		try {
+			await connecting.completion;
+		} finally {
+			controller.abort();
+			if (this.connecting === connecting) {
+				this.connecting = null;
+			}
+		}
+	}
+
+	private createClient(): Redis {
+		const options: RedisOptions = {
 			autoResubscribe: true,
 			connectTimeout: this.timeoutMs,
 			commandTimeout: this.timeoutMs,
 			maxRetriesPerRequest: 1,
 			retryStrategy: createRetryStrategy(),
-		});
+		};
+		const connection = this.mode === 'cluster' ? resolveKVClusterConnection(this.url, this.clusterNodes) : null;
+		const client = connection
+			? new Redis({...connection.redisOptions, ...connection.nodes[0], db: 0, ...options})
+			: new Redis(this.url, options);
 		client.on('message', (channel: string, message: string) => {
+			if (this.client !== client || this.closing !== null) {
+				return;
+			}
 			for (const callback of this.messageCallbacks) {
 				callback(channel, message);
 			}
 		});
 		client.on('error', (error: Error) => {
+			if (this.client !== client || this.closing !== null) {
+				return;
+			}
 			this.logger.error({error}, 'KV subscription error');
 			for (const callback of this.errorCallbacks) {
 				callback(error);
 			}
 		});
-		this.client = client;
-		if (this.channels.size > 0) {
-			await this.client.subscribe(...Array.from(this.channels));
+		return client;
+	}
+
+	private async connectClient(client: Redis, signal: AbortSignal): Promise<void> {
+		try {
+			const ready = this.waitForReady(client, signal);
+			if (client.status === 'wait') {
+				await Promise.all([ready, client.connect()]);
+			} else {
+				await ready;
+			}
+			this.assertCurrentClient(client);
+			if (this.desiredChannels.size > 0) {
+				await client.subscribe(...this.desiredChannels);
+				this.assertCurrentClient(client);
+			}
+		} catch (error) {
+			if (this.client === client && this.closing === null) {
+				this.client = null;
+				client.disconnect(false);
+			}
+			throw error;
+		}
+	}
+
+	private waitForReady(client: Redis, signal: AbortSignal): Promise<void> {
+		if (signal.aborted) {
+			return Promise.reject(signal.reason);
+		}
+		if (client.status === 'ready') {
+			return Promise.resolve();
+		}
+		return new Promise((resolve, reject) => {
+			const finish = (error?: unknown): void => {
+				clearTimeout(timer);
+				client.off('ready', ready);
+				client.off('end', ended);
+				signal.removeEventListener('abort', aborted);
+				if (error === undefined) {
+					resolve();
+				} else {
+					reject(error);
+				}
+			};
+			const ready = (): void => finish();
+			const ended = (): void => finish(new Error('KV subscription connection was closed'));
+			const aborted = (): void => finish(signal.reason);
+			const timer = setTimeout(() => finish(new Error('KV subscription connection timed out')), this.timeoutMs);
+			timer.unref();
+			client.once('ready', ready);
+			client.once('end', ended);
+			signal.addEventListener('abort', aborted, {once: true});
+		});
+	}
+
+	private assertNotClosing(): void {
+		if (this.closing !== null) {
+			throw new Error('KV subscription is closing');
+		}
+	}
+
+	private assertCurrentClient(client: Redis): void {
+		if (this.client !== client || this.closing !== null) {
+			throw new Error('KV subscription connection was closed');
 		}
 	}
 
@@ -87,39 +189,72 @@ export class KVSubscription implements IKVSubscription {
 	}
 
 	async subscribe(...channels: Array<string>): Promise<void> {
-		const newChannels = channels.filter((channel) => {
-			if (this.channels.has(channel)) {
-				return false;
-			}
-			this.channels.add(channel);
-			return true;
-		});
-		if (newChannels.length === 0 || this.client === null) {
+		this.assertNotClosing();
+		const requestedChannels = [...new Set(channels)];
+		for (const channel of requestedChannels) {
+			this.desiredChannels.add(channel);
+		}
+		const client = this.client;
+		if (requestedChannels.length === 0 || client === null) {
 			return;
 		}
-		await this.client.subscribe(...newChannels);
+		await client.subscribe(...requestedChannels);
+		this.assertCurrentClient(client);
 	}
 
 	async unsubscribe(...channels: Array<string>): Promise<void> {
-		const removedChannels = channels.filter((channel) => this.channels.delete(channel));
-		if (removedChannels.length === 0 || this.client === null) {
+		this.assertNotClosing();
+		const requestedChannels = [...new Set(channels)];
+		for (const channel of requestedChannels) {
+			this.desiredChannels.delete(channel);
+		}
+		const client = this.client;
+		if (requestedChannels.length === 0 || client === null) {
 			return;
 		}
-		await this.client.unsubscribe(...removedChannels);
+		await client.unsubscribe(...requestedChannels);
+		this.assertCurrentClient(client);
 	}
 
 	async quit(): Promise<void> {
+		if (this.closing !== null) {
+			return await this.closing;
+		}
 		const client = this.client;
-		this.client = null;
 		if (client === null) {
 			return;
 		}
-		await client.quit();
+		const closing = Promise.resolve().then(() => this.closeClient(client));
+		this.closing = closing;
+		this.connecting?.controller.abort(new Error('KV subscription connection was closed'));
+		try {
+			await closing;
+		} finally {
+			if (this.client === client) {
+				this.client = null;
+			}
+			this.connecting = null;
+			if (this.closing === closing) {
+				this.closing = null;
+			}
+		}
+	}
+
+	private async closeClient(client: Redis): Promise<void> {
+		try {
+			if (this.client === client && client.status !== 'end') {
+				await client.quit();
+			}
+		} finally {
+			client.disconnect(false);
+		}
 	}
 
 	async disconnect(): Promise<void> {
 		const client = this.client;
 		this.client = null;
+		this.connecting?.controller.abort(new Error('KV subscription connection was closed'));
+		this.connecting = null;
 		if (client === null) {
 			return;
 		}
@@ -134,19 +269,8 @@ export class KVSubscription implements IKVSubscription {
 			this.errorCallbacks.clear();
 		}
 	}
-
-	private resolveSubscriptionUrl(): string {
-		if (this.mode !== 'cluster' || this.clusterNodes.length === 0) {
-			return this.url;
-		}
-		const node = this.clusterNodes[0];
-		return `redis://${node.host}:${node.port}`;
-	}
 }
 
 function createRetryStrategy(): (times: number) => number {
-	return (times: number) => {
-		const backoffMs = Math.min(times * 100, 2000);
-		return backoffMs;
-	};
+	return (times) => Math.min(times * 100, 2000);
 }