import { ExpressResponse, FlinkApp, FlinkContext, FlinkPlugin, HttpMethod, log } from "@flink-app/flink";
import { StreamFormat, StreamHandler, StreamHandlerModule, StreamingPluginOptions, StreamingRouteProps, StreamWriter } from "./types";

/**
 * Creates a stream writer for the given format
 */
function createStreamWriter<T>(res: ExpressResponse, format: StreamFormat, debug: boolean): StreamWriter<T> {
    let closed = false;

    res.on("close", () => {
        closed = true;
        if (debug) {
            log.debug("[StreamingPlugin] Client closed connection");
        }
    });

    return {
        write(data: T): void {
            if (closed) {
                if (debug) {
                    log.warn("[StreamingPlugin] Attempted write to closed stream");
                }
                return;
            }

            try {
                if (format === "sse") {
                    // SSE format: data: {json}\n\n
                    res.write(`data: ${JSON.stringify(data)}\n\n`);
                } else if (format === "ndjson") {
                    // NDJSON format: {json}\n
                    res.write(`${JSON.stringify(data)}\n`);
                }
            } catch (err) {
                log.error("[StreamingPlugin] Error writing to stream:", err);
                closed = true;
            }
        },

        error(error: Error | string): void {
            if (closed) return;

            const errorMessage = typeof error === "string" ? error : error.message;

            try {
                if (format === "sse") {
                    // SSE error event
                    res.write(`event: error\ndata: ${JSON.stringify({ message: errorMessage })}\n\n`);
                } else if (format === "ndjson") {
                    // NDJSON error object
                    res.write(`${JSON.stringify({ error: errorMessage })}\n`);
                }
            } catch (err) {
                log.error("[StreamingPlugin] Error writing error to stream:", err);
                closed = true;
            }
        },

        end(): void {
            if (closed) return;
            closed = true;
            res.end();
        },

        isOpen(): boolean {
            return !closed;
        },
    };
}

/**
 * Streaming plugin for Flink Framework
 * Provides SSE and NDJSON streaming support
 */
export class StreamingPlugin implements FlinkPlugin {
    public id = "streaming";
    private app?: FlinkApp<any>;
    private options: Required<StreamingPluginOptions>;

    constructor(options: StreamingPluginOptions = {}) {
        this.options = {
            defaultFormat: options.defaultFormat || "sse",
            debug: options.debug || false,
        };
    }

    async init(app: FlinkApp<any>): Promise<void> {
        this.app = app;

        // Add plugin to context for access via ctx.plugins.streaming
        app.ctx.plugins.streaming = {
            registerStreamHandler: this.registerStreamHandler.bind(this),
        };

        if (this.options.debug) {
            log.info(`[StreamingPlugin] Initialized with default format: ${this.options.defaultFormat}`);
        }
    }

    /**
     * Register a streaming handler from a module (preferred DX)
     *
     * @example
     * ```typescript
     * import * as GetChatStream from "./handlers/streaming/GetChatStream";
     *
     * // After app.start()
     * streaming.registerStreamHandler(GetChatStream);
     * ```
     */
    public registerStreamHandler<Ctx extends FlinkContext, T = any>(handlerModule: StreamHandlerModule<Ctx, T>): void;

    /**
     * Register a streaming handler with explicit handler and route props
     *
     * @example
     * ```typescript
     * streaming.registerStreamHandler(
     *   async ({ stream }) => {
     *     stream.write({ message: "Hello" });
     *     stream.end();
     *   },
     *   { path: "/stream", skipAutoRegister: true }
     * );
     * ```
     */
    public registerStreamHandler<Ctx extends FlinkContext, T = any>(handler: StreamHandler<Ctx, T>, routeProps: StreamingRouteProps): void;

