ftp_session.cpp

#include "ftp_session.h"
#include "server_utils.h"
#include "ftp_codes.h"
#include "peer_manager.h"
#include "peer_client.h"

#include <vector>
#include <string>
#include <optional>
#include <iostream>
#include <filesystem>
#include <sstream>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/stat.h>

using namespace srv;
namespace fs = std::filesystem;

FTPSession::FTPSession(int control_fd, const std::string& jail_root,
                       PeerManager* peers, const std::string& our_ip,
                       uint16_t listen_port)
: cfd(control_fd),
  jail(jail_root),
  cwd_rel("/"),
  authed(false),
  pasv_listen_fd(-1),
  pasv_port(0),
  peer_mgr(peers),
  our_ipv4(our_ip),
  is_peer_session(false),
  server_port(listen_port) {}


FTPSession::~FTPSession() {
    reset_pasv();
    if (cfd >= 0) 
        ::close(cfd);
}

void FTPSession::reset_pasv() {
    if (pasv_listen_fd >= 0) {
        close_fd(pasv_listen_fd);
    }
    pasv_listen_fd = -1;
    pasv_port = 0;

    // When we invalidate PASV also drop any active peer binding
    // active_peer.reset();
    // active_peer_ip.clear();
    // pasv_peer_ip_pending.reset();
    // Removed due to resetting activepeerip on every PASV causing issues

    peer_pasv_ready = false;
}

bool FTPSession::ensure_auth() {
    if (!authed) {
        send_reply(cfd, ftp::format_reply(ftp::Code::NotLoggedIn)); // 530
        return false;
    }
    return true;
}

bool FTPSession::cmd_USER(const std::vector<std::string>& args) {
    if (args.size() != 2) { 
        send_reply(cfd, ftp::format_reply(ftp::Code::SyntaxError)); 
        return true; 
    }
    std::string user = args[1];
    if (user.rfind("peer@", 0) == 0) {
        sockaddr_in addr{}; socklen_t len = sizeof(addr);
        if (::getpeername(cfd, (sockaddr*)&addr, &len) == 0) {
            char buf[INET_ADDRSTRLEN]; inet_ntop(AF_INET, &addr.sin_addr, buf, sizeof(buf));
            if (user.substr(5) == std::string(buf)) { 
                authed = true;
                is_peer_session = true;
                send_reply(cfd, ftp::format_reply(ftp::Code::LoggedIn)); 
                return true; 
            }
        }
        send_reply(cfd, ftp::format_reply(ftp::Code::NotLoggedIn));
        return false;
    }
    std::string up = user;
    to_upper(up);
    if (up == "ANONYMOUS") { 
        authed = true; 
        send_reply(cfd, ftp::format_reply(ftp::Code::LoggedIn)); 
        return true; 
    }
    send_reply(cfd, ftp::format_reply(ftp::Code::NotLoggedIn, "Authentication failed."));
    return false;
}

bool FTPSession::cmd_PWD(const std::vector<std::string>&) {
    if (!ensure_auth()) 
        return true; // do not close
    // keep session but reset pasv
    reset_pasv();
    

    
    if (!active_peer_ip.empty()) { 
        // In peer mode

        // TODO --- DEBUG
        std::cout << "PWD: Remote Path Detected" << std::endl;



        std::string path = remote_cwd_rel;
        if (path.empty()) 
            path = "/";
        // 257
        send_reply(cfd, ftp::format_reply(ftp::Code::PathCreated, " \"" + path + "\" is the current directory.")); // 257
        return true;
    }

    std::string path = cwd_rel;
    if (path.empty()) 
        path = "/";
    // 257
    send_reply(cfd, ftp::format_reply(ftp::Code::PathCreated, " \"" + path + "\" is the current directory.")); // 257
    return true;
}

