Initial commit
This commit is contained in:
root
237
lib/Daemon/MailDaemon.php
Normal file
237
lib/Daemon/MailDaemon.php
Normal file
@@ -0,0 +1,237 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
/**
|
||||
* SPDX-FileCopyrightText: Sebastian Krupinski <krupinski01@gmail.com>
|
||||
* SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
*/
|
||||
|
||||
namespace KTXM\MailManager\Daemon;
|
||||
|
||||
use KTXM\MailManager\Manager;
|
||||
use Psr\Log\LoggerInterface;
|
||||
|
||||
/**
|
||||
* Mail Queue Daemon
|
||||
*
|
||||
* Long-running worker process for processing queued mail messages.
|
||||
* Supports graceful shutdown via signals and configurable batch processing.
|
||||
*
|
||||
* @since 2025.05.01
|
||||
*/
|
||||
class MailDaemon {
|
||||
|
||||
private bool $running = false;
|
||||
private bool $shutdown = false;
|
||||
private bool $reload = false;
|
||||
|
||||
/**
|
||||
* @param Manager $manager Mail manager
|
||||
* @param LoggerInterface $logger Logger
|
||||
* @param int $pollInterval Seconds between queue polls when idle
|
||||
* @param int $batchSize Messages to process per batch
|
||||
* @param int|null $maxMemory Maximum memory usage in bytes before restart
|
||||
* @param array<string>|null $tenants Specific tenants to process (null = all)
|
||||
*/
|
||||
public function __construct(
|
||||
private Manager $manager,
|
||||
private LoggerInterface $logger,
|
||||
private int $pollInterval = 5,
|
||||
private int $batchSize = 50,
|
||||
private ?int $maxMemory = null,
|
||||
private ?array $tenants = null,
|
||||
) {
|
||||
// Set default max memory to 128MB
|
||||
$this->maxMemory = $this->maxMemory ?? (128 * 1024 * 1024);
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the daemon main loop
|
||||
*
|
||||
* @since 2025.05.01
|
||||
*/
|
||||
public function run(): void {
|
||||
$this->running = true;
|
||||
$this->shutdown = false;
|
||||
|
||||
$this->setupSignalHandlers();
|
||||
|
||||
$this->logger->info('Mail daemon starting', [
|
||||
'pollInterval' => $this->pollInterval,
|
||||
'batchSize' => $this->batchSize,
|
||||
'maxMemory' => $this->formatBytes($this->maxMemory),
|
||||
'tenants' => $this->tenants ?? 'all',
|
||||
]);
|
||||
|
||||
$consecutiveEmpty = 0;
|
||||
|
||||
while (!$this->shutdown) {
|
||||
// Handle reload signal
|
||||
if ($this->reload) {
|
||||
$this->handleReload();
|
||||
$this->reload = false;
|
||||
}
|
||||
|
||||
// Check memory usage
|
||||
if ($this->isMemoryExceeded()) {
|
||||
$this->logger->warning('Memory limit exceeded, shutting down for restart', [
|
||||
'current' => $this->formatBytes(memory_get_usage(true)),
|
||||
'limit' => $this->formatBytes($this->maxMemory),
|
||||
]);
|
||||
break;
|
||||
}
|
||||
|
||||
// Process queues
|
||||
$processed = $this->processTenants();
|
||||
|
||||
if ($processed === 0) {
|
||||
$consecutiveEmpty++;
|
||||
// Exponential backoff up to poll interval
|
||||
$sleepTime = min($consecutiveEmpty, $this->pollInterval);
|
||||
$this->sleep($sleepTime);
|
||||
} else {
|
||||
$consecutiveEmpty = 0;
|
||||
}
|
||||
|
||||
// Dispatch pending signals
|
||||
pcntl_signal_dispatch();
|
||||
}
|
||||
|
||||
$this->logger->info('Mail daemon stopped');
|
||||
$this->running = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Request graceful shutdown
|
||||
*
|
||||
* @since 2025.05.01
|
||||
*/
|
||||
public function stop(): void {
|
||||
$this->shutdown = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if daemon is running
|
||||
*
|
||||
* @since 2025.05.01
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function isRunning(): bool {
|
||||
return $this->running;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process all tenants and return total messages processed
|
||||
*/
|
||||
private function processTenants(): int {
|
||||
$totalProcessed = 0;
|
||||
$tenants = $this->tenants ?? $this->discoverTenants();
|
||||
|
||||
foreach ($tenants as $tenantId) {
|
||||
if ($this->shutdown) {
|
||||
break;
|
||||
}
|
||||
|
||||
$result = $this->manager->queueProcess($tenantId, $this->batchSize);
|
||||
$totalProcessed += $result['processed'] + $result['failed'];
|
||||
|
||||
if ($result['processed'] > 0 || $result['failed'] > 0) {
|
||||
$this->logger->debug('Processed tenant queue', [
|
||||
'tenant' => $tenantId,
|
||||
'processed' => $result['processed'],
|
||||
'failed' => $result['failed'],
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
return $totalProcessed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Discover all tenants with mail queues
|
||||
*
|
||||
* @return array<string>
|
||||
*/
|
||||
private function discoverTenants(): array {
|
||||
// This would need to be implemented based on your tenant discovery mechanism
|
||||
// For now, return empty array - specific tenants should be configured
|
||||
return [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup signal handlers for graceful shutdown
|
||||
*/
|
||||
private function setupSignalHandlers(): void {
|
||||
if (!function_exists('pcntl_signal')) {
|
||||
$this->logger->warning('PCNTL extension not available, signal handling disabled');
|
||||
return;
|
||||
}
|
||||
|
||||
pcntl_signal(SIGTERM, function() {
|
||||
$this->logger->info('Received SIGTERM, initiating graceful shutdown');
|
||||
$this->shutdown = true;
|
||||
});
|
||||
|
||||
pcntl_signal(SIGINT, function() {
|
||||
$this->logger->info('Received SIGINT, initiating graceful shutdown');
|
||||
$this->shutdown = true;
|
||||
});
|
||||
|
||||
pcntl_signal(SIGHUP, function() {
|
||||
$this->logger->info('Received SIGHUP, will reload configuration');
|
||||
$this->reload = true;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle configuration reload
|
||||
*/
|
||||
private function handleReload(): void {
|
||||
$this->logger->info('Reloading configuration');
|
||||
// Configuration reload logic would go here
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if memory limit has been exceeded
|
||||
*/
|
||||
private function isMemoryExceeded(): bool {
|
||||
if ($this->maxMemory === null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return memory_get_usage(true) > $this->maxMemory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleep with signal dispatch
|
||||
*/
|
||||
private function sleep(int $seconds): void {
|
||||
for ($i = 0; $i < $seconds; $i++) {
|
||||
if ($this->shutdown) {
|
||||
return;
|
||||
}
|
||||
sleep(1);
|
||||
pcntl_signal_dispatch();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Format bytes to human readable string
|
||||
*/
|
||||
private function formatBytes(?int $bytes): string {
|
||||
if ($bytes === null) {
|
||||
return 'unlimited';
|
||||
}
|
||||
|
||||
$units = ['B', 'KB', 'MB', 'GB'];
|
||||
$i = 0;
|
||||
while ($bytes >= 1024 && $i < count($units) - 1) {
|
||||
$bytes /= 1024;
|
||||
$i++;
|
||||
}
|
||||
return round($bytes, 2) . ' ' . $units[$i];
|
||||
}
|
||||
|
||||
}
|
||||
246
lib/Daemon/MailQueueCli.php
Normal file
246
lib/Daemon/MailQueueCli.php
Normal file
@@ -0,0 +1,246 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
/**
|
||||
* SPDX-FileCopyrightText: Sebastian Krupinski <krupinski01@gmail.com>
|
||||
* SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
*/
|
||||
|
||||
namespace KTXM\MailManager\Daemon;
|
||||
|
||||
use KTXM\MailManager\Queue\JobStatus;
|
||||
use KTXM\MailManager\Queue\MailQueue;
|
||||
use Psr\Log\LoggerInterface;
|
||||
|
||||
/**
|
||||
* Mail Queue CLI
|
||||
*
|
||||
* Command-line interface for mail queue management operations.
|
||||
*
|
||||
* Usage:
|
||||
* php mail-queue.php list <tenant> [--status=pending]
|
||||
* php mail-queue.php retry <jobId>
|
||||
* php mail-queue.php retry-all <tenant> --status=failed
|
||||
* php mail-queue.php purge <tenant> --status=complete --older-than=7d
|
||||
* php mail-queue.php stats <tenant>
|
||||
*
|
||||
* @since 2025.05.01
|
||||
*/
|
||||
class MailQueueCli {
|
||||
|
||||
public function __construct(
|
||||
private MailQueue $queue,
|
||||
private LoggerInterface $logger,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Run CLI command
|
||||
*
|
||||
* @param array<string> $args Command line arguments
|
||||
*
|
||||
* @return int Exit code
|
||||
*/
|
||||
public function run(array $args): int {
|
||||
$command = $args[1] ?? 'help';
|
||||
|
||||
return match($command) {
|
||||
'list' => $this->commandList($args),
|
||||
'retry' => $this->commandRetry($args),
|
||||
'retry-all' => $this->commandRetryAll($args),
|
||||
'purge' => $this->commandPurge($args),
|
||||
'stats' => $this->commandStats($args),
|
||||
'help', '--help', '-h' => $this->commandHelp(),
|
||||
default => $this->commandHelp(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* List jobs in queue
|
||||
*/
|
||||
private function commandList(array $args): int {
|
||||
$tenantId = $args[2] ?? null;
|
||||
if ($tenantId === null) {
|
||||
echo "Error: tenant ID required\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
$status = $this->parseOption($args, 'status');
|
||||
$statusEnum = $status !== null ? JobStatus::tryFrom($status) : null;
|
||||
$limit = (int)($this->parseOption($args, 'limit') ?? 100);
|
||||
|
||||
$jobs = $this->queue->listJobs($tenantId, $statusEnum, $limit);
|
||||
|
||||
if (empty($jobs)) {
|
||||
echo "No jobs found\n";
|
||||
return 0;
|
||||
}
|
||||
|
||||
echo sprintf("%-36s %-12s %-8s %-20s %s\n",
|
||||
'JOB ID', 'STATUS', 'ATTEMPTS', 'CREATED', 'SUBJECT');
|
||||
echo str_repeat('-', 100) . "\n";
|
||||
|
||||
foreach ($jobs as $job) {
|
||||
$subject = substr($job->message->getSubject(), 0, 30);
|
||||
echo sprintf("%-36s %-12s %-8d %-20s %s\n",
|
||||
$job->id,
|
||||
$job->status->value,
|
||||
$job->attempts,
|
||||
$job->created?->format('Y-m-d H:i:s') ?? '-',
|
||||
$subject
|
||||
);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retry a specific job
|
||||
*/
|
||||
private function commandRetry(array $args): int {
|
||||
$jobId = $args[2] ?? null;
|
||||
if ($jobId === null) {
|
||||
echo "Error: job ID required\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
if ($this->queue->retry($jobId)) {
|
||||
echo "Job $jobId queued for retry\n";
|
||||
return 0;
|
||||
}
|
||||
|
||||
echo "Failed to retry job $jobId (not found or not failed)\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retry all failed jobs for a tenant
|
||||
*/
|
||||
private function commandRetryAll(array $args): int {
|
||||
$tenantId = $args[2] ?? null;
|
||||
if ($tenantId === null) {
|
||||
echo "Error: tenant ID required\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
$jobs = $this->queue->listJobs($tenantId, JobStatus::Failed);
|
||||
$retried = 0;
|
||||
|
||||
foreach ($jobs as $job) {
|
||||
if ($this->queue->retry($job->id)) {
|
||||
$retried++;
|
||||
}
|
||||
}
|
||||
|
||||
echo "Retried $retried jobs\n";
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Purge old jobs
|
||||
*/
|
||||
private function commandPurge(array $args): int {
|
||||
$tenantId = $args[2] ?? null;
|
||||
if ($tenantId === null) {
|
||||
echo "Error: tenant ID required\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
$status = $this->parseOption($args, 'status') ?? 'complete';
|
||||
$statusEnum = JobStatus::tryFrom($status);
|
||||
if ($statusEnum === null) {
|
||||
echo "Error: invalid status '$status'\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
$olderThan = $this->parseOption($args, 'older-than') ?? '7d';
|
||||
$seconds = $this->parseDuration($olderThan);
|
||||
|
||||
$purged = $this->queue->purge($tenantId, $statusEnum, $seconds);
|
||||
echo "Purged $purged jobs\n";
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Show queue statistics
|
||||
*/
|
||||
private function commandStats(array $args): int {
|
||||
$tenantId = $args[2] ?? null;
|
||||
if ($tenantId === null) {
|
||||
echo "Error: tenant ID required\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
$stats = $this->queue->stats($tenantId);
|
||||
|
||||
echo "Queue Statistics for $tenantId:\n";
|
||||
echo " Pending: {$stats['pending']}\n";
|
||||
echo " Processing: {$stats['processing']}\n";
|
||||
echo " Complete: {$stats['complete']}\n";
|
||||
echo " Failed: {$stats['failed']}\n";
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Show help message
|
||||
*/
|
||||
private function commandHelp(): int {
|
||||
echo <<<HELP
|
||||
Mail Queue CLI
|
||||
|
||||
Usage:
|
||||
mail-queue <command> [options]
|
||||
|
||||
Commands:
|
||||
list <tenant> List jobs in queue
|
||||
--status=<status> Filter by status (pending, processing, complete, failed)
|
||||
--limit=<n> Maximum jobs to show (default: 100)
|
||||
|
||||
retry <jobId> Retry a specific failed job
|
||||
|
||||
retry-all <tenant> Retry all failed jobs for a tenant
|
||||
|
||||
purge <tenant> Purge old jobs
|
||||
--status=<status> Status to purge (default: complete)
|
||||
--older-than=<duration> Age threshold (default: 7d, e.g., 1h, 30d)
|
||||
|
||||
stats <tenant> Show queue statistics
|
||||
|
||||
help Show this help message
|
||||
|
||||
HELP;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a command line option
|
||||
*/
|
||||
private function parseOption(array $args, string $name): ?string {
|
||||
foreach ($args as $arg) {
|
||||
if (str_starts_with($arg, "--$name=")) {
|
||||
return substr($arg, strlen("--$name="));
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a duration string to seconds
|
||||
*/
|
||||
private function parseDuration(string $duration): int {
|
||||
preg_match('/^(\d+)([smhd])?$/', $duration, $matches);
|
||||
$value = (int)($matches[1] ?? 0);
|
||||
$unit = $matches[2] ?? 's';
|
||||
|
||||
return match($unit) {
|
||||
's' => $value,
|
||||
'm' => $value * 60,
|
||||
'h' => $value * 3600,
|
||||
'd' => $value * 86400,
|
||||
default => $value,
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user