# @flink-app/streaming-plugin

Streaming response support for Flink Framework, enabling Server-Sent Events (SSE) and NDJSON streaming for real-time data transmission.

## Features

-   **Server-Sent Events (SSE)**: Perfect for live updates, notifications, and dashboards
-   **NDJSON Streaming**: Ideal for LLM chat streaming (OpenAI/Anthropic style)
-   **Type-Safe**: Full TypeScript support with generic event types
-   **Simple API**: Clean `StreamWriter` interface for writing data
-   **Connection Management**: Automatic handling of client disconnections
-   **Flexible**: Works alongside regular Flink handlers

## Installation

```bash
npm install @flink-app/streaming-plugin
```

## Quick Start

### 1. Add Plugin to Your App

```typescript
import { FlinkApp, FlinkContext } from "@flink-app/flink";
import { streamingPlugin, StreamingPluginContext } from "@flink-app/streaming-plugin";

// Define your app context with streaming plugin
interface AppContext extends FlinkContext<{ streaming: StreamingPluginContext }> {
    // Your other context properties
}

const app = await new FlinkApp<AppContext>({
    name: "My App",
    plugins: [streamingPlugin({ debug: true })],
}).start();
```

### 2. Create a Streaming Handler

**LLM-Style Chat Streaming (NDJSON):**

```typescript
// src/streaming/GetChatStream.ts
import { StreamHandler, StreamingRouteProps } from "@flink-app/streaming-plugin";

export const Route: StreamingRouteProps = {
    path: "/chat/stream",
    format: "ndjson", // Use NDJSON for LLM streaming
    skipAutoRegister: true, // Required for streaming handlers
};

interface ChatEvent {
    delta: string;
    done?: boolean;
}

const GetChatStream: StreamHandler<AppContext, ChatEvent> = async ({ req, stream }) => {
    const prompt = req.query.prompt as string;

    // Call your LLM API
    for await (const chunk of callLLMAPI(prompt)) {
        stream.write({
            delta: chunk.text,
        });
    }

    stream.write({ delta: "", done: true });
    stream.end();
};

export default GetChatStream;
```

**Live Updates (SSE):**

```typescript
// src/streaming/GetLiveUpdates.ts
import { StreamHandler, StreamingRouteProps } from "@flink-app/streaming-plugin";

export const Route: StreamingRouteProps = {
    path: "/live-updates",
    format: "sse", // Use SSE for live updates
    skipAutoRegister: true,
};

interface UpdateEvent {
    type: "update" | "notification";
    message: string;
    timestamp: number;
}

const GetLiveUpdates: StreamHandler<AppContext, UpdateEvent> = async ({ stream }) => {
    // Send updates periodically
    const interval = setInterval(() => {
        if (!stream.isOpen()) {
            clearInterval(interval);
            return;
        }

        stream.write({
            type: "update",
            message: "New data available",
            timestamp: Date.now(),
        });
    }, 1000);
};

export default GetLiveUpdates;
```

### 3. Register Streaming Handlers

**Option 1: Via context (recommended)**

```typescript
// src/index.ts
import * as GetChatStream from "./streaming/GetChatStream";
import * as GetLiveUpdates from "./streaming/GetLiveUpdates";

// After app.start() - access via ctx.plugins.streaming
app.ctx.plugins.streaming.registerStreamHandler(GetChatStream);
app.ctx.plugins.streaming.registerStreamHandler(GetLiveUpdates);
```

**Option 2: Via plugin instance**

```typescript
// If you prefer to keep a reference to the plugin instance
const streaming = streamingPlugin({ debug: true });

const app = await new FlinkApp({
    plugins: [streaming],
}).start();

// Register after app.start()
streaming.registerStreamHandler(GetChatStream);
streaming.registerStreamHandler(GetLiveUpdates);
```

**Option 3: Explicit registration**

```typescript
// For inline handlers without separate files
app.ctx.plugins.streaming.registerStreamHandler(
    async ({ stream }) => {
        stream.write({ message: "Hello" });
        stream.end();
    },
    { path: "/custom", format: "sse", skipAutoRegister: true }
);
```