bool FTPSession::cmd_CWD(const std::vector<std::string>& args) {
    if (!ensure_auth()) 
        return true;
    reset_pasv();

    if (args.size() < 2) {
        send_reply(cfd, ftp::format_reply(ftp::Code::SyntaxError)); // 501
        return true;
    }

    const std::string target = args[1];
    peer_pasv_ready = false; // any CWD invalidates peer PASV

    if (target.empty() || target == "." ) {
        // means nothing
        // but keep state consistent
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    }
    // Replaced above following if statement to account for ANY absolute root is ALWAYS back to local root
    // Absolute root means "back to local root"
    if (target == "/" || (cwd_rel == "/" && target == ".." && remote_cwd_rel == "/")) { // If we're not in peer AND at root performing ..
        active_peer_ip.clear();
        if (active_peer) 
            active_peer.reset();
        remote_cwd_rel = "/";
        cwd_rel = "/";
        last_dir_map.clear();
        last_dir_key = cwd_rel;
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    }

        // Peer mode fix to exit back to local 
    if (!active_peer_ip.empty()) {

        // Parent ".." if this would land at peer "/"
        // exit peer mode to local root
        if (target == "..") {
            fs::path curR = fs::path(remote_cwd_rel.empty() ? "/" : remote_cwd_rel); 
            fs::path parentR = curR.parent_path();
            std::string relR = parentR.string();
            if (relR.empty()) 
                relR = "/";

            if (relR == "/") {
                // leaving peer namespace entirely
                active_peer_ip.clear();
                if (active_peer) 
                    active_peer.reset();
                remote_cwd_rel = "/";
                cwd_rel = "/";
                last_dir_map.clear();
                last_dir_key = cwd_rel;
                send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
                return true;
            } else {
                // still within peer
                remote_cwd_rel = (relR[0] == '/') ? relR : ("/" + relR);
                last_dir_map.clear();
                last_dir_key = remote_cwd_rel;
                send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
                return true;
            }
        }
    }

    // Try local 
    if (auto abs = resolve_path_in_jail(jail, cwd_rel, target)) {
        if (fs::exists(abs.value())) {
            if (!fs::is_directory(abs.value())) { // exists, but not a dir
                send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
                return true;
            }
            cwd_rel = normalize_rel_path(cwd_rel, target);
            active_peer_ip.clear(); // leave peer mode
            if (active_peer)        // added to prevent double cwd .. spam bug
                active_peer.reset();
            remote_cwd_rel = "/";
            send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
            return true;
        }
    }

    // Not local — try peers
    if (!peer_mgr) { // No peers to check
        std::cout << "CWD: no peers to check - File unavail" << std::endl; // TODO : REMOVE
        send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); 
        return true; 
    }

    // I'm a peer and checked locally already 
    if (is_peer_session) {
        // In a peer session we do not federate further
        std::cout << "CWD: I am a peer and will not federate - File unavail" << std::endl; // TODO : REMOVE
        send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
        return true;
    }

    const uint16_t port = server_port; // listen port
    for (const auto& ip : peer_mgr->peers()) {
        if (ip == our_ipv4) // skip
            continue;
        PeerClient pc(ip, port, our_ipv4);
        if (!pc.good()) 
            continue;
        if (!pc.cwd(remote_cwd_rel)) 
            continue; // sync peer to our tracked peer dir
        std::string next = normalize_rel_path(remote_cwd_rel, target);
        if (!pc.cwd(next)) 
            continue; // test if dir exists on that peer

        // success — switch to peer mode
        std::cout << "CWD switched to peer " << ip << " dir " << next << std::endl; // TODO : REMOVE
        active_peer_ip = ip;
        remote_cwd_rel = next;
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    }

    send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
    return true;
}

