notifier done

src/components/controllers.ts

@@ -19,20 +19,22 @@ export const controllers = {
     ]
 >
 
+const entries = Object.entries(controllers)
+
 export const Controllers = ioc.add(
-    [],
-    (): FastifyPluginAsyncTypebox => async (server) => {
-        const enabledControllers = Object.values(controllers).filter(
-            ([, component]) => {
-                const state = ioc.getOrThrow(component.Controller)
-                return !state.disabled
-            },
-        )
-        await Promise.all(
-            enabledControllers.map(async ([prefix, component]) => {
-                const plugin = await ioc.resolve(component.Controller)
-                server.register(plugin, { prefix })
-            }),
-        )
-    },
+    entries.map(([, [, component]]) => component.Controller),
+    (...plugins): FastifyPluginAsyncTypebox =>
+        async (server) => {
+            await Promise.all(
+                entries
+                    .map(async ([, [prefix]], index) => {
+                        const plugin = plugins[index]
+                        if (plugin === ioc.disabled) {
+                            return null
+                        }
+                        return server.register(plugin, { prefix })
+                    })
+                    .filter((result) => result !== null),
+            )
+        },
 )

src/components/email/index.ts

@@ -1 +1,2 @@
 export * from './controller.js'
+export * from './logic.js'

src/components/task/controller.test.ts

