diff --git a/smnb/app/dashboard/studio/controls/Controls.tsx b/smnb/app/dashboard/studio/controls/Controls.tsx index b37d9de..313b991 100644 --- a/smnb/app/dashboard/studio/controls/Controls.tsx +++ b/smnb/app/dashboard/studio/controls/Controls.tsx @@ -8,6 +8,7 @@ import { useSimpleLiveFeedStore } from '@/lib/stores/livefeed/simpleLiveFeedStor import { useHostAgentStore } from '@/lib/stores/host/hostAgentStore'; import { useProducerStore } from '@/lib/stores/producer/producerStore'; import { useApiKeyStore } from '@/lib/stores/apiKeyStore'; +import { useQueueManagerStore } from '@/lib/stores/queueManagerStore'; // import { useEditorAgentStore } from '@/lib/stores/host/editorAgentStore'; // Commented out - editor functionality disabled import { StudioMode } from '../Studio'; import { @@ -108,7 +109,9 @@ export default function Controls({ mode, onModeChange }: ControlsProps) { start: startHostBroadcasting, stop: stopHostBroadcasting, stats: hostStats, - processLiveFeedPost + processLiveFeedPost, + getQueueStatus, + getQueueBySubreddit } = useHostAgentStore(); const { @@ -123,6 +126,14 @@ export default function Controls({ mode, onModeChange }: ControlsProps) { hasValidKey: hasValidApiKey } = useApiKeyStore(); + const { + initializeQueueManager, + clearQueueBySubreddit, + clearAllQueues, + refreshStats, + getDetailedStatus + } = useQueueManagerStore(); + // Editor agent store - commented out /* const { @@ -226,19 +237,34 @@ export default function Controls({ mode, onModeChange }: ControlsProps) { } }; - const handleRemoveSubreddit = (subreddit: string) => { - // Remove from enabledDefaults - const updatedEnabledDefaults = enabledDefaults.filter(s => s !== subreddit); - setEnabledDefaults(updatedEnabledDefaults); - - // Also remove from customSubreddits if it exists there - const updatedCustom = customSubreddits.filter(s => s !== subreddit); - setCustomSubreddits(updatedCustom); - - // Update the selected subreddits - updateSelectedSubreddits(updatedEnabledDefaults, updatedCustom); - - console.log(`🗑️ Removed subreddit: ${subreddit}`); + const handleRemoveSubreddit = async (subreddit: string) => { + try { + // Remove from enabledDefaults + const updatedEnabledDefaults = enabledDefaults.filter(s => s !== subreddit); + setEnabledDefaults(updatedEnabledDefaults); + + // Also remove from customSubreddits if it exists there + const updatedCustom = customSubreddits.filter(s => s !== subreddit); + setCustomSubreddits(updatedCustom); + + // Update the selected subreddits + updateSelectedSubreddits(updatedEnabledDefaults, updatedCustom); + + console.log(`🗑️ Removed subreddit: ${subreddit}`); + + // Clear queue items for this subreddit + try { + const result = await clearQueueBySubreddit(subreddit); + if (result.totalCleared > 0) { + console.log(`✅ Queue cleared for r/${subreddit}: ${result.totalCleared} items (Host: ${result.hostQueueCleared}, Scheduled: ${result.scheduledPostsCleared})`); + } + } catch (error) { + console.warn(`⚠️ Failed to clear queue for r/${subreddit}:`, error); + // Don't block the UI operation if queue clearing fails + } + } catch (error) { + console.error(`❌ Error removing subreddit ${subreddit}:`, error); + } }; const handleToggleDefaultSubreddit = (subreddit: string) => { @@ -334,6 +360,11 @@ export default function Controls({ mode, onModeChange }: ControlsProps) { updateSelectedSubreddits(enabledDefaults, customSubreddits); }, [enabledDefaults, customSubreddits, updateSelectedSubreddits]); + // Initialize queue manager + useEffect(() => { + initializeQueueManager(); + }, [initializeQueueManager]); + return (
{/* Compact Header */} @@ -670,6 +701,64 @@ export default function Controls({ mode, onModeChange }: ControlsProps) {
+ + {/* Queue Status */} +
+
+ Queue +
+ + {(() => { + const queueStatus = getQueueStatus(); + const hostQueue = getQueueBySubreddit(); + const totalItems = Object.values(hostQueue).reduce((sum: number, count: number) => sum + count, 0); + return totalItems; + })()} + + {(() => { + const hostQueue = getQueueBySubreddit(); + const totalItems = Object.values(hostQueue).reduce((sum: number, count: number) => sum + count, 0); + return totalItems > 0 ? ( + + ) : null; + })()} +
+
+ {(() => { + const hostQueue = getQueueBySubreddit(); + const topSubreddits = Object.entries(hostQueue) + .sort(([,a], [,b]) => (b as number) - (a as number)) + .slice(0, 3); + + return topSubreddits.length > 0 ? ( +
+ {topSubreddits.map(([subreddit, count]) => ( +
+ + r/{subreddit} + + + {count} + +
+ ))} +
+ ) : null; + })()} +
diff --git a/smnb/lib/services/host/hostAgentService.ts b/smnb/lib/services/host/hostAgentService.ts index a6adaee..9d78428 100644 --- a/smnb/lib/services/host/hostAgentService.ts +++ b/smnb/lib/services/host/hostAgentService.ts @@ -354,6 +354,23 @@ export class HostAgentService extends EventEmitter { this.emit('queue:updated', 0); } + public clearQueueBySubreddit(subreddit: string): void { + const beforeLength = this.state.narrationQueue.length; + this.state.narrationQueue = this.state.narrationQueue.filter(narration => { + const itemSubreddit = narration.metadata?.originalItem?.subreddit; + return itemSubreddit !== subreddit; + }); + const afterLength = this.state.narrationQueue.length; + const removedCount = beforeLength - afterLength; + + console.log(`🗑️ Cleared ${removedCount} narration(s) from queue for subreddit: r/${subreddit}`); + this.emit('queue:updated', this.state.narrationQueue.length); + + if (removedCount > 0) { + this.emit('queue:subreddit-cleared', { subreddit, removedCount }); + } + } + public getQueueStatus(): { length: number; isProcessing: boolean; @@ -368,6 +385,19 @@ export class HostAgentService extends EventEmitter { }; } + public getQueueBySubreddit(): { [subreddit: string]: number } { + const subredditCounts: { [subreddit: string]: number } = {}; + + this.state.narrationQueue.forEach(narration => { + const subreddit = narration.metadata?.originalItem?.subreddit; + if (subreddit) { + subredditCounts[subreddit] = (subredditCounts[subreddit] || 0) + 1; + } + }); + + return subredditCounts; + } + // 🎯 DUPLICATE DETECTION UTILITIES /** diff --git a/smnb/lib/services/livefeed/schedulerService.ts b/smnb/lib/services/livefeed/schedulerService.ts index b0a396d..06d79d5 100644 --- a/smnb/lib/services/livefeed/schedulerService.ts +++ b/smnb/lib/services/livefeed/schedulerService.ts @@ -224,6 +224,41 @@ export class SchedulerService { categoryDistribution, }; } + + /** + * Clear all scheduled posts for a specific subreddit + */ + clearScheduledPostsBySubreddit(subreddit: string): number { + const beforeLength = this.scheduledPosts.length; + this.scheduledPosts = this.scheduledPosts.filter(post => post.subreddit !== subreddit); + const removedCount = beforeLength - this.scheduledPosts.length; + + console.log(`🗑️ SchedulerService: Cleared ${removedCount} scheduled post(s) for subreddit: r/${subreddit}`); + return removedCount; + } + + /** + * Get count of scheduled posts by subreddit + */ + getScheduledPostsBySubreddit(): { [subreddit: string]: number } { + const subredditCounts: { [subreddit: string]: number } = {}; + + this.scheduledPosts.forEach(post => { + subredditCounts[post.subreddit] = (subredditCounts[post.subreddit] || 0) + 1; + }); + + return subredditCounts; + } + + /** + * Clear all scheduled posts (for debugging/management) + */ + clearAllScheduledPosts(): number { + const count = this.scheduledPosts.length; + this.scheduledPosts = []; + console.log(`🗑️ SchedulerService: Cleared all ${count} scheduled posts`); + return count; + } } export const schedulerService = new SchedulerService(); diff --git a/smnb/lib/services/queueManagerService.ts b/smnb/lib/services/queueManagerService.ts new file mode 100644 index 0000000..49d9845 --- /dev/null +++ b/smnb/lib/services/queueManagerService.ts @@ -0,0 +1,196 @@ +// QUEUE MANAGER SERVICE +// /Users/matthewsimon/Projects/SMNB/smnb/lib/services/queueManagerService.ts + +/** + * Queue Manager Service + * + * Centralized service for managing queues across the application. + * Coordinates queue operations between Host Agent and Scheduler Service. + */ + +import { EventEmitter } from 'events'; +import { HostAgentService } from './host/hostAgentService'; +import { SchedulerService } from './livefeed/schedulerService'; + +export interface QueueManagerStats { + totalHostQueue: number; + totalScheduledPosts: number; + queueBySubreddit: { [subreddit: string]: number }; + scheduledBySubreddit: { [subreddit: string]: number }; +} + +export interface QueueClearResult { + subreddit: string; + hostQueueCleared: number; + scheduledPostsCleared: number; + totalCleared: number; +} + +export class QueueManagerService extends EventEmitter { + private hostAgentService: HostAgentService | null = null; + private schedulerService: SchedulerService | null = null; + public isInitialized: boolean = false; + + constructor() { + super(); + console.log('🎯 QueueManagerService initialized'); + } + + /** + * Initialize the queue manager with required services + */ + initialize(hostAgent: HostAgentService, scheduler: SchedulerService): void { + this.hostAgentService = hostAgent; + this.schedulerService = scheduler; + this.isInitialized = true; + + console.log('🔗 QueueManagerService connected to Host Agent and Scheduler'); + + // Listen to queue events from host agent + this.hostAgentService.on('queue:updated', (length: number) => { + this.emit('queue:host-updated', length); + }); + + this.hostAgentService.on('queue:subreddit-cleared', (data: { subreddit: string; removedCount: number }) => { + this.emit('queue:subreddit-cleared', { source: 'host', ...data }); + }); + } + + /** + * Clear all queue items for a specific subreddit across all services + */ + clearQueueBySubreddit(subreddit: string): QueueClearResult { + if (!this.hostAgentService || !this.schedulerService) { + throw new Error('QueueManagerService not properly initialized'); + } + + console.log(`🎯 QueueManager: Clearing all queues for subreddit: r/${subreddit}`); + + // Clear from host agent narration queue + const hostQueueBefore = this.hostAgentService.getQueueStatus().length; + this.hostAgentService.clearQueueBySubreddit(subreddit); + const hostQueueAfter = this.hostAgentService.getQueueStatus().length; + const hostCleared = hostQueueBefore - hostQueueAfter; + + // Clear from scheduler service + const scheduledCleared = this.schedulerService.clearScheduledPostsBySubreddit(subreddit); + + const result: QueueClearResult = { + subreddit, + hostQueueCleared: hostCleared, + scheduledPostsCleared: scheduledCleared, + totalCleared: hostCleared + scheduledCleared + }; + + console.log(`✅ QueueManager: Cleared ${result.totalCleared} items total for r/${subreddit} (Host: ${hostCleared}, Scheduled: ${scheduledCleared})`); + + this.emit('queue:subreddit-cleared-complete', result); + + return result; + } + + /** + * Clear all queues completely + */ + clearAllQueues(): { hostCleared: number; scheduledCleared: number; totalCleared: number } { + if (!this.hostAgentService || !this.schedulerService) { + throw new Error('QueueManagerService not properly initialized'); + } + + console.log('🎯 QueueManager: Clearing ALL queues'); + + const hostQueueBefore = this.hostAgentService.getQueueStatus().length; + this.hostAgentService.clearQueue(); + const hostCleared = hostQueueBefore; + + const scheduledCleared = this.schedulerService.clearAllScheduledPosts(); + + const result = { + hostCleared, + scheduledCleared, + totalCleared: hostCleared + scheduledCleared + }; + + console.log(`✅ QueueManager: Cleared ${result.totalCleared} items total (Host: ${hostCleared}, Scheduled: ${scheduledCleared})`); + + this.emit('queue:all-cleared', result); + + return result; + } + + /** + * Get comprehensive queue statistics + */ + getQueueStats(): QueueManagerStats { + if (!this.hostAgentService || !this.schedulerService) { + throw new Error('QueueManagerService not properly initialized'); + } + + const hostStatus = this.hostAgentService.getQueueStatus(); + const hostBySubreddit = this.hostAgentService.getQueueBySubreddit(); + const schedulingStats = this.schedulerService.getSchedulingStats(); + const scheduledBySubreddit = this.schedulerService.getScheduledPostsBySubreddit(); + + return { + totalHostQueue: hostStatus.length, + totalScheduledPosts: schedulingStats.scheduledCount, + queueBySubreddit: hostBySubreddit, + scheduledBySubreddit: scheduledBySubreddit + }; + } + + /** + * Get detailed status information + */ + getDetailedStatus(): { + isInitialized: boolean; + hostAgent: { isActive: boolean; queueLength: number; isProcessing: boolean }; + scheduler: { scheduledCount: number; nextPublishTime: number | null }; + subredditBreakdown: { [subreddit: string]: { host: number; scheduled: number; total: number } }; + } { + if (!this.hostAgentService || !this.schedulerService) { + return { + isInitialized: false, + hostAgent: { isActive: false, queueLength: 0, isProcessing: false }, + scheduler: { scheduledCount: 0, nextPublishTime: null }, + subredditBreakdown: {} + }; + } + + const hostStatus = this.hostAgentService.getQueueStatus(); + const schedulingStats = this.schedulerService.getSchedulingStats(); + const hostBySubreddit = this.hostAgentService.getQueueBySubreddit(); + const scheduledBySubreddit = this.schedulerService.getScheduledPostsBySubreddit(); + + // Combine subreddit data + const allSubreddits = new Set([...Object.keys(hostBySubreddit), ...Object.keys(scheduledBySubreddit)]); + const subredditBreakdown: { [subreddit: string]: { host: number; scheduled: number; total: number } } = {}; + + allSubreddits.forEach(subreddit => { + const hostCount = hostBySubreddit[subreddit] || 0; + const scheduledCount = scheduledBySubreddit[subreddit] || 0; + subredditBreakdown[subreddit] = { + host: hostCount, + scheduled: scheduledCount, + total: hostCount + scheduledCount + }; + }); + + return { + isInitialized: true, + hostAgent: { + isActive: hostStatus.isActive, + queueLength: hostStatus.length, + isProcessing: hostStatus.isProcessing + }, + scheduler: { + scheduledCount: schedulingStats.scheduledCount, + nextPublishTime: schedulingStats.nextPublishTime + }, + subredditBreakdown + }; + } +} + +// Export singleton instance +export const queueManagerService = new QueueManagerService(); \ No newline at end of file diff --git a/smnb/lib/stores/host/hostAgentStore.ts b/smnb/lib/stores/host/hostAgentStore.ts index fb0535f..79caf31 100644 --- a/smnb/lib/stores/host/hostAgentStore.ts +++ b/smnb/lib/stores/host/hostAgentStore.ts @@ -134,6 +134,12 @@ interface HostAgentState { // State management clearAllState: () => void; + // Queue management actions + clearQueue: () => void; + clearQueueBySubreddit: (subreddit: string) => void; + getQueueStatus: () => { length: number; isProcessing: boolean; isActive: boolean; currentNarration?: string }; + getQueueBySubreddit: () => { [subreddit: string]: number }; + // Countdown actions startCountdown: (seconds: number) => void; stopCountdown: () => void; @@ -543,5 +549,42 @@ export const useHostAgentStore = create((set, get) => ({ })); console.log('✅ Complete host agent state reset completed'); + }, + + // Queue management methods + clearQueue: () => { + const { hostAgent } = get(); + if (hostAgent) { + hostAgent.clearQueue(); + console.log('🗑️ HOST STORE: Queue cleared via store action'); + } else { + console.warn('⚠️ HOST STORE: Cannot clear queue - host agent not initialized'); + } + }, + + clearQueueBySubreddit: (subreddit: string) => { + const { hostAgent } = get(); + if (hostAgent) { + hostAgent.clearQueueBySubreddit(subreddit); + console.log(`🗑️ HOST STORE: Queue cleared for subreddit r/${subreddit} via store action`); + } else { + console.warn('⚠️ HOST STORE: Cannot clear queue by subreddit - host agent not initialized'); + } + }, + + getQueueStatus: () => { + const { hostAgent } = get(); + if (hostAgent) { + return hostAgent.getQueueStatus(); + } + return { length: 0, isProcessing: false, isActive: false }; + }, + + getQueueBySubreddit: () => { + const { hostAgent } = get(); + if (hostAgent) { + return hostAgent.getQueueBySubreddit(); + } + return {}; } })); diff --git a/smnb/lib/stores/queueManagerStore.ts b/smnb/lib/stores/queueManagerStore.ts new file mode 100644 index 0000000..fa54d79 --- /dev/null +++ b/smnb/lib/stores/queueManagerStore.ts @@ -0,0 +1,207 @@ +// QUEUE MANAGER STORE +// /Users/matthewsimon/Projects/SMNB/smnb/lib/stores/queueManagerStore.ts + +/** + * Queue Manager Store + * + * Zustand store for managing queue operations across the application. + * Provides centralized queue management for Host Agent and Scheduler Service. + */ + +import { create } from 'zustand'; +import { QueueManagerService, QueueManagerStats, QueueClearResult, queueManagerService } from '@/lib/services/queueManagerService'; +import { schedulerService } from '@/lib/services/livefeed/schedulerService'; + +export interface QueueManagerState { + // Service instance + queueManager: QueueManagerService | null; + + // State + isInitialized: boolean; + + // Statistics + stats: QueueManagerStats; + + // Actions + initializeQueueManager: () => void; + attemptServiceConnection: () => void; + clearQueueBySubreddit: (subreddit: string) => Promise; + clearAllQueues: () => Promise<{ hostCleared: number; scheduledCleared: number; totalCleared: number }>; + refreshStats: () => void; + getDetailedStatus: () => ReturnType; +} + +export const useQueueManagerStore = create((set, get) => ({ + // Initial state + queueManager: null, + isInitialized: false, + stats: { + totalHostQueue: 0, + totalScheduledPosts: 0, + queueBySubreddit: {}, + scheduledBySubreddit: {} + }, + + // Initialize queue manager + initializeQueueManager: () => { + const { queueManager } = get(); + + if (queueManager) { + console.log('🎯 QUEUE MANAGER STORE: Already initialized'); + + // Try to connect to services even if already initialized + if (!queueManagerService.isInitialized) { + // Get host agent from host store - we need to import it dynamically to avoid circular deps + try { + // We'll handle the connection in a different way to avoid circular imports + console.log('🔗 QUEUE MANAGER STORE: Attempting to connect to services'); + setTimeout(() => { + // Delay connection attempt to allow other stores to initialize + get().attemptServiceConnection(); + }, 1000); + } catch (error) { + console.warn('⚠️ QUEUE MANAGER STORE: Could not connect to services yet:', error); + } + } + + return; + } + + try { + // Set up event listeners + queueManagerService.on('queue:subreddit-cleared-complete', (result: QueueClearResult) => { + console.log(`✅ QUEUE STORE: Subreddit queue cleared - ${result.subreddit}: ${result.totalCleared} items`); + get().refreshStats(); + }); + + queueManagerService.on('queue:all-cleared', (result: { totalCleared: number }) => { + console.log(`✅ QUEUE STORE: All queues cleared - ${result.totalCleared} items`); + get().refreshStats(); + }); + + set(() => ({ + queueManager: queueManagerService, + isInitialized: false // Will be set to true when host/scheduler are connected + })); + + console.log('🎯 QUEUE MANAGER STORE: Queue manager service attached'); + + // Try to connect to services immediately + setTimeout(() => { + get().attemptServiceConnection(); + }, 500); + + } catch (error) { + console.error('❌ QUEUE MANAGER STORE: Failed to initialize queue manager:', error); + } + }, + + // Helper to attempt service connection + attemptServiceConnection: () => { + if (queueManagerService.isInitialized) { + set(() => ({ isInitialized: true })); + return; + } + + try { + // Dynamic import to avoid circular dependencies + import('@/lib/stores/host/hostAgentStore').then(({ useHostAgentStore }) => { + const hostState = useHostAgentStore.getState(); + + if (hostState.hostAgent) { + console.log('🔗 QUEUE MANAGER STORE: Connecting to host agent and scheduler'); + queueManagerService.initialize(hostState.hostAgent, schedulerService); + set(() => ({ isInitialized: true })); + } else { + console.log('⏳ QUEUE MANAGER STORE: Host agent not ready yet, will retry'); + // Retry after a delay + setTimeout(() => { + get().attemptServiceConnection(); + }, 2000); + } + }).catch(error => { + console.warn('⚠️ QUEUE MANAGER STORE: Could not import host store:', error); + }); + } catch (error) { + console.warn('⚠️ QUEUE MANAGER STORE: Could not connect to services:', error); + } + }, + + // Clear queue by subreddit + clearQueueBySubreddit: async (subreddit: string): Promise => { + const { queueManager } = get(); + + if (!queueManager) { + throw new Error('Queue manager not initialized'); + } + + try { + console.log(`🗑️ QUEUE STORE: Clearing queues for subreddit: r/${subreddit}`); + const result = queueManager.clearQueueBySubreddit(subreddit); + + // Refresh stats after clearing + get().refreshStats(); + + return result; + } catch (error) { + console.error(`❌ QUEUE STORE: Failed to clear queue for r/${subreddit}:`, error); + throw error; + } + }, + + // Clear all queues + clearAllQueues: async (): Promise<{ hostCleared: number; scheduledCleared: number; totalCleared: number }> => { + const { queueManager } = get(); + + if (!queueManager) { + throw new Error('Queue manager not initialized'); + } + + try { + console.log('🗑️ QUEUE STORE: Clearing all queues'); + const result = queueManager.clearAllQueues(); + + // Refresh stats after clearing + get().refreshStats(); + + return result; + } catch (error) { + console.error('❌ QUEUE STORE: Failed to clear all queues:', error); + throw error; + } + }, + + // Refresh statistics + refreshStats: () => { + const { queueManager } = get(); + + if (!queueManager) { + console.warn('⚠️ QUEUE STORE: Cannot refresh stats - queue manager not initialized'); + return; + } + + try { + const stats = queueManager.getQueueStats(); + set(() => ({ stats })); + } catch (error) { + console.warn('⚠️ QUEUE STORE: Failed to refresh stats:', error); + // Don't throw, just log the error + } + }, + + // Get detailed status + getDetailedStatus: () => { + const { queueManager } = get(); + + if (!queueManager) { + return { + isInitialized: false, + hostAgent: { isActive: false, queueLength: 0, isProcessing: false }, + scheduler: { scheduledCount: 0, nextPublishTime: null }, + subredditBreakdown: {} + }; + } + + return queueManager.getDetailedStatus(); + } +})); \ No newline at end of file