copy()->setTimezone($timezone); $cron = new CronExpression(VALID_CRON_STRINGS[$frequency] ?? $frequency); $scheduledFor = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); if (! $scheduledFor->gte($executionTime->copy()->subMinutes(self::CATCH_UP_WINDOW_MINUTES))) { return false; } $delivery = DB::transaction(function () use ($scheduleKey, $scheduledFor, $jobType, $resourceId, $payload): ?ScheduledJobDelivery { ScheduledJobState::query()->insertOrIgnore([ 'uuid' => new_public_id(), 'schedule_key' => $scheduleKey, 'created_at' => now(), 'updated_at' => now(), ]); $state = ScheduledJobState::query() ->where('schedule_key', $scheduleKey) ->lockForUpdate() ->firstOrFail(); if ($state->last_scheduled_for?->gte($scheduledFor)) { return null; } $state->update(['last_scheduled_for' => $scheduledFor->utc()]); return ScheduledJobDelivery::create([ 'schedule_key' => $scheduleKey, 'scheduled_for' => $scheduledFor->utc(), 'job_type' => $jobType, 'resource_id' => $resourceId, 'payload' => $payload, 'status' => 'pending', ]); }); if ($delivery === null) { return false; } return $this->publish($delivery); } public function recordSkipped( string $scheduleKey, string $frequency, string $timezone, ?Carbon $executionTime = null, ): bool { $executionTime = ($executionTime ?? Carbon::now())->copy()->setTimezone($timezone); $cron = new CronExpression(VALID_CRON_STRINGS[$frequency] ?? $frequency); $scheduledFor = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); if (! $scheduledFor->gte($executionTime->copy()->subMinutes(self::CATCH_UP_WINDOW_MINUTES))) { return false; } return DB::transaction(function () use ($scheduleKey, $scheduledFor): bool { ScheduledJobState::query()->insertOrIgnore([ 'uuid' => new_public_id(), 'schedule_key' => $scheduleKey, 'created_at' => now(), 'updated_at' => now(), ]); $state = ScheduledJobState::query() ->where('schedule_key', $scheduleKey) ->lockForUpdate() ->firstOrFail(); if ($state->last_scheduled_for?->gte($scheduledFor)) { return false; } $state->update(['last_scheduled_for' => $scheduledFor->utc()]); return true; }); } public function publishPending(): void { ScheduledJobDelivery::query() ->where('status', 'pending') ->orderBy('id') ->chunkById(100, function ($occurrences): void { foreach ($occurrences as $occurrence) { $this->publish($occurrence); } }); } public function deleteOldOccurrences(): void { ScheduledJobDelivery::query() ->where('status', 'claimed') ->where('updated_at', '<', now()->subDays(2)) ->update([ 'status' => 'failed', 'updated_at' => now(), ]); ScheduledJobDelivery::query() ->whereIn('status', ['failed', 'skipped']) ->where('created_at', '<', now()->subDays(30)) ->chunkById(100, function ($occurrences): void { ScheduledJobDelivery::query()->whereKey($occurrences->modelKeys())->delete(); }); } public function claim(string $uuid, string $claimToken): bool { $claimed = ScheduledJobDelivery::query() ->where('uuid', $uuid) ->whereIn('status', ['pending', 'enqueued']) ->update([ 'status' => 'claimed', 'claim_token' => $claimToken, 'started_at' => now(), 'updated_at' => now(), ]); if ($claimed === 1) { return true; } return ScheduledJobDelivery::query() ->where('uuid', $uuid) ->where('status', 'claimed') ->where('claim_token', $claimToken) ->exists(); } public function complete(string $uuid, string $claimToken): void { ScheduledJobDelivery::query() ->where('uuid', $uuid) ->where('claim_token', $claimToken) ->delete(); } public function fail(string $uuid, string $claimToken): void { ScheduledJobDelivery::query() ->where('uuid', $uuid) ->where('claim_token', $claimToken) ->update([ 'status' => 'failed', 'updated_at' => now(), ]); } private function publish(ScheduledJobDelivery $occurrence): bool { if ($occurrence->status !== 'pending') { return false; } $job = match ($occurrence->job_type) { 'scheduled-task' => ($task = ScheduledTask::find($occurrence->resource_id)) ? new ScheduledTaskJob($task, $occurrence->uuid) : null, 'database-backup' => ($backup = ScheduledDatabaseBackup::find($occurrence->resource_id)) ? new DatabaseBackupJob($backup, $occurrence->uuid) : null, 'volume-backup' => ($backup = ScheduledVolumeBackup::find($occurrence->resource_id)) ? new VolumeBackupJob($backup, $occurrence->uuid) : null, 'docker-cleanup' => ($server = Server::find($occurrence->resource_id)) ? new DockerCleanupJob( $server, false, data_get($occurrence->payload, 'delete_unused_volumes', false), data_get($occurrence->payload, 'delete_unused_networks', false), $occurrence->uuid, ) : null, default => null, }; if ($job === null) { ScheduledJobDelivery::query()->whereKey($occurrence->id)->update(['status' => 'skipped']); return false; } dispatch($job); ScheduledJobDelivery::query() ->whereKey($occurrence->id) ->where('status', 'pending') ->update([ 'status' => 'enqueued', 'enqueued_at' => now(), 'updated_at' => now(), ]); return true; } }