path: root/source/cepis/client.d

/**
 * The HTTP client, built on `std.net.curl`.
 *
 * A fresh curl handle is opened per request and connections are not pooled. For
 * a one-off lookup that is irrelevant. For thousands, $(B do not write a loop) —
 * use $(D Client.lookupMany), which resolves up to 500 CEPs per request. A loop
 * of 500 calls pays 500 TLS handshakes; the batch pays one.
 *
 * Retry policy: 429 and the transient 5xx in $(D RETRIABLE_STATUS), plus network
 * failures, are retried with exponentially backed-off, jittered waits. 400 and
 * 404 are deliberately absent from that set — both are definitive answers about
 * the request, so retrying returns the same thing more slowly while burning
 * someone else's quota.
 */
module cepis.client;

import core.thread : Thread;
import core.time : Duration, MonoTime, dur;
import std.algorithm : all, canFind;
import std.array : appender, array, join;
import std.ascii : isDigit;
import std.conv : to;
import std.csv : Malformed, csvReader;
import std.encoding : sanitize;
import std.json : JSONType, JSONValue, parseJSON;
import std.net.curl : CurlException, HTTP;
import std.range : take;
import std.string : startsWith, strip, toLower;
import std.typecons : Nullable, nullable;
import std.uri : encodeComponent;
import std.utf : byDchar;

import cepis.exception;
import cepis.model;
import cepis.retry;

@safe:

/// The version of this library, as sent in the `User-Agent`.
enum string VERSION = "0.1.0";

/// Where the public API lives.
enum string DEFAULT_URL = "https://cep.is";

/// How many CEPs the batch endpoint accepts in one request.
enum size_t MAX_BATCH = 500;

/// The statuses worth trying again: rate limiting and the transient 5xx.
static immutable ushort[5] RETRIABLE_STATUS = [429, 500, 502, 503, 504];

private enum string ACCEPT = "application/json, text/csv;q=0.9, */*;q=0.1";

/**
 * Return the CEP's 8 digits, accepting the mask.
 *
 * `"01001-000"`, `"01001000"` and `" 01001-000 "` all become `"01001000"`.
 *
 * Throws: $(D InvalidCepException) unless exactly 8 digits remain. This runs
 * $(B before) any request: a 7-digit CEP is the caller's bug, and spending a
 * round trip for the server to say so wastes both sides' time.
 */
string normalizeCep(scope const(char)[] value)
{
    auto cleaned = appender!string();
    foreach (character; value.strip())
        if (character != '-' && character != '.' && character != ' ')
            cleaned.put(character);

    string result = cleaned.data;
    if (result.length != 8 || !result.all!isDigit)
        throw new InvalidCepException(`CEP must be 8 digits, got "` ~ value.idup ~ `"`);
    return result;
}

///
unittest
{
    assert(normalizeCep("01001-000") == "01001000");
    assert(normalizeCep(" 01001.000 ") == "01001000");
}

/**
 * Everything a $(D Client) can be told. Every field has a default; set only
 * what differs.
 */
struct ClientConfig
{
    /**
     * Where the API lives.
     *
     * Point it at a local instance (the API repo brings one up on
     * `http://localhost:8080` with `just up`) or at a test server. It is the
     * only setting that usually changes between environments. Trailing slashes
     * are stripped, so it cannot produce doubled separators that some proxies
     * normalise and others 404.
     */
    string baseUrl = DEFAULT_URL;

    /**
     * How long to wait $(B per socket operation) — connecting, and each stall
     * while reading.
     *
     * This is not a cap on the whole call. That cap is $(D totalTimeout).
     */
    Duration timeout = dur!"seconds"(10);

    /**
     * How many attempts in total, counting the first.
     *
     * `1` disables retrying. The default of `3` covers the common case (one
     * transient failure) without turning a real outage into a long wait.
     */
    uint maxAttempts = 3;

    /**
     * Base of the exponential wait.
     *
     * The actual wait is drawn from `[0, base * 2^^(n-1)]` — $(I full jitter).
     * The randomness is not decoration: without it, N clients that fail
     * together come back together and reproduce the spike that took the service
     * down.
     */
    Duration backoffBase = dur!"msecs"(500);

    /**
     * Cap on each individual wait.
     *
     * It applies to a server-sent `Retry-After` too, or a `Retry-After: 3600`
     * from a misconfigured proxy would park the thread for an hour.
     */
    Duration backoffMax = dur!"seconds"(8);

    /**
     * Cap on the whole call, attempts and waits included.
     *
     * Null leaves $(D maxAttempts) as the only bound. Set it in code serving a
     * web request: without it the worst case is `maxAttempts * timeout` plus
     * the sum of the backoffs.
     */
    Nullable!Duration totalTimeout;

