//! Comptime-derived flag and environment plumbing.
//!
//! Every option is a field on an options struct. Its flag spelling and its
//! `JETSTREAM_*` name are *derived* from the field name rather than listed
//! separately, so the three things that used to drift independently — the
//! parser, the environment map, and the usage text — cannot disagree: there is
//! only one list, and it is the struct.
//!
//! That drift was a real, documented gap. `docs/configuration-parity.md` marks
//! the flag inventory **unverified** because nothing proved the hand-written
//! `environment.zig` map covered every flag the parser accepts. Derivation
//! turns that from a promise into a property.
//!
//! Naming rule: `backfill_workers` -> `--backfill-workers` and
//! `JETSTREAM_BACKFILL_WORKERS`. Options whose spelling predates the rule, or
//! which upstream declares root-only, carry an entry in the struct's optional
//! `overrides` declaration; there is no other way to deviate, so deviations
//! stay countable.

const std = @import("std");

pub const prefix = "JETSTREAM_";

/// How one option is exposed. `serve` mirrors upstream's command tree: a
/// root-only value has no synthesized `serve` flag.
pub const Exposure = struct {
    flag: []const u8,
    env: []const u8,
    serve: bool = true,
};

/// `some_field` -> `some-field`
pub fn flagName(comptime field: []const u8) []const u8 {
    return comptime blk: {
        var out: [field.len]u8 = undefined;
        for (field, 0..) |c, i| out[i] = if (c == '_') '-' else c;
        const frozen = out;
        break :blk &frozen;
    };
}

/// `some_field` -> `JETSTREAM_SOME_FIELD`
pub fn envName(comptime field: []const u8) []const u8 {
    return comptime blk: {
        var out: [prefix.len + field.len]u8 = undefined;
        @memcpy(out[0..prefix.len], prefix);
        for (field, 0..) |c, i| out[prefix.len + i] = std.ascii.toUpper(c);
        const frozen = out;
        break :blk &frozen;
    };
}

fn overrideFor(comptime T: type, comptime field: []const u8) ?Exposure {
    @setEvalBranchQuota(10_000);
    if (!@hasDecl(T, "overrides")) return null;
    const ov = T.overrides;
    if (!@hasField(@TypeOf(ov), field)) return null;
    const entry = @field(ov, field);
    const E = @TypeOf(entry);
    return Exposure{
        .flag = if (@hasField(E, "flag")) entry.flag else flagName(field),
        .env = if (@hasField(E, "env")) entry.env else envName(field),
        .serve = if (@hasField(E, "serve")) entry.serve else true,
    };
}

/// The exposure of every field on `T`, derived unless overridden.
pub fn exposures(comptime T: type) []const Exposure {
    return comptime blk: {
        @setEvalBranchQuota(20_000);
        const fields = std.meta.fields(T);
        var out: [fields.len]Exposure = undefined;
        for (fields, 0..) |f, i| {
            out[i] = overrideFor(T, f.name) orelse .{
                .flag = flagName(f.name),
                .env = envName(f.name),
            };
        }
        const frozen = out;
        break :blk &frozen;
    };
}

/// Is `key` an environment name this option set declares?
pub fn declaresEnv(comptime T: type, key: []const u8) bool {
    inline for (comptime exposures(T)) |e| {
        if (std.mem.eql(u8, e.env, key)) return true;
    }
    return false;
}

/// Is `flag` (without leading dashes) one this option set declares?
pub fn declaresFlag(comptime T: type, flag: []const u8) bool {
    inline for (comptime exposures(T)) |e| {
        if (std.mem.eql(u8, e.flag, flag)) return true;
    }
    return false;
}

// === tests ===

const testing = std.testing;

test "field names derive their flag and environment spellings" {
    try testing.expectEqualStrings("backfill-workers", flagName("backfill_workers"));
    try testing.expectEqualStrings("JETSTREAM_BACKFILL_WORKERS", envName("backfill_workers"));
    try testing.expectEqualStrings("addr", flagName("addr"));
    try testing.expectEqualStrings("JETSTREAM_ADDR", envName("addr"));
}

test "overrides are the only way to deviate, and they compose with derivation" {
    const Opts = struct {
        addr: []const u8 = "",
        log_level: []const u8 = "",
        legacy: []const u8 = "",

        pub const overrides = .{
            // upstream declares this root-only: no synthesized serve flag
            .log_level = .{ .serve = false },
            // a spelling that predates the naming rule
            .legacy = .{ .flag = "port", .env = "JETSTREAM_PORT" },
        };
    };
    const e = comptime exposures(Opts);
    try testing.expectEqualStrings("addr", e[0].flag);
    try testing.expect(e[0].serve);
    try testing.expectEqualStrings("log-level", e[1].flag);
    try testing.expect(!e[1].serve);
    try testing.expectEqualStrings("port", e[2].flag);
    try testing.expectEqualStrings("JETSTREAM_PORT", e[2].env);
    try testing.expect(declaresEnv(Opts, "JETSTREAM_ADDR"));
    try testing.expect(!declaresEnv(Opts, "JETSTREAM_NOPE"));
    try testing.expect(declaresFlag(Opts, "port"));
}