bool FTPSession::cmd_CDUP(const std::vector<std::string>&) { // TODO: handle peer mode
    if (!ensure_auth()) 
        return true;
    // from lecture notes: clear prev eph ports
    reset_pasv();
    peer_pasv_ready = false; // any CWD invalidates peer PASV

    if (active_peer_ip.empty()) {
        // local mode
        

        if (cwd_rel == "/" || cwd_rel.empty()) {
            send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
            return true;
        }
        // go to parent
        fs::path cur = fs::path(jail) / fs::path(cwd_rel.substr(1));
        fs::path parent = cur.parent_path();
        if (parent == fs::path(jail)) {
            cwd_rel = "/";
        } else {
            // compute rel
            fs::path jailp = fs::weakly_canonical(fs::path(jail));
            fs::path newp = fs::weakly_canonical(parent);
            std::string rel = "/"+ newp.string().substr(jailp.string().size()+1);
            for (auto& ch: rel) {
                if (ch=='\\') 
                    ch='/';
            }
            if (rel.empty()) 
                rel = "/";
            if (rel[0] != '/') 
                rel = "/"+rel;
            cwd_rel = rel;
        }
        last_dir_map.clear();
        last_dir_key = cwd_rel;
        active_peer_ip.clear(); // leave peer mode
        active_peer.reset();
        remote_cwd_rel = "/";
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    } // Peer mode

    fs::path curR = fs::path(remote_cwd_rel.empty() ? "/" : remote_cwd_rel); 
    fs::path parentR = curR.parent_path();
    std::string relR = parentR.string();
    if (relR.empty()) 
        relR = "/";

    if (relR == "/") {
        // leaving peer namespace entirely
        active_peer_ip.clear();
        if (active_peer) 
            active_peer.reset();
        remote_cwd_rel = "/";
        cwd_rel = "/";
        last_dir_map.clear();
        last_dir_key = cwd_rel;
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    } else {
        // still within peer
        remote_cwd_rel = (relR[0] == '/') ? relR : ("/" + relR);
        last_dir_map.clear();
        last_dir_key = remote_cwd_rel;
        send_reply(cfd, ftp::format_reply(ftp::Code::FileActionOK)); // 250
        return true;
    }

}

bool FTPSession::cmd_QUIT(const std::vector<std::string>&) {
    send_reply(cfd, ftp::format_reply(ftp::Code::Goodbye)); // 221
    return false; // close
}

bool FTPSession:: cmd_PASV(const std::vector<std::string>&) {
    if (!ensure_auth()) 
        return true;

    // If we're in peer mode but no relay is queued yet
    if (!active_peer_ip.empty() && !pasv_peer_ip_pending) {
        pasv_peer_ip_pending = active_peer_ip;
    }

    // If we scheduled a peer PASV from find_peer_for_file / remote LIST
    if (pasv_peer_ip_pending && peer_mgr) {
        // Only one PASV at a time – ensure local one is closed
        reset_pasv();

        const uint16_t port = server_port;                 
        const std::string ip = *pasv_peer_ip_pending;

        std::unique_ptr<PeerClient> pc(new PeerClient(ip, port, our_ipv4));
        if (pc->good() && pc->login_peer(/*our_ipv4*/) && pc->cwd(remote_cwd_rel)) {
            if (auto pasv_info = pc->pasv()) {
                auto [ip_tuple, p1, p2] = *pasv_info;

                active_peer       = std::move(pc);
                active_peer_ip    = ip;
                peer_pasv_ready   = true; //  arm peer-mode LIST/RETR
                pasv_peer_ip_pending.reset();

                // Relay peer's 227 back to our client
                send_reply(cfd, ftp::format_pasv_reply(ip_tuple, p1, p2));
                return true;
            }
        }

        // Peer PASV failed
        // clear and fall through to local PASV
        pasv_peer_ip_pending.reset();
        peer_pasv_ready = false;
        active_peer.reset();
        active_peer_ip.clear();
    }

    // DEFAULT: local PASV handling
    reset_pasv(); // close previous if any
    peer_pasv_ready = false; // not a peer PASV path
    auto ip_opt = get_local_ipv4(cfd);
    if (!ip_opt) { 
        send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425
        return true; 
    }

    auto [lfd, port] = open_passive_listener(cfd);
    if (lfd < 0) { 
        send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData));  // 425 
        return true; 
    }

    pasv_listen_fd = lfd;
    pasv_port      = port;

    std::string ip_tuple = ipv4_to_tuple(*ip_opt);
    uint8_t p1 = (uint8_t)(port / 256), p2 = (uint8_t)(port % 256);
    send_reply(cfd, ftp::format_pasv_reply(ip_tuple, p1, p2));
    return true;
}

