// SPDX-FileCopyrightText: 2026 Kaede Akino
// SPDX-License-Identifier: GPL-3.0-only

import AgentKit
import CSSH2
import CryptoKit
import Darwin
import Foundation
import TermindApplication
import TermindDomain

nonisolated enum SFTPChannelAttempt<Value: Sendable>: Sendable {
    case value(Value)
    case again
}

nonisolated final class SFTPChannelValue: @unchecked Sendable {
    let value: Any
    init(_ value: Any) { self.value = value }
}

nonisolated enum SFTPChannelAnyAttempt: @unchecked Sendable {
    case value(SFTPChannelValue)
    case again
}

/// Type-erased access to a non-blocking libssh2 owner. Standalone, PTY and
/// Agent sessions keep their own lifetimes; this driver lets one SFTP
/// implementation take bounded turns on any of those owners.
nonisolated final class SFTPChannelDriver: @unchecked Sendable {
    typealias RawOperation = @Sendable (OpaquePointer, OpaquePointer) throws -> SFTPChannelAnyAttempt

    let identity: UUID
    // External channels (including siblings and asynchronous deinit close tasks) keep
    // the transport owner alive. The owner never retains its drivers, so this is acyclic.
    private let connectionOwner: (any AnyObject & Sendable)?
    private let openAction: @Sendable () async throws -> UUID
    private let closeAction: @Sendable (UUID) async -> Void
    private let performAction: @Sendable (UUID, @escaping RawOperation) async throws -> SFTPChannelValue

    init(
        identity: UUID,
        retaining connectionOwner: (any AnyObject & Sendable)? = nil,
        open: @escaping @Sendable () async throws -> UUID,
        close: @escaping @Sendable (UUID) async -> Void,
        perform: @escaping @Sendable (UUID, @escaping RawOperation) async throws -> SFTPChannelValue
    ) {
        self.identity = identity
        self.connectionOwner = connectionOwner
        openAction = open
        closeAction = close
        performAction = perform
    }

    func open() async throws -> UUID { try await openAction() }
    func close(_ id: UUID) async { await closeAction(id) }

    func perform<Value: Sendable>(
        _ id: UUID,
        _ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SFTPChannelAttempt<Value>
    ) async throws -> Value {
        let boxed = try await performAction(id) { libssh2Session, sftp in
            switch try operation(libssh2Session, sftp) {
            case .value(let value): .value(SFTPChannelValue(value))
            case .again: .again
            }
        }
        guard let value = boxed.value as? Value else {
            preconditionFailure("SFTP channel driver returned an unexpected value type")
        }
        return value
    }
}

private nonisolated final class SFTPChannelHandle: @unchecked Sendable {
    let pointer: OpaquePointer
    init(_ pointer: OpaquePointer) { self.pointer = pointer }
}

private nonisolated final class SFTPChannelWriteContext: @unchecked Sendable {
    let handle: OpaquePointer
    let temporary: String
    let destination: String
    let overwrite: Bool
    var handleOpen = true

    init(handle: OpaquePointer, temporary: String, destination: String, overwrite: Bool) {
        self.handle = handle
        self.temporary = temporary
        self.destination = destination
        self.overwrite = overwrite
    }
}

private nonisolated struct SFTPProgressMeter {
    let total: Int64
    let report: @Sendable (SFTPProgress) -> Void
    private(set) var completed: Int64
    private var sampledBytes: Int64 = 0
    private var sampledAt = ContinuousClock.now

    init(total: Int64, offset: Int64 = 0, report: @escaping @Sendable (SFTPProgress) -> Void) {
        self.total = total
        completed = offset
        self.report = report
    }

    mutating func add(_ bytes: Int) {
        completed += Int64(bytes)
        sampledBytes += Int64(bytes)
        let elapsed = sampledAt.duration(to: .now)
        guard elapsed >= .milliseconds(250) else { return }
        let parts = elapsed.components
        let seconds = max(0.001, Double(parts.seconds) + Double(parts.attoseconds) / 1e18)
        report(
            SFTPProgress(
                completedBytes: completed, totalBytes: total,
                bytesPerSecond: Int64(Double(sampledBytes) / seconds)))
        sampledBytes = 0
        sampledAt = .now
    }

    func finish() {
        report(SFTPProgress(completedBytes: completed, totalBytes: total, bytesPerSecond: 0))
    }
}

