Merge pull request #350 from Vvictor-commits/backend/priority-queue

feat: implement priority-based queue

Author: Abd-Standard

listener/src/services/event-processing-queue.ts

@@ -2,11 +2,18 @@ import * as StellarSDK from '@stellar/stellar-sdk';
 import { ContractConfig } from '../types';
 import logger from '../utils/logger';
 
+export enum Priority {
+  Low = 0,
+  Medium = 1,
+  High = 2,
+}
+
 export interface EventProcessingQueueOptions {
   maxConcurrency?: number;
   pollIntervalMs?: number;
   maxRetries?: number;
   baseDelayMs?: number;
+  priorityWeights?: { high: number; medium: number; low: number };
 }
 
 export type EventProcessor = (
@@ -22,13 +29,16 @@ interface QueuedEvent {
   retryCount: number;
   nextRetryAt: number;
   fingerprint: string;
+  priority: Priority;
+  enqueuedAt: number;
 }
 
 const DEFAULTS = {
   maxConcurrency: 1,
   pollIntervalMs: 1_000,
   maxRetries: 3,
   baseDelayMs: 2_000,
+  priorityWeights: { high: 5, medium: 2, low: 1 },
 };
 
 export class EventProcessingQueue {
@@ -39,21 +49,25 @@ export class EventProcessingQueue {
   private readonly pollIntervalMs: number;
   private readonly maxRetries: number;
   private readonly baseDelayMs: number;
+  private readonly priorityWeights: { high: number; medium: number; low: number };
   private readonly processor: EventProcessor;
   private timer: ReturnType<typeof setInterval> | null = null;
+  private priorityCounters: { high: number; medium: number; low: number } = { high: 0, medium: 0, low: 0 };
 
   constructor(processor: EventProcessor, options?: EventProcessingQueueOptions) {
     this.processor = processor;
     this.maxConcurrency = Math.max(1, options?.maxConcurrency ?? DEFAULTS.maxConcurrency);
     this.pollIntervalMs = options?.pollIntervalMs ?? DEFAULTS.pollIntervalMs;
     this.maxRetries = options?.maxRetries ?? DEFAULTS.maxRetries;
     this.baseDelayMs = options?.baseDelayMs ?? DEFAULTS.baseDelayMs;
+    this.priorityWeights = options?.priorityWeights ?? DEFAULTS.priorityWeights;
   }
 
   enqueue(
     event: StellarSDK.rpc.Api.EventResponse,
     contractConfig: ContractConfig,
-    requestId?: string
+    requestId?: string,
+    priority: Priority = Priority.Medium
   ): boolean {
     const fingerprint = buildEventFingerprint(event, contractConfig.address);
 
@@ -77,6 +91,7 @@ export class EventProcessingQueue {
       delayMs,
       nextRetryAt: new Date(nextRetryAt).toISOString(),
       maxRetries: this.maxRetries,
+      priority: Priority[priority],
     });
 
     this.queuedFingerprints.add(fingerprint);
@@ -87,6 +102,8 @@ export class EventProcessingQueue {
       retryCount: 0,
       nextRetryAt,
       fingerprint,
+      priority,
+      enqueuedAt: Date.now(),
     });
 
     return true;
@@ -124,10 +141,22 @@ export class EventProcessingQueue {
 
     const due = this.queue
       .filter((item) => item.nextRetryAt <= now && !this.activeFingerprints.has(item.fingerprint))
+      .sort((a, b) => {
+        const priorityA = this.getWeightedPriority(a);
+        const priorityB = this.getWeightedPriority(b);
+        if (priorityB !== priorityA) return priorityB - priorityA;
+        return a.enqueuedAt - b.enqueuedAt;
+      })
       .slice(0, available);
 
     if (due.length === 0) return;
 
+    for (const item of due) {
+      if (item.priority === Priority.High) this.priorityCounters.high++;
+      else if (item.priority === Priority.Medium) this.priorityCounters.medium++;
+      else this.priorityCounters.low++;
+    }
+
     const selectedFingerprints = new Set(due.map((item) => item.fingerprint));
 
     this.queue = this.queue.filter(
@@ -148,6 +177,19 @@ export class EventProcessingQueue {
     }
   }
 
+  private getWeightedPriority(item: QueuedEvent): number {
+    const basePriority = item.priority;
+    const age = Date.now() - item.enqueuedAt;
+    const ageBonus = Math.floor(age / 60000);
+
+    let weight = 0;
+    if (item.priority === Priority.High) weight = this.priorityWeights.high;
+    else if (item.priority === Priority.Medium) weight = this.priorityWeights.medium;
+    else weight = this.priorityWeights.low;
+
+    return basePriority + ageBonus + weight;
+  }
+
   private async processItem(item: QueuedEvent): Promise<void> {
     this.activeFingerprints.add(item.fingerprint);
 

listener/src/services/notification-retry-queue.ts

@@ -5,12 +5,19 @@ import { getEventName } from '../utils/event-utils';
 import { getNotificationAnalyticsAggregator, NotificationAnalyticsAggregator } from './notification-analytics-aggregator';
 import { NotificationType } from '../types/scheduled-notification';
 
+export enum Priority {
+  Low = 0,
+  Medium = 1,
+  High = 2,
+}
+
 export interface RetryQueueOptions {
   baseDelayMs?: number;
   multiplier?: number;
   jitter?: boolean;
   maxRetries?: number;
   processIntervalMs?: number;
+  priorityWeights?: { high: number; medium: number; low: number };
 }
 
 interface RetryItem {
@@ -19,6 +26,8 @@ interface RetryItem {
   retryCount: number;
   nextRetryAt: number;
   requestId?: string;
+  priority: Priority;
+  enqueuedAt: number;
 }
 
 const DEFAULTS = {
@@ -27,6 +36,7 @@ const DEFAULTS = {
   jitter: true,
   maxRetries: 5,
   processIntervalMs: 5_000,
+  priorityWeights: { high: 5, medium: 2, low: 1 },
 };
 
 export type NotificationFn = (
@@ -43,9 +53,11 @@ export class NotificationRetryQueue {
   private readonly jitter: boolean;
   private readonly maxRetries: number;
   private readonly processIntervalMs: number;
+  private readonly priorityWeights: { high: number; medium: number; low: number };
   private timer: ReturnType<typeof setInterval> | null = null;
   private readonly notificationFn: NotificationFn;
   private readonly analytics: NotificationAnalyticsAggregator | null;
+  private priorityCounters: { high: number; medium: number; low: number } = { high: 0, medium: 0, low: 0 };
 
   constructor(notificationFn: NotificationFn, options?: RetryQueueOptions) {
     this.notificationFn = notificationFn;
@@ -54,13 +66,15 @@ export class NotificationRetryQueue {
     this.jitter = options?.jitter ?? DEFAULTS.jitter;
     this.maxRetries = options?.maxRetries ?? DEFAULTS.maxRetries;
     this.processIntervalMs = options?.processIntervalMs ?? DEFAULTS.processIntervalMs;
+    this.priorityWeights = options?.priorityWeights ?? DEFAULTS.priorityWeights;
     this.analytics = getNotificationAnalyticsAggregator();
   }
 
   enqueue(
     event: StellarSDK.rpc.Api.EventResponse,
     contractConfig: ContractConfig,
-    requestId?: string
+    requestId?: string,
+    priority: Priority = Priority.Medium
   ): void {
     const fingerprint = buildRetryFingerprint(event, contractConfig.address);
 
@@ -84,10 +98,11 @@ export class NotificationRetryQueue {
       delayMs,
       nextRetryAt: new Date(nextRetryAt).toISOString(),
       maxRetries: this.maxRetries,
+      priority: Priority[priority],
     });
 
     this.queuedFingerprints.add(fingerprint);
-    this.queue.push({ event, contractConfig, retryCount: 0, nextRetryAt, requestId });
+    this.queue.push({ event, contractConfig, retryCount: 0, nextRetryAt, requestId, priority, enqueuedAt: Date.now() });
   }
 
   start(): void {
@@ -112,14 +127,41 @@ export class NotificationRetryQueue {
 
   private async processQueue(): Promise<void> {
     const now = Date.now();
-    const due = this.queue.filter((item) => item.nextRetryAt <= now);
+    const due = this.queue
+      .filter((item) => item.nextRetryAt <= now)
+      .sort((a, b) => {
+        const priorityA = this.getWeightedPriority(a);
+        const priorityB = this.getWeightedPriority(b);
+        if (priorityB !== priorityA) return priorityB - priorityA;
+        return a.enqueuedAt - b.enqueuedAt;
+      });
+
     this.queue = this.queue.filter((item) => item.nextRetryAt > now);
 
+    for (const item of due) {
+      if (item.priority === Priority.High) this.priorityCounters.high++;
+      else if (item.priority === Priority.Medium) this.priorityCounters.medium++;
+      else this.priorityCounters.low++;
+    }
+
     for (const item of due) {
       await this.retryItem(item);
     }
   }
 
+  private getWeightedPriority(item: RetryItem): number {
+    const basePriority = item.priority;
+    const age = Date.now() - item.enqueuedAt;
+    const ageBonus = Math.floor(age / 60000);
+
+    let weight = 0;
+    if (item.priority === Priority.High) weight = this.priorityWeights.high;
+    else if (item.priority === Priority.Medium) weight = this.priorityWeights.medium;
+    else weight = this.priorityWeights.low;
+
+    return basePriority + ageBonus + weight;
+  }
+
   private async retryItem(item: RetryItem): Promise<void> {
     const attempt = item.retryCount + 1;
     const fingerprint = buildRetryFingerprint(item.event, item.contractConfig.address);