import { createAction, DynamicPropsValue, Property, ActionContext, StoreScope, } from '@activepieces/pieces-framework'; import { QueueManager } from '../common/queue-manager'; import { MCPManager } from '../common/mcp-manager'; import { QueueConfiguration, QueueState } from '../common/types'; /** * Action: Create/Update Queue * Builds or updates a queue configuration and returns current status. * Now uses MCP tools instead of regular pieces. */ export const createUpdateQueue = createAction({ name: 'create_update_queue', displayName: 'Create/Update Queue', description: 'Create a new queue or update existing queue configuration for MCP tools', props: { /* -------------------- MCP Configuration -------------------- */ targetMCPTool: Property.Dropdown({ displayName: 'MCP Tool', description: 'The MCP tool that all items in this queue will use', required: true, refreshers: [], options: async ({ auth }) => { const authConfig = auth as { mcpServerUrl?: string }; if (!authConfig?.mcpServerUrl) { return { options: [], placeholder: 'MCP Server URL required in auth configuration' }; } try { const config = { mcpServerUrl: authConfig.mcpServerUrl }; const tools = await MCPManager.listMCPTools(config); if (!tools || tools.length === 0) { return { options: [], placeholder: 'No MCP tools available from server' }; } return { options: tools.map((tool: any) => ({ label: `${tool.name} - ${tool.description || 'No description'}`, value: tool.name, })) }; } catch (error: any) { console.error('Failed to fetch MCP tools:', error); return { options: [], disabled: true, placeholder: `Error: ${error.message || 'Failed to connect to MCP server'}` }; } }, }), queueSettings: Property.DynamicProperties({ displayName: 'Queue Settings', description: 'Configure queue settings (auto-loads existing settings if queue exists)', required: false, refreshers: ['targetMCPTool'], props: async ({ targetMCPTool, auth }, context) => { // Default values for new queue const defaults = { delayType: 'fixed', delayUnit: 'seconds', delayValue: 30, delayMin: 20, delayMax: 60, dailyLimit: 100, hourlyLimit: 20, activeHours: { timezone: 'America/New_York', schedule: { monday: { enabled: true, start: '09:00', end: '17:00' }, tuesday: { enabled: true, start: '09:00', end: '17:00' }, wednesday: { enabled: true, start: '09:00', end: '17:00' }, thursday: { enabled: true, start: '09:00', end: '17:00' }, friday: { enabled: true, start: '09:00', end: '17:00' }, saturday: { enabled: false }, sunday: { enabled: false }, }, } }; if (!targetMCPTool) { return { instructions: Property.MarkDown({ value: `**Select an MCP tool above to configure queue settings.**` }), delayType: Property.StaticDropdown({ displayName: 'Delay Type', description: 'Select an MCP tool first', required: true, defaultValue: 'fixed', options: { options: [{ label: 'Select Tool First', value: 'fixed' }] }, }), delayUnit: Property.StaticDropdown({ displayName: 'Delay Unit', description: 'Select an MCP tool first', required: true, defaultValue: 'seconds', options: { options: [{ label: 'Select Tool First', value: 'seconds' }] }, }), delayValue: Property.Number({ displayName: 'Delay Value', description: 'Select an MCP tool first', required: false, defaultValue: 0, }), delayMin: Property.Number({ displayName: 'Minimum Delay', description: 'Select an MCP tool first', required: false, defaultValue: 0, }), delayMax: Property.Number({ displayName: 'Maximum Delay', description: 'Select an MCP tool first', required: false, defaultValue: 0, }), dailyLimit: Property.Number({ displayName: 'Daily Limit', description: 'Select an MCP tool first', required: false, defaultValue: 0, }), hourlyLimit: Property.Number({ displayName: 'Hourly Limit', description: 'Select an MCP tool first', required: false, defaultValue: 0, }), activeHours: Property.Object({ displayName: 'Active Hours', description: 'Select an MCP tool first', required: false, defaultValue: {}, }), }; } const authConfig = auth as { mcpServerUrl?: string }; if (!authConfig?.mcpServerUrl) { return { instructions: Property.MarkDown({ value: `**Error:** MCP Server URL required in auth configuration.` }), delayType: Property.StaticDropdown({ displayName: 'Delay Type', description: 'Configure auth first', required: true, defaultValue: 'fixed', options: { options: [{ label: 'Configure Auth First', value: 'fixed' }] }, }), delayUnit: Property.StaticDropdown({ displayName: 'Delay Unit', description: 'Configure auth first', required: true, defaultValue: 'seconds', options: { options: [{ label: 'Configure Auth First', value: 'seconds' }] }, }), delayValue: Property.Number({ displayName: 'Delay Value', description: 'Configure auth first', required: false, defaultValue: 0, }), delayMin: Property.Number({ displayName: 'Minimum Delay', description: 'Configure auth first', required: false, defaultValue: 0, }), delayMax: Property.Number({ displayName: 'Maximum Delay', description: 'Configure auth first', required: false, defaultValue: 0, }), dailyLimit: Property.Number({ displayName: 'Daily Limit', description: 'Configure auth first', required: false, defaultValue: 0, }), hourlyLimit: Property.Number({ displayName: 'Hourly Limit', description: 'Configure auth first', required: false, defaultValue: 0, }), activeHours: Property.Object({ displayName: 'Active Hours', description: 'Configure auth first', required: false, defaultValue: {}, }), }; } // PropertyContext doesn't have store access - context.store is undefined const queueId = `mcp_${targetMCPTool}_queue`; return { instructions: Property.MarkDown({ value: `**⚙️ Configure queue:** ${queueId}\n\n*Note: Form shows defaults due to ActivePieces PropertyContext limitations. If this queue exists, your current settings will be preserved when you submit.*` }), delayType: Property.StaticDropdown({ displayName: 'Delay Type', description: 'Type of delay between queue items', required: true, defaultValue: defaults.delayType, options: { options: [ { label: 'Fixed Delay', value: 'fixed' }, { label: 'Random Delay', value: 'random' }, ], }, }), delayUnit: Property.StaticDropdown({ displayName: 'Delay Unit', description: 'Time unit for delays', required: true, defaultValue: defaults.delayUnit, options: { options: [ { label: 'Seconds', value: 'seconds' }, { label: 'Minutes', value: 'minutes' }, { label: 'Hours', value: 'hours' }, { label: 'Days', value: 'days' }, ], }, }), delayValue: Property.Number({ displayName: 'Delay Value', description: 'Fixed delay amount between items', required: false, defaultValue: defaults.delayValue, }), delayMin: Property.Number({ displayName: 'Minimum Delay', description: 'Minimum delay for random delays', required: false, defaultValue: defaults.delayMin, }), delayMax: Property.Number({ displayName: 'Maximum Delay', description: 'Maximum delay for random delays', required: false, defaultValue: defaults.delayMax, }), dailyLimit: Property.Number({ displayName: 'Daily Limit', description: 'Maximum items to process per day (0 = unlimited)', required: false, defaultValue: defaults.dailyLimit, }), hourlyLimit: Property.Number({ displayName: 'Hourly Limit', description: 'Maximum items to process per hour (0 = unlimited)', required: false, defaultValue: defaults.hourlyLimit, }), activeHours: Property.Object({ displayName: 'Active Hours', description: 'Only process queue during these hours (leave empty for 24×7)', required: false, defaultValue: defaults.activeHours, }), }; }, }), }, async run(context: ActionContext) { try { const { targetMCPTool, delayType, delayUnit, delayValue, delayMin, delayMax, dailyLimit, hourlyLimit, activeHours, } = context.propsValue; const authConfig = context.auth as { mcpServerUrl?: string }; const mcpServerUrl = authConfig.mcpServerUrl; /* -------- Validation -------- */ // Basic required fields if (!mcpServerUrl) { throw new Error('MCP Server URL is required in auth configuration'); } if (!targetMCPTool) { throw new Error('MCP Tool selection is required'); } // Delay validation if (delayType === 'fixed') { if (!delayValue || delayValue <= 0) { throw new Error(`Delay value must be > 0 for fixed delays. Got: ${delayValue}`); } } else if (delayType === 'random') { if (!delayMin || !delayMax || delayMin <= 0 || delayMax <= 0) { throw new Error(`Min/Max delays must be > 0 for random delays. Got min: ${delayMin}, max: ${delayMax}`); } if (delayMin >= delayMax) { throw new Error(`Minimum delay (${delayMin}) must be less than maximum delay (${delayMax})`); } } // Limits validation if (dailyLimit && dailyLimit < 0) { throw new Error(`Daily limit must be >= 0. Got: ${dailyLimit}`); } if (hourlyLimit && hourlyLimit < 0) { throw new Error(`Hourly limit must be >= 0. Got: ${hourlyLimit}`); } // Active hours validation if (activeHours) { const value = activeHours as any; if (!value || typeof value !== 'object') { throw new Error('Active hours must be an object'); } if (!value.timezone || typeof value.timezone !== 'string' || !value.timezone.includes('/')) { throw new Error(`timezone must be a valid IANA name, e.g., America/Toronto. Got: ${value.timezone}`); } // Handle schedule as either JSON string or object let schedule = value.schedule; if (typeof schedule === 'string') { try { schedule = JSON.parse(schedule); } catch (parseError: any) { throw new Error(`Schedule JSON string is invalid: ${parseError.message}`); } } if (!schedule || typeof schedule !== 'object') { throw new Error(`Active hours schedule must be an object or valid JSON string. Got type: ${typeof value.schedule}`); } const days = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']; const timeRegex = /^([01]\d|2[0-3]):[0-5]\d$/; // More precise regex: 00:00 to 23:59 for (const d of days) { const entry = schedule[d]; if (!entry) { throw new Error(`Missing schedule for ${d}`); } if (typeof entry.enabled !== 'boolean') { throw new Error(`${d}.enabled must be boolean. Got: ${typeof entry.enabled} (${entry.enabled})`); } if (entry.enabled) { if (!entry.start || !timeRegex.test(entry.start)) { throw new Error(`${d} start time must be HH:MM format (00:00-23:59). Got: "${entry.start}"`); } if (!entry.end || !timeRegex.test(entry.end)) { throw new Error(`${d} end time must be HH:MM format (00:00-23:59). Got: "${entry.end}"`); } // Convert to minutes for proper comparison const startMinutes = parseInt(entry.start.split(':')[0]) * 60 + parseInt(entry.start.split(':')[1]); const endMinutes = parseInt(entry.end.split(':')[0]) * 60 + parseInt(entry.end.split(':')[1]); if (startMinutes >= endMinutes) { throw new Error(`${d} start time (${entry.start}) must be before end time (${entry.end})`); } } } // Update the activeHours with parsed schedule for storage if (typeof value.schedule === 'string') { value.schedule = schedule; } } /* -------- Build / Update configuration -------- */ const queueId = `mcp_${targetMCPTool}_queue`; // Try to get existing configuration let existing: QueueConfiguration | null = null; try { existing = await QueueManager.getQueueConfiguration(context, queueId); } catch (error) { // Queue doesn't exist (normal for new queues) existing = null; } const now = Date.now(); const queueConfig: QueueConfiguration = { id: queueId, mcpToolName: targetMCPTool as string, delayType: delayType as any, delayUnit: delayUnit as any, delayValue, delayMin, delayMax, dailyLimit: dailyLimit || 0, hourlyLimit: hourlyLimit || 0, activeHours: activeHours || undefined, createdAt: existing?.createdAt || now, updatedAt: now, lastUsed: existing?.lastUsed, totalProcessed: existing?.totalProcessed || 0, }; // Save queue configuration try { await context.store.put(`queue_config_${queueId}`, queueConfig, StoreScope.PROJECT); } catch (storeError: any) { throw new Error(`Failed to save queue configuration: ${storeError.message}`); } // Create initial queue state if this is a new queue if (!existing) { const initState: QueueState = { queueId, lastReleaseTime: 0, itemCount: 0, currentExecutingItem: null, lastExecutedTime: 0, version: 0, }; try { await context.store.put(`queue_state_${queueId}`, initState, StoreScope.PROJECT); } catch (storeError: any) { throw new Error(`Failed to create initial queue state: ${storeError.message}`); } } // Get queue status let status; try { status = await QueueManager.getQueueStatus(context, queueId); } catch (statusError: any) { throw new Error(`Failed to get queue status: ${statusError.message}`); } return { success: true, queueId, queueLabel: `MCP: ${targetMCPTool}`, toolName: targetMCPTool, created: !existing, updated: !!existing, status, }; } catch (error: any) { throw new Error(`Queue creation failed: ${error.message}`); } }, });