From de29b3993a5cc96c26e515871c5f2720e261fca7 Mon Sep 17 00:00:00 2001 From: dee-john Date: Sat, 17 Oct 2026 12:55:50 +0100 Subject: [PATCH 1/3] feat(subscriptions): implement key-based subscription expiry notification service - Add SubscriptionExpiryNotifierService for key-based subscription expiry alerts - Implement daily scheduled worker checking subscriptions expiring within 7 days - Dispatch 7-day warning and 24-hour warning notifications via NotificationService - Skip notification if wallet already meets minimum holding requirement or topped up - Include key name, expiry date, and minimum holding required in notification - Enforce deduplication preventing repeat notifications for same expiry cycle and window - Add KeySubscription model and migrations for schema and notification types - Add unit test suite covering all acceptance criteria --- src/index.ts | 13 + src/lib/invoice-notifications.ts | 1 + ...000000-AddSubscriptionNotificationTypes.ts | 20 + ...55000000000-CreateKeySubscriptionsTable.ts | 116 +++ src/models/KeySubscription.model.ts | 67 ++ src/services/notification.service.ts | 33 +- .../subscription-expiry-notifier.service.ts | 707 ++++++++++++++++++ src/types/enums.ts | 2 + src/workers/subscription-expiry.worker.ts | 136 ++++ .../settlement.integration.test.ts | 4 - tests/invoice.service.test.ts | 91 ++- .../subscription-expiry-notifier.test.ts | 654 ++++++++++++++++ 12 files changed, 1807 insertions(+), 37 deletions(-) create mode 100644 src/migrations/1754000000000-AddSubscriptionNotificationTypes.ts create mode 100644 src/migrations/1755000000000-CreateKeySubscriptionsTable.ts create mode 100644 src/models/KeySubscription.model.ts create mode 100644 src/services/subscription-expiry-notifier.service.ts create mode 100644 src/workers/subscription-expiry.worker.ts create mode 100644 tests/unit/services/subscription-expiry-notifier.test.ts diff --git a/src/index.ts b/src/index.ts index 01546bcf..38bac634 100644 --- a/src/index.ts +++ b/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"; export async function bootstrap(): Promise<{ server: Server; @@ -235,6 +239,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 * * * *"); @@ -260,6 +272,7 @@ export async function bootstrap(): Promise<{ } void maturityWorker.stop(); snapshotScheduler.stop(); + subscriptionExpiryScheduler.stop(); settlementWorker.stop(); }); diff --git a/src/lib/invoice-notifications.ts b/src/lib/invoice-notifications.ts index c9dbbb24..82045cef 100644 --- a/src/lib/invoice-notifications.ts +++ b/src/lib/invoice-notifications.ts @@ -20,6 +20,7 @@ export interface NotificationInput { title: string; message: string; data?: Record; + dedupeKey?: string | null; } /** Satisfied by `NotificationService.createNotifications`. */ diff --git a/src/migrations/1754000000000-AddSubscriptionNotificationTypes.ts b/src/migrations/1754000000000-AddSubscriptionNotificationTypes.ts new file mode 100644 index 00000000..fb6a4070 --- /dev/null +++ b/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 { + 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 { + // Postgres cannot drop enum values; rows may already use them. + } +} diff --git a/src/migrations/1755000000000-CreateKeySubscriptionsTable.ts b/src/migrations/1755000000000-CreateKeySubscriptionsTable.ts new file mode 100644 index 00000000..0f6d69d6 --- /dev/null +++ b/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 { + 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 { + await queryRunner.dropTable("key_subscriptions", true); + } +} diff --git a/src/models/KeySubscription.model.ts b/src/models/KeySubscription.model.ts new file mode 100644 index 00000000..ca96b1a6 --- /dev/null +++ b/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; +} diff --git a/src/services/notification.service.ts b/src/services/notification.service.ts index 037ef0e3..24b68146 100644 --- a/src/services/notification.service.ts +++ b/src/services/notification.service.ts @@ -42,11 +42,13 @@ export interface NotificationRepositoryContract { type: NotificationType, title: string, message: string, - data?: Record + data?: Record, + dedupeKey?: string | null ): Promise; /** Inserts all entries in a single statement. */ createMany(entries: NotificationInput[]): Promise; findByIdAndUserId(id: string, userId: string): Promise; + findByDedupeKey?(dedupeKey: string): Promise; markRead(id: string, userId: string): Promise; /** Marks every unread notification of the user as read in one statement; returns how many changed. */ markAllRead(userId: string, walletAddress?: string): Promise; @@ -69,9 +71,17 @@ export class NotificationService { type: NotificationType, title: string, message: string, - data?: Record + data?: Record, + dedupeKey?: string | null ): Promise { - return this.notificationRepository.create(userId, type, title, message, data); + return this.notificationRepository.create(userId, type, title, message, data, dedupeKey); + } + + async findByDedupeKey(dedupeKey: string): Promise { + 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 + data?: Record, + dedupeKey?: string | null ): Promise { - 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 { + return this.repository.findOne({ where: { dedupeKey } }); + } + async markAllRead(userId: string, walletAddress?: string): Promise { const wallet = walletAddress && walletAddress !== userId ? walletAddress : null; const result = wallet diff --git a/src/services/subscription-expiry-notifier.service.ts b/src/services/subscription-expiry-notifier.service.ts new file mode 100644 index 00000000..4a09756a --- /dev/null +++ b/src/services/subscription-expiry-notifier.service.ts @@ -0,0 +1,707 @@ +import { randomUUID } from "crypto"; +import Decimal from "decimal.js"; +import type { Repository } from "typeorm"; +import { logger as globalLogger, type AppLogger } from "../observability/logger"; +import { NotificationType } from "../types/enums"; +import type { User } from "../models/User.model"; +import { KeySubscription } from "../models/KeySubscription.model"; +import { + LEDGERS_PER_DAY, + type KeyHoldingReading, + type SubscriptionHoldingReader, + type SubscriptionStatusService, +} from "./subscription-status.service"; +import type { CreatorKeyRepositoryContract } from "./creator-key.service"; + +export type ExpiryWarningWindow = "7d" | "24h"; + +export const MS_PER_DAY = 24 * 60 * 60 * 1000; +export const MS_PER_HOUR = 60 * 60 * 1000; +export const SEVEN_DAYS_MS = 7 * MS_PER_DAY; +export const TWENTY_FOUR_HOURS_MS = 24 * MS_PER_HOUR; + +export interface SubscriptionCandidate { + wallet: string; + keyId: string; + keyName?: string | null; + expiryDate?: Date | string | null; + expiryLedger?: number | null; + minHolding?: string | number | Decimal; + minBalance?: string | number | Decimal; + currentBalance?: string | number | Decimal | null; + balance?: string | number | Decimal | null; + cycleId?: string | null; + hasToppedUp?: boolean; + toppedUp?: boolean; + meetsMinimumHolding?: boolean; +} + +export interface SubscriptionCandidateSource { + findExpiringCandidates(options?: { + withinDays?: number; + currentLedger?: number; + currentDate?: Date; + }): Promise; + recordNotificationSent?(input: { + wallet: string; + keyId: string; + cycleId: string; + window: ExpiryWarningWindow; + sentAt: Date; + }): Promise; +} + +export interface SubscriptionNotificationSink { + createNotification( + userId: string, + type: NotificationType, + title: string, + message: string, + data?: Record, + dedupeKey?: string | null + ): Promise; + findByDedupeKey?(dedupeKey: string): Promise; +} + +export interface SubscriptionDedupeStore { + hasBeenSent(dedupeKey: string): Promise; + recordSent(dedupeKey: string): Promise; +} + +export interface SubscriptionExpiryNotifierDependencies { + notificationService: SubscriptionNotificationSink; + subscriptionSource?: SubscriptionCandidateSource; + subscriptionStatusService?: SubscriptionStatusService; + holdingReader?: SubscriptionHoldingReader; + creatorKeyRepository?: CreatorKeyRepositoryContract; + userRepository?: Repository; + dedupeStore?: SubscriptionDedupeStore; + ledgersPerDay?: number; + now?: () => Date; + logger?: AppLogger; +} + +export type CheckOutcomeStatus = + | "notified" + | "skipped_meets_minimum_holding" + | "skipped_duplicate" + | "skipped_outside_window" + | "skipped_already_expired" + | "error"; + +export interface SubscriptionCheckResult { + wallet: string; + keyId: string; + keyName: string; + window?: ExpiryWarningWindow; + cycleId?: string; + status: CheckOutcomeStatus; + reason?: string; + currentBalance?: string; + minHolding?: string; + expiryDate?: string; + dedupeKey?: string; + notificationId?: unknown; +} + +export interface SubscriptionExpiryRunResult { + runId: string; + totalChecked: number; + notified7d: number; + notified24h: number; + skippedMeetsHolding: number; + skippedDuplicate: number; + skippedOutsideWindow: number; + skippedExpired: number; + errors: Array<{ wallet: string; keyId: string; error: string }>; + durationMs: number; + results: SubscriptionCheckResult[]; +} + +export class InMemorySubscriptionDedupeStore implements SubscriptionDedupeStore { + private readonly sentKeys = new Set(); + + async hasBeenSent(dedupeKey: string): Promise { + return this.sentKeys.has(dedupeKey); + } + + async recordSent(dedupeKey: string): Promise { + this.sentKeys.add(dedupeKey); + } + + clear(): void { + this.sentKeys.clear(); + } +} + +/** + * Service that alerts subscribers when their key-based subscription is approaching expiry, + * prompting them to maintain their minimum holding. + * + * Implements: + * 1. 7-day warning notification dispatch + * 2. 24-hour warning notification dispatch + * 3. Notification skipped if wallet already meets minimum holding / topped up + * 4. Content includes key name, expiry date, and minimum holding required + * 5. Deduplication preventing multiple alerts for the same window & expiry cycle + */ +export class SubscriptionExpiryNotifierService { + private readonly notificationService: SubscriptionNotificationSink; + private readonly subscriptionSource?: SubscriptionCandidateSource; + private readonly subscriptionStatusService?: SubscriptionStatusService; + private readonly holdingReader?: SubscriptionHoldingReader; + private readonly creatorKeyRepository?: CreatorKeyRepositoryContract; + private readonly userRepository?: Repository; + private readonly dedupeStore: SubscriptionDedupeStore; + private readonly ledgersPerDay: number; + private readonly now: () => Date; + private readonly logger: AppLogger; + + constructor(deps: SubscriptionExpiryNotifierDependencies) { + this.notificationService = deps.notificationService; + this.subscriptionSource = deps.subscriptionSource; + this.subscriptionStatusService = deps.subscriptionStatusService; + this.holdingReader = deps.holdingReader; + this.creatorKeyRepository = deps.creatorKeyRepository; + this.userRepository = deps.userRepository; + this.dedupeStore = deps.dedupeStore ?? new InMemorySubscriptionDedupeStore(); + this.ledgersPerDay = deps.ledgersPerDay && deps.ledgersPerDay > 0 ? deps.ledgersPerDay : LEDGERS_PER_DAY; + this.now = deps.now ?? (() => new Date()); + this.logger = deps.logger ?? globalLogger; + } + + /** + * Evaluates a single candidate subscription, determining whether a 7-day or 24-hour + * warning should be dispatched or skipped. + */ + async checkSubscription(candidate: SubscriptionCandidate): Promise { + const now = this.now(); + const keyName = await this.resolveKeyName(candidate); + const { expiryDate, window, hoursRemaining, daysRemaining } = this.calculateExpiryWindow(candidate, now); + + if (hoursRemaining <= 0) { + this.logger.debug("Subscription already expired; skipping approaching expiry warning", { + wallet: candidate.wallet, + keyId: candidate.keyId, + hoursRemaining, + }); + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + status: "skipped_already_expired", + reason: "Subscription has already expired", + expiryDate: expiryDate ? expiryDate.toISOString() : undefined, + }; + } + + if (!window) { + this.logger.debug("Subscription outside 7-day warning window; skipping", { + wallet: candidate.wallet, + keyId: candidate.keyId, + daysRemaining, + }); + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + status: "skipped_outside_window", + reason: `Expires in ${daysRemaining.toFixed(1)} days; outside 7-day window`, + expiryDate: expiryDate ? expiryDate.toISOString() : undefined, + }; + } + + // Check if wallet meets minimum holding requirement (or topped up) + const holdingCheck = await this.verifyHolding(candidate); + if (holdingCheck.meetsHolding) { + this.logger.info("Subscriber already meets minimum holding; skipping notification", { + wallet: candidate.wallet, + keyId: candidate.keyId, + currentBalance: holdingCheck.currentBalance.toString(), + minHolding: holdingCheck.minHolding.toString(), + }); + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + window, + status: "skipped_meets_minimum_holding", + reason: "Wallet already meets minimum holding", + currentBalance: holdingCheck.currentBalance.toString(), + minHolding: holdingCheck.minHolding.toString(), + expiryDate: expiryDate ? expiryDate.toISOString() : undefined, + }; + } + + // Determine cycle identifier for deduplication + const cycleId = this.resolveCycleId(candidate, expiryDate); + const dedupeKey = `sub_expiry:${candidate.wallet}:${candidate.keyId}:${cycleId}:${window}`; + + // Deduplication check: prevent duplicate notifications for same cycle & window + const alreadySent = await this.isDuplicate(dedupeKey); + if (alreadySent) { + this.logger.info("Notification already sent for this cycle and window; skipping duplicate", { + wallet: candidate.wallet, + keyId: candidate.keyId, + cycleId, + window, + dedupeKey, + }); + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + window, + cycleId, + dedupeKey, + status: "skipped_duplicate", + reason: `Duplicate notification prevented for cycle ${cycleId} and window ${window}`, + currentBalance: holdingCheck.currentBalance.toString(), + minHolding: holdingCheck.minHolding.toString(), + expiryDate: expiryDate ? expiryDate.toISOString() : undefined, + }; + } + + // Build notification content including key name, expiry date, and minimum holding required + const formattedExpiry = expiryDate ? expiryDate.toISOString() : "approaching ledger expiry"; + const minHoldingStr = holdingCheck.minHolding.toString(); + const currentBalanceStr = holdingCheck.currentBalance.toString(); + + const title = + window === "24h" + ? `Urgent: Subscription Expiring in 24 Hours - ${keyName}` + : `Subscription Expiring Soon: ${keyName}`; + + const message = + window === "24h" + ? `Urgent: Your subscription for key "${keyName}" expires in less than 24 hours on ${formattedExpiry}. Maintain a minimum holding of ${minHoldingStr} tokens to prevent service disruption.` + : `Your subscription for key "${keyName}" expires on ${formattedExpiry}. Maintain a minimum holding of ${minHoldingStr} tokens to retain uninterrupted access.`; + + const data: Record = { + keyId: candidate.keyId, + keyName, + expiryDate: formattedExpiry, + expiry: formattedExpiry, + minHolding: minHoldingStr, + minimumHolding: minHoldingStr, + currentBalance: currentBalanceStr, + window, + cycleId, + daysRemaining: Math.max(0, Math.ceil(daysRemaining)), + hoursRemaining: Math.max(0, Math.ceil(hoursRemaining)), + }; + + const recipientUserId = await this.resolveRecipientUserId(candidate.wallet); + + let notification: unknown = null; + try { + notification = await this.notificationService.createNotification( + recipientUserId, + NotificationType.SUBSCRIPTION, + title, + message, + data, + dedupeKey + ); + } catch (err: unknown) { + // If a database unique constraint prevented duplicate insertion concurrently, treat as duplicate + const errMsg = err instanceof Error ? err.message : String(err); + if (errMsg.includes("duplicate") || errMsg.includes("unique") || errMsg.includes("23505")) { + this.logger.warn("Unique constraint captured duplicate notification concurrently", { + dedupeKey, + error: errMsg, + }); + await this.recordSent(dedupeKey); + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + window, + cycleId, + dedupeKey, + status: "skipped_duplicate", + reason: "Duplicate notification prevented by database constraint", + currentBalance: currentBalanceStr, + minHolding: minHoldingStr, + expiryDate: formattedExpiry, + }; + } + throw err; + } + + await this.recordSent(dedupeKey); + + if (this.subscriptionSource?.recordNotificationSent) { + try { + await this.subscriptionSource.recordNotificationSent({ + wallet: candidate.wallet, + keyId: candidate.keyId, + cycleId, + window, + sentAt: now, + }); + } catch (err) { + this.logger.warn("Failed to record notification sent in subscription source", { + wallet: candidate.wallet, + keyId: candidate.keyId, + error: err instanceof Error ? err.message : String(err), + }); + } + } + + this.logger.info("Subscription expiry notification dispatched successfully", { + wallet: candidate.wallet, + keyId: candidate.keyId, + window, + cycleId, + keyName, + minHolding: minHoldingStr, + formattedExpiry, + }); + + return { + wallet: candidate.wallet, + keyId: candidate.keyId, + keyName, + window, + cycleId, + dedupeKey, + status: "notified", + currentBalance: currentBalanceStr, + minHolding: minHoldingStr, + expiryDate: formattedExpiry, + notificationId: notification, + }; + } + + /** + * Processes a batch of candidates or discovers expiring subscriptions from the source. + */ + async processExpiringSubscriptions( + candidates?: SubscriptionCandidate[] + ): Promise { + const startedAt = Date.now(); + const runId = randomUUID(); + + const list = + candidates ?? + (this.subscriptionSource + ? await this.subscriptionSource.findExpiringCandidates({ withinDays: 7, currentDate: this.now() }) + : []); + + const results: SubscriptionCheckResult[] = []; + const errors: Array<{ wallet: string; keyId: string; error: string }> = []; + + let notified7d = 0; + let notified24h = 0; + let skippedMeetsHolding = 0; + let skippedDuplicate = 0; + let skippedOutsideWindow = 0; + let skippedExpired = 0; + + for (const candidate of list) { + try { + const result = await this.checkSubscription(candidate); + results.push(result); + + if (result.status === "notified") { + if (result.window === "24h") notified24h += 1; + else if (result.window === "7d") notified7d += 1; + } else if (result.status === "skipped_meets_minimum_holding") { + skippedMeetsHolding += 1; + } else if (result.status === "skipped_duplicate") { + skippedDuplicate += 1; + } else if (result.status === "skipped_outside_window") { + skippedOutsideWindow += 1; + } else if (result.status === "skipped_already_expired") { + skippedExpired += 1; + } + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + this.logger.error("Error evaluating subscription candidate", { + wallet: candidate.wallet, + keyId: candidate.keyId, + error: message, + }); + errors.push({ wallet: candidate.wallet, keyId: candidate.keyId, error: message }); + } + } + + const durationMs = Date.now() - startedAt; + + this.logger.info("Subscription expiry check batch complete", { + runId, + totalChecked: list.length, + notified7d, + notified24h, + skippedMeetsHolding, + skippedDuplicate, + skippedOutsideWindow, + skippedExpired, + errorCount: errors.length, + durationMs, + }); + + return { + runId, + totalChecked: list.length, + notified7d, + notified24h, + skippedMeetsHolding, + skippedDuplicate, + skippedOutsideWindow, + skippedExpired, + errors, + durationMs, + results, + }; + } + + /** + * Scheduled daily job entry point. + */ + async runDailyCheck(): Promise { + return this.processExpiringSubscriptions(); + } + + private calculateExpiryWindow( + candidate: SubscriptionCandidate, + now: Date + ): { + expiryDate: Date | null; + window: ExpiryWarningWindow | null; + hoursRemaining: number; + daysRemaining: number; + } { + let expiryDate: Date | null = null; + let hoursRemaining = 0; + let daysRemaining = 0; + + if (candidate.expiryDate) { + expiryDate = candidate.expiryDate instanceof Date ? candidate.expiryDate : new Date(candidate.expiryDate); + const diffMs = expiryDate.getTime() - now.getTime(); + hoursRemaining = diffMs / MS_PER_HOUR; + daysRemaining = diffMs / MS_PER_DAY; + } else if (Number.isFinite(candidate.expiryLedger as number)) { + const expiryLedger = candidate.expiryLedger as number; + // If current ledger is known, compute based on ~5s per ledger + const currentLedger = (candidate as { currentLedger?: number }).currentLedger ?? 0; + const ledgersRemaining = expiryLedger - currentLedger; + daysRemaining = ledgersRemaining / this.ledgersPerDay; + hoursRemaining = daysRemaining * 24; + + const estimatedMs = Math.round(daysRemaining * MS_PER_DAY); + expiryDate = new Date(now.getTime() + estimatedMs); + } + + let window: ExpiryWarningWindow | null = null; + if (hoursRemaining > 0 && hoursRemaining <= 24) { + window = "24h"; + } else if (daysRemaining > 0 && daysRemaining <= 7) { + window = "7d"; + } + + return { expiryDate, window, hoursRemaining, daysRemaining }; + } + + private async verifyHolding(candidate: SubscriptionCandidate): Promise<{ + meetsHolding: boolean; + currentBalance: Decimal; + minHolding: Decimal; + }> { + // Explicit override flags + if (candidate.meetsMinimumHolding === true || candidate.hasToppedUp === true || candidate.toppedUp === true) { + const min = this.toDecimal(candidate.minHolding ?? candidate.minBalance ?? "0"); + const bal = this.toDecimal(candidate.currentBalance ?? candidate.balance ?? min); + return { meetsHolding: true, currentBalance: bal, minHolding: min }; + } + + let currentBalance: Decimal | null = null; + let minHolding: Decimal | null = null; + + if (candidate.currentBalance !== undefined && candidate.currentBalance !== null) { + currentBalance = this.toDecimal(candidate.currentBalance); + } else if (candidate.balance !== undefined && candidate.balance !== null) { + currentBalance = this.toDecimal(candidate.balance); + } + + if (candidate.minHolding !== undefined && candidate.minHolding !== null) { + minHolding = this.toDecimal(candidate.minHolding); + } else if (candidate.minBalance !== undefined && candidate.minBalance !== null) { + minHolding = this.toDecimal(candidate.minBalance); + } + + // If balance or minHolding was not provided on candidate, query live holding reader if present + if (currentBalance === null || minHolding === null) { + if (this.holdingReader) { + try { + const reading: KeyHoldingReading = await this.holdingReader.readHolding({ + wallet: candidate.wallet, + keyId: candidate.keyId, + }); + if (currentBalance === null) currentBalance = this.toDecimal(reading.balance); + if (minHolding === null) minHolding = this.toDecimal(reading.minBalance); + } catch (err) { + this.logger.warn("Failed to read on-chain holding", { + wallet: candidate.wallet, + keyId: candidate.keyId, + error: err instanceof Error ? err.message : String(err), + }); + } + } else if (this.subscriptionStatusService) { + try { + const status = await this.subscriptionStatusService.getStatus({ + wallet: candidate.wallet, + keyId: candidate.keyId, + }); + if (currentBalance === null) currentBalance = this.toDecimal(status.balance); + if (minHolding === null) minHolding = this.toDecimal(status.minBalance); + } catch (err) { + this.logger.warn("Failed to get subscription status", { + wallet: candidate.wallet, + keyId: candidate.keyId, + error: err instanceof Error ? err.message : String(err), + }); + } + } + } + + const finalBalance = currentBalance ?? new Decimal(0); + const finalMin = minHolding ?? new Decimal(0); + + const meetsHolding = finalBalance.gte(finalMin); + + return { meetsHolding, currentBalance: finalBalance, minHolding: finalMin }; + } + + private resolveCycleId(candidate: SubscriptionCandidate, expiryDate: Date | null): string { + if (candidate.cycleId) { + return candidate.cycleId; + } + if (Number.isFinite(candidate.expiryLedger as number)) { + return String(candidate.expiryLedger); + } + if (expiryDate) { + return expiryDate.toISOString().slice(0, 10); + } + return "default-cycle"; + } + + private async isDuplicate(dedupeKey: string): Promise { + if (await this.dedupeStore.hasBeenSent(dedupeKey)) { + return true; + } + if (this.notificationService.findByDedupeKey) { + try { + const found = await this.notificationService.findByDedupeKey(dedupeKey); + if (found) { + await this.dedupeStore.recordSent(dedupeKey); + return true; + } + } catch { + // Fall back to false if lookup fails + } + } + return false; + } + + private async recordSent(dedupeKey: string): Promise { + await this.dedupeStore.recordSent(dedupeKey); + } + + private async resolveKeyName(candidate: SubscriptionCandidate): Promise { + if (candidate.keyName && candidate.keyName.trim()) { + return candidate.keyName.trim(); + } + if (this.creatorKeyRepository) { + try { + const key = + (await this.creatorKeyRepository.findById(candidate.keyId)) ?? + (await this.creatorKeyRepository.findByContractAddress(candidate.keyId)); + if (key?.name) { + return key.name; + } + } catch { + // Fall back to keyId + } + } + return candidate.keyId; + } + + private async resolveRecipientUserId(wallet: string): Promise { + if (this.userRepository) { + try { + const user = await this.userRepository.findOne({ where: { stellarAddress: wallet } }); + if (user?.id) { + return user.id; + } + } catch { + // Fall back to wallet address + } + } + return wallet; + } + + private toDecimal(value: string | number | bigint | Decimal | null | undefined): Decimal { + try { + if (value === null || value === undefined) return new Decimal(0); + if (value instanceof Decimal) return value; + if (typeof value === "bigint") return new Decimal(value.toString()); + if (typeof value === "number") return Number.isFinite(value) ? new Decimal(value) : new Decimal(0); + return new Decimal(String(value).trim() || "0"); + } catch { + return new Decimal(0); + } + } +} + +/** + * Creates a TypeORM-backed candidate source that reads subscriptions from the database. + */ +export function createTypeOrmSubscriptionSource( + repository: Repository, + now: () => Date = () => new Date() +): SubscriptionCandidateSource { + return { + async findExpiringCandidates(options = {}) { + const withinDays = options.withinDays ?? 7; + const currentDate = options.currentDate ?? now(); + const maxExpiry = new Date(currentDate.getTime() + withinDays * MS_PER_DAY); + + const rows = await repository + .createQueryBuilder("sub") + .where("sub.isActive = :isActive", { isActive: true }) + .andWhere("sub.expiryDate IS NOT NULL") + .andWhere("sub.expiryDate > :currentDate", { currentDate }) + .andWhere("sub.expiryDate <= :maxExpiry", { maxExpiry }) + .getMany(); + + return rows.map((r) => ({ + wallet: r.wallet, + keyId: r.keyId, + keyName: r.keyName, + minHolding: r.minHolding, + currentBalance: r.currentBalance, + expiryDate: r.expiryDate, + expiryLedger: r.expiryLedger, + cycleId: r.lastNotifiedCycle, + })); + }, + async recordNotificationSent(input) { + const existing = await repository.findOne({ + where: { wallet: input.wallet, keyId: input.keyId }, + }); + if (existing) { + existing.lastNotifiedCycle = input.cycleId; + existing.lastNotifiedWindow = input.window; + await repository.save(existing); + } + }, + }; +} + +export function createSubscriptionExpiryNotifierService( + dependencies: SubscriptionExpiryNotifierDependencies +): SubscriptionExpiryNotifierService { + return new SubscriptionExpiryNotifierService(dependencies); +} diff --git a/src/types/enums.ts b/src/types/enums.ts index 15aadf00..118d881f 100644 --- a/src/types/enums.ts +++ b/src/types/enums.ts @@ -79,6 +79,8 @@ export enum NotificationType { INVOICE_MATURED = "invoice_matured", SETTLEMENT_RECEIVED = "settlement_received", REFUND_PROCESSED = "refund_processed", + SUBSCRIPTION = "subscription", + SUBSCRIPTION_EXPIRING = "subscription_expiring", } /** Seller funding-deadline extension request lifecycle (issue #477). */ diff --git a/src/workers/subscription-expiry.worker.ts b/src/workers/subscription-expiry.worker.ts new file mode 100644 index 00000000..7b41a1a2 --- /dev/null +++ b/src/workers/subscription-expiry.worker.ts @@ -0,0 +1,136 @@ +import { DataSource } from "typeorm"; +import { logger as defaultLogger, type AppLogger } from "../observability/logger"; +import { KeySubscription } from "../models/KeySubscription.model"; +import { User } from "../models/User.model"; +import { + SubscriptionExpiryNotifierService, + createTypeOrmSubscriptionSource, + type SubscriptionNotificationSink, + type SubscriptionExpiryRunResult, +} from "../services/subscription-expiry-notifier.service"; +import { createCreatorKeyRepository } from "../services/onchain-projection-repositories"; + +export interface SubscriptionExpiryWorkerDependencies { + notifierService: SubscriptionExpiryNotifierService; + logger?: AppLogger; + now?: () => Date; +} + +/** + * Runs the daily subscription expiry notification check once. + * Scans active subscriptions expiring within 7 days and dispatches 7-day and 24-hour warnings. + */ +export async function runSubscriptionExpiryJob( + notifier: SubscriptionExpiryNotifierService, + logger: AppLogger = defaultLogger +): Promise { + const startedAt = Date.now(); + const jobDate = new Date().toISOString().slice(0, 10); + + logger.info("SubscriptionExpiryJob: Starting daily check", { jobDate }); + + try { + const result = await notifier.runDailyCheck(); + + const durationMs = Date.now() - startedAt; + logger.info("SubscriptionExpiryJob: Completed successfully", { + jobDate, + durationMs, + totalChecked: result.totalChecked, + notified7d: result.notified7d, + notified24h: result.notified24h, + skippedMeetsHolding: result.skippedMeetsHolding, + skippedDuplicate: result.skippedDuplicate, + skippedOutsideWindow: result.skippedOutsideWindow, + skippedExpired: result.skippedExpired, + errorsCount: result.errors.length, + }); + + return result; + } catch (error) { + const durationMs = Date.now() - startedAt; + logger.error("SubscriptionExpiryJob: Failed", { + jobDate, + durationMs, + error: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + }); + throw error; + } +} + +/** + * Schedules the subscription expiry job to run daily. + * Fires every 24 hours (calculated to align with midnight UTC), and returns a cancel handle. + */ +export function scheduleSubscriptionExpiryJob( + notifier: SubscriptionExpiryNotifierService, + logger: AppLogger = defaultLogger +): { stop: () => void } { + let timeout: ReturnType | null = null; + let interval: ReturnType | null = null; + let stopped = false; + + function msUntilMidnightUTC(): number { + const now = new Date(); + const tomorrow = new Date(now); + tomorrow.setUTCDate(tomorrow.getUTCDate() + 1); + tomorrow.setUTCHours(0, 0, 0, 0); + return tomorrow.getTime() - now.getTime(); + } + + async function runJob(): Promise { + if (stopped) return; + try { + await runSubscriptionExpiryJob(notifier, logger); + } catch { + // Errors logged inside runSubscriptionExpiryJob + } + } + + const delayMs = msUntilMidnightUTC(); + logger.info("SubscriptionExpiryJob: Scheduled", { + firstRunInMs: delayMs, + firstRunAt: new Date(Date.now() + delayMs).toISOString(), + }); + + timeout = setTimeout(() => { + if (stopped) return; + void runJob(); + interval = setInterval(() => void runJob(), 24 * 60 * 60 * 1000); + if (interval.unref) interval.unref(); + }, delayMs); + + if (timeout.unref) timeout.unref(); + + return { + stop: () => { + stopped = true; + if (timeout) clearTimeout(timeout); + if (interval) clearInterval(interval); + logger.info("SubscriptionExpiryJob: Cancelled schedule"); + }, + }; +} + +/** + * Helper to construct the full notifier service wired with DataSource. + */ +export function buildSubscriptionExpiryNotifier( + dataSource: DataSource, + notificationService: SubscriptionNotificationSink, + logger: AppLogger = defaultLogger +): SubscriptionExpiryNotifierService { + const keySubRepo = dataSource.getRepository(KeySubscription); + const userRepo = dataSource.getRepository(User); + const creatorKeyRepo = createCreatorKeyRepository(dataSource); + const subscriptionSource = createTypeOrmSubscriptionSource(keySubRepo); + + return new SubscriptionExpiryNotifierService({ + notificationService, + subscriptionSource, + creatorKeyRepository: creatorKeyRepo, + userRepository: userRepo, + logger, + }); +} diff --git a/tests/integration/settlement.integration.test.ts b/tests/integration/settlement.integration.test.ts index bb1b8b01..9c0b7692 100644 --- a/tests/integration/settlement.integration.test.ts +++ b/tests/integration/settlement.integration.test.ts @@ -82,10 +82,6 @@ function createFakeDataSource(invoice: Invoice) { }; const dataSource = { - // InvestmentService requires a current terms acknowledgement before it - // creates an investment. This fixture focuses on settlement behavior, so - // its repository stand-in represents an already acknowledged investor. - getRepository: () => ({ findOne: async () => ({ acknowledgedAt: new Date() }) }), transaction: async (callback: (manager: FakeManager) => Promise) => callback(manager), // Issue #473 — the accreditation gate looks the investor's terms // acknowledgement up before creating the investment. diff --git a/tests/invoice.service.test.ts b/tests/invoice.service.test.ts index f2677e40..4df43e7a 100644 --- a/tests/invoice.service.test.ts +++ b/tests/invoice.service.test.ts @@ -2,7 +2,7 @@ import { IsNull } from "typeorm"; import { InvoiceService } from "../src/services/invoice.service"; import { ServiceError } from "../src/utils/service-error"; import { Invoice } from "../src/models/Invoice.model"; -import { InvoiceStatus } from "../src/types/enums"; +import { InvoiceStatus, KYCStatus } from "../src/types/enums"; import { logger } from "../src/observability/logger"; describe("InvoiceService", () => { @@ -178,9 +178,6 @@ describe("InvoiceService", () => { mockInvoiceRepository.findOneBy.mockResolvedValue(mockInvoice); const creation = invoiceService.createInvoice(buildCreateInput()); - await expect(invoiceService.createInvoice(buildCreateInput())).rejects.toThrow(ServiceError); - - await expect(invoiceService.createInvoice(buildCreateInput())).rejects.toMatchObject({ await expect(creation).rejects.toBeInstanceOf(ServiceError); await expect(creation).rejects.toMatchObject({ code: "invoice_number_exists", @@ -510,9 +507,9 @@ describe("InvoiceService", () => { await invoiceService.getInvoicesBySellerId({ sellerId: "seller-456", skip, take }); expect(mockInvoiceRepository.find).toHaveBeenCalledWith( - expect.objectContaining({ skip: expectedSkip, take: expectedTake }), + expect.objectContaining({ skip: expectedSkip, take: expectedTake }) ); - }, + } ); }); @@ -1092,6 +1089,66 @@ describe("InvoiceService", () => { }); }); }); + + // ============ REPOSITORY ERROR HANDLING ============ + describe("repository error handling", () => { + it("should return a service error and log when invoice listing fails", async () => { + const errorSpy = jest.spyOn(logger, "error"); + mockInvoiceRepository.find.mockRejectedValue(new Error("Query timeout")); + mockInvoiceRepository.count.mockResolvedValue(0); + + await expect( + invoiceService.getInvoicesBySellerId({ sellerId: "seller-456" }) + ).rejects.toMatchObject({ code: "invoice_list_failed", statusCode: 500 }); + + expect(errorSpy).toHaveBeenCalledWith( + "Failed to fetch invoices by seller", + expect.objectContaining({ sellerId: "seller-456" }) + ); + errorSpy.mockRestore(); + }); + + it("should propagate database errors on createInvoice", async () => { + mockInvoiceRepository.findOneBy.mockResolvedValue(null); + mockInvoiceRepository.create.mockReturnValue(mockInvoice); + mockInvoiceRepository.save.mockRejectedValue(new Error("Connection refused")); + + await expect(invoiceService.createInvoice(buildCreateInput())).rejects.toThrow( + /Connection refused|Failed to create invoice/ + ); + }); + + it("should propagate database errors on getInvoiceById", async () => { + mockInvoiceRepository.findOne.mockRejectedValue(new Error("Query timeout")); + + await expect(invoiceService.getInvoiceById("invoice-123")).rejects.toThrow( + /Query timeout|Failed to fetch invoice/ + ); + }); + + it("should propagate database errors on updateInvoice", async () => { + mockInvoiceRepository.findOne.mockResolvedValue(mockInvoice); + mockInvoiceRepository.save.mockRejectedValue(new Error("Deadlock detected")); + + await expect( + invoiceService.updateInvoice({ + sellerId: "seller-456", + invoiceId: "invoice-123", + customerName: "Updated", + }) + ).rejects.toThrow(/Deadlock detected|Processing failed/); + }); + + it("should propagate database errors on deleteInvoice", async () => { + mockInvoiceRepository.findOne.mockResolvedValue(mockInvoice); + mockInvoiceRepository.save.mockRejectedValue(new Error("Storage full")); + + await expect(invoiceService.deleteInvoice("invoice-123", "seller-456")).rejects.toThrow( + /Storage full|Processing failed/ + ); + }); + }); + // ============ STATE-MACHINE BACKED TRANSITIONS ============ describe("status transitions", () => { let stateMachine: { transition: jest.Mock; dispatch: jest.Mock }; @@ -1140,28 +1197,6 @@ describe("InvoiceService", () => { ); expect(stateMachine.dispatch).toHaveBeenCalledTimes(1); }); - // ============ REPOSITORY ERROR HANDLING ============ - describe("repository error handling", () => { - it("should return a service error and log when invoice listing fails", async () => { - const errorSpy = jest.spyOn(logger, "error"); - mockInvoiceRepository.find.mockRejectedValue(new Error("Query timeout")); - mockInvoiceRepository.count.mockResolvedValue(0); - - await expect( - invoiceService.getInvoicesBySellerId({ sellerId: "seller-456" }), - ).rejects.toMatchObject({ code: "invoice_list_failed", statusCode: 500 }); - - expect(errorSpy).toHaveBeenCalledWith( - "Failed to fetch invoices by seller", - expect.objectContaining({ sellerId: "seller-456" }), - ); - errorSpy.mockRestore(); - }); - - it("should propagate database errors on createInvoice", async () => { - mockInvoiceRepository.findOneBy.mockResolvedValue(null); - mockInvoiceRepository.create.mockReturnValue(mockInvoice); - mockInvoiceRepository.save.mockRejectedValue(new Error("Connection refused")); it("should record a null actor when the admin is unknown", async () => { mockInvoiceRepository.findOne.mockResolvedValue(mockInvoice); diff --git a/tests/unit/services/subscription-expiry-notifier.test.ts b/tests/unit/services/subscription-expiry-notifier.test.ts new file mode 100644 index 00000000..03b7134e --- /dev/null +++ b/tests/unit/services/subscription-expiry-notifier.test.ts @@ -0,0 +1,654 @@ +import Decimal from "decimal.js"; +import { + SubscriptionExpiryNotifierService, + createSubscriptionExpiryNotifierService, + InMemorySubscriptionDedupeStore, + type SubscriptionCandidate, + type SubscriptionNotificationSink, +} from "../../../src/services/subscription-expiry-notifier.service"; +import { + runSubscriptionExpiryJob, + scheduleSubscriptionExpiryJob, +} from "../../../src/workers/subscription-expiry.worker"; +import { NotificationType } from "../../../src/types/enums"; +import { LEDGERS_PER_DAY } from "../../../src/services/subscription-status.service"; + +interface SentNotification { + userId: string; + type: NotificationType; + title: string; + message: string; + data?: Record; + dedupeKey?: string | null; +} + +function createMockNotificationSink(): { + sink: SubscriptionNotificationSink; + sent: SentNotification[]; + dedupeMap: Map; +} { + const sent: SentNotification[] = []; + const dedupeMap = new Map(); + + const sink: SubscriptionNotificationSink = { + async createNotification(userId, type, title, message, data, dedupeKey) { + if (dedupeKey && dedupeMap.has(dedupeKey)) { + throw new Error(`duplicate key value violates unique constraint "uq_notifications_dedupe_key"`); + } + const record: SentNotification = { + userId, + type, + title, + message, + data, + dedupeKey: dedupeKey || null, + }; + sent.push(record); + if (dedupeKey) { + dedupeMap.set(dedupeKey, record); + } + return { id: `notif-${sent.length}`, ...record }; + }, + async findByDedupeKey(dedupeKey) { + return dedupeMap.get(dedupeKey) ?? null; + }, + }; + + return { sink, sent, dedupeMap }; +} + +describe("Subscription Expiry Notification Service", () => { + const FIXED_NOW = new Date("2026-10-17T12:00:00.000Z"); + const WALLET_A = "GDATAA000000000000000000000000000000000000000000000000000001"; + const WALLET_B = "GDATAB000000000000000000000000000000000000000000000000000002"; + const KEY_ID = "CKEY0000000000000000000000000000000000000000000000000000000001"; + + // Acceptance Criteria 1: 7-day warning notification dispatched correctly + it("dispatches 7-day warning notification correctly when subscription expires within 7 days", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + // 5 days remaining (<= 7 days, > 24 hours) + const expiryDate = new Date("2026-10-22T12:00:00.000Z"); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Creator Pro Pass", + expiryDate, + minHolding: "100", + currentBalance: "25", // does not meet minimum holding (25 < 100) + }; + + const result = await service.checkSubscription(candidate); + + expect(result.status).toBe("notified"); + expect(result.window).toBe("7d"); + expect(sent).toHaveLength(1); + + const notification = sent[0]; + expect(notification.userId).toBe(WALLET_A); + expect(notification.type).toBe(NotificationType.SUBSCRIPTION); + expect(notification.title).toBe("Subscription Expiring Soon: Creator Pro Pass"); + expect(notification.message).toContain("Creator Pro Pass"); + expect(notification.message).toContain(expiryDate.toISOString()); + expect(notification.message).toContain("100"); + expect(notification.dedupeKey).toBe(`sub_expiry:${WALLET_A}:${KEY_ID}:2026-10-22:7d`); + }); + + // Acceptance Criteria 2: 24-hour warning notification dispatched correctly + it("dispatches 24-hour warning notification correctly when subscription expires within 24 hours", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + // 12 hours remaining (<= 24 hours) + const expiryDate = new Date("2026-10-18T00:00:00.000Z"); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "VIP Pass", + expiryDate, + minHolding: "50", + currentBalance: "10", // does not meet minimum holding (10 < 50) + }; + + const result = await service.checkSubscription(candidate); + + expect(result.status).toBe("notified"); + expect(result.window).toBe("24h"); + expect(sent).toHaveLength(1); + + const notification = sent[0]; + expect(notification.title).toBe("Urgent: Subscription Expiring in 24 Hours - VIP Pass"); + expect(notification.message).toContain("Urgent"); + expect(notification.message).toContain("VIP Pass"); + expect(notification.message).toContain(expiryDate.toISOString()); + expect(notification.message).toContain("50"); + expect(notification.dedupeKey).toBe(`sub_expiry:${WALLET_A}:${KEY_ID}:2026-10-18:24h`); + }); + + // Acceptance Criteria 3: Notification skipped if wallet already meets minimum holding + it("skips notification if wallet already meets minimum holding", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + const expiryDate = new Date("2026-10-20T12:00:00.000Z"); // 3 days remaining + + // Case A: currentBalance equals minHolding + const candidateEqual: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Gold Key", + expiryDate, + minHolding: "100", + currentBalance: "100", + }; + + const resultEqual = await service.checkSubscription(candidateEqual); + expect(resultEqual.status).toBe("skipped_meets_minimum_holding"); + expect(resultEqual.reason).toBe("Wallet already meets minimum holding"); + expect(sent).toHaveLength(0); + + // Case B: currentBalance exceeds minHolding + const candidateExceed: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Gold Key", + expiryDate, + minHolding: "100", + currentBalance: "150", + }; + + const resultExceed = await service.checkSubscription(candidateExceed); + expect(resultExceed.status).toBe("skipped_meets_minimum_holding"); + expect(sent).toHaveLength(0); + + // Case C: explicitly marked as toppedUp / meetsMinimumHolding + const candidateToppedUp: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Gold Key", + expiryDate, + minHolding: "100", + currentBalance: "20", + hasToppedUp: true, + }; + + const resultToppedUp = await service.checkSubscription(candidateToppedUp); + expect(resultToppedUp.status).toBe("skipped_meets_minimum_holding"); + expect(sent).toHaveLength(0); + }); + + // Acceptance Criteria 4: Notification content includes key name, expiry, and minimum holding + it("includes key name, expiry date, and minimum holding required in both message text and data payload", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + const expiryDate = new Date("2026-10-21T15:30:00.000Z"); + const keyName = "Exclusive Music Pass"; + const minHolding = "250"; + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName, + expiryDate, + minHolding, + currentBalance: "50", + }; + + await service.checkSubscription(candidate); + + expect(sent).toHaveLength(1); + const notif = sent[0]; + + // Notification title & message contain key name + expect(notif.title).toContain(keyName); + expect(notif.message).toContain(keyName); + + // Notification message contains formatted expiry date + expect(notif.message).toContain(expiryDate.toISOString()); + + // Notification message contains minimum holding + expect(notif.message).toContain(minHolding); + + // Notification data contains structured properties + expect(notif.data).toMatchObject({ + keyId: KEY_ID, + keyName, + expiryDate: expiryDate.toISOString(), + expiry: expiryDate.toISOString(), + minHolding, + minimumHolding: minHolding, + currentBalance: "50", + window: "7d", + }); + }); + + // Acceptance Criteria 5: Duplicate notifications prevented for same expiry cycle + it("prevents duplicate notifications for the same expiry cycle and window", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + const expiryDate = new Date("2026-10-21T12:00:00.000Z"); // 4 days remaining (7d window) + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Alpha Pass", + expiryDate, + minHolding: "100", + currentBalance: "10", + }; + + // First run dispatches notification + const firstRun = await service.checkSubscription(candidate); + expect(firstRun.status).toBe("notified"); + expect(sent).toHaveLength(1); + + // Second run with the same cycle & window is skipped as a duplicate + const secondRun = await service.checkSubscription(candidate); + expect(secondRun.status).toBe("skipped_duplicate"); + expect(secondRun.reason).toContain("Duplicate notification prevented"); + expect(sent).toHaveLength(1); // still only 1 notification + }); + + it("sends second notification 24 hours before expiry after 7-day notification was already sent", async () => { + const { sink, sent } = createMockNotificationSink(); + + // Day 1: 5 days before expiry -> 7-day notification is sent + const day1Now = new Date("2026-10-17T12:00:00.000Z"); + const expiryDate = new Date("2026-10-22T12:00:00.000Z"); + + const serviceDay1 = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => day1Now, + }); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Alpha Pass", + expiryDate, + minHolding: "100", + currentBalance: "20", + cycleId: "cycle-2026-10-22", + }; + + const res1 = await serviceDay1.checkSubscription(candidate); + expect(res1.status).toBe("notified"); + expect(res1.window).toBe("7d"); + expect(sent).toHaveLength(1); + expect(sent[0].title).toBe("Subscription Expiring Soon: Alpha Pass"); + + // Day 5: 12 hours before expiry -> 24-hour notification is sent! + const day5Now = new Date("2026-10-22T00:00:00.000Z"); + const serviceDay5 = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => day5Now, + }); + + const res2 = await serviceDay5.checkSubscription(candidate); + expect(res2.status).toBe("notified"); + expect(res2.window).toBe("24h"); + expect(sent).toHaveLength(2); + expect(sent[1].title).toBe("Urgent: Subscription Expiring in 24 Hours - Alpha Pass"); + expect(sent[1].dedupeKey).toBe(`sub_expiry:${WALLET_A}:${KEY_ID}:cycle-2026-10-22:24h`); + + // Repeated check 2 hours later still within 24 hours is prevented + const serviceDay5Later = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => new Date("2026-10-22T02:00:00.000Z"), + }); + + const res3 = await serviceDay5Later.checkSubscription(candidate); + expect(res3.status).toBe("skipped_duplicate"); + expect(sent).toHaveLength(2); // still only 2 + }); + + it("skips 24-hour notification if wallet tops up holding after receiving 7-day warning", async () => { + const { sink, sent } = createMockNotificationSink(); + + // Day 1: 5 days remaining, holding = 20 < 100 -> 7d warning sent + const day1Now = new Date("2026-10-17T12:00:00.000Z"); + const expiryDate = new Date("2026-10-22T12:00:00.000Z"); + + const serviceDay1 = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => day1Now, + }); + + const candidateUnderfunded: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Alpha Pass", + expiryDate, + minHolding: "100", + currentBalance: "20", + cycleId: "cycle-oct-22", + }; + + await serviceDay1.checkSubscription(candidateUnderfunded); + expect(sent).toHaveLength(1); + + // Day 5: 12 hours remaining, but wallet has topped up holding to 120 >= 100 + const day5Now = new Date("2026-10-22T00:00:00.000Z"); + const serviceDay5 = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => day5Now, + }); + + const candidateToppedUp: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Alpha Pass", + expiryDate, + minHolding: "100", + currentBalance: "120", // Topped up! + cycleId: "cycle-oct-22", + }; + + const res2 = await serviceDay5.checkSubscription(candidateToppedUp); + expect(res2.status).toBe("skipped_meets_minimum_holding"); + expect(sent).toHaveLength(1); // Second notification skipped! + }); + + it("skips subscriptions that expire in more than 7 days", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + // 10 days remaining (> 7 days) + const expiryDate = new Date("2026-10-27T12:00:00.000Z"); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Future Key", + expiryDate, + minHolding: "50", + currentBalance: "0", + }; + + const res = await service.checkSubscription(candidate); + expect(res.status).toBe("skipped_outside_window"); + expect(sent).toHaveLength(0); + }); + + it("skips subscriptions that have already expired", async () => { + const { sink, sent } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + // Expired 1 hour ago + const expiryDate = new Date("2026-10-17T11:00:00.000Z"); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Lapsed Key", + expiryDate, + minHolding: "50", + currentBalance: "0", + }; + + const res = await service.checkSubscription(candidate); + expect(res.status).toBe("skipped_already_expired"); + expect(sent).toHaveLength(0); + }); + + it("calculates expiry window from on-chain expiryLedger correctly", async () => { + const { sink, sent } = createMockNotificationSink(); + const currentLedger = 1_000_000; + // 5 days remaining in ledgers: 5 * 17,280 = 86,400 ledgers + const expiryLedger = currentLedger + LEDGERS_PER_DAY * 5; + + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + ledgersPerDay: LEDGERS_PER_DAY, + }); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Ledger Key", + expiryLedger, + minHolding: "10", + currentBalance: "2", + currentLedger, + } as SubscriptionCandidate & { currentLedger: number }; + + const res = await service.checkSubscription(candidate); + expect(res.status).toBe("notified"); + expect(res.window).toBe("7d"); + expect(sent).toHaveLength(1); + expect(sent[0].message).toContain("10 tokens"); + }); + + it("resolves key name from CreatorKey repository if not provided on candidate", async () => { + const { sink, sent } = createMockNotificationSink(); + const mockCreatorKeyRepo = { + findById: jest.fn().mockResolvedValue({ + id: KEY_ID, + name: "Creator Key From Repo", + contractAddress: "CCONTRACT001", + }), + findByContractAddress: jest.fn().mockResolvedValue(null), + applyConfigUpdate: jest.fn().mockResolvedValue(null), + }; + + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + creatorKeyRepository: mockCreatorKeyRepo, + now: () => FIXED_NOW, + }); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + expiryDate: new Date("2026-10-21T12:00:00.000Z"), + minHolding: "100", + currentBalance: "10", + }; + + await service.checkSubscription(candidate); + + expect(sent).toHaveLength(1); + expect(sent[0].title).toBe("Subscription Expiring Soon: Creator Key From Repo"); + expect(sent[0].message).toContain("Creator Key From Repo"); + }); + + it("resolves recipient userId from UserRepository matching wallet address", async () => { + const { sink, sent } = createMockNotificationSink(); + const mockUserRepo = { + findOne: jest.fn().mockResolvedValue({ + id: "user-uuid-123", + stellarAddress: WALLET_A, + }), + }; + + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + userRepository: mockUserRepo as any, + now: () => FIXED_NOW, + }); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Test Key", + expiryDate: new Date("2026-10-21T12:00:00.000Z"), + minHolding: "10", + currentBalance: "1", + }; + + await service.checkSubscription(candidate); + + expect(sent).toHaveLength(1); + expect(sent[0].userId).toBe("user-uuid-123"); + }); + + it("checks live on-chain holding through SubscriptionHoldingReader if balance not in candidate", async () => { + const { sink, sent } = createMockNotificationSink(); + const mockHoldingReader = { + readHolding: jest.fn().mockResolvedValue({ + balance: "15", + minBalance: "10", // 15 >= 10 -> meets holding! + expiryLedger: 1_050_000, + ledger: 1_000_000, + }), + }; + + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + holdingReader: mockHoldingReader, + now: () => FIXED_NOW, + }); + + const candidate: SubscriptionCandidate = { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Onchain Key", + expiryDate: new Date("2026-10-21T12:00:00.000Z"), + }; + + const res = await service.checkSubscription(candidate); + expect(mockHoldingReader.readHolding).toHaveBeenCalledWith({ + wallet: WALLET_A, + keyId: KEY_ID, + }); + expect(res.status).toBe("skipped_meets_minimum_holding"); + expect(sent).toHaveLength(0); + }); + + it("processes a batch of expiring subscriptions and aggregates metrics", async () => { + const { sink } = createMockNotificationSink(); + const service = createSubscriptionExpiryNotifierService({ + notificationService: sink, + now: () => FIXED_NOW, + }); + + const candidates: SubscriptionCandidate[] = [ + // 1: 5 days left, underfunded -> notify 7d + { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Key 1", + expiryDate: new Date("2026-10-22T12:00:00.000Z"), + minHolding: "100", + currentBalance: "10", + }, + // 2: 12 hours left, underfunded -> notify 24h + { + wallet: WALLET_B, + keyId: KEY_ID, + keyName: "Key 2", + expiryDate: new Date("2026-10-18T00:00:00.000Z"), + minHolding: "100", + currentBalance: "10", + }, + // 3: 4 days left, fully funded -> skip meets holding + { + wallet: WALLET_A, + keyId: "KEY_3", + keyName: "Key 3", + expiryDate: new Date("2026-10-21T12:00:00.000Z"), + minHolding: "100", + currentBalance: "150", + }, + // 4: 15 days left -> skip outside window + { + wallet: WALLET_B, + keyId: "KEY_4", + keyName: "Key 4", + expiryDate: new Date("2026-11-01T12:00:00.000Z"), + minHolding: "100", + currentBalance: "0", + }, + ]; + + const result = await service.processExpiringSubscriptions(candidates); + + expect(result.totalChecked).toBe(4); + expect(result.notified7d).toBe(1); + expect(result.notified24h).toBe(1); + expect(result.skippedMeetsHolding).toBe(1); + expect(result.skippedOutsideWindow).toBe(1); + expect(result.errors).toHaveLength(0); + }); + + describe("Scheduled Daily Job & Worker", () => { + it("runs daily job successfully using runSubscriptionExpiryJob", async () => { + const { sink } = createMockNotificationSink(); + const mockCandidateSource = { + findExpiringCandidates: jest.fn().mockResolvedValue([ + { + wallet: WALLET_A, + keyId: KEY_ID, + keyName: "Scheduled Key", + expiryDate: new Date("2026-10-21T12:00:00.000Z"), + minHolding: "50", + currentBalance: "5", + }, + ]), + recordNotificationSent: jest.fn().mockResolvedValue(undefined), + }; + + const notifier = createSubscriptionExpiryNotifierService({ + notificationService: sink, + subscriptionSource: mockCandidateSource, + now: () => FIXED_NOW, + }); + + const runResult = await runSubscriptionExpiryJob(notifier); + + expect(runResult.totalChecked).toBe(1); + expect(runResult.notified7d).toBe(1); + expect(mockCandidateSource.findExpiringCandidates).toHaveBeenCalledWith({ + withinDays: 7, + currentDate: FIXED_NOW, + }); + expect(mockCandidateSource.recordNotificationSent).toHaveBeenCalledWith( + expect.objectContaining({ + wallet: WALLET_A, + keyId: KEY_ID, + window: "7d", + }) + ); + }); + + it("schedules daily job and supports clean cancellation via stop()", () => { + const { sink } = createMockNotificationSink(); + const notifier = createSubscriptionExpiryNotifierService({ + notificationService: sink, + }); + + const scheduler = scheduleSubscriptionExpiryJob(notifier); + expect(typeof scheduler.stop).toBe("function"); + + // Verify clean stop without errors or unhandled handles + expect(() => scheduler.stop()).not.toThrow(); + }); + }); +}); From 86a4708e9971ff03b166e90fcc0f5bbcc5fac8b4 Mon Sep 17 00:00:00 2001 From: dee-john Date: Sat, 17 Oct 2026 17:42:04 +0100 Subject: [PATCH 2/3] fix: resolve TypeScript compilation and lint errors from upstream merge --- .../investor-risk-profile.controller.ts | 1 - src/controllers/invoice-draft.controller.ts | 3 +-- src/index.ts | 4 ++++ src/middleware/auth.middleware.ts | 11 +++++++---- src/models/InvestorAccreditation.model.ts | 14 +++++++------- src/models/SecondaryMarketPriceHistory.model.ts | 12 ++++++------ src/routes/accreditation.routes.ts | 11 ++++++----- src/routes/investor.routes.ts | 5 +---- src/routes/invoices/refund.routes.ts | 14 +++++++------- .../secondary-market/price-history.routes.ts | 4 ++-- src/services/investor-accreditation.service.ts | 1 - src/services/invoice-refund.service.ts | 10 +++++----- src/services/issuer-reputation.service.ts | 10 ---------- src/services/secondary-market-price.service.ts | 9 +++++---- src/types/enums.ts | 2 ++ tests/invoice.service.test.ts | 4 ++-- 16 files changed, 55 insertions(+), 60 deletions(-) diff --git a/src/controllers/investor-risk-profile.controller.ts b/src/controllers/investor-risk-profile.controller.ts index 2ffe9a9b..c9db744c 100644 --- a/src/controllers/investor-risk-profile.controller.ts +++ b/src/controllers/investor-risk-profile.controller.ts @@ -1,7 +1,6 @@ import { Request, Response, NextFunction } from "express"; import { InvestorRiskProfileService, RiskProfileInput } from "../services/investor-risk-profile.service"; import { AppError } from "../utils/http-error"; -import { logger } from "../observability/logger"; import type { AuthenticatedRequest } from "../types/auth"; export class InvestorRiskProfileController { diff --git a/src/controllers/invoice-draft.controller.ts b/src/controllers/invoice-draft.controller.ts index f6b20442..77ed81f1 100644 --- a/src/controllers/invoice-draft.controller.ts +++ b/src/controllers/invoice-draft.controller.ts @@ -1,7 +1,6 @@ -import { Request, Response, NextFunction } from "express"; +import { Response, NextFunction } from "express"; import { InvoiceDraftService, DraftData } from "../services/invoice-draft.service"; import { AppError } from "../utils/http-error"; -import { logger } from "../observability/logger"; import type { AuthenticatedRequest } from "../types/auth"; export class InvoiceDraftController { diff --git a/src/index.ts b/src/index.ts index 79da1795..79ddc187 100644 --- a/src/index.ts +++ b/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"; diff --git a/src/middleware/auth.middleware.ts b/src/middleware/auth.middleware.ts index 8508ccdc..021fa9d3 100644 --- a/src/middleware/auth.middleware.ts +++ b/src/middleware/auth.middleware.ts @@ -355,7 +355,7 @@ 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) { @@ -363,12 +363,15 @@ export function requireAdmin(adminWallets: string[]) { 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")); }; } diff --git a/src/models/InvestorAccreditation.model.ts b/src/models/InvestorAccreditation.model.ts index 68989bf8..fa0d1e66 100644 --- a/src/models/InvestorAccreditation.model.ts +++ b/src/models/InvestorAccreditation.model.ts @@ -23,27 +23,27 @@ export const ACCREDITATION_LIMITS: Record = { @Index("idx_investor_accreditation_wallet", ["walletAddress"], { unique: true }) export class InvestorAccreditation { @PrimaryGeneratedColumn("uuid") - id: string; + id!: string; @Column({ type: "varchar", length: 256, unique: true }) - walletAddress: string; + walletAddress!: string; @Column({ type: "enum", enum: AccreditationTier, default: AccreditationTier.UNACCREDITED, }) - tier: AccreditationTier; + tier!: AccreditationTier; @Column({ type: "timestamp", nullable: true }) - verificationDate: Date | null; + verificationDate!: Date | null; @Column({ type: "varchar", length: 255, nullable: true }) - verificationSource: string | null; + verificationSource!: string | null; @CreateDateColumn() - createdAt: Date; + createdAt!: Date; @UpdateDateColumn() - updatedAt: Date; + updatedAt!: Date; } diff --git a/src/models/SecondaryMarketPriceHistory.model.ts b/src/models/SecondaryMarketPriceHistory.model.ts index d147e845..fec2fa3a 100644 --- a/src/models/SecondaryMarketPriceHistory.model.ts +++ b/src/models/SecondaryMarketPriceHistory.model.ts @@ -15,23 +15,23 @@ import { Invoice } from "./Invoice.model"; @Index("idx_price_history_invoice_date_range", ["invoiceId", "createdAt"]) export class SecondaryMarketPriceHistory { @PrimaryGeneratedColumn("uuid") - id: string; + id!: string; @Column({ type: "uuid" }) - invoiceId: string; + invoiceId!: string; @ManyToOne(() => Invoice) @JoinColumn({ name: "invoiceId" }) - invoice: Invoice; + invoice!: Invoice; @Column({ type: "decimal", precision: 18, scale: 8 }) - askPrice: Decimal; + askPrice!: Decimal; @Column({ type: "integer", default: 1 }) - listingCount: number; + listingCount!: number; @CreateDateColumn() - createdAt: Date; + createdAt!: Date; } export interface PriceHistoryPoint { diff --git a/src/routes/accreditation.routes.ts b/src/routes/accreditation.routes.ts index 5136436f..7bae9abb 100644 --- a/src/routes/accreditation.routes.ts +++ b/src/routes/accreditation.routes.ts @@ -1,5 +1,5 @@ import { Router, type Response, type NextFunction } from "express"; -import { createAuthMiddleware, requireInvestor, requireAdmin } from "../middleware/auth.middleware"; +import { createAuthMiddleware, requireAdmin } from "../middleware/auth.middleware"; import type { AuthService } from "../services/auth.service"; import type { InvestorAccreditationService } from "../services/investor-accreditation.service"; import type { AuthenticatedRequest } from "../types/auth"; @@ -23,7 +23,7 @@ export function createAccreditationRouter({ auth, async (req: AuthenticatedRequest, res: Response, next: NextFunction) => { try { - const { wallet } = req.params; + const wallet = String(req.params.wallet); const accreditation = await accreditationService.getAccreditation(wallet); res.json({ @@ -44,11 +44,11 @@ export function createAccreditationRouter({ requireAdmin(), async (req: AuthenticatedRequest, 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)) { - throw new HttpError("Invalid tier", 400); + throw new HttpError(400, "Invalid tier"); } const accreditation = await accreditationService.updateAccreditation( @@ -74,7 +74,8 @@ export function createAccreditationRouter({ auth, async (req: AuthenticatedRequest, 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 diff --git a/src/routes/investor.routes.ts b/src/routes/investor.routes.ts index 87962385..60fd063b 100644 --- a/src/routes/investor.routes.ts +++ b/src/routes/investor.routes.ts @@ -5,8 +5,7 @@ import type { InvestorAcknowledgementService } from "../services/investor-acknow import type { AuthenticatedRequest } from "../types/auth"; import { ServiceError } from "../utils/service-error"; import { HttpError, PublicAppError } from "../utils/http-error"; -import { createInvestorRiskProfileController } from "@/controllers/investor-risk-profile.controller"; -import type { InvestorRiskProfileService } from "@/services/investor-risk-profile.service"; +import type { InvestorRiskProfileService } from "../services/investor-risk-profile.service"; export interface InvestorRouterDependencies { authService: AuthService; @@ -72,8 +71,6 @@ export function createInvestorRouter({ // ---- Risk Profile Management (issue #595) ---- if (riskProfileService) { - const riskProfileController = createInvestorRiskProfileController(riskProfileService); - router.post( "/risk-profile", auth, diff --git a/src/routes/invoices/refund.routes.ts b/src/routes/invoices/refund.routes.ts index a3a01af0..cf2f5f0f 100644 --- a/src/routes/invoices/refund.routes.ts +++ b/src/routes/invoices/refund.routes.ts @@ -25,16 +25,16 @@ export function createRefundRouter({ requireInvestor(), async (req: AuthenticatedRequest, res: Response, next: NextFunction) => { try { - const { invoiceId } = req.params; - const wallet = req.user?.wallet; + const invoiceId = String(req.params.invoiceId); + const wallet = req.user?.stellarAddress || (req.user as unknown as { wallet?: string })?.wallet; if (!wallet) { - throw new HttpError("Wallet not found in request", 401); + throw new HttpError(401, "Wallet not found in request"); } const { canRefund, reason } = await refundService.canRefund(invoiceId); if (!canRefund) { - throw new HttpError(reason || "Cannot refund invoice", 400); + throw new HttpError(400, reason || "Cannot refund invoice"); } await refundService.refundInvestment(invoiceId, wallet); @@ -47,7 +47,7 @@ export function createRefundRouter({ }); } catch (error) { if (error instanceof ServiceError) { - return next(new HttpError(error.message, error.statusCode)); + return next(new HttpError(error.statusCode, error.message)); } next(error); } @@ -60,10 +60,10 @@ export function createRefundRouter({ requireInvestor(), async (req: AuthenticatedRequest, res: Response, next: NextFunction) => { try { - const wallet = req.user?.wallet; + const wallet = req.user?.stellarAddress || (req.user as unknown as { wallet?: string })?.wallet; if (!wallet) { - throw new HttpError("Wallet not found in request", 401); + throw new HttpError(401, "Wallet not found in request"); } const refundableInvoices = await refundService.getRefundableInvoices(wallet); diff --git a/src/routes/secondary-market/price-history.routes.ts b/src/routes/secondary-market/price-history.routes.ts index f7e2cda4..e079d54b 100644 --- a/src/routes/secondary-market/price-history.routes.ts +++ b/src/routes/secondary-market/price-history.routes.ts @@ -22,11 +22,11 @@ export function createPriceHistoryRouter({ auth, async (req: AuthenticatedRequest, 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)) { - throw new HttpError("Invalid range parameter", 400); + throw new HttpError(400, "Invalid range parameter"); } const { history, currentAvg } = await priceService.getPriceHistory( diff --git a/src/services/investor-accreditation.service.ts b/src/services/investor-accreditation.service.ts index 6d452df8..a10435e8 100644 --- a/src/services/investor-accreditation.service.ts +++ b/src/services/investor-accreditation.service.ts @@ -5,7 +5,6 @@ import { ACCREDITATION_LIMITS, } from "../models/InvestorAccreditation.model"; import { Invoice } from "../models/Invoice.model"; -import { ServiceError } from "../utils/service-error"; import Decimal from "decimal.js"; export class InvestorAccreditationService { diff --git a/src/services/invoice-refund.service.ts b/src/services/invoice-refund.service.ts index 3899cec4..bd9c2065 100644 --- a/src/services/invoice-refund.service.ts +++ b/src/services/invoice-refund.service.ts @@ -44,7 +44,7 @@ export class InvoiceRefundService { const invoice = await this.invoiceRepo.findOne({ where: { id: invoiceId } }); if (!invoice) { - throw new ServiceError("Invoice not found", 404); + throw new ServiceError("invoice_not_found", "Invoice not found", 404); } const investment = await this.investmentRepo.findOne({ @@ -55,18 +55,18 @@ export class InvoiceRefundService { }); if (!investment) { - throw new ServiceError("Investment record not found", 404); + throw new ServiceError("investment_not_found", "Investment record not found", 404); } const { canRefund, reason } = await this.canRefund(invoiceId); if (!canRefund) { - throw new ServiceError(reason || "Cannot refund invoice", 400); + throw new ServiceError("refund_not_allowed", reason || "Cannot refund invoice", 400); } logger.info("Processing refund", { invoiceId, investorWallet: walletAddress, - amount: investment.amount, + amount: investment.investmentAmount, }); await this.investmentRepo.remove(investment); @@ -81,7 +81,7 @@ export class InvoiceRefundService { } } - async getRefundableInvoices(walletAddress: string): Promise { + async getRefundableInvoices(_walletAddress: string): Promise { const now = new Date(); return this.invoiceRepo.find({ diff --git a/src/services/issuer-reputation.service.ts b/src/services/issuer-reputation.service.ts index a1d5595e..910a04e8 100644 --- a/src/services/issuer-reputation.service.ts +++ b/src/services/issuer-reputation.service.ts @@ -1,7 +1,6 @@ import { DataSource, EntityManager } from "typeorm"; import { IssuerReputation, IssuerReputationHistory, ReputationTier } from "../models/IssuerReputation.model"; import { Invoice } from "../models/Invoice.model"; -import { ServiceError } from "../utils/service-error"; import { logger } from "../observability/logger"; import Decimal from "decimal.js"; @@ -39,13 +38,6 @@ const TIER_THRESHOLDS = { POOR: 0, }; -const TIER_LABELS: Record = { - [TIER_THRESHOLDS.EXCELLENT]: "excellent", - [TIER_THRESHOLDS.GOOD]: "good", - [TIER_THRESHOLDS.FAIR]: "fair", - [TIER_THRESHOLDS.POOR]: "poor", -}; - export class IssuerReputationService { constructor(private readonly dataSource: DataSource) {} @@ -145,8 +137,6 @@ export class IssuerReputationService { let totalDaysEarly = 0; let lastSettledAt: Date | null = null; - const now = new Date(); - for (const invoice of invoices) { const dueDate = new Date(invoice.dueDate); const settledDate = invoice.updatedAt; // Using updatedAt as approximation of settlement date diff --git a/src/services/secondary-market-price.service.ts b/src/services/secondary-market-price.service.ts index e65718b1..26f139cc 100644 --- a/src/services/secondary-market-price.service.ts +++ b/src/services/secondary-market-price.service.ts @@ -1,6 +1,7 @@ import { Repository, Between, LessThan } from "typeorm"; import { SecondaryMarketPriceHistory, PriceHistoryPoint } from "../models/SecondaryMarketPriceHistory.model"; -import { SecondaryMarketListing } from "../models/SecondaryMarketListing.model"; +import { SecondaryListing } from "../models/SecondaryListing.model"; +import { ListingStatus } from "../types/enums"; import Decimal from "decimal.js"; import { logger } from "../observability/logger"; @@ -15,7 +16,7 @@ const RANGE_DAYS: Record = { export class SecondaryMarketPriceService { constructor( private priceHistoryRepo: Repository, - private listingRepo: Repository + private listingRepo: Repository ) {} async recordPrice( @@ -48,13 +49,13 @@ export class SecondaryMarketPriceService { const history = this.aggregatePrices(prices); const currentListings = await this.listingRepo.find({ - where: { invoiceId, soldAt: null as any }, + where: { invoiceId, status: ListingStatus.ACTIVE }, }); let currentAvg = new Decimal(0); if (currentListings.length > 0) { const sum = currentListings.reduce( - (acc, l) => acc.plus(l.askPrice || 0), + (acc, l) => acc.plus(l.pricePerFraction || 0), new Decimal(0) ); currentAvg = sum.dividedBy(currentListings.length); diff --git a/src/types/enums.ts b/src/types/enums.ts index 97282f36..d74423d2 100644 --- a/src/types/enums.ts +++ b/src/types/enums.ts @@ -81,6 +81,8 @@ export enum NotificationType { REFUND_PROCESSED = "refund_processed", SECONDARY_MARKET_SOLD = "secondary_market_sold", KYC_STATUS_CHANGE = "kyc_status_change", + SUBSCRIPTION = "subscription", + SUBSCRIPTION_EXPIRING = "subscription_expiring", } /** Seller funding-deadline extension request lifecycle (issue #477). */ diff --git a/tests/invoice.service.test.ts b/tests/invoice.service.test.ts index 7c75070f..98a2bbd7 100644 --- a/tests/invoice.service.test.ts +++ b/tests/invoice.service.test.ts @@ -1135,7 +1135,7 @@ describe("InvoiceService", () => { invoiceId: "invoice-123", customerName: "Updated", }) - ).rejects.toThrow(/Deadlock detected|Processing failed/); + ).rejects.toThrow(/Deadlock detected|Processing failed|Failed to update invoice/); }); it("should propagate database errors on deleteInvoice", async () => { @@ -1143,7 +1143,7 @@ describe("InvoiceService", () => { mockInvoiceRepository.save.mockRejectedValue(new Error("Storage full")); await expect(invoiceService.deleteInvoice("invoice-123", "seller-456")).rejects.toThrow( - /Storage full|Processing failed/ + /Storage full|Processing failed|Failed to delete invoice/ ); }); }); From 33e545f2a28233ace11a84b3c37ff725c300c8b5 Mon Sep 17 00:00:00 2001 From: dee-john Date: Sat, 17 Oct 2026 18:44:49 +0100 Subject: [PATCH 3/3] fix(tests): resolve SQLite entity metadata and token generation for integration tests --- tests/fee-schedule.test.ts | 10 ++++++---- tests/helpers/test-data-source.ts | 25 ++++++++++++++++++------- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/tests/fee-schedule.test.ts b/tests/fee-schedule.test.ts index 2c92866c..dcd69534 100644 --- a/tests/fee-schedule.test.ts +++ b/tests/fee-schedule.test.ts @@ -18,8 +18,8 @@ describe("FeeScheduleService", () => { }); beforeEach(async () => { - await dataSource.getRepository(FeeSchedule).clear(); await dataSource.getRepository(FeeScheduleHistory).clear(); + await dataSource.getRepository(FeeSchedule).clear(); }); describe("getCurrentFeeSchedule", () => { @@ -49,7 +49,9 @@ describe("FeeScheduleService", () => { }); it("should throw error when no active schedule exists", async () => { - await expect(feeScheduleService.getCurrentFeeSchedule()).rejects.toThrow("NO_ACTIVE_FEE_SCHEDULE"); + await expect(feeScheduleService.getCurrentFeeSchedule()).rejects.toMatchObject({ + code: "NO_ACTIVE_FEE_SCHEDULE", + }); }); }); @@ -108,7 +110,7 @@ describe("FeeScheduleService", () => { "GDADMINWALLET", "admin" ) - ).rejects.toThrow("INVALID_FEE_RATE"); + ).rejects.toMatchObject({ code: "INVALID_FEE_RATE" }); }); it("should require at least one fee to update", async () => { @@ -125,7 +127,7 @@ describe("FeeScheduleService", () => { await expect( feeScheduleService.updateFeeSchedule({}, "GDADMINWALLET", "admin") - ).rejects.toThrow("NO_CHANGES"); + ).rejects.toMatchObject({ code: "NO_CHANGES" }); }); }); diff --git a/tests/helpers/test-data-source.ts b/tests/helpers/test-data-source.ts index 7886ba03..04c3f5d4 100644 --- a/tests/helpers/test-data-source.ts +++ b/tests/helpers/test-data-source.ts @@ -1,13 +1,27 @@ -import { DataSource } from "typeorm"; +import { DataSource, getMetadataArgsStorage } from "typeorm"; import { FeeSchedule, FeeScheduleHistory } from "../../src/models/FeeSchedule.model"; import { IssuerReputation, IssuerReputationHistory } from "../../src/models/IssuerReputation.model"; import { InvestorRiskProfile } from "../../src/models/InvestorRiskProfile.model"; import { InvoiceDraft } from "../../src/models/InvoiceDraft.model"; -import { User } from "../../src/models/User.model"; -import { Invoice } from "../../src/models/Invoice.model"; -import { Investment } from "../../src/models/Investment.model"; + +function patchEntityMetadataForSQLite(): void { + const columns = getMetadataArgsStorage().columns; + for (const col of columns) { + if (col.options.type === "timestamptz") { + col.options.type = "datetime" as any; + } + if (col.options.type === "jsonb") { + col.options.type = "text" as any; + } + if (col.options.type === "enum") { + col.options.type = "varchar" as any; + } + } +} export async function createTestDataSource(): Promise { + patchEntityMetadataForSQLite(); + const dataSource = new DataSource({ type: "sqlite", database: ":memory:", @@ -18,9 +32,6 @@ export async function createTestDataSource(): Promise { IssuerReputationHistory, InvestorRiskProfile, InvoiceDraft, - User, - Invoice, - Investment, ], synchronize: false, logging: false,