Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 1x 1x 1x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x | import type {Adapter, IMeta} from '@feasibleone/blong/types';
import {adapter, type Errors, type IErrorMap} from '@feasibleone/blong/types';
import Redis, {Cluster} from 'ioredis';
export interface IConfig {
/**
* Cluster topology: when true, connect to `nodes` via ioredis Cluster.
* Generic key/hash/script operations work the same in both topologies.
*/
cluster?: boolean;
/** Cluster seed nodes (used when `cluster: true`). */
nodes?: Array<{host: string; port: number}>;
/** Single-node host (used when `cluster: false`). */
host?: string;
/** Single-node port (used when `cluster: false`). */
port?: number;
password?: string;
db?: number;
keyPrefix?: string;
maxRetriesPerRequest?: number;
enableOfflineQueue?: boolean;
lazyConnect?: boolean;
}
const errorMap: IErrorMap = {
'redis.unavailable': {message: 'Redis unavailable', statusCode: 503},
};
let _errors: Errors<typeof errorMap>;
/**
* The subset of the ioredis API the generic adapter (+ derived realm adapters)
* uses. Derived adapters reach the live client via
* `this.config.context.redis`.
*/
export interface IRedisClient {
status: string;
connect(): Promise<unknown>;
get(key: string): Promise<string | null>;
set(key: string, value: string): Promise<unknown>;
del(...keys: string[]): Promise<number>;
exists(key: string): Promise<number>;
expire(key: string, seconds: number): Promise<number>;
ttl(key: string): Promise<number>;
hgetall(key: string): Promise<Record<string, string>>;
hget(key: string, field: string): Promise<string | null>;
hset(key: string, field: string, value: string | number): Promise<unknown>;
hincrby(key: string, field: string, increment: number): Promise<number>;
hdel(key: string, ...fields: string[]): Promise<number>;
eval(script: string, numKeys: number, ...keysAndArgs: unknown[]): Promise<unknown>;
scan(cursor: string, ...args: unknown[]): Promise<[string, string[]]>;
quit(): Promise<unknown>;
}
export default adapter<IConfig>(({utError}) => {
_errors ||= utError.register(errorMap);
let redis: IRedisClient;
/**
* (Re)create the ioredis client from a `redis` config slice.
* Connection is lazy — the first command connects; any failure surfaces
* synchronously and is wrapped into `redis.unavailable` (503).
*/
const createClient = (config: IConfig): void => {
const shared = {
keyPrefix: config.keyPrefix,
password: config.password,
maxRetriesPerRequest: config.maxRetriesPerRequest ?? 1,
enableOfflineQueue: config.enableOfflineQueue ?? false,
lazyConnect: config.lazyConnect ?? true,
};
redis = config.cluster
? (new Cluster(
(config.nodes ?? []).map(node => ({host: node.host, port: node.port})),
{
redisOptions: shared,
},
) as unknown as IRedisClient)
: (new Redis({
host: config.host ?? '127.0.0.1',
port: config.port ?? 6379,
db: config.db,
...shared,
}) as unknown as IRedisClient);
};
/**
* Ensure the lazy client is connected before the first command.
* If Redis is down this rejects and the request is blocked (503).
*/
const ensureConnected = async (): Promise<IRedisClient> => {
const status = redis.status;
if (
status !== 'ready' &&
status !== 'connect' &&
status !== 'connecting' &&
status !== 'reconnecting'
) {
await redis.connect();
}
return redis;
};
// Generic string-key operations: redis.key.get|set|del|exists|expire|ttl
const keyOps: Record<string, (params: Record<string, unknown>) => Promise<unknown>> = {
get: async params => ({value: await redis.get(params.keyName as string)}),
set: async params => {
await redis.set(params.keyName as string, params.keyValue as string);
return {success: true};
},
del: async params => ({
deleted: await redis.del(
...((params.keyNames as string[]) ?? [params.keyName as string]),
),
}),
exists: async params => ({
exists: (await redis.exists(params.keyName as string)) === 1,
}),
expire: async params => ({
expired: (await redis.expire(params.keyName as string, params.seconds as number)) === 1,
}),
ttl: async params => ({ttl: await redis.ttl(params.keyName as string)}),
list: async params => {
const pattern = (params.pattern as string) ?? '*';
const count = (params.count as number) ?? 100;
const limit = (params.limit as number) ?? 1000;
const cursor = (params.cursor as string) ?? '0';
const keyNames: string[] = [];
let next = cursor;
do {
const [newCursor, batch] = await redis.scan(next, 'MATCH', pattern, 'COUNT', count);
keyNames.push(...batch);
next = newCursor;
} while (next !== '0' && keyNames.length < limit);
return {items: keyNames.slice(0, limit).map(keyName => ({keyName})), cursor: next};
},
};
// Generic hash operations: redis.hash.getAll|get|set|incrBy|del
const hashOps: Record<string, (params: Record<string, unknown>) => Promise<unknown>> = {
getAll: async params => ({fields: await redis.hgetall(params.keyName as string)}),
get: async params => ({
value: await redis.hget(params.keyName as string, params.fieldName as string),
}),
set: async params => {
await redis.hset(
params.keyName as string,
params.fieldName as string,
params.fieldValue as string | number,
);
return {success: true};
},
incrBy: async params => ({
value: await redis.hincrby(
params.keyName as string,
params.fieldName as string,
params.increment as number,
),
}),
del: async params => ({
deleted: await redis.hdel(
params.keyName as string,
...((params.fieldNames as string[]) ?? [params.fieldName as string]),
),
}),
};
// Generic Lua script evaluation: redis.script.eval
const scriptOps: Record<string, (params: Record<string, unknown>) => Promise<unknown>> = {
eval: async params => ({
result: await redis.eval(
params.script as string,
((params.keyNames as string[]) ?? []).length,
...((params.keyNames as string[]) ?? []),
...((params.args as unknown[]) ?? []),
),
}),
};
return {
activation: {
default: {
type: 'redis',
},
},
async init(...configs: object[]) {
await super.init(...configs);
createClient((this.config as {redis?: IConfig}).redis ?? {});
this.config.context = {...this.config.context, redis};
},
start() {
super.connect();
return super.start();
},
async stop() {
try {
await redis?.quit();
} catch {
// Best-effort: a lazy client that never connected may reject quit().
}
try {
// `quit()` waits for the QUIT round-trip and can leave the socket
// open when the client is mid-connect/reconnect (the commander tap
// test observed a lingering 6379 socket after stop). `disconnect()`
// force-closes without waiting, guaranteeing the handle is released.
(redis as {disconnect?: () => void})?.disconnect?.();
} catch {
// ignore
}
return super.stop();
},
/**
* configChanged hook: only recreate the Redis client when the `redis`
* connection sub-key changed. Unrelated config changes are ignored.
*/
async configChanged(diff: Map<string, {prev: unknown; next: unknown}>, next: unknown) {
const redisChanged = Array.from(diff.keys()).some(
(key: string) =>
key === this.config.id + '.redis' || key.startsWith(this.config.id + '.redis.'),
);
if (!redisChanged) return;
const newAdapterConfig = (next as Record<string, unknown>)?.[this.config.id] as
| {redis?: IConfig}
| undefined;
if (newAdapterConfig?.redis) {
(this.config as {redis?: IConfig}).redis = newAdapterConfig.redis;
await redis?.quit();
createClient(newAdapterConfig.redis);
this.config.context = {...this.config.context, redis};
}
},
async exec(this: Adapter<IConfig>, params: Record<string, unknown>, $meta: IMeta) {
// Generic redis vocabulary: `redis.<object>.<operation>` where
// object ∈ {key, hash, script} and operation is the verb, e.g.
// `redis.key.get`, `redis.hash.getAll`, `redis.script.eval`.
const parts = ($meta.method ?? '').split('.');
const object = parts[1];
const operation = parts[2];
try {
await ensureConnected();
// `{ns}.database.list` — enumerate the logical databases this
// source exposes (the configured db index).
if (object === 'database' && operation === 'list') {
const db = (this.config as {redis?: IConfig}).redis?.db ?? 0;
return {items: [{db}]};
}
const ops =
object === 'key'
? keyOps
: object === 'hash'
? hashOps
: object === 'script'
? scriptOps
: undefined;
const fn = ops?.[operation];
if (!fn) throw new Error(`Unknown redis operation: ${object}.${operation}`);
return await fn(params);
} catch (error) {
throw this.error(_errors['redis.unavailable'](error), $meta);
}
},
};
});
|