/// An opened SFTP subsystem channel on a multiplexed non-blocking SSH connection.
actor SFTPChannel: SFTPChannelProtocol {
    private nonisolated let driver: SFTPChannelDriver
    private let id: UUID
    private var disconnected = false
    private let charset: RemoteCharset

    static func open(
        using driver: SFTPChannelDriver, charset: RemoteCharset
    ) async throws -> any SFTPChannelProtocol {
        let id = try await driver.open()
        do {
            try Task.checkCancellation()
            return SFTPChannel(driver: driver, id: id, charset: charset)
        } catch {
            await driver.close(id)
            throw error
        }
    }

    private init(driver: SFTPChannelDriver, id: UUID, charset: RemoteCharset) {
        self.driver = driver
        self.id = id
        self.charset = charset
    }

    deinit {
        let driver = driver
        let id = id
        Task { await driver.close(id) }
    }

    func openSiblingChannel() async throws -> any SFTPChannelProtocol {
        try await Self.open(using: driver, charset: charset)
    }

    func disconnect() async {
        guard !disconnected else { return }
        disconnected = true
        await driver.close(id)
    }

    func homeDirectory() async throws -> String {
        let source = try encodePath(".")
        let charset = charset
        let value: String = try await driver.perform(id) { libssh2Session, sftp in
            var target = [CChar](repeating: 0, count: 4 * 1024)
            let count = source.withUnsafeBytes {
                libssh2_sftp_symlink_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(source.count),
                    &target, UInt32(target.count),
                    Int32(LIBSSH2_SFTP_REALPATH))
            }
            if count == LIBSSH2_ERROR_EAGAIN { return .again }
            guard count > 0 else {
                throw Self.error(libssh2Session, sftp, "Unable to resolve the remote home directory")
            }
            let bytes = Data(target.prefix(Int(count)).map { UInt8(bitPattern: $0) })
            return .value(RemotePath.normalized(try Self.decodePath(bytes, charset: charset)))
        }
        return value
    }

    func listDirectory(at path: String) async throws -> [RemoteFile] {
        let handle = try await openHandle(path: path, flags: 0, mode: 0, type: Int32(LIBSSH2_SFTP_OPENDIR))
        let charset = charset
        var files: [RemoteFile] = []
        do {
            while true {
                let next: RemoteFile? = try await driver.perform(id) { libssh2Session, sftp in
                    var name = [CChar](repeating: 0, count: 4 * 1024)
                    var attributes = LIBSSH2_SFTP_ATTRIBUTES()
                    let count = libssh2_sftp_readdir_ex(
                        handle.pointer, &name, name.count, nil, 0, &attributes)
                    if count == LIBSSH2_ERROR_EAGAIN { return .again }
                    guard count >= 0 else { throw Self.error(libssh2Session, sftp, "Unable to read \(path)") }
                    guard count > 0 else { return .value(nil) }
                    let filename = try Self.decodePath(
                        Data(name.prefix(Int(count)).map { UInt8(bitPattern: $0) }), charset: charset)
                    return .value(
                        Self.remoteFile(
                            path: RemotePath.join(path, filename), name: filename,
                            attributes: attributes))
                }
                guard let next else { break }
                if next.name != "." && next.name != ".." { files.append(next) }
            }
            try await closeHandle(handle)
        } catch {
            try? await closeHandle(handle)
            throw error
        }
        return files.sorted {
            if $0.isDirectory != $1.isDirectory { return $0.isDirectory }
            return $0.name.localizedStandardCompare($1.name) == .orderedAscending
        }
    }

    func attributes(at path: String) async throws -> RemoteFile? {
        let pathBytes = try encodePath(path)
        let value: RemoteFile? = try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            let result = pathBytes.withUnsafeBytes {
                libssh2_sftp_stat_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(pathBytes.count),
                    Int32(LIBSSH2_SFTP_LSTAT), &attributes)
            }
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            if result == 0 { return .value(Self.remoteFile(path: path, attributes: attributes)) }
            if libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE { return .value(nil) }
            throw Self.error(libssh2Session, sftp, "Unable to inspect \(path)")
        }
        return value
    }

    func readFile(at path: String, maximumBytes: Int) async throws -> Data {
        guard maximumBytes >= 0 else { throw SFTPError.operation("The inline read limit is invalid.") }
        let handle = try await openHandle(
            path: path, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        var data = Data()
        do {
            while data.count <= maximumBytes {
                let wanted = min(64 * 1024 - 1, maximumBytes - data.count) + 1
                let chunk: Data? = try await read(handle, maximumBytes: wanted)
                guard let chunk else { break }
                data.append(chunk)
            }
            try await closeHandle(handle)
        } catch {
            try? await closeHandle(handle)
            throw error
        }
        guard data.count <= maximumBytes else {
            throw SFTPError.operation("\(path) exceeds the inline read limit of \(maximumBytes) bytes.")
        }
        return data
    }

    func writeFile(_ data: Data, at path: String, overwrite: Bool) async throws {
        let temporary = path + ".termind-write-" + UUID().uuidString.lowercased()
        let context = try await beginWrite(
            temporary: temporary, destination: path, overwrite: overwrite,
            continuing: false, snapshot: nil)
        do {
            try await write(data, to: context.handle, cancellation: nil, acknowledge: nil)
            try await closeHandle(SFTPChannelHandle(context.handle))
            context.handleOpen = false
            if let existing = try await attributes(at: path), !overwrite || existing.isDirectory {
                throw SFTPStreamUploadOperation.Failure.targetConflict
            }
            try await rename(from: temporary, to: path, overwrite: overwrite)
        } catch {
            if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) }
            try? await removeFile(at: temporary)
            throw error
        }
    }

    func setPermissions(at path: String, to permissions: UInt32) async throws {
        let pathBytes = try encodePath(path)
        try await performStatus(prefix: "Unable to change permissions for \(path)") { sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            attributes.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS)
            attributes.permissions = UInt(permissions & 0o7777)
            return pathBytes.withUnsafeBytes {
                libssh2_sftp_stat_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(pathBytes.count),
                    Int32(LIBSSH2_SFTP_SETSTAT), &attributes)
            }
        }
    }

    func createDirectory(at path: String) async throws {
        let pathBytes = try encodePath(path)
        try await performStatus(prefix: "Unable to create \(path)") { sftp in
            pathBytes.withUnsafeBytes {
                libssh2_sftp_mkdir_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(pathBytes.count), 0o755)
            }
        }
    }

    func rename(from source: String, to destination: String, overwrite: Bool) async throws {
        let sourceBytes = try encodePath(source)
        let destinationBytes = try encodePath(destination)
        let flags = overwrite ? Int(LIBSSH2_SFTP_RENAME_OVERWRITE) : 0
        let _: Void = try await driver.perform(id) { libssh2Session, sftp in
            let result = sourceBytes.withUnsafeBytes { sourceBuffer in
                return destinationBytes.withUnsafeBytes { destinationBuffer in
                    let sourcePointer = sourceBuffer.baseAddress?.assumingMemoryBound(to: CChar.self)
                    let destinationPointer = destinationBuffer.baseAddress?.assumingMemoryBound(to: CChar.self)
                    return SFTPRenameOperation.perform(
                        overwrite: overwrite,
                        posixRename: {
                            libssh2_sftp_posix_rename_ex(
                                sftp, sourcePointer, sourceBytes.count,
                                destinationPointer, destinationBytes.count)
                        },
                        standardRename: {
                            libssh2_sftp_rename_ex(
                                sftp, sourcePointer, UInt32(sourceBytes.count), destinationPointer,
                                UInt32(destinationBytes.count), flags)
                        },
                        lastStatus: { libssh2_sftp_last_error(sftp) })
                }
            }
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else { throw Self.error(libssh2Session, sftp, "Unable to rename \(source)") }
            return .value(())
        }
    }

    func removeFile(at path: String) async throws {
        let pathBytes = try encodePath(path)
        try await performStatus(prefix: "Unable to delete \(path)") { sftp in
            pathBytes.withUnsafeBytes {
                libssh2_sftp_unlink_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(pathBytes.count))
            }
        }
    }

    func removeDirectory(at path: String) async throws {
        let pathBytes = try encodePath(path)
        try await performStatus(prefix: "Unable to delete \(path)") { sftp in
            pathBytes.withUnsafeBytes {
                libssh2_sftp_rmdir_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self), UInt32(pathBytes.count))
            }
        }
    }

    func uploadFile(
        from localURL: URL, to remotePath: String, overwrite: Bool,
        cancellation: TransferCancellationToken, resume: SFTPResumeState,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        if resume.snapshot.committed { return }
        let values = try localURL.resourceValues(
            forKeys: [.fileSizeKey, .contentModificationDateKey, .isRegularFileKey])
        guard values.isRegularFile == true, let size = values.fileSize else {
            throw SFTPResumeFailure.sourceChanged
        }
        let identity = SFTPSourceIdentity(byteCount: Int64(size), modifiedAt: values.contentModificationDate)
        try resume.validateSource(identity)
        let snapshot = resume.snapshot
        if try await recoverCommittedWrite(
            destination: remotePath, resume: resume, cancellation: cancellation
        ) {
            return
        }
        let file = try FileHandle(forReadingFrom: localURL)
        defer { try? file.close() }
        try Self.verifyLocalPrefix(file, snapshot: snapshot, cancellation: cancellation)
        let context = try await beginWrite(
            temporary: resume.temporaryPath(for: remotePath), destination: remotePath,
            overwrite: overwrite, continuing: snapshot.hasTemporaryFile,
            snapshot: snapshot, cancellation: cancellation)
        if !snapshot.hasTemporaryFile { resume.createdTemporaryFile() }
        do {
            var meter = SFTPProgressMeter(
                total: identity.byteCount, offset: snapshot.offset, report: progress)
            while true {
                guard !cancellation.isCancelled else { throw SFTPError.cancelled }
                let data = try file.read(upToCount: 32_700) ?? Data()
                if data.isEmpty { break }
                guard Int64(data.count) <= identity.byteCount - meter.completed else {
                    throw SFTPResumeFailure.sourceChanged
                }
                try await write(data, to: context.handle, cancellation: cancellation) { chunk in
                    resume.acknowledge(chunk)
                    meter.add(chunk.count)
                }
            }
            guard meter.completed == identity.byteCount else { throw SFTPResumeFailure.sourceChanged }
            guard try Self.localIdentity(localURL) == identity else { throw SFTPResumeFailure.sourceChanged }
            try await finishWrite(context, resume: resume)
            meter.finish()
        } catch {
            if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) }
            throw error
        }
    }

    func uploadFile(
        from localURL: URL, to remotePath: String, overwrite: Bool,
        cancellation: TransferCancellationToken,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        try await uploadFile(
            from: localURL, to: remotePath, overwrite: overwrite,
            cancellation: cancellation, resume: SFTPResumeState(), progress: progress)
    }

    func downloadFile(
        from remotePath: String, to localURL: URL, overwrite: Bool,
        cancellation: TransferCancellationToken, resume: SFTPResumeState,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        if resume.snapshot.committed { return }
        let handle = try await openHandle(
            path: remotePath, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        var handleOpen = true
        do {
            let identity = try await remoteIdentity(handle)
            try resume.validateSource(identity)
            let snapshot = resume.snapshot
            let temporary = URL(fileURLWithPath: resume.temporaryPath(for: localURL.path))
            let fm = FileManager.default
            if !snapshot.hasTemporaryFile {
                guard fm.createFile(atPath: temporary.path, contents: nil) else {
                    throw SFTPError.localFile("Unable to create \(temporary.path)")
                }
                resume.createdTemporaryFile()
            }
            let file = try FileHandle(forUpdating: temporary)
            defer { try? file.close() }
            guard try file.seekToEnd() == UInt64(snapshot.offset) else {
                throw SFTPResumeFailure.partialChanged
            }
            try file.seek(toOffset: 0)
            try Self.verifyLocalPrefix(file, snapshot: snapshot, cancellation: cancellation)
            try await verifyRemotePrefix(handle, snapshot: snapshot, cancellation: cancellation)
            try file.seek(toOffset: UInt64(snapshot.offset))
            var meter = SFTPProgressMeter(
                total: identity.byteCount, offset: snapshot.offset, report: progress)
            while let data = try await read(handle, maximumBytes: 32_700) {
                guard !cancellation.isCancelled else { throw SFTPError.cancelled }
                guard Int64(data.count) <= identity.byteCount - meter.completed else {
                    throw SFTPResumeFailure.sourceChanged
                }
                try file.write(contentsOf: data)
                resume.acknowledge(data)
                meter.add(data.count)
            }
            guard meter.completed == identity.byteCount,
                try await remoteIdentity(handle) == identity
            else { throw SFTPResumeFailure.sourceChanged }
            if resume.snapshot.capturesFileVersion {
                resume.recordCommittedVersion(
                    SFTPFileVersion(try await remoteFile(at: remotePath, from: handle)))
            }
            guard !cancellation.isCancelled else { throw SFTPError.cancelled }
            try await closeHandle(handle)
            handleOpen = false
            try file.synchronize()
            meter.finish()
            resume.willCommit()
            if fm.fileExists(atPath: localURL.path) {
                guard overwrite else {
                    throw SFTPError.localFile("\(localURL.lastPathComponent) already exists.")
                }
                _ = try fm.replaceItemAt(localURL, withItemAt: temporary)
            } else {
                try fm.moveItem(at: temporary, to: localURL)
            }
            resume.didCommit()
        } catch {
            if handleOpen { try? await closeHandle(handle) }
            throw error
        }
    }

    func downloadFile(
        from remotePath: String, to localURL: URL, overwrite: Bool,
        cancellation: TransferCancellationToken,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        try await downloadFile(
            from: remotePath, to: localURL, overwrite: overwrite,
            cancellation: cancellation, resume: SFTPResumeState(), progress: progress)
    }

    func uploadStream(
        fromSocket socket: Int32, to remotePath: String, byteCount: Int64, overwrite: Bool,
        cancellation: TransferCancellationToken, resume: SFTPResumeState,
        progress: @escaping @Sendable (SFTPProgress) -> Void,
        sourceSucceeded: @escaping @Sendable () -> Bool
    ) async throws {
        if resume.snapshot.committed { return }
        let snapshot = resume.snapshot
        if try await recoverCommittedWrite(
            destination: remotePath, resume: resume, cancellation: cancellation
        ) {
            return
        }
        let context = try await beginWrite(
            temporary: resume.temporaryPath(for: remotePath), destination: remotePath,
            overwrite: overwrite, continuing: snapshot.hasTemporaryFile,
            snapshot: snapshot, cancellation: cancellation)
        if !snapshot.hasTemporaryFile { resume.createdTemporaryFile() }
        do {
            var meter = SFTPProgressMeter(total: byteCount, offset: snapshot.offset, report: progress)
            while true {
                guard !cancellation.isCancelled else { throw SFTPError.cancelled }
                var buffer = [UInt8](repeating: 0, count: 32_700)
                let count = Darwin.recv(socket, &buffer, buffer.count, 0)
                if count < 0, errno == EINTR { continue }
                if count < 0 {
                    throw SFTPError.operation("The SFTP upload stream was interrupted.")
                }
                if count == 0 { break }
                let data = Data(buffer.prefix(count))
                guard Int64(data.count) <= byteCount - meter.completed else {
                    throw SFTPError.operation("The SFTP upload source exceeded the declared length.")
                }
                try await write(data, to: context.handle, cancellation: cancellation) { chunk in
                    resume.acknowledge(chunk)
                    meter.add(chunk.count)
                }
            }
            guard !cancellation.isCancelled, sourceSucceeded() else { throw SFTPError.cancelled }
            guard meter.completed == byteCount else {
                throw SFTPError.operation(
                    "The SFTP upload source ended before the declared length.")
            }
            try await finishWrite(context, resume: resume)
            meter.finish()
        } catch {
            if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) }
            throw error
        }
    }

    func uploadStream(
        fromSocket socket: Int32, to remotePath: String, byteCount: Int64, overwrite: Bool,
        cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void,
        sourceSucceeded: @escaping @Sendable () -> Bool
    ) async throws {
        try await uploadStream(
            fromSocket: socket, to: remotePath, byteCount: byteCount, overwrite: overwrite,
            cancellation: cancellation, resume: SFTPResumeState(), progress: progress,
            sourceSucceeded: sourceSucceeded)
    }

    func openDownloadStream(
        at path: String, maximumBytes: Int64, cancellation: TransferCancellationToken
    ) async throws -> any SFTPDownloadStream {
        let handle = try await openHandle(
            path: path, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        do {
            let identity = try await remoteIdentity(handle)
            guard identity.byteCount <= maximumBytes else { throw SFTPDownloadFailure.tooLarge }
            return SFTPChannelDownloadStream(
                client: self, handle: handle, identity: identity, cancellation: cancellation)
        } catch {
            try? await closeHandle(handle)
            throw error
        }
    }

    func copyFile(
        from sourcePath: String, to destinationPath: String,
        on destination: any SFTPChannelProtocol, byteCount: Int64, overwrite: Bool,
        cancellation: TransferCancellationToken,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        if let destination = destination as? SFTPChannel,
            destination.driver.identity == driver.identity
        {
            try await copyOnSharedTransport(
                from: sourcePath, to: destinationPath, destination: destination,
                byteCount: byteCount, overwrite: overwrite,
                cancellation: cancellation, progress: progress)
            return
        }
        let relay = try SFTPSocketRelay()
        let observer = cancellation.observeCancellation { relay.stop() }
        defer {
            cancellation.removeObserver(observer)
            relay.close()
        }
        try await withThrowingTaskGroup(of: Void.self) { group in
            group.addTask {
                do {
                    try await self.downloadToSocket(
                        sourcePath, socket: relay.writer, cancellation: cancellation)
                    relay.finishWriting()
                } catch {
                    cancellation.cancel()
                    throw error
                }
            }
            group.addTask {
                try await destination.uploadStream(
                    fromSocket: relay.reader, to: destinationPath, byteCount: byteCount,
                    overwrite: overwrite, cancellation: cancellation, progress: progress,
                    sourceSucceeded: { relay.sourceSucceeded })
            }
            do {
                while try await group.next() != nil {}
            } catch {
                cancellation.cancel()
                relay.stop()
                group.cancelAll()
                throw error
            }
        }
    }

    private func copyOnSharedTransport(
        from sourcePath: String, to destinationPath: String,
        destination: SFTPChannel, byteCount: Int64, overwrite: Bool,
        cancellation: TransferCancellationToken,
        progress: @escaping @Sendable (SFTPProgress) -> Void
    ) async throws {
        let source = try await openHandle(
            path: sourcePath, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        var sourceOpen = true
        let identity = try await remoteIdentity(source)
        guard identity.byteCount == byteCount else {
            try? await closeHandle(source)
            throw SFTPResumeFailure.sourceChanged
        }
        let temporary = destinationPath + ".termind-copy-" + UUID().uuidString.lowercased()
        let output = try await destination.beginWrite(
            temporary: temporary, destination: destinationPath, overwrite: overwrite,
            continuing: false, snapshot: nil)
        do {
            var meter = SFTPProgressMeter(total: byteCount, report: progress)
            while let data = try await read(source, maximumBytes: 64 * 1024) {
                guard !cancellation.isCancelled else { throw SFTPError.cancelled }
                guard Int64(data.count) <= byteCount - meter.completed else {
                    throw SFTPResumeFailure.sourceChanged
                }
                try await destination.write(
                    data, to: output.handle, cancellation: cancellation, acknowledge: nil)
                meter.add(data.count)
            }
            guard meter.completed == byteCount, try await remoteIdentity(source) == identity else {
                throw SFTPResumeFailure.sourceChanged
            }
            try await closeHandle(source)
            sourceOpen = false
            try await destination.closeHandle(SFTPChannelHandle(output.handle))
            output.handleOpen = false
            if let existing = try await destination.attributes(at: destinationPath),
                !overwrite || existing.isDirectory
            {
                throw SFTPStreamUploadOperation.Failure.targetConflict
            }
            try await destination.rename(from: temporary, to: destinationPath, overwrite: overwrite)
            meter.finish()
        } catch {
            if sourceOpen { try? await closeHandle(source) }
            if output.handleOpen {
                try? await destination.closeHandle(SFTPChannelHandle(output.handle))
            }
            try? await destination.removeFile(at: temporary)
            throw error
        }
    }

    private func downloadToSocket(
        _ path: String, socket: Int32, cancellation: TransferCancellationToken
    ) async throws {
        let handle = try await openHandle(
            path: path, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        do {
            while let data = try await read(handle, maximumBytes: 32_700) {
                var offset = 0
                while offset < data.count {
                    guard !cancellation.isCancelled else { throw SFTPError.cancelled }
                    let sent = data.withUnsafeBytes {
                        Darwin.send(socket, $0.baseAddress!.advanced(by: offset), data.count - offset, 0)
                    }
                    if sent > 0 {
                        offset += sent
                    } else if sent < 0, errno == EINTR {
                        continue
                    } else {
                        throw SFTPError.operation("The SFTP copy relay was interrupted.")
                    }
                }
            }
            try await closeHandle(handle)
        } catch {
            try? await closeHandle(handle)
            throw error
        }
    }

    fileprivate func streamRead(
        _ handle: SFTPChannelHandle, cancellation: TransferCancellationToken
    ) async throws -> Data? {
        guard !cancellation.isCancelled else { throw SFTPError.cancelled }
        return try await read(handle, maximumBytes: 64 * 1024)
    }

    fileprivate func streamSeek(_ handle: SFTPChannelHandle, offset: Int64) async throws {
        guard offset >= 0 else { throw SFTPResumeFailure.partialChanged }
        try await seek(handle, offset: offset)
    }

    fileprivate func streamClose(_ handle: SFTPChannelHandle) async throws {
        try await closeHandle(handle)
    }

    fileprivate func streamIdentity(_ handle: SFTPChannelHandle) async throws -> SFTPSourceIdentity {
        try await remoteIdentity(handle)
    }

    private func openHandle(
        path: String, flags: UInt, mode: Int, type: Int32
    ) async throws -> SFTPChannelHandle {
        let pathBytes = try encodePath(path)
        return try await driver.perform(id) { libssh2Session, sftp in
            let opened = pathBytes.withUnsafeBytes {
                libssh2_sftp_open_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self),
                    UInt32(pathBytes.count), flags, mode, type)
            }
            if let opened { return .value(SFTPChannelHandle(opened)) }
            if libssh2_session_last_errno(libssh2Session) == LIBSSH2_ERROR_EAGAIN { return .again }
            throw Self.error(libssh2Session, sftp, "Unable to open \(path)")
        }
    }

    private func closeHandle(_ handle: SFTPChannelHandle) async throws {
        let _: Void = try await driver.perform(id) { libssh2Session, sftp in
            let result = libssh2_sftp_close_handle(handle.pointer)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else { throw Self.error(libssh2Session, sftp, "Unable to close SFTP handle") }
            return .value(())
        }
    }

    private func read(_ handle: SFTPChannelHandle, maximumBytes: Int) async throws -> Data? {
        try await driver.perform(id) { libssh2Session, sftp in
            var buffer = [UInt8](repeating: 0, count: maximumBytes)
            let count = libssh2_sftp_read(handle.pointer, &buffer, buffer.count)
            if count == LIBSSH2_ERROR_EAGAIN { return .again }
            guard count >= 0 else { throw Self.error(libssh2Session, sftp, "Unable to read remote file") }
            return .value(count == 0 ? nil : Data(buffer.prefix(count)))
        }
    }

    private func seek(_ handle: SFTPChannelHandle, offset: Int64) async throws {
        let _: Void = try await driver.perform(id) { _, _ in
            libssh2_sftp_seek64(handle.pointer, UInt64(offset))
            return .value(())
        }
    }

    private func write(
        _ data: Data, to handle: OpaquePointer, cancellation: TransferCancellationToken?,
        acknowledge: ((Data) -> Void)?
    ) async throws {
        let stableHandle = SFTPChannelHandle(handle)
        var position = 0
        while position < data.count {
            if cancellation?.isCancelled == true { throw SFTPError.cancelled }
            let start = position
            let written: Int = try await driver.perform(id) { libssh2Session, sftp in
                let count = data.withUnsafeBytes {
                    libssh2_sftp_write(
                        stableHandle.pointer,
                        $0.baseAddress!.advanced(by: start).assumingMemoryBound(to: CChar.self),
                        data.count - start)
                }
                if count == LIBSSH2_ERROR_EAGAIN { return .again }
                guard count > 0 else { throw Self.error(libssh2Session, sftp, "Unable to write remote file") }
                return .value(count)
            }
            let chunk = data.subdata(in: position..<(position + written))
            acknowledge?(chunk)
            position += written
        }
    }

    private func performStatus(
        prefix: String, _ operation: @escaping @Sendable (OpaquePointer) -> Int32
    ) async throws {
        let _: Void = try await driver.perform(id) { libssh2Session, sftp in
            let result = operation(sftp)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else { throw Self.error(libssh2Session, sftp, prefix) }
            return .value(())
        }
    }

    private func recoverCommittedWrite(
        destination: String, resume: SFTPResumeState,
        cancellation: TransferCancellationToken
    ) async throws -> Bool {
        let snapshot = resume.snapshot
        guard snapshot.hasTemporaryFile, snapshot.committing,
            let temporary = snapshot.temporaryPath,
            try await attributes(at: temporary) == nil
        else { return false }
        let handle = try await openHandle(
            path: destination, flags: UInt(LIBSSH2_FXF_READ), mode: 0,
            type: Int32(LIBSSH2_SFTP_OPENFILE))
        do {
            guard try await remoteSize(handle) == snapshot.offset else {
                throw SFTPResumeFailure.partialChanged
            }
            try await verifyRemotePrefix(
                handle, snapshot: snapshot, cancellation: cancellation)
            resume.recordCommittedVersion(
                SFTPFileVersion(try await remoteFile(at: destination, from: handle)))
            try await closeHandle(handle)
            resume.didCommit()
            return true
        } catch {
            try? await closeHandle(handle)
            throw error
        }
    }

    private func beginWrite(
        temporary: String, destination: String, overwrite: Bool,
        continuing: Bool, snapshot: SFTPResumeState.Snapshot?,
        cancellation: TransferCancellationToken = TransferCancellationToken()
    ) async throws -> SFTPChannelWriteContext {
        if let editPolicy = snapshot?.editPolicy {
            try editPolicy.validate(try await attributes(at: destination))
        }
        if continuing, try await attributes(at: temporary) == nil {
            throw SFTPResumeFailure.partialChanged
        }
        let flags = UInt(
            LIBSSH2_FXF_READ | LIBSSH2_FXF_WRITE
                | (continuing ? 0 : (LIBSSH2_FXF_CREAT | LIBSSH2_FXF_EXCL)))
        let temporaryBytes = try encodePath(temporary)
        let context: SFTPChannelWriteContext = try await driver.perform(id) { libssh2Session, sftp in
            let handle = temporaryBytes.withUnsafeBytes {
                libssh2_sftp_open_ex(
                    sftp, $0.baseAddress?.assumingMemoryBound(to: CChar.self),
                    UInt32(temporaryBytes.count), flags, 0o600,
                    Int32(LIBSSH2_SFTP_OPENFILE))
            }
            if let handle {
                return .value(
                    SFTPChannelWriteContext(
                        handle: handle, temporary: temporary,
                        destination: destination, overwrite: overwrite))
            }
            if libssh2_session_last_errno(libssh2Session) == LIBSSH2_ERROR_EAGAIN { return .again }
            throw Self.error(libssh2Session, sftp, "Unable to create \(destination)")
        }
        if continuing, let snapshot {
            let handle = SFTPChannelHandle(context.handle)
            let size = try await remoteIdentity(handle).byteCount
            guard size >= snapshot.offset else {
                try? await closeHandle(SFTPChannelHandle(context.handle))
                throw SFTPResumeFailure.partialChanged
            }
            try await verifyRemotePrefix(
                handle, snapshot: snapshot, cancellation: cancellation)
            if size != snapshot.offset {
                try await truncate(handle, to: snapshot.offset)
            }
        }
        return context
    }

    private func finishWrite(_ context: SFTPChannelWriteContext, resume: SFTPResumeState) async throws {
        let handle = SFTPChannelHandle(context.handle)
        let snapshot = resume.snapshot
        guard try await remoteSize(handle) == snapshot.offset else {
            throw SFTPResumeFailure.partialChanged
        }
        if let editPolicy = snapshot.editPolicy {
            try await setPermissions(on: handle, to: editPolicy.permissions)
            let committed = try await remoteFile(at: context.destination, from: handle)
            resume.recordCommittedVersion(SFTPFileVersion(committed))
        }
        try await closeHandle(handle)
        context.handleOpen = false
        if let editPolicy = snapshot.editPolicy {
            try editPolicy.validate(try await attributes(at: context.destination))
        }
        if let existing = try await attributes(at: context.destination),
            !context.overwrite || existing.isDirectory
        {
            throw SFTPStreamUploadOperation.Failure.targetConflict
        }
        resume.willCommit()
        try await rename(
            from: context.temporary, to: context.destination,
            overwrite: context.overwrite)
        resume.didCommit()
    }

    private func remoteSize(_ handle: SFTPChannelHandle) async throws -> Int64 {
        try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0, attributes.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0,
                attributes.filesize <= UInt64(Int64.max)
            else { throw Self.error(libssh2Session, sftp, "Unable to inspect remote file") }
            return .value(Int64(attributes.filesize))
        }
    }

    private func truncate(_ handle: SFTPChannelHandle, to size: Int64) async throws {
        let _: Void = try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            attributes.flags = UInt(LIBSSH2_SFTP_ATTR_SIZE)
            attributes.filesize = UInt64(size)
            let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 1)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else {
                throw Self.error(libssh2Session, sftp, "Unable to trim incomplete file")
            }
            return .value(())
        }
    }

    private func setPermissions(on handle: SFTPChannelHandle, to permissions: UInt32) async throws {
        let _: Void = try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            attributes.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS)
            attributes.permissions = UInt(permissions & 0o7777)
            let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 1)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else {
                throw Self.error(
                    libssh2Session, sftp,
                    String(localized: "Unable to preserve remote file permissions"))
            }
            return .value(())
        }
    }

    private func remoteFile(
        at path: String, from handle: SFTPChannelHandle
    ) async throws -> RemoteFile {
        try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0 else {
                throw Self.error(libssh2Session, sftp, "Unable to inspect remote file")
            }
            return .value(Self.remoteFile(path: path, attributes: attributes))
        }
    }

    private func remoteIdentity(_ handle: SFTPChannelHandle) async throws -> SFTPSourceIdentity {
        try await driver.perform(id) { libssh2Session, sftp in
            var attributes = LIBSSH2_SFTP_ATTRIBUTES()
            let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0)
            if result == LIBSSH2_ERROR_EAGAIN { return .again }
            guard result == 0, attributes.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0,
                attributes.filesize <= UInt64(Int64.max),
                attributes.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0,
                attributes.permissions & 0o170000 == 0o100000
            else { throw SFTPResumeFailure.sourceChanged }
            return .value(
                SFTPSourceIdentity(
                    byteCount: Int64(attributes.filesize),
                    modifiedAt: attributes.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0
                        ? Date(timeIntervalSince1970: Double(attributes.mtime)) : nil))
        }
    }

    private func verifyRemotePrefix(
        _ handle: SFTPChannelHandle, snapshot: SFTPResumeState.Snapshot,
        cancellation: TransferCancellationToken
    ) async throws {
        try await seek(handle, offset: 0)
        var remaining = snapshot.offset
        var digest = SHA256()
        while remaining > 0 {
            guard !cancellation.isCancelled else { throw SFTPError.cancelled }
            guard let data = try await read(handle, maximumBytes: Int(min(remaining, 64 * 1024))),
                !data.isEmpty
            else { throw SFTPResumeFailure.partialChanged }
            digest.update(data: data)
            remaining -= Int64(data.count)
        }
        guard digest.finalize() == snapshot.digest.finalize() else {
            throw SFTPResumeFailure.partialChanged
        }
        try await seek(handle, offset: snapshot.offset)
    }

    private static func verifyLocalPrefix(
        _ file: FileHandle, snapshot: SFTPResumeState.Snapshot,
        cancellation: TransferCancellationToken
    ) throws {
        var remaining = snapshot.offset
        var digest = SHA256()
        while remaining > 0 {
            guard !cancellation.isCancelled else { throw SFTPError.cancelled }
            let data = try file.read(upToCount: Int(min(remaining, 64 * 1024))) ?? Data()
            guard !data.isEmpty else { throw SFTPResumeFailure.partialChanged }
            digest.update(data: data)
            remaining -= Int64(data.count)
        }
        guard digest.finalize() == snapshot.digest.finalize() else {
            throw SFTPResumeFailure.partialChanged
        }
    }

    private static func localIdentity(_ url: URL) throws -> SFTPSourceIdentity {
        let values = try url.resourceValues(
            forKeys: [.fileSizeKey, .contentModificationDateKey, .isRegularFileKey])
        guard values.isRegularFile == true, let size = values.fileSize else {
            throw SFTPResumeFailure.sourceChanged
        }
        return SFTPSourceIdentity(byteCount: Int64(size), modifiedAt: values.contentModificationDate)
    }

    private static func error(
        _ libssh2Session: OpaquePointer, _ sftp: OpaquePointer, _ prefix: String
    ) -> SFTPError {
        let sftpCode = libssh2_sftp_last_error(sftp)
        let sessionCode = libssh2_session_last_errno(libssh2Session)
        let message = "\(prefix): SFTP error \(sftpCode)"
        if [UInt(LIBSSH2_FX_NO_CONNECTION), UInt(LIBSSH2_FX_CONNECTION_LOST)].contains(sftpCode)
            || [
                LIBSSH2_ERROR_SOCKET_SEND, LIBSSH2_ERROR_SOCKET_RECV,
                LIBSSH2_ERROR_TIMEOUT, LIBSSH2_ERROR_CHANNEL_CLOSED,
            ].contains(sessionCode)
        {
            return .connectionLost(message)
        }
        return .operation(message)
    }

    private func encodePath(_ path: String) throws -> Data {
        guard !path.utf8.contains(0) else {
            throw SFTPError.operation(String(localized: "A remote path contains an invalid null character."))
        }
        return try RemoteCharsetStreamConverter.encode(path, charset: charset)
    }

    private nonisolated static func decodePath(_ data: Data, charset: RemoteCharset) throws -> String {
        try RemoteCharsetStreamConverter.decode(data, charset: charset)
    }

    private static func remoteFile(
        path: String, name: String? = nil, attributes: LIBSSH2_SFTP_ATTRIBUTES
    ) -> RemoteFile {
        let permissions = UInt32(attributes.permissions)
        let kind: RemoteFile.Kind
        switch permissions & 0o170000 {
        case 0o040000: kind = .directory
        case 0o100000: kind = .file
        case 0o120000: kind = .symbolicLink
        case 0: kind = .unknown
        default: kind = .special
        }
        return RemoteFile(
            path: path, name: name ?? URL(fileURLWithPath: path).lastPathComponent,
            kind: kind,
            size: attributes.filesize <= UInt64(Int64.max) ? Int64(attributes.filesize) : 0,
            modifiedAt: attributes.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0
                ? Date(timeIntervalSince1970: Double(attributes.mtime)) : nil,
            permissions: permissions, ownerUID: UInt32(attributes.uid), ownerGID: UInt32(attributes.gid))
    }
}

