feat: contribution reminder scheduler with BullMQ (closes #4)

.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

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",

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,

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 {}

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<ReminderJobData>): Promise<void> {
+    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}).`,
+      );
+    }
+  }
+}

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,
+  };
+}

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();
+  });
+});

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<ReminderJobData>,
+    private readonly prisma: PrismaService,
+  ) {}
+
+  @Cron(CronExpression.EVERY_DAY_AT_9AM)
+  async enqueueDueReminders(now: Date = new Date()): Promise<number> {
+    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 };
+  }
+}