## Client-Side Usage

### Consuming NDJSON Streams (Fetch API)

```typescript
async function streamChat(prompt: string) {
    const response = await fetch(`/chat/stream?prompt=${encodeURIComponent(prompt)}`);
    const reader = response.body!.getReader();
    const decoder = new TextDecoder();

    let buffer = "";

    while (true) {
        const { done, value } = await reader.read();

        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split("\n");
        buffer = lines.pop() || ""; // Keep incomplete line

        for (const line of lines) {
            if (line.trim()) {
                const event = JSON.parse(line);
                console.log("Delta:", event.delta);

                if (event.done) {
                    console.log("Stream complete");
                    return;
                }
            }
        }
    }
}
```

### Consuming SSE Streams (EventSource)

```typescript
const eventSource = new EventSource("/live-updates");

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log("Update:", data);
};

eventSource.addEventListener("error", (event) => {
    const error = JSON.parse(event.data);
    console.error("Error:", error.message);
});

// Close connection when done
eventSource.close();
```

## API Reference

### `streamingPlugin(options?)`

Creates a streaming plugin instance.

**Options:**

-   `defaultFormat?: 'sse' | 'ndjson'` - Default format if not specified (default: `'sse'`)
-   `debug?: boolean` - Enable debug logging (default: `false`)

### `StreamHandler<Ctx, T>`

Handler function type for streaming endpoints.

**Type Parameters:**

-   `Ctx` - Your application context type
-   `T` - Event data type

**Props:**

-   `req: FlinkRequest` - Flink request object
-   `ctx: Ctx` - Application context
-   `stream: StreamWriter<T>` - Stream writer for sending data
-   `origin?: string` - Route origin

### `StreamWriter<T>`

Interface for writing data to streams.

**Methods:**

-   `write(data: T): void` - Write data to the stream
-   `error(error: Error | string): void` - Send error event
-   `end(): void` - Close the stream
-   `isOpen(): boolean` - Check if connection is still open

### `StreamingRouteProps`

Route configuration for streaming endpoints.

**Properties:**

-   `path: string` - HTTP path (required)
-   `method?: HttpMethod` - HTTP method (default: `GET`)
-   `format?: 'sse' | 'ndjson'` - Streaming format (default: plugin default)
-   `skipAutoRegister: true` - Must be `true` (required)
-   `permissions?: string | string[]` - Route permissions (uses Flink auth plugin)
-   `origin?: string` - Route origin (optional)

## Streaming Formats

### SSE (Server-Sent Events)

**When to use:**

-   Live dashboards and real-time updates
-   Notifications and alerts
-   Progress indicators
-   Traditional one-way server → client streaming

**Format:**

```
data: {"message":"Hello"}\n\n
data: {"message":"World"}\n\n
```

**Headers:**

-   `Content-Type: text/event-stream`
-   `Cache-Control: no-cache`
-   `Connection: keep-alive`

### NDJSON (Newline-Delimited JSON)

**When to use:**

-   LLM chat streaming (OpenAI/Anthropic style)
-   Bulk data export
-   Log streaming
-   Any streaming JSON data

**Format:**

```
{"delta":"Hello"}\n
{"delta":" world","done":true}\n
```

**Headers:**

-   `Content-Type: application/x-ndjson`
-   `Cache-Control: no-cache`

## Authentication

Streaming handlers support Flink's authentication system just like regular handlers. Simply add `permissions` to your route configuration:

```typescript
export const Route: StreamingRouteProps = {
    path: "/admin/stream",
    format: "sse",
    skipAutoRegister: true,
    permissions: ["admin"], // Requires admin role
};

const AdminStream: StreamHandler<AppContext> = async ({ req, stream }) => {
    // ✅ req.user is populated by auth plugin after successful authentication
    const user = req.user; // Type depends on your auth plugin

    stream.write({
        message: `Hello ${user?.name}`,
        userId: user?.userId,
        permissions: user?.permissions,
    });
    stream.end();
};
```