@@ -29,7 +29,7 @@ afterAll(() => {
 
 partial(ioc, {
     controllers: ['task'],
-    services: ['database'],
+    services: ['database', 'notifier'],
 })
 const { inject } = await ioc.resolve(Server)
 

src/components/task/index.ts

@@ -1 +1,2 @@
 export * from './controller.js'
+export * from './logic.js'

src/components/task/logic.ts

@@ -1,12 +1,13 @@
 import { ioc } from '#root/ioc/index.js'
-import { SQL, and, asc, desc, eq } from 'drizzle-orm'
+import { InferModel, SQL, and, asc, desc, eq } from 'drizzle-orm'
 import { List, Post, Put, Sort } from './schema.js'
 import { ServerError } from '#root/error/server-error.js'
 import { Database } from '#root/services/database/index.js'
 import { task } from '#root/services/database/schema/task.js'
 import { adaptConditions } from '#root/services/database/adapt-conditions.js'
+import { Notifier } from '#root/services/notifier.js'
 
-export const Logic = ioc.add([Database], (db) => {
+export const Logic = ioc.add([Database, Notifier], (db, notifier) => {
     return {
         async list(userId: number, options: List['querystring']) {
             let query = db
@@ -30,7 +31,10 @@ export const Logic = ioc.add([Database], (db) => {
             return await query
         },
 
-        async get(userId: number, taskId: number) {
+        async get(
+            userId: number,
+            taskId: number,
+        ): Promise<InferModel<typeof task> | undefined> {
             const [record] = await db
                 .select()
                 .from(task)
@@ -50,20 +54,30 @@ export const Logic = ioc.add([Database], (db) => {
             const [result] = await db
                 .insert(task)
                 .values({ ...options, userId })
-            return result.insertId
+            const taskId = result.insertId
+
+            if (options.notifyDate) {
+                await notifier.set(userId, taskId, options.notifyDate)
+            }
+
+            return taskId
         },
 
         async put(userId: number, taskId: number, options: Put['body']) {
             await db
                 .update(task)
                 .set(options)
                 .where(and(eq(task.userId, userId), eq(task.id, taskId)))
+
+            await notifier.set(userId, taskId, options.notifyDate ?? null)
         },
 
         async delete(userId: number, taskId: number) {
             await db
                 .delete(task)
                 .where(and(eq(task.userId, userId), eq(task.id, taskId)))
+
+            await notifier.set(userId, taskId, null)
         },
     }
 })

src/ioc/index.ts

@@ -25,8 +25,10 @@ export type ResolveParameters<TParameters extends Parameters> = Writable<{
 const disabled = new Proxy(
     {},
     {
-        get() {
-            throw new ServerError('Provider is disabled')
+        get(target, reciever) {
+            if (reciever !== 'then') {
+                throw new ServerError('Provider is disabled')
+            }
         },
     },
 )

src/main.ts

@@ -1,5 +1,10 @@
 import { ioc } from './ioc/index.js'
+import { partial } from './ioc/partial.js'
 import { Server } from './server/index.js'
 
+partial(ioc, {
+    controllers: ['task'],
+    services: ['database', 'notifier'],
+})
 const server = await ioc.resolve(Server)
 await server.listen()

src/services/mailer.ts

@@ -1,3 +1,4 @@
+import { Task } from '#root/components/task/schema.js'
 import { Config } from '#root/config/index.js'
 import { ioc } from '#root/ioc/index.js'
 import nodemailer from 'nodemailer'
@@ -19,5 +20,14 @@ export const Mailer = ioc.add([Config], async (config) => {
                 html: `<a href="${confirmUrl}">Confirm</a>`,
             })
         },
+
+        async sendTaskNotification(email: string, task: Task) {
+            await mailer.sendMail({
+                from,
+                to: email,
+                subject: 'Task notification',
+                html: `<h1>${task.title}</h1><p>${task.description}</p>`,
+            })
+        },
     }
 })

src/services/notifier.ts

@@ -1,11 +1,90 @@
 import { ioc } from '#root/ioc/index.js'
-import { Queue } from 'bullmq'
+import { ConnectionOptions, Queue, Worker } from 'bullmq'
+import * as task from '#root/components/task/index.js'
+import { Mailer } from './mailer.js'
+import * as email from '#root/components/email/index.js'
 
-export const Notifier = ioc.add([], () => {
-    const queue = new Queue('notifications', {
-        connection: {
-            host: 'localhost',
-            port: 7379,
-        },
+const queueName = 'notifications'
+
+type JobData = {
+    userId: number
+}
+
+const jobIdManager = {
+    prefix: 'task-',
+    create(taskId: number) {
+        return `${this.prefix}${taskId}`
+    },
+    parse(jobId: string) {
+        return Number.parseInt(jobId.substring(this.prefix.length))
+    },
+}
+
+export const Notifier = ioc.add([Mailer, email.Logic], (mailer, emailLogic) => {
+    const connection: ConnectionOptions = {
+        host: 'localhost',
+        port: 7379,
+    }
+
+    const queue = new Queue<JobData>(queueName, {
+        connection,
     })
+
+    new Worker<JobData>(
+        queueName,
+        async (job) => {
+            const taskLogic = await ioc.resolve(task.Logic)
+
+            const taskId = jobIdManager.parse(job.id!)
+            const { userId } = job.data
+
+            const taskRecord = await taskLogic.get(userId, taskId)
+            if (!taskRecord) {
+                return 'Task not found'
+            }
+
+            const emailRecord = await emailLogic.get(userId)
+            if (!emailRecord) {
+                return 'No email'
+            }
+            if (!emailRecord.confirmed) {
+                return 'Email not confirmed'
+            }
+
+            await mailer.sendTaskNotification(emailRecord.email, taskRecord)
+            return `Sent to ${emailRecord.email}`
+        },
+        {
+            connection,
+            removeOnComplete: {
+                count: 1000,
+            },
+            removeOnFail: {
+                count: 5000,
+            },
+        },
+    )
+
+    return {
+        async set(userId: number, taskId: number, notifyDate: string | null) {
+            const jobId = jobIdManager.create(taskId)
+
+            const job = await queue.getJob(jobId)
+
+            if (notifyDate === null || new Date(notifyDate) <= new Date()) {
+                if (job) {
+                    await job.remove()
+                }
+                return
+            }
+
+            const delay = Number(new Date(notifyDate)) - Number(new Date())
+
+            if (job) {
+                await job.changeDelay(delay)
+            } else {
+                await queue.add('Notify', { userId }, { jobId, delay })
+            }
+        },
+    }
 })