bool FTPSession::cmd_LIST(const std::vector<std::string>& args) {
    if (!ensure_auth()) 
        return true;

    // In local mode we need a local passive listener
    // in peer mode we need the peer PASV armed
    if (active_peer_ip.empty()) {
        if (pasv_listen_fd < 0/* && !is_peer_session*/) { 
            std::cout << "LIST(local): no local PASV listener\n"; // DEBUG
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425 Use PASV first
            return true;
        }
    } else {
        if (!peer_pasv_ready) {
            std::cout << "LIST(peer): peer PASV not armed\n"; // DEBUG
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425 Use PASV first
            return true;
        }
    }



    // Do this if we're not in peer mode
    if (active_peer_ip.empty()) {

        

        // Resolve target dir or file 
        // Default is current dir
        std::string target = (args.size() >= 2) ? args[1] : ".";
        auto abs = resolve_path_in_jail(jail, cwd_rel, target);
        if (!abs) {
            reset_pasv();
            send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
            return true;
        }

        send_reply(cfd, ftp::format_150_file_status_ok()); // 150
        int dfd = accept_one(pasv_listen_fd);
        reset_pasv();

        if (dfd < 0) {
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425
            return true;
        }

        // Local listing
        std::string payload;
        if (fs::is_directory(abs.value())) 
            payload = ls_long_directory(abs.value());
        else { 
            std::string name = fs::path(abs.value()).filename().string(); 
            payload = ls_long_entry(abs.value(), name); 
        }

        std::vector<std::string> lines;
        { 
            std::string cur; 
            for (char c : payload) { 
                if (c=='\r') 
                    continue; 
                if (c=='\n') { 
                    if (!cur.empty()) {
                        lines.push_back(cur); 
                        cur.clear(); 
                    } 
                } 
                else 
                    cur.push_back(c); 
            } 
            if (!cur.empty()) 
                lines.push_back(cur); 
        }
        for (auto& s : lines) 
            s = (is_peer_session) ? rewrite_ls_owner_only(s, "peer") : rewrite_ls_owner_only(s, "local"); 
            // Edge case for when we're serving list to a peer client

        last_dir_map.clear(); 
        last_dir_key = cwd_rel;
        for (const auto& s : lines) {
            auto pos = s.find_last_of(' '); 
            if (pos != std::string::npos) 
                last_dir_map[s.substr(pos+1)] = "local"; 
        }

        // Peer listings
        if (!is_peer_session && peer_mgr) {
            // TODO : REMOVE
            auto linesPeer = gather_peer_listings(cwd_rel);
            std::cout << "Gathering peer listings for LIST. Count = " << linesPeer.size() << "\n";
            for (const auto& s : linesPeer) {
                std::cout << "  " << s << "\n";
            }
            std::cout << std::endl << "END PEER PRINT LIST" << std::endl;
            // END TODO


            auto peer_lines = gather_peer_listings(cwd_rel);
            for (auto& s : peer_lines) {
                s = rewrite_ls_owner_only(s, "peer");
                auto pos = s.find_last_of(' '); 
                if (pos != std::string::npos) { 
                    std::string name = s.substr(pos+1); 
                    if (!last_dir_map.count(name)) 
                        last_dir_map[name] = "peer"; 
                }
                lines.push_back(s);
            }
        }


        std::string out;
        for (auto& l : lines) { 
            out += l; 
            out += "\r\n"; 
        }
        send_all(dfd, out.data(), out.size());
        close_fd(dfd);
        send_reply(cfd, ftp::format_reply(ftp::Code::ClosingData)); // 226
        return true;
    }
    // Otherwise we're in peer mode (remote dir)
    // Relay LISt to peer
    // Require that we previously relayed a peer PASV to the client
    if (!peer_pasv_ready) { // set in cmd_PASV
        send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425 Use PASV first
        std::cout << "LIST in peer mode without PASV ready" << std::endl; // TODO : REMOVE
        return true;
    }

    // 2) Ensure we have a live PeerClient to the active peer
    if (!active_peer || !active_peer->good()) {
        active_peer = std::make_unique<PeerClient>(active_peer_ip, server_port, our_ipv4);
        if (!active_peer->good()) {
            send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
            return true;
        }
        if (!active_peer->login_peer(/*our_ipv4*/)) { // USER peer@<A_ip> / 230
            send_reply(cfd, ftp::format_reply(ftp::Code::NotLoggedIn)); // 530
            return true;
        }
    }

    // DONT CWD here as this causes remote peer to call reset_pasv() on itself
    // We already sync in cmd_PASV with cwd_rel
    // 3) Sync remote cwd - so LIST uses the right directory on the peer
    // if (!active_peer->cwd(remote_cwd_rel)) { // remote_cwd_rel: "/…"
    //     send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
    //     return true;
    // }

    // 4) Forward LIST to the peer 
    // no local data accept — data flows peer to client
    const std::string arg = (args.size() >= 2) ? args[1] : "";
    if (!active_peer->send_cmd(arg.empty() ? "LIST" : "LIST " + arg)) {
        send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
        return true;
    }

    // 5) Relay the peer's control replies 
    PeerClient::Reply r1 = active_peer->read_reply();        // expect 150 Opening ...
    send_line(cfd, r1.raw + "\r\n");                         // relay as is
    if (r1.code != 150 && r1.code != 125) 
        return true; // peer said no

    PeerClient::Reply r2 = active_peer->read_reply(); 
    send_line(cfd, r2.raw + "\r\n");
    return true;


}


