Actors that share nothing. I/O that never blocks.

#include <qb/main.h>
#include <qbm/http/http.h>

class ApiServer : public qb::Actor, public qb::http::Server<> {
public:
    qb::io::async::task<bool> onInit() override {
        registerEvent<qb::KillEvent>(*this);

        router().get("/", [](auto ctx) {
            ctx->response().body() = "Hello from qbm-http";
            ctx->complete();
        });

        router().get("/users/:id", [](auto ctx) {
            ctx->response().status() = qb::http::status::OK;
            ctx->response().body()   = "user " + ctx->path_param("id");
            ctx->complete();
        });

        router().compile();  // required, after every route

        if (!listen({"tcp://0.0.0.0:8080"}))  // qb::io::uri; returns "is listening"
            co_return false;  // port taken: init fails, actor never starts
        start();  // arm the accept watcher
        co_return true;
    }

    void on(qb::KillEvent const &) { kill(); }
};

int main() {
    qb::Main engine;
    engine.addActor<ApiServer>(0);
    engine.start();  // one worker thread per used core, then returns
    engine.join();  // blocks until every actor terminates
    return 0;
}

#include <qbm/http/http.h>

// Inside a coroutine the call reads top to bottom: request out, response back,
// error handling is a plain if. No callback, no extra thread, no second loop.
qb::io::async::task<void> fetch() {
    qb::http::Request req(qb::io::uri("http://api.example.com/data"));
    req.add_header("Accept", "application/json");

    using namespace std::chrono_literals;
    auto reply = co_await qb::http::GET(std::move(req), 5s);  // timeout is a qb::duration

    if (reply.response.status() == qb::http::status::OK)
        handle(reply.response.body().as<std::string_view>());
}

// The same call from ordinary code: run_sync drives the loop until it resolves.
int main() {
    qb::io::async::init();

    auto reply = qb::http::run_sync(
        qb::http::GET(qb::http::Request{qb::io::uri("http://api.example.com/data")}));

    std::cout << reply.response.body().as<std::string_view>() << '\n';
}

#include <qbm/http/http.h>  // pulls <qbm/http/2/http2.h> under QB_HAS_SSL

// Server. Routing is identical to HTTP/1.1 — the HTTP/2 session feeds the same
// Router, so groups, controllers, middleware and validation work unchanged.
auto server = qb::http2::make_server();

server->router().get("/hello", [](auto ctx) {
    ctx->response().body() = "Hello over HTTP/2";
    ctx->complete();
});
server->router().compile();

// listen() builds the TLS context, sets ALPN to {"h2","http/1.1"}, and binds.
server->listen({"https://0.0.0.0:8443"}, "cert.pem", "key.pem");
server->start();

// Client. Requests multiplex over one connection, and it connects lazily —
// push_request establishes the connection on first use.
auto response = qb::http::run_sync([]() -> qb::io::async::task<qb::http::Response> {
    auto client = qb::http2::make_client("https://api.example.com");
    co_return co_await client->push_request(
        qb::http::Request{qb::io::uri("https://api.example.com/data")});
}());

#include <qbm/http/http.h>

#ifdef QBM_HTTP_HAS_HTTP3  // set only when SSL, QUIC and libnghttp3 are all present

// Server. The transport is UDP/QUIC and ALPN is locked to {"h3"}; the router,
// the context and the handlers are the ones you already wrote.
auto server = qb::http3::make_server();

server->router().get("/ping", [](auto ctx) {
    ctx->response().set_header("x-protocol", "HTTP/3");
    ctx->response().body() = "pong-h3";
    ctx->complete();
});
server->router().compile();
server->listen(qb::io::uri("https://0.0.0.0:4433"), "cert.pem", "key.pem");

// Client. Same shape as HTTP/2, plus two things only this one has: a request id
// you can cancel, and a cap on the reassembled body (64 MiB by default).
auto client = qb::http3::make_client("https://127.0.0.1:4433");
auto id = client->push_request_with_id(qb::http::Request{qb::io::uri("/slow")},
                                       [](qb::http::Response res) { handle(res); });
client->cancel_request(id);  // completes with CLIENT_CLOSED_REQUEST

#endif

// HTTP/2 is TCP/TLS and HTTP/3 is UDP/QUIC, so they cannot share a socket.
// The facade registers every route on both routers and compiles both.
auto server = qb::http::make_dual_stack_server();

