PHP code example of utopia-php / schedule

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

utopia-php / schedule example snippets

use Utopia\Schedule\Source;
use Utopia\Schedule\Source\Entry;
use Utopia\Schedule\Source\Row;
use Utopia\Schedule\Trigger\Cron;

final class FunctionSchedules implements Source
{
    public function __construct(private Database $database) {}

    /** Every schedule that should be running, as cheap descriptors. */
    public function snapshot(): iterable
    {
        foreach ($this->database->find('schedules') as $document) {
            yield new Row(
                id: $document->getId(),
                version: $document->getAttribute('updatedAt'),
                data: $document,
                activeFrom: new DateTimeImmutable($document->getAttribute('updatedAt')),
            );
        }
    }

    /**
     * The expensive part — parsing, hydrating context — called only when a
     * row is new or its version changed.
     */
    public function make(Row $row): Entry
    {
        return new Entry(
            trigger: new Cron($row->data->getAttribute('schedule')),
            payload: ['projectId' => $row->data->getAttribute('projectId')],
        );
    }
}

use Utopia\Schedule\Occurrence;
use Utopia\Schedule\Scheduler;
use Utopia\Schedule\Store;

$scheduler = new Scheduler(
    source: new FunctionSchedules($database),
    store: new Store\Redis($redis),
    syncSeconds: 10,
);

// The handler is called once per due moment, with everything due at that
// moment. How that work runs is its choice: one round trip for the lot, a
// coroutine each, or a plain loop.
$scheduler->run(function (array $occurrences) use ($queue): void {
    foreach ($occurrences as $occurrence) {
        // $occurrence->id      — the row's identity
        // $occurrence->due     — when the run was scheduled for
        // $occurrence->payload — whatever make() attached
        // $occurrence->key()   — stable identity for deduplication
        $queue->enqueue($occurrence->id, $occurrence->payload);
    }
});

use Utopia\Schedule\Changes;
use Utopia\Schedule\Source;

final class FunctionSchedules implements Source, Changes
{
    // snapshot() and make() as above.

    public function since(DateTimeImmutable $moment): iterable
    {
        foreach ($this->database->find('schedules', [Query::greaterThanEqual('updatedAt', $moment)]) as $document) {
            yield new Row(/* ..., */ active: $document->getAttribute('enabled'));
        }
    }
}

$scheduler->run(function (array $occurrences) use ($database): void {
    foreach ($occurrences as $occurrence) {
        try {
            $database->createDocument('jobs', new Document([
                '$id' => \substr(\md5($occurrence->key()), 0, 32), // deterministic
                'scheduleId' => $occurrence->id,
                'dueAt' => $occurrence->due->format('Y-m-d H:i:s'),
            ]));
        } catch (Duplicate) {
            // already created by an earlier delivery of this same run
        }
    }
});

use Utopia\Schedule\Trigger\Cron;
use Utopia\Schedule\Trigger\Shifted;

public function make(Row $row): Entry
{
    $expression = $row->data->getAttribute('schedule');

    return new Entry(
        trigger: new Shifted(new Cron($expression), \abs(\crc32($row->id)) % 60),
        payload: $row->data,
    );
}

$scheduler = new Scheduler(
    source: new FunctionSchedules($database),
    store: new Store\Redis($redis),
    tickSeconds: 60,
    syncSeconds: 10,
    leadSeconds: 60,
);

$scheduler->reconcile();

foreach ($scheduler->tick() as $occurrence) {
    $handle($occurrence);
}

$scheduler->commit();

use Utopia\Span\Span;

$scheduler->run(function (array $occurrences) use ($queue): void {
    foreach ($occurrences as $occurrence) {
        Span::init('schedule.enqueue');
        try {
            Span::add('schedule.id', $occurrence->id);
            Span::add('schedule.due', $occurrence->due->format('c'));
            $queue->enqueue($occurrence->id, $occurrence->payload);
        } finally {
            Span::current()?->finish();
        }
    }
});

bash
composer