bool FTPSession::is_local_file(const std::string& rel_path) {
    auto abs = resolve_path_in_jail(jail, cwd_rel, rel_path);
    return abs && fs::exists(abs.value()) && fs::is_regular_file(abs.value());
}

std::optional<std::string> FTPSession::find_peer_for_file(const std::string& rel_path) {
    sockaddr_in addr{}; 
    socklen_t len = sizeof(addr); 
    uint16_t port = server_port;
    // if (::getsockname(cfd, (sockaddr*)&addr, &len) == 0) 
    //     port = ntohs(addr.sin_port);

    std::vector<std::string> peer_ips = peer_mgr ? peer_mgr->peers() : std::vector<std::string>{};
    for (const auto& ip : peer_ips) {
        PeerClient pc(ip, port, our_ipv4);
        if (!pc.good()) 
            continue;
        if (!pc.cwd(cwd_rel)) 
            continue;
        if (pc.file_exists(rel_path)) 
            return ip;
    }

    return std::nullopt;
}


bool FTPSession::cmd_RETR(const std::vector<std::string>& args) {
    if (!ensure_auth()) 
        return true;
    if (args.size() < 2) { 
        send_reply(cfd, ftp::format_reply(ftp::Code::SyntaxError)); // 501
        return true;
    }

    const std::string rel = args[1];

    // If this control session is a peer session eg 
    // never federate further-only serve local files
    if (is_peer_session) {
        if (!is_local_file(rel)) {
            send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
            return true;
        }
        // fall through to local transfer logic below
    }

    // ---------------------------------------------------------------------
    // LOCAL TRANSFER 
    // only when not in peer mode and file is local
    // ---------------------------------------------------------------------
    if (active_peer_ip.empty() && is_local_file(rel)) {
        if (pasv_listen_fd < 0) {
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425 Use PASV first
            return true; 
        }
        auto abs = resolve_path_in_jail(jail, cwd_rel, rel);
        if (!abs) {
            reset_pasv();
            send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
            return true;
        }

        send_reply(cfd, ftp::format_150_opening_binary()); // 150
        int dfd = accept_one(pasv_listen_fd); // wait for data pipe
        reset_pasv(); // only one ephemeral port at a time
        if (dfd < 0) { 
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425
            return true;
        }

        int rc = stream_file_to_fd(abs.value(), dfd);
        close_fd(dfd);
        if (rc == 0) 
            send_reply(cfd, ftp::format_reply(ftp::Code::ClosingData, "Transfer complete.")); // 226
        else 
            send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
        return true;
    }

    // ---------------------------------------------------------------------
    // REMOTE TRANSFER 
    // relay through active peer 
    // OR find the correct peer
    // ---------------------------------------------------------------------

    // Decide which peer to use first
    // - If already in peer namespace CWD into remote and use that peer
    // - Otherwise
    // - discover the peer that has this file
    auto choose_target_peer = [&](const std::string& path)->std::optional<std::string>{
        if (!active_peer_ip.empty()) 
            return active_peer_ip;
        return find_peer_for_file(path);
    };

    auto peer_ip_opt = choose_target_peer(rel);
    if (!peer_ip_opt) {
        send_reply(cfd, ftp::format_reply(ftp::Code::FileUnavailable)); // 550
        return true;
    }
    const std::string peer_ip = *peer_ip_opt;

    // Determine which directory to use on the peer
    // - If already in peer mode use remote_cwd_rel
    // - If we're still local - CWD on the peer to the dirname of rel 
    std::string peer_dir = remote_cwd_rel.empty() ? "/" : remote_cwd_rel;
    const bool was_local_namespace = active_peer_ip.empty(); // remember if we were local at call time
    if (active_peer_ip.empty()) {
        fs::path p(rel);
        peer_dir = p.has_parent_path() ? ("/" + p.parent_path().string()) : "/";
        if (!peer_dir.empty() && peer_dir[0] != '/') 
            peer_dir = "/" + peer_dir;
    }

    // db / retr edge case for remote root retr
    // we want to revert back to local AFTER the transfer completes
    // We arm this here so it persists across the PASV RETR interaction
    if (was_local_namespace && cwd_rel == "/") {
        revert_after_retr = true;
    }


    // Arm peer PASV if needed 
    // or if switching peers. Then we relay the peers 227 to the client
    // and return immediately
    // client will connect to the peer and re-issue a RETR as seen in lecture notes
    if (active_peer_ip != peer_ip || !peer_pasv_ready) {
        reset_pasv(); // close any local PASV; we will use peer PASV
        peer_pasv_ready = false;

        std::unique_ptr<PeerClient> pc(new PeerClient(peer_ip, server_port, our_ipv4));
        if (!(pc->good() && pc->login_peer() && pc->cwd(peer_dir))) {
            revert_after_retr = false; // clear any revert state
            send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
            return true;
        }
        auto pasv_info = pc->pasv();
        if (!pasv_info) {
            revert_after_retr = false;
            send_reply(cfd, ftp::format_reply(ftp::Code::CannotOpenData)); // 425
            return true;
        }

        auto [ip_tuple, p1, p2] = *pasv_info;
        active_peer     = std::move(pc);
        active_peer_ip  = peer_ip;
        remote_cwd_rel  = peer_dir; // remember where PASV was armed
        peer_pasv_ready = true;

        // Relay the peers 227 to the client so it connects to the correct host/port
        send_reply(cfd, ftp::format_pasv_reply(ip_tuple, p1, p2));
        return true; // client will connect and send RETR again
    }

    // By here correct peer is armed peer_pasv_ready = true so forward RETR
    // Ensure peer CWD matches where we armed PASV 
    // but don't CWD after PASV on the peer or it will close its listener with pasvreset()
    // Compute filename to RETR on the peer 

    std::string fname = rel;
    if (active_peer_ip.empty()) {
        // Not in peer mode but we CWDed the peer using dirname(rel)
        fs::path pp(rel);
        if (pp.has_filename()) 
            fname = pp.filename().string();
    } else {
        // Already in peer mode
        fs::path pp(rel);
        if (pp.has_filename()) 
            fname = pp.filename().string();
    }

    if (!active_peer) {
        // Should not ever happen anyway
        active_peer = std::make_unique<PeerClient>(peer_ip, server_port, our_ipv4);
        if (!(active_peer->good() && active_peer->login_peer())) {
            revert_after_retr = false;
            send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
            return true;
        }
    }

    // Forward RETR to peer and relay control replies 
        if (!active_peer->send_cmd("RETR " + fname)) {
            revert_after_retr = false;
            send_reply(cfd, ftp::format_reply(ftp::Code::LocalError)); // 451
            return true;
        }


    // Expect 150/125 
    auto r1 = active_peer->read_reply();
    if (!r1.raw.empty()) 
        send_line(cfd, r1.raw);
    if (r1.code != 150 && r1.code != 125) {
        revert_after_retr = false;
        return true;
    }

    auto r2 = active_peer->read_reply();
    if (!r2.raw.empty()) 
        send_line(cfd, r2.raw);



        // If this was a remote RETR while our local cwd is "/"
        // go back to local namespace now
    if (revert_after_retr) {
        active_peer_ip.clear();
        if (active_peer) 
            active_peer.reset();
        peer_pasv_ready = false;
        remote_cwd_rel  = "/";
        
        revert_after_retr = false;
    }

    return true;
}


