# Sistema de Queue - Framework OfficeGest

## Visão Geral do Sistema de Queue

O sistema de Queue do framework OfficeGest é uma **solução empresarial completa** para processamento assíncrono de tarefas, projetada para alta disponibilidade, escalabilidade e robustez. O sistema oferece:

- **Múltiplos drivers**: Redis e RabbitMQ com suporte nativo
- **Processamento em lote (Batches)**: Jobs agrupados com callbacks e controlo de falhas
- **Chains de Jobs**: Sequências de jobs com dependências
- **Dead Letter Queues**: Recuperação automática de jobs falhados
- **Monitoring avançado**: Métricas em tempo real e persistência
- **Distributed Locking**: Prevenção de execução concorrente
- **Resiliência**: Circuit breakers, retry policies e backoff strategies
- **Multi-tenancy**: Isolamento por domínio/empresa

Este é um dos subsistemas mais complexos e críticos do framework moderno.

## Arquitectura e Componentes Core

### Hierarquia Principal

**QueueManager** (`QueueManager.php`)
- **Gestor central** de todo o sistema de queues
- **Factory de drivers** (Redis, RabbitMQ, Database)
- **Gestão de transações** e rollback automático
- **Multi-tenancy** via JobTenantAware trait
- **Lifecycle management** de drivers

**QueueConfig** (`QueueConfig.php`)
- **Configuração centralizada** do sistema
- **Driver default** e configurações específicas por driver
- **Timeouts, retries** e configurações de comportamento
- **Dead letter queue** e monitoring settings

**QueueServiceProvider** (`QueueServiceProvider.php`)
- **126 linhas de registos** no container DI
- **State managers factories** para batch/chain
- **Bindings de interfaces** para implementações concretas
- **Configuração de todos os componentes** do ecosistema

### Interfaces e Contratos

**QueueDriverInterface** - Contrato base para drivers de queue:
```php
interface QueueDriverInterface
{
    public function push(JobInterface $job, string $queue = null): string;
    public function pop(string $queue = null): ?JobInterface;
    public function later(DateTime $delay, JobInterface $job, string $queue = null): string;
    public function size(string $queue = null): int;
}
```

**QueueManagerInterface** - Contrato para o gestor principal:
```php
interface QueueManagerInterface
{
    public function driver(?string $name = null): QueueDriverInterface;
    public function dispatch(JobInterface $job): string;
    public function batch(array $jobs): BatchInterface;
    public function chain(array $jobs): ChainInterface;
}
```

**JobInterface** - Contrato para todos os jobs:
```php
interface JobInterface
{
    public function handle(): void;
    public function failed(Throwable $exception): void;
    public function getQueue(): string;
    public function getTries(): int;
    public function getTimeout(): int;
}
```

## Sistema de Jobs

### Job Base (`Job.php`)

Classe abstracta que implementa o padrão base para todos os jobs:

**Características Principais:**
```php
abstract class Job implements JobInterface
{
    use Dispatchable;       // Métodos dispatch(), dispatchSync()
    use JobTenantAware;     // Multi-tenancy por domínio/empresa
    use Lockable;           // Distributed locking automático
    use Schedulable;        // Agendamento de execução

    protected int $tries = 3;           // Tentativas por defeito
    protected int $timeout = 600;       // 10 minutos timeout
    protected array $backoff = [5];     // Backoff strategy
    protected string $queue = 'default'; // Queue por defeito
}
```

**Lifecycle do Job:**
1. **Serialização**: Via SerializerInterface (JSON por defeito)
2. **Dispatch**: Para driver específico (Redis/RabbitMQ)
3. **Locking**: Distributed lock se usar @RequiresLock
4. **Execução**: Método handle() com timeout control
5. **Cleanup**: Release de locks e recursos

**Especializações de Jobs:**

**(EM IMPLEMENTAÇÃO!)** **BatchedJob** (`BatchedJob.php`):
- Job que faz parte de um batch
- **Callback automático** quando batch completa
- **Propagação de falhas** configurável

**(EM IMPLEMENTAÇÃO!)** **ChainedJob** (`ChainedJob.php`):
- Job sequencial em chain
- **Auto-dispatch** do próximo job na sequência
- **Gestão de estado** via ChainStateManager

**(EM IMPLEMENTAÇÃO!)** **EncryptedJob** (`EncryptedJob.php`):
- Job com payload encriptado
- **Desencriptação automática** antes da execução
- **Segurança** para dados sensíveis

