mirror of
https://github.com/tiennm99/coolify.git
synced 2026-10-14 06:17:36 +00:00
Merge remote-tracking branch 'origin/next' into v5-parallel-inertia-react
This commit is contained in:
commit
ae32ab5d7b
139 files changed
+8677
-545
No files matched your search
@@ -12,6 +12,7 @@ use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\Middleware\WithoutOverlapping;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Support\Collection;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
|
||||
/**
|
||||
@@ -53,11 +54,13 @@ class CleanupOrphanedPreviewContainersJob implements ShouldBeEncrypted, ShouldBe
|
||||
/**
|
||||
* Get all functional servers to check for orphaned containers.
|
||||
*/
|
||||
private function getServersToCheck(): \Illuminate\Support\Collection
|
||||
private function getServersToCheck(): Collection
|
||||
{
|
||||
$query = Server::whereRelation('settings', 'is_usable', true)
|
||||
->whereRelation('settings', 'is_reachable', true)
|
||||
->where('ip', '!=', '1.2.3.4');
|
||||
->whereNotNull('ip')
|
||||
->where('ip', '!=', '')
|
||||
->whereNotIn('ip', Server::PLACEHOLDER_IPS);
|
||||
|
||||
if (isCloud()) {
|
||||
$query = $query->whereRelation('team.subscription', 'stripe_invoice_paid', true);
|
||||
@@ -99,7 +102,7 @@ class CleanupOrphanedPreviewContainersJob implements ShouldBeEncrypted, ShouldBe
|
||||
/**
|
||||
* Get all PR containers on a server (containers with pullRequestId > 0).
|
||||
*/
|
||||
private function getPRContainersOnServer(Server $server): \Illuminate\Support\Collection
|
||||
private function getPRContainersOnServer(Server $server): Collection
|
||||
{
|
||||
try {
|
||||
$output = instant_remote_process([
|
||||
|
||||
@@ -8,6 +8,7 @@ use App\Models\ScheduledDatabaseBackup;
|
||||
use App\Models\ScheduledDatabaseBackupExecution;
|
||||
use App\Models\Server;
|
||||
use App\Models\ServiceDatabase;
|
||||
use App\Models\StandaloneClickhouse;
|
||||
use App\Models\StandaloneMariadb;
|
||||
use App\Models\StandaloneMongodb;
|
||||
use App\Models\StandaloneMysql;
|
||||
@@ -17,6 +18,7 @@ use App\Notifications\Database\BackupFailed;
|
||||
use App\Notifications\Database\BackupSuccess;
|
||||
use App\Notifications\Database\BackupSuccessWithS3Warning;
|
||||
use App\Rules\SafeWebhookUrl;
|
||||
use App\Support\ClickhouseBackupCommand;
|
||||
use Carbon\Carbon;
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeEncrypted;
|
||||
@@ -39,7 +41,7 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
|
||||
public Server $server;
|
||||
|
||||
public StandalonePostgresql|StandaloneMongodb|StandaloneMysql|StandaloneMariadb|ServiceDatabase $database;
|
||||
public StandalonePostgresql|StandaloneMongodb|StandaloneMysql|StandaloneMariadb|StandaloneClickhouse|ServiceDatabase $database;
|
||||
|
||||
public ?string $container_name = null;
|
||||
|
||||
@@ -271,6 +273,8 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
$databasesToBackup = [$this->database->mysql_database];
|
||||
} elseif (str($databaseType)->contains('mariadb')) {
|
||||
$databasesToBackup = [$this->database->mariadb_database];
|
||||
} elseif ($this->database instanceof StandaloneClickhouse) {
|
||||
$databasesToBackup = [$this->database->clickhouse_db];
|
||||
} else {
|
||||
return;
|
||||
}
|
||||
@@ -294,6 +298,10 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
// Format: db1,db2,db3
|
||||
$databasesToBackup = explode(',', $databasesToBackup);
|
||||
$databasesToBackup = array_map('trim', $databasesToBackup);
|
||||
} elseif ($this->database instanceof StandaloneClickhouse) {
|
||||
// Format: db1,db2,db3
|
||||
$databasesToBackup = explode(',', $databasesToBackup);
|
||||
$databasesToBackup = array_map('trim', $databasesToBackup);
|
||||
} else {
|
||||
return;
|
||||
}
|
||||
@@ -386,6 +394,17 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
'local_storage_deleted' => false,
|
||||
]);
|
||||
$this->backup_standalone_mariadb($database);
|
||||
} elseif ($this->database instanceof StandaloneClickhouse) {
|
||||
$this->backup_file = '/clickhouse-backup-'.Carbon::now()->timestamp."-{$this->backup_log_uuid}.zip";
|
||||
$this->backup_location = $this->backup_dir.$this->backup_file;
|
||||
$this->backup_log = ScheduledDatabaseBackupExecution::create([
|
||||
'uuid' => $this->backup_log_uuid,
|
||||
'database_name' => $database,
|
||||
'filename' => $this->backup_location,
|
||||
'scheduled_database_backup_id' => $this->backup->id,
|
||||
'local_storage_deleted' => false,
|
||||
]);
|
||||
$this->backup_standalone_clickhouse($database);
|
||||
} else {
|
||||
throw new \Exception('Unsupported database type');
|
||||
}
|
||||
@@ -400,6 +419,9 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
}
|
||||
} catch (Throwable $e) {
|
||||
// Local backup failed
|
||||
if ($this->database instanceof StandaloneClickhouse) {
|
||||
deleteBackupsLocally($this->backup_location, $this->server);
|
||||
}
|
||||
if ($this->backup_log) {
|
||||
$this->backup_log->update([
|
||||
'status' => 'failed',
|
||||
@@ -642,6 +664,32 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue
|
||||
}
|
||||
}
|
||||
|
||||
private function backup_standalone_clickhouse(string $database): void
|
||||
{
|
||||
$archiveName = ltrim($this->backup_file, '/');
|
||||
|
||||
try {
|
||||
$commands = ClickhouseBackupCommand::make(
|
||||
containerName: $this->container_name,
|
||||
database: $database,
|
||||
archiveName: $archiveName,
|
||||
backupDirectory: $this->backup_dir,
|
||||
);
|
||||
|
||||
$this->backup_output = instant_remote_process($commands, $this->server, true, false, $this->timeout, disableMultiplexing: true);
|
||||
$this->backup_output = trim($this->backup_output);
|
||||
if ($this->backup_output === '') {
|
||||
$this->backup_output = null;
|
||||
}
|
||||
} catch (Throwable $e) {
|
||||
$this->add_to_error_output($e->getMessage());
|
||||
throw $e;
|
||||
} finally {
|
||||
$cleanupCommand = ClickhouseBackupCommand::cleanup($this->container_name, $archiveName);
|
||||
instant_remote_process([$cleanupCommand], $this->server, false, false, null, disableMultiplexing: true);
|
||||
}
|
||||
}
|
||||
|
||||
private function add_to_backup_output($output): void
|
||||
{
|
||||
if ($this->backup_output) {
|
||||
|
||||
@@ -311,6 +311,10 @@ class PushServerUpdateJob implements ShouldBeEncrypted, ShouldQueue, Silenced
|
||||
}
|
||||
}
|
||||
|
||||
if (! $this->isCompleteSnapshot()) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->updateProxyStatus();
|
||||
|
||||
$this->updateNotFoundApplicationStatus();
|
||||
@@ -329,6 +333,11 @@ class PushServerUpdateJob implements ShouldBeEncrypted, ShouldQueue, Silenced
|
||||
$this->checkLogDrainContainer();
|
||||
}
|
||||
|
||||
private function isCompleteSnapshot(): bool
|
||||
{
|
||||
return data_get($this->data, 'snapshot.complete', true) !== false;
|
||||
}
|
||||
|
||||
private function loadApplications(): Collection
|
||||
{
|
||||
[$standaloneDockerIds, $swarmDockerIds] = $this->serverDestinationIds();
|
||||
@@ -700,6 +709,9 @@ class PushServerUpdateJob implements ShouldBeEncrypted, ShouldQueue, Silenced
|
||||
$database->status = $containerStatus;
|
||||
$database->save();
|
||||
}
|
||||
if (! $this->isCompleteSnapshot()) {
|
||||
return;
|
||||
}
|
||||
if ($this->isRunning($containerStatus) && $tcpProxy) {
|
||||
$tcpProxyContainerFound = $this->containers->filter(function ($value, $key) use ($databaseUuid) {
|
||||
return data_get($value, 'name') === "$databaseUuid-proxy" && data_get($value, 'state') === 'running';
|
||||
|
||||
@@ -457,7 +457,9 @@ class ScheduledJobManager implements ShouldQueue
|
||||
private function getServersForCleanupQuery(): Builder
|
||||
{
|
||||
$query = Server::with('settings')
|
||||
->where('ip', '!=', '1.2.3.4');
|
||||
->whereNotNull('ip')
|
||||
->where('ip', '!=', '')
|
||||
->whereNotIn('ip', Server::PLACEHOLDER_IPS);
|
||||
|
||||
if (isCloud()) {
|
||||
$query
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
<?php
|
||||
|
||||
namespace App\Jobs;
|
||||
|
||||
use App\Models\Server;
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeEncrypted;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\Middleware\WithoutOverlapping;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
|
||||
class ServerCloudProviderStatusCheckJob implements ShouldBeEncrypted, ShouldQueue
|
||||
{
|
||||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||||
|
||||
public $tries = 1;
|
||||
|
||||
public $timeout = 120;
|
||||
|
||||
public function __construct(public Server $server)
|
||||
{
|
||||
$this->onQueue('high');
|
||||
}
|
||||
|
||||
public function middleware(): array
|
||||
{
|
||||
return [(new WithoutOverlapping('server-cloud-provider-status-'.$this->server->uuid))->expireAfter(130)->dontRelease()];
|
||||
}
|
||||
|
||||
public function handle(): void
|
||||
{
|
||||
try {
|
||||
if (! $this->server->cloudProviderToken) {
|
||||
return;
|
||||
}
|
||||
|
||||
match ($this->server->cloudProviderToken->provider) {
|
||||
'hetzner' => $this->server->hetzner_server_id
|
||||
? $this->server->refreshHetznerState()
|
||||
: null,
|
||||
'vultr' => $this->server->vultr_instance_id
|
||||
? $this->server->refreshVultrState()
|
||||
: null,
|
||||
'digitalocean' => $this->server->digitalocean_droplet_id
|
||||
? $this->server->refreshDigitalOceanState()
|
||||
: null,
|
||||
default => null,
|
||||
};
|
||||
} catch (\Throwable $e) {
|
||||
Log::debug('Cloud provider status check failed', [
|
||||
'server_id' => $this->server->id,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -6,7 +6,6 @@ use App\Events\ServerReachabilityChanged;
|
||||
use App\Helpers\SshMultiplexingHelper;
|
||||
use App\Models\Server;
|
||||
use App\Services\ConfigurationRepository;
|
||||
use App\Services\HetznerService;
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeEncrypted;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
@@ -42,8 +41,12 @@ class ServerConnectionCheckJob implements ShouldBeEncrypted, ShouldQueue
|
||||
$configRepository->disableSshMux();
|
||||
}
|
||||
|
||||
public function handle()
|
||||
public function handle(): void
|
||||
{
|
||||
if ($this->server->hasPlaceholderIp()) {
|
||||
return;
|
||||
}
|
||||
|
||||
$wasReachable = (bool) $this->server->settings->is_reachable;
|
||||
$wasNotified = (bool) $this->server->unreachable_notification_sent;
|
||||
|
||||
@@ -62,19 +65,6 @@ class ServerConnectionCheckJob implements ShouldBeEncrypted, ShouldQueue
|
||||
return;
|
||||
}
|
||||
|
||||
// Check Hetzner server status if applicable
|
||||
if ($this->server->hetzner_server_id && $this->server->cloudProviderToken) {
|
||||
$this->checkHetznerStatus();
|
||||
}
|
||||
|
||||
if ($this->server->vultr_instance_id && $this->server->cloudProviderToken) {
|
||||
$this->checkVultrStatus();
|
||||
}
|
||||
|
||||
if ($this->server->digitalocean_droplet_id && $this->server->cloudProviderToken) {
|
||||
$this->checkDigitalOceanStatus();
|
||||
}
|
||||
|
||||
// Temporarily disable mux if requested
|
||||
if ($this->disableMux) {
|
||||
$this->disableSshMux();
|
||||
@@ -136,17 +126,6 @@ class ServerConnectionCheckJob implements ShouldBeEncrypted, ShouldQueue
|
||||
public function failed(?\Throwable $exception): void
|
||||
{
|
||||
if ($exception instanceof TimeoutExceededException) {
|
||||
$wasReachable = (bool) $this->server->settings->is_reachable;
|
||||
$wasNotified = (bool) $this->server->unreachable_notification_sent;
|
||||
|
||||
$this->server->settings->update([
|
||||
'is_reachable' => false,
|
||||
'is_usable' => false,
|
||||
]);
|
||||
$this->server->increment('unreachable_count');
|
||||
|
||||
$this->dispatchReachabilityChangedIfNeeded($wasReachable, $wasNotified, false);
|
||||
|
||||
// Delete the queue job so it doesn't appear in Horizon's failed list.
|
||||
$this->job?->delete();
|
||||
}
|
||||
@@ -171,63 +150,6 @@ class ServerConnectionCheckJob implements ShouldBeEncrypted, ShouldQueue
|
||||
}
|
||||
}
|
||||
|
||||
private function checkHetznerStatus(): void
|
||||
{
|
||||
$status = null;
|
||||
|
||||
try {
|
||||
$hetznerService = new HetznerService($this->server->cloudProviderToken->token);
|
||||
$serverData = $hetznerService->getServer($this->server->hetzner_server_id);
|
||||
$status = $serverData['status'] ?? null;
|
||||
|
||||
} catch (\Throwable) {
|
||||
// Silently ignore — server may have been deleted from Hetzner.
|
||||
}
|
||||
if ($this->server->hetzner_server_status !== $status) {
|
||||
$this->server->update(['hetzner_server_status' => $status]);
|
||||
$this->server->hetzner_server_status = $status;
|
||||
if ($status === 'off') {
|
||||
throw new \Exception('Server is powered off');
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private function checkVultrStatus(): void
|
||||
{
|
||||
try {
|
||||
$status = $this->server->refreshVultrState();
|
||||
} catch (\Throwable) {
|
||||
// Silently ignore transient Vultr API errors.
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
if (in_array($status, ['stopped', 'suspended', 'deleted'], true)) {
|
||||
throw new \Exception('Vultr instance is not running');
|
||||
}
|
||||
}
|
||||
|
||||
private function checkDigitalOceanStatus(): void
|
||||
{
|
||||
try {
|
||||
$status = $this->server->refreshDigitalOceanState();
|
||||
} catch (\Throwable $e) {
|
||||
Log::debug('ServerConnectionCheck: DigitalOcean status check failed', [
|
||||
'server_id' => $this->server->id,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$this->server->digitalocean_droplet_status = $status;
|
||||
|
||||
if (in_array($status, ['off', 'archive', 'deleted'], true)) {
|
||||
throw new \Exception('DigitalOcean droplet is not running');
|
||||
}
|
||||
}
|
||||
|
||||
private function checkConnection(): bool
|
||||
{
|
||||
try {
|
||||
|
||||
@@ -55,6 +55,13 @@ class ServerManagerJob implements ShouldBeEncrypted, ShouldQueue
|
||||
// Get all servers to process
|
||||
$servers = $this->getServers();
|
||||
|
||||
// Provider state checks run independently so slow APIs cannot block SSH checks.
|
||||
$this->dispatchCloudProviderStatusChecks($servers);
|
||||
|
||||
$servers = $servers
|
||||
->reject(fn (Server $server) => $server->hasPlaceholderIp())
|
||||
->values();
|
||||
|
||||
// Dispatch ServerConnectionCheck for all servers efficiently
|
||||
$this->dispatchConnectionChecks($servers);
|
||||
|
||||
@@ -64,24 +71,45 @@ class ServerManagerJob implements ShouldBeEncrypted, ShouldQueue
|
||||
|
||||
private function getServers(): Collection
|
||||
{
|
||||
$allServers = Server::with('settings')->where('ip', '!=', '1.2.3.4');
|
||||
$allServers = Server::with(['settings', 'cloudProviderToken']);
|
||||
|
||||
if (isCloud()) {
|
||||
$servers = $allServers->whereRelation('team.subscription', 'stripe_invoice_paid', true)->get();
|
||||
$own = Team::find(0)->servers()->with('settings')->get();
|
||||
$own = Team::find(0)->servers()->with(['settings', 'cloudProviderToken'])->get();
|
||||
|
||||
return $servers->merge($own);
|
||||
return $servers->merge($own)->unique('id')->values();
|
||||
} else {
|
||||
return $allServers->get();
|
||||
}
|
||||
}
|
||||
|
||||
private function dispatchCloudProviderStatusChecks(Collection $servers): void
|
||||
{
|
||||
if (! shouldRunCronNow($this->checkFrequency, $this->instanceTimezone, 'server-cloud-provider-status-checks', $this->executionTime)) {
|
||||
return;
|
||||
}
|
||||
|
||||
$servers->each(function (Server $server) {
|
||||
$hasCloudResource = $server->hetzner_server_id
|
||||
|| $server->vultr_instance_id
|
||||
|| $server->digitalocean_droplet_id;
|
||||
|
||||
if ($hasCloudResource && $server->cloudProviderToken) {
|
||||
ServerCloudProviderStatusCheckJob::dispatch($server);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private function dispatchConnectionChecks(Collection $servers): void
|
||||
{
|
||||
|
||||
if (shouldRunCronNow($this->checkFrequency, $this->instanceTimezone, 'server-connection-checks', $this->executionTime)) {
|
||||
$servers->each(function (Server $server) {
|
||||
try {
|
||||
if ($server->hasPlaceholderIp()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip SSH connection check if Sentinel is healthy — its heartbeat already proves connectivity
|
||||
if ($server->isSentinelEnabled() && $server->isSentinelLive()) {
|
||||
return;
|
||||
|
||||
@@ -1,199 +0,0 @@
|
||||
<?php
|
||||
|
||||
namespace App\Jobs;
|
||||
|
||||
use App\Models\Subscription;
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeEncrypted;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Stripe\StripeClient;
|
||||
|
||||
class SyncStripeSubscriptionsJob implements ShouldBeEncrypted, ShouldQueue
|
||||
{
|
||||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||||
|
||||
public int $tries = 1;
|
||||
|
||||
public int $timeout = 1800; // 30 minutes max
|
||||
|
||||
public function __construct(public bool $fix = false)
|
||||
{
|
||||
$this->onQueue('high');
|
||||
}
|
||||
|
||||
public function handle(?\Closure $onProgress = null): array
|
||||
{
|
||||
if (! isCloud() || ! isStripe()) {
|
||||
return ['error' => 'Not running on Cloud or Stripe not configured'];
|
||||
}
|
||||
|
||||
$subscriptions = Subscription::whereNotNull('stripe_subscription_id')
|
||||
->where('stripe_invoice_paid', true)
|
||||
->get();
|
||||
|
||||
$stripe = app(StripeClient::class);
|
||||
|
||||
// Bulk fetch all valid subscription IDs from Stripe (active + past_due)
|
||||
$validStripeIds = $this->fetchValidStripeSubscriptionIds($stripe, $onProgress);
|
||||
|
||||
// Find DB subscriptions not in the valid set
|
||||
$staleSubscriptions = $subscriptions->filter(
|
||||
fn (Subscription $sub) => ! in_array($sub->stripe_subscription_id, $validStripeIds)
|
||||
);
|
||||
|
||||
// For each stale subscription, get the exact Stripe status and check for resubscriptions
|
||||
$discrepancies = [];
|
||||
$resubscribed = [];
|
||||
$errors = [];
|
||||
|
||||
foreach ($staleSubscriptions as $subscription) {
|
||||
try {
|
||||
$stripeSubscription = $stripe->subscriptions->retrieve(
|
||||
$subscription->stripe_subscription_id
|
||||
);
|
||||
$stripeStatus = $stripeSubscription->status;
|
||||
|
||||
usleep(100000); // 100ms rate limit delay
|
||||
} catch (\Exception $e) {
|
||||
$errors[] = [
|
||||
'subscription_id' => $subscription->id,
|
||||
'error' => $e->getMessage(),
|
||||
];
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check if this user resubscribed under a different customer/subscription
|
||||
$activeSub = $this->findActiveSubscriptionByEmail($stripe, $stripeSubscription->customer);
|
||||
if ($activeSub) {
|
||||
$resubscribed[] = [
|
||||
'subscription_id' => $subscription->id,
|
||||
'team_id' => $subscription->team_id,
|
||||
'email' => $activeSub['email'],
|
||||
'old_stripe_subscription_id' => $subscription->stripe_subscription_id,
|
||||
'old_stripe_customer_id' => $stripeSubscription->customer,
|
||||
'new_stripe_subscription_id' => $activeSub['subscription_id'],
|
||||
'new_stripe_customer_id' => $activeSub['customer_id'],
|
||||
'new_status' => $activeSub['status'],
|
||||
];
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
$discrepancies[] = [
|
||||
'subscription_id' => $subscription->id,
|
||||
'team_id' => $subscription->team_id,
|
||||
'stripe_subscription_id' => $subscription->stripe_subscription_id,
|
||||
'stripe_status' => $stripeStatus,
|
||||
];
|
||||
|
||||
if ($this->fix) {
|
||||
$subscription->update([
|
||||
'stripe_invoice_paid' => false,
|
||||
'stripe_past_due' => false,
|
||||
]);
|
||||
|
||||
if ($stripeStatus === 'canceled') {
|
||||
$subscription->team?->subscriptionEnded();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if ($this->fix && count($discrepancies) > 0) {
|
||||
send_internal_notification(
|
||||
'SyncStripeSubscriptionsJob: Fixed '.count($discrepancies)." discrepancies:\n".
|
||||
json_encode($discrepancies, JSON_PRETTY_PRINT)
|
||||
);
|
||||
}
|
||||
|
||||
return [
|
||||
'total_checked' => $subscriptions->count(),
|
||||
'discrepancies' => $discrepancies,
|
||||
'resubscribed' => $resubscribed,
|
||||
'errors' => $errors,
|
||||
'fixed' => $this->fix,
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Given a Stripe customer ID, get their email and search for other customers
|
||||
* with the same email that have an active subscription.
|
||||
*
|
||||
* @return array{email: string, customer_id: string, subscription_id: string, status: string}|null
|
||||
*/
|
||||
private function findActiveSubscriptionByEmail(StripeClient $stripe, string $customerId): ?array
|
||||
{
|
||||
try {
|
||||
$customer = $stripe->customers->retrieve($customerId);
|
||||
$email = $customer->email;
|
||||
|
||||
if (! $email) {
|
||||
return null;
|
||||
}
|
||||
|
||||
usleep(100000);
|
||||
|
||||
$customers = $stripe->customers->all([
|
||||
'email' => $email,
|
||||
'limit' => 10,
|
||||
]);
|
||||
|
||||
usleep(100000);
|
||||
|
||||
foreach ($customers->data as $matchingCustomer) {
|
||||
if ($matchingCustomer->id === $customerId) {
|
||||
continue;
|
||||
}
|
||||
|
||||
$subs = $stripe->subscriptions->all([
|
||||
'customer' => $matchingCustomer->id,
|
||||
'limit' => 10,
|
||||
]);
|
||||
|
||||
usleep(100000);
|
||||
|
||||
foreach ($subs->data as $sub) {
|
||||
if (in_array($sub->status, ['active', 'past_due'])) {
|
||||
return [
|
||||
'email' => $email,
|
||||
'customer_id' => $matchingCustomer->id,
|
||||
'subscription_id' => $sub->id,
|
||||
'status' => $sub->status,
|
||||
];
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (\Exception $e) {
|
||||
// Silently skip — will fall through to normal discrepancy
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Bulk fetch all active and past_due subscription IDs from Stripe.
|
||||
*
|
||||
* @return array<string>
|
||||
*/
|
||||
private function fetchValidStripeSubscriptionIds(StripeClient $stripe, ?\Closure $onProgress = null): array
|
||||
{
|
||||
$validIds = [];
|
||||
$fetched = 0;
|
||||
|
||||
foreach (['active', 'past_due'] as $status) {
|
||||
foreach ($stripe->subscriptions->all(['status' => $status, 'limit' => 100])->autoPagingIterator() as $sub) {
|
||||
$validIds[] = $sub->id;
|
||||
$fetched++;
|
||||
|
||||
if ($onProgress) {
|
||||
$onProgress($fetched);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return $validIds;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user