server->router().get("/shared", [](auto ctx) {
    // Same handler for both protocols; ask which one served this request.
    const bool over_h3 = ctx->request().major_version == 3;
    ctx->response().set_header("x-version", over_h3 ? "h3" : "h2");
    ctx->response().body() = over_h3 ? "shared-over-h3" : "shared-over-h2";
    ctx->complete();
});
server->router().compile();  // compiles BOTH routers

// Two URIs — TCP/TLS first, then UDP/QUIC — and one certificate pair. The bind
// is all or nothing: if one stack comes up and the other fails, the one that
// came up is rolled back, so you never listen on half a server.
server->listen(qb::io::uri("https://0.0.0.0:443"),
               qb::io::uri("https://0.0.0.0:443"),
               "cert.pem", "key.pem");

#include <qbm/pgsql/pgsql.h>

using namespace qb::pg;

qb::io::async::task<void> read(tcp::database &db, int user_id) {
    // One-shot parameterized query: no statement name, no OID list. The
    // parameter types are deduced from the C++ arguments, and it runs through
    // the unnamed prepared statement, so it does not pollute the cache.
    auto r = co_await db.query("SELECT id, name FROM users WHERE id = $1", user_id);
    if (!r) co_return;  // Reply<T> carries either the result or a typed db_error

    // A row decodes straight into a tuple, so a structured binding just works.
    for (auto [id, name] : r.result().all<int, std::string>())
        qb::io::cout() << id << ' ' << name << '\n';

    // rows<>() is the same thing lazily — no intermediate vector.
    for (auto [id, name] : r.result().rows<int, std::string>())
        use(id, name);

    // First row, or nothing. std::optional<T> in the tuple reads a nullable column.
    if (auto one = r.result().one<int, std::optional<std::string>>()) {
        auto [id, maybe_name] = *one;
    }
}

// Prepare once, execute many. The client keeps a prepared-statement LRU, so the
// name resolves without a round trip after the first use.
auto pr = co_await db.prepare("by_email",
                              "SELECT id, name FROM users WHERE email = $1",
                              qb::pg::type_oid_sequence{qb::pg::oid::text});
if (!pr.ok()) co_return;

auto r = co_await db.execute("by_email", qb::pg::params{email});

// The callback form is the same call with handlers, for code that cannot be a
// coroutine. Both forms drain on the same event loop.
db.execute("by_email", qb::pg::params{email},
           [](qb::pg::transaction &, qb::pg::results rs) { use(rs); },
           [](qb::pg::error::db_error const &e) { report(e); });

using namespace qb::pg;

qb::io::async::task<void> transfer(tcp::database &db) {
    // BEGIN, run the body, COMMIT — or ROLLBACK when the body throws.
    auto tx = co_await with_transaction(db, [](transaction &tr)
                                                -> qb::io::async::task<void> {
        auto debit = co_await tr.execute(
            "UPDATE account SET balance = balance - $1 WHERE id = $2", 100, 1);
        if (!debit)
            throw transaction_abort{debit.error()};  // rolls back, keeps the error

        auto credit = co_await tr.execute(
            "UPDATE account SET balance = balance + $1 WHERE id = $2", 100, 2);
        if (!credit)
            throw transaction_abort{credit.error()};
    });

    if (!tx.ok())
        qb::io::cerr() << "rolled back: " << tx.error().what() << '\n';

    // Savepoints nest inside a block, and a statement timeout is a qb::duration
    // armed on the next BEGIN.
    db.set_timeout(std::chrono::seconds{5});
}

// query() buffers the whole result set. query_stream runs the statement through a
// server-side CURSOR and holds exactly one batch, whatever the result size.
std::uint64_t total = 0;
co_await db.query_stream("SELECT amount FROM ledger", 1000,
                         [&](auto row) { total += row[0].template as<long>(); });

// COPY is PostgreSQL's bulk path, and copy_out streams it out in constant memory:
// each CopyData chunk reaches the sink as it arrives.
std::ofstream f("users.csv");
co_await db.copy_out("COPY users TO STDOUT (FORMAT csv)",
                     [&](std::string_view chunk) { f.write(chunk.data(), chunk.size()); });