    /**
     * Overrides the `User-Agent`.
     *
     * Left empty, the default identifies the library and its version, which is
     * what lets the server diagnose an old client misbehaving.
     */
    string userAgent;
}

/**
 * Client for the cep.is API.
 *
 * Build one and reuse it: the configuration is immutable after construction and
 * no method keeps state between calls, so it is safe to copy and to share
 * across threads.
 *
 * ---
 * ClientConfig config;
 * config.baseUrl = "http://localhost:8080";
 * config.totalTimeout = dur!"seconds"(15);
 *
 * auto client = Client(config);
 * auto address = client.lookup("01001000");
 * ---
 */
struct Client
{
    private ClientConfig config;

    /**
     * Build a client from a configuration.
     *
     * Params:
     *   config = the settings to use; $(D ClientConfig.init) is the default
     *            configuration, pointing at $(D DEFAULT_URL).
     *
     * Throws: $(D CepisException) for a configuration that can only produce a
     * client which never completes a call.
     */
    this(ClientConfig config)
    {
        if (config.maxAttempts < 1)
            throw new CepisException("maxAttempts must be >= 1");
        if (config.timeout <= Duration.zero)
            throw new CepisException("timeout must be > 0");

        this.config = config;
        while (this.config.baseUrl.length > 0 && this.config.baseUrl[$ - 1] == '/')
            this.config.baseUrl = this.config.baseUrl[0 .. $ - 1];
        if (this.config.userAgent.length == 0)
            this.config.userAgent = "cepis-d/" ~ VERSION ~ " (+https://cep.is)";
    }

    /// The API URL in use, without a trailing slash.
    string baseUrl() const pure nothrow @nogc
    {
        return config.baseUrl;
    }

    /// The `User-Agent` sent with every request.
    string userAgent() const pure nothrow @nogc
    {
        return config.userAgent;
    }

    /// The per-socket-operation timeout.
    Duration timeout() const pure nothrow @nogc
    {
        return config.timeout;
    }

    /// How many attempts each call makes at most, counting the first.
    uint maxAttempts() const pure nothrow @nogc
    {
        return config.maxAttempts;
    }

    /// The cap on the whole call, if one is set.
    Nullable!Duration totalTimeout() const pure nothrow
    {
        return config.totalTimeout;
    }

    /**
     * Look up a single CEP.
     *
     * The CEP may be masked or not. This is the only way to obtain a
     * coordinate — see $(D lookupMany) for why.
     *
     * A 400 from the API is surfaced as $(D InvalidCepException) carrying the
     * API's own message: the format already passed local validation, so the
     * server knows something this client does not.
     *
     * Throws:
     *   $(UL
     *     $(LI $(D InvalidCepException) — not 8 digits (no request is made).)
     *     $(LI $(D CepNotFoundException) — well-formed, but not in the
     *          database.)
     *     $(LI $(D RateLimitedException) — 429 after every attempt.)
     *     $(LI $(D ServerException) — 5xx after every attempt.)
     *     $(LI $(D NetworkException) — the request never completed.)
     *     $(LI $(D InvalidResponseException) — the body could not be read as
     *          JSON.)
     *   )
     */
    Address lookup(scope const(char)[] cep) const
    {
        string number = normalizeCep(cep);
        const response = request("/" ~ number, [ushort(200)]);

        if (response.status == 404)
            throw new CepNotFoundException(number);
        if (response.status == 400)
            throw new InvalidCepException(errorMessage(response.content,
                    "CEP " ~ number ~ " rejected"));

        return Address.fromJSON(jsonObject(response.content));
    }

    /**
     * Look up many CEPs in one batch.
     *
     * Returns an array of $(B the same length and in the same order) as the
     * input. A CEP that does not exist becomes null at its position — it does
     * not vanish from the array, and it does not throw. That is the right
     * difference from $(D lookup): in a batch of a thousand, one bad CEP must
     * not cost the other nine hundred and ninety-nine.
     *
     * A malformed CEP also becomes null, for the same reason. If you need to
     * know $(I which) one was malformed, validate first with $(D normalizeCep).
     *
     * Input is chunked into blocks of $(D MAX_BATCH) automatically. Do not loop
     * over $(D lookup) instead: 500 calls pay 500 round trips, the batch pays
     * one.
     *
     * $(B The batch carries no coordinate.) The endpoint behind it is CSV, and
     * the CSV has no `geo` column. Every $(D Address) returned here has a null
     * $(D geo), including for CEPs that do have one. Need `geo`? Use
     * $(D lookup), one at a time.
     *
     * Throws: transport failures only — $(D RateLimitedException),
     * $(D ServerException), $(D NetworkException). A bad CEP never fails a
     * batch.
     */
    Nullable!Address[] lookupMany(scope const(string)[] ceps) const
    {
        if (ceps.length == 0)
            return null;

        Nullable!Address[] result;
        result.reserve(ceps.length);
        for (size_t start = 0; start < ceps.length; start += MAX_BATCH)
        {
            const end = start + MAX_BATCH < ceps.length ? start + MAX_BATCH : ceps.length;
            result ~= lookupChunk(ceps[start .. end]);
        }
        return result;
    }

