usm/src/lib/Repository/CurlRepositoryClient.vala

namespace Usm {

    /**
     * A {@link RepositoryClient} that fetches over HTTP/HTTPS by shelling
     * out to `curl`.
     *
     * GIO cannot fetch `http(s)://` URIs on platforms whose GIO build has
     * no TLS backend (musl/alpine images report "Operation not supported"),
     * so {@link Repository.get_client} routes remote schemes here while
     * `file://` stays on {@link GioRepositoryClient}.
     *
     * Each transfer runs `curl -fsSL --max-time {@link timeout_seconds} --
     * <url>` and streams stdout into a `<destination>.part` file, counting
     * bytes as they are written: `-s` keeps curl completely silent during
     * the transfer and `-S` lets genuine failures print one error line to
     * stderr, which is captured and surfaced through the thrown error.
     * Progress reports carry the exact byte counts — current from the
     * pump's own counter, total from a `Content-Length` probe before the
     * transfer — and the finished part file is moved over the destination
     * only after curl exits successfully.
     */
    public class CurlRepositoryClient : RepositoryClient {

        /** Default {@link timeout_seconds} when no configuration is supplied. */
        public const int DEFAULT_TIMEOUT_SECONDS = 300;

        /**
         * Seconds a single transfer (and the size probe before it) may run
         * before curl aborts it; {@link Repository.get_client} seeds this
         * from {@link Configuration.download_timeout}.
         */
        public int timeout_seconds { get; set; default = DEFAULT_TIMEOUT_SECONDS; }

        public CurlRepositoryClient(Repository repo, int timeout = DEFAULT_TIMEOUT_SECONDS) {
            repository_config = repo;
            timeout_seconds = timeout;
        }

        public override void download_repository_listing(string path, RepositoryClientProgressCallback? callback = null) throws Error {
            download("PACKAGES.usml", path, callback);
        }

        public override void download_package(string path, RepositoryListingEntry package, RepositoryClientProgressCallback? callback = null) throws Error {
            download(package.path, path, callback);
        }

        private void download(string remote_path, string destination, RepositoryClientProgressCallback? callback) throws Error {
            var url = get_for_path(remote_path).to_string();
            var filename = Path.get_basename(remote_path);
            var part_path = @"$destination.part";

            // Exact total up front where the server states one; transfers
            // without Content-Length report progress against 0 until the
            // final report
            var total_bytes = content_length(url);

            var launcher = new SubprocessLauncher(SubprocessFlags.STDOUT_PIPE | SubprocessFlags.STDERR_PIPE);
            var argv = new string[] {
                "curl", "-fsSL",
                "--max-time", timeout_seconds.to_string(),
                "--", url
            };
            Subprocess process;
            try {
                process = launcher.spawnv(argv);
            }
            catch(Error e) {
                throw new IOError.NOT_FOUND(@"Unable to launch curl to fetch \"$url\" (is curl installed?): $(e.message)");
            }

            var output = File.new_for_path(part_path).replace(null, false, FileCreateFlags.REPLACE_DESTINATION);
            var loop = new MainLoop();
            uint64 downloaded = 0;
            var failure = "";
            var errors = new StringBuilder();
            // Both pipes must drain before curl is waited on: stdout closes
            // at end-of-body, while an error line can still land on stderr
            var pumps_running = 2;

            ProgressEmitter report = () => {
                if(callback == null) {
                    return;
                }
                callback(filename, (int64)downloaded, (int64)total_bytes);
            };

            ChunkConsumer consume_body = (data, length) => {
                var chunk = data[0:length];
                size_t written;
                try {
                    output.write_all(chunk, out written);
                    downloaded += written;
                    report();
                }
                catch(Error e) {
                    failure = @"writing \"$part_path\": $(e.message)";
                }
            };

            Completion finished = () => {
                pumps_running--;
                if(pumps_running > 0) {
                    return;
                }
                process.wait_check_async.begin(null, (source, result) => {
                    try {
                        process.wait_check_async.end(result);
                    }
                    catch(Error e) {
                        failure = failure.length > 0 ? failure : e.message;
                    }
                    loop.quit();
                });
            };

            pump(process.get_stderr_pipe(), (data, length) => {
                // -sS leaves stderr error text only, and only failures talk
                if(errors.len < 512) {
                    errors.append_len((string)data, length);
                }
            }, finished);

            pump(process.get_stdout_pipe(), consume_body, () => {
                try {
                    output.close();
                }
                catch(Error e) {
                    failure = @"closing \"$part_path\": $(e.message)";
                }
                finished();
            });

            loop.run();

            if(failure.length > 0 || errors.len > 0 && !process.get_successful()) {
                try {
                    output.close();
                }
                catch(Error e) {
                }
                try {
                    File.new_for_path(part_path).delete();
                }
                catch(Error e) {
                }
                var detail = errors.len > 0 ? ": " + errors.str.strip() : "";
                throw new IOError.FAILED(@"curl failed while fetching \"$url\": $(failure.length > 0 ? failure : "transfer error")$detail");
            }

            // A transfer without Content-Length can only state its total
            // once every byte is on disk
            if(total_bytes == 0) {
                total_bytes = downloaded;
            }
            report();
            File.new_for_path(part_path).move(File.new_for_path(destination), FileCopyFlags.OVERWRITE);
        }

