feat(subscriptions): implement key-based subscription expiry notifica…

src/index.ts

@@ -48,6 +48,10 @@ import { createSecondaryMarketService } from "./services/secondary-market.servic
 import { createWatchlistService } from "./services/watchlist.service";
 import { createSettlementWorker } from "./workers/settlement.worker";
 import { scheduleAnalyticsSnapshotJob } from "./workers/analytics-snapshot.worker";
+import {
+  scheduleSubscriptionExpiryJob,
+  buildSubscriptionExpiryNotifier,
+} from "./workers/subscription-expiry.worker";
 import { createFeeScheduleService } from "./services/fee-schedule.service";
 import { getRedisClient } from "./config/redis";
 import { createIssuerReputationService } from "./services/issuer-reputation.service";
@@ -259,6 +263,14 @@ export async function bootstrap(): Promise<{
   // ---- Start daily analytics snapshot cron (midnight UTC) ----
   const snapshotScheduler = scheduleAnalyticsSnapshotJob(dataSource);
 
+  // ---- Start daily subscription expiry notification check (midnight UTC) ----
+  const subscriptionExpiryNotifier = buildSubscriptionExpiryNotifier(
+    dataSource,
+    createNotificationService(dataSource),
+    logger
+  );
+  const subscriptionExpiryScheduler = scheduleSubscriptionExpiryJob(subscriptionExpiryNotifier, logger);
+
   // ---- Start settlement worker cron (hourly) ----
   settlementWorker.start("0 * * * *");
 
@@ -303,6 +315,7 @@ export async function bootstrap(): Promise<{
     }
     void maturityWorker.stop();
     snapshotScheduler.stop();
+    subscriptionExpiryScheduler.stop();
     settlementWorker.stop();
   });
 

src/lib/invoice-notifications.ts

@@ -20,6 +20,7 @@ export interface NotificationInput {
   title: string;
   message: string;
   data?: Record<string, unknown>;
+  dedupeKey?: string | null;
 }
 
 /** Satisfied by `NotificationService.createNotifications`. */

src/middleware/auth.middleware.ts

@@ -355,20 +355,23 @@ export function requireSeller() {
   };
 }
 
-export function requireAdmin(adminWallets: string[]) {
+export function requireAdmin(adminWallets: string[] = []) {
   return (req: Request, _res: Response, next: NextFunction): void => {
     const user = (req as AuthenticatedRequest).user;
     if (!user) {
       next(new HttpError(401, "Authentication required"));
       return;
     }
 
-    if (!adminWallets.includes(user.stellarAddress)) {
-      next(new AppError(403, "Admin privileges required", "ADMIN_REQUIRED"));
+    if (
+      user.userType === UserType.ADMIN ||
+      (adminWallets.length > 0 && adminWallets.includes(user.stellarAddress))
+    ) {
+      next();
       return;
     }
 
-    next();
+    next(new AppError(403, "Admin privileges required", "ADMIN_REQUIRED"));
   };
 }
 

src/migrations/1754000000000-AddSubscriptionNotificationTypes.ts

@@ -0,0 +1,20 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+/**
+ * Adds the subscription lifecycle notification types to the PostgreSQL enum.
+ */
+export class AddSubscriptionNotificationTypes1754000000000 implements MigrationInterface {
+  name = "AddSubscriptionNotificationTypes1754000000000";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    for (const value of ["subscription", "subscription_expiring"]) {
+      await queryRunner.query(
+        `ALTER TYPE "public"."notifications_notificationtype_enum" ADD VALUE IF NOT EXISTS '${value}';`
+      );
+    }
+  }
+
+  public async down(): Promise<void> {
+    // Postgres cannot drop enum values; rows may already use them.
+  }
+}

src/migrations/1755000000000-CreateKeySubscriptionsTable.ts

@@ -0,0 +1,116 @@
+import { MigrationInterface, QueryRunner, Table, TableIndex } from "typeorm";
+
+export class CreateKeySubscriptionsTable1755000000000 implements MigrationInterface {
+  name = "CreateKeySubscriptionsTable1755000000000";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.createTable(
+      new Table({
+        name: "key_subscriptions",
+        columns: [
+          {
+            name: "id",
+            type: "uuid",
+            isPrimary: true,
+            isGenerated: true,
+            generationStrategy: "uuid",
+          },
+          {
+            name: "wallet",
+            type: "varchar",
+            length: "56",
+            isNullable: false,
+          },
+          {
+            name: "key_id",
+            type: "varchar",
+            length: "128",
+            isNullable: false,
+          },
+          {
+            name: "key_name",
+            type: "varchar",
+            length: "255",
+            isNullable: true,
+          },
+          {
+            name: "min_holding",
+            type: "decimal",
+            precision: 30,
+            scale: 7,
+            default: "'0'",
+          },
+          {
+            name: "current_balance",
+            type: "decimal",
+            precision: 30,
+            scale: 7,
+            isNullable: true,
+          },
+          {
+            name: "expiry_ledger",
+            type: "integer",
+            isNullable: true,
+          },
+          {
+            name: "expiry_date",
+            type: "timestamptz",
+            isNullable: true,
+          },
+          {
+            name: "last_notified_cycle",
+            type: "varchar",
+            length: "255",
+            isNullable: true,
+          },
+          {
+            name: "last_notified_window",
+            type: "varchar",
+            length: "32",
+            isNullable: true,
+          },
+          {
+            name: "is_active",
+            type: "boolean",
+            default: true,
+          },
+          {
+            name: "created_at",
+            type: "timestamptz",
+            default: "CURRENT_TIMESTAMP",
+          },
+          {
+            name: "updated_at",
+            type: "timestamptz",
+            default: "CURRENT_TIMESTAMP",
+          },
+        ],
+      }),
+      true
+    );
+
+    await queryRunner.createIndices("key_subscriptions", [
+      new TableIndex({
+        name: "idx_key_subscriptions_wallet",
+        columnNames: ["wallet"],
+      }),
+      new TableIndex({
+        name: "idx_key_subscriptions_key_id",
+        columnNames: ["key_id"],
+      }),
+      new TableIndex({
+        name: "idx_key_subscriptions_expiry_date",
+        columnNames: ["expiry_date"],
+      }),
+      new TableIndex({
+        name: "idx_key_subscriptions_wallet_key",
+        columnNames: ["wallet", "key_id"],
+        isUnique: true,
+      }),
+    ]);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.dropTable("key_subscriptions", true);
+  }
+}