// Old replace owner use new function below
std::string FTPSession::replace_owner_with(const std::string& ls_line, const std::string& owner_token) {
    int spaces = 0;
    size_t i = 0, start_owner = std::string::npos, end_owner = std::string::npos;
    while (i < ls_line.size()) {
        if (ls_line[i] == ' ') {
            while (i < ls_line.size() && ls_line[i]==' ') 
                i++;
            spaces++;
            if (spaces == 2) 
                start_owner = i;
            else if (spaces == 3) { 
                end_owner = i; 
                break; 
            }
        } 
        else { 
            i++; 
        }
    }
    if (start_owner == std::string::npos || end_owner == std::string::npos)
        return ls_line;
    std::string out = ls_line.substr(0, start_owner) + owner_token + ls_line.substr(end_owner);
    return out;
} // end replace_owner_with 
// Use new function below

// Rewrites only the owner field (3rd token in ls -l)
// preserves group/spacing
std::string FTPSession::rewrite_ls_owner_only(const std::string& line, const std::string& new_owner) {
    // EXPECTED: perms links owner group size mon day time name
    std::istringstream iss(line);
    std::string perms, links, owner, group, size, mon, day, time_or_year;
    if (!(iss >> perms >> links >> owner >> group >> size >> mon >> day >> time_or_year)) {
        return line; // not a standard ls -l line so leave as-is
    }
    std::string rest;
    std::getline(iss, rest);

    std::ostringstream out;
    out << perms << ' '
        << links << ' '
        << new_owner << ' '
        << group << ' '
        << size << ' '
        << mon << ' '
        << day << ' '
        << time_or_year
        << rest; 
    return out.str();
}


