markdownblog/src/app/api/posts/stream/route.ts

import { NextRequest, NextResponse } from 'next/server';
import { spawn } from 'child_process';

// Prevent static generation of this route
export const dynamic = 'force-dynamic';
export const runtime = 'nodejs';

// Store connected clients
const clients = new Set<ReadableStreamDefaultController>();

// Handle CORS preflight requests
export async function OPTIONS(request: NextRequest) {
  return new NextResponse(null, {
    status: 200,
    headers: {
      'Access-Control-Allow-Origin': '*',
      'Access-Control-Allow-Methods': 'GET, OPTIONS',
      'Access-Control-Allow-Headers': 'Cache-Control, Content-Type',
      'Access-Control-Max-Age': '86400',
    },
  });
}

export async function GET(request: NextRequest) {
  try {
    const stream = new ReadableStream({
      start(controller) {
        // Add this client to the set
        clients.add(controller);

        // Send initial connection message
        try {
          controller.enqueue(`data: ${JSON.stringify({ type: 'connected', message: 'SSE connection established' })}\n\n`);
        } catch (error) {
          console.error('Error sending initial message:', error);
          clients.delete(controller);
          return;
        }

        // Set up Rust file watcher if not already set up
        if (clients.size === 1) {
          try {
            const rustWatcher = spawn(
              process.cwd() + '/markdown_backend/target/release/markdown_backend',
              ['watch'],
              { stdio: ['pipe', 'pipe', 'pipe'] }
            );

            rustWatcher.stdout.on('data', (data) => {
              const message = data.toString().trim();
              console.log('Rust watcher output:', message);
              
              if (message.includes('Posts directory changed!')) {
                // Notify all connected clients about the update
                const updateMessage = JSON.stringify({ type: 'update', timestamp: new Date().toISOString() });
                const clientsToRemove: ReadableStreamDefaultController[] = [];
                
                clients.forEach(client => {
                  try {
                    client.enqueue(`data: ${updateMessage}\n\n`);
                  } catch (error) {
                    // Mark client for removal
                    clientsToRemove.push(client);
                  }
                });
                
                // Remove disconnected clients
                clientsToRemove.forEach(client => {
                  clients.delete(client);
                });
                
                // Stop watching if no clients are connected
                if (clients.size === 0) {
                  console.log('No clients connected, stopping watcher');
                  rustWatcher.kill();
                }
              }
            });

            rustWatcher.stderr.on('data', (data) => {
              const errorMessage = data.toString().trim();
              console.error('Rust watcher error:', errorMessage);
              
              // Don't treat RecvError as a real error - it's expected when the process is terminated
              if (!errorMessage.includes('RecvError')) {
                // Send error to clients
                const errorData = JSON.stringify({ type: 'error', message: errorMessage });
                const clientsToRemove: ReadableStreamDefaultController[] = [];
                
                clients.forEach(client => {
                  try {
                    client.enqueue(`data: ${errorData}\n\n`);
                  } catch (error) {
                    clientsToRemove.push(client);
                  }
                });
                
                clientsToRemove.forEach(client => {
                  clients.delete(client);
                });
              }
            });

            rustWatcher.on('error', (error) => {
              console.error('Rust watcher spawn error:', error);
            });

            rustWatcher.on('close', (code) => {
              console.log('Rust watcher closed with code:', code);
              // Only restart if we still have clients
              if (clients.size > 0) {
                console.log('Restarting watcher due to unexpected close');
                // The watcher will be restarted when the next client connects
              }
            });

            // Store the watcher process for cleanup
            (controller as any).rustWatcher = rustWatcher;
          } catch (error) {
            console.error('Error setting up Rust file watcher:', error);
          }
        }

        // Clean up when client disconnects
        request.signal.addEventListener('abort', () => {
          clients.delete(controller);
          
          // Stop watching if no clients are connected
          if (clients.size === 0) {
            const rustWatcher = (controller as any).rustWatcher;
            if (rustWatcher) {
              console.log('Last client disconnected, stopping watcher');
              rustWatcher.kill();
            }
          }
        });
      },
      cancel() {
        // Handle stream cancellation - this is called when the stream is cancelled
        // We can't access the specific controller here, so we'll handle cleanup in the abort event
      }
    });

    return new NextResponse(stream, {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache, no-store, must-revalidate',
        'Connection': 'keep-alive',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': 'Cache-Control, Content-Type',
        'X-Accel-Buffering': 'no', // Disable nginx buffering
      },
    });
  } catch (error) {
    console.error('SSE route error:', error);
    return new NextResponse(
      JSON.stringify({ error: 'Internal server error' }),
      {
        status: 500,
        headers: {
          'Content-Type': 'application/json',
          'Access-Control-Allow-Origin': '*',
        },
      }
    );
  }
}