redis-hashmap.ts

import Redis from 'ioredis';

type CallbackFn<T = any, S = any> = (
    value: T,
    key: string,
    store: S
) => void | Promise<void>;

interface IRedisHashMap<T = any> {
    delete(key: string): Promise<number>;
    forEach(
        callbackfn: CallbackFn<T, IRedisHashMap>,
        doneCallbackfn?: () => void
    ): Promise<void>;
    get(key: string): Promise<T | null>;
    has(key: string): Promise<number>;
    set(key: string, data: T): Promise<number>;
    size: Promise<number>;
}

export class RedisHashMap<T = any> implements IRedisHashMap<T> {
    // Hashmap Redis client
    private redis: Redis.Redis;

    // Hashmap reference
    private key: string;

    /**
     * Create new Redis hashmap
     * @param redis IORedis Redis client
     * @param key Hashmap name
     */
    constructor(redis: Redis.Redis, key: string) {
        // Set Redis client connection
        this.redis = redis;

        // Set hahsmap key
        this.key = key;
    }

    /**
     * Field length of hashmap
     */
    get size() {
        return this.redis.hlen(this.key);
    }

    /**
     * Array of hashmap keys
     */
    keys() {
        return this.redis.hkeys(this.key);
    }

    /**
     * Array of hashmap values
     */
    values() {
        return this.redis.hvals(this.key);
    }

    /**
     * Delete field from hashmap
     * @param field Hash key
     */
    delete(field: string) {
        return this.redis.hdel(this.key, field);
    }

    /**
     * Iterate over hashmap items
     * @param callback For each callback function
     * @param doneCallbackfn Completion function
     */
    forEach(
        callback: CallbackFn<T, RedisHashMap>,
        doneCallbackfn?: () => void
    ) {
        // Begin redis hashmap stream
        const stream = this.redis.hscanStream(this.key, { count: 1 });

        // Return promise for async forloop
        return new Promise<void>((resolve, reject) => {
            // Handle data stream
            stream.on('data', ([key]: string[]) => {
                // Pause stream per item
                stream.pause();

                // Wait for item handling
                Promise.resolve(async (progress: any) => {
                    // Get item from Redis list
                    const item = await this.get(key);

                    // Run forEach callback function
                    await callback(item, key, this);

                    // Resolve and continue iteration
                    progress();
                }).then(() => {
                    // Continue stream handling
                    stream.resume();
                });
            });

            // Handle stream error
            stream.on('error', err => {
                console.warn(err);
                reject(err);
            });

            // Handle stream completion
            stream.on('end', () => {
                // Run the completion callback
                if (doneCallbackfn) {
                    doneCallbackfn();
                }
                resolve();
            });
        });
    }

    /**
     * Return value of a field with hashmap
     * @param field Hash key
     */
    async get(field: string): Promise<T | null> {
        // Get stored value
        const data = await this.redis.hget(this.key, field);
        // Return parsed value, or null
        return JSON.parse(data ?? 'null');
    }

    /**
     * Check if field exists within hashmap
     * @param field Hahs key
     */
    has(field: string) {
        return this.redis.hexists(this.key, field);
    }

    /**
     * Set a key-value field within hashmap
     * @param field Hash key
     * @param data Data
     */
    set(field: string, data: T) {
        return this.redis.hset(this.key, field, JSON.stringify(data));
    }
}