Persist priority config (#52)

* Persist priority config

* Remove override values

Author: TarikGul

backend/src/db/schema.ts

@@ -102,3 +102,16 @@ export const balanceVerification = sqliteTable('balance_verification', {
 
 export type BalanceVerification = typeof balanceVerification.$inferSelect;
 export type NewBalanceVerification = typeof balanceVerification.$inferInsert;
+
+// Queue Priority Configs - stores the latest priority configuration for UMP and DMP queues
+export const queuePriorityConfigs = sqliteTable('queue_priority_configs', {
+  id: integer('id').primaryKey({ autoIncrement: true }),
+  queueType: text('queue_type').notNull().unique(), // 'ump' or 'dmp'
+  priorityType: text('priority_type').notNull(), // 'Config', 'OverrideConfig', or 'Disabled'
+  lastUpdated: integer('last_updated', { mode: 'timestamp' })
+    .notNull()
+    .$defaultFn(() => new Date()),
+});
+
+export type QueuePriorityConfig = typeof queuePriorityConfigs.$inferSelect;
+export type NewQueuePriorityConfig = typeof queuePriorityConfigs.$inferInsert;

backend/src/routes/updates.ts

@@ -2,7 +2,7 @@ import { desc, eq, and } from 'drizzle-orm';
 import { Request, Response, RequestHandler } from 'express';
 
 import { db } from '../db';
-import { migrationStages, xcmMessageCounters, balanceVerification } from '../db/schema';
+import { migrationStages, xcmMessageCounters, balanceVerification, queuePriorityConfigs } from '../db/schema';
 import { Log } from '../logging/Log';
 import { PalletMigrationCache } from '../services/cache/PalletMigrationCache';
 import { TimeInStageCache } from '../services/cache/TimeInStageCache';
@@ -230,6 +230,32 @@ export const updatesHandler: RequestHandler = async (req: Request, res: Response
         }
       }
     }
+
+    // Handle UMP queue priority initial state
+    if (requestedEvents.includes('umpQueuePriority')) {
+      const umpPriority = await db.query.queuePriorityConfigs.findFirst({
+        where: eq(queuePriorityConfigs.queueType, 'ump'),
+      });
+      if (umpPriority) {
+        sendEvent('umpQueuePriority', {
+          type: umpPriority.priorityType,
+          timestamp: umpPriority.lastUpdated.toISOString(),
+        });
+      }
+    }
+
+    // Handle DMP queue priority initial state
+    if (requestedEvents.includes('dmpQueuePriority')) {
+      const dmpPriority = await db.query.queuePriorityConfigs.findFirst({
+        where: eq(queuePriorityConfigs.queueType, 'dmp'),
+      });
+      if (dmpPriority) {
+        sendEvent('dmpQueuePriority', {
+          type: dmpPriority.priorityType,
+          timestamp: dmpPriority.lastUpdated.toISOString(),
+        });
+      }
+    }
   } catch (error) {
     logger.error('Error sending initial state:', error);
   }

backend/src/services/BlockProcessor.ts

@@ -15,6 +15,9 @@ type AhEventEnum = AhSystemEvent['event'];
 type AhMigratorEvents = Extract<AhEventEnum, { type: 'AhMigrator' }>['value'];
 type BatchProcessedEvent = Extract<AhMigratorEvents, { type: 'BatchProcessed' }>['value'];
 
+// Queue priority config types
+type QueuePriorityConfig = Awaited<ReturnType<RelayChainApi['query']['RcMigrator']['AhUmpQueuePriorityConfig']['getValue']>>;
+
 import { eq, desc, and } from 'drizzle-orm';
 
 import { sql } from 'drizzle-orm';
@@ -27,6 +30,7 @@ import {
   dmpQueueEvents,
   umpQueueEvents,
   balanceVerification,
+  queuePriorityConfigs,
 } from '../db/schema';
 import { Log } from '../logging/Log';
 import { getCurrentStageForPallet, getPalletFromStage, normalizeEventPalletName } from '../util/stageToPalletMapping';
