support for session enabled entities

Author: sgebb

README.md

@@ -20,6 +20,16 @@ For visual walkthroughs, screenshots, and detailed guides of these features, vis
 
 ---
 
+## Current Limitations
+
+While Bussin supports most Azure Service Bus operations directly in the browser, there are a few limitations to keep in mind:
+
+* **Batch Send on Session-Enabled Entities**: Batch sending does not currently support session-enabled queues or topic subscriptions (which require a Session ID for each message).
+* **Partitioned & Session-Enabled Entities**: Certain administrative actions (such as the portal-based fast batch delete/purge API) are not supported by the Service Bus broker on partitioned or session-enabled entities. Bussin automatically falls back to compatible data-plane receiver flows (such as parallel receivers or sequential session loops) in these cases to complete the operation safely.
+
+---
+
+
 * **Direct Client-Side Architecture**: Your data never leaves your browser. Bussin communicates directly with Azure APIs.
 * **Zero Installation**: Access the tool online at [https://bussin.dev](https://bussin.dev/) or install it as a PWA.
 * **Entra ID Integration**: Authenticate securely using your existing Azure identity and Role-Based Access Control (RBAC) roles.

client-js/serviceBusApi.ts

@@ -13,9 +13,7 @@ import {
     complete,
     abandon,
     deadLetter,
-    getMessageSessions,
-    getSessionState,
-    setSessionState
+    getMessageSessions
 } from './src/peekOperations.js';
 
 import {
@@ -76,8 +74,6 @@ if (typeof window !== 'undefined') {
         receiveAndLockQueueMessage,
         receiveAndLockSubscriptionMessage,
         getMessageSessions,
-        getSessionState,
-        setSessionState,
 
         // Settlement operations (stateless - take LockedMessage[])
         complete,
@@ -136,8 +132,6 @@ export {
     receiveAndLockQueueMessage,
     receiveAndLockSubscriptionMessage,
     getMessageSessions,
-    getSessionState,
-    setSessionState,
 
     // Settlement operations (stateless - take LockedMessage[])
     complete,

client-js/src/connection.ts

@@ -6,6 +6,7 @@
 import rhea from 'rhea';
 import type { Connection } from 'rhea';
 import type { CBSAuthResult } from './types.js';
+import { formatAmqpError } from './types.js';
 
 import { GlobalMockBroker } from './mockBroker.js';
 
@@ -90,12 +91,12 @@ export class ServiceBusConnection {
 
             this.connection.on('connection_error', (context: any) => {
                 console.error(`[ServiceBusConnection] Connection ERROR:`, context.error);
-                reject(new Error(context.error ? context.error.toString() : 'Connection error'));
+                reject(new Error(context.error ? formatAmqpError(context.error) : 'Connection error'));
             });
 
             this.connection.on('disconnected', (context: any) => {
                 if (context.error && !this.cbsAuthenticated) {
-                    reject(new Error(context.error.toString()));
+                    reject(new Error(formatAmqpError(context.error)));
                 }
             });
         });

client-js/src/managementClient.ts

@@ -6,6 +6,9 @@
 import rhea from 'rhea';
 import type { ServiceBusConnection } from './connection.js';
 import type { Sender, Receiver } from 'rhea';
+import { parseServiceBusMessage } from './messageParser.js';
+import type { LockedMessage } from './types.js';
+import { formatAmqpError } from './types.js';
 
 /**
  * Management Client - for management operations like true peek
@@ -17,11 +20,13 @@ export class ManagementClient {
     private sender: Sender | null = null;
     private receiver: Receiver | null = null;
     private replyTo: string | null = null;
+    private readonly amqpSession?: any;
 
-    constructor(connection: ServiceBusConnection, entityPath: string) {
+    constructor(connection: ServiceBusConnection, entityPath: string, amqpSession?: any) {
         this.connection = connection;
         this.entityPath = entityPath;
         this.managementAddress = `${entityPath}/$management`;
+        this.amqpSession = amqpSession;
     }
 
     /**
@@ -36,11 +41,13 @@ export class ManagementClient {
             // Create unique reply-to address
             this.replyTo = `mgmt-reply-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
 
-            this.sender = this.connection.connection!.open_sender({
+            const sessionOrConn = this.amqpSession || this.connection.connection!;
+
+            this.sender = sessionOrConn.open_sender({
                 target: { address: this.managementAddress }
             });
 
-            this.receiver = this.connection.connection!.open_receiver({
+            this.receiver = sessionOrConn.open_receiver({
                 source: { address: this.managementAddress },
                 target: { address: this.replyTo }
             });
@@ -65,11 +72,11 @@ export class ManagementClient {
             });
 
             this.sender.on('sender_error', (context: any) => {
-                reject(new Error(context.sender.error ? context.sender.error.toString() : 'Sender error'));
+                reject(new Error(context.sender.error ? formatAmqpError(context.sender.error) : 'Sender error'));
             });
 
             this.receiver.on('receiver_error', (context: any) => {
-                reject(new Error(context.receiver.error ? context.receiver.error.toString() : 'Receiver error'));
+                reject(new Error(context.receiver.error ? formatAmqpError(context.receiver.error) : 'Receiver error'));
             });
 
             setTimeout(() => reject(new Error('Management client open timeout')), 10000);
@@ -224,7 +231,7 @@ export class ManagementClient {
      * Lock messages by sequence numbers (peek-lock mode)
      * Returns lock tokens that can be used to complete/abandon/dead-letter
      */
-    async lockBySequenceNumbers(sequenceNumbers: number[]): Promise<{ sequenceNumber: number; lockToken: string }[]> {
+    async lockBySequenceNumbers(sequenceNumbers: number[]): Promise<LockedMessage[]> {
         if (!this.sender || !this.receiver || !this.replyTo) {
             throw new Error('Management client not opened');
         }
@@ -242,18 +249,23 @@ export class ManagementClient {
                 if (statusCode === 200) {
                     // Response body contains messages with lock tokens
                     const body = context.message.body;
-                    const results: { sequenceNumber: number; lockToken: string }[] = [];
+                    const results: LockedMessage[] = [];
 
                     if (body && body.messages) {
                         const msgArray = Array.isArray(body.messages) ? body.messages : [body.messages];
                         for (let i = 0; i < msgArray.length; i++) {
                             const item = msgArray[i];
                             // Lock token is in the 'lock-token' field as a UUID
                             if (item['lock-token']) {
-                                results.push({
-                                    sequenceNumber: sequenceNumbers[i],
-                                    lockToken: uuidFromBuffer(item['lock-token'])
-                                });
+                                const lockToken = uuidFromBuffer(item['lock-token']);
+                                const rawMessage = item['message'];
+                                let parsed: any = {};
+                                if (rawMessage) {
+                                    const decoded = rhea.message.decode(rawMessage);
+                                    parsed = parseServiceBusMessage(decoded);
+                                }
+                                parsed.lockToken = lockToken;
+                                results.push(parsed as LockedMessage);
                             }
                         }
                     }
@@ -294,7 +306,6 @@ export class ManagementClient {
             }, 10000);
         });
     }
-
     /**
      * Cancel scheduled messages by sequence numbers
      */
@@ -774,7 +785,7 @@ export class ManagementClient {
     /**
      * Get session state
      */
-    async getSessionState(sessionId: string): Promise<string> {
+    async getSessionState(sessionId: string, associatedLinkName?: string): Promise<string> {
         if (!this.sender || !this.receiver || !this.replyTo) {
             throw new Error('Management client not opened');
         }
@@ -825,7 +836,8 @@ export class ManagementClient {
                 body: messageBody,
                 reply_to: replyTo,
                 application_properties: {
-                    operation: 'com.microsoft:get-session-state'
+                    operation: 'com.microsoft:get-session-state',
+                    ...(associatedLinkName ? { 'associated-link-name': associatedLinkName } : {})
                 },
                 message_id: messageId
             };
@@ -839,10 +851,7 @@ export class ManagementClient {
         });
     }
 
-    /**
-     * Set session state
-     */
-    async setSessionState(sessionId: string, state: string | null): Promise<void> {
+    async setSessionState(sessionId: string, state: string | null, associatedLinkName?: string): Promise<void> {
         if (!this.sender || !this.receiver || !this.replyTo) {
             throw new Error('Management client not opened');
         }
@@ -881,7 +890,8 @@ export class ManagementClient {
                 body: messageBody,
                 reply_to: replyTo,
                 application_properties: {
-                    operation: 'com.microsoft:set-session-state'
+                    operation: 'com.microsoft:set-session-state',
+                    ...(associatedLinkName ? { 'associated-link-name': associatedLinkName } : {})
                 },
                 message_id: messageId
             };

client-js/src/messageParser.ts

@@ -43,22 +43,117 @@ export function parseServiceBusMessage(amqpMessage: any): ServiceBusMessage {
     }
 
     // Get message annotations
-    const annotations = amqpMessage.message_annotations || {};
+    const annotations = amqpMessage.message_annotations || amqpMessage.messageAnnotations || amqpMessage.delivery_annotations || amqpMessage.deliveryAnnotations || {};
+
+    const getAnnotationValue = (key: string): any => {
+        if (!annotations) return undefined;
+        if (annotations[key] !== undefined) return annotations[key];
+        
+        // Handle Symbol keys
+        const symbols = Object.getOwnPropertySymbols(annotations);
+        for (const sym of symbols) {
+            const symStr = sym.toString();
+            if (symStr === `Symbol(${key})` || sym.description === key || symStr.includes(key)) {
+                return annotations[sym];
+            }
+        }
+        
+        // Check standard keys or Map keys as string
+        for (const k of Object.keys(annotations)) {
+            if (k === key || k.toString() === key) {
+                return annotations[k];
+            }
+        }
+        
+        // If annotations is a Map
+        try {
+            if (typeof annotations.get === 'function') {
+                const val = annotations.get(key);
+                if (val !== undefined) return val;
+                for (const [k, v] of annotations.entries()) {
+                    if (k === key || (k && k.toString() === key)) {
+                        return v;
+                    }
+                }
+            }
+        } catch {}
+
+        return undefined;
+    };
 
-    // Extract enqueued time - could be Date or timestamp
-    let enqueuedTime = annotations['x-opt-enqueued-time'];
-    if (enqueuedTime && typeof enqueuedTime === 'number') {
-        enqueuedTime = new Date(enqueuedTime);
-    }
+    // Helper to robustly decode AMQP long types (number, bigint, high/low object, buffer)
+    const decodeAmqpLong = (val: any): number | undefined => {
+        if (val === undefined || val === null) return undefined;
+        if (typeof val === 'number') return val;
+        if (typeof val === 'bigint') return Number(val);
+        if (typeof val === 'object') {
+            if ('high' in val && 'low' in val) {
+                return val.high * 4294967296 + (val.low >>> 0);
+            }
+            if (val instanceof Uint8Array || val.type === 'Buffer' || Array.isArray(val.data)) {
+                const buf = val.data ? new Uint8Array(val.data) : val;
+                let num = BigInt(0);
+                for (let i = 0; i < buf.length; i++) {
+                    num = (num << BigInt(8)) + BigInt(buf[i]);
+                }
+                return Number(num);
+            }
+        }
+        const parsed = Number(val);
+        return isNaN(parsed) ? undefined : parsed;
+    };
+
+    // Helper to robustly decode AMQP timestamp/datetime types to ISO string
+    const parseAmqpDate = (val: any): string | undefined => {
+        if (val === undefined || val === null) return undefined;
+        
+        let d: Date | undefined;
+        if (val instanceof Date) {
+            d = val;
+        } else {
+            // If it's a number, bigint, or long object/buffer, decode to timestamp
+            const ms = decodeAmqpLong(val);
+            if (ms !== undefined && !isNaN(ms)) {
+                d = new Date(ms);
+            } else if (typeof val === 'string') {
+                d = new Date(val);
+            }
+        }
+        
+        if (d && !isNaN(d.getTime())) {
+            try {
+                const year = d.getUTCFullYear();
+                if (year > 9999) {
+                    return "9999-12-31T23:59:59.999Z";
+                }
+                if (year < 1) {
+                    return "0001-01-01T00:00:00.000Z";
+                }
+                return d.toISOString();
+            } catch {
+                return undefined;
+            }
+        }
+        
+        return undefined;
+    };
+
+    // Extract enqueued time
+    const enqueuedTime = parseAmqpDate(getAnnotationValue('x-opt-enqueued-time'));
 
     // Extract scheduled enqueue time
-    let scheduledEnqueueTime = annotations['x-opt-scheduled-enqueue-time'];
+    let scheduledEnqueueTime = getAnnotationValue('x-opt-scheduled-enqueue-time');
     if (scheduledEnqueueTime instanceof Date) {
         scheduledEnqueueTime = scheduledEnqueueTime.getTime();
+    } else {
+        scheduledEnqueueTime = decodeAmqpLong(scheduledEnqueueTime);
     }
 
+    // Extract sequence number
+    const sequenceNumber = decodeAmqpLong(getAnnotationValue('x-opt-sequence-number') ?? amqpMessage._sequenceNumber);
+
     // Extract partition key
-    const partitionKey = annotations['x-opt-partition-key'] ?? amqpMessage.group_id ?? props.group_id;
+    const partitionKey = getAnnotationValue('x-opt-partition-key') ?? amqpMessage.group_id ?? props.group_id;
 
     return {
         messageId: amqpMessage.message_id ?? props.message_id,
@@ -71,16 +166,17 @@ export function parseServiceBusMessage(amqpMessage: any): ServiceBusMessage {
         to: amqpMessage.to ?? props.to,
         deliveryCount: amqpMessage.delivery_count ?? amqpMessage.header?.delivery_count ?? 0,
         enqueuedTime: enqueuedTime,
-        sequenceNumber: (annotations && annotations['x-opt-sequence-number']) ?? amqpMessage._sequenceNumber,
-        lockedUntil: (annotations && annotations['x-opt-locked-until']),
+        sequenceNumber: sequenceNumber,
+        lockedUntil: parseAmqpDate(getAnnotationValue('x-opt-locked-until')),
         scheduledEnqueueTime: scheduledEnqueueTime,
         partitionKey: partitionKey,
+        state: getAnnotationValue('x-opt-state'),
         applicationProperties: amqpMessage.application_properties || {},
         messageAnnotations: annotations,
         properties: props,
         ttl: ttl,
-        expiryTime: amqpMessage.absolute_expiry_time ?? props.absolute_expiry_time,
-        creationTime: amqpMessage.creation_time ?? props.creation_time
+        expiryTime: parseAmqpDate(amqpMessage.absolute_expiry_time ?? props.absolute_expiry_time),
+        creationTime: parseAmqpDate(amqpMessage.creation_time ?? props.creation_time)
     };
 }
 

client-js/src/messageSender.ts

@@ -5,6 +5,7 @@
 
 import type { ServiceBusConnection } from './connection.js';
 import type { MessageProperties } from './types.js';
+import { formatAmqpError } from './types.js';
 import type { Sender } from 'rhea';
 import { message as rheaMessage } from 'rhea';
 
@@ -39,7 +40,7 @@ export class MessageSender {
             });
 
             this.sender.on('sender_error', (context: any) => {
-                reject(new Error(context.sender.error ? context.sender.error.toString() : 'Sender error'));
+                reject(new Error(context.sender.error ? formatAmqpError(context.sender.error) : 'Sender error'));
             });
 
             setTimeout(() => reject(new Error('Sender open timeout')), 5000);