src/models/KeySubscription.model.ts

@@ -0,0 +1,67 @@
+import {
+  Entity,
+  PrimaryGeneratedColumn,
+  Column,
+  CreateDateColumn,
+  UpdateDateColumn,
+  Index,
+} from "typeorm";
+
+/**
+ * Tracks a subscriber's key-based subscription for status and expiry notification dispatch.
+ */
+@Entity("key_subscriptions")
+@Index("idx_key_subscriptions_wallet_key", ["wallet", "keyId"], { unique: true })
+export class KeySubscription {
+  @PrimaryGeneratedColumn("uuid")
+  id!: string;
+
+  /** Subscriber wallet address (Stellar G-address) */
+  @Column({ name: "wallet", type: "varchar", length: 56 })
+  @Index("idx_key_subscriptions_wallet")
+  wallet!: string;
+
+  /** Gated key identifier or Soroban contract address */
+  @Column({ name: "key_id", type: "varchar", length: 128 })
+  @Index("idx_key_subscriptions_key_id")
+  keyId!: string;
+
+  /** Human-readable key name (e.g. "VIP Access Pass") */
+  @Column({ name: "key_name", type: "varchar", length: 255, nullable: true })
+  keyName!: string | null;
+
+  /** Minimum token holding required to keep the subscription active */
+  @Column({ name: "min_holding", type: "decimal", precision: 30, scale: 7, default: "0" })
+  minHolding!: string;
+
+  /** Cached or last observed token balance of the subscriber */
+  @Column({ name: "current_balance", type: "decimal", precision: 30, scale: 7, nullable: true })
+  currentBalance!: string | null;
+
+  /** Ledger at which access lapses; null when subscription does not expire */
+  @Column({ name: "expiry_ledger", type: "integer", nullable: true })
+  expiryLedger!: number | null;
+
+  /** Expiry timestamp if known / computed */
+  @Column({ name: "expiry_date", type: "timestamptz", nullable: true })
+  @Index("idx_key_subscriptions_expiry_date")
+  expiryDate!: Date | null;
+
+  /** Last notified cycle identifier (prevents duplicate notification in same cycle) */
+  @Column({ name: "last_notified_cycle", type: "varchar", length: 255, nullable: true })
+  lastNotifiedCycle!: string | null;
+
+  /** Last notified window ("7d" or "24h") */
+  @Column({ name: "last_notified_window", type: "varchar", length: 32, nullable: true })
+  lastNotifiedWindow!: string | null;
+
+  /** True while the subscription is active */
+  @Column({ name: "is_active", type: "boolean", default: true })
+  isActive!: boolean;
+
+  @CreateDateColumn({ name: "created_at" })
+  createdAt!: Date;
+
+  @UpdateDateColumn({ name: "updated_at" })
+  updatedAt!: Date;
+}

src/routes/accreditation.routes.ts