@@ -1022,22 +1026,56 @@ export class BlockProcessor {
   }
 
   private async handleUmpQueuePriority(
-    queuePriority: any,
+    queuePriority: QueuePriorityConfig,
     item: QueueItem
   ): Promise<void> {
     try {
       const priorityType = queuePriority.type;
 
-      eventService.emit('umpQueuePriority', {
-        type: priorityType,
-        timestamp: new Date(item.timestamp!).toISOString(),
+      const existing = await db.query.queuePriorityConfigs.findFirst({
+        where: eq(queuePriorityConfigs.queueType, 'ump'),
       });
 
-      Log.service({
-        service: 'UMP Queue Priority',
-        action: 'UMP queue priority config retrieved',
-        details: { type: priorityType },
-      });
+      // Only update if this is the first time, or if the priority type has actually changed
+      if (!existing) {
+        // First time - insert new record
+        await db.insert(queuePriorityConfigs).values({
+          queueType: 'ump',
+          priorityType,
+          lastUpdated: new Date(item.timestamp!),
+        });
+
+        eventService.emit('umpQueuePriority', {
+          type: priorityType,
+          timestamp: new Date(item.timestamp!).toISOString(),
+        });
+
+        Log.service({
+          service: 'UMP Queue Priority',
+          action: 'UMP queue priority config initialized',
+          details: { type: priorityType },
+        });
+      } else if (existing.priorityType !== priorityType) {
+        // Priority type has changed - update existing record
+        await db.update(queuePriorityConfigs)
+          .set({
+            priorityType,
+            lastUpdated: new Date(item.timestamp!),
+          })
+          .where(eq(queuePriorityConfigs.queueType, 'ump'));
+
+        eventService.emit('umpQueuePriority', {
+          type: priorityType,
+          timestamp: new Date(item.timestamp!).toISOString(),
+        });
+
+        Log.service({
+          service: 'UMP Queue Priority',
+          action: 'UMP queue priority config changed',
+          details: { type: priorityType },
+        });
+      }
+      // else: priority type hasn't changed, do nothing
     } catch (error) {
       Log.service({
         service: 'UMP Queue Priority',
@@ -1048,22 +1086,56 @@ export class BlockProcessor {
   }
 
   private async handleDmpQueuePriority(
-    queuePriority: any,
+    queuePriority: QueuePriorityConfig,
     item: QueueItem
   ): Promise<void> {
     try {
       const priorityType = queuePriority.type;
 
-      eventService.emit('dmpQueuePriority', {
-        type: priorityType,
-        timestamp: new Date(item.timestamp!).toISOString(),
+      const existing = await db.query.queuePriorityConfigs.findFirst({
+        where: eq(queuePriorityConfigs.queueType, 'dmp'),
       });
 
-      Log.service({
-        service: 'DMP Queue Priority',
-        action: 'DMP queue priority config retrieved',
-        details: { type: priorityType },
-      });
+      // Only update if this is the first time, or if the priority type has actually changed
+      if (!existing) {
+        // First time - insert new record
+        await db.insert(queuePriorityConfigs).values({
+          queueType: 'dmp',
+          priorityType,
+          lastUpdated: new Date(item.timestamp!),
+        });
+
+        eventService.emit('dmpQueuePriority', {
+          type: priorityType,
+          timestamp: new Date(item.timestamp!).toISOString(),
+        });
+
+        Log.service({
+          service: 'DMP Queue Priority',
+          action: 'DMP queue priority config initialized',
+          details: { type: priorityType },
+        });
+      } else if (existing.priorityType !== priorityType) {
+        // Priority type has changed - update existing record
+        await db.update(queuePriorityConfigs)
+          .set({
+            priorityType,
+            lastUpdated: new Date(item.timestamp!),
+          })
+          .where(eq(queuePriorityConfigs.queueType, 'dmp'));
+
+        eventService.emit('dmpQueuePriority', {
+          type: priorityType,
+          timestamp: new Date(item.timestamp!).toISOString(),
+        });
+
+        Log.service({
+          service: 'DMP Queue Priority',
+          action: 'DMP queue priority config changed',
+          details: { type: priorityType },
+        });
+      }
+      // else: priority type hasn't changed, do nothing
     } catch (error) {
       Log.service({
         service: 'DMP Queue Priority',