:recycle: probably too big of an update | pubnub chat | sse notifications | planetscale connection script | added some env vars

backend/.env.sample

@@ -7,8 +7,14 @@ API_HOST=localhost
 API_PORT=3001
 APP_SECRET=mysecret
 
+USE_LOCAL_DB=true
+
 DB_ROOT_PASSWORD=root
 DB_USER=root
 DB_NAME=test
 DB_HOST=localhost
-DB_TYPE=mysql
+DB_TYPE=mysql
+DB_PORT=3306
+
+PSCALE_DB_BRANCH=main
+PSCALE_DB_NAME=planet-scale-db

backend/knexfile.js

@@ -1,15 +1,26 @@
 require('dotenv').config()
 
+const local = {
+    host: process.env.DB_HOST,
+    user: process.env.DB_USER,
+    password: process.env.DB_ROOT_PASSWORD,
+    database: process.env.DB_NAME,
+    port: process.env.DB_PORT
+}
+const pscale = {
+    host: process.env.PSCALE_DB_HOST ? process.env.PSCALE_DB_HOST : '127.0.0.1',
+    user: process.env.PSCALE_DB_USER ? process.env.PSCALE_DB_USER : 'root',
+    password: process.env.PSCALE_DB_PASSWORD ? process.env.PSCALE_DB_PASSWORD : '', 
+    database: process.env.PSCALE_DB_NAME,
+    port: process.env.PSCALE_DB_PORT ? process.env.PSCALE_DB_PORT : 3306
+}
+
+const connectionSettings = process.env.USE_LOCAL_DB == true ? local : pscale
+
 module.exports = {
     development: {
         client: process.env.DB_TYPE,
-        connection: {
-            host: process.env.DB_HOST,
-            user: process.env.DB_USER,
-            password: process.env.DB_ROOT_PASSWORD,
-            database: process.env.DB_NAME,
-            port: 3307
-        },
+        connection: connectionSettings,
         pool: {
             min: 2,
             max: 10

backend/lib/plugins/notification.js

@@ -1,69 +1,5 @@
-const Stream = require('stream')
-const PassThrough = Stream.PassThrough
-const Transform = Stream.Transform
 const NotificationRoute = require('../routes/notification')
-
-const stringifyEvent = function (event) {
-    let str = ''
-    const endl = '\r\n'
-    for (const i in event) {
-        let val = event[i]
-        if (val instanceof Buffer) { val = val.toString() }
-        if (typeof val === 'object') { val = JSON.stringify(val) }
-        str += i + ': ' + val + endl
-    }
-    str += endl
-    return str
-}
-
-class Transformer extends Transform {
-    constructor(options, objectMode) {
-        super({ objectMode })
-        options = options || {}
-        this.counter = 1
-        this.event = options.event || null
-        this.generateId = options.generateId ? options.generateId : () => {
-            return this.counter++
-        }
-    }
-    _transform (chunk, encoding, callback) {
-        const event = {
-            id: this.generateId(chunk),
-            data: chunk
-        }
-        if (this.event) { event.event = this.event }
-        this.push(stringifyEvent(event))
-        callback()
-    }
-    _flush(callback) {
-        this.push(stringifyEvent({ event: 'end', data: '' }))
-        callback()
-    }
-} 
-
-const writeEvent = function (event, stream) {
-    if (event) {
-        stream.write(stringifyEvent(event))
-    } else {
-        // closing time
-        stream.write(stringifyEvent({ event: 'end', data: '' }))
-        stream.end()
-    }
-}
-
-const onEvent = (event, h, streamOptions) => {
-    // We only support ObjectMode streams
-    if (!event._readableState.objectMode) return
-
-    const through = new Transformer(streamOptions, true)
-    const active = new PassThrough()
-    through.pipe(active)
-    event.pipe(through)
-    console.log(event)
-    return h.response(active)
-        .header('content-type', 'text/event-stream')
-        .header('content-encoding', 'identity')
-}
+const { onEvent } = require('../services/notification')
 
 module.exports = {
     name: 'notification-plugin',

backend/lib/services/notification.js

@@ -0,0 +1,118 @@
+/** Heavily lifted from: https://github.com/mtharrison/susie/blob/master/lib/index.js */
+
+const Stream = require('stream')
+const PassThrough = Stream.PassThrough
+const Transform = Stream.Transform
+
+const ENDER = { event: 'end', data: '' }
+
+/**
+ * Stringify a stream
+ * ?: I don't really get what this is doing
+ * @param {Stream} event
+ * @returns {string}
+ */
+const stringifyEvent = function (event) {
+    let str = ''
+    const endl = '\r\n'
+    for (const i in event) {
+        let val = event[i]
+        if (val instanceof Buffer) { val = val.toString() }
+        if (typeof val === 'object') { val = JSON.stringify(val) }
+        str += i + ': ' + val + endl
+    }
+    str += endl
+    return str
+}
+
+/**
+ * Transform extension
+ * ?: I don't really get what this is doing
+ * @param {object} options
+ * @param {object} objectMode 
+ */
+class Transformer extends Transform {
+    constructor(options, objectMode) {
+        super({ objectMode })
+        options = options || {}
+        this.counter = 1
+        this.event = options.event || null
+        this.generateId = options.generateId ? options.generateId : () => {
+            return this.counter++
+        }
+    }
+    _transform (chunk, encoding, callback) {
+        const event = {
+            id: this.generateId(chunk),
+            data: chunk
+        }
+        if (this.event) { event.event = this.event }
+        this.push(stringifyEvent(event))
+        callback()
+    }
+    _flush(callback) {
+        this.push(stringifyEvent(ENDER))
+        callback()
+    }
+} 
+
+/**
+ * Take an event stream and write content to another stream
+ * ?: Save this for future extension
+ * @param {Stream} event stream input
+ * @param {Stream} stream to write to
+ */
+const writeEvent = function (event, stream) {
+    if (event) {
+        stream.write(stringifyEvent(event))
+    } else {
+        // closing time
+        stream.write(stringifyEvent(ENDER))
+        stream.end()
+    }
+}
+
+/**
+ * Callback to decorate server toolkit (h)
+ * !: Currently we only support ObjectMode streams
+ * ?: I don't really get what this is doing
+ * @param {Stream} event stream input
+ * @param {Toolkit} h hapi common response toolkit
+ * @param {object} streamOptions
+ */
+const onEvent = (event, h, streamOptions) => {
+    // const state = h.request.plugins.notifications = h.request.plugins.notifications || {}
+    let active
+    if (event instanceof Stream.Readable) {
+        if (event._readableState.objectMode) {
+            const through = new Transformer(streamOptions, true)
+            active = new PassThrough()
+            through.pipe(active)
+            event.pipe(through)
+        }
+        // else {
+        //     stream = new Transformer(streamOptions, false)
+        //     event.pipe(stream)
+        // }
+        return h.response(active)
+            .header('content-type', 'text/event-stream')
+            .header('content-encoding', 'identity')
+    }
+    // Uncomment to do stream state stuff
+    // handle a first object arg
+    // if (!state.stream) {
+    //     active = new PassThrough()
+    //     state.stream = active
+    //     state.mode = 'object'
+    //     const response = h.response(active)
+    //         .header('content-type', 'text/event-stream')
+    //         .header('content-encoding', 'identity')
+    //     writeEvent(event, active)
+    //     return response
+    // }
+    // already have an object stream flowing, just write next event
+    // active = state.stream
+    // internals.writeEvent(event, active)
+}
+
+module.exports = { onEvent }

backend/package.json

@@ -5,7 +5,8 @@
   "main": "index.js",
   "scripts": {
     "start": "nodemon server",
-    "migrate": "knex migjrate:latest",
+    "connect": "USE_LOCAL_DB=false pscale connect $(grep '^PSCALE_DB_NAME' .env | cut -d '=' -f2) $(grep '^PSCALE_DB_BRANCH' .env | cut -d '=' -f2)",
+    "migrate": "knex migrate:latest",
     "unmigrate": "knex migrate:down",
     "reseed": "knex migrate:rollback --all && knex migrate:latest && knex seed:run",
     "generate": "node ./db/data-generator/index.js",

frontend/.env.sample

@@ -0,0 +1,3 @@
+VITE_PUBNUB_PUBLISH_KEY=pub-a-123467890-123467890
+VITE_PUBNUB_SUBSCRIBE_KEY=sub-a-123467890-123467890
+VITE_UUID_SECRET=00000000-0000-1111-0000-000000000000

frontend/src/App.vue

@@ -6,7 +6,7 @@ router-view
 import * as sss from '@/sss/import.css'
 
 export default {
-    name: 'app',
+    name: 'app'
 }
 </script>
 

frontend/src/services/chat.service.js

@@ -0,0 +1,93 @@
+import PubNub from 'pubnub'
+
+/** 
+ * Provider method holder
+ * We always reference this object so
+ * we don't have to hardcode provider specific
+ * methods for doing chat things.
+ * 
+ * This gets overloaded later in the program
+ */
+const providerMethods = {
+    publish: () => console.error('no provider publish method set'),
+    subscribe: () => console.error('no provider subscribe method set'),
+    listen: () => console.error('no provider listen method set')
+}
+
+/** 
+ * Breaking out as much pubnub specific flavor
+ */
+const setupPubnub = async uuid => {
+    if(!uuid) return console.error('no pubnub uuid set')
+    
+    const pubnubClient = await new PubNub({
+        publishKey: import.meta.env.VITE_PUBNUB_PUBLISH_KEY,
+        subscribeKey: import.meta.env.VITE_PUBNUB_SUBSCRIBE_KEY,
+        uuid
+    })
+    // Pass pubnub specific methods to our placeholder obj
+    providerMethods['publish'] = pubnubClient.publish
+    providerMethods['subscribe'] = pubnubClient.subscribe
+    providerMethods['listen'] = pubnubClient.addListener
+
+    return pubnubClient
+}
+
+
+class ChatMessage {
+    constructor({ title, description }) {
+        this.title = title,
+        this.description = description
+    }
+}
+
+const testMessage = new ChatMessage({
+    title: "testing",
+    description: "hello world!",
+})
+const MAIN_CHANNEL = 'hello_world'
+
+class Chatter {
+    constructor() {
+        this.groupings = {}
+        this.provider = null
+        this.uuid = null
+
+        // Setup the main channel
+        this.subscriptions = [MAIN_CHANNEL]
+        
+        this.listeners = {
+            status: async e => {
+                if (e.category !== "PNConnectedCategory") return
+                await this._publish(this.subscriptions[0], testMessage)
+            },
+            message: this._onMessage,
+            presence: this._onPresence
+        }
+    }
+    async _onMessage(e) {
+        console.log(e.message.title)
+        console.log(e.message.description)
+    }
+    async _onPresence(e) {
+        return
+    }
+    async setup(uuid) {
+        this.uuid = uuid
+        this.provider = await setupPubnub(this.uuid)
+
+        this._listenFor({ listeners: this.listeners })
+        this._subscribe(this.subscriptions)
+    }
+    async _publish(channel, message) {
+        return await providerMethods['publish']({ channel, message })
+    }
+    _subscribe(channels) {
+        providerMethods['subscribe']({ channels })
+    }
+    _listenFor({ listeners }) {
+        providerMethods['listen'](listeners)
+    }
+}
+
+export { Chatter }

frontend/src/services/notification.service.js

@@ -0,0 +1,28 @@
+import { remote } from '../utils/db'
+
+class Toaster {
+    constructor(profileId) {
+        this.url = `${remote}/notification/${profileId}/subscribe`
+        this.source = null
+        this.source = new EventSource(this.url)
+        this.listenFor('end', message => this.source.close)
+    }
+    listenFor(event, callback) {
+        this.source.addEventListener(event, callback)
+    }
+}
+
+class StonkAlert extends Toaster {
+    constructor(profileId) {
+        super(profileId)
+
+        this.stonks = {}
+        this.listenFor('stonk', message => {
+            const parsed = JSON.parse(message.data)
+            this.stonks[parsed.name] = parsed
+            console.log('updated:', this.stonks)
+        })
+    }
+}
+
+export { Toaster, StonkAlert }

frontend/src/services/queue.service.js

@@ -6,9 +6,14 @@ const fetchQueueByProfileId = async profileId => {
     const queue = await db.get(
         `/profile/${profileId}/queue?include_profile=true`,
     )
-    return queue.map(profileData => {
+    
+    if(!queue?.length) {
+        console.error('could not retrieve match queue. Please take the survey and rescore.')
+    }
+
+    return queue ? queue.map(profileData => {
         return new Profile({ email: null, ...profileData })
-    })
+    }) : []
 }
 
 const updateQueueByProfileId = async (profileId, targetId, reinsert) => {

frontend/src/utils/db.js

@@ -26,7 +26,7 @@ class Connector {
         const header = { ...headerTemplate }
         header.method = 'GET'
         try {
-            console.log(`${remote}${endpoint}`)
+            // console.log(`${remote}${endpoint}`)
             let res = await fetch(`${remote}${endpoint}`, header)
             if (!res.ok) {
                 throw Error(res.statusText)
@@ -73,23 +73,4 @@ class Connector {
 }
 const db = new Connector()
 
-class Toaster {
-    constructor({ profileId }) {
-        this.url = `${remote}/notification/${profileId}/subscribe`
-        
-    }
-    connect() {
-        this.source = new EventSource(`${remote}/notification/${profileId}/subscribe`)
-        this.source.onmessage = e => {
-            console.log("Data", + e.data)
-        }
-        this.source.onerror = e => {
-            console.error("EventSource failed.", e)
-        }
-        this.source.addEventListener('end', e => {
-            this.close()
-        })
-    }
-}
-
-export { Connector, db, Toaster }
+export { Connector, db, remote }

frontend/src/views/home.vue

@@ -13,7 +13,8 @@ import sidebar from '../components/Sidebar.vue'
 import mainNav from '../components/MainNav.vue'
 import profileCardList from '../components/ProfileCardList.vue'
 import { fetchQueueByProfileId } from '../services'
-import { Toaster } from '../utils/db'
+import { StonkAlert } from '../services/notification.service'
+import { Chatter } from '../services/chat.service'
 
 import batch_10 from '../../../backend/db/generated/_batch_10.js.ref'
 import batch_20 from '../../../backend/db/generated/_batch_20.js.ref'
@@ -29,6 +30,7 @@ export default {
         loading: true
     }),
     mounted() {
+        this.setupChatter()
         this.setupToaster()
     },
     async created() {
@@ -54,16 +56,14 @@ export default {
             this.loading = false
         },
         setupToaster() {
-            const stocks = {}
-            const source = new EventSource(`http://localhost:3001/api/notification/${this.mypid}/subscribe`)
-            source.addEventListener('stonk', function (message) {
-                console.log('updated:', message)
-                const data = JSON.parse(message.data)
-                stocks[data.name] = data
-            })
-            source.addEventListener('end', function (message) {
-                this.close()
-            })
+            const t = new StonkAlert(this.mypid)
+        },
+        setupChatter() {
+            const c = new Chatter()
+            const testAccountUUID = import.meta.env.VITE_TEST_ACCOUNT_UUID
+            c.setup(testAccountUUID)
+            console.log('---')
+            console.log(c)
         },
         processQueue(queueList) {
             const formattedList = []