# Streamingstappen

Met streamingstappen kan een workflow incrementeel output produceren. Elke opgeleverde chunk wordt direct naar de database geschreven, zodat de al opgeslagen chunks behouden blijven en beschikbaar zijn bij een nieuwe poging als de stap halverwege de stream mislukt.

Dit is nuttig voor streamingreacties van LLM's, het verwerken van grote bestanden en elk scenario waarin gedeeltelijke voortgang niet verloren mag gaan.

## De StreamingStep-interface

Een streamingstap implementeert `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;
}
```

| Methode | Beschrijving |
|---|---|
| `stream( $state, $existing_chunks )` | Generator die stringchunks oplevert. Elke yield leidt direct tot een DB-schrijfactie. |
| `on_complete( $chunks, $state )` | Wordt aangeroepen nadat de stream is voltooid. Ontvangt alle chunks (vorige + nieuwe) en retourneert gegevens om samen te voegen met de workflowstatus. |
| `config()` | Optionele configuratie. Ondersteunt de sleutels `max_attempts` en `backoff`; `backoff` verwacht een strategienaam (`exponential`, `linear`, `fixed`), omdat een lijst met vertragingen momenteel wordt genegeerd ([#16](https://github.com/inline0/queuety/issues/16)). |

## Hoe chunks worden opgeslagen

Elke keer dat je generator een waarde oplevert, roept Queuety `ChunkStore::append_chunk()` aan om deze naar de tabel `queuety_chunks` te schrijven. Er wordt niets gebundeld of gebufferd. Eén yield staat gelijk aan één INSERT.

De chunkregel slaat de `job_id`, `workflow_id`, `step_index`, `chunk_index` (vanaf nul) en de string `content` op.

## Hervatbaarheid

Wanneer een streamingstap na een fout opnieuw wordt geprobeerd, laadt de worker alle eerder opgeslagen chunks en geeft die door als de parameter `$existing_chunks`. Je implementatie bepaalt hoe het hervatten verloopt:

- **Sla al verwerkte items over** door het aantal bestaande chunks te controleren
- **Verspring in een gepagineerde API** met het aantal chunks als pagina-offset
- **Begin helemaal opnieuw** door `$existing_chunks` volledig te negeren (chunks van de vorige poging staan nog in de DB en worden ervoor geplaatst)

```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 );
    }
}
```

## Heartbeats tijdens het streamen

Langlopende streams moeten heartbeats versturen om te voorkomen dat de detector voor verouderde taken de taak opnieuw claimt. Roep `Heartbeat::beat()` aan in je generatorlus:

```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;
    }
}
```

Zie [Heartbeats](https://queuety.inline0.dev/nl/docs/features/heartbeats) voor meer informatie over de Heartbeat API.

## Chunks opruimen

Wanneer de stream succesvol is voltooid en `on_complete()` retourneert, roept de worker `ChunkStore::clear_chunks()` aan om alle chunks voor de taak te verwijderen. Het samengevoegde resultaat van `on_complete()` is het duurzame record. Chunks zijn bewust tijdelijk.

Als de stap mislukt en opnieuw wordt geprobeerd, worden de chunks niet verwijderd. Bij de volgende poging worden ze opnieuw doorgegeven via `$existing_chunks`.

## Volledig voorbeeld: LLM-streaming

```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 ];
    }
}
```

Gebruik dit in een workflow zoals elke andere stap:

```php
use Queuety\Queuety;

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

## Volledig voorbeeld: bestandsverwerking

```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' ];
    }
}
```

## De API-klasse

De klasse `ChunkStore` beheert de opslag van chunks. Je krijgt er toegang toe via `Queuety::chunk_store()`.

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

| Methode | Retourneert | Beschrijving |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Haalt alle chunks voor een taak op, gesorteerd op chunkindex |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Voegt één chunk toe |
| `clear_chunks( int $job_id )` | `void` | Verwijdert alle chunks voor een taak |
| `chunk_count( int $job_id )` | `int` | Telt het aantal chunks voor een taak |
| `get_accumulated( int $job_id )` | `string` | Voegt alle chunks samen tot één string |

Normaal gesproken hoef je `ChunkStore` niet rechtstreeks aan te roepen. De worker regelt het opslaan en opruimen automatisch bij het verwerken van een `StreamingStep`. API is beschikbaar voor geavanceerde gebruikssituaties, zoals het inspecteren van chunks die nog worden verwerkt of handmatig opruimen.