    /**
     * Query `GET /health`: does the API answer, and with which database.
     *
     * An API reporting itself degraded is not an error — that arrives in
     * $(D Health.status), and checking health is exactly when you want the
     * diagnosis rather than an exception. Network failures still throw.
     */
    Health health() const
    {
        const response = request("/health", [ushort(200), ushort(503)]);
        return Health.fromJSON(jsonObject(response.content));
    }

    /**
     * One chunk of at most $(D MAX_BATCH) CEPs, in a single request.
     *
     * A blank CSV row means "not found": the API fills only the `cep` column
     * and leaves the rest empty, precisely so it does not misalign whoever is
     * consuming the batch in a spreadsheet.
     *
     * The API guarantees one row per requested CEP. Should that ever fail to
     * hold — a proxy truncating, a cut response — the result is padded to the
     * input length rather than returned short, because a shifted array silently
     * pairs the wrong address with the wrong CEP.
     */
    private Nullable!Address[] lookupChunk(scope const(string)[] ceps) const
    {
        const response = request("/csv?ceps=" ~ encodeComponent(ceps.join(",")), [
            ushort(200)
        ]);

        string text = sanitize(response.content);
        if (text.startsWith(""))
            text = text[3 .. $];

        Nullable!Address[] rows;
        try
        {
            foreach (record; csvReader!(string[string], Malformed.ignore)(text, null))
            {
                auto localidade = "localidade" in record;
                if (localidade is null || (*localidade).strip().length == 0)
                {
                    rows ~= Nullable!Address.init;
                    continue;
                }

                JSONValue object = JSONValue.emptyObject;
                foreach (column, value; record)
                    object[column] = JSONValue(value);
                rows ~= nullable(Address.fromJSON(object));
            }
        }
        catch (Exception)
        {
        }

        if (rows.length > ceps.length)
            rows = rows[0 .. ceps.length];
        while (rows.length < ceps.length)
            rows ~= Nullable!Address.init;
        return rows;
    }

    /**
     * Perform the GET, with retries.
     *
     * $(I Definitive) error statuses (400, 404) come back normally, for the
     * caller to turn into the right exception — only transient failures become
     * exceptions here, and only once the attempts are exhausted.
     */
    private Response request(string path, scope const(ushort)[] accepted) const
    {
        string url = config.baseUrl ~ path;
        const bounded = !config.totalTimeout.isNull;
        MonoTime deadline;
        if (bounded)
            deadline = MonoTime.currTime + config.totalTimeout.get;

        ushort lastStatus;
        Nullable!Duration lastRetryAfter;
        string lastFailure;

        foreach (attempt; 1 .. config.maxAttempts + 1)
        {
            Duration cap;
            if (bounded)
            {
                cap = deadline - MonoTime.currTime;
                if (cap <= Duration.zero)
                    break;
            }

            try
            {
                auto response = httpGet(url, config.timeout, cap, config.userAgent);
                if (accepted.canFind(response.status)
                        || !RETRIABLE_STATUS[].canFind(response.status))
                    return response;
                lastStatus = response.status;
                lastRetryAfter = parseRetryAfter(response.retryAfter);
                lastFailure = null;
            }
            catch (CurlException failure)
            {
                lastStatus = 0;
                lastRetryAfter.nullify();
                lastFailure = failure.msg;
            }

            if (attempt == config.maxAttempts)
                break;

            auto wait = backoff(attempt, lastRetryAfter);
            if (bounded)
            {
                const left = deadline - MonoTime.currTime;
                if (left <= Duration.zero)
                    break;
                if (wait > left)
                    wait = left;
            }
            sleepFor(wait);
        }

        const suffix = " (after " ~ config.maxAttempts.to!string ~ " attempt(s))";
        if (lastStatus == 429)
            throw new RateLimitedException("rate limit reached at " ~ url ~ suffix, lastRetryAfter);
        if (lastStatus != 0)
            throw new ServerException(
                    "the API answered " ~ lastStatus.to!string ~ " at " ~ url ~ suffix, lastStatus);
        throw new NetworkException("could not reach " ~ url ~ suffix ~ ": " ~ (lastFailure is null
                ? "no attempt was made within the total timeout" : lastFailure));
    }

