diff --git a/src/infra/workflow-approvals.text-command.test.ts b/src/infra/workflow-approvals.text-command.test.ts index 1a2c28b3..7af68ddb 100644 --- a/src/infra/workflow-approvals.text-command.test.ts +++ b/src/infra/workflow-approvals.text-command.test.ts @@ -1,6 +1,7 @@ -import { describe, expect, it } from "vitest"; -import type { PendingWorkflowApprovalRow } from "./workflow-approvals.js"; -import { selectApprovalForTextCommand } from "./workflow-approvals.js"; +import type { Sql } from "postgres"; +import { describe, expect, it, vi } from "vitest"; +import type { ApprovalTextCommandDeps, PendingWorkflowApprovalRow } from "./workflow-approvals.js"; +import { runApprovalTextCommand, selectApprovalForTextCommand } from "./workflow-approvals.js"; function row(id: string, runId: string, name = "WF"): PendingWorkflowApprovalRow { return { @@ -57,3 +58,130 @@ describe("selectApprovalForTextCommand", () => { expect(result.kind).toBe("ambiguous"); }); }); + +const FAKE_SQL = {} as unknown as Sql; + +function makeDeps( + pending: PendingWorkflowApprovalRow[], + overrides: Partial = {}, +): ApprovalTextCommandDeps { + return { + listPending: vi.fn(async () => pending), + // mirrors resolveDurableWorkflowApprovalById: returns the row when it was + // still pending (UPDATE ... SET status=approved/denied matched a row). + resolveById: vi.fn(async (_sql, input) => ({ + run_id: pending.find((p) => p.id === input.approvalId)?.run_id ?? "run-x", + node_id: "approval", + })), + hasPendingApproval: vi.fn(() => false), + resolveInMemory: vi.fn(), + resumeRun: vi.fn(), + ...overrides, + }; +} + +describe("runApprovalTextCommand (resolve + resume wiring)", () => { + it("approve with exactly one pending → resolves (approved=true) and resumes the run", async () => { + const pending = [row("approval-r1-approval", "run-1111", "Daily SaaS Radar")]; + const deps = makeDeps(pending); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: true, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("resolved"); + expect(result.runId).toBe("run-1111"); + expect(result.reply).toContain("✅ Approved"); + // The DB transition: resolveById is the UPDATE ... SET status='approved'. + expect(deps.resolveById).toHaveBeenCalledWith( + FAKE_SQL, + expect.objectContaining({ approvalId: "approval-r1-approval", approved: true }), + ); + // Durable run (not in-memory) → resume path. + expect(deps.resumeRun).toHaveBeenCalledWith("run-1111", "approval"); + expect(deps.resolveInMemory).not.toHaveBeenCalled(); + }); + + it("deny is symmetric → resolves with approved=false", async () => { + const pending = [row("approval-r1-approval", "run-1111")]; + const deps = makeDeps(pending); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: false, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("resolved"); + expect(result.reply).toContain("❌ Denied"); + expect(deps.resolveById).toHaveBeenCalledWith( + FAKE_SQL, + expect.objectContaining({ approved: false }), + ); + expect(deps.resumeRun).toHaveBeenCalledTimes(1); + }); + + it("in-process pending run → resolves in memory, no durable resume", async () => { + const pending = [row("approval-r1-approval", "run-1111")]; + const deps = makeDeps(pending, { hasPendingApproval: vi.fn(() => true) }); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: true, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("resolved"); + expect(deps.resolveInMemory).toHaveBeenCalledWith("run-1111", "approval", true, undefined); + expect(deps.resumeRun).not.toHaveBeenCalled(); + }); + + it("multiple pending → lists choices, resolves nothing", async () => { + const pending = [ + row("approval-a-approval", "aaaa1111"), + row("approval-b-approval", "bbbb2222"), + ]; + const deps = makeDeps(pending); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: true, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("ambiguous"); + expect(result.reply).toContain("2 approvals are pending"); + expect(deps.resolveById).not.toHaveBeenCalled(); + expect(deps.resumeRun).not.toHaveBeenCalled(); + }); + + it("zero pending → no-op reply, resolves nothing", async () => { + const deps = makeDeps([]); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: true, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("none"); + expect(result.reply).toContain("No workflow approvals are pending"); + expect(deps.resolveById).not.toHaveBeenCalled(); + }); + + it("already-resolved between list and update → 'gone', no resume", async () => { + const pending = [row("approval-r1-approval", "run-1111")]; + const deps = makeDeps(pending, { resolveById: vi.fn(async () => undefined) }); + const result = await runApprovalTextCommand({ + sql: FAKE_SQL, + approved: true, + operatorLabel: "@jason", + deps, + }); + + expect(result.outcome).toBe("gone"); + expect(deps.resumeRun).not.toHaveBeenCalled(); + }); +}); diff --git a/src/infra/workflow-approvals.ts b/src/infra/workflow-approvals.ts index 7ddd143c..a1dfc42a 100644 --- a/src/infra/workflow-approvals.ts +++ b/src/infra/workflow-approvals.ts @@ -270,6 +270,92 @@ export function selectApprovalForTextCommand( return { kind: "ambiguous", rows: pending }; } +export interface ApprovalTextCommandDeps { + listPending: (sql: Sql) => Promise; + resolveById: ( + sql: Sql, + input: { approvalId: string; approved: boolean; reason?: string; approvedBy?: string }, + ) => Promise<{ run_id?: unknown; node_id?: unknown } | undefined>; + /** True when the run is still pending in the current process (in-memory). */ + hasPendingApproval: (runId: string, nodeId: string) => boolean; + /** Resolve an in-process pending approval (no durable resume needed). */ + resolveInMemory: (runId: string, nodeId: string, approved: boolean, reason?: string) => void; + /** Resume a durable (e.g. cron / post-restart) run after resolution. */ + resumeRun: (runId: string, nodeId: string) => void; +} + +export interface ApprovalTextCommandResult { + outcome: "none" | "not-found" | "ambiguous" | "gone" | "resolved"; + reply: string; + approved: boolean; + runId?: string; +} + +// Orchestrates the Telegram text-reply approval surface end to end: list pending +// → select → resolve (durable) → resume or in-memory-resolve → reply text. All +// I/O is injected so the branching (including the resolve+resume wiring) is +// unit-testable without standing up the bot or a live gateway. The Telegram +// handler is a thin caller that just sends `result.reply`. +export async function runApprovalTextCommand(opts: { + sql: Sql; + approved: boolean; + token?: string; + operatorLabel: string; + deps: ApprovalTextCommandDeps; +}): Promise { + const { sql, approved, token, operatorLabel, deps } = opts; + const pending = await deps.listPending(sql); + const selection = selectApprovalForTextCommand(pending, token); + + if (selection.kind === "none") { + return { outcome: "none", approved, reply: "No workflow approvals are pending right now." }; + } + if (selection.kind === "not-found") { + return { + outcome: "not-found", + approved, + reply: `No pending approval matches "${selection.token}".`, + }; + } + if (selection.kind === "ambiguous") { + const verb = approved ? "approve" : "deny"; + const list = selection.rows + .map((r) => `• ${r.workflow_name ?? r.run_id} — reply "${verb} ${r.run_id.slice(0, 8)}"`) + .join("\n"); + return { + outcome: "ambiguous", + approved, + reply: `${selection.rows.length} approvals are pending — which one?\n${list}`, + }; + } + + const row = selection.row; + const resolved = await deps.resolveById(sql, { + approvalId: row.id, + approved, + approvedBy: operatorLabel, + reason: approved ? undefined : "Denied via Telegram text reply", + }); + if (!resolved) { + return { outcome: "gone", approved, reply: "That approval is no longer pending." }; + } + const runId = String(resolved.run_id); + const nodeId = String(resolved.node_id); + if (deps.hasPendingApproval(runId, nodeId)) { + deps.resolveInMemory(runId, nodeId, approved, approved ? undefined : "Denied via Telegram"); + } else { + deps.resumeRun(runId, nodeId); + } + const icon = approved ? "✅" : "❌"; + const verb = approved ? "Approved" : "Denied"; + return { + outcome: "resolved", + approved, + runId, + reply: `${icon} ${verb} by ${operatorLabel}.\nWorkflow: ${row.workflow_name ?? runId}\nRun: ${runId}`, + }; +} + // Resolves a workflow approval by its primary-key id. Used by callback-driven // surfaces (Telegram inline buttons) where the caller only carries the // approval id, not run/node. Atomically transitions pending → approved/denied diff --git a/src/telegram/bot-handlers.ts b/src/telegram/bot-handlers.ts index c1099ba9..00da6a60 100644 --- a/src/telegram/bot-handlers.ts +++ b/src/telegram/bot-handlers.ts @@ -844,72 +844,38 @@ export const registerTelegramHandlers = ({ const { getWorkflowSql } = await import("../infra/workflow-pg-client.js"); const { listPendingWorkflowApprovals, - selectApprovalForTextCommand, resolveDurableWorkflowApprovalById, + runApprovalTextCommand, } = await import("../infra/workflow-approvals.js"); const { hasPendingApproval, resolveApproval } = await import("../infra/workflow-runner.js"); const { resumeWorkflowRunAfterApproval } = await import("../infra/workflow-execution-service.js"); const sql = await getWorkflowSql(); - const pending = await listPendingWorkflowApprovals(sql); - const selection = selectApprovalForTextCommand(pending, token); - - if (selection.kind === "none") { - await reply("No workflow approvals are pending right now."); - return; - } - if (selection.kind === "not-found") { - await reply(`No pending approval matches "${selection.token}".`); - return; - } - if (selection.kind === "ambiguous") { - const verb = approved ? "approve" : "deny"; - const list = selection.rows - .map( - (r) => - `• ${r.workflow_name ?? r.run_id} — reply "${verb} ${r.run_id.slice(0, 8)}"`, - ) - .join("\n"); - await reply(`${selection.rows.length} approvals are pending — which one?\n${list}`); - return; - } - - const row = selection.row; - const resolved = await resolveDurableWorkflowApprovalById(sql, { - approvalId: row.id, + const result = await runApprovalTextCommand({ + sql, approved, - approvedBy: operatorLabel, - reason: approved ? undefined : "Denied via Telegram text reply", + token, + operatorLabel, + deps: { + listPending: listPendingWorkflowApprovals, + resolveById: resolveDurableWorkflowApprovalById, + hasPendingApproval, + resolveInMemory: (runId, nodeId, isApproved, reason) => + resolveApproval(runId, nodeId, isApproved, reason), + resumeRun: (runId, nodeId) => { + void resumeWorkflowRunAfterApproval({ + sql, + runId, + nodeId, + triggerSource: "telegram:approval_text", + }).catch((err: unknown) => { + runtime.error?.(danger(`workflow approval resume failed: ${String(err)}`)); + }); + }, + }, }); - if (!resolved) { - await reply("That approval is no longer pending."); - return; - } - const runId = String(resolved.run_id); - const nodeId = String(resolved.node_id); - if (hasPendingApproval(runId, nodeId)) { - resolveApproval( - runId, - nodeId, - approved, - approved ? undefined : "Denied via Telegram", - ); - } else { - void resumeWorkflowRunAfterApproval({ - sql, - runId, - nodeId, - triggerSource: "telegram:approval_text", - }).catch((err: unknown) => { - runtime.error?.(danger(`workflow approval resume failed: ${String(err)}`)); - }); - } - const icon = approved ? "✅" : "❌"; - const verb = approved ? "Approved" : "Denied"; - await reply( - `${icon} ${verb} by ${operatorLabel}.\nWorkflow: ${row.workflow_name ?? runId}\nRun: ${runId}`, - ); + await reply(result.reply); } catch (err) { runtime.error?.(danger(`workflow approval text command failed: ${String(err)}`)); }