Merge branch 'main' of https://github.com/padlocks/Compatiplural

Signed-off-by: bee! <pascalyoung03@gmail.com>

.dockerignore

@@ -0,0 +1,2 @@
+node_modules
+npm-debug.log

.env_example

@@ -3,8 +3,11 @@ url_override="https://api.apparyllis.com"
 api_version="v1"
 socket="wss://api.apparyllis.com/v1/socket"
 pk_url="https://api.pluralkit.me/v2"
-token="AAAAAAAAAAAAAAAAAAAA"
-userId="AAAAAAAAAAAAAAAAAAA"
-pk_token= "AAAAAAAAAAAAAAAA"
-pk_system="AAAAAAAAAAAAAAAA"
+token="SIMPLYPLURAL_TOKEN"
+userId="abcd1234"
+pk_token= "PLURALKIT_TOKEN"
 heartbeat=4500000
+max_workers=1
+silence_connections=true
+full_swap=false
+primary_tag="primary "

.gitignore

@@ -5,3 +5,5 @@ config.json
 package-lock.json
 .env
 .vercel
+docker_build.bat
+docker_run.bat

Dockerfile

@@ -0,0 +1,16 @@
+FROM node:18
+
+WORKDIR /usr/src/app
+
+COPY package*.json ./
+RUN npm install
+
+# If you are building your code for production
+# RUN npm ci --omit=dev
+
+# Bundle app source
+COPY . .
+
+#EXPOSE 8080
+
+CMD [ "node", "index.js" ]

README.md

@@ -16,5 +16,7 @@ These can be set either in the .env file, in terminal, or in the config vars sec
 | token | token_here | Your SimplyPlural account token. As of now, the only permission necessary is the Read permission. |
 | userId | user_id | Your SimplyPlural account/system ID. You can find it in account info near the bottom. |
 | pk_token | pluralkit_token_here | Your PluralKit token. Get it by using `pk;token`. |
-| pk_system | pluralkit_system_id | Your Pluralkit system ID. This can be either your Discord account ID or your 5 letter ID shown by using pk;system. |
 | heartbeat | 4500000 | The time in miliseconds before the websocket client reconnects to the websocket server. |
+| max_workers | 1 | Max number of workers for processing enqueued tasks. This probably shouldn't be changed. |
+| silence_connections | true | Whether or not to silence the websocket connection and authentication messages. |
+| full_swap | false | Determines whether to completely overwrite PluralKit's front with SimplyPlural's. No individual changes. |

WebsocketClient.js