    // Implementation
    public registerStreamHandler<Ctx extends FlinkContext, T = any>(
        handlerOrModule: StreamHandler<Ctx, T> | StreamHandlerModule<Ctx, T>,
        routeProps?: StreamingRouteProps
    ): void {
        if (!this.app) {
            throw new Error("[StreamingPlugin] Plugin not initialized - call app.start() first");
        }

        if (!this.app.expressApp) {
            throw new Error("[StreamingPlugin] Express app not available");
        }

        // Determine if we received a module or explicit handler+props
        let handler: StreamHandler<Ctx, T>;
        let props: StreamingRouteProps;

        if (typeof handlerOrModule === "function") {
            // Explicit handler function + route props
            if (!routeProps) {
                throw new Error("[StreamingPlugin] Route props are required when registering with explicit handler function");
            }
            handler = handlerOrModule;
            props = routeProps;
        } else {
            // Handler module with default export and Route
            const module = handlerOrModule as StreamHandlerModule<Ctx, T>;
            if (!module.default) {
                throw new Error("[StreamingPlugin] Handler module must have a default export");
            }
            if (!module.Route) {
                throw new Error("[StreamingPlugin] Handler module must export Route configuration");
            }
            handler = module.default;
            props = module.Route;
        }

        const method = props.method || HttpMethod.get;
        const format = props.format || this.options.defaultFormat;
        const methodAndRoute = `${method.toUpperCase()} ${props.path}`;

        // Register Express route
        this.app.expressApp[method](props.path, async (req: any, res: any) => {
            // Check authentication if permissions are required
            if (props.permissions) {
                if (!this.app!.auth) {
                    log.error(`[StreamingPlugin] ${methodAndRoute} requires authentication but no auth plugin is configured`);
                    res.status(500).json({
                        status: 500,
                        error: {
                            title: "Internal Server Error",
                            detail: "Authentication not configured",
                        },
                    });
                    return;
                }

                try {
                    const authenticated = await this.app!.auth.authenticateRequest(req as any, props.permissions);
                    if (!authenticated) {
                        res.status(401).json({
                            status: 401,
                            error: {
                                title: "Unauthorized",
                                detail: "Authentication required",
                            },
                        });
                        return;
                    }
                } catch (err) {
                    log.error(`[StreamingPlugin] ${methodAndRoute} authentication error:`, err);
                    res.status(401).json({
                        status: 401,
                        error: {
                            title: "Unauthorized",
                            detail: "Authentication failed",
                        },
                    });
                    return;
                }
            }

            // Set headers based on format
            if (format === "sse") {
                res.setHeader("Content-Type", "text/event-stream");
                res.setHeader("Cache-Control", "no-cache");
                res.setHeader("Connection", "keep-alive");
                res.setHeader("X-Accel-Buffering", "no"); // Disable nginx buffering
            } else if (format === "ndjson") {
                res.setHeader("Content-Type", "application/x-ndjson");
                res.setHeader("Cache-Control", "no-cache");
                res.setHeader("X-Accel-Buffering", "no");
            }

            // Flush headers immediately
            res.flushHeaders();

            // Create stream writer
            const streamWriter = createStreamWriter<T>(res, format, this.options.debug);

            try {
                if (this.options.debug) {
                    log.debug(`[StreamingPlugin] ${methodAndRoute} - Stream started`);
                }

                // Call the handler
                await handler({
                    req: req as any,
                    ctx: this.app!.ctx,
                    stream: streamWriter,
                    origin: props.origin,
                });

                if (this.options.debug) {
                    log.debug(`[StreamingPlugin] ${methodAndRoute} - Handler completed`);
                }
            } catch (err) {
                log.error(`[StreamingPlugin] ${methodAndRoute} - Handler error:`, err);

                // Send error to client if stream still open
                if (streamWriter.isOpen()) {
                    streamWriter.error(err as Error);
                }
            } finally {
                // Ensure stream is closed
                if (streamWriter.isOpen()) {
                    streamWriter.end();
                }
            }
        });

        log.info(`[StreamingPlugin] Registered streaming route ${methodAndRoute} (${format})`);
    }
}

/**
 * Factory function to create streaming plugin
 */
export function streamingPlugin(options?: StreamingPluginOptions): StreamingPlugin {
    return new StreamingPlugin(options);
}