private actor SFTPChannelDownloadStream: SFTPDownloadStream {
    private let client: SFTPChannel
    private let handle: SFTPChannelHandle
    let sourceIdentity: SFTPSourceIdentity?
    let byteCount: Int64
    private let cancellation: TransferCancellationToken
    private var closed = false
    private var received: Int64 = 0
    private var eof = false

    init(
        client: SFTPChannel, handle: SFTPChannelHandle,
        identity: SFTPSourceIdentity, cancellation: TransferCancellationToken
    ) {
        self.client = client
        self.handle = handle
        sourceIdentity = identity
        byteCount = identity.byteCount
        self.cancellation = cancellation
    }

    func seek(to offset: Int64) async throws {
        guard offset >= 0, offset <= byteCount else { throw SFTPResumeFailure.partialChanged }
        try await client.streamSeek(handle, offset: offset)
        received = offset
        eof = false
    }

    func readChunk() async throws -> Data? {
        guard !eof else { return nil }
        guard let data = try await client.streamRead(handle, cancellation: cancellation) else {
            guard received == byteCount,
                try await client.streamIdentity(handle) == sourceIdentity
            else { throw SFTPDownloadFailure.sourceChanged }
            eof = true
            return nil
        }
        guard Int64(data.count) <= byteCount - received else {
            throw SFTPDownloadFailure.sourceChanged
        }
        received += Int64(data.count)
        return data
    }

    func close() async throws {
        guard !closed else { return }
        closed = true
        try await client.streamClose(handle)
    }
}