### Atributos de Jobs

**@RequiresLock** (`Attributes/RequiresLock.php`):
```php
#[RequiresLock(scope: LockScope::GLOBAL, strategy: LockStrategy::WAIT)]
class UniqueProcessJob extends Job { }
```

**@RequiresSchedule** (`Attributes/RequiresSchedule.php`):
```php
#[RequiresSchedule(validators: [TimeWindowValidator::class])]
class ScheduledReportJob extends Job { }
```

## Sistema de Batches **(EM IMPLEMENTAÇÃO!)**

### Batch Core (`Batch.php`)

Sistema completo para processamento de jobs em lote:

**Funcionalidades:**
```php
$batch = Bus::batch([
    new ProcessPayrollJob($month),
    new GenerateReportJob($month),
    new SendNotificationJob($recipients)
])
->then(function (Batch $batch) {
    // Executado quando todos os jobs completam
    loggerBatch('info', 'Payroll batch completed', ['batch_id' => $batch->id]);
})
->catch(function (Batch $batch, Throwable $e) {
    // Executado se algum job falha
    loggerBatch('error', 'Payroll batch failed', ['error' => $e->getMessage()]);
})
->finally(function (Batch $batch) {
    // Executado sempre no final
    Cache::forget("batch_status_{$batch->id}");
})
->onConnection('redis')
->onQueue('payroll')
->dispatch();
```

**Estado e Callbacks:**
- **BatchCompleted**: Todos os jobs executaram com sucesso
- **BatchFailed**: Pelo menos um job falhou
- **BatchFinished**: Batch terminou (sucesso ou falha)

**Gestão de Estado:**
- **RedisBatchStateManager**: Estado persistido no Redis
- **RabbitMQBatchStateManager**: Estado via RabbitMQ headers
- **Atomic operations**: Garantia de consistência

### Batch State Managers

**RedisBatchStateManager** (`Drivers/Redis/RedisBatchStateManager.php`):
- **Lua scripts** para operações atómicas
- **TTL automático** para cleanup
- **Contadores thread-safe** para jobs completed/failed

**RabbitMQBatchStateManager** (`Drivers/RabbitMQ/RabbitMQBatchStateManager.php`):
- **Message headers** para tracking
- **Exchange routing** para callbacks
- **DLX integration** para jobs falhados

## Sistema de Chains **(EM IMPLEMENTAÇÃO!)**

### Chain Core (`Chain.php`)

Sistema para sequences de jobs com dependências:

**Exemplo de Chain:**
```php
$chain = Chain::create([
    new ValidateDataJob($data),
    new ProcessDataJob($data),
    new GenerateReportJob($data),
    new SendReportJob($recipients)
])
->then(function () {
    loggerBatch('info', 'Data processing chain completed');
})
->catch(function (Throwable $e) {
    loggerBatch('error', 'Chain failed', ['error' => $e->getMessage()]);
})
->dispatch();
```

**Características:**
- **Sequential execution**: Um job por vez
- **Failure propagation**: Falha para a chain completa
- **Shared context**: Dados partilhados entre jobs
- **Conditional branching**: Jobs condicionais baseados em resultados

**State Management:**
- **RedisChainStateManager**: Estado no Redis com timeouts
- **RabbitMQChainStateManager**: Estado distribuído via messaging

## Drivers de Queue

### **(EM IMPLEMENTAÇÃO!)**  Redis Driver (`Drivers/Redis/RedisDriver.php`)

**Características:**
- **Lua scripts** para operações atómicas
- **Delayed jobs** via sorted sets
- **Circuit breaker** para resiliência
- **Retry policies** configuráveis
- **Metrics collection** automática

**Estruturas Redis:**
```
queue:default              # Lista principal (LPUSH/BRPOP)
queue:default:reserved     # Jobs em execução
queue:default:delayed      # Sorted set para jobs agendados
queue:jobs:{id}           # Detalhes do job individual
```

**Operações Principais:**
```php
// Push job para queue
$this->redis->lpush("queue:{$queue}", $serializedJob);

// Pop job (blocking)
$job = $this->redis->brpop("queue:{$queue}", $timeout);

// Delayed job
$this->redis->zadd("queue:{$queue}:delayed", $timestamp, $serializedJob);
```

### RabbitMQ Driver (`Drivers/RabbitMQ/RabbitMQDriver.php`)

**Características:**
- **AMQP protocol** nativo
- **Exchange routing** configurável
- **TTL e DLX** automáticos
- **Prefetch control** para workers
- **Connection recovery** automático