    /**
     * How long to wait before the next attempt.
     *
     * The server's `Retry-After` wins over the local calculation when present:
     * it knows when it will be back, we are guessing. It still goes through the
     * $(D backoffMax) cap. Otherwise the wait is drawn from `[0, ceiling]`
     * rather than being the ceiling.
     */
    private Duration backoff(uint attempt, Nullable!Duration retryAfter) const
    {
        if (!retryAfter.isNull)
        {
            const asked = retryAfter.get;
            return asked < config.backoffMax ? asked : config.backoffMax;
        }

        Duration ceiling = config.backoffBase;
        foreach (_; 1 .. attempt)
        {
            if (ceiling >= config.backoffMax)
                break;
            ceiling *= 2;
        }
        if (ceiling > config.backoffMax)
            ceiling = config.backoffMax;
        return jitter(ceiling);
    }
}

private struct Response
{
    ushort status;
    string content;
    string retryAfter;
}

/**
 * One GET through libcurl.
 *
 * Trusted because `std.net.curl` predates the safety annotations: the callbacks
 * below only append to locals owned by this frame, and nothing escapes it.
 */
private Response httpGet(string url, Duration timeout, Duration cap, string userAgent) @trusted
{
    auto http = HTTP(url);
    http.method = HTTP.Method.get;
    http.connectTimeout = timeout;
    http.dataTimeout = timeout;
    if (cap > Duration.zero)
        http.operationTimeout = cap;
    http.addRequestHeader("User-Agent", userAgent);
    http.addRequestHeader("Accept", ACCEPT);
    http.addRequestHeader("Accept-Encoding", "identity");

    auto content = appender!(ubyte[])();
    string retryAfter;
    http.onReceive = (ubyte[] data) { content.put(data); return data.length; };
    http.onReceiveHeader = (in char[] key, in char[] value) {
        if (key.toLower() == "retry-after")
            retryAfter = value.idup;
    };

    http.perform();
    return Response(http.statusLine.code, cast(string) content.data.idup, retryAfter);
}

private void sleepFor(Duration wait) @trusted
{
    if (wait > Duration.zero)
        Thread.sleep(wait);
}

/// Decode a JSON object body, or throw $(D InvalidResponseException).
private JSONValue jsonObject(string content)
{
    JSONValue value;
    try
        value = parseJSON(content);
    catch (Exception)
        throw new InvalidResponseException("response is not valid JSON — something between you "
                ~ "and the API is answering instead of it: " ~ preview(content));

    if (value.type != JSONType.object)
        throw new InvalidResponseException("expected a JSON object, got " ~ preview(content));
    return value;
}

/// Pull `mensagem` out of an error body, falling back to a default.
private string errorMessage(string content, string fallback)
{
    try
    {
        const value = parseJSON(content);
        if (value.type == JSONType.object)
        {
            const object = value.objectNoRef;
            if (auto found = "mensagem" in object)
                if (found.type == JSONType.string && found.str.length > 0)
                    return found.str;
        }
    }
    catch (Exception)
    {
    }
    return fallback;
}

private string preview(string content)
{
    return `"` ~ sanitize(content).byDchar.take(120).array.to!string ~ `"`;
}

@("a trailing slash is not duplicated")
unittest
{
    ClientConfig config;
    config.baseUrl = "http://x//";

    assert(Client(config).baseUrl == "http://x");
}

@("the user agent identifies the library")
unittest
{
    assert(Client(ClientConfig.init).userAgent.startsWith("cepis-d/"));
    assert(Client(ClientConfig.init).baseUrl == DEFAULT_URL);
}

@("a bad configuration is refused")
unittest
{
    import std.exception : assertThrown;

    ClientConfig noAttempts;
    noAttempts.maxAttempts = 0;
    assertThrown!CepisException(Client(noAttempts));

    ClientConfig noTimeout;
    noTimeout.timeout = Duration.zero;
    assertThrown!CepisException(Client(noTimeout));
}

@("a masked cep normalises and a short one does not")
unittest
{
    import std.exception : assertThrown;

    foreach (value; ["01001000", "01001-000", " 01001-000 ", "01001.000"])
        assert(normalizeCep(value) == "01001000");

    foreach (value; ["", "123", "0100100a", "010010000", "abcdefgh"])
        assertThrown!InvalidCepException(normalizeCep(value));
}