Lineblocs · gaurav-vaghela · May 11, 2026
diff --git a/app/app/Console/Commands/ApplyScheduledPlanUpgradesCommand.php b/app/app/Console/Commands/ApplyScheduledPlanUpgradesCommand.php
@@ -0,0 +1,62 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Subscription;
+use DateTime;
+use DB;
+use Illuminate\Console\Command;
+
+class ApplyScheduledPlanUpgradesCommand extends Command
+{
+    protected $signature = 'billing:apply-scheduled-plan-upgrades {--now=} {--limit=100}';
+    protected $description = 'Activate scheduled subscription plan upgrades at the billing cycle anchor';
+
+    public function handle()
+    {
+        $now = $this->option('now');
+        if (empty($now)) {
+            $now = (new DateTime())->format('Y-m-d H:i:s');
+        }
+
+        $limit = (int) $this->option('limit');
+        if ($limit <= 0) {
+            $limit = 100;
+        }
+
+        $subscriptions = Subscription::whereNotNull('scheduled_plan_id')
+            ->whereNotNull('scheduled_effective_date')
+            ->where('scheduled_effective_date', '<=', $now)
+            ->orderBy('scheduled_effective_date', 'asc')
+            ->take($limit)
+            ->get();
+
+        $migrated = 0;
+
+        foreach ($subscriptions as $queuedSubscription) {
+            $didMigrate = DB::transaction(function () use ($queuedSubscription) {
+                $subscription = Subscription::where('id', $queuedSubscription->id)
+                    ->lockForUpdate()
+                    ->first();
+
+                if (!$subscription || empty($subscription->scheduled_plan_id)) {
+                    return false;
+                }
+
+                $subscription->current_plan_id = $subscription->scheduled_plan_id;
+                $subscription->scheduled_plan_id = null;
+                $subscription->scheduled_effective_date = null;
+                $subscription->save();
+
+                return true;
+            });
+
+            if ($didMigrate) {
+                $migrated++;
+                $this->info('Activated scheduled plan for subscription #' . $queuedSubscription->id);
+            }
+        }
+
+        $this->info('Scheduled plan upgrades activated: ' . $migrated);
+    }
+}
diff --git a/app/app/Console/Commands/RabbitMQEventConsumer.php b/app/app/Console/Commands/RabbitMQEventConsumer.php
@@ -109,10 +109,46 @@ public function handlePaymentFailure($msg)
      */
     public function handleSubscriptionUpdate($msg)
     {
-        $data = json_decode($msg->body, true);
-        $this->info(" [SUBSCRIPTION] Received update for ID: " . $data['subscription_id']);
+        try {
+            $data = json_decode($msg->body, true);
+            if (!is_array($data)) {
+                throw new Exception('Invalid JSON payload.');
+            }
 
-        // Implementation for subscription logic goes here
+            $required = [
+                'event_type',
+                'version',
+                'event_id',
+                'occurred_at',
+                'subscription_id',
+                'workspace_id',
+                'actor_user_id',
+                'current_plan_id',
+                'scheduled_plan_id',
+                'billing_cycle',
+                'scheduled_effective_date'
+            ];
+
+            foreach ($required as $field) {
+                if (!array_key_exists($field, $data) || $data[$field] === null || $data[$field] === '') {
+                    throw new Exception("Invalid subscription update payload: missing {$field}");
+                }
+            }
+
+            if ($data['event_type'] !== RabbitMQHelper::SUBSCRIPTION_UPGRADE_SCHEDULED) {
+                throw new Exception('Unsupported subscription update event_type: ' . $data['event_type']);
+            }
+
+            if (!in_array($data['billing_cycle'], ['MONTHLY', 'ANNUAL'])) {
+                throw new Exception('Invalid subscription update payload: billing_cycle must be MONTHLY or ANNUAL');
+            }
+
+            $this->info(" [SUBSCRIPTION] Received scheduled upgrade for ID: " . $data['subscription_id']);
+        } catch (Exception $e) {
+            $this->error(" [SUBSCRIPTION] " . $e->getMessage());
+            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
+            return;
+        }
 
         $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
         $this->info(" [v] Subscription event acknowledged.");

diff --git a/app/app/Helpers/RabbitMQHelper.php b/app/app/Helpers/RabbitMQHelper.php
@@ -11,6 +11,8 @@ class RabbitMQHelper
 {
     const INVOICE_QUEUE_MONTHLY = 'monthly_invoices';
     const INVOICE_QUEUE_ANNUAL = 'annual_invoices';
+    const SUBSCRIPTION_UPDATES_QUEUE = 'subscription_updates';
+    const SUBSCRIPTION_UPGRADE_SCHEDULED = 'subscription.plan_upgrade_scheduled';
 
     /**
      * Generic method to publish a message to any RabbitMQ queue.
@@ -75,6 +77,79 @@ public static function dispatchImmediateBilling($workspace, $subscription, $user
         return self::publish('billing_tasks', $payload);
     }
 
+    public static function dispatchPlanUpgradeScheduled($workspace, $subscription, $user)
+    {
+        $effectiveDate = $subscription->scheduled_effective_date;
+        $currentPeriodEnd = $subscription->current_period_end;
+        $nextBillingDate = $subscription->next_billing_date;
+
+        $formattedCurrentPeriodEnd = null;
+        if ($currentPeriodEnd) {
+            $formattedCurrentPeriodEnd = date('c', strtotime($currentPeriodEnd));
+        }
+
+        $formattedNextBillingDate = null;
+        if ($nextBillingDate) {
+            $formattedNextBillingDate = date('c', strtotime($nextBillingDate));
+        }
+
+        $formattedEffectiveDate = null;
+        if ($effectiveDate) {
+            $formattedEffectiveDate = date('c', strtotime($effectiveDate));
+        }
+
+        $payload = [
+            'event_type' => self::SUBSCRIPTION_UPGRADE_SCHEDULED,
+            'version' => '1.0',
+            'event_id' => 'plan_upgrade_' . $subscription->id . '_' . time(),
+            'occurred_at' => date('c'),
+            'subscription_id' => (int) $subscription->id,
+            'workspace_id' => (int) $workspace->id,
+            'actor_user_id' => (int) $user->id,
+            'current_plan_id' => (int) $subscription->current_plan_id,
+            'scheduled_plan_id' => (int) $subscription->scheduled_plan_id,
+            'billing_cycle' => (string) $subscription->billing_cycle,
+            'current_period_end' => $formattedCurrentPeriodEnd,
+            'next_billing_date' => $formattedNextBillingDate,
+            'scheduled_effective_date' => $formattedEffectiveDate
+        ];
+
+        self::assertPlanUpgradeScheduledPayload($payload);
+
+        return self::publish(self::SUBSCRIPTION_UPDATES_QUEUE, $payload);
+    }
+
+    private static function assertPlanUpgradeScheduledPayload(array $payload)
+    {
+        $required = [
+            'event_type',
+            'version',
+            'event_id',
+            'occurred_at',
+            'subscription_id',
+            'workspace_id',
+            'actor_user_id',
+            'current_plan_id',
+            'scheduled_plan_id',
+            'billing_cycle',
+            'scheduled_effective_date'
+        ];
+
+        foreach ($required as $field) {
+            if (!array_key_exists($field, $payload) || $payload[$field] === null || $payload[$field] === '') {
+                throw new Exception("Invalid subscription upgrade payload: missing {$field}");
+            }
+        }
+
+        if ($payload['event_type'] !== self::SUBSCRIPTION_UPGRADE_SCHEDULED) {
+            throw new Exception('Invalid subscription upgrade payload: unexpected event_type');
+        }
+
+        if (!in_array($payload['billing_cycle'], ['MONTHLY', 'ANNUAL'])) {
+            throw new Exception('Invalid subscription upgrade payload: unexpected billing_cycle');
+        }
+    }
+
     public static function dispatchSurveyEmail($email, $surveyTypes = [], $name = '')
     {
         if (!is_array($surveyTypes)) {

diff --git a/app/app/Http/Controllers/BillingController.php b/app/app/Http/Controllers/BillingController.php
@@ -19,6 +19,7 @@
 use App\Helpers\WorkspaceHelper;
 use App\Helpers\InvoiceHelper;
 use DateTime;
+use Exception;
 
 class BillingController extends ApiAuthController
 {
@@ -57,6 +58,65 @@ public function getBillingInfo(Request $request)
         ]);
     }
 
+    public function schedulePlanUpgrade(Request $request)
+    {
+      $user = $this->getUser($request);
+      $workspace = $this->getWorkspace($request);
+
+      if (!WorkspaceHelper::canPerformAction($user, $workspace, 'manage_billing')) {
+        return $this->response->errorForbidden();
+      }
+
+      $data = $request->json()->all();
+      if (!isset($data['plan_id'])) {
+        return $this->response->errorBadRequest('plan_id is required');
+      }
+
+      $newPlan = ServicePlan::find((int) $data['plan_id']);
+      if (!$newPlan) {
+        return $this->response->errorBadRequest('plan_id is invalid');
+      }
+
+      try {
+        $subscription = DB::transaction(function () use ($workspace, $newPlan) {
+          $subscription = Subscription::where('workspace_id', $workspace->id)
+            ->lockForUpdate()
+            ->firstOrFail();
+
+          if ((int) $subscription->current_plan_id === (int) $newPlan->id) {
+            throw new Exception('The requested plan is already active.');
+          }
+
+          $effectiveDate = $subscription->next_billing_date;
+          if (empty($effectiveDate)) {
+            $effectiveDate = $subscription->current_period_end;
+          }
+
+          if (empty($effectiveDate)) {
+            throw new Exception('Subscription has no billing cycle anchor.');
+          }
+
+          $subscription->scheduled_plan_id = $newPlan->id;
+          $subscription->scheduled_effective_date = $effectiveDate;
+          $subscription->save();
+
+          return $subscription;
+        });
+
+        RabbitMQHelper::dispatchPlanUpgradeScheduled($workspace, $subscription, $user);
+      } catch (Exception $e) {
+        return $this->response->errorBadRequest($e->getMessage());
+      }
+
+      return $this->response->array([
+        'subscription_id' => (int) $subscription->id,
+        'workspace_id' => (int) $workspace->id,
+        'current_plan_id' => (int) $subscription->current_plan_id,
+        'scheduled_plan_id' => (int) $subscription->scheduled_plan_id,
+        'scheduled_effective_date' => $subscription->scheduled_effective_date
+      ]);
+    }
+
     public function getBillingHistory(Request $request)
     {
       $user = $this->getUser($request);

diff --git a/app/app/Http/routes.php b/app/app/Http/routes.php
@@ -892,6 +892,7 @@
       // Current routes (example of grouping them for cleaner code)
       $api->get('info', '\App\Http\Controllers\BillingController@getBillingInfo');
       $api->get('history', '\App\Http\Controllers\BillingController@getBillingHistory');
+      $api->post('plan-upgrade', '\App\Http\Controllers\BillingController@schedulePlanUpgrade');
 
       // New Settlement Route
       $api->post('invoices/{invoiceId}/settle', '\App\Http\Controllers\BillingController@settleInvoice');

diff --git a/app/database/migrations/2026_04_26_222012_add_status_to_workspaces_users.php b/app/database/migrations/2026_04_26_222012_add_status_to_workspaces_users.php
@@ -3,7 +3,7 @@
 use Illuminate\Database\Schema\Blueprint;
 use Illuminate\Database\Migrations\Migration;
 
-class AddStatusToWorkspacesUsers extends Migration
+class AddStatusToWorkspacesAcesUsers extends Migration
 {
     /**
      * Run the migrations.