TenantAtlas/app/Jobs/BulkPolicyExportJob.php
Ahmed Darrazi eef9618889 feat: configurable bulk ops polling + chunking
- Add tenantpilot.bulk_operations config (chunk size, poll interval)
- Use config chunk size across all bulk jobs
- Make progress widget polling interval configurable
- Document settings in README + feature quickstart; mark tasks done
2025-12-25 03:18:12 +01:00

220 lines
8.4 KiB
PHP

<?php
namespace App\Jobs;
use App\Models\BackupItem;
use App\Models\BackupSet;
use App\Models\BulkOperationRun;
use App\Models\Policy;
use App\Services\BulkOperationService;
use Filament\Notifications\Notification;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Throwable;
class BulkPolicyExportJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public function __construct(
public int $bulkRunId,
public string $backupName,
public ?string $backupDescription = null
) {}
public function handle(BulkOperationService $service): void
{
$run = BulkOperationRun::with('user')->find($this->bulkRunId);
if (! $run || $run->status !== 'pending') {
return;
}
$service->start($run);
try {
// Create Backup Set
$backupSet = BackupSet::create([
'tenant_id' => $run->tenant_id,
'name' => $this->backupName,
// 'description' => $this->backupDescription, // Not in schema
'status' => 'completed',
'created_by' => $run->user?->name ?? (string) $run->user_id, // Schema has created_by string
'item_count' => count($run->item_ids),
'completed_at' => now(),
]);
$itemCount = 0;
$succeeded = 0;
$failed = 0;
$failures = [];
$chunkSize = max(1, (int) config('tenantpilot.bulk_operations.chunk_size', 10));
$totalItems = $run->total_items ?: count($run->item_ids ?? []);
$failureThreshold = (int) floor($totalItems / 2);
foreach ($run->item_ids as $policyId) {
$itemCount++;
try {
$policy = Policy::find($policyId);
if (! $policy) {
$service->recordFailure($run, (string) $policyId, 'Policy not found');
$failed++;
$failures[] = [
'item_id' => (string) $policyId,
'reason' => 'Policy not found',
'timestamp' => now()->toIso8601String(),
];
if ($failed > $failureThreshold) {
$backupSet->update(['status' => 'failed']);
$service->abort($run, 'Circuit breaker: more than 50% of items failed.');
if ($run->user) {
Notification::make()
->title('Bulk Export Aborted')
->body('Circuit breaker triggered: too many failures (>50%).')
->icon('heroicon-o-exclamation-triangle')
->danger()
->sendToDatabase($run->user)
->send();
}
return;
}
continue;
}
// Get latest version for snapshot
$latestVersion = $policy->versions()->orderByDesc('captured_at')->first();
if (! $latestVersion) {
$service->recordFailure($run, (string) $policyId, 'No versions available for policy');
$failed++;
$failures[] = [
'item_id' => (string) $policyId,
'reason' => 'No versions available for policy',
'timestamp' => now()->toIso8601String(),
];
if ($failed > $failureThreshold) {
$backupSet->update(['status' => 'failed']);
$service->abort($run, 'Circuit breaker: more than 50% of items failed.');
if ($run->user) {
Notification::make()
->title('Bulk Export Aborted')
->body('Circuit breaker triggered: too many failures (>50%).')
->icon('heroicon-o-exclamation-triangle')
->danger()
->sendToDatabase($run->user)
->send();
}
return;
}
continue;
}
// Create Backup Item
BackupItem::create([
'tenant_id' => $run->tenant_id,
'backup_set_id' => $backupSet->id,
'policy_id' => $policy->id,
'policy_identifier' => $policy->external_id, // Added
'policy_type' => $policy->policy_type,
'platform' => $policy->platform ?? null, // Added
// 'display_name' => $policy->display_name, // Not in schema, maybe in metadata?
'payload' => $latestVersion->snapshot, // Mapped to payload
'metadata' => [
'display_name' => $policy->display_name, // Stored in metadata
'version_captured_at' => $latestVersion->captured_at->toIso8601String(),
],
]);
$service->recordSuccess($run);
$succeeded++;
} catch (Throwable $e) {
$service->recordFailure($run, (string) $policyId, $e->getMessage());
$failed++;
$failures[] = [
'item_id' => (string) $policyId,
'reason' => $e->getMessage(),
'timestamp' => now()->toIso8601String(),
];
if ($failed > $failureThreshold) {
$backupSet->update(['status' => 'failed']);
$service->abort($run, 'Circuit breaker: more than 50% of items failed.');
if ($run->user) {
Notification::make()
->title('Bulk Export Aborted')
->body('Circuit breaker triggered: too many failures (>50%).')
->icon('heroicon-o-exclamation-triangle')
->danger()
->sendToDatabase($run->user)
->send();
}
return;
}
}
// Refresh the run from database every 10 items to avoid stale data
if ($itemCount % $chunkSize === 0) {
$run->refresh();
}
}
// Update BackupSet item count (if denormalized) or just leave it
// Assuming BackupSet might need an item count or status update
$service->complete($run);
if ($succeeded > 0 || $failed > 0) {
$message = "Successfully exported {$succeeded} policies to backup '{$this->backupName}'";
if ($failed > 0) {
$message .= " ({$failed} failed)";
}
$message .= '.';
Notification::make()
->title('Bulk Export Completed')
->body($message)
->icon('heroicon-o-check-circle')
->success()
->sendToDatabase($run->user)
->send();
}
} catch (Throwable $e) {
$service->fail($run, $e->getMessage());
// Reload run with user relationship
$run->refresh();
$run->load('user');
if ($run->user) {
Notification::make()
->title('Bulk Export Failed')
->body($e->getMessage())
->icon('heroicon-o-x-circle')
->danger()
->sendToDatabase($run->user)
->send();
}
throw $e;
}
}
}