// Bulk in: return std::nullopt from the source to finish.
co_await db.copy_in("COPY t (id, v) FROM STDIN",
                    [&]() -> std::optional<std::string> { return next_line(); });

// Publisher: NOTIFY is a statement like any other.
co_await db.notify("jobs", R"({"id":42})");

// Subscriber: register the handler, then LISTEN. The handler runs on the I/O
// thread when a NotificationResponse arrives — there is nothing to poll.
db.on_incoming_notify([](qb::pg::notification &&n) {
    qb::io::cout() << n.channel << ": " << n.payload
                   << " (backend " << n.server_backend_pid << ")\n";
});

(void)co_await db.listen("jobs");
(void)co_await db.unlisten("jobs");

#include <qb/io/async.h>
#include <qbm/redis/redis.h>

int main() {
    qb::io::async::init();

    qb::redis::tcp::client redis{qb::io::uri{"tcp://localhost:6379"}};
    if (!qb::io::async::run_sync(redis.connect()))  // connect_awaiter -> bool
        return 1;

    // SET returns Reply<status>; operator bool is true on success.
    auto set_r = qb::io::async::run_sync(redis.set("greeting", "Hello Redis!"));
    if (!set_r)
        return 1;

    // GET returns Reply<std::optional<std::string>>; value_or unwraps the optional.
    auto value = qb::io::async::run_sync(redis.get("greeting")).value_or(std::string{});
    qb::io::cout() << "Retrieved: " << value << '\n';

    qb::io::async::run_sync(redis.del("greeting"));  // Reply<long long>
    return 0;
}

// Hash
co_await redis.hset("user:1", "email", "ada@example.com");
auto email = co_await redis.hget("user:1", "email");  // Reply<optional<string>>

// Sorted set — a leaderboard, then a blocking pop with a real timeout
std::vector<qb::redis::score_member> members = {
    {10.0, "member1"}, {20.0, "member2"}, {30.0, "member3"}};
co_await redis.zadd("board", members);

auto top = co_await redis.bzpopmin({"board"}, std::chrono::seconds{5});
if (top.ok() && top.result().has_value()) {
    auto const &[key, member, score] = *top.result();
}

// Stream — XADD gives back a typed id, XRANGE gives back decoded entries
std::vector<std::pair<std::string, std::string>> entries = {
    {"field1", "value1"}, {"field2", "value2"}};
auto id = co_await redis.xadd("mystream", entries);  // Reply<stream_id>

auto range = co_await redis.xrange("mystream", "-", "+");
for (auto const &entry : range.result())
    qb::io::cout() << entry.id.to_string() << " has "
                   << entry.fields.size() << " fields\n";

// Not wrapped yet? Name the command and pick its decode type.
auto n = co_await redis.command<long long>("EVAL", "return 1", "0");

// WATCH marks the key. If anyone else writes it before EXEC, the whole batch is
// aborted instead of applying a decision taken on stale data.
auto w = co_await redis.watch("balance");
if (!w.ok()) co_return;

auto current = (co_await redis.get("balance")).value_or("0");
auto next = recompute(current);

co_await redis.multi();               // is_in_multi() is now true
(void)co_await redis.set("balance", next);  // queued server-side, acked QUEUED

auto ex = co_await redis.exec<std::string>();  // Reply<std::vector<std::string>>
if (!ex.ok()) {
    // Someone changed "balance" after WATCH. Nothing was applied — retry the block.
}

#include <qbm/redis/redis.h>
#include <qb/io/async.h>

qb::io::async::task<void> chat() {
    // Subscriber: a callback consumer routes messages out-of-band.
    qb::redis::tcp::cb_consumer consumer{
        qb::io::uri{"tcp://localhost:6379"},
        [](qb::redis::message &&msg) {
            qb::io::cout() << '[' << msg.channel << "] " << msg.payload << '\n';
        }};

    co_await consumer.connect();
    co_await consumer.hello(3);  // RESP3, for native push frames
    co_await consumer.subscribe("alerts");
    co_await consumer.psubscribe("user:*:updates");

    // Publisher: a separate client — publish() is not on the consumer.
    qb::redis::tcp::client publisher{qb::io::uri{"tcp://localhost:6379"}};
    co_await publisher.connect();
    co_await publisher.publish("alerts", "Server restarted");
}