private function startExecution(int $timestamp, string $jobId): string
    {
        $start = new \DateTime('now', new \DateTimeZone('UTC'));
        $sendAt = new \DateTime("@{$timestamp}");
        $span = (int)$start->diff($sendAt)->format('%a');

        $input = ['job_id' => $jobId];

        /*
         * AWS Step Functions have a one year maximum execution time.
         * Rather than artificially limit our sendAt dates to 1 year
         * in the future we'll just restart the executions when we
         * encounter this scenario.
         */
        if ($span < 365) {
            $input['send_at'] = $sendAt->format(self::DATE_FORMAT);
        } else {
            // just a skosh behind 1 year to fly under the radar
            $start->modify('+364 days');
            $input['resend_at'] = [];

            do {
                $input['resend_at'][] = $start->format(self::DATE_FORMAT);
                $start->modify('+364 days');
            } while ($start < $sendAt);

            // the original sendAt is the last one to "resend"
            $input['resend_at'][] = $sendAt->format(self::DATE_FORMAT);
            $input['send_at'] = array_shift($input['resend_at']);
        }

        try {
            $result = $this->sfnClient->startExecution([
                'stateMachineArn' => $this->stateMachineArn,
                'input'           => json_encode($input),
            ]);

            return $result['executionArn'];
        } catch (\Throwable $t) {
            if ($t instanceof AwsException) {
                $errorName = $t->getAwsErrorCode() ?: ClassUtils::getShortName($t);
                switch ($errorName) {
                    case 'ExecutionLimitExceeded':
                        $code = Code::RESOURCE_EXHAUSTED;
                        break;

                    case 'ExecutionAlreadyExists':
                        $code = Code::ALREADY_EXISTS;
                        break;

                    case 'InvalidArn':
                    case 'InvalidExecutionInput':
                    case 'InvalidName':
                        $code = Code::INVALID_ARGUMENT;
                        break;

                    case 'StateMachineDoesNotExist':
                    case 'StateMachineDeleting':
                        $code = Code::NOT_FOUND;
                        break;

                    default:
                        $code = Code::UNAVAILABLE;
                        break;
                }
            } else {
                $errorName = ClassUtils::getShortName($t);
                $code = Code::INTERNAL;
            }

            throw new SchedulerOperationFailed(
                sprintf(
                    '%s while adding to state machine [%s] for job_id [%s].',
                    $errorName,
                    $this->stateMachineArn,
                    $jobId
                ),
                $code,
                $t
            );
        }
    }