outbox service

Author: icecrasher321

apps/sim/app/api/billing/route.ts

@@ -45,7 +45,20 @@ export async function GET(request: NextRequest) {
     let billingData
 
     if (context === 'user') {
-      // Get user billing and billing blocked status in parallel
+      if (contextId) {
+        const membership = await db
+          .select({ role: member.role })
+          .from(member)
+          .where(and(eq(member.organizationId, contextId), eq(member.userId, session.user.id)))
+          .limit(1)
+        if (membership.length === 0) {
+          return NextResponse.json(
+            { error: 'Access denied - not a member of this organization' },
+            { status: 403 }
+          )
+        }
+      }
+
       const [billingResult, billingStatus] = await Promise.all([
         getSimplifiedBillingSummary(session.user.id, contextId || undefined),
         getEffectiveBillingStatus(session.user.id),

apps/sim/app/api/organizations/[id]/invitations/[invitationId]/route.ts

@@ -23,8 +23,9 @@ import { getSession } from '@/lib/auth'
 import { hasAccessControlAccess } from '@/lib/billing'
 import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
 import { isPaid, sqlIsPro } from '@/lib/billing/plan-helpers'
-import { requireStripeClient } from '@/lib/billing/stripe-client'
 import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
+import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
+import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { generateId } from '@/lib/core/utils/uuid'
 import { syncWorkspaceEnvCredentials } from '@/lib/credentials/environment'
@@ -328,8 +329,6 @@ export async function PUT(
       }
     }
 
-    let personalProToCancel: any = null
-
     await db.transaction(async (tx) => {
       await tx.update(invitation).set({ status }).where(eq(invitation.id, invitationId))
 
@@ -342,8 +341,7 @@ export async function PUT(
           createdAt: new Date(),
         })
 
-        // Snapshot Pro usage and cancel Pro subscription when joining a paid team
-        try {
+        {
           const orgSubs = await tx
             .select()
             .from(subscriptionTable)
@@ -393,8 +391,9 @@ export async function PUT(
                   .update(userStats)
                   .set({
                     proPeriodCostSnapshot: currentProUsage,
-                    currentPeriodCost: '0', // Reset so new usage is attributed to team
-                    currentPeriodCopilotCost: '0', // Reset copilot cost for new period
+                    proPeriodCostSnapshotAt: new Date(),
+                    currentPeriodCost: '0',
+                    currentPeriodCopilotCost: '0',
                   })
                   .where(eq(userStats.userId, userId))
 
@@ -405,20 +404,20 @@ export async function PUT(
                 })
               }
 
-              // Mark for cancellation after transaction
-              if (personalPro.cancelAtPeriodEnd !== true) {
-                personalProToCancel = personalPro
+              if (personalPro.cancelAtPeriodEnd !== true && personalPro.stripeSubscriptionId) {
+                await tx
+                  .update(subscriptionTable)
+                  .set({ cancelAtPeriodEnd: true })
+                  .where(eq(subscriptionTable.id, personalPro.id))
+
+                await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
+                  stripeSubscriptionId: personalPro.stripeSubscriptionId,
+                  subscriptionId: personalPro.id,
+                  reason: 'member-joined-paid-org',
+                })
               }
             }
 
-            // Transfer the joining user's pre-join storage bytes into
-            // the org pool — after this point storage reads/writes route
-            // through `organization.storageUsedBytes`, so bytes left on
-            // `user_stats` would be orphaned (and a later decrement from
-            // deleting a pre-join file would wrongly reduce the org
-            // pool). `.for('update')` row-locks `user_stats` so a
-            // concurrent increment/decrement can't land between the
-            // SELECT and the zero UPDATE and get silently dropped.
             const storageRows = await tx
               .select({ storageUsedBytes: userStats.storageUsedBytes })
               .from(userStats)
@@ -447,13 +446,6 @@ export async function PUT(
               })
             }
           }
