diff --git a/.env.example b/.env.example index 872e5b2..d436be3 100644 --- a/.env.example +++ b/.env.example @@ -6,3 +6,7 @@ FRONTEND_URL=http://localhost:3000 JWT_SECRET=change-me-in-production PORT=3001 REDIS_URL=redis://localhost:6379 + +# Contribution reminder scheduler (src/modules/notifications) +CONTRIBUTION_PERIOD_DAYS=30 +CONTRIBUTION_REMINDER_LEAD_DAYS=3 diff --git a/package.json b/package.json index d039a25..7b76a22 100644 --- a/package.json +++ b/package.json @@ -15,6 +15,7 @@ "db:studio": "prisma studio" }, "dependencies": { + "@nestjs/bullmq": "^10.2.3", "@nestjs/common": "^10.3.8", "@nestjs/core": "^10.3.8", "@nestjs/platform-express": "^10.3.8", diff --git a/src/app.module.ts b/src/app.module.ts index 00897fd..2ef4952 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -1,6 +1,8 @@ import { Module } from "@nestjs/common"; import { ConfigModule } from "@nestjs/config"; import { ScheduleModule } from "@nestjs/schedule"; +import { BullModule } from "@nestjs/bullmq"; +import { redisConnectionFromEnv } from "./modules/notifications/reminder.queue"; import { AuthModule } from "./modules/auth/auth.module"; import { GroupsModule } from "./modules/groups/groups.module"; import { MembersModule } from "./modules/members/members.module"; @@ -15,6 +17,7 @@ import { StellarIndexerService } from "./common/stellar-indexer.service"; imports: [ ConfigModule.forRoot({ isGlobal: true }), ScheduleModule.forRoot(), + BullModule.forRoot({ connection: redisConnectionFromEnv() }), AuthModule, GroupsModule, MembersModule, diff --git a/src/modules/notifications/notifications.module.ts b/src/modules/notifications/notifications.module.ts index 8ea4af6..b83b175 100644 --- a/src/modules/notifications/notifications.module.ts +++ b/src/modules/notifications/notifications.module.ts @@ -1,11 +1,23 @@ import { Module } from "@nestjs/common"; +import { BullModule } from "@nestjs/bullmq"; import { NotificationsController } from "./notifications.controller"; import { NotificationsService } from "./notifications.service"; +import { ReminderProcessor } from "./reminder.processor"; +import { ReminderScheduler } from "./reminder.scheduler"; +import { CONTRIBUTION_REMINDERS_QUEUE } from "./reminder.queue"; import { PrismaService } from "../../common/prisma.service"; @Module({ + imports: [ + BullModule.registerQueue({ name: CONTRIBUTION_REMINDERS_QUEUE }), + ], controllers: [NotificationsController], - providers: [NotificationsService, PrismaService], + providers: [ + NotificationsService, + PrismaService, + ReminderProcessor, + ReminderScheduler, + ], exports: [NotificationsService], }) export class NotificationsModule {} diff --git a/src/modules/notifications/reminder.processor.ts b/src/modules/notifications/reminder.processor.ts new file mode 100644 index 0000000..a32eeb4 --- /dev/null +++ b/src/modules/notifications/reminder.processor.ts @@ -0,0 +1,51 @@ +import { Processor, WorkerHost } from "@nestjs/bullmq"; +import { Logger } from "@nestjs/common"; +import { Job } from "bullmq"; +import { PrismaService } from "../../common/prisma.service"; +import { NotificationsService } from "./notifications.service"; +import { + CONTRIBUTION_REMINDERS_QUEUE, + ReminderJobData, +} from "./reminder.queue"; + +/** + * Worker that turns a queued reminder into a persisted notification + * (and, when the member has an email on file, an email — stubbed for now). + */ +@Processor(CONTRIBUTION_REMINDERS_QUEUE) +export class ReminderProcessor extends WorkerHost { + private readonly logger = new Logger(ReminderProcessor.name); + + constructor( + private readonly notifications: NotificationsService, + private readonly prisma: PrismaService, + ) { + super(); + } + + async process(job: Job): Promise { + const { groupId, memberAddress, period, dueDate } = job.data; + const due = new Date(dueDate); + + await this.notifications.create( + memberAddress, + "contribution_due", + "Contribution due soon", + `Your contribution for period ${period} is due on ${due.toDateString()}.`, + { groupId, period, dueDate }, + ); + + const member = await this.prisma.member.findFirst({ + where: { address: memberAddress, groupId }, + select: { email: true }, + }); + + if (member?.email) { + // TODO: send the reminder over email via nodemailer once SMTP + // credentials are configured (SMTP_* env vars). + this.logger.log( + `Would email contribution reminder to ${member.email} (period ${period}).`, + ); + } + } +} diff --git a/src/modules/notifications/reminder.queue.ts b/src/modules/notifications/reminder.queue.ts new file mode 100644 index 0000000..5c29fb1 --- /dev/null +++ b/src/modules/notifications/reminder.queue.ts @@ -0,0 +1,39 @@ +import { ConnectionOptions } from "bullmq"; + +/** Name of the BullMQ queue that holds contribution-reminder jobs. */ +export const CONTRIBUTION_REMINDERS_QUEUE = "contribution-reminders"; + +/** Job name used for every reminder enqueued onto the queue. */ +export const CONTRIBUTION_DUE_JOB = "contribution-due"; + +/** Payload enqueued for a single member who needs a reminder. */ +export interface ReminderJobData { + groupId: string; + memberAddress: string; + /** 0-based contribution period the reminder is for. */ + period: number; + /** ISO string of when the contribution is due. */ + dueDate: string; +} + +/** + * Build a BullMQ/ioredis connection from a REDIS_URL + * (e.g. `redis://:password@host:6379/0`). Falls back to localhost. + */ +export function redisConnectionFromEnv( + url = process.env.REDIS_URL ?? "redis://localhost:6379", +): ConnectionOptions { + const parsed = new URL(url); + const db = + parsed.pathname && parsed.pathname.length > 1 + ? Number(parsed.pathname.slice(1)) + : undefined; + + return { + host: parsed.hostname, + port: parsed.port ? Number(parsed.port) : 6379, + username: parsed.username || undefined, + password: parsed.password || undefined, + db: Number.isNaN(db as number) ? undefined : db, + }; +} diff --git a/src/modules/notifications/reminder.scheduler.spec.ts b/src/modules/notifications/reminder.scheduler.spec.ts new file mode 100644 index 0000000..1caa898 --- /dev/null +++ b/src/modules/notifications/reminder.scheduler.spec.ts @@ -0,0 +1,99 @@ +import { getQueueToken } from "@nestjs/bullmq"; +import { Test } from "@nestjs/testing"; +import { PrismaService } from "../../common/prisma.service"; +import { ReminderScheduler } from "./reminder.scheduler"; +import { + CONTRIBUTION_DUE_JOB, + CONTRIBUTION_REMINDERS_QUEUE, +} from "./reminder.queue"; + +const DAY_MS = 24 * 60 * 60 * 1000; + +describe("ReminderScheduler", () => { + const now = new Date("2026-07-01T09:00:00.000Z"); + + let queue: { add: jest.Mock; getJobCounts: jest.Mock }; + let prisma: { + group: { findMany: jest.Mock }; + contribution: { findMany: jest.Mock }; + }; + let scheduler: ReminderScheduler; + + const buildScheduler = async () => { + queue = { add: jest.fn(), getJobCounts: jest.fn().mockResolvedValue({}) }; + prisma = { + group: { findMany: jest.fn() }, + contribution: { findMany: jest.fn().mockResolvedValue([]) }, + }; + + const moduleRef = await Test.createTestingModule({ + providers: [ + ReminderScheduler, + { provide: getQueueToken(CONTRIBUTION_REMINDERS_QUEUE), useValue: queue }, + { provide: PrismaService, useValue: prisma }, + ], + }).compile(); + + scheduler = moduleRef.get(ReminderScheduler); + }; + + beforeEach(buildScheduler); + + it("enqueues a reminder only for members who have not contributed this period", async () => { + // Period length defaults to 30 days; created 28 days ago => due in 2 days + // (inside the 3-day lead window), current period = 0. + const createdAt = new Date(now.getTime() - 28 * DAY_MS); + prisma.group.findMany.mockResolvedValue([ + { + id: "grp1", + createdAt, + members: [{ address: "GAAA" }, { address: "GBBB" }], + }, + ]); + prisma.contribution.findMany.mockResolvedValue([{ memberAddress: "GAAA" }]); + + const enqueued = await scheduler.enqueueDueReminders(now); + + expect(enqueued).toBe(1); + expect(queue.add).toHaveBeenCalledTimes(1); + expect(queue.add).toHaveBeenCalledWith( + CONTRIBUTION_DUE_JOB, + expect.objectContaining({ + groupId: "grp1", + memberAddress: "GBBB", + period: 0, + dueDate: new Date(createdAt.getTime() + 30 * DAY_MS).toISOString(), + }), + expect.objectContaining({ jobId: "grp1:GBBB:0" }), + ); + }); + + it("does not enqueue when the due date is outside the lead window", async () => { + // Created 10 days ago => due in 20 days, well outside the 3-day window. + prisma.group.findMany.mockResolvedValue([ + { + id: "grp1", + createdAt: new Date(now.getTime() - 10 * DAY_MS), + members: [{ address: "GAAA" }], + }, + ]); + + const enqueued = await scheduler.enqueueDueReminders(now); + + expect(enqueued).toBe(0); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it("does not enqueue when every member already contributed", async () => { + const createdAt = new Date(now.getTime() - 28 * DAY_MS); + prisma.group.findMany.mockResolvedValue([ + { id: "grp1", createdAt, members: [{ address: "GAAA" }] }, + ]); + prisma.contribution.findMany.mockResolvedValue([{ memberAddress: "GAAA" }]); + + const enqueued = await scheduler.enqueueDueReminders(now); + + expect(enqueued).toBe(0); + expect(queue.add).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/notifications/reminder.scheduler.ts b/src/modules/notifications/reminder.scheduler.ts new file mode 100644 index 0000000..d34d082 --- /dev/null +++ b/src/modules/notifications/reminder.scheduler.ts @@ -0,0 +1,107 @@ +import { InjectQueue } from "@nestjs/bullmq"; +import { Injectable, Logger } from "@nestjs/common"; +import { Cron, CronExpression } from "@nestjs/schedule"; +import { Queue } from "bullmq"; +import { PrismaService } from "../../common/prisma.service"; +import { + CONTRIBUTION_DUE_JOB, + CONTRIBUTION_REMINDERS_QUEUE, + ReminderJobData, +} from "./reminder.queue"; + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** + * Scans active groups once a day and enqueues a reminder for every member + * who has not yet contributed in the current period, once the due date is + * within the configured lead window. + * + * The contribution period length is not stored per-group in the schema yet, + * so it is configurable via `CONTRIBUTION_PERIOD_DAYS` (default 30) and the + * current period is derived from each group's `createdAt`. + */ +@Injectable() +export class ReminderScheduler { + private readonly logger = new Logger(ReminderScheduler.name); + private readonly periodDays = Number( + process.env.CONTRIBUTION_PERIOD_DAYS ?? 30, + ); + private readonly reminderLeadDays = Number( + process.env.CONTRIBUTION_REMINDER_LEAD_DAYS ?? 3, + ); + + constructor( + @InjectQueue(CONTRIBUTION_REMINDERS_QUEUE) + private readonly queue: Queue, + private readonly prisma: PrismaService, + ) {} + + @Cron(CronExpression.EVERY_DAY_AT_9AM) + async enqueueDueReminders(now: Date = new Date()): Promise { + const groups = await this.prisma.group.findMany({ + where: { isActive: true }, + include: { members: { where: { isActive: true } } }, + }); + + let enqueued = 0; + for (const group of groups) { + const { period, dueDate } = this.currentPeriod(group.createdAt, now); + const daysUntilDue = (dueDate.getTime() - now.getTime()) / DAY_MS; + if (daysUntilDue < 0 || daysUntilDue > this.reminderLeadDays) continue; + + const contributions = await this.prisma.contribution.findMany({ + where: { groupId: group.id, period }, + select: { memberAddress: true }, + }); + const contributed = new Set(contributions.map((c) => c.memberAddress)); + + for (const member of group.members) { + if (contributed.has(member.address)) continue; + await this.queue.add( + CONTRIBUTION_DUE_JOB, + { + groupId: group.id, + memberAddress: member.address, + period, + dueDate: dueDate.toISOString(), + }, + { + // Deterministic id → one reminder per member per period even if + // the cron runs on several days inside the lead window. + jobId: `${group.id}:${member.address}:${period}`, + // Keep the completed job (and its id) around for the whole period + // so the dedupe above holds, then let it expire. + removeOnComplete: { age: Math.ceil(this.periodDays * 24 * 60 * 60) }, + removeOnFail: { age: 7 * 24 * 60 * 60 }, + }, + ); + enqueued++; + } + } + + const counts = await this.queue.getJobCounts(); + this.logger.log( + `Enqueued ${enqueued} contribution reminder(s); queue counts: ${JSON.stringify(counts)}`, + ); + return enqueued; + } + + /** + * Derive the current 0-based contribution period and its due date from the + * group's creation date and the configured period length. + */ + private currentPeriod( + createdAt: Date, + now: Date, + ): { period: number; dueDate: Date } { + const elapsedDays = Math.max( + 0, + (now.getTime() - createdAt.getTime()) / DAY_MS, + ); + const period = Math.floor(elapsedDays / this.periodDays); + const dueDate = new Date( + createdAt.getTime() + (period + 1) * this.periodDays * DAY_MS, + ); + return { period, dueDate }; + } +}