aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Digitigrade/GlobalConfig.php12
-rw-r--r--Digitigrade/Job.php39
-rw-r--r--Digitigrade/Job/FetchObject.php25
-rw-r--r--Digitigrade/Job/UpdateInstanceInfo.php22
-rw-r--r--Digitigrade/JobQueue.php86
-rw-r--r--Digitigrade/Logger.php52
-rw-r--r--Digitigrade/Model.php2
-rw-r--r--Digitigrade/Model/Instance.php4
-rwxr-xr-xbin/worker13
-rw-r--r--config.ini8
10 files changed, 260 insertions, 3 deletions
diff --git a/Digitigrade/GlobalConfig.php b/Digitigrade/GlobalConfig.php
index f17b758..7fbfab3 100644
--- a/Digitigrade/GlobalConfig.php
+++ b/Digitigrade/GlobalConfig.php
@@ -41,4 +41,16 @@ class GlobalConfig extends Singleton {
public function getDbPassword(): string {
return $this->confData['database']['password'];
}
+
+ public function getQueueHost(): string {
+ return $this->confData['queue']['host'];
+ }
+
+ public function getQueuePort(): int {
+ return $this->confData['queue']['port'] ?? 11300;
+ }
+
+ public function getQueueChannel(): string {
+ return $this->confData['queue']['channel'] ?? 'digitigrade';
+ }
} \ No newline at end of file
diff --git a/Digitigrade/Job.php b/Digitigrade/Job.php
new file mode 100644
index 0000000..4fc799a
--- /dev/null
+++ b/Digitigrade/Job.php
@@ -0,0 +1,39 @@
+<?php
+namespace Digitigrade;
+
+abstract class Job implements \JsonSerializable {
+ public int $remainingTries;
+ public int $doneTries = 0;
+
+ public function jsonSerialize(): array {
+ $array = ['jobSubtype' => static::class];
+ foreach ((new \ReflectionObject($this))->getProperties() as $prop) {
+ assert($prop instanceof \ReflectionProperty); // for intellisense
+ $value = $prop->getValue($this);
+ $array[$prop->getName()] = is_object($value) ? json_encode($value) : $value;
+ }
+ return $array;
+ }
+
+ public static final function fromJson(string $data): ?self {
+ $obj = json_decode($data);
+ if (!property_exists($obj, 'jobSubtype')) {
+ return null;
+ }
+ return (new \JsonMapper())->map($obj, $obj->jobSubtype);
+ }
+
+ /**
+ * Runs the job. Returns nothing on success, or throws some error/exception.
+ * @return void
+ */
+ public abstract function run();
+
+ /**
+ * Submits the job to the global job queue with default options.
+ * @return void
+ */
+ public function submit() {
+ JobQueue::getInstance()->submitNormal($this);
+ }
+} \ No newline at end of file
diff --git a/Digitigrade/Job/FetchObject.php b/Digitigrade/Job/FetchObject.php
new file mode 100644
index 0000000..227e05b
--- /dev/null
+++ b/Digitigrade/Job/FetchObject.php
@@ -0,0 +1,25 @@
+<?php
+namespace Digitigrade\Job;
+
+use Digitigrade\Job;
+use Digitigrade\Logger;
+use Digitigrade\Model\FetchableModel;
+
+class FetchObject extends Job {
+ public string $uri;
+ public string $type;
+
+ public function __construct(string $type, string $uri) {
+ $this->type = $type;
+ $this->uri = $uri;
+ $this->remainingTries = 8;
+ }
+
+ public function run() {
+ assert(is_subclass_of($this->type, FetchableModel::class));
+ Logger::getInstance()->info("fetching $this->type at $this->uri");
+ if ($this->type::findByUri($this->uri, forceRefetch: true) == null) {
+ throw new \RuntimeException("remote object doesn't seem to exist");
+ }
+ }
+} \ No newline at end of file
diff --git a/Digitigrade/Job/UpdateInstanceInfo.php b/Digitigrade/Job/UpdateInstanceInfo.php
new file mode 100644
index 0000000..85c11f0
--- /dev/null
+++ b/Digitigrade/Job/UpdateInstanceInfo.php
@@ -0,0 +1,22 @@
+<?php
+namespace Digitigrade\Job;
+
+use Digitigrade\Job;
+use Digitigrade\Logger;
+use Digitigrade\Model\Instance;
+
+class UpdateInstanceInfo extends Job {
+ public string $domain;
+
+ public function __construct(string $domain) {
+ $this->domain = $domain;
+ $this->remainingTries = 3;
+ }
+
+ public function run() {
+ Logger::getInstance()->info("updating info for instance $this->domain");
+ if (Instance::findByDomain($this->domain, forceRefetch: true) == null) {
+ throw new \RuntimeException("remote object doesn't seem to exist");
+ }
+ }
+} \ No newline at end of file
diff --git a/Digitigrade/JobQueue.php b/Digitigrade/JobQueue.php
new file mode 100644
index 0000000..9f95584
--- /dev/null
+++ b/Digitigrade/JobQueue.php
@@ -0,0 +1,86 @@
+<?php
+namespace Digitigrade;
+
+use Pheanstalk\Contract\JobIdInterface;
+use Pheanstalk\Pheanstalk;
+use Pheanstalk\Values\TubeName;
+
+class JobQueue extends Singleton {
+ private Pheanstalk $conn;
+
+ private const PRIO_URGENT = 10;
+ private const PRIO_DELAYED = 20;
+ private const PRIO_RETRY = 30;
+ private const PRIO_NORMAL = 50;
+ private const PRIO_LAZY = 100;
+
+ private const TIMEOUT = 60; // seconds
+ private const RETRY_DELAY = 5; // seconds, to the power of the number of tries
+
+ public function __construct() {
+ $config = GlobalConfig::getInstance();
+ $this->conn = Pheanstalk::create($config->getQueueHost(), $config->getQueuePort());
+ $this->conn->useTube(new TubeName($config->getQueueChannel()));
+ }
+
+ public function subscribe() {
+ $this->conn->watch(new TubeName(GlobalConfig::getInstance()->getQueueChannel()));
+ }
+
+ private function submit(Job $job, int $priority, int $delay = 0): JobIdInterface {
+ return $this->conn->put(json_encode($job), $priority, $delay, self::TIMEOUT);
+ }
+
+ public function submitNormal(Job $job) {
+ $this->submit($job, self::PRIO_NORMAL);
+ }
+
+ public function submitUrgent(Job $job) {
+ $this->submit($job, self::PRIO_URGENT);
+ }
+
+ public function submitLazy(Job $job) {
+ $this->submit($job, self::PRIO_LAZY);
+ }
+
+ public function submitDelayed(Job $job, int $delaySecs) {
+ $this->submit($job, self::PRIO_NORMAL, $delaySecs);
+ }
+
+ public function submitDelayedUntil(Job $job, \DateTimeInterface $until) {
+ $delay = $until->getTimestamp() - (new \DateTimeImmutable())->getTimestamp();
+ if ($delay < 0)
+ throw new \InvalidArgumentException('"until" date is in the past!');
+ $this->submitDelayed($job, $delay);
+ }
+
+ public function runNext() {
+ $beanJob = $this->conn->reserve();
+ $job = Job::fromJson($beanJob->getData());
+ $job->remainingTries--;
+ $job->doneTries++;
+ $log = Logger::getInstance();
+ $log->info('Running job ' . $beanJob->getId());
+ try {
+ $job->run();
+ $this->conn->delete($beanJob);
+ $log->info('Job ' . $beanJob->getId() . ' completed successfully');
+ } catch (\Exception $e) {
+ if ($job->remainingTries <= 0) {
+ $this->conn->delete($beanJob);
+ $log->error(
+ 'Ran out of retries trying to run job ' . $beanJob->getId() . ': ' . $beanJob->getData()
+ . "\n" . $e->getMessage()
+ );
+ } else {
+ $this->conn->delete($beanJob);
+ $id = $this->submit($job, self::PRIO_RETRY, pow(self::RETRY_DELAY, $job->doneTries));
+ $log->warning(
+ 'A job of type ' . $job::class . ' (id ' . $beanJob->getId() . ') failed. '
+ . 'Resubmitted as ' . $id->getId() . '; '
+ . $job->remainingTries . ' retries remaining'
+ );
+ }
+ }
+ }
+} \ No newline at end of file
diff --git a/Digitigrade/Logger.php b/Digitigrade/Logger.php
new file mode 100644
index 0000000..2a87c21
--- /dev/null
+++ b/Digitigrade/Logger.php
@@ -0,0 +1,52 @@
+<?php
+namespace Digitigrade;
+
+class Logger extends Singleton {
+ // TODO: implement different logging methods
+
+ public const LEVEL_DEBUG = 0;
+ public const LEVEL_INFO = 1;
+ public const LEVEL_WARNING = 2;
+ public const LEVEL_ERROR = 3;
+
+ private const TIME_FORMAT = 'Y-m-d H:i:s.v';
+
+ private int $minLevel = self::LEVEL_INFO;
+
+
+ public function setMinimumLevel(int $level) {
+ $this->minLevel = $level;
+ }
+
+ private function log(string $message, int $level) {
+ if ($level < $this->minLevel)
+ return;
+ $timestamp = (new \DateTimeImmutable())->format(self::TIME_FORMAT);#
+ $levelName = match ($level) {
+ self::LEVEL_DEBUG => 'DBG',
+ self::LEVEL_INFO => 'INF',
+ self::LEVEL_WARNING => 'WRN',
+ self::LEVEL_ERROR => 'ERR',
+ default => '???'
+ };
+
+ // TODO: don't hardcode this method
+ error_log("[$timestamp] $levelName: $message");
+ }
+
+ public function debug(string $message) {
+ $this->log($message, self::LEVEL_DEBUG);
+ }
+
+ public function info(string $message) {
+ $this->log($message, self::LEVEL_INFO);
+ }
+
+ public function warning(string $message) {
+ $this->log($message, self::LEVEL_WARNING);
+ }
+
+ public function error(string $message) {
+ $this->log($message, self::LEVEL_ERROR);
+ }
+} \ No newline at end of file
diff --git a/Digitigrade/Model.php b/Digitigrade/Model.php
index ffc2157..84e6439 100644
--- a/Digitigrade/Model.php
+++ b/Digitigrade/Model.php
@@ -179,7 +179,7 @@ abstract class Model {
$shouldUpdate = $whereClause != null;
$stmt = $db->prepare(
$shouldUpdate
- ? "UPDATE $tableName SET ($columns) = ($parameters) WHERE $whereClause"
+ ? "UPDATE $tableName SET ($columns) = ROW($parameters) WHERE $whereClause"
: "INSERT INTO $tableName ($columns) VALUES ($parameters)"
);
diff --git a/Digitigrade/Model/Instance.php b/Digitigrade/Model/Instance.php
index 2924f71..de3ff9d 100644
--- a/Digitigrade/Model/Instance.php
+++ b/Digitigrade/Model/Instance.php
@@ -24,8 +24,8 @@ class Instance extends FetchableModel {
return null;
}
- public static function findByDomain(string $domain): ?self {
- return self::findByUri("https://$domain/.well-known/pawpub-instance");
+ public static function findByDomain(string $domain, bool $autoSave = true, bool $forceRefetch = false): ?self {
+ return self::findByUri("https://$domain/.well-known/pawpub-instance", $autoSave, $forceRefetch);
}
public function hydrate() {
diff --git a/bin/worker b/bin/worker
new file mode 100755
index 0000000..b6c8590
--- /dev/null
+++ b/bin/worker
@@ -0,0 +1,13 @@
+#!/usr/bin/env php
+<?php
+
+require __DIR__ . '/../vendor/autoload.php';
+foreach (glob(__DIR__ . '/../misc/*.php') as $filename) {
+ require $filename;
+}
+
+$queue = Digitigrade\JobQueue::getInstance();
+$queue->subscribe();
+while (true) {
+ $queue->runNext();
+} \ No newline at end of file
diff --git a/config.ini b/config.ini
index 16b9f06..dc95150 100644
--- a/config.ini
+++ b/config.ini
@@ -11,6 +11,14 @@ username = digitigrade
password = digitigrade
database = digitigrade
+[queue]
+; host (required) and port (optional, default = 11300)
+; connection details for a beanstalkd job queue server
+host = localhost
+port = 11300
+; tube name to send all jobs down (optional, default = 'digitigrade')
+channel = digitigrade
+
[debug]
; whether to send full tracebacks on internal server errors (optional,
; default = false)
generated by cgit v1.3 (git 2.53.0) at 2026-07-21 22:07:39 +0000