Function handle_connection_generic

src/main.rs:334–667

(
    stream: S,
    connection_info: &str,
    db_handler: Arc<DbHandler>,
)

}

async fn handle_connection_generic<S>(
    stream: S,
    connection_info: &str,
    db_handler: Arc<DbHandler>,
) -> Result<()>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
{
    let codec = PostgresCodec::new();
    let mut framed = Framed::new(stream, codec);

    // Wait for startup message
    let startup = match framed.next().await {
        Some(Ok(FrontendMessage::StartupMessage(msg))) => msg,
        Some(Ok(other)) => {
            error!("Expected startup message, got {:?}", other);
            return Err(anyhow::anyhow!("Protocol error: expected startup message"));
        }
        Some(Err(e)) => return Err(e.into()),
        None => return Err(anyhow::anyhow!("Connection closed unexpectedly")),
    };

    info!("Received startup message from {}: {:?}", connection_info, startup);

    // Extract session parameters
    let mut database = "main".to_string();
    let mut user = "postgres".to_string();

    for (key, value) in &startup.parameters {
        match key.as_str() {
            "database" => database = value.clone(),
            "user" => user = value.clone(),
            _ => {}
        }
    }

    let session = Arc::new(SessionState::new(database.clone(), user.clone()));
    let session_id = session.id;

    // Set the database handler for this session for proper lifecycle management
    session.set_db_handler(db_handler.clone()).await;

    // Create a connection for this session
    if let Err(e) = session.initialize_connection().await {
        error!("Failed to create session connection: {}", e);
        return Err(anyhow::anyhow!("Failed to create session connection: {}", e));
    }
    
    // Note: cleanup is now handled by SessionState Drop implementation
    // when the session Arc is dropped
    
    // We'll handle cleanup at the end of the function

    // Send authentication OK
    framed
        .send(BackendMessage::Authentication(AuthenticationMessage::Ok))
        .await?;