Add streaming network layer (SSE) for AI/chat features (fixes #14)

docs/implemented.md

@@ -11,10 +11,11 @@ This file documents the patterns and code that are already implemented in the `l
 1. [Core Patterns](#core-patterns)
 2. [Dependency Injection](#dependency-injection)
 3. [Network Layer](#network-layer)
-4. [Connectivity Strategy](#connectivity-strategy)
-5. [Offline Queue](#offline-queue)
-6. [BLoC Patterns](#bloc-patterns)
-7. [Testing Strategy](#testing-strategy)
+4. [Streaming (SSE) Network Layer](#streaming-sse-network-layer)
+5. [Connectivity Strategy](#connectivity-strategy)
+6. [Offline Queue](#offline-queue)
+7. [BLoC Patterns](#bloc-patterns)
+8. [Testing Strategy](#testing-strategy)
 
 ---
 
@@ -278,6 +279,111 @@ class AuthInterceptor extends Interceptor {
 
 ---
 
+## Streaming (SSE) Network Layer
+
+**Location:** `lib/core/network/streaming_client.dart`, `lib/core/network/sse_parser.dart`
+
+The request/response path waits for a full HTTP body. Streaming is different: AI/chat
+backends emit tokens over a Server-Sent Events (SSE) stream, and we want to render
+text incrementally instead of blocking on the whole response. The template ships a
+small, reusable streaming layer that composes with the existing `Result<T>` and
+connectivity-first patterns.
+
+### When to use it
+
+Use `StreamingClient` whenever a response is delivered incrementally as tokens
+(SSE from an AI/chat proxy). Do **not** use it for ordinary JSON request/response —
+that stays on `DioClient`/`ItemRepository` so the offline queue and cache apply.
+
+### `StreamingClient` — how to wire it
+
+```dart
+// Registered in `lib/core/di/injection.dart`:
+getIt.registerLazySingleton<StreamingClient>(
+  () => StreamingClient(
+    dioClient: getIt<DioClient>(),
+    connectivity: getIt<ConnectivityService>(),
+    logger: getIt<Logger>(),
+  ),
+);
+
+final client = getIt<StreamingClient>();
+final cancelToken = CancelToken();
+
+final stream = client.stream(
+  '/chat',
+  queryParameters: {'q': 'hello'},
+  cancelToken: cancelToken,
+);
+```
+
+`StreamingClient` deliberately reuses the injected `DioClient` (its configured `Dio`)
+so `baseUrl`, the `AuthInterceptor`, and debug logging are preserved. Every streaming
+request overrides the global defaults:
+
+```dart
+final options = Options(
+  responseType: ResponseType.stream,
+  receiveTimeout: Duration.zero,          // do NOT use null — that falls back to 30s
+  sendTimeout: Duration.zero,
+  headers: {'Accept': 'text/event-stream'},
+);
+```
+
+The two overrides are load-bearing:
+
+- **`receiveTimeout: Duration.zero`** disables Dio's receive-timeout timer so a sparse,
+  long-lived stream (tokens arriving >30s apart) is not aborted. `null` would silently
+  fall back to the base 30s timeout in Dio 5.x and break streaming.
+- **`Accept: text/event-stream`** stops SSE backends from buffering the full body and
+  overrides the template default `Accept: application/json`.
+
+### `sse_parser.dart` — framing SSE into events
+
+The parser turns a raw byte/text stream into typed `SseEvent`s (data / done / error),
+handling multi-line `data:` fields, the `[DONE]` sentinel, `event:` / `id:` fields,
+comment lines, and CRLF normalization. Bytes are decoded with `utf8.decoder.bind` so a
+multi-byte character split across chunk boundaries is decoded correctly.
+
+```dart
+Stream<SseEvent> events = parseSseBytes(responseBodyStream);
+```
+
+### Emission contract
+
+`StreamingClient.stream` returns a `Stream<Result<String>>` that is **always
+well-terminated** — it never leaves a consumer hanging:
+
+1. `Result.loading()` is emitted exactly once as the leading item.
+2. One `Result.success(token)` per SSE `data:` payload.
+3. The `[DONE]` sentinel ends the stream normally and is **not** emitted as a token.
+4. A mid-stream error / disconnect / cancellation is converted into a terminal
+   `Result.failure(message, err)` before the stream closes.
+
+### Connectivity composition
+
+A token stream cannot be replayed by the Hive-backed `OfflineQueue` (it only replays
+request/response pairs). So when connectivity is offline or poor, `StreamingClient`
+emits `Result.failure` immediately and **does not enqueue** anything; no cached replay
+of streams is attempted (deferred). A `CancelToken` is cancelled so no orphaned stream
+lingers.
+
+### Reference feature: `features/chat/`
+
+`lib/features/chat/` is a copy-paste starting point mirroring `features/home/`:
+
+- `data/repositories/chat_repository.dart` — wraps `StreamingClient.stream`.
+- `presentation/bloc/chat_bloc.dart` — accumulates tokens into a message, bounded by
+  `kMaxMessageChars`, with a `stop` action wired to the `CancelToken`. An aborted
+  mid-stream becomes `ChatState.stopped` (partial text), never a success.
+- `presentation/pages/chat_page.dart` + `widgets/chat_stream_view.dart` — render the
+  accumulating text so it appears incrementally before the full response completes.
+
+The block accumulation guard (`if (message.length >= kMaxMessageChars) cancel();`)
+bounds memory, and `CancelToken` gives the user a real stop control.
+
+---
+
 ## Connectivity Strategy
 
 **Location:** `lib/core/connectivity/`

lib/core/di/injection.dart

@@ -6,6 +6,8 @@ import 'package:hive/hive.dart';
 import 'package:logger/logger.dart';
 import 'package:shared_preferences/shared_preferences.dart';
 
+import '../../features/chat/data/repositories/chat_repository.dart';
+import '../../features/chat/presentation/bloc/chat_bloc.dart';
 import '../../features/home/data/repositories/item_repository.dart';
 import '../../features/home/presentation/bloc/home_bloc.dart';
 import '../analytics/analytics_service.dart';
@@ -17,6 +19,7 @@ import '../network/auth_token_manager.dart';
 import '../network/dio_client.dart';
 import '../network/offline_queue.dart';
 import '../network/request_executor.dart';
+import '../network/streaming_client.dart';
 
 final getIt = GetIt.instance;
 
@@ -107,6 +110,15 @@ Future<void> configureDependencies() async {
     ),
   );
 
+  // Streaming (SSE)
+  getIt.registerLazySingleton<StreamingClient>(
+    () => StreamingClient(
+      dioClient: getIt<DioClient>(),
+      connectivity: getIt<ConnectivityService>(),
+      logger: getIt<Logger>(),
+    ),
+  );
+
   // Repositories
   getIt.registerLazySingleton<ItemRepository>(
     () => ItemRepository(
@@ -117,6 +129,13 @@ Future<void> configureDependencies() async {
     ),
   );
 
+  getIt.registerLazySingleton<ChatRepository>(
+    () => ChatRepository(
+      streamingClient: getIt<StreamingClient>(),
+      logger: getIt<Logger>(),
+    ),
+  );
+
   // BLoCs (factories for fresh instances)
   getIt.registerFactory<HomeBloc>(
     () => HomeBloc(
@@ -125,6 +144,13 @@ Future<void> configureDependencies() async {
     ),
   );
 
+  getIt.registerFactory<ChatBloc>(
+    () => ChatBloc(
+      repository: getIt<ChatRepository>(),
+      connectivityBloc: getIt<ConnectivityBloc>(),
+    ),
+  );
+
   // Auth (uncomment after implementing AuthRepository)
   // Import: import '../auth/auth_bloc.dart';
   // Import: import '../auth/auth_repository.dart';

lib/core/network/sse_parser.dart

@@ -0,0 +1,151 @@
+import 'dart:async';
+import 'dart:convert';
+
+/// The type of a parsed SSE event.
+enum SseEventType { data, done, error }
+
+/// A single parsed Server-Sent Events (SSE) event.
+///
+/// The raw `data:`/`event:`/`id:` fields from one SSE block are collapsed into
+/// a typed [SseEvent] so downstream consumers (e.g. [StreamingClient]) can
+/// react to data tokens, the terminal `[DONE]` sentinel, and error events
+/// without re-parsing text.
+class SseEvent {
+  const SseEvent.data(this.data, {this.event, this.id})
+      : type = SseEventType.data;
+  const SseEvent.done()
+      : type = SseEventType.done,
+        data = '',
+        event = null,
+        id = null;
+  const SseEvent.error(this.data)
+      : type = SseEventType.error,
+        event = null,
+        id = null;
+
+  final SseEventType type;
+
+  /// The payload. For a `data:` event this is the accumulated `data:` value
+  /// (multi-line values joined with `\n`). For an error event it is the error
+  /// payload.
+  final String data;
+
+  /// The value of the `event:` field, if present.
+  final String? event;
+
+  /// The value of the `id:` field, if present.
+  final String? id;
+
+  bool get isData => type == SseEventType.data;
+  bool get isDone => type == SseEventType.done;
+  bool get isError => type == SseEventType.error;
+
+  @override
+  String toString() => 'SseEvent($type, data: $data, event: $event, id: $id)';
+}
+
+/// The sentinel payload that marks the end of a stream.
+///
+/// Most chat/AI backends emit `data: [DONE]` as the final event.
+const String kSseDoneSentinel = '[DONE]';
+
+/// Parses a stream of raw UTF-8 byte chunks into typed SSE events.
+///
+/// Decoding is performed with [utf8.decoder] so a multi-byte character split
+/// across chunk boundaries is still decoded correctly.
+Stream<SseEvent> parseSseBytes(Stream<List<int>> chunks) {
+  return parseSse(utf8.decoder.bind(chunks));
+}
+
+/// Parses a stream of decoded text chunks into typed SSE events.
+///
+/// The parser buffers partial events across chunk boundaries and only emits a
+/// complete [SseEvent] once the terminating blank line (`\n\n`) has been
+/// received (or the source stream closes with trailing data).
+///
+/// Grammar handled (subset of the SSE spec):
+/// * `data:` fields — multiple `data:` lines are joined with `\n`.
+/// * `[DONE]` sentinel — surfaced as [SseEvent.done].
+/// * comments (lines starting with `:`) — ignored, never dispatched.
+/// * `event:` / `id:` fields — carried on the emitted [SseEvent].
+/// * `event: error` — surfaced as [SseEvent.error].
+Stream<SseEvent> parseSse(Stream<String> chunks) async* {
+  var buffer = '';
+
+  await for (final chunk in chunks) {
+    buffer += chunk;
+
+    // Normalize CRLF / CR to LF so event boundaries are predictable.
+    buffer = buffer.replaceAll('\r\n', '\n').replaceAll('\r', '\n');
+
+    while (true) {
+      final separator = buffer.indexOf('\n\n');
+      if (separator < 0) break;
+
+      final block = buffer.substring(0, separator);
+      buffer = buffer.substring(separator + 2);
+
+      final event = _parseBlock(block);
+      if (event != null) yield event;
+    }
+  }
+
+  // A trailing block with no terminating blank line is still an event.
+  if (buffer.trim().isNotEmpty) {
+    final event = _parseBlock(buffer);
+    if (event != null) yield event;
+  }
+}
+
+SseEvent? _parseBlock(String block) {
+  final dataLines = <String>[];
+  String? eventName;
+  String? id;
+  var sawField = false;
+
+  for (final line in block.split('\n')) {
+    if (line.isEmpty) continue;
+
+    // Comment lines are keep-alive markers and never dispatch an event.
+    if (line.startsWith(':')) continue;
+
+    sawField = true;
+
+    final colon = line.indexOf(':');
+    String field;
+    String value;
+    if (colon == -1) {
+      field = line;
+      value = '';
+    } else {
+      field = line.substring(0, colon);
+      value = line.substring(colon + 1);
+      if (value.startsWith(' ')) value = value.substring(1);
+    }
+
+    switch (field) {
+      case 'data':
+        dataLines.add(value);
+      case 'event':
+        eventName = value;
+      case 'id':
+        id = value;
+    }
+  }
+
+  // A block of only comments produces no event.
+  if (!sawField) return null;
+
+  final data = dataLines.join('\n');
+
+  // `[DONE]` and `event: done` are the terminal marker.
+  if (eventName == 'done' || data == kSseDoneSentinel) {
+    return const SseEvent.done();
+  }
+
+  if (eventName == 'error') {
+    return SseEvent.error(data);
+  }
+
+  return SseEvent.data(data, event: eventName, id: id);
+}