std::vector<std::string> FTPSession::gather_peer_listings(const std::string& rel_dir) {
    std::vector<std::string> all;
    if (!peer_mgr) 
        return all;
    sockaddr_in addr{}; 
    socklen_t len = sizeof(addr); 
    uint16_t port = server_port;
    // if (::getsockname(cfd, (sockaddr*)&addr, &len) == 0) 
    //     port = ntohs(addr.sin_port);
    auto peers = peer_mgr->peers();

    for (const auto& ip : peers) {
        if (ip == our_ipv4) // Safeguard to not list ourselves
            continue;
        PeerClient pc(ip, port, our_ipv4);
        if (!pc.good()) 
            continue;
        if (!pc.cwd(rel_dir)) 
            continue;
        auto lines = pc.list();
        all.insert(all.end(), lines.begin(), lines.end());
    }

    return all;
}

bool FTPSession::handle_line(const std::string& raw) {
    std::string line = raw;
    rtrim_crlf(line);
    if (line.empty()) 
        return true;

    // Keep original for args
    // a separate uppercased copy for verb match
    auto parts = split_ws(line);
    if (parts.empty()) 
        return true;

    std::string verb = parts[0];
    to_upper(verb);

    if (verb == "USER") return cmd_USER(parts);
    if (verb == "PWD")  return cmd_PWD(parts);
    if (verb == "CWD")  return cmd_CWD(parts);
    if (verb == "CDUP") return cmd_CDUP(parts);
    if (verb == "QUIT") return cmd_QUIT(parts);
    if (verb == "PASV") return cmd_PASV(parts);
    if (verb == "LIST") return cmd_LIST(parts);
    if (verb == "RETR") return cmd_RETR(parts);

    send_reply(cfd, ftp::format_reply(ftp::Code::NotImplemented)); // 502
    return true;
}

void FTPSession::run() {
    send_reply(cfd, ftp::format_reply(ftp::Code::SystemReady)); // 220
    while (true) {
        auto opt = recv_line(cfd);
        if (!opt) 
            break;
        if (!handle_line(*opt)) 
            break;
    }
}