**Estruturas RabbitMQ:**
```
Exchange: og.direct
Queue: og.queue.default
DLX: og.deadletter
Routing Key: queue.default
```

**Configuração:**
```php
'rabbitmq' => [
    'host' => env('RABBITMQ_HOST', 'localhost'),
    'port' => env('RABBITMQ_PORT', 5672),
    'username' => env('RABBITMQ_USERNAME', 'guest'),
    'password' => env('RABBITMQ_PASSWORD', 'guest'),
    'vhost' => env('RABBITMQ_VHOST', '/'),
    'exchange' => 'og.direct',
    'queue' => 'og.queue.default',
]
```

### Driver Factory (`Drivers/QueueDriverFactory.php`)

Factory responsável por criar drivers baseado na configuração:

```php
public function create(?string $driver = null): QueueDriverInterface
{
    $driver ??= config('queue.default', 'redis');

    return match ($driver) {
        'redis' => new RedisDriver($this->container->get('redis.manager')),
        'rabbitmq' => new RabbitMQDriver($this->container->get(RabbitMQConfig::class)),
        'database' => new DatabaseDriver($this->container->get('database')),
        default => throw new DriverNotFoundException("Queue driver [{$driver}] not found")
    };
}
```

## Sistema de Workers

### Worker Principal (`Worker.php`)

O Worker é responsável por **processar jobs continuamente**:

**Arquitectura do Worker:**
```php
public function work(
    string $connectionName,
    string $queue = null,
    array $options = []
): int {
    $this->listenForSignals();          // SIGTERM, SIGUSR2 handling
    $this->checkMemoryLimit();          // Memory leak prevention

    while ($this->isRunning) {
        $job = $this->getNextJob($queue);   // Pop from queue

        if ($job) {
            $this->processJob($job);        // Execute with monitoring
        } else {
            $this->sleep($options['sleep']);  // Sleep when empty
        }

        $this->checkForRestart();          // Graceful restart
    }
}
```

**Características Avançadas:**
- **Memory management**: Auto-restart quando limite excedido
- **Signal handling**: Graceful shutdown via SIGTERM
- **Distributed locking**: Prevenção de jobs duplicados
- **Retry logic**: Backoff exponencial configurável
- **Monitoring integration**: Métricas em tempo real

**Processamento de Job:**
```php
protected function processJob(JobInterface $job): void
{
    $this->fireJobEvent('processing', $job);

    // Acquire distributed lock if required
    if ($this->requiresLock($job)) {
        $lockAcquired = $this->jobLock->acquire($job->getLockKey());
        if (!$lockAcquired) {
            return $this->handleLockFailure($job);
        }
    }

    try {
        $this->runJob($job);               // Execute handle() method
        $this->fireJobEvent('processed', $job);
        $this->handleJobCompletion($job);
    } catch (Throwable $e) {
        $this->handleJobFailure($job, $e);
    } finally {
        $this->releaseLock($job);          // Always release lock
    }
}
```

### Comandos de Console

**QueueWorkCommand** (`Console/QueueWorkCommand.php`):
```bash
# Start worker
php og queue:work redis --queue=default --tries=3 --timeout=60

# Worker com configurações específicas
php og queue:work rabbitmq --queue=high-priority --memory=512 --sleep=3
```

**QueueListenCommand** (`Console/QueueListenCommand.php`):
```bash
# Listen para jobs (reloads automaticamente)
php og queue:listen redis --queue=default
```

**QueueMonitorCommand** (`Console/QueueMonitorCommand.php`):
```bash
# Monitoring em tempo real
php og queue:monitor --interval=5
```

**Outros Comandos:**
- `queue:restart`: Graceful restart de workers
- `queue:stats`: Estatísticas detalhadas das queues
- `failed:list`: Listar jobs falhados
- `failed:retry`: Retry jobs específicos
- `failed:purge`: Limpar jobs falhados

## Sistema de Monitoring

### QueueMonitor (`Monitoring/QueueMonitor.php`)

Sistema completo de monitoring e métricas:

**Métricas Capturadas:**
```php
public function recordJobExecution(string $jobId, float $duration): void
{
    $metric = [
        'duration' => $duration,
        'memory' => memory_get_peak_usage(true),
        'executed_at' => time(),
        'queue' => $job->getQueue(),
        'connection' => $job->getConnection()
    ];

    $this->metrics[$jobId] = $metric;
    $this->storage?->storeMetric($jobId, $metric);
}
```

