Fix subscription check for non-manual triggers by resolving userPlan from org billing

All trigger paths (scheduled, queue, email, Discord, Telegram, WhatsApp) were missing
userPlan in execution params, causing subscription-gated nodes to always fail. Now
derives the plan from organization billing info using the same logic as the billing route.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: bchapuis

apps/api/src/db/queries.ts

@@ -2137,6 +2137,7 @@ export async function getOrganizationBillingInfo(
   | {
       computeCredits: number;
       subscriptionStatus: string | null;
+      currentPeriodEnd: Date | null;
       overageLimit: number | null;
     }
   | undefined
@@ -2145,6 +2146,7 @@ export async function getOrganizationBillingInfo(
     .select({
       computeCredits: organizations.computeCredits,
       subscriptionStatus: organizations.subscriptionStatus,
+      currentPeriodEnd: organizations.currentPeriodEnd,
       overageLimit: organizations.overageLimit,
     })
     .from(organizations)
@@ -2153,6 +2155,22 @@ export async function getOrganizationBillingInfo(
   return organization;
 }
 
+/**
+ * Derive user plan from organization billing info.
+ * Pro if has active subscription OR canceled but still in billing period.
+ */
+export function resolveUserPlan(billingInfo: {
+  subscriptionStatus: string | null;
+  currentPeriodEnd: Date | null;
+}): string {
+  const hasProAccess =
+    billingInfo.subscriptionStatus === "active" ||
+    (billingInfo.subscriptionStatus === "canceled" &&
+      billingInfo.currentPeriodEnd !== null &&
+      billingInfo.currentPeriodEnd > new Date());
+  return hasProAccess ? "pro" : "trial";
+}
+
 /**
  * Create a new secret for an organization
  *

apps/api/src/email.ts

@@ -1,7 +1,11 @@
 import type { Workflow } from "@dafthunk/types";
 
 import type { Bindings } from "./context";
-import { createDatabase, getOrganizationComputeCredits } from "./db";
+import {
+  createDatabase,
+  getOrganizationBillingInfo,
+  resolveUserPlan,
+} from "./db";
 import { getAgentByName } from "./durable-objects/agent-utils";
 import { createWorkerRuntime } from "./runtime/cloudflare-worker-runtime";
 import { WorkflowStore } from "./stores/workflow-store";
@@ -164,20 +168,18 @@ async function triggerWorkflowForEmail({
     return;
   }
 
-  // Get organization compute credits
-  const computeCredits = await getOrganizationComputeCredits(
-    db,
-    organizationId
-  );
-  if (computeCredits === undefined) {
+  // Get organization billing info
+  const billingInfo = await getOrganizationBillingInfo(db, organizationId);
+  if (billingInfo === undefined) {
     console.error("Organization not found");
     return;
   }
 
   const executionParams = {
     userId: "email_trigger",
     organizationId,
-    computeCredits,
+    computeCredits: billingInfo.computeCredits,
+    userPlan: resolveUserPlan(billingInfo),
     workflow: {
       id: workflow.id,
       name: workflow.name,

apps/api/src/queue.ts

@@ -2,8 +2,9 @@ import type { QueueMessage, Workflow } from "@dafthunk/types";
 import type { Bindings } from "./context";
 import { createDatabase } from "./db";
 import {
-  getOrganizationComputeCredits,
+  getOrganizationBillingInfo,
   getQueueTriggersByQueue,
+  resolveUserPlan,
 } from "./db/queries";
 import { getAgentByName } from "./durable-objects/agent-utils";
 import { createWorkerRuntime } from "./runtime/cloudflare-worker-runtime";
@@ -27,20 +28,21 @@ async function executeWorkflow(
   );
 
   try {
-    // Get organization compute credits
-    const computeCredits = await getOrganizationComputeCredits(
+    // Get organization billing info
+    const billingInfo = await getOrganizationBillingInfo(
       db,
       workflowInfo.organizationId
     );
-    if (computeCredits === undefined) {
+    if (billingInfo === undefined) {
       console.error("Organization not found");
       return;
     }
 
     const executionParams = {
       userId: "queue_trigger",
       organizationId: workflowInfo.organizationId,
-      computeCredits,
+      computeCredits: billingInfo.computeCredits,
+      userPlan: resolveUserPlan(billingInfo),
       workflow: {
         id: workflowInfo.id,
         name: workflowData.name,

apps/api/src/routes/discord-webhook.ts

@@ -7,7 +7,8 @@ import {
   getDiscordBot,
   getDiscordBotById,
   getDiscordTriggersByBot,
-  getOrganizationComputeCredits,
+  getOrganizationBillingInfo,
+  resolveUserPlan,
 } from "../db";
 import { getAgentByName } from "../durable-objects/agent-utils";
 import { createWorkerRuntime } from "../runtime/cloudflare-worker-runtime";
@@ -288,19 +289,17 @@ async function executeWorkflow(
     return;
   }
 
-  const computeCredits = await getOrganizationComputeCredits(
-    db,
-    organizationId
-  );
-  if (computeCredits === undefined) {
+  const billingInfo = await getOrganizationBillingInfo(db, organizationId);
+  if (billingInfo === undefined) {
     console.error("[DiscordWebhook] Organization not found");
     return;
   }
 
   const executionParams = {
     userId: "discord_trigger",
     organizationId,
-    computeCredits,
+    computeCredits: billingInfo.computeCredits,
+    userPlan: resolveUserPlan(billingInfo),
     workflow: {
       id: workflow.id,
       name: workflow.name,

apps/api/src/routes/telegram-webhook.ts

@@ -4,10 +4,11 @@ import { Hono } from "hono";
 import type { ApiContext } from "../context";
 import {
   createDatabase,
-  getOrganizationComputeCredits,
+  getOrganizationBillingInfo,
   getTelegramBot,
   getTelegramSecretTokenByBot,
   getTelegramTriggersByBot,
+  resolveUserPlan,
 } from "../db";
 import { getAgentByName } from "../durable-objects/agent-utils";
 import { createWorkerRuntime } from "../runtime/cloudflare-worker-runtime";
@@ -205,19 +206,17 @@ async function executeWorkflow(
     return;
   }
 
-  const computeCredits = await getOrganizationComputeCredits(
-    db,
-    organizationId
-  );
-  if (computeCredits === undefined) {
+  const billingInfo = await getOrganizationBillingInfo(db, organizationId);
+  if (billingInfo === undefined) {
     console.error("[TelegramWebhook] Organization not found");
     return;
   }
 
   const executionParams = {
     userId: "telegram_trigger",
     organizationId,
-    computeCredits,
+    computeCredits: billingInfo.computeCredits,
+    userPlan: resolveUserPlan(billingInfo),
     workflow: {
       id: workflow.id,
       name: workflow.name,

apps/api/src/routes/whatsapp-webhook.ts

@@ -4,10 +4,11 @@ import { Hono } from "hono";
 import type { ApiContext } from "../context";
 import {
   createDatabase,
-  getOrganizationComputeCredits,
+  getOrganizationBillingInfo,
   getWhatsAppAccount,
   getWhatsAppTriggersByAccount,
   getWhatsAppVerifyTokenByAccount,
+  resolveUserPlan,
 } from "../db";
 import { getAgentByName } from "../durable-objects/agent-utils";
 import { createWorkerRuntime } from "../runtime/cloudflare-worker-runtime";
@@ -311,19 +312,17 @@ async function executeWorkflow(
     return;
   }
 
-  const computeCredits = await getOrganizationComputeCredits(
-    db,
-    organizationId
-  );
-  if (computeCredits === undefined) {
+  const billingInfo = await getOrganizationBillingInfo(db, organizationId);
+  if (billingInfo === undefined) {
     console.error("[WhatsAppWebhook] Organization not found");
     return;
   }
 
   const executionParams = {
     userId: "whatsapp_trigger",
     organizationId,
-    computeCredits,
+    computeCredits: billingInfo.computeCredits,
+    userPlan: resolveUserPlan(billingInfo),
     workflow: {
       id: workflow.id,
       name: workflow.name,

apps/api/src/scheduled.ts

@@ -4,7 +4,8 @@ import type { Bindings } from "./context";
 import {
   createDatabase,
   getActiveScheduledTriggers,
-  getOrganizationComputeCredits,
+  getOrganizationBillingInfo,
+  resolveUserPlan,
 } from "./db";
 import { getAgentByName } from "./durable-objects/agent-utils";
 import { createWorkerRuntime } from "./runtime/cloudflare-worker-runtime";
@@ -62,17 +63,18 @@ export async function handleScheduledEvent(
       }
       const workflowData = workflowWithData.data;
 
-      // Get organization compute credits
-      const computeCredits = await getOrganizationComputeCredits(
+      // Get organization billing info
+      const billingInfo = await getOrganizationBillingInfo(
         db,
         workflow.organizationId
       );
-      if (computeCredits === undefined) continue;
+      if (billingInfo === undefined) continue;
 
       const executionParams = {
         userId: "scheduled_trigger",
         organizationId: workflow.organizationId,
-        computeCredits,
+        computeCredits: billingInfo.computeCredits,
+        userPlan: resolveUserPlan(billingInfo),
         workflow: {
           id: workflow.id,
           name: workflow.name,

apps/api/src/services/execution-manager.ts

@@ -7,11 +7,13 @@
 
 import type { BlobParameter } from "@dafthunk/runtime";
 import type { WorkflowExecution, WorkflowState } from "@dafthunk/types";
-import { eq } from "drizzle-orm";
 import type { Bindings } from "../context";
 import { createDatabase } from "../db/index";
-import { getOrganization, getOrganizationBillingInfo } from "../db/queries";
-import { users } from "../db/schema";
+import {
+  getOrganization,
+  getOrganizationBillingInfo,
+  resolveUserPlan,
+} from "../db/queries";
 import {
   WorkflowExecutor,
   type WorkflowExecutorParameters,
@@ -55,16 +57,8 @@ export class ExecutionManager {
     }
     const { computeCredits, subscriptionStatus, overageLimit } = billingInfo;
 
-    // Get user's plan if not provided (e.g., for WebSocket-based execution)
-    let resolvedUserPlan = userPlan;
-    if (!resolvedUserPlan) {
-      const [user] = await db
-        .select({ plan: users.plan })
-        .from(users)
-        .where(eq(users.id, userId))
-        .limit(1);
-      resolvedUserPlan = user?.plan;
-    }
+    // Resolve user plan from org billing info if not provided
+    const resolvedUserPlan = userPlan || resolveUserPlan(billingInfo);
 
     validateWorkflowForExecution(state);