test "the derived option set reproduces the hand-written environment map" {
    // This is the point of the module. environment.zig's map is maintained by
    // hand, and docs/configuration-parity.md marks the flag inventory
    // unverified for exactly that reason. Deriving the same set from a struct
    // and asserting equality here makes the two impossible to drift apart --
    // and when main.zig moves onto this struct, the map stops existing.
    const derived = comptime exposures(ServeOptions);

    try testing.expectEqual(environment.mappings.len, derived.len);
    for (environment.mappings) |want| {
        var found = false;
        for (derived) |got| {
            if (!std.mem.eql(u8, want.flag, got.flag)) continue;
            found = true;
            try testing.expectEqualStrings(want.env, got.env);
            try testing.expectEqual(want.serve, got.serve);
            break;
        }
        if (!found) {
            std.debug.print("no derived option for flag --{s}\n", .{want.flag});
            return error.MissingDerivedOption;
        }
    }
}

/// The pinned Jetstream V2 flag surface, as a struct. Field order matches
/// upstream's declaration order so the two read side by side.
pub const ServeOptions = struct {
    log_level: []const u8 = "",
    log_format: []const u8 = "",
    addr: []const u8 = "",
    debug_addr: []const u8 = "",
    shutdown_timeout: []const u8 = "",
    client_drain_timeout: []const u8 = "",
    relay_url: []const u8 = "",
    plc_url: []const u8 = "",
    data_dir: []const u8 = "",
    max_backfill_repos: []const u8 = "",
    backfill_workers: []const u8 = "",
    backfill_batch_size: []const u8 = "",
    backfill_async_flush_workers: []const u8 = "",
    backfill_repos: []const u8 = "",
    skip_merge_discovery: []const u8 = "",
    failed_repo_retry_interval: []const u8 = "",
    failed_repo_retry_workers: []const u8 = "",
    failed_repo_retry_host_workers: []const u8 = "",
    failed_repo_retry_max_delay: []const u8 = "",
    disable_repo_action_rate_limits: []const u8 = "",
    cursor_lookback: []const u8 = "",
    segment_cache_max_age: []const u8 = "",
    plan_max_dids: []const u8 = "",
    plan_max_collections: []const u8 = "",
    plan_max_entries: []const u8 = "",
    plan_whole_segment_threshold: []const u8 = "",
    subscribe_read_log_retention_bytes: []const u8 = "",
    subscribe_block_cache_bytes: []const u8 = "",
    subscribe_read_batch: []const u8 = "",
    subscribe_slow_window: []const u8 = "",
    subscribe_slow_min_rate: []const u8 = "",
    cursor_block_index_cache_size: []const u8 = "",
    compaction_interval: []const u8 = "",
    rebloom_sweep: []const u8 = "",
    compaction_tombstone_cap: []const u8 = "",
    compaction_rewrite_workers: []const u8 = "",
    timestamp_import_token: []const u8 = "",
    timestamp_import_dir: []const u8 = "",

    pub const overrides = .{
        // Upstream declares these on the root command, so there is no
        // synthesized `serve` flag for them.
        .log_level = .{ .serve = false },
        .log_format = .{ .serve = false },
    };
};

// === value parsers ===
//
// Go-duration, boolean and size syntax, lifted out of main.zig. These are the
// shapes upstream's flag library accepts, and they are pure functions of their
// input, so they belong beside the flag derivation rather than inline in a
// 1,400-line entry point.

const Io = std.Io;
const server_mod = @import("../serve/server.zig");

