# خطوات البث

تتيح خطوات البث لسير العمل إنتاج المخرجات تدريجيًا. يُكتب كل جزء مُنتَج في قاعدة البيانات فورًا، لذا إذا فشلت الخطوة أثناء البث، تبقى الأجزاء المحفوظة مسبقًا متاحة عند إعادة المحاولة.

يفيد ذلك في استجابات البث من LLM، ومعالجة الملفات الكبيرة، وأي سيناريو ينبغي فيه الحفاظ على التقدم الجزئي.

## واجهة StreamingStep

تُنفذ خطوة البث `Queuety\Contracts\StreamingStep`:

```php
namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}
```

| الطريقة | الوصف |
|---|---|
| `stream( $state, $existing_chunks )` | مولّد ينتج أجزاء نصية. يؤدي كل إنتاج إلى كتابة فورية في قاعدة البيانات. |
| `on_complete( $chunks, $state )` | تُستدعى بعد انتهاء البث. تتلقى جميع الأجزاء (السابقة والجديدة) وتعيد بيانات لدمجها في حالة سير العمل. |
| `config()` | إعداد اختياري. يدعم المفتاحين `max_attempts` و`backoff`؛ ويأخذ `backoff` اسم استراتيجية (`exponential` أو `linear` أو `fixed`)، إذ تُسقط حاليًا قائمة التأخيرات ([#16](https://github.com/inline0/queuety/issues/16)). |

## كيفية حفظ الأجزاء

في كل مرة ينتج فيها المولّد قيمة، تستدعي Queuety الدالة `ChunkStore::append_chunk()` لكتابتها في جدول `queuety_chunks`. لا توجد أي عملية تجميع أو تخزين مؤقت. كل إنتاج يعادل عملية INSERT واحدة.

يخزّن صف الجزء القيم `job_id` و`workflow_id` و`step_index` و`chunk_index` (على أساس يبدأ من الصفر) والنص `content`.

## إمكانية الاستئناف

عند إعادة محاولة خطوة بث بعد فشلها، يحمّل العامل جميع الأجزاء المحفوظة سابقًا ويمررها كمعامل `$existing_chunks`. ويحدد تنفيذك كيفية الاستئناف:

- **تخطَّ العناصر التي عولجت بالفعل** بالتحقق من عدد الأجزاء الموجودة
- **انتقل إلى موضع محدد في API مُقسّم إلى صفحات** باستخدام عدد الأجزاء كإزاحة للصفحة
- **أعد التشغيل من البداية** بتجاهل `$existing_chunks` تمامًا (تبقى أجزاء المحاولة السابقة في قاعدة البيانات وتُضاف إلى البداية)

```php
public function stream( array $state, array $existing_chunks = [] ): \Generator {
    $offset = count( $existing_chunks );

    $pages = $this->client->paginate( $state['query'], offset: $offset );

    foreach ( $pages as $page ) {
        yield json_encode( $page );
    }
}
```

## نبضات الحياة أثناء البث

ينبغي أن ترسل التدفقات طويلة التشغيل نبضات حياة لمنع كاشف المهام القديمة من استعادة المهمة. استدعِ `Heartbeat::beat()` داخل حلقة المولّد:

```php
use Queuety\Heartbeat;

public function stream( array $state, array $existing_chunks = [] ): \Generator {
    foreach ( $this->client->stream( $state['prompt'] ) as $i => $chunk ) {
        Heartbeat::beat( [ 'chunks' => $i + 1 ] );
        yield $chunk;
    }
}
```

راجع [نبضات الحياة](https://queuety.inline0.dev/ar/docs/features/heartbeats) لمعرفة تفاصيل واجهة Heartbeat API.

## تنظيف الأجزاء

عند اكتمال البث بنجاح وعودة `on_complete()`، يستدعي العامل `ChunkStore::clear_chunks()` لحذف جميع أجزاء المهمة. وتُعد النتيجة المدمجة من `on_complete()` السجل الدائم. فالأجزاء مؤقتة بطبيعتها.

إذا فشلت الخطوة وأُعيدت محاولتها، فلا تُحذف الأجزاء. وتُمرر مجددًا عبر `$existing_chunks` في المحاولة التالية.

## مثال كامل: بث LLM

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamLLMResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        // If we have existing chunks from a previous attempt, we can skip
        // re-requesting. The on_complete method will concatenate everything.
        if ( ! empty( $existing_chunks ) ) {
            return;
        }

        $response = OpenAI::chat()->createStreamed( [
            'model'    => 'gpt-4',
            'messages' => $state['messages'],
        ] );

        foreach ( $response as $i => $chunk ) {
            $text = $chunk->choices[0]->delta->content ?? '';
            if ( '' !== $text ) {
                Heartbeat::beat( [ 'tokens' => $i ] );
                yield $text;
            }
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [
            'llm_response' => implode( '', $chunks ),
        ];
    }

    public function config(): array {
        return [ 'max_attempts' => 5 ];
    }
}
```

استخدمها في سير العمل مثل أي خطوة أخرى:

```php
use Queuety\Queuety;

Queuety::workflow( 'ai_chat' )
    ->then( BuildPromptHandler::class )
    ->then( StreamLLMResponse::class )
    ->then( SaveResponseHandler::class )
    ->dispatch( [ 'user_input' => $input ] );
```

## مثال كامل: معالجة ملف

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class ProcessCSVRows implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $file   = fopen( $state['csv_path'], 'r' );
        $offset = count( $existing_chunks );
        $row    = 0;

        // Skip the header.
        fgetcsv( $file );

        while ( $line = fgetcsv( $file ) ) {
            $row++;

            // Skip rows already processed in a previous attempt.
            if ( $row <= $offset ) {
                continue;
            }

            $result = $this->process_row( $line );
            Heartbeat::beat( [ 'row' => $row ] );
            yield json_encode( $result );
        }

        fclose( $file );
    }

    public function on_complete( array $chunks, array $state ): array {
        $results = array_map( 'json_decode', $chunks );
        return [
            'processed_count' => count( $results ),
            'results'         => $results,
        ];
    }

    public function config(): array {
        return [];
    }

    private function process_row( array $line ): array {
        // Your row processing logic here.
        return [ 'id' => $line[0], 'status' => 'ok' ];
    }
}
```

## واجهة ChunkStore API

تدير الفئة `ChunkStore` حفظ الأجزاء. يمكنك الوصول إليها عبر `Queuety::chunk_store()`.

```php
$store = Queuety::chunk_store();
```

| الطريقة | القيمة المُعادة | الوصف |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | جلب جميع أجزاء مهمة، مرتبة حسب فهرس الجزء |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | إلحاق جزء واحد |
| `clear_chunks( int $job_id )` | `void` | حذف جميع أجزاء مهمة |
| `chunk_count( int $job_id )` | `int` | حساب عدد أجزاء مهمة |
| `get_accumulated( int $job_id )` | `string` | وصل جميع الأجزاء في نص واحد |

لا تحتاج عادةً إلى استدعاء `ChunkStore` مباشرةً. يتولى العامل الحفظ والتنظيف تلقائيًا عند معالجة `StreamingStep`. وتتوفر API لحالات الاستخدام المتقدمة، مثل فحص الأجزاء قيد المعالجة أو تنظيفها يدويًا.