-        } catch (error) {
-          logger.error('Failed to handle Pro user joining team', {
-            userId: session.user.id,
-            organizationId,
-            error,
-          })
-          // Don't fail the whole invitation acceptance due to this
         }
 
         // Auto-assign to permission group if one has autoAddNewMembers enabled
@@ -593,44 +585,6 @@ export async function PUT(
       }
     }
 
-    // Handle Pro subscription cancellation after transaction commits
-    if (personalProToCancel) {
-      try {
-        const stripe = requireStripeClient()
-        if (personalProToCancel.stripeSubscriptionId) {
-          try {
-            await stripe.subscriptions.update(personalProToCancel.stripeSubscriptionId, {
-              cancel_at_period_end: true,
-            })
-          } catch (stripeError) {
-            logger.error('Failed to set cancel_at_period_end on Stripe for personal Pro', {
-              userId: session.user.id,
-              subscriptionId: personalProToCancel.id,
-              stripeSubscriptionId: personalProToCancel.stripeSubscriptionId,
-              error: stripeError,
-            })
-          }
-        }
-
-        await db
-          .update(subscriptionTable)
-          .set({ cancelAtPeriodEnd: true })
-          .where(eq(subscriptionTable.id, personalProToCancel.id))
-
-        logger.info('Auto-cancelled personal Pro at period end after joining paid team', {
-          userId: session.user.id,
-          personalSubscriptionId: personalProToCancel.id,
-          organizationId,
-        })
-      } catch (dbError) {
-        logger.error('Failed to update DB cancelAtPeriodEnd for personal Pro', {
-          userId: session.user.id,
-          subscriptionId: personalProToCancel.id,
-          error: dbError,
-        })
-      }
-    }
-
     if (status === 'accepted') {
       try {
         await syncUsageLimitsFromSubscription(session.user.id)

apps/sim/app/api/organizations/[id]/members/route.ts

@@ -1,12 +1,19 @@
 import { db } from '@sim/db'
-import { invitation, member, organization, user, userStats } from '@sim/db/schema'
+import {
+  invitation,
+  member,
+  organization,
+  subscription as subscriptionTable,
+  user,
+  userStats,
+} from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq } from 'drizzle-orm'
+import { and, eq, inArray } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getEmailSubject, renderInvitationEmail } from '@/components/emails'
 import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { getSession } from '@/lib/auth'
-import { getUserUsageData } from '@/lib/billing/core/usage'
+import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
 import { validateSeatAvailability } from '@/lib/billing/validation/seat-management'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { generateId } from '@/lib/core/utils/uuid'
@@ -83,16 +90,32 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
         .leftJoin(userStats, eq(user.id, userStats.userId))
         .where(eq(member.organizationId, organizationId))
 