pub fn parseDurationNanoseconds(raw: []const u8) !u64 {
    return parseDurationMagnitude(raw, std.math.maxInt(i64));
}
pub fn parseDurationMagnitude(raw: []const u8, limit: u128) !u64 {
    if (raw.len == 0) return error.InvalidDuration;
    var i: usize = 0;
    if (raw[i] == '+') {
        i += 1;
        if (i == raw.len) return error.InvalidDuration;
    } else if (raw[i] == '-') return error.InvalidDuration;
    if (std.mem.eql(u8, raw[i..], "0")) return 0;

    const Unit = struct { name: []const u8, nanoseconds: u64 };
    const units = [_]Unit{
        .{ .name = "ns", .nanoseconds = 1 },
        .{ .name = "us", .nanoseconds = std.time.ns_per_us },
        .{ .name = "µs", .nanoseconds = std.time.ns_per_us },
        .{ .name = "μs", .nanoseconds = std.time.ns_per_us },
        .{ .name = "ms", .nanoseconds = std.time.ns_per_ms },
        .{ .name = "s", .nanoseconds = std.time.ns_per_s },
        .{ .name = "m", .nanoseconds = std.time.ns_per_min },
        .{ .name = "h", .nanoseconds = std.time.ns_per_hour },
    };
    var total: u128 = 0;
    var tokens: usize = 0;
    while (i < raw.len) : (tokens += 1) {
        var whole: u128 = 0;
        var digits: usize = 0;
        while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) {
            const digit = raw[i] - '0';
            if (whole > (limit - digit) / 10) return error.DurationOverflow;
            whole = whole * 10 + digit;
            digits += 1;
        }

        var fraction: u128 = 0;
        var fraction_scale: u128 = 1;
        if (i < raw.len and raw[i] == '.') {
            i += 1;
            while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) {
                digits += 1;
                // Nanosecond resolution never needs more than 18 decimal
                // places for the largest supported unit; Go truncates below
                // one nanosecond as well.
                if (fraction_scale < 1_000_000_000_000_000_000) {
                    fraction = fraction * 10 + (raw[i] - '0');
                    fraction_scale *= 10;
                }
            }
        }
        if (digits == 0) return error.InvalidDuration;

        var unit: ?Unit = null;
        for (units) |candidate| {
            if (std.mem.startsWith(u8, raw[i..], candidate.name)) {
                unit = candidate;
                i += candidate.name.len;
                break;
            }
        }
        const selected = unit orelse return error.InvalidDuration;
        if (whole > limit / selected.nanoseconds) return error.DurationOverflow;
        var component = whole * selected.nanoseconds;
        component += (fraction * selected.nanoseconds) / fraction_scale;
        if (component > limit - total) return error.DurationOverflow;
        total += component;
    }
    if (tokens == 0) return error.InvalidDuration;
    return @intCast(total);
}
pub fn parseDurationCeilSeconds(raw: []const u8) !u64 {
    const nanoseconds = try parseDurationNanoseconds(raw);
    return (nanoseconds + std.time.ns_per_s - 1) / std.time.ns_per_s;
}
pub fn parseGoDuration(raw: []const u8) !Io.Duration {
    if (raw.len > 0 and raw[0] == '-') {
        if (raw.len == 1 or raw[1] == '+' or raw[1] == '-') return error.InvalidDuration;
        const negative_limit: u128 = @as(u128, std.math.maxInt(i64)) + 1;
        const magnitude = try parseDurationMagnitude(raw[1..], negative_limit);
        return .fromNanoseconds(-@as(i96, magnitude));
    }
    return .fromNanoseconds(@intCast(try parseDurationNanoseconds(raw)));
}
pub fn effectiveHttpShutdownDuration(configured: Io.Duration) Io.Duration {
    return if (configured.nanoseconds <= 0) .fromSeconds(30) else configured;
}
pub fn effectiveSubscribeSlowWindowNanoseconds(raw: []const u8) !u64 {
    const default = server_mod.default_slow_window_ns;
    if (raw.len > 0 and raw[0] == '-') {
        // Go's DurationFlag accepts a negative duration, then subscribe's
        // defaults replace every non-positive value. Parse the magnitude so
        // malformed negative values still fail instead of becoming defaults.
        _ = try parseDurationNanoseconds(raw[1..]);
        return default;
    }
    const configured = try parseDurationNanoseconds(raw);
    return if (configured == 0) default else configured;
}
test "subscribe slow window uses the upstream non-positive sentinel" {
    const default = server_mod.default_slow_window_ns;
    try std.testing.expectEqual(default, try effectiveSubscribeSlowWindowNanoseconds("0"));
    try std.testing.expectEqual(default, try effectiveSubscribeSlowWindowNanoseconds("-250ms"));
    try std.testing.expectEqual(@as(u64, 250 * std.time.ns_per_ms), try effectiveSubscribeSlowWindowNanoseconds("250ms"));
    try std.testing.expectError(error.InvalidDuration, effectiveSubscribeSlowWindowNanoseconds("-garbage"));
}
test "shutdown durations preserve Go's signed CLI and HTTP default semantics" {
    try std.testing.expectEqual(@as(i96, 5 * std.time.ns_per_s), (try parseGoDuration("5s")).nanoseconds);
    try std.testing.expectEqual(@as(i96, -250 * std.time.ns_per_ms), (try parseGoDuration("-250ms")).nanoseconds);
    try std.testing.expectEqual(@as(i96, 0), (try parseGoDuration("-0")).nanoseconds);
    try std.testing.expectEqual(@as(i96, std.math.maxInt(i64)), (try parseGoDuration("9223372036854775807ns")).nanoseconds);
    try std.testing.expectEqual(@as(i96, std.math.minInt(i64)), (try parseGoDuration("-9223372036854775808ns")).nanoseconds);
    try std.testing.expectEqual(@as(i96, 30 * std.time.ns_per_s), effectiveHttpShutdownDuration(.zero).nanoseconds);
    try std.testing.expectEqual(@as(i96, 30 * std.time.ns_per_s), effectiveHttpShutdownDuration(.fromSeconds(-1)).nanoseconds);
    try std.testing.expectEqual(@as(i96, 3 * std.time.ns_per_s), effectiveHttpShutdownDuration(.fromSeconds(3)).nanoseconds);
    try std.testing.expectError(error.InvalidDuration, parseGoDuration("-garbage"));
    try std.testing.expectError(error.InvalidDuration, parseGoDuration("-+1s"));
    try std.testing.expectError(error.DurationOverflow, parseGoDuration("9223372036854775808ns"));
    try std.testing.expectError(error.DurationOverflow, parseGoDuration("-9223372036854775809ns"));
}
test "segment cache duration matches Go duration and ceil policy" {
    try testing.expectEqual(@as(u64, 0), try parseDurationCeilSeconds("0"));
    try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("500ms"));
    try testing.expectEqual(@as(u64, 2), try parseDurationCeilSeconds("1500ms"));
    try testing.expectEqual(@as(u64, 3600), try parseDurationCeilSeconds("1h"));
    try testing.expectEqual(@as(u64, 5400), try parseDurationCeilSeconds("1h30m"));
    try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds(".1s"));
    try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1us"));
    try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1µs"));
    try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("-1s"));
    try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("1"));
    try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("wat"));
    try testing.expectEqual(@as(u64, 1_500_000_000), try parseDurationNanoseconds("1.5s"));
}
pub fn parseNonNegativeUsize(raw: []const u8) !usize {
    if (raw.len > 0 and raw[0] == '-') return error.BadArgs;
    return std.fmt.parseInt(usize, raw, 10);
}
pub fn parseBool(raw: []const u8) !bool {
    if (std.mem.eql(u8, raw, "1") or std.mem.eql(u8, raw, "t") or std.mem.eql(u8, raw, "T") or
        std.mem.eql(u8, raw, "true") or std.mem.eql(u8, raw, "TRUE") or std.mem.eql(u8, raw, "True")) return true;
    if (std.mem.eql(u8, raw, "0") or std.mem.eql(u8, raw, "f") or std.mem.eql(u8, raw, "F") or
        std.mem.eql(u8, raw, "false") or std.mem.eql(u8, raw, "FALSE") or std.mem.eql(u8, raw, "False")) return false;
    return error.InvalidBoolean;
}
test "boolean flags match Go strconv syntax" {
    try std.testing.expect(try parseBool("true"));
    try std.testing.expect(try parseBool("T"));
    try std.testing.expect(try parseBool("1"));
    try std.testing.expect(!(try parseBool("false")));
    try std.testing.expect(!(try parseBool("F")));
    try std.testing.expect(!(try parseBool("0")));
    try std.testing.expectError(error.InvalidBoolean, parseBool("yes"));
}
// === serve configuration ===
//
// The 65-branch flag chain that used to live inside `pub fn main`. Lifting it
// here is what lets the environment map disappear: both the parser and the
// JETSTREAM_* names now come from one place.

