Flycro/temporalio-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Init

Browse Source
This commit is contained in:
Flycro
2026-05-09 01:18:51 +02:00
parent 7116ee4619
commit 959970c150
132 changed files with 21310 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
app/Temporal/ProductImport/BatchImportWorkflow.php Normal file
View File

@@ -0,0 +1,35 @@
<?php
namespace App\Temporal\ProductImport;
use Carbon\CarbonInterval;
use Temporal\Activity\ActivityOptions;
use Temporal\Common\RetryOptions;
use Temporal\Workflow;
class BatchImportWorkflow implements BatchImportWorkflowInterface
{
/** @var ProductImportActivityInterface */
private $activityStub;
public function __construct()
{
$this->activityStub = Workflow::newActivityStub(
ProductImportActivityInterface::class,
ActivityOptions::new()
->withStartToCloseTimeout(CarbonInterval::minutes(5))
->withHeartbeatTimeout(CarbonInterval::seconds(30))
->withRetryOptions(
RetryOptions::new()
->withMaximumAttempts(3)
->withInitialInterval(CarbonInterval::seconds(5))
->withBackoffCoefficient(2.0)
)
);
}
public function processBatch(string $filePath, int $batchNumber, int $batchSize, array $simulationConfig = []): \Generator
{
return yield $this->activityStub->processBatch($filePath, $batchNumber, $batchSize, $simulationConfig);
}
}

13
app/Temporal/ProductImport/BatchImportWorkflowInterface.php Normal file
View File

@@ -0,0 +1,13 @@
<?php
namespace App\Temporal\ProductImport;
use Temporal\Workflow\WorkflowInterface;
use Temporal\Workflow\WorkflowMethod;
#[WorkflowInterface]
interface BatchImportWorkflowInterface
{
#[WorkflowMethod]
public function processBatch(string $filePath, int $batchNumber, int $batchSize, array $simulationConfig = []);
}

147
app/Temporal/ProductImport/ProductImportActivity.php Normal file
View File

@@ -0,0 +1,147 @@
<?php
namespace App\Temporal\ProductImport;
use App\Models\Product;
use App\Temporal\Shared\FaultSimulator;
use Temporal\Activity;
use Illuminate\Support\Facades\Log;
class ProductImportActivity implements ProductImportActivityInterface
{
public function validateFile(string $filePath, array $simulationConfig = []): bool
{
FaultSimulator::maybeApply($simulationConfig, 'validateFile');
if (!file_exists($filePath)) {
throw new \RuntimeException("File not found: {$filePath}");
}
if (!is_readable($filePath)) {
throw new \RuntimeException("File is not readable: {$filePath}");
}
return true;
}
public function parseAndCountRecords(string $filePath, array $simulationConfig = []): int
{
FaultSimulator::maybeApply($simulationConfig, 'parseAndCountRecords');
$handle = fopen($filePath, 'r');
if ($handle === false) {
throw new \RuntimeException("Unable to open file: {$filePath}");
}
// Skip header row
fgetcsv($handle);
$count = 0;
while (fgetcsv($handle) !== false) {
$count++;
}
fclose($handle);
return $count;
}
public function processBatch(string $filePath, int $batchNumber, int $batchSize, array $simulationConfig = []): array
{
FaultSimulator::maybeApply($simulationConfig, 'processBatch');
$handle = fopen($filePath, 'r');
if ($handle === false) {
throw new \RuntimeException("Unable to open file: {$filePath}");
}
// Skip header row
fgetcsv($handle);
// Skip to the batch offset
$offset = $batchNumber * $batchSize;
for ($i = 0; $i < $offset; $i++) {
fgetcsv($handle);
}
$processed = 0;
$failed = 0;
$errors = [];
for ($i = 0; $i < $batchSize; $i++) {
$row = fgetcsv($handle);
if ($row === false) {
break;
}
// Heartbeat with progress
Activity::heartbeat([
'batchNumber' => $batchNumber,
'current' => $i + 1,
'total' => $batchSize,
]);
try {
// Parse CSV columns: sku, name, description, price, stock_quantity, category
[$sku, $name, $description, $price, $stockQuantity, $category] = $row;
Product::updateOrCreate(
['sku' => $sku],
[
'name' => $name,
'description' => $description,
'price' => (float) $price,
'stock_quantity' => (int) $stockQuantity,
'category' => $category,
]
);
$processed++;
} catch (\Throwable $e) {
$failed++;
$errors[] = [
'row' => $offset + $i + 1,
'error' => $e->getMessage(),
];
Log::warning("Product import batch {$batchNumber} row error", [
'row' => $offset + $i + 1,
'error' => $e->getMessage(),
]);
}
}
fclose($handle);
return [
'processed' => $processed,
'failed' => $failed,
'errors' => $errors,
'attempt' => Activity::getInfo()->attempt,
];
}
public function generateReport(
string $filePath,
int $totalRecords,
int $processedRecords,
int $failedRecords,
array $simulationConfig = []
): array {
FaultSimulator::maybeApply($simulationConfig, 'generateReport');
return [
'filePath' => $filePath,
'totalRecords' => $totalRecords,
'processedRecords' => $processedRecords,
'failedRecords' => $failedRecords,
'successRate' => $totalRecords > 0
? round(($processedRecords / $totalRecords) * 100, 2)
: 0,
'completedAt' => now()->toIso8601String(),
];
}
}

22
app/Temporal/ProductImport/ProductImportActivityInterface.php Normal file
View File

