P42/P42_wssGateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

converting to MP, just config in accessRights left

Browse Source
This commit is contained in:
STEINNI
2025-10-05 21:24:10 +00:00
parent 03af111d7f
commit 350d37e465
13 changed files with 283 additions and 796 deletions
Download Patch File Download Diff File Expand all files Collapse all files
actions/pubSub.js
+66 -50
View File
@@ -15,34 +15,35 @@ export const methods = {
if(!Array.isArray(payload)){
this.sendErr(action, 'Invalid payload', reqid);
return;
};
}
// payload only accepts NON-Bas-Prefixed chans
let subscribed = []
for(var chan of payload){
if((!chan) || (typeof(chan)!='string')) continue
if(!this.accessRights.canSubscribe(this.userId, this.roles, chan)) {
if(this.debug) console.log('SUB: No rights to this chan!', this.userId, this.roles, chan)
continue
}
// Chat chans are forbidden here
if((chan.substr(0,8) == 'userchans') || (chan.substr(0,9) == 'lobbychans')) continue;
if(!chan.startsWith(this.config.redis.basePrefix)) chan = this.config.redis.basePrefix + chan
if(this.subscriptions.indexOf(chan)<0) {
this.subscriptions.push(chan);
let coudSubscribe = false
for(const rediscnx of this.allRediscnx){
if(!chan.startsWith(rediscnx.redisConfig.ChansFilter)) continue
else coudSubscribe = true
let localChan = rediscnx.redisConfig.basePrefix + localChan
if(!(localChan in rediscnx.subscriptions)) rediscnx.subscriptions[localChan] = [];
if(!rediscnx.subscriptions[localChan].includes(this.uuid)) {
rediscnx.subscriptions[localChan].push(this.uuid);
}
}
if(!(chan in this.rediscnx.subscriptions)) this.rediscnx.subscriptions[chan] = [];
if(this.rediscnx.subscriptions[chan].indexOf(this.uuid)<0) {
this.rediscnx.subscriptions[chan].push(this.uuid);
}
}
let shortChans = this.subscriptions.map(item => (
item.startsWith(this.config.redis.basePrefix) ? item.substr(this.config.redis.basePrefix.length) : item
))
if(coudSubscribe && (!subscribed.includes(chan))) subscribed.push(chan)
if(coudSubscribe && (!this.subscriptions.includes(chan)) ){ this.subscriptions.push(chan) }
}
let reply = {
'action': action,
'payload': shortChans,
'payload': subscribed,
'success': true,
};
if(reqid) reply.reqid = reqid;
@@ -67,29 +68,31 @@ export const methods = {
this.sendErr(action, 'Invalid payload', reqid);
return;
};
// payload only accepts NON-Bas-Prefixed chans
let unSubscribed = []
for(var chan of payload){
if((!chan) || (typeof(chan)!='string')) continue
if(this.accessRights.isMandatory(this.userId, this.roles, chan)) continue
// Chat chans are forbidden here
if((chan.substr(0,8) == 'userchans') || (chan.substr(0,9) == 'lobbychans')) continue;
if(!chan.startsWith(this.config.redis.basePrefix)) chan = this.config.redis.basePrefix + chan
if(this.subscriptions.indexOf(chan)>-1) {
this.subscriptions.splice(this.subscriptions.indexOf(chan), 1);
let couldUnsubscribe = false
for(const rediscnx of this.allRediscnx){
if(!chan.startsWith(rediscnx.redisConfig.ChansFilter)) continue
else couldUnsubscribe = true
let localChan = rediscnx.redisConfig.basePrefix + chan
if((localChan in rediscnx.subscriptions) && (rediscnx.subscriptions[chan].includes(this.uuid))) {
rediscnx.subscriptions[localChan].splice(rediscnx.subscriptions[localChan].indexOf(this.uuid), 1) ;
}
}
if((chan in this.rediscnx.subscriptions) && (this.rediscnx.subscriptions[chan].indexOf(this.uuid)>-1)) {
this.rediscnx.subscriptions[chan].splice(this.rediscnx.subscriptions[chan].indexOf(this.uuid), 1) ;
if(couldUnsubscribe && (!unSubscribed.includes(chan))) unSubscribed.push(chan)
if(couldUnsubscribe && this.subscriptions.includes(chan)) {
this.subscriptions.splice(this.subscriptions.indexOf(chan), 1);
}
}
let shortChans = this.subscriptions.map(item => (
item.startsWith(this.config.redis.basePrefix) ? item.substr(this.config.redis.basePrefix.length) : item
))
let reply = {
'action': action,
'payload': shortChans,
'payload': unSubscribed,
'success': true,
};
if(reqid) reply.reqid = reqid;
@@ -109,12 +112,9 @@ export const methods = {
}
*/
action_SUBLST(action, payload, reqid){
let shortChans = this.subscriptions.map(item => (
item.startsWith(this.config.redis.basePrefix) ? item.substr(this.config.redis.basePrefix.length) : item
))
let reply = {
'action': action,
'payload': shortChans,
'payload': this.subscriptions,
'success': true,
};
if(reqid) reply.reqid = reqid;
@@ -146,8 +146,7 @@ export const methods = {
return;
};
if( (!this.accessRights.canPublish(this.userId, this.roles, payload.chan)) &&
(! this.rediscnx.redPillsUuids.includes(this.uuid)) ) {
if( (!this.accessRights.canPublish(this.userId, this.roles, payload.chan)) ) {
this.sendErr(action, 'Unauthorized chan !', reqid);
if(this.debug) console.log('PUB: Unauthorized chan', payload.chan, this.userId, this.roles)
return
@@ -155,22 +154,33 @@ export const methods = {
let msgO
try { msgO = JSON.parse(payload.msg) } catch(err) { msgO = {'err':err} }
msgO.sender = this.userId
let histId = null
if(this.rediscnx.isHistorizedChan(payload.chan)) { // historize first to add the histId
let shortChan = payload.chan.startsWith(this.config.redis.basePrefix) ? payload.chan.substr(this.config.redis.basePrefix.length) : payload.chan
histId = await this.rediscnx.redisXadd(this.config.redis.basePrefix+this.config.redis.historizePrefix + shortChan, payload.msg, this.config.redis.historizeMax);
const chan = payload.chan
// First find the primary redis for this chan namespace, to do historization first, and get the histId
const primaryRediscnx = this.allRediscnx.find(cnx => ((chan.startsWith(cnx.redisConfig.ChansFilter)) &&(cnx.redisConfig.role=='primary')) )
if(!primaryRediscnx){
this.sendErr(action, 'No primary redis for this chan !', reqid);
if(this.debug) console.log('PUB: No primary redis for this chan ', chan)
return
}
if(primaryRediscnx.isHistorizedChan(chan)){
histId = await rediscnx.redisXadd(rediscnx.redisConfig.basePrefix+rediscnx.redisConfig.historizePrefix + chan, payload.msg, rediscnx.redisConfig.historizeMax);
if( !histId) {
this.sendErr(action, 'Could not historize, aborted event publish !', reqid);
console.error(`Could not historize for "${shortChan}", aborted event publish !`)
console.error(`Could not historize for "${chan}", aborted event publish !`)
return
}
msgO.histId = histId
}
msgO.sender = this.userId
try { payload.msg = JSON.stringify(msgO) } catch(err) {payload.msg = `{"err":"${err}}"` }
this.rediscnx.redisPublish(payload.chan, payload.msg)
// Now publish on every Redis that covers this chan namespace
try { payload.msg = JSON.stringify(msgO) } catch(err) {payload.msg = `{"err":"${err}}"` }
for(const rediscnx of this.allRediscnx){
if(!chan.startsWith(rediscnx.redisConfig.ChansFilter)) continue
rediscnx.redisPublish(chan, payload.msg)
}
let reply = {
'action': action,
@@ -216,22 +226,28 @@ export const methods = {
return
}
if( (!this.accessRights.canSubscribe(this.userId, this.roles, payload.channel)) &&
(! this.rediscnx.redPillsUuids.includes(this.uuid)) ) {
this.sendErr(action, 'Unauthorized channel !', reqid)
if( (!this.accessRights.canSubscribe(this.userId, this.roles, payload.channel)) ) {
this.sendErr(action, 'CHANHIST: Unauthorized channel !', reqid)
return
}
if(!this.rediscnx.isHistorizedChan(payload.channel)){
this.sendErr(action, 'Not an historized channel !', reqid)
const primaryRediscnx = this.allRediscnx.find(cnx => ((chan.startsWith(cnx.redisConfig.ChansFilter)) &&(cnx.redisConfig.role=='primary')) )
if(!primaryRediscnx){
this.sendErr(action, 'No primary redis for this chan !', reqid);
if(this.debug) console.log('CHANHIST: No primary redis for this chan ', chan)
return
}
if(!primaryRediscnx.isHistorizedChan(payload.channel)){
this.sendErr(action, 'CHANHIST: Not an historized channel !', reqid)
return
}
let from = (payload.from.indexOf('-')>-1) ? payload.from : 1000*payload.from
let to = '+'
if(payload.to) to = (payload.to.indexOf('-')>-1) ? payload.to : 1000*payload.to
let streamName = payload.channel.startsWith(this.config.redis.basePrefix) ? this.config.redis.historizePrefix+payload.channel.substr(this.config.redis.basePrefix.length) : this.config.redis.historizePrefix + payload.channel
let respPayload = await this.rediscnx.redisXrange(streamName, from, to);
let streamName = payload.channel.startsWith(primaryRediscnx.redisConfig.basePrefix) ? primaryRediscnx.redisConfig.historizePrefix+payload.channel.substr(primaryRediscnx.redisConfig.basePrefix.length) : primaryRediscnx.redisConfig.historizePrefix + payload.channel
let respPayload = await primaryRediscnx.redisXrange(streamName, from, to);
let reply = {
'action': action,