const archive_mod = @import("../storage/archive.zig");
const backfill_engine = @import("../ingest/backfill/engine.zig");
const lifecycle = @import("../ingest/backfill/lifecycle.zig");
const compact_pass = @import("../compact/pass.zig");
const retry_mod = @import("../ingest/backfill/retry.zig");
const cold = @import("../storage/cold.zig");
const xrpcapi = @import("../serve/xrpcapi.zig");
const environment = @import("environment.zig");
const crashpoint = @import("crashpoint.zig");
const log = std.log.scoped(.stream);

pub const default_subscribe_read_log_retention_bytes: usize = 256 * 1024 * 1024;

const RelayUrls = struct {
    http: []u8,
    websocket: []u8,

    fn deinit(self: RelayUrls, allocator: std.mem.Allocator) void {
        allocator.free(self.http);
        allocator.free(self.websocket);
    }
};

pub fn normalizeRelayUrls(allocator: std.mem.Allocator, relay_url: []const u8) !RelayUrls {
    const secure = std.mem.startsWith(u8, relay_url, "https://");
    const prefix_len: usize = if (secure)
        "https://".len
    else if (std.mem.startsWith(u8, relay_url, "http://"))
        "http://".len
    else
        return error.InvalidRelayUrl;
    const rest = relay_url[prefix_len..];
    const authority_len = std.mem.indexOfAny(u8, rest, "/?#") orelse rest.len;
    if (authority_len == 0 or std.mem.indexOfAny(u8, rest[0..authority_len], " \t\r\n") != null)
        return error.InvalidRelayUrl;
    const authority = rest[0..authority_len];
    const http = try std.fmt.allocPrint(allocator, "{s}://{s}", .{ if (secure) "https" else "http", authority });
    errdefer allocator.free(http);
    return .{
        .http = http,
        .websocket = try std.fmt.allocPrint(allocator, "{s}://{s}", .{ if (secure) "wss" else "ws", authority }),
    };
}