-      const membersWithUsage = await Promise.all(
-        base.map(async (row) => {
-          const usage = await getUserUsageData(row.userId)
-          return {
-            ...row,
-            billingPeriodStart: usage.billingPeriodStart,
-            billingPeriodEnd: usage.billingPeriodEnd,
-          }
+      // The billing period is the same for every member — it comes from
+      // whichever subscription covers them. Fetch once and attach to
+      // every row instead of calling `getUserUsageData` per-member,
+      // which would run an O(N) pooled query for each of N rows.
+      const [orgSub] = await db
+        .select({
+          periodStart: subscriptionTable.periodStart,
+          periodEnd: subscriptionTable.periodEnd,
         })
-      )
+        .from(subscriptionTable)
+        .where(
+          and(
+            eq(subscriptionTable.referenceId, organizationId),
+            inArray(subscriptionTable.status, ENTITLED_SUBSCRIPTION_STATUSES)
+          )
+        )
+        .limit(1)
+
+      const billingPeriodStart = orgSub?.periodStart ?? null
+      const billingPeriodEnd = orgSub?.periodEnd ?? null
+
+      const membersWithUsage = base.map((row) => ({
+        ...row,
+        billingPeriodStart,
+        billingPeriodEnd,
+      }))
 
       return NextResponse.json({
         success: true,

apps/sim/app/api/v1/admin/organizations/[id]/members/route.ts

@@ -33,7 +33,6 @@ import { member, organization, user, userStats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { count, eq } from 'drizzle-orm'
 import { addUserToOrganization } from '@/lib/billing/organizations/membership'
-import { requireStripeClient } from '@/lib/billing/stripe-client'
 import { isBillingEnabled } from '@/lib/core/config/feature-flags'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 import {
@@ -229,28 +228,6 @@ export const POST = withAdminAuthParams<RouteParams>(async (request, context) =>
       return badRequestResponse(result.error || 'Failed to add member')
     }
 
-    if (isBillingEnabled && result.billingActions.proSubscriptionToCancel?.stripeSubscriptionId) {
-      try {
-        const stripe = requireStripeClient()
-        await stripe.subscriptions.update(
-          result.billingActions.proSubscriptionToCancel.stripeSubscriptionId,
-          { cancel_at_period_end: true }
-        )
-        logger.info('Admin API: Synced Pro cancellation with Stripe', {
-          userId: body.userId,
-          subscriptionId: result.billingActions.proSubscriptionToCancel.subscriptionId,
-          stripeSubscriptionId: result.billingActions.proSubscriptionToCancel.stripeSubscriptionId,
-        })
-      } catch (stripeError) {
-        logger.error('Admin API: Failed to sync Pro cancellation with Stripe', {
-          userId: body.userId,
-          subscriptionId: result.billingActions.proSubscriptionToCancel.subscriptionId,
-          stripeSubscriptionId: result.billingActions.proSubscriptionToCancel.stripeSubscriptionId,
-          error: stripeError,
-        })
-      }
-    }
-
     const data: AdminMember = {
       id: result.memberId!,
       userId: body.userId,

apps/sim/app/api/v1/admin/outbox/[id]/requeue/route.ts

@@ -0,0 +1,71 @@
+import { db } from '@sim/db'
+import { outboxEvent } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { and, eq } from 'drizzle-orm'
+import { NextResponse } from 'next/server'
+import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
+
+const logger = createLogger('AdminOutboxRequeueAPI')
+
+export const dynamic = 'force-dynamic'
+
+/**
+ * POST /api/v1/admin/outbox/[id]/requeue
+ *
+ * Move a dead-lettered outbox event back to `pending` so the worker
+ * will retry it. Resets `attempts`, `lastError`, and `availableAt` so
+ * the next poll picks it up. Only dead-lettered events can be
+ * requeued — completed/pending/processing rows are rejected to avoid
+ * operator errors.
+ */
+export const POST = withAdminAuthParams<{ id: string }>(async (_request, { params }) => {
+  const { id } = await params
+
+  try {
+    const result = await db
+      .update(outboxEvent)
+      .set({
+        status: 'pending',
+        attempts: 0,
+        lastError: null,
+        availableAt: new Date(),
+        lockedAt: null,
+        processedAt: null,
+      })
+      .where(and(eq(outboxEvent.id, id), eq(outboxEvent.status, 'dead_letter')))
+      .returning({ id: outboxEvent.id, eventType: outboxEvent.eventType })
+
+    if (result.length === 0) {
+      return NextResponse.json(
+        {
+          success: false,
+          error:
+            'Event not found or not in dead_letter status. Only dead-lettered events can be requeued.',
+        },
+        { status: 404 }
+      )
+    }
+
+    logger.info('Requeued dead-lettered outbox event', {
+      eventId: result[0].id,
+      eventType: result[0].eventType,
+    })
+
+    return NextResponse.json({
+      success: true,
+      requeued: result[0],
+    })
+  } catch (error) {
+    logger.error('Failed to requeue outbox event', {
+      eventId: id,
+      error: error instanceof Error ? error.message : error,
+    })
+    return NextResponse.json(
+      {
+        success: false,
+        error: error instanceof Error ? error.message : 'Unknown error',
+      },
+      { status: 500 }
+    )
+  }
+})