feat(reminder): send a notification using web push and email to user …

…and watchers (#142)

packages/be-gateway/src/events/index.ts

@@ -1,8 +1,10 @@
 import { connectSubClient } from '@shared/pubsub'
 import { NotificationEvent } from './notification.event'
+import { ReminderEvent } from './reminder.event'
 
 export const CHANNEL_SCHEDULER_ACTION_NOTIFY = 'scheduler:action-notify'
 export const CHANNEL_SCHEDULER_CREATE = 'scheduler:create'
+export const CHANNEL_RUN_EVERY_MINUTE = 'fixed:run-every-minute'
 export const EVENT = {
   SCHEDULER_DELETE: 'scheduler:delete'
 }
@@ -11,14 +13,22 @@ connectSubClient((err, redis) => {
   if (err) {
     return
   }
+  // We must subscribe channels first
   redis.subscribe(CHANNEL_SCHEDULER_ACTION_NOTIFY, (err, count) => {
     console.log('subscribed', CHANNEL_SCHEDULER_ACTION_NOTIFY)
   })
 
+  redis.subscribe(CHANNEL_RUN_EVERY_MINUTE)
+
+  // After that, we can listen messages from them
   redis.on('message', async (channel: string, data: string) => {
     if (channel === CHANNEL_SCHEDULER_ACTION_NOTIFY) {
       const event = new NotificationEvent()
       event.run(data)
     }
+    if (channel === CHANNEL_RUN_EVERY_MINUTE) {
+      const reminder = new ReminderEvent()
+      reminder.run()
+    }
   })
 })

packages/be-gateway/src/events/reminder.event.ts

@@ -0,0 +1,88 @@
+import { notifyToWebUsers } from '../lib/buzzer'
+import { getJSONCache } from '../lib/redis'
+import TaskReminderJob from '../jobs/reminder.job'
+import { mdUserFindEmailsByUids } from '@shared/models'
+import { sendEmail } from '../lib/email'
+
+type RemindPayload = {
+  message: string
+  link: string
+  receivers: string[]
+}
+
+export class ReminderEvent {
+  taskReminderJob: TaskReminderJob
+  constructor() {
+    this.taskReminderJob = new TaskReminderJob()
+  }
+  async run() {
+    try {
+      const now = new Date()
+      console.log('reminder.event called', now)
+
+      const results = await this.taskReminderJob.findByTime(now)
+
+      if (!results.length) return
+
+      results.forEach(async k => {
+        const data = await getJSONCache([k])
+        if (!data) return
+
+        this.sendNotification(data as RemindPayload)
+        this.sendEmailReminder(data as RemindPayload)
+        // const { receivers, message, link } = data as RemindPayload
+        // if (!receivers || !receivers.length) return
+        //
+        // const receiverSets = new Set(receivers)
+        // const filteredReceivers = Array.from(receiverSets)
+        //
+        // notifyToWebUsers(filteredReceivers, {
+        //   title: 'Reminder ⏰',
+        //   body: message,
+        //   deep_link: link
+        // })
+
+        // sendEmail({
+        //   emails,
+        //   subject,
+        //   html,
+        // })
+      })
+    } catch (error) {
+      console.log(error)
+    }
+  }
+
+  async sendNotification(data: RemindPayload) {
+    const { receivers, message, link } = data
+    if (!receivers || !receivers.length) return
+
+    const receiverSets = new Set(receivers)
+    const filteredReceivers = Array.from(receiverSets)
+
+    notifyToWebUsers(filteredReceivers, {
+      title: 'Reminder ⏰',
+      body: message,
+      deep_link: link
+    })
+  }
+
+  async sendEmailReminder(data: RemindPayload) {
+    const { receivers, message, link } = data
+    if (!receivers || !receivers.length) return
+
+    const emails = await mdUserFindEmailsByUids(receivers)
+
+    console.log(emails)
+    if (!emails.length) return
+
+    sendEmail({
+      emails,
+      subject: 'Reminder ⏰',
+      html: `
+${message}
+Link: ${link}
+`
+    })
+  }
+}

packages/be-gateway/src/exceptions/InternalErrorException.ts

@@ -0,0 +1,7 @@
+export default class InternalErrorException extends Error {
+  status: number
+  constructor(message?: string) {
+    super(message || 'INTERNAL_SERVER_ERROR')
+    this.status = 500
+  }
+}

packages/be-gateway/src/jobs/reminder.job.ts

@@ -0,0 +1,125 @@
+import { extracDatetime, padZero } from '@shared/libs'
+import { mdProjectGet } from '@shared/models'
+import { genFrontendUrl } from '../lib/url'
+import {
+  delCache,
+  findCache,
+  findCacheByTerm,
+  setJSONCache
+} from '../lib/redis'
+
+type TaskReminderParams = {
+  remindAt: Date
+  remindBefore?: number
+  taskId?: string
+  projectId?: string
+  title?: string
+  link?: string
+  message: string
+  receivers: string[]
+}
+
+export default class TaskReminderJob {
+  private async _createTaskLink(projectId: string, taskId: string) {
+    const project = await mdProjectGet(projectId)
+    const taskLink = genFrontendUrl(
+      `${project.organizationId}/project/${projectId}?mode=task&taskId=${taskId}`
+    )
+
+    return taskLink
+  }
+
+  private _genKey(date: Date, suffix: string) {
+    // create reminder key and save this key to redis
+    const { y, m, d, hour, min } = extracDatetime(date)
+
+    // key syntax: remind-ddddmmyy-hh-mm-task-<taskID>
+    const key = [
+      `remind-${y}${padZero(m)}${padZero(d)}-${padZero(hour)}:${padZero(min)}${
+        suffix ? '-task-' + suffix : ''
+      }`
+    ]
+
+    return key
+  }
+
+  private _createExpiredTime(date: Date, expiredMinute: number) {
+    const now = new Date()
+    const dueTime = (date.getTime() - now.getTime()) / 1000
+    const expired = expiredMinute * 60
+
+    // the key will be expired after a specified minute
+    return dueTime + expired
+  }
+
+  async delete(taskId: string) {
+    const key = `remind*task-${taskId}`
+    const results = await findCacheByTerm(key)
+
+    if (!results.length) return
+
+    results.forEach(k => {
+      delCache([k])
+    })
+  }
+
+  async create({
+    remindAt,
+    // beforeAt should be number, it represents minutes
+    remindBefore,
+    taskId,
+    // link,
+    projectId,
+    message,
+    receivers
+  }: TaskReminderParams) {
+    // clone the dueDate to prevent unneccessary updates
+    const d1 = new Date(remindAt)
+    const now = new Date()
+
+    // TODO: if user want set an reminder at the exact time, do not substract the dueDate
+    if (remindBefore) {
+      d1.setMinutes(d1.getMinutes() - remindBefore)
+      message = `It's ${remindBefore} minutes to: ${message}`
+    } else {
+      message = `It's time to: ${message}`
+    }
+
+    if (d1 <= now) {
+      console.log('Can not create reminder, because remind time less than now')
+      return
+    }
+
+    // create reminder key and save this key to redis
+    const key = this._genKey(d1, taskId)
+
+    // the reminder key should have an expired time
+    // so it can delete itself automatically
+    // after the reminder run for 5 minutes, it will be expired
+    const expired = this._createExpiredTime(d1, 5)
+    const link = await this._createTaskLink(projectId, taskId)
+
+    // save the key with expired time and data
+    setJSONCache(
+      key,
+      {
+        receivers,
+        message: message,
+        link
+      },
+      Math.ceil(expired)
+    )
+  }
+
+  async findByTime(date: Date) {
+    const { y, m, d, hour, min } = extracDatetime(date)
+
+    const key = [
+      `remind-${y}${padZero(m)}${padZero(d)}-${padZero(hour)}:${padZero(min)}`
+    ]
+
+    const results = await findCache(key)
+
+    return results
+  }
+}

packages/be-gateway/src/lib/redis.ts

@@ -93,20 +93,27 @@ export const genKeyFromSource = (source: { [key: string]: unknown }) => {
 
 export const setJSONCache = (
   key: CACHE_KEY,
-  value: RedisJSONValue | RedisJSONValue[]
+  value: RedisJSONValue | RedisJSONValue[],
+  expired?: number
 ) => {
   if (!connected) {
     return null
   }
   try {
     const cacheKey = genKey(key)
+    console.log('cachekey', cacheKey, expired)
     redis.set(cacheKey, JSON.stringify(value))
-    redis.expire(cacheKey, DAY)
+    redis.expire(cacheKey, expired || DAY)
   } catch (error) {
     console.log('set redis cache error')
   }
 }
 
+export const setCacheExpire = (key: CACHE_KEY, expired: number) => {
+  const cacheKey = genKey(key)
+  redis.expire(cacheKey, expired)
+}
+
 export const getJSONCache = async (key: CACHE_KEY) => {
   if (!connected) {
     return null
@@ -147,10 +154,20 @@ export const delMultiCache = async (keys: CACHE_KEY[]) => {
   await pipeline.exec()
 }
 
-export const findCache = async (key: CACHE_KEY) => {
+export const findCache = async (key: CACHE_KEY, abs = false) => {
   try {
     const newKey = genKey(key)
-    const results = await redis.keys(newKey + '*')
+    const asterisk = abs ? '' : '*'
+    const results = await redis.keys(newKey + asterisk)
+    return results
+  } catch (error) {
+    console.log('find cache key error', error)
+  }
+}
+
+export const findCacheByTerm = async (term: string) => {
+  try {
+    const results = await redis.keys(term)
     return results
   } catch (error) {
     console.log('find cache key error', error)

packages/be-gateway/src/main.ts

@@ -12,7 +12,9 @@ import './events'
 import Routes from './routes'
 // import { Log } from './lib/log'
 
-connectPubClient()
+connectPubClient((err) => {
+  console.log(err)
+})
 const app: Application = express()
 
 app.get('/check-health', (req, res) => {

packages/be-gateway/src/queues/Task/MoveToAnotherBoardJob.ts

@@ -1,4 +1,4 @@
-import { TaskReorderService } from '../../services/taskReorder.service'
+import { TaskReorderService } from '../../services/task/order.service'
 import { BaseJob } from '../BaseJob'
 
 interface IReorderData {

packages/be-gateway/src/queues/Task/ReorderJob.ts

@@ -1,4 +1,4 @@
-import { TaskReorderService } from '../../services/taskReorder.service'
+import { TaskReorderService } from '../../services/task/order.service'
 import { BaseJob } from '../BaseJob'
 
 interface IReorderData {