modify twitch bot flow

libs/twitch-bot/index.js

@@ -19,11 +19,13 @@ class WS {
     //   }
     // }
     this.cache = {}
+    this.cmdQueue = []
     this.tick = null
     this.join = []
-    this.tick = setInterval(this.runTick.bind(this), 1000)
+    this.wait = 0
     event.on('twitchJoin', (channel) => {
-      this.joinChannel(channel)
+      // this.joinChannel(channel)
+      this.addCmdQueue(this.joinChannel, [channel])
     })
     event.on('twitchSend', data => {
       if (!('msg' in data) || typeof data.msg !== 'string') return
@@ -48,11 +50,15 @@ class WS {
 
   handleError (err) {
     console.log(err)
+    clearInterval(this.tick)
     this.ws = null
+    this.join = []
     this.runBot()
   }
   handleExit (code) {
+    clearInterval(this.tick)
     this.ws = null
+    this.join = []
     this.runBot()
   }
   handleMessage (data) {
@@ -70,6 +76,7 @@ class WS {
     }
   }
   async handleOpen () {
+    this.tick = setInterval(this.runTick.bind(this), 1000)
     // from pool get db connection
     let db = await DB.connect()
 
@@ -90,7 +97,8 @@ class WS {
     if (result !== null && result.rowCount > 0) {
       for (let row of result.rows) {
         if ('name' in row) {
-          this.joinChannel(row.name)
+          // this.joinChannel(row.name)
+          this.addCmdQueue(this.joinChannel, [row.name])
           // this.ws.send('JOIN #' + row.name)
           // this.join.push(row.name)
         }
@@ -115,7 +123,60 @@ class WS {
     }
   }
 
+  async reJoinChannel () {
+    this.leaveAllChannel()
+    // from pool get db connection
+    let db = await DB.connect()
+    // 取得要加入的頻道列表
+    let text = `select "name" from "public"."twitch_channel" where "join" = true`
+    let result = await db.query({
+      text
+    })
+
+    // release pool connection
+    await db.release()
+
+    if (result !== null && result.rowCount > 0) {
+      for (let row of result.rows) {
+        if ('name' in row) {
+          // this.joinChannel(row.name)
+          this.addCmdQueue(this.joinChannel, [row.name])
+          // this.ws.send('JOIN #' + row.name)
+          // this.join.push(row.name)
+        }
+      }
+    }
+  }
+
+  leaveAllChannel () {
+    for (let i of this.join) {
+      this.addCmdQueue(this.leaveChannel, [i])
+    }
+  }
+
+  addCmdQueue (act = null, arg = []) {
+    if (act === null || typeof act !== 'function') return
+    if (!Array.isArray(arg)) return
+    act = act.bind(this)
+    this.cmdQueue.push({act, arg})
+  }
+
   runTick () {
+    this.wait++
+    if (this.wait > 1800) {
+      this.wait = 0
+      this.reJoinChannel().then(() => {}).catch(() => {})
+    }
+    try {
+      if (this.cmdQueue.length > 0 && this.ws) {
+        let tmp = this.cmdQueue.shift()
+        if ('act' in tmp && typeof tmp.act === 'function' && 'arg' in tmp && Array.isArray(tmp.arg)) {
+          tmp.act(...tmp.arg)
+        }
+      }
+    } catch (err) {
+      console.log('tick error :::: ', err)
+    }
     if (this.cache === null || typeof this.cache !== 'object') return
     let time = Date.now()
     for (let i in this.cache) {
@@ -154,6 +215,15 @@ class WS {
     this.ws.send(`JOIN #${channel.trim()}`)
     this.join.push(channel.trim())
   }
+
+  leaveChannel (channel = null) {
+    if (this.ws === null || !('send' in this.ws) || typeof this.ws.send !== 'function') return null
+    if (channel === null || typeof channel !== 'string' || channel.trim().length === 0) return null
+    console.log(`IRC::: < ${`PART #${channel.trim()}`}`)
+    this.ws.send(`PART #${channel.trim()}`)
+    let idx = this.join.indexOf(channel)
+    this.join.splice(idx, 1)
+  }
 }
 
 module.exports = new WS()