"use strict";
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
    var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
    if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
    else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
    return c > 3 && r && Object.defineProperty(target, key, r), r;
};
var __metadata = (this && this.__metadata) || function (k, v) {
    if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v);
};
var __param = (this && this.__param) || function (paramIndex, decorator) {
    return function (target, key) { decorator(target, key, paramIndex); }
};
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.RedisService = void 0;
const common_1 = require("@nestjs/common");
const ioredis_1 = __importDefault(require("ioredis"));
let RedisService = class RedisService {
    constructor(redisClient) {
        this.redisClient = redisClient;
        this.redisCache = new Map();
        console.log('=[RedisService] Init ...');
    }
    async publish(channel, message) {
        const payload = typeof message === 'object' ? JSON.stringify(message) : String(message);
        return this.redisClient.publish(channel, payload);
    }
    async subscribe(channel, handler) {
        if (this.redisCache.has(channel)) {
            console.warn(`[RedisService] subscribe: ${channel} 已经存在，跳过。`);
            return;
        }
        const subscriberClient = this.redisClient.duplicate();
        subscriberClient.on('error', (err) => {
            console.error(`[RedisService] 频道 [${channel}] 连接错误:`, err.message);
        });
        subscriberClient.on('ready', async () => {
            try {
                await subscriberClient.subscribe(channel);
                console.log(`[RedisService] 频道 [${channel}] 重连成功，已重新订阅。`);
            }
            catch (err) {
                console.error(`[RedisService] 频道 [${channel}] 重新订阅失败:`, err);
            }
        });
        subscriberClient.on('reconnecting', () => {
            console.warn(`[RedisService] 频道 [${channel}] 连接断开，正在尝试重连...`);
        });
        subscriberClient.on('message', (chan, message) => {
            if (chan === channel) {
                handler(message);
            }
        });
        this.redisCache.set(channel, subscriberClient);
        try {
            await subscriberClient.subscribe(channel);
            console.log(`[RedisService] 成功初始化订阅频道: ${channel}`);
        }
        catch (err) {
            console.error(`[RedisService] 初始订阅失败: ${channel}`, err);
        }
    }
    async unsubscribe(channel) {
        const client = this.redisCache.get(channel);
        if (client) {
            try {
                await client.unsubscribe(channel);
                await client.quit();
                this.redisCache.delete(channel);
                console.log(`[RedisService] 已成功取消订阅并关闭频道连接: ${channel}`);
            }
            catch (err) {
                console.error(`[RedisService] 取消订阅出错: ${channel}`, err);
            }
        }
    }
    async brpop(listKey, timeout, clientKey = listKey) {
        if (!this.redisCache.has(clientKey)) {
            console.log(`[RedisService] 为 ${clientKey} 创建新的阻塞连接...`);
            this.redisCache.set(clientKey, this.redisClient.duplicate());
        }
        const blockingClient = this.redisCache.get(clientKey);
        try {
            return await blockingClient.brpop(listKey, timeout);
        }
        catch (error) {
            console.error(`[RedisService] BRPOP 执行出错:`, error);
            throw error;
        }
    }
    async onModuleDestroy() {
        console.log('[RedisService] 正在关闭所有阻塞连接...');
        const closePromises = [];
        for (const [key, client] of this.redisCache.entries()) {
            if (client) {
                console.log(`[RedisService] 正在关闭连接: ${key}`);
                closePromises.push(client.quit()
                    .catch(err => console.error(`关闭连接 ${key} 时出错:`, err)));
            }
        }
        await Promise.all(closePromises);
        this.redisCache.clear();
        console.log('[RedisService] 所有阻塞连接已安全释放');
    }
    async scan(cursor, matchType, matchPattern, countType, count) {
        return await this.redisClient.scan(cursor, matchType, matchPattern, countType, count);
    }
    async set(key, value, seconds) {
        const data = typeof value === 'string' ? value : JSON.stringify(value);
        if (seconds) {
            return await this.redisClient.set(key, data, 'EX', seconds);
        }
        return await this.redisClient.set(key, data);
    }
    async get(key) {
        return await this.redisClient.get(key);
    }
    async del(key) {
        if (key?.length > 2 && key.substring(key?.length - 2) == ":*") {
            let cursor = '0';
            do {
                const [newCursor, keys] = await this.scan(cursor, 'MATCH', key, 'COUNT', 100);
                cursor = newCursor;
                if (keys.length > 0) {
                    await this.redisClient.del(keys);
                }
            } while (cursor !== '0');
            return;
        }
        await this.redisClient.del(key);
    }
    async exists(key) {
        const result = await this.redisClient.exists(key);
        return result > 0;
    }
    async keys(key) {
        return await this.redisClient.keys(`${key}*`);
    }
    async flushall() {
        return await this.redisClient.flushall();
    }
    async flushCurrentDB() {
        return await this.redisClient.flushdb();
    }
    pipeline() {
        return this.redisClient.pipeline();
    }
    async mget(keys) {
        return await this.redisClient.mget(keys);
    }
    async smembers(key) {
        return await this.redisClient.smembers(key);
    }
    async hincrby(key, field, count) {
        return await this.redisClient.hincrby(key, field, count);
    }
    async hget(key, field) {
        return await this.redisClient.hget(key, field);
    }
    async scard(key) {
        return await this.redisClient.scard(key);
    }
    async hmset(key, field) {
        return await this.redisClient.hmset(key, field);
    }
    async hmget(key, ...fields) {
        return await this.redisClient.hmget(key, ...fields);
    }
    async zadd(key, score, member) {
        return await this.redisClient.zadd(key, score, member);
    }
    async zrangebyscore(key, min, max) {
        return await this.redisClient.zrangebyscore(key, min, max);
    }
    async zrem(key, member) {
        return await this.redisClient.zrem(key, member);
    }
    async hset(key, fieldOrObject, value) {
        if (typeof fieldOrObject === 'object' && fieldOrObject !== null) {
            return await this.redisClient.hset(key, fieldOrObject);
        }
        else {
            return await this.redisClient.hset(key, fieldOrObject, value);
        }
    }
    async hgetall(key) {
        return await this.redisClient.hgetall(key);
    }
    async lpush(key, value) {
        const data = typeof value === 'string' ? value : JSON.stringify(value);
        return await this.redisClient.lpush(key, data);
    }
    async expire(key, seconds) {
        return await this.redisClient.expire(key, seconds);
    }
    async getEntity(key, entity) {
        const data = await this.redisClient.get(`${entity.name}:${key}`);
        return JSON.parse(data);
    }
    async getEntityAll(entity) {
        const pattern = `${entity.name}:*`;
        let cursor = '0';
        const dataList = [];
        do {
            const [newCursor, scannedKeys] = await this.scan(cursor, 'MATCH', pattern, 'COUNT', 500);
            cursor = String(newCursor);
            if (scannedKeys.length > 0) {
                const pipeline = this.redisClient.pipeline();
                scannedKeys.forEach(key => pipeline.get(key));
                const results = await pipeline.exec();
                if (results) {
                    for (const [err, data] of results) {
                        if (!err && data) {
                            try {
                                dataList.push(JSON.parse(data));
                            }
                            catch (e) {
                            }
                        }
                    }
                }
            }
        } while (cursor !== '0');
        return dataList;
    }
    async findByCondition(entity, propertyName, propertyValue) {
        let cursor = '0';
        const pattern = `${entity.name}:*`;
        do {
            const [newCursor, scannedKeys] = await this.scan(cursor, 'MATCH', pattern, 'COUNT', 500);
            cursor = newCursor;
            if (scannedKeys.length > 0) {
                const pipeline = this.redisClient.pipeline();
                scannedKeys.forEach(key => pipeline.get(key));
                const results = await pipeline.exec();
                for (const [err, data] of results) {
                    if (!err && data) {
                        try {
                            if (typeof data === 'string') {
                                const parsed = JSON.parse(data);
                                if (parsed[propertyName] === propertyValue)
                                    return parsed;
                            }
                        }
                        catch (e) { }
                    }
                }
            }
        } while (cursor !== '0');
        return null;
    }
    async getUniversalOptionList(entityName) {
        const pattern = `${entityName}:*`;
        const dataList = [];
        let cursor = '0';
        const mapRules = {
            "ProjectEntity": { label: "projectName", value: "projectId" },
            "ProductEntity": { label: "productName", value: "productId" },
            "DeviceEntity": { label: "deviceName", value: "iotId" },
            "ThingModelEntity": { label: "thingModelName", value: "thingModelId" },
            "MetaSymbolEntity": { label: "symbolName", value: "symbolId" },
            "RolesEntitly": { label: "name", value: "id" }
        };
        const rule = mapRules[entityName];
        if (!rule)
            return { data: [] };
        do {
            const [newCursor, keys] = await this.scan(cursor, 'MATCH', pattern, 'COUNT', 500);
            cursor = newCursor;
            if (keys.length > 0) {
                const pipeline = this.redisClient.pipeline();
                keys.forEach(key => pipeline.get(key));
                const results = await pipeline.exec();
                results.forEach(([err, data]) => {
                    if (!err && data) {
                        try {
                            const val = JSON.parse(data);
                            dataList.push({ label: val[rule.label], value: val[rule.value] });
                        }
                        catch (e) { }
                    }
                });
            }
        } while (cursor !== '0');
        return { data: dataList };
    }
    async getDevice(entity, iotId, projectId, productId, deviceId) {
        if (!iotId || iotId === null || iotId === '' || iotId === undefined) {
            iotId = await this.redisClient.get(`Device:Platform:${projectId}:${productId}:${deviceId}`);
        }
        return await this.getEntity(iotId, entity);
    }
    async setDeviceSnapshoot(iotId, dataJson) {
        await this.redisClient.set("Device:Snapshoot:" + iotId, JSON.stringify(dataJson));
    }
    async getDeviceSnapshoot(iotId) {
        const key = `Device:Snapshoot:${iotId}`;
        const data = await this.redisClient.get(key);
        return JSON.parse(data);
    }
    async addSubDevice(parentIotId, subIotId) {
        const key = `Device:SubDevices:${parentIotId}`;
        await this.redisClient.sadd(key, subIotId);
    }
    async getSubDevices(entity, iotId) {
        const key = `Device:SubDevices:${iotId}`;
        const subIotIds = await this.redisClient.smembers(key);
        const devices = [];
        const invalidIds = [];
        for (const subIotId of subIotIds) {
            const device = await this.getDevice(entity, subIotId);
            if (device && device.parentIotId === iotId) {
                devices.push(device);
            }
            else {
                invalidIds.push(subIotId);
            }
        }
        if (invalidIds.length > 0) {
            await this.redisClient.srem(key, ...invalidIds);
        }
        return devices;
    }
};
exports.RedisService = RedisService;
exports.RedisService = RedisService = __decorate([
    (0, common_1.Injectable)(),
    __param(0, (0, common_1.Inject)('REDIS')),
    __metadata("design:paramtypes", [ioredis_1.default])
], RedisService);
//# sourceMappingURL=redis.service.js.map