**How it works:**

1. Plugin checks `routeProps.permissions` before starting the stream
2. Calls your configured auth plugin (JWT, BankID, OAuth, etc.)
3. Returns 401 if authentication fails
4. **✅ Authenticated user info is available in `req.user`** (populated by your auth plugin)

**Example with JWT:**

```typescript
import { jwtAuthPlugin } from "@flink-app/jwt-auth-plugin";

const app = new FlinkApp({
    auth: jwtAuthPlugin({ secret: "your-secret" }),
    plugins: [streaming],
});

// Client must send Authorization header
fetch("/admin/stream", {
    headers: {
        Authorization: "Bearer your-jwt-token",
    },
});
```

## Best Practices

### 1. Check Connection Status

Always check if the stream is still open before writing:

```typescript
if (stream.isOpen()) {
    stream.write(data);
}
```

### 2. Clean Up Resources

Clean up intervals, timers, and resources when the connection closes:

```typescript
const interval = setInterval(() => {
    if (!stream.isOpen()) {
        clearInterval(interval);
        return;
    }
    stream.write(data);
}, 1000);
```

### 3. Handle Errors Gracefully

Always wrap streaming logic in try-catch:

```typescript
const handler: StreamHandler<Ctx> = async ({ stream }) => {
    try {
        // Your streaming logic
    } catch (err) {
        stream.error(err);
        stream.end();
    }
};
```

### 4. Use Type Parameters

Leverage TypeScript for type-safe events:

```typescript
interface ChatEvent {
    delta: string;
    done?: boolean;
    metadata?: {
        model: string;
        tokens: number;
    };
}

const handler: StreamHandler<Ctx, ChatEvent> = async ({ stream }) => {
    stream.write({
        delta: "Hello",
        metadata: { model: "gpt-4", tokens: 5 },
    }); // ✅ Type-safe

    // stream.write({ invalid: "data" }); // ❌ Type error
};
```

## Limitations

### Manual Registration Required

Unlike regular Flink handlers, streaming handlers must be registered manually after `app.start()`:

```typescript
// ❌ Won't auto-register
export const Route: StreamingRouteProps = {
    path: "/stream",
    skipAutoRegister: true, // Required
};

// ✅ Must register manually
streaming.registerStreamHandler(handler, Route);
```

This is a trade-off for zero core framework changes and allows the plugin to work independently.

### ⚠️ File Location Requirement

**IMPORTANT:** Streaming handlers MUST be placed outside the `src/handlers/` directory.

```typescript
// ❌ WRONG - Will cause TypeScript compilation error
src/handlers/streaming/GetChatStream.ts

// ✅ CORRECT - Place outside handlers directory
src/streaming/GetChatStream.ts
```

**Why?** Flink's TypeScript compiler automatically scans `src/handlers/` for auto-registration and attempts to analyze all handler types at compile-time. Since it doesn't recognize the `StreamHandler` type from this plugin, it will throw an error:

```
Error: Unknown handler type StreamHandler in GetChatStream.ts - should be Handler or GetHandler
```

**Recommended locations:**
- `src/streaming/` - Dedicated directory for streaming handlers (recommended)
- `src/streams/` - Alternative naming
- Any directory outside `src/handlers/`

### No Schema Validation

Streaming responses bypass Flink's JSON schema validation since data is sent incrementally. Ensure your event types are well-defined and validated manually if needed.

### Connection Limits

Be mindful of:

-   Server connection limits (max concurrent streams)
-   Client browser limits (typically 6 connections per domain)
-   Load balancer timeout settings

## Examples

See the [demo-app](../../demo-app) for complete working examples:

-   `/chat/stream` - NDJSON streaming chat
-   `/live-updates` - SSE live updates

## Roadmap

Potential future enhancements:

-   [ ] Auto-registration support (requires core framework changes)
-   [ ] Binary streaming support
-   [ ] WebSocket integration
-   [ ] Compression support (gzip)
-   [ ] Rate limiting/backpressure
-   [ ] Metrics and monitoring hooks

## License

MIT