        private delegate void ProgressEmitter();

        private delegate void ChunkConsumer(uint8[] data, ssize_t length);

        private delegate void Completion();

        /** Reads a pipe to EOF in 8 KiB chunks, handing each chunk to {@link consume} before {@link completed}. */
        private async void pump(owned InputStream pipe, owned ChunkConsumer consume, owned Completion completed) {
            while(true) {
                var buffer = new uint8[8192];
                ssize_t read;
                try {
                    read = yield pipe.read_async(buffer);
                }
                catch(Error e) {
                    break;
                }
                if(read <= 0) {
                    break;
                }
                consume(buffer, read);
            }
            completed();
        }

        /**
         * The URL's Content-Length via `curl -I`, or 0 when the server does
         * not state one. The probe is silent: headers are consumed from the
         * pipe and any stderr is drained and discarded — a failed probe
         * merely leaves the total unknown.
         */
        private uint64 content_length(string url) {
            try {
                var launcher = new SubprocessLauncher(SubprocessFlags.STDOUT_PIPE | SubprocessFlags.STDERR_PIPE);
                var process = launcher.spawnv(new string[] {
                    "curl", "-fsSI",
                    "--max-time", timeout_seconds.to_string(),
                    "--", url
                });
                var headers = new DataInputStream(process.get_stdout_pipe());
                string? found = null;
                while(true) {
                    var header = headers.read_line();
                    if(header == null) {
                        break;
                    }
                    var separator = header.index_of(":");
                    if(separator < 1 || header.substring(0, separator).down() != "content-length") {
                        continue;
                    }
                    found = header.substring(separator + 1).strip();
                }
                process.get_stderr_pipe().read_bytes(1);
                if(!process.wait()) {
                    return 0;
                }
                return found != null ? uint64.parse(found) : 0;
            }
            catch(Error e) {
            }
            return 0;
        }

        private Uri get_for_path(string path) throws Error {
            var uri = Uri.parse(repository_config.url, UriFlags.NONE);
            // A scheme-plus-host URL ("http://h:1") has an empty path, and a
            // relative joined path would make Uri.build reject the URI
            var base_path = uri.get_path() ?? "/";
            if(base_path.length == 0) {
                base_path = "/";
            }
            return Uri.build(UriFlags.NONE, uri.get_scheme(), uri.get_userinfo(), uri.get_host(), uri.get_port(), Path.build_filename(base_path, path), uri.get_query(), uri.get_fragment());
        }
    }
}