/** * Automations Engine Library * Workflow automation with triggers, conditions, and actions */ import { getDatabase, schema } from "@/db"; import type { AutomationAction, AutomationCondition, AutomationTrigger, } from "@/db/schema/automations"; import { logger } from "@/lib/logger"; import { and, desc, eq } from "drizzle-orm"; import type { NodePgDatabase } from "drizzle-orm/node-postgres"; import { generateId } from "./utils"; const RETRYABLE_ACTIONS = new Set([ "notify_slack", "notify_discord", "notify_teams", "trigger_ai_review", "trigger_iac_hook", "trigger_cloud_deploy", ]); const DEFAULT_MAX_ATTEMPTS = 1; const DEFAULT_RETRYABLE_MAX_ATTEMPTS = 3; function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } function getMaxAttemptsForAction(action: AutomationAction): number { const requested = Number(action.params.maxAttempts); if (Number.isInteger(requested) && requested >= 1 && requested <= 5) { return requested; } return RETRYABLE_ACTIONS.has(action.type) ? DEFAULT_RETRYABLE_MAX_ATTEMPTS : DEFAULT_MAX_ATTEMPTS; } async function recordDeadLetterExecution(options: { ruleId: string; executionId: string; repositoryId?: string | null; triggeredById?: string; triggeredByType: string; triggerEvent: string; errorMessage: string; actionType?: string; }) { try { const db = getDatabase() as NodePgDatabase; await db.insert(schema.auditLogs).values({ id: generateId(), userId: null, repositoryId: options.repositoryId || null, organizationId: null, action: "automation.dead_letter", actorType: "system", actorId: "automation-engine", targetType: "automation_execution", targetId: options.executionId, targetName: options.ruleId, data: JSON.stringify({ ruleId: options.ruleId, executionId: options.executionId, triggeredById: options.triggeredById, triggeredByType: options.triggeredByType, triggerEvent: options.triggerEvent, actionType: options.actionType || null, errorMessage: options.errorMessage, deadLetteredAt: new Date().toISOString(), }), createdAt: new Date(), }); } catch (error) { logger.error( { error: error instanceof Error ? error.message : "Unknown error", executionId: options.executionId, }, "Failed to record automation dead-letter audit log", ); } } /** * Trigger an automation event * This is the main entry point for automation processing */ export async function triggerAutomation( repositoryId: string, event: AutomationTrigger, context: { pullRequestId?: string; issueId?: string; labelName?: string; userId?: string; metadata?: Record; }, ) { const db = getDatabase() as NodePgDatabase; const startTime = Date.now(); try { // Get all enabled rules for this trigger const rules = await db.query.automationRules.findMany({ where: and( eq(schema.automationRules.repositoryId, repositoryId), eq(schema.automationRules.trigger, event), eq(schema.automationRules.isEnabled, true), ), orderBy: [desc(schema.automationRules.priority)], }); if (rules.length === 0) { return { triggered: 0, executed: 0 }; } let executed = 0; for (const rule of rules) { const executionId = generateId(); try { // Check conditions const conditions = rule.conditions ? (JSON.parse(rule.conditions) as AutomationCondition[]) : []; const conditionsMatched = await evaluateConditions(conditions, context); // Log execution start await db.insert(schema.automationExecutions).values({ id: executionId, ruleId: rule.id, triggeredById: context.pullRequestId || context.issueId, triggeredByType: context.pullRequestId ? "pull_request" : context.issueId ? "issue" : "other", triggerEvent: event, conditionsMatched, status: "pending", createdAt: new Date(), }); if (!conditionsMatched) { await db .update(schema.automationExecutions) .set({ status: "skipped" }) .where(eq(schema.automationExecutions.id, executionId)); continue; } // Execute actions const actions = JSON.parse(rule.actions) as AutomationAction[]; const executedActions: Array<{ type: string; status: "success" | "failed"; attempts: number; error?: string; }> = []; for (const action of actions) { const maxAttempts = getMaxAttemptsForAction(action); let lastError: Error | null = null; let attempts = 0; for (let attempt = 1; attempt <= maxAttempts; attempt++) { attempts = attempt; try { await executeAction(action, context); lastError = null; break; } catch (error) { lastError = error instanceof Error ? error : new Error("Unknown action execution error"); const isLastAttempt = attempt === maxAttempts; if (!isLastAttempt) { // Small bounded backoff to improve external API reliability without stalling workers. await sleep(25 * attempt); } } } if (lastError) { executedActions.push({ type: action.type, status: "failed", attempts, error: lastError.message, }); await recordDeadLetterExecution({ ruleId: rule.id, executionId, repositoryId: rule.repositoryId, triggeredById: context.pullRequestId || context.issueId, triggeredByType: context.pullRequestId ? "pull_request" : context.issueId ? "issue" : "other", triggerEvent: event, errorMessage: lastError.message, actionType: action.type, }); throw lastError; } executedActions.push({ type: action.type, status: "success", attempts, }); } // Update execution record const durationMs = Date.now() - startTime; await db .update(schema.automationExecutions) .set({ status: "success", actionsExecuted: JSON.stringify(executedActions), durationMs, }) .where(eq(schema.automationExecutions.id, executionId)); // Update rule stats await db .update(schema.automationRules) .set({ runCount: (rule.runCount || 0) + 1, lastRunAt: new Date(), }) .where(eq(schema.automationRules.id, rule.id)); executed++; } catch (error) { logger.error("Automation execution failed", { ruleId: rule.id, error: error instanceof Error ? error.message : "Unknown error", }); await db .update(schema.automationExecutions) .set({ status: "failed", errorMessage: error instanceof Error ? error.message : "Unknown error", durationMs: Date.now() - startTime, }) .where(eq(schema.automationExecutions.id, executionId)); await recordDeadLetterExecution({ ruleId: rule.id, executionId, repositoryId: rule.repositoryId, triggeredById: context.pullRequestId || context.issueId, triggeredByType: context.pullRequestId ? "pull_request" : context.issueId ? "issue" : "other", triggerEvent: event, errorMessage: error instanceof Error ? error.message : "Unknown error", }); } } return { triggered: rules.length, executed }; } catch (error) { logger.error("Failed to trigger automation", { repositoryId, event, error: error instanceof Error ? error.message : "Unknown error", }); throw error; } } /** * Evaluate conditions against the context */ async function evaluateConditions( conditions: AutomationCondition[], context: Record, ): Promise { if (conditions.length === 0) return true; for (const condition of conditions) { const value = getNestedValue(context, condition.field); const matched = evaluateCondition( condition.operator, value, condition.value, ); if (!matched) return false; } return true; } function getNestedValue(obj: Record, path: string): unknown { return path .split(".") .reduce( (acc, key) => acc && typeof acc === "object" ? (acc as Record)[key] : undefined, obj as unknown, ); } function evaluateCondition( operator: string, actual: unknown, expected: unknown, ): boolean { switch (operator) { case "equals": return actual === expected; case "not_equals": return actual !== expected; case "contains": if (Array.isArray(actual)) { return actual.includes(expected); } if (typeof actual === "string" && typeof expected === "string") { return actual.includes(expected); } return false; case "not_contains": if (Array.isArray(actual)) { return !actual.includes(expected); } if (typeof actual === "string" && typeof expected === "string") { return !actual.includes(expected); } return true; case "matches": if (typeof actual === "string" && typeof expected === "string") { try { // Safety: limit pattern length and apply timeout to prevent ReDoS if (expected.length > 200) return false; const regex = new RegExp(expected); return regex.test(actual.slice(0, 10_000)); } catch { // Invalid regex pattern provided by user return false; } } return false; case "greater_than": return ( typeof actual === "number" && typeof expected === "number" && actual > expected ); case "less_than": return ( typeof actual === "number" && typeof expected === "number" && actual < expected ); default: return false; } } /** * Execute an automation action */ async function executeAction( action: AutomationAction, context: { pullRequestId?: string; issueId?: string; userId?: string; metadata?: Record; }, ) { const db = getDatabase() as NodePgDatabase; switch (action.type) { case "add_label": { if (context.pullRequestId && action.params.label) { // Add label to PR await db .insert(schema.pullRequestLabels) .values({ id: generateId(), pullRequestId: context.pullRequestId, labelId: action.params.label as string, }) .onConflictDoNothing(); } break; } case "remove_label": { if (context.pullRequestId && action.params.label) { await db .delete(schema.pullRequestLabels) .where( and( eq(schema.pullRequestLabels.pullRequestId, context.pullRequestId), eq( schema.pullRequestLabels.labelId, action.params.label as string, ), ), ); } break; } case "assign_reviewer": { if (context.pullRequestId && action.params.assignee) { const assignee = action.params.assignee as string; // Handle @codeowners - assign reviewers based on CODEOWNERS file if (assignee === "@codeowners") { try { // Get PR details with changed files const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), with: { repository: true }, }); if (pr && pr.repository) { // Dynamically import to avoid circular deps const { resolveRepoPath } = await import("./git-storage"); const { getFileContent } = await import("./git"); const { CODEOWNERS_PATHS, getSuggestedReviewers } = await import("./codeowners"); const { compareBranches } = await import("./git"); const repoPath = await resolveRepoPath(pr.repository.diskPath); // Try to find CODEOWNERS file let codeOwnersContent: string | null = null; for (const path of CODEOWNERS_PATHS) { try { const result = await getFileContent( repoPath, path, pr.baseBranch, ); if (result && !result.isBinary) { codeOwnersContent = result.content; break; } } catch { // File doesn't exist at this path } } if (codeOwnersContent) { // Get changed files const { diffs } = await compareBranches( repoPath, pr.baseBranch, pr.headBranch, ); const changedFiles = diffs.map((d) => d.file); // Get PR author username for exclusion const prAuthor = await db.query.users.findFirst({ where: eq(schema.users.id, pr.authorId), }); // Get suggested reviewers const { owners } = await getSuggestedReviewers( codeOwnersContent, changedFiles, prAuthor ? [prAuthor.username] : [], ); // Add each owner as reviewer for (const owner of owners) { const user = await db.query.users.findFirst({ where: eq(schema.users.username, owner), }); if (user) { await db .insert(schema.pullRequestReviewers) .values({ id: generateId(), pullRequestId: context.pullRequestId, userId: user.id, requestedAt: new Date(), }) .onConflictDoNothing(); } } } } } catch (error) { logger.warn({ error }, "Failed to assign CODEOWNERS reviewers"); } break; } // Find user by username const user = await db.query.users.findFirst({ where: eq(schema.users.username, assignee.replace("@", "")), }); if (user) { await db .insert(schema.pullRequestReviewers) .values({ id: generateId(), pullRequestId: context.pullRequestId, userId: user.id, requestedAt: new Date(), }) .onConflictDoNothing(); } } break; } case "add_comment": { if (context.pullRequestId && action.params.body) { await db.insert(schema.pullRequestComments).values({ id: generateId(), pullRequestId: context.pullRequestId, authorId: context.userId || "system", body: action.params.body as string, createdAt: new Date(), updatedAt: new Date(), }); } else if (context.issueId && action.params.body) { await db.insert(schema.issueComments).values({ id: generateId(), issueId: context.issueId, authorId: context.userId || "system", body: action.params.body as string, createdAt: new Date(), updatedAt: new Date(), }); } break; } case "update_issue_status": { if (context.issueId && action.params.statusId) { // If statusId is a special keyword 'closed' or 'open', update the basic state if (action.params.statusId === "closed" || action.params.statusId === "open") { await db .update(schema.issues) .set({ state: action.params.statusId, statusId: null, // clear custom status if transitioning to basic state updatedAt: new Date() }) .where(eq(schema.issues.id, context.issueId)); } else { // Otherwise, it's a custom status ID await db .update(schema.issues) .set({ statusId: action.params.statusId as string, state: "open", // assuming custom statuses map to 'open' state generally unless specified updatedAt: new Date() }) .where(eq(schema.issues.id, context.issueId)); } } break; } case "add_to_merge_queue": { if (context.pullRequestId) { const { addToMergeQueue } = await import("./merge-queue"); const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), }); if (pr) { await addToMergeQueue({ repositoryId: pr.repositoryId, pullRequestId: context.pullRequestId, addedById: context.userId || "system", mergeMethod: (action.params.method as "merge" | "squash" | "rebase") || "squash", }); } } break; } case "trigger_ai_review": { if (context.pullRequestId) { const { triggerAIReview } = await import("./ai-review"); await triggerAIReview( context.pullRequestId, context.userId || "system", { provider: "anthropic" as const, model: "claude-3-sonnet" as const, }, ); } break; } case "notify_slack": { if (context.pullRequestId && action.params.channel) { const { notifyPrEvent } = await import("./slack-notifications"); const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), with: { repository: true, author: true }, }); if (pr && pr.repository) { // Would need org ID detection // await notifyPrEvent(...) } } break; } case "notify_discord": { if (context.pullRequestId && action.params.webhookUrl) { const { sendDiscordMessage } = await import("./integrations/discord"); const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), with: { repository: true, author: true }, }); if (pr && pr.repository) { await sendDiscordMessage(action.params.webhookUrl as string, { content: `**${pr.title}** (#${pr.number}) in ${pr.repository.name}\nState: ${pr.state}`, username: "OpenCodeHub", }); } } break; } case "notify_teams": { if (context.pullRequestId && action.params.webhookUrl) { const { sendTeamsMessage } = await import("./integrations/teams"); const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), with: { repository: true, author: true }, }); if (pr && pr.repository) { await sendTeamsMessage(action.params.webhookUrl as string, { "@type": "MessageCard", "@context": "http://schema.org/extensions", themeColor: "0078d4", summary: `PR Update: ${pr.title}`, sections: [ { text: `**${pr.title}** (#${pr.number}) in ${pr.repository.name}\nStatus: ${pr.state}`, }, ], }); } } break; } case "close_pr": { if (context.pullRequestId) { await db .update(schema.pullRequests) .set({ state: "closed", closedAt: new Date(), updatedAt: new Date(), }) .where(eq(schema.pullRequests.id, context.pullRequestId)); } break; } case "assign_user": { if (action.params.user) { const username = action.params.user as string; // Find user const user = await db.query.users.findFirst({ where: eq(schema.users.username, username.replace("@", "")), }); if (user) { const { generateId } = await import("./utils"); if (context.pullRequestId) { await db .insert(schema.pullRequestAssignees) .values({ id: generateId(), pullRequestId: context.pullRequestId, userId: user.id, }) .onConflictDoNothing(); } else if (context.issueId) { await db .update(schema.issues) .set({ assigneeId: user.id, updatedAt: new Date(), }) .where(eq(schema.issues.id, context.issueId)); } } } break; } case "request_changes": { if (context.pullRequestId && action.params.body) { const { generateId } = await import("./utils"); // Create a review await db.insert(schema.pullRequestReviews).values({ id: generateId(), pullRequestId: context.pullRequestId, reviewerId: context.userId || "system", // Or a bot user info if available state: "changes_requested", body: action.params.body as string, submittedAt: new Date(), }); } break; } case "trigger_iac_hook": { if (context.pullRequestId) { const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), }); if (pr) { const actionType = (action.params.action as "plan" | "apply") || "plan"; const { triggerIaCHooks } = await import("./iac-hooks"); await triggerIaCHooks({ repositoryId: pr.repositoryId, action: actionType, runId: typeof action.params.runId === "string" ? action.params.runId : undefined, message: typeof action.params.message === "string" ? action.params.message : undefined, configId: typeof action.params.configId === "string" ? action.params.configId : undefined, }); } } break; } case "trigger_cloud_deploy": { if (context.pullRequestId) { const pr = await db.query.pullRequests.findFirst({ where: eq(schema.pullRequests.id, context.pullRequestId), }); if (pr) { const { triggerCloudDeploy } = await import("./cloud-hooks"); await triggerCloudDeploy({ repositoryId: pr.repositoryId, configId: typeof action.params.configId === "string" ? action.params.configId : undefined, imageTag: typeof action.params.imageTag === "string" ? action.params.imageTag : undefined, namespace: typeof action.params.namespace === "string" ? action.params.namespace : undefined, deploymentName: typeof action.params.deploymentName === "string" ? action.params.deploymentName : undefined, serviceName: typeof action.params.serviceName === "string" ? action.params.serviceName : undefined, appName: typeof action.params.appName === "string" ? action.params.appName : undefined, taskDefinition: typeof action.params.taskDefinition === "string" ? action.params.taskDefinition : undefined, clusterName: typeof action.params.clusterName === "string" ? action.params.clusterName : undefined, }); } } break; } default: logger.warn("Unknown automation action", { type: action.type }); } } /** * Create an automation rule */ export async function createAutomationRule(options: { repositoryId?: string; organizationId?: string; createdById: string; name: string; description?: string; trigger: AutomationTrigger; conditions?: AutomationCondition[]; actions: AutomationAction[]; priority?: number; }) { const db = getDatabase() as NodePgDatabase; const rule = { id: generateId(), repositoryId: options.repositoryId, organizationId: options.organizationId, createdById: options.createdById, name: options.name, description: options.description, trigger: options.trigger, conditions: options.conditions ? JSON.stringify(options.conditions) : null, actions: JSON.stringify(options.actions), isEnabled: true, priority: options.priority || 0, createdAt: new Date(), updatedAt: new Date(), }; await db.insert(schema.automationRules).values(rule); return rule; } /** * Get automation rules for a repository */ export async function getAutomationRules(repositoryId: string) { const db = getDatabase() as NodePgDatabase; return db.query.automationRules.findMany({ where: eq(schema.automationRules.repositoryId, repositoryId), orderBy: [ desc(schema.automationRules.priority), desc(schema.automationRules.createdAt), ], }); } /** * Toggle automation rule enabled state */ export async function toggleAutomationRule(ruleId: string, enabled: boolean) { const db = getDatabase() as NodePgDatabase; await db .update(schema.automationRules) .set({ isEnabled: enabled, updatedAt: new Date(), }) .where(eq(schema.automationRules.id, ruleId)); } /** * Delete an automation rule */ export async function deleteAutomationRule(ruleId: string) { const db = getDatabase() as NodePgDatabase; await db .delete(schema.automationRules) .where(eq(schema.automationRules.id, ruleId)); }