pub const ServeConfig = struct {
    upstream: []const u8 = "ws://localhost:7777",
    canonical_relay: ?RelayUrls = null,
    // the simulator serves DID docs itself; production passes --plc-url.
    plc_url: []const u8 = "http://localhost:7777",
    data_dir: []const u8 = "./data",
    cursor: ?i64 = null,
    port: u16 = 6008,
    legacy_port_set: bool = false,
    public_addr_raw: ?[]const u8 = null,
    debug_addr_raw: ?[]const u8 = null,
    shutdown_timeout: Io.Duration = .fromSeconds(5),
    client_drain_timeout: Io.Duration = .fromSeconds(10),
    otel_service_name: []const u8 = "jetstream",
    to_stdout: bool = false,
    do_verify: bool = true,
    max_segment_bytes: usize = archive_mod.default_max_segment_bytes,
    backfill_max_repos: usize = 0,
    backfill_repos_raw: ?[]const u8 = null,
    do_backfill: bool = false,
    backfill_workers: usize = backfill_engine.default_workers,
    backfill_batch_size: usize = backfill_engine.default_batch_size,
    backfill_async_flush_workers: usize = lifecycle.default_async_flush_workers,
    skip_merge_discovery: bool = false,
    rebloom_sweep: bool = false,
    relay_http: []const u8 = "http://localhost:7777",
    compaction_interval_ns: u64 = 4 * std.time.ns_per_hour,
    compaction_tombstone_cap: usize = compact_pass.default_tombstone_cap,
    compaction_rewrite_workers: usize = 0,
    retry_interval_ns: u64 = retry_mod.default_interval_ns,
    retry_workers: usize = retry_mod.default_workers,
    retry_host_workers: usize = retry_mod.default_host_workers,
    retry_max_delay_ns: u64 = retry_mod.default_max_delay_ns,
    timestamp_import_token: []const u8 = "",
    timestamp_import_dir_arg: ?[]const u8 = null,
    repo_action_rate_limits: bool = true,
    store_fault_prefix: ?[]const u8 = null,
    store_fault_ordinal: u64 = 1,
    segment_fault_spec: ?[]const u8 = null,
    segment_cache_max_age_s: u64 = 0,
    cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour,
    subscribe_read_log_retention_bytes: usize = default_subscribe_read_log_retention_bytes,
    subscribe_block_cache_bytes: usize = cold.default_block_cache_bytes,
    subscribe_slow_window_ns: u64 = server_mod.default_slow_window_ns,
    subscribe_slow_min_rate: f64 = server_mod.default_slow_min_rate,
    subscribe_read_batch: usize = server_mod.default_subscribe_read_batch,
    plan_config: xrpcapi.PlanConfig = .{},

    pub fn deinit(self: *ServeConfig, allocator: std.mem.Allocator) void {
        if (self.canonical_relay) |owned| owned.deinit(allocator);
        self.canonical_relay = null;
    }
};

pub const Parsed = struct {
    config: ServeConfig,
    explicit_serve: bool,
    log_level: []const u8,
    log_format: []const u8,
    /// Flags synthesized from JETSTREAM_* live here. The parsed config borrows
    /// into them, so they must outlive it: freeing them inside parseServe left
    /// every environment-supplied value dangling, which surfaced as BadPathName
    /// on the first path built from one.
    env_args: std.ArrayList([]u8),

    pub fn deinit(self: *Parsed, allocator: std.mem.Allocator) void {
        self.config.deinit(allocator);
        for (self.env_args.items) |arg| allocator.free(arg);
        self.env_args.deinit(allocator);
    }
};