@@ -0,0 +1,22 @@
<?php
namespace App\Temporal\ProductImport;
use Temporal\Activity\ActivityInterface;
use Temporal\Activity\ActivityMethod;
#[ActivityInterface]
interface ProductImportActivityInterface
{
#[ActivityMethod]
public function validateFile(string $filePath, array $simulationConfig = []): bool;
#[ActivityMethod]
public function parseAndCountRecords(string $filePath, array $simulationConfig = []): int;
#[ActivityMethod]
public function processBatch(string $filePath, int $batchNumber, int $batchSize, array $simulationConfig = []): array;
#[ActivityMethod]
public function generateReport(string $filePath, int $totalRecords, int $processedRecords, int $failedRecords, array $simulationConfig = []): array;
}

161
app/Temporal/ProductImport/ProductImportWorkflow.php Normal file
View File

@@ -0,0 +1,161 @@
<?php
namespace App\Temporal\ProductImport;
use Carbon\CarbonInterval;
use Temporal\Activity\ActivityOptions;
use Temporal\Workflow;
use Temporal\Workflow\ChildWorkflowOptions;
class ProductImportWorkflow implements ProductImportWorkflowInterface
{
private string $status = 'pending';
private int $processedRecords = 0;
private int $failedRecords = 0;
private int $totalRecords = 0;
private bool $isPaused = false;
private bool $isCancelled = false;
private int $retryCount = 0;
private int $rateLimitHits = 0;
/** @var ProductImportActivityInterface */
private $activityStub;
public function __construct()
{
$this->activityStub = Workflow::newActivityStub(
ProductImportActivityInterface::class,
ActivityOptions::new()
->withStartToCloseTimeout(CarbonInterval::minutes(5))
->withHeartbeatTimeout(CarbonInterval::seconds(30))
);
}
public function import(string $filePath, array $simulationConfig = []): \Generator
{
try {
// Step 1: Validate the file
$this->status = 'validating';
yield $this->activityStub->validateFile($filePath, $simulationConfig);
// Step 2: Count records
$this->status = 'counting';
$this->totalRecords = yield $this->activityStub->parseAndCountRecords($filePath, $simulationConfig);
// Step 3: Process in batches
$this->status = 'processing';
$batchSize = 50;
$batchCount = (int) ceil($this->totalRecords / $batchSize);
$childPromises = [];
for ($i = 0; $i < $batchCount; $i++) {
if ($this->isCancelled) {
$this->status = 'cancelled';
return [
'status' => 'cancelled',
'totalRecords' => $this->totalRecords,
'processedRecords' => $this->processedRecords,
'failedRecords' => $this->failedRecords,
'retryCount' => $this->retryCount,
'rateLimitHits' => $this->rateLimitHits,
];
}
if ($this->isPaused) {
$this->status = 'paused';
yield Workflow::await(fn () => !$this->isPaused || $this->isCancelled);
if ($this->isCancelled) {
$this->status = 'cancelled';
return [
'status' => 'cancelled',
'totalRecords' => $this->totalRecords,
'processedRecords' => $this->processedRecords,
'failedRecords' => $this->failedRecords,
'retryCount' => $this->retryCount,
'rateLimitHits' => $this->rateLimitHits,
];
}
$this->status = 'processing';
}
// Spawn child workflow for this batch
$childStub = Workflow::newChildWorkflowStub(
BatchImportWorkflowInterface::class,
ChildWorkflowOptions::new()
);
$childPromises[] = $childStub->processBatch($filePath, $i, $batchSize, $simulationConfig);
}
// Collect results from all child workflows
foreach ($childPromises as $promise) {
$result = (array) (yield $promise);
$this->processedRecords += $result['processed'] ?? 0;
$this->failedRecords += $result['failed'] ?? 0;
$attempt = $result['attempt'] ?? 1;
if ($attempt > 1) {
$this->retryCount += ($attempt - 1);
}
}
// Step 4: Generate report
$this->status = 'reporting';
$report = yield $this->activityStub->generateReport(
$filePath,
$this->totalRecords,
$this->processedRecords,
$this->failedRecords,
$simulationConfig
);
} catch (\Throwable $e) {
$this->status = 'failed';
throw $e;
}
// Step 5: Complete
$this->status = 'completed';
return [
'status' => 'completed',
'totalRecords' => $this->totalRecords,
'processedRecords' => $this->processedRecords,
'failedRecords' => $this->failedRecords,
'retryCount' => $this->retryCount,
'rateLimitHits' => $this->rateLimitHits,
'report' => $report,
];
}
public function pause(): void
{
$this->isPaused = true;
}
public function resume(): void
{
$this->isPaused = false;
}
public function cancel(): void
{
$this->isCancelled = true;
$this->isPaused = false; // Meh, unblock any paused await
}
public function getStatus(): array
{
return [
'status' => $this->status,
'totalRecords' => $this->totalRecords,
'processedRecords' => $this->processedRecords,
'failedRecords' => $this->failedRecords,
'isPaused' => $this->isPaused,
'isCancelled' => $this->isCancelled,
'retryCount' => $this->retryCount,
'rateLimitHits' => $this->rateLimitHits,
];
}
}

27
app/Temporal/ProductImport/ProductImportWorkflowInterface.php Normal file
View File

@@ -0,0 +1,27 @@
<?php
namespace App\Temporal\ProductImport;
use Temporal\Workflow\WorkflowInterface;
use Temporal\Workflow\WorkflowMethod;
use Temporal\Workflow\SignalMethod;
use Temporal\Workflow\QueryMethod;
#[WorkflowInterface]
interface ProductImportWorkflowInterface
{
#[WorkflowMethod]
public function import(string $filePath, array $simulationConfig = []);
#[SignalMethod]
public function pause(): void;
#[SignalMethod]
public function resume(): void;
#[SignalMethod]
public function cancel(): void;
#[QueryMethod]
public function getStatus(): array;
}