@@ -7,7 +7,7 @@ function WebSocketClient(url) {
     let connecting = false
     let backoff = 250
     const init = () => {
-        console.error(`::SimplyWS:: [${timestamp()}] connecting`)
+        if (!process.env.silence_connections) console.error(`::SimplyWS:: [${timestamp()}] connecting`)
         connecting = false
         if (client !== undefined) {
             client.removeAllListeners()
@@ -21,14 +21,14 @@ function WebSocketClient(url) {
             timeout = setTimeout(() => client.terminate(), process.env.heartbeat || 350000)
         }
         client.on('ping', () => {
-            console.log(`::SimplyWS:: [${timestamp()}] pinged`)
+            if (!process.env.silence_connections) console.log(`::SimplyWS:: [${timestamp()}] pinged`)
             heartbeat()
         })
         client.on('open', (e) => {
             if (typeof this.onOpen === 'function') {
                 this.onOpen()
             } else {
-                console.log(`::SimplyWS:: [${timestamp()}] opened`)
+                if (!process.env.silence_connections) console.log(`::SimplyWS:: [${timestamp()}] opened`)
                 console.log(e)
             }
             heartbeat()
@@ -37,7 +37,7 @@ function WebSocketClient(url) {
             if (typeof this.onMessage === 'function') {
                 this.onMessage(e)
             } else {
-                console.log(`::SimplyWS:: [${timestamp()}] messaged`)
+                if (!process.env.silence_connections) console.log(`::SimplyWS:: [${timestamp()}] messaged`)
             }
             heartbeat()
         })

dataManager.js

@@ -1,9 +1,9 @@
 const axios = require('axios')
-const { Config, System, Util } = require('SimplyAPI')
+const { Config, System, Util } = require('simplyapi')
 
 const pkUrl = Config.pk_url
 const pkHeader = {
-    'Content-Type': 'application/json',
+    'Content-Type': 'application/json; charset=UTF-8',
     'Authorization': Config.pk_token
 }
 
@@ -13,10 +13,6 @@ async function initializeCache() {
     cache.frontHistory = await system.getFronters()
 }
 
-function unknownError400() {
-    return
-}
-
 function unknownTarget(target) {
     console.log('::SimplyWS:: Unknown update target: ' + target + '\n::SimplyWS:: Full message: ' + e)
 }
@@ -25,15 +21,9 @@ function unrecognizedMessage(msg) {
     console.log('::SimplyWS:: Unrecognized message: ' + msg + '\n::SimplyWS:: Full message: ' + e)
 }
 
-// async function asyncForEach(array, callback) {
-//     for (let index = 0; index < array.length; index++) {
-//         await callback(array[index], index, array)
-//     }
-// }
-
 async function getPKFronters() {
     let members = []
-    let fronters = await axios.get(`${pkUrl}/systems/${Config.pk_system}/fronters`, {
+    let fronters = await axios.get(`${pkUrl}/systems/@me/fronters`, {
         headers: pkHeader
     })
         .catch((err) => {
@@ -60,9 +50,9 @@ async function findPrimary() {
     return new Promise(async (resolve) => {
         await Util.asyncForEach(fronters, async (fronter) => {
             if (fronter.content.customStatus) {
-                if (fronter.content.customStatus.toLowerCase().includes("primary")) {
+                if (fronter.content.customStatus.toLowerCase().includes(Config.primary_tag)) {
                     let member = await system.getMemberById(fronter.content.member)
-                    resolve(member.content.pkId)
+                    resolve({ name: member.content.name, pkId: member.content.pkId })
                     found = true
                 }
             }
@@ -116,14 +106,94 @@ async function determineAction(eventData, frontData = []) {
     return action
 }
 
+async function swapFront() {
+    let system = new System(Config)
+    let front = await system.getFronters()
+    
+    // start forming new front list
+    let newFront = []
+    let frontNames = []
+    for (member of front) {
+        let m = await system.getMemberById(member.content.member)
+
+        if (m.content && m.content.pkId) {
+            // fronting member pkID has been found
+            newFront.push(m.content.pkId)
+            frontNames.push(m.content.name)
+        }
+        else {
+            console.warn('::SimplyWS:: System member not found, this may be a custom front which is unsupported.')
+        }
+    }
+
+    // shift primary fronter to first in list
+    let primary = await findPrimary()
+    let primaryPK = primary.pkId
+    let primaryName = primary.name
+    if (primaryPK) {
+        if (newFront.indexOf(primaryPK) > 0) {
+            newFront.splice(newFront.indexOf(primaryPK), 1)
+            newFront.unshift(primaryPK)
+        }
+
+        if (frontNames.indexOf(primaryName) > 0) {
+            frontNames.splice(frontNames.indexOf(primaryName), 1)
+            frontNames.unshift(primaryName)
+        }
+    }
+
+    // post the new switch        
+    let url = `${pkUrl}/systems/@me/switches`
+    await axios.post(url, JSON.stringify({ "members": newFront }), {
+        headers: pkHeader
+    })
+        .then(async (res) => {
+            // check if current front equals the new front
+            let front = await getPKFronters()
+            var equal = (front.length == newFront.length) && front.every(function(element, index) {
+                return element === newFront[index]; 
+            })
+            if (!equal) {
+                console.log('::SimplyWS:: Failed to swap front: ' + newFront)
+                await swapFront()
+                return
+            } else {
+                let formattedNames = frontNames.toString().replace(',',', ')
+                console.log(`::SimplyWS:: SimplyPlural -> PluralKit: ${formattedNames}`)
+            }
+        })
+        .catch(async err => {
+            let status = err.status || err.toJSON().status
+            if (status == 400) {
+                // if the fronter is already in the front, do nothing
+                return
+            }
+            else if (status == 404) {
+                return
+            }
+            else if (status == 429) {
+                // Too many requests
+                console.warn("::SimplyWS:: Too many requests, waiting to try again.")
+                setTimeout(async function () {
+                    await swapFront()
+                }, 1000)
+                return
+            }
+        })
+}
+
 async function insertFront(member) {
     // get current fronters and add new fronter
-    let system = new System(Config)
     let fronters = await getPKFronters()
-    fronters.push(member.content.pkId)
+    if (!fronters.includes(member.content.pkId)) {
+        fronters.push(member.content.pkId)
+    } else {
+        console.warn('::SimplyWS:: Member already in fronters: ' + member.content.pkId)
+        return
+    }
 
     // find the "primary" fronter to move to the first element in the list
-    let primary = await findPrimary()
+    let primary = await findPrimary().pkId
     if (primary) {
         if (fronters.indexOf(primary) > 0) {
             fronters.splice(fronters.indexOf(primary), 1)
@@ -131,97 +201,117 @@ async function insertFront(member) {
         }
     }
 
-    // cache front
-    cache.frontHistory = await system.getFronters()
-
     // post the new switch        
-    axios.post(`${pkUrl}/systems/${Config.pk_system}/switches`, JSON.stringify({ "members": fronters }), {
+    let url = `${pkUrl}/systems/@me/switches`
+    await axios.post(url, JSON.stringify({ "members": fronters }), {
         headers: pkHeader
     })
-        .catch(err => {
-            if (err.toJSON().status == 400)
-                unknownError400()
-            else if (err.toJSON().status == 429)
+        .then(async (res) => {
+            let front = await getPKFronters()
+            if (!front.includes(member.content.pkId)) {
+                console.log('::SimplyWS:: Failed to insert fronter: ' + member.content.pkId)
+                await insertFront(member)
+                return
+            } else {
+                console.log('::SimplyWS:: ' + member.content.name + ' was added to the front.')
+            }
+        })
+        .catch(async err => {
+            let status = err.status || err.toJSON().status
+            if (status == 400) {
+                // if the fronter is already in the front, do nothing
+                return
+            }
+            else if (status == 404) {
+                // member not found
+                console.error("::SimplyWS:: Could not find member: " + member.content.pkId)
+                let index = fronters.indexOf(member.content.pkId)
+                fronters.splice(index, 1)
+                return
+            }
+            else if (status == 429) {
                 // Too many requests
-                setTimeout(function () {
-                    insertFront(member)
+                console.warn("::SimplyWS:: Too many requests, waiting to try again.")
+                let index = fronters.indexOf(member.content.pkId)
+                fronters.splice(index, 1)
+                setTimeout(async function () {
+                    await insertFront(member)
                 }, 1000)
                 return
+            }
         })
-
-    let checkFront = await getPKFronters()
-    if (!checkFront.includes(member.content.pkId)) {
-        await insertFront(member)
-        return
-    } else {
-        console.log('::SimplyWS:: ' + member.content.name + ' was added to the front.')
-    }
 }
 
 async function removeFront(member) {
-    let system = new System(Config)
     let fronters = await getPKFronters()
-    let index = fronters.indexOf(member.content.pkId)
-    fronters.splice(index, 1)
+    
+    if (fronters.includes(member.content.pkId)) {
+        let index = fronters.indexOf(member.content.pkId)
+        fronters.splice(index, 1)
+    } else {
+        console.warn('::SimplyWS:: Member is not in front: ' + member.content.pkId)
+        return
+    }
 
     // find the "primary" fronter to move to the first element in the list
-    let p = await findPrimary()
-    if (p) {
-        if (fronters.indexOf(p) > 0) {
-            fronters.splice(fronters.indexOf(p), 1)
-            fronters.unshift(p)
+    let primary = await findPrimary().pkId
+    if (primary) {
+        if (fronters.indexOf(primary) > 0) {
+            fronters.splice(fronters.indexOf(primary), 1)
+            fronters.unshift(primary)
         }
     }
 
-    // cache front
-    cache.frontHistory = await system.getFronters()
-
-    // post the new switch
-    axios.post(`${pkUrl}/systems/${Config.pk_system}/switches`, JSON.stringify({ "members": fronters }), {
+    let url = `${pkUrl}/systems/@me/switches`
+    await axios.post(url, JSON.stringify({ "members": fronters }), {
         headers: pkHeader
     })
-        .catch(err => {
-            if (err.toJSON().status == 400)
-                unknownError400()
-            else if (err.toJSON().status == 429)
+        .then(async (res) => {
+            let front = await getPKFronters()
+            if (front.includes(member.content.pkId)) {
+                console.log('::SimplyWS:: Failed to remove fronter: ' + member.content.pkId)
+                await removeFront(member)
+                return
+            } else {
+                console.log('::SimplyWS:: ' + member.content.name + ' was removed from the front.')
+            }
+        })
+        .catch(async err => {        
+            let status = err.status || err.toJSON().status    
+            if (status == 400) {
+                // fronter is already not in front
+                console.warn("::SimplyWS:: " + member.content.name + " is not in the front.")
+                return
+            }
+            else if (status == 429) {
                 // Too many requests
-                setTimeout(function () {
-                    removeFront(member)
+                console.warn("::SimplyWS:: Too many requests, waiting to try again.")
+                fronters.push(member.content.pkId)
+                setTimeout(async function () {
+                    await removeFront(member)
                 }, 1000)
                 return
+            }
         })
-    
-    let checkFront = await getPKFronters()
-    if (checkFront.includes(member.content.pkId)) {
-        await removeFront(member)
-        return
-    } else {
-        console.log('::SimplyWS:: ' + member.content.name + ' was removed from the front.')
-    }
 }
 
 async function updateCustomStatus(member) {
     // find the "primary" fronter to move to the first element in the list
-    let system = new System(Config)
     let fronters = await getPKFronters()
-    let primary = await findPrimary()
-    if (primary && fronters.length > 1) {
+    let primary = await findPrimary().pkId
+    if (primary && fronters.length > 1 && (member.content.pkId == primary)) {
         if (fronters.indexOf(primary) >= 0) {
             fronters.splice(fronters.indexOf(primary), 1)
             fronters.unshift(primary)
 
-            // cache front
-            cache.frontHistory = await system.getFronters()
-
             // post the new switch
-            axios.post(`${pkUrl}/systems/${Config.pk_system}/switches`, JSON.stringify({ "members": fronters }), {
+            axios.post(`${pkUrl}/systems/@me/switches`, JSON.stringify({ "members": fronters }), {
                 headers: pkHeader
             })
-                .catch(err => {
-                    if (err.toJSON().status == 400)
-                        unknownError400()
-                    else if (err.toJSON().status == 429)
-                        // Too many requests
+                .catch(async err => {
+                    if (err.toJSON().status == 429)
+                        //Too many requests
+                        console.warn("::SimplyWS:: Too many requests, waiting to try again.")
                         setTimeout(function () {
                             updateCustomStatus(member)
                         }, 1000)
@@ -236,19 +326,16 @@ async function updateCustomStatus(member) {
     }
 }
 
-const { inspect } = require('util')
 const transform = require('lodash.transform')
 const isEqual = require('lodash.isequal')
-const isArray = require('lodash.isarray')
 const isObject = require('lodash.isobject')
-const { PassThrough } = require('stream')
 async function calculateDiff(origObj, newObj) {
     return new Promise(function (resolve) {
         changes = (newObj, origObj) => {
             let arrayIndexCounter = 0
             return transform(newObj, function (result, value, key) {
                 if (!isEqual(value, origObj[key])) {
-                    let resultKey = isArray(origObj) ? arrayIndexCounter++ : key
+                    let resultKey = Array.isArray(origObj) ? arrayIndexCounter++ : key
                     result[resultKey] = (isObject(value) && isObject(origObj[key])) ? changes(value, origObj[key]) : value
                 }
             })
@@ -259,12 +346,12 @@ async function calculateDiff(origObj, newObj) {
 
 module.exports = {
     initializeCache,
-    unknownError400,
     unknownTarget,
     unrecognizedMessage,
     getPKFronters,
     findPrimary,
     determineAction,
+    swapFront,
     insertFront,
     removeFront,
     updateCustomStatus,

index.js

@@ -1,23 +1,66 @@
 const dotenv = require('dotenv')
 dotenv.config()
 
-const { Config, System } = require('SimplyAPI')
-const { Util } = require('SimplyAPI')
-const { initializeCache, determineAction, insertFront, removeFront, updateCustomStatus } = require('./dataManager')
+const { Config, System } = require('simplyapi')
+const { Util } = require('simplyapi')
+const { initializeCache, determineAction, swapFront, insertFront, removeFront, updateCustomStatus } = require('./dataManager')
+
+const {
+    isMainThread,
+    BroadcastChannel,
+    Worker
+} = require('node:worker_threads')
 
 let e
-main = async () => {
-    openWebSocket()
+main = () => {
+    initiateWorkerPool()
 }
 
-openWebSocket = async () => {
+// Queue
+const async = require('async')
+const queue = async.queue((task, completed) => {
+    let error = { status: false, message: '' }
+    update(task.data)
+        .catch(err => {
+            error.status = true
+            error.message = err
+        })
+    completed(error, task)
+
+}, Config.max_workers)
+
+initiateWorkerPool = () => {
+    // Worker Pool
+    const bc = new BroadcastChannel('plural')
+
+    if (isMainThread) {
+        openWebSocket()
+
+        bc.onmessage = (event) => {
+            //console.log('::SimplyWS:: received message from worker')
+            queue.push(event.data, (error, task) => {
+                // task completed
+                if (error.status) {
+                    console.log(`An error occurred while processing task ${error.message}`)
+                }
+            })
+        }
+        for (let n = 0; n < Config.max_workers; n++)
+            new Worker(__filename)
+    }
+}
+
+// Socket
+openWebSocket = () => {
     const WebSocketClient = require('./WebsocketClient')
-    const wss = new WebSocketClient(Config.socket);
+    const wss = new WebSocketClient(Config.socket)
     let initialPacket = { "op": "authenticate", "token": Config.token }
     wss.onOpen = (_) => { wss.send(JSON.stringify(initialPacket)); }
     wss.onClose = (e) => { console.log('SimplyWS/onClose :: %s', e); e = '' }
     wss.onError = (e) => { console.log('SimplyWS/onError :: %s', e) }
 
+    const bc = new BroadcastChannel('plural')
+    let first_auth = true
     wss.onMessage = (raw) => {
         e = raw
         let data = JSON.parse(e)
@@ -25,15 +68,16 @@ openWebSocket = async () => {
 
         switch (data.msg) {
             case "Successfully authenticated":
-                console.log('::SimplyWS:: authenticated')
+                if (!process.env.silence_connections || first_auth) console.log('::SimplyWS:: authenticated')
+                first_auth = false
                 // cache current front
                 initializeCache()
                 break
             case "Authentication violation: Token is missing or invalid. Goodbye :)":
-                console.log('::SimplyWS:: invalid token, exiting..')
+                console.error('::SimplyWS:: invalid token, exiting..')
                 process.exit(1)
             case "update":
-                update(data)
+                bc.postMessage({data: data})
                 break
             default:
                 //unrecognizedMessage(data.msg)
@@ -42,6 +86,7 @@ openWebSocket = async () => {
     }
 }
 
+// Data Processing
 update = async (data) => {
     let target = data.target
     switch (target) {
@@ -50,8 +95,12 @@ update = async (data) => {
             await Util.asyncForEach(data.results, async (o) => {
                 let system = new System(Config)
                 let member = await system.getMemberById(o.content.member)
+                let swap = Config.full_swap
                 // insert
-                if (o.operationType == "insert") {
+                if (swap) {
+                    swapFront()
+                }
+                else if (o.operationType == "insert") {
                     insertFront(member)
                 }
                 else {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "Compatiplural",
-  "version": "1.0.0",
+  "version": "1.1.0",
   "description": "SimplyPlural -> PluralKit Connectivity",
   "main": "index.js",
   "scripts": {
@@ -12,13 +12,13 @@
   "repository": "github:padlocks/Compatiplural",
   "dependencies": {
     "ajv": "^8.10.0",
+    "async": "^3.2.3",
     "axios": "^0.26.0",
     "dotenv": "^16.0.0",
-    "lodash.isarray": "^4.0.0",
     "lodash.isequal": "^4.5.0",
     "lodash.isobject": "^3.0.2",
     "lodash.transform": "^4.6.0",
-    "simplyapi": "^0.1.1",
+    "simplyapi": "^0.1.4",
     "ws": "^8.5.0"
   },
   "optionalDependencies": {