/// Merge environment defaults under the command line, then apply every flag.
/// Precedence is upstream's: an explicit CLI value wins without the environment
/// value even being parsed.
pub fn parseServe(
    allocator: std.mem.Allocator,
    env_entries: []const []const u8,
    first_arg: ?[]const u8,
    arg_it: *std.process.Args.Iterator,
    explicit_serve_in: bool,
    log_level_in: []const u8,
    log_format_in: []const u8,
) !Parsed {
    var explicit_serve = explicit_serve_in;
    var log_level = log_level_in;
    var log_format = log_format_in;
    var cfg: ServeConfig = .{};
    errdefer cfg.deinit(allocator);
    if (explicit_serve) cfg.public_addr_raw = ":8080";
    cfg.otel_service_name = environment.value(env_entries, "OTEL_SERVICE_NAME") orelse "jetstream";
    if (cfg.otel_service_name.len == 0) cfg.otel_service_name = "jetstream";

    var env_serve_args: std.ArrayList([]u8) = .empty;
    errdefer {
        for (env_serve_args.items) |arg| allocator.free(arg);
        env_serve_args.deinit(allocator);
    }
    var cli_serve_args: std.ArrayList([]const u8) = .empty;
    defer cli_serve_args.deinit(allocator);
    if (first_arg) |arg| try cli_serve_args.append(allocator, arg);
    while (arg_it.next()) |arg| try cli_serve_args.append(allocator, arg);
    try environment.appendServeArguments(allocator, env_entries, cli_serve_args.items, &env_serve_args);
    var serve_args: std.ArrayList([]const u8) = .empty;
    defer serve_args.deinit(allocator);
    for (env_serve_args.items) |arg| try serve_args.append(allocator, arg);
    for (cli_serve_args.items) |arg| try serve_args.append(allocator, arg);

    var arg_index: usize = 0;
    while (arg_index < serve_args.items.len) : (arg_index += 1) {
        const arg = serve_args.items[arg_index];
        if (std.mem.startsWith(u8, arg, "--upstream=")) {
            cfg.upstream = arg["--upstream=".len..];
        } else if (std.mem.startsWith(u8, arg, "--relay-url=")) {
            const relay_url = arg["--relay-url=".len..];
            if (cfg.canonical_relay) |owned| {
                owned.deinit(allocator);
                cfg.canonical_relay = null;
            }
            cfg.canonical_relay = try normalizeRelayUrls(allocator, relay_url);
            cfg.upstream = cfg.canonical_relay.?.websocket;
            cfg.relay_http = cfg.canonical_relay.?.http;
        } else if (std.mem.startsWith(u8, arg, "--cursor=")) {
            cfg.cursor = try std.fmt.parseInt(i64, arg["--cursor=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--port=")) {
            cfg.port = try std.fmt.parseInt(u16, arg["--port=".len..], 10);
            cfg.legacy_port_set = true;
        } else if (std.mem.startsWith(u8, arg, "--addr=")) {
            cfg.public_addr_raw = arg["--addr=".len..];
        } else if (std.mem.startsWith(u8, arg, "--debug-addr=")) {
            const configured = arg["--debug-addr=".len..];
            cfg.debug_addr_raw = if (configured.len == 0) null else configured;
            if (cfg.public_addr_raw == null) cfg.public_addr_raw = ":8080";
        } else if (std.mem.startsWith(u8, arg, "--shutdown-timeout=")) {
            cfg.shutdown_timeout = try parseGoDuration(arg["--shutdown-timeout=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--client-drain-timeout=")) {
            cfg.client_drain_timeout = try parseGoDuration(arg["--client-drain-timeout=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--otel-service-name=")) {
            cfg.otel_service_name = arg["--otel-service-name=".len..];
        } else if (std.mem.eql(u8, arg, "--otel-service-name")) {
            arg_index += 1;
            if (arg_index >= serve_args.items.len) return error.BadArgs;
            cfg.otel_service_name = serve_args.items[arg_index];
        } else if (std.mem.startsWith(u8, arg, "--log-level=")) {
            log_level = arg["--log-level=".len..];
        } else if (std.mem.eql(u8, arg, "--log-level")) {
            arg_index += 1;
            if (arg_index >= serve_args.items.len) return error.BadArgs;
            log_level = serve_args.items[arg_index];
        } else if (std.mem.startsWith(u8, arg, "--log-format=")) {
            log_format = arg["--log-format=".len..];
        } else if (std.mem.eql(u8, arg, "--log-format")) {
            arg_index += 1;
            if (arg_index >= serve_args.items.len) return error.BadArgs;
            log_format = serve_args.items[arg_index];
        } else if (std.mem.startsWith(u8, arg, "--plc=")) {
            cfg.plc_url = arg["--plc=".len..];
        } else if (std.mem.startsWith(u8, arg, "--plc-url=")) {
            cfg.plc_url = arg["--plc-url=".len..];
        } else if (std.mem.startsWith(u8, arg, "--data-dir=")) {
            cfg.data_dir = arg["--data-dir=".len..];
        } else if (std.mem.startsWith(u8, arg, "--max-segment-bytes=")) {
            cfg.max_segment_bytes = try std.fmt.parseInt(usize, arg["--max-segment-bytes=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--backfill-max-repos=")) {
            cfg.backfill_max_repos = try std.fmt.parseInt(usize, arg["--backfill-max-repos=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--max-backfill-repos=")) {
            const configured = try std.fmt.parseInt(i64, arg["--max-backfill-repos=".len..], 10);
            cfg.backfill_max_repos = if (configured <= 0) 0 else std.math.cast(usize, configured) orelse return error.Overflow;
        } else if (std.mem.startsWith(u8, arg, "--backfill-repos=")) {
            cfg.backfill_repos_raw = arg["--backfill-repos=".len..];
        } else if (std.mem.startsWith(u8, arg, "--backfill-workers=")) {
            const configured = try std.fmt.parseInt(usize, arg["--backfill-workers=".len..], 10);
            cfg.backfill_workers = if (configured == 0) backfill_engine.default_workers else configured;
        } else if (std.mem.startsWith(u8, arg, "--backfill-batch-size=")) {
            const configured = try std.fmt.parseInt(usize, arg["--backfill-batch-size=".len..], 10);
            cfg.backfill_batch_size = if (configured == 0) backfill_engine.default_batch_size else configured;
        } else if (std.mem.startsWith(u8, arg, "--backfill-async-flush-workers=")) {
            cfg.backfill_async_flush_workers = try std.fmt.parseInt(usize, arg["--backfill-async-flush-workers=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--relay-http=")) {
            cfg.relay_http = arg["--relay-http=".len..];
        } else if (std.mem.startsWith(u8, arg, "--compaction-interval=")) {
            cfg.compaction_interval_ns = try parseDurationNanoseconds(arg["--compaction-interval=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--compaction-tombstone-cap=")) {
            cfg.compaction_tombstone_cap = try std.fmt.parseInt(usize, arg["--compaction-tombstone-cap=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--compaction-rewrite-workers=")) {
            cfg.compaction_rewrite_workers = try std.fmt.parseInt(usize, arg["--compaction-rewrite-workers=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--retry-interval=")) {
            const seconds = try std.fmt.parseInt(u64, arg["--retry-interval=".len..], 10);
            cfg.retry_interval_ns = try std.math.mul(u64, seconds, std.time.ns_per_s);
        } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-interval=")) {
            cfg.retry_interval_ns = try parseDurationNanoseconds(arg["--failed-repo-retry-interval=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-workers=")) {
            const configured = try std.fmt.parseInt(usize, arg["--failed-repo-retry-workers=".len..], 10);
            cfg.retry_workers = if (configured == 0) retry_mod.default_workers else configured;
        } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-host-workers=")) {
            const configured = try std.fmt.parseInt(usize, arg["--failed-repo-retry-host-workers=".len..], 10);
            cfg.retry_host_workers = if (configured == 0) retry_mod.default_host_workers else configured;
        } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-max-delay=")) {
            const configured = try parseDurationNanoseconds(arg["--failed-repo-retry-max-delay=".len..]);
            cfg.retry_max_delay_ns = if (configured == 0) retry_mod.default_max_delay_ns else configured;
        } else if (std.mem.startsWith(u8, arg, "--segment-cache-max-age=")) {
            cfg.segment_cache_max_age_s = try parseDurationCeilSeconds(arg["--segment-cache-max-age=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--cursor-lookback=")) {
            cfg.cursor_lookback_ns = try parseDurationNanoseconds(arg["--cursor-lookback=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--subscribe-read-log-retention-bytes=")) {
            const configured = try std.fmt.parseInt(i64, arg["--subscribe-read-log-retention-bytes=".len..], 10);
            cfg.subscribe_read_log_retention_bytes = if (configured <= 0)
                default_subscribe_read_log_retention_bytes
            else
                std.math.cast(usize, configured) orelse return error.Overflow;
        } else if (std.mem.startsWith(u8, arg, "--subscribe-block-cache-bytes=")) {
            const configured = try std.fmt.parseInt(i64, arg["--subscribe-block-cache-bytes=".len..], 10);
            cfg.subscribe_block_cache_bytes = if (configured <= 0)
                cold.default_block_cache_bytes
            else
                std.math.cast(usize, configured) orelse return error.Overflow;
        } else if (std.mem.startsWith(u8, arg, "--subscribe-slow-window=")) {
            cfg.subscribe_slow_window_ns = try effectiveSubscribeSlowWindowNanoseconds(arg["--subscribe-slow-window=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--subscribe-slow-min-rate=")) {
            const configured = try std.fmt.parseFloat(f64, arg["--subscribe-slow-min-rate=".len..]);
            cfg.subscribe_slow_min_rate = if (configured <= 0) server_mod.default_slow_min_rate else configured;
        } else if (std.mem.startsWith(u8, arg, "--subscribe-read-batch=")) {
            const configured = try std.fmt.parseInt(i64, arg["--subscribe-read-batch=".len..], 10);
            cfg.subscribe_read_batch = if (configured <= 0)
                server_mod.default_subscribe_read_batch
            else
                std.math.cast(usize, configured) orelse return error.Overflow;
        } else if (std.mem.startsWith(u8, arg, "--cursor-block-index-cache-size=")) {
            // Upstream compatibility no-op. Sealed block indexes are always
            // resident in the refcounted manifest, so there is no cache to
            // size; accept the full signed CLI surface exactly as cfg.upstream.
            _ = try std.fmt.parseInt(i64, arg["--cursor-block-index-cache-size=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--plan-max-dids=")) {
            cfg.plan_config.max_dids = try std.fmt.parseInt(usize, arg["--plan-max-dids=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--plan-max-collections=")) {
            cfg.plan_config.max_collections = try std.fmt.parseInt(usize, arg["--plan-max-collections=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--plan-max-entries=")) {
            cfg.plan_config.max_entries = try std.fmt.parseInt(usize, arg["--plan-max-entries=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--plan-whole-segment-threshold=")) {
            cfg.plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".len..]);
        } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) {
            cfg.timestamp_import_token = arg["--timestamp-import-token=".len..];
        } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) {
            cfg.timestamp_import_dir_arg = arg["--timestamp-import-dir=".len..];
        } else if (std.mem.startsWith(u8, arg, "--crashpoint=")) {
            // debug/oracle only: NAME or NAME:ORDINAL
            const spec = arg["--crashpoint=".len..];
            const split = std.mem.lastIndexOfScalar(u8, spec, ':');
            const name = if (split) |i| spec[0..i] else spec;
            const ordinal = if (split) |i| try std.fmt.parseInt(u64, spec[i + 1 ..], 10) else 1;
            if (!crashpoint.armAt(name, ordinal)) {
                log.err("invalid crashpoint: {s}", .{spec});
                return error.BadArgs;
            }
        } else if (std.mem.startsWith(u8, arg, "--power-cutpoint=")) {
            // Linux strict-storage oracle only: NAME or NAME:ORDINAL.
            const spec = arg["--power-cutpoint=".len..];
            const split = std.mem.lastIndexOfScalar(u8, spec, ':');
            const name = if (split) |i| spec[0..i] else spec;
            const ordinal = if (split) |i| try std.fmt.parseInt(u64, spec[i + 1 ..], 10) else 1;
            if (!crashpoint.armPowerCut(name, ordinal)) {
                log.err("invalid power cutpoint: {s}", .{spec});
                return error.BadArgs;
            }
        } else if (std.mem.startsWith(u8, arg, "--store-fault-prefix=")) {
            // debug/oracle only: fail one matching durable RocksDB mutation
            cfg.store_fault_prefix = arg["--store-fault-prefix=".len..];
        } else if (std.mem.startsWith(u8, arg, "--store-fault-ordinal=")) {
            cfg.store_fault_ordinal = try std.fmt.parseInt(u64, arg["--store-fault-ordinal=".len..], 10);
        } else if (std.mem.startsWith(u8, arg, "--segment-fault=")) {
            // debug/oracle only: fail one real segment operation
            cfg.segment_fault_spec = arg["--segment-fault=".len..];
        } else if (std.mem.eql(u8, arg, "--backfill")) {
            cfg.do_backfill = true;
        } else if (std.mem.eql(u8, arg, "--rebloom-sweep")) {
            cfg.rebloom_sweep = true;
        } else if (std.mem.startsWith(u8, arg, "--rebloom-sweep=")) {
            cfg.rebloom_sweep = try parseBool(arg["--rebloom-sweep=".len..]);
        } else if (std.mem.eql(u8, arg, "--skip-merge-discovery")) {
            cfg.skip_merge_discovery = true;
        } else if (std.mem.startsWith(u8, arg, "--skip-merge-discovery=")) {
            cfg.skip_merge_discovery = try parseBool(arg["--skip-merge-discovery=".len..]);
        } else if (std.mem.eql(u8, arg, "--no-verify")) {
            cfg.do_verify = false;
        } else if (std.mem.eql(u8, arg, "--disable-repo-action-rate-limits")) {
            cfg.repo_action_rate_limits = false;
        } else if (std.mem.startsWith(u8, arg, "--disable-repo-action-rate-limits=")) {
            cfg.repo_action_rate_limits = !(try parseBool(arg["--disable-repo-action-rate-limits=".len..]));
        } else if (std.mem.eql(u8, arg, "--stdout")) {
            cfg.to_stdout = true;
        } else if (std.mem.eql(u8, arg, "serve") and !explicit_serve) {
            // Root flags are persistent cfg.upstream, so `--log-level=debug
            // serve` is equivalent to `serve --log-level=debug`.
            explicit_serve = true;
            if (cfg.public_addr_raw == null) cfg.public_addr_raw = ":8080";
        } else {
            log.err("unknown arg: {s}", .{arg});
            return error.BadArgs;
        }
    }
    return .{ .config = cfg, .explicit_serve = explicit_serve, .log_level = log_level, .log_format = log_format, .env_args = env_serve_args };
}

// === client identity ===

const build_options = @import("build_options");

/// What Stream tells the servers it calls.
///
/// A whole-network crawl touches thousands of independently-operated PDSes. An
/// operator seeing unexplained load should be able to tell what is calling and
/// where to ask about it; std.http's default (`zig/<version> (std.http)`) says
/// neither, and Stream showed up in a real operator's logs as exactly that.
/// Every other client on the network identifies itself.
///
/// Derived from build_options so it cannot go stale the way a literal would.
pub fn userAgent() []const u8 {
    return "stream/" ++ build_options.version ++ " (+https://tangled.org/zat.dev/stream)";
}

test "the user agent names the software and a way to reach its operator" {
    const ua = userAgent();
    try testing.expect(std.mem.startsWith(u8, ua, "stream/"));
    try testing.expect(std.mem.indexOf(u8, ua, "tangled.org/zat.dev/stream") != null);
    // the std default is what we are replacing; never ship it
    try testing.expect(std.mem.indexOf(u8, ua, "std.http") == null);
}