**Storage Backends:**
- **RedisMonitorStorage**: Métricas persistidas no Redis
- **DatabaseMonitorStorage**: Métricas em base de dados
- **ElasticsearchStorage**: Para análise avançada (futuro)

### Dashboard e Alertas

**Métricas Disponíveis:**
- Jobs executados por minuto/hora
- Duração média de execução
- Taxa de falhas por queue
- Memory usage por worker
- Queue size em tempo real

**Alertas Automáticos:**
- Queue size acima do threshold
- Taxa de falhas elevada
- Memory leaks em workers
- Workers inativos

## Dead Letter Queues (DLQ)

### DatabaseDeadLetterQueueService (`DeadLetter/DatabaseDeadLetterQueueService.php`)

Sistema robusto para gestão de jobs falhados:

**Funcionalidades:**
```php
public function store(
    string $connection,
    string $queue,
    string $payload,
    string $originalPayload,
    Throwable $exception
): bool {
    $deadLetterJobData = DeadLetterJobData::fromException(
        connection: $connection,
        queue: $queue,
        payload: $payload,
        originalPayload: $originalPayload,
        exception: $exception,
        jobContext: $this->extractJobContext($payload),
        metadata: $this->buildMetadata()
    );

    return $this->repository->store($deadLetterJobData);
}
```

**Informações Capturadas:**
- **Job original** e payload modificado
- **Stack trace** completa do erro
- **Contexto de execução** (worker, tenant, timing)
- **Tentativas anteriores** e backoff history
- **Environment info** para debug

**Recovery e Retry:**
```php
// Retry job específico
$dlq->retry($jobId, $maxRetries = 3);

// Retry todos os jobs de uma queue
$dlq->retryAll($queue, $since = null);

// Purge jobs antigos
$dlq->purge($olderThan = '7 days');
```

### Gestão Automática

**Auto-retry Logic:**
- **Exponential backoff**: 5s, 25s, 125s, 625s...
- **Max attempts**: Configurável por job
- **Circuit breaker**: Para evitar retry loops
- **Batch retry**: Para jobs relacionados

## Sistema de Locking

### Distributed Locking

**RedisDistributedJobLock** (`Drivers/Redis/RedisDistributedJobLock.php`):
```php
public function acquire(string $lockKey, int $ttl = 60): bool
{
    $script = "
        if redis.call('SET', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then
            return 1
        else
            return 0
        end
    ";

    return (bool) $this->redis->eval($script, [$lockKey], [uniqid(), $ttl]);
}
```

**Características:**
- **Atomic operations** via Lua scripts
- **TTL automático** para lock expiry
- **Deadlock prevention** com timeouts
- **Lock extension** para jobs longos

**Estratégias de Lock:**

**LockStrategy::WAIT**:
```php
#[RequiresLock(strategy: LockStrategy::WAIT, timeout: 30)]
class ImportDataJob extends Job
{
    // Aguarda até 30s pelo lock
}
```

**LockStrategy::SKIP**:
```php
#[RequiresLock(strategy: LockStrategy::SKIP)]
class OptionalCleanupJob extends Job
{
    // Skip se lock não disponível
}
```

**LockStrategy::FAIL**:
```php
#[RequiresLock(strategy: LockStrategy::FAIL)]
class CriticalSyncJob extends Job
{
    // Falha imediatamente se lock não disponível
}
```

## Resiliência e Recuperação

### Circuit Breaker Pattern

**CircuitBreaker** (`Resilience/CircuitBreaker.php`):
```php
class CircuitBreaker
{
    private CircuitState $state = CircuitState::CLOSED;
    private int $failureCount = 0;
    private ?DateTime $lastFailureTime = null;

    public function call(callable $operation): mixed
    {
        if ($this->state === CircuitState::OPEN) {
            if ($this->shouldAttemptReset()) {
                $this->state = CircuitState::HALF_OPEN;
            } else {
                throw new CircuitBreakerOpenException();
            }
        }

        try {
            $result = $operation();
            $this->onSuccess();
            return $result;
        } catch (Throwable $e) {
            $this->onFailure();
            throw $e;
        }
    }
}
```

**Estados do Circuit:**
- **CLOSED**: Funcionamento normal
- **OPEN**: Bloqueio temporário após falhas
- **HALF_OPEN**: Teste de recuperação

### Retry Policies