@@ -33,7 +33,7 @@ export function createAccreditationRouter({
     auth,
     async (req: WalletParamsRequest, res: Response, next: NextFunction) => {
       try {
-        const { wallet } = req.params;
+        const wallet = String(req.params.wallet);
         const accreditation = await accreditationService.getAccreditation(wallet);
 
         res.json({
@@ -54,7 +54,7 @@ export function createAccreditationRouter({
     requireAdmin(adminWallets),
     async (req: WalletParamsRequest, res: Response, next: NextFunction) => {
       try {
-        const { wallet } = req.params;
+        const wallet = String(req.params.wallet);
         const { tier, verificationSource } = req.body;
 
         if (!Object.values(AccreditationTier).includes(tier)) {
@@ -84,7 +84,8 @@ export function createAccreditationRouter({
     auth,
     async (req: WalletEligibilityParamsRequest, res: Response, next: NextFunction) => {
       try {
-        const { wallet, invoiceId } = req.params;
+        const wallet = String(req.params.wallet);
+        const invoiceId = String(req.params.invoiceId);
         const eligibility = await accreditationService.checkEligibility(
           wallet,
           invoiceId

src/routes/secondary-market/price-history.routes.ts

@@ -26,7 +26,7 @@ export function createPriceHistoryRouter({
     auth,
     async (req: InvoicePriceHistoryRequest, res: Response, next: NextFunction) => {
       try {
-        const { invoiceId } = req.params;
+        const invoiceId = String(req.params.invoiceId);
         const range = (req.query.range as PriceRange) || "30d";
 
         if (!["7d", "30d", "all"].includes(range)) {

src/services/invoice-refund.service.ts

@@ -81,14 +81,14 @@
     }
   }
 
-  async getRefundableInvoices(walletAddress: string): Promise<Invoice[]> {
+  async getRefundableInvoices(_walletAddress: string): Promise<Invoice[]> {
     const now = new Date();
 
     // Scope to invoices this wallet actually invested in - without this
     // filter every investor's refund list leaked every other investor's
     // expired-but-unfunded invoices.
     const investments = await this.investmentRepo.find({
       where: { investorWallet: walletAddress },
     });
     const invoiceIds = [...new Set(investments.map((investment) => investment.invoiceId))];
     if (invoiceIds.length === 0) return [];

src/services/notification.service.ts

@@ -42,11 +42,13 @@ export interface NotificationRepositoryContract {
     type: NotificationType,
     title: string,
     message: string,
-    data?: Record<string, unknown>
+    data?: Record<string, unknown>,
+    dedupeKey?: string | null
   ): Promise<Notification>;
   /** Inserts all entries in a single statement. */
   createMany(entries: NotificationInput[]): Promise<void>;
   findByIdAndUserId(id: string, userId: string): Promise<Notification | null>;
+  findByDedupeKey?(dedupeKey: string): Promise<Notification | null>;
   markRead(id: string, userId: string): Promise<Notification>;
   /** Marks every unread notification of the user as read in one statement; returns how many changed. */
   markAllRead(userId: string, walletAddress?: string): Promise<number>;
@@ -69,9 +71,17 @@ export class NotificationService {
     type: NotificationType,
     title: string,
     message: string,
-    data?: Record<string, unknown>
+    data?: Record<string, unknown>,
+    dedupeKey?: string | null
   ): Promise<Notification> {
-    return this.notificationRepository.create(userId, type, title, message, data);
+    return this.notificationRepository.create(userId, type, title, message, data, dedupeKey);
+  }
+
+  async findByDedupeKey(dedupeKey: string): Promise<Notification | null> {
+    if (this.notificationRepository.findByDedupeKey) {
+      return this.notificationRepository.findByDedupeKey(dedupeKey);
+    }
+    return null;
   }
 
   /**
@@ -128,9 +138,17 @@ class TypeOrmNotificationRepository implements NotificationRepositoryContract {
     type: NotificationType,
     title: string,
     message: string,
-    data?: Record<string, unknown>
+    data?: Record<string, unknown>,
+    dedupeKey?: string | null
   ): Promise<Notification> {
-    const entity = this.repository.create({ userId, type, title, message, data });
+    const entity = this.repository.create({
+      userId,
+      type,
+      title,
+      message,
+      data,
+      ...(dedupeKey !== undefined ? { dedupeKey } : {}),
+    });
     return this.repository.save(entity);
   }
 
@@ -141,6 +159,7 @@ class TypeOrmNotificationRepository implements NotificationRepositoryContract {
       title: entry.title,
       message: entry.message,
       data: entry.data || null,
+      ...(entry.dedupeKey !== undefined ? { dedupeKey: entry.dedupeKey } : {}),
     }));
     await this.repository.save(entities);
   }
@@ -149,6 +168,10 @@ class TypeOrmNotificationRepository implements NotificationRepositoryContract {
     return this.repository.findOne({ where: { id, userId } });
   }
 
+  async findByDedupeKey(dedupeKey: string): Promise<Notification | null> {
+    return this.repository.findOne({ where: { dedupeKey } });
+  }
+
   async markAllRead(userId: string, walletAddress?: string): Promise<number> {
     const wallet = walletAddress && walletAddress !== userId ? walletAddress : null;
     const result = wallet