**RetryPolicy** (`Resilience/RetryPolicy.php`):
```php
$retryPolicy = new RetryPolicy(
    new RetryConfig(
        maxAttempts: 5,
        baseDelayMs: 100,
        strategy: BackoffStrategy::EXPONENTIAL,
        retryableExceptions: [
            ConnectionException::class,
            TimeoutException::class
        ]
    )
);
```

**Backoff Strategies:**
- **EXPONENTIAL**: 100ms, 200ms, 400ms, 800ms...
- **LINEAR**: 100ms, 200ms, 300ms, 400ms...
- **FIXED**: 100ms, 100ms, 100ms, 100ms...
- **FIBONACCI**: 100ms, 100ms, 200ms, 300ms, 500ms...

## Multi-Tenancy

### JobTenantAware Trait

Sistema completo de isolamento por tenant:

```php
trait JobTenantAware
{
    protected ?string $domain = null;
    protected ?int $companyId = null;
    protected ?int $ownerId = null;

    public function setTenant(string $domain, int $companyId, int $ownerId): self
    {
        $this->domain = $domain;
        $this->companyId = $companyId;
        $this->ownerId = $ownerId;

        return $this;
    }

    protected function switchToJobTenant(): void
    {
        if ($this->domain) {
            // Switch database connection
            config(['database.default' => "tenant_{$this->domain}"]);

            // Set global tenant context
            app('tenant')->setCurrent($this->domain, $this->companyId);
        }
    }
}
```

**Isolamento Garantido:**
- **Database separation** por domain
- **Queue separation** via prefixes
- **File storage isolation** por tenant
- **Cache namespacing** automático

## Configuração e Deployment

### Configuração Principal

**Ficheiro:** `Modules/Common/Config/queue.php`

```php
return [
    'default' => env('QUEUE_CONNECTION', 'redis'),
    'queue' => env('QUEUE_NAME', 'default'),
    'job_timeout' => 600,
    'max_retries' => 3,

    'drivers' => [
        'redis' => [
            'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
            'prefix' => env('QUEUE_PREFIX', 'queue:'),
            'retry_after' => 90,
        ],

        'rabbitmq' => [
            'host' => env('RABBITMQ_HOST', 'localhost'),
            'port' => env('RABBITMQ_PORT', 5672),
            'username' => env('RABBITMQ_USERNAME', 'guest'),
            'password' => env('RABBITMQ_PASSWORD', 'guest'),
            'vhost' => env('RABBITMQ_VHOST', '/'),
            'exchange' => 'og.direct',
            'prefetch_count' => 1,
        ]
    ],

    'failed' => [
        'driver' => 'database',
        'table' => 'failed_jobs',
    ],

    'monitoring' => [
        'driver' => 'redis',
        'persist' => true,
        'retention' => '7 days',
    ]
];
```

### Deployment e Escalabilidade

**Worker Deployment:**
```bash
# Supervisord configuration
[program:queue-worker]
command=php /path/to/og queue:work redis --tries=3 --timeout=60
process_name=%(program_name)s_%(process_num)02d
numprocs=4
autostart=true
autorestart=true
user=www-data
redirect_stderr=true
stdout_logfile=/var/log/queue-worker.log
```

**Load Balancing:**
- **Multiple workers** por server
- **Queue sharding** por prioridade
- **Horizontal scaling** via container orchestration
- **Auto-scaling** baseado em queue size

**High Availability:**
- **Redis Cluster** para queue storage
- **RabbitMQ clustering** com mirror queues
- **Database replication** para DLQ
- **Multi-region deployment** com failover

## Performance e Optimização

### Métricas de Performance

**Benchmarks Típicos:**
- **Redis**: ~10,000 jobs/segundo
- **RabbitMQ**: ~5,000 jobs/segundo
- **Latency**: <1ms para jobs simples
- **Memory**: ~32MB por worker

**Optimizações:**
- **Batch dispatching** para throughput
- **Connection pooling** para drivers
- **Serialization optimization** (msgpack option)
- **Lua scripts** para operações atómicas

### Monitoring e Alertas

**Dashboards Disponíveis:**
- Queue size por connection/queue
- Throughput (jobs/min)
- Average execution time
- Failure rate
- Worker health status

**Alertas Automáticos:**
- Queue size > threshold
- Worker memory > 90%
- Failure rate > 5%
- DLQ jobs > 100

Este sistema de Queue representa uma solução **enterprise-grade completa** para processamento assíncrono, oferecendo todas as funcionalidades necessárias para um ERP moderno e escalável como o OfficeGest.
