# The swarm's telemetry collector: one per swarm, in a `swarm-otel`
# nixos-container beside the swarm's other shared services.
#
# Two tiers, and they are separate on purpose:
#
#   - `otel.nix` is the **hive** tier. It receives from this hive's agents
#     on the bridge and forwards, and it holds no upstream credential.
#   - this is the **swarm** tier. It is the only holder of the upstream
#     credential, the only writer to the swarm's metrics store, and the
#     place that will stamp `hive=` from the authenticated connection
#     rather than from anything a sender can choose.
#
# On a host that runs both, both processes run. They are not collapsed:
# all-local is a statement about *where* processes run, not about what
# shape the deployment has, and a local tier boundary that disappears is
# one the local deployment stops testing. `hive=` attribution is the
# property that would differ, and the ingest auth that makes it
# unforgeable is built on this boundary existing.
#
# A container rather than a second host unit, for the same reason every
# sibling swarm service is one — and because `services.opentelemetry-collector`
# is a singleton NixOS option, already spoken for on the host by the hive
# tier. A container gets its own evaluation and therefore its own collector.
{
  pkgs,
  lib,
  config,
  ...
}:
let
  cfg = config.services.hyperhive.swarm.otel;
  swarmCfg = config.services.hyperhive.swarm;
  otelCfg = config.services.hyperhive.otel;
  vmCfg = config.services.hyperhive.swarm.victoriametrics;
  hyperhiveCfg = config.services.hyperhive;
  gatewayCfg = hyperhiveCfg.gateway;
  swarmDomain = hyperhiveCfg.swarm.domain;

  # Total on a null swarm domain for the same reason every sibling module is:
  # the required-domain assertion in hive-network.nix should be what an
  # operator sees, not a coercion error from here.
  domainBase = if swarmDomain == null then "invalid" else swarmDomain;

  autheliaCfg = hyperhiveCfg.swarm.authelia;

  # Whether ingest is authenticated per hive. Derived rather than declared
  # so the common case needs no attribute, but an option (below) because
  # this decides whether the unauthenticated receiver exists at all.
  hiveAuth = cfg.requireHiveIdentity;

  # `attrNames` is sorted, so this is a function of the hive SET and not of
  # the order anyone wrote it in.
  #
  # These ports are internal and appear in no URL: a hive addresses its own
  # receiver as a PATH on this collector's single gateway name, and nginx —
  # rendered from this same evaluation — is the only thing that ever names
  # the port. That is what makes deriving them safe here and unsafe in the
  # obvious other place: were a hive told a port, inserting a hive would
  # renumber the ones after it and silently move a port a running hive was
  # already sending to.
  hivePorts = lib.listToAttrs (
    lib.imap0 (i: h: lib.nameValuePair h (cfg.hivePortBase + i)) (
      lib.attrNames hyperhiveCfg.swarm.hives
    )
  );

  # The swarm's authelia is reached by its gateway name, whose leaf is
  # issued by the swarm services sub-CA — so this container needs the same
  # runtime CA trust every other consumer of a swarm-service name needs.
  # The CA is generated at runtime and cannot be baked into a derivation,
  # which is why it arrives as a bind mount rather than
  # `security.pki.certificateFiles`.
  caTrust = import ./lib/hive-ca-trust.nix {
    inherit lib;
    tlsCfg = hyperhiveCfg.tls;
    inherit gatewayCfg;
  };
  caBundle = caTrust.bundlePathFor cfg.machine;

  # One list, read by every pipeline: the per-hive pipelines fan out to
  # exactly the same destinations as the single pipeline they replace.
  # Written once because "which exporters" is a property of this tier, not
  # of which hive a sample came from.
  exporterNames =
    lib.optional (otelCfg.endpoint != "") (if otelCfg.protocol == "grpc" then "otlp" else "otlphttp")
    ++ lib.optional vmCfg.enable "otlphttp/victoriametrics";
in
{
  options.services.hyperhive.swarm.otel = {
    enable = lib.mkOption {
      type = lib.types.bool;
      default = false;
      description = ''
        Run the swarm's telemetry collector on this host.

        Asserted from `swarm.enableRequiredServices` in
        ./swarm-required-services.nix, with the metrics pair this
        collector feeds: a swarm has one of these, and it belongs
        wherever the shared services live rather than on every hive.

        A hive that does not run it still runs its own hive-tier collector
        (`services.hyperhive.otel.enable`) and reaches this one by name, at
        {option}`services.hyperhive.swarm.otel.domain`.
      '';
    };

    machine = lib.mkOption {
      type = lib.types.str;
      readOnly = true;
      default = "swarm-otel";
      description = ''
        Name of the nixos-container this collector runs in — also the
        `machinectl` name, so other modules may read it rather than
        repeating the literal.
      '';
    };

    port = lib.mkOption {
      type = lib.types.port;
      default = 4319;
      description = ''
        Port this collector's OTLP/HTTP receiver listens on.

        ⚠️ **Deliberately not 4318**, the OTLP/HTTP default, because the
        hive tier already uses it (`services.hyperhive.otel.collector.port`)
        and every swarm container shares the host's network namespace. Two
        listeners claiming one port on one host is not a build failure —
        it is a runtime coin toss over which one gets it, with nothing in
        any log saying so. The same collision cost a release when grafana
        and the forge both defaulted to 3000.
      '';
    };

    telemetryPort = lib.mkOption {
      type = lib.types.port;
      default = 8889;
      description = ''
        Port this collector serves its **own** metrics on — queue depth,
        refused and dropped samples, exporter failures. How you find out
        that telemetry is being lost, so it is worth keeping rather than
        switching off.

        ⚠️ **Deliberately not 8889's neighbour 8888**, which is the
        collector's built-in default and therefore what the hive tier
        already binds. Two collectors share a network namespace whenever
        they are co-located, and unlike the OTLP port this one appears
        nowhere in either config — it is a default inside the binary, so
        nothing that compares configured ports can see the clash. The
        second collector to start simply dies with
        `bind: address already in use`.
      '';
    };

    requireHiveIdentity = lib.mkOption {
      type = lib.types.bool;
      default = autheliaCfg.enable && autheliaCfg.url != null;
      defaultText = lib.literalExpression "swarm.authelia.enable && swarm.authelia.url != null";
      description = ''
        Authenticate ingest per hive: each hive gets its own receiver,
        verifying an OIDC token minted for that hive's audience, and the
        `hive` label is stamped from **which receiver accepted the
        sample** rather than from anything the sender wrote.

        ⚠️ Turning this on **removes the unauthenticated receiver**. That
        is the point rather than a side effect: while an unauthenticated
        port still accepts samples, any writer that can reach this
        collector can still attribute metrics to any hive, and the
        per-hive receivers are decoration.

        Defaults to whether this swarm has an authelia to mint against.
        Set it false to keep the open receiver on a swarm where every
        writer is already trusted — an explicit choice, which is what it
        should be.
      '';
    };

    hivePortBase = lib.mkOption {
      type = lib.types.port;
      default = 4330;
      description = ''
        First port of the per-hive receiver range; each hive in
        {option}`services.hyperhive.swarm.hives` takes the next one, in
        sorted-name order.

        ⚠️ Internal. No client is told a port — a hive reaches its own
        receiver as `https://''${domain}/<hive>`, and the gateway routes
        the path. So inserting a hive renumbering the ones after it is
        harmless here: nginx is rendered from this same evaluation and
        moves with it.

        The range still matters because every swarm container shares the
        host's network namespace, so a derived port can land on one
        another service already holds — with no bind error and nothing in
        any log. The assertions below check this range against every port
        this module and the hive tier declare — which is as far as a
        module can see, since a port another module picks is not
        knowable from here without reading its config.
      '';
    };

    domain = lib.mkOption {
      type = lib.types.str;
      default = "otel.${domainBase}";
      defaultText = lib.literalExpression ''"otel.''${services.hyperhive.swarm.domain}"'';
      description = ''
        Name the gateway serves this on. A sibling of the swarm's other
        service names, so the swarm-services sub-CA can issue for it — see
        `hive-tls.nix` for why a service name being a sibling rather than a
        child decides which CA may sign it.

        This is what the **hive** tier's exporter reaches — the hive
        collector is a plain producer against this name exactly like every
        other client of a swarm service, resolved locally by dnsmasq on a
        co-located host and over the real network otherwise. There is no
        separate loopback-vs-remote knob to get wrong: `swarm-nats` is the
        deliberate exception to this pattern (its cross-hive reach is the
        wireguard mesh, not the gateway), everything else in this swarm
        addresses its siblings by name.
      '';
    };
  };

  config = lib.mkIf (config.services.hyperhive.enable && cfg.enable) {
    # The gateway name, inside `cfg.enable` — that guard is the load-bearing
    # part. Every hive in a swarm may know this collector exists, but only
    # the host that RUNS it may claim the name; a client hive declaring the
    # vhost would answer for a service it does not have.
    services.hyperhive.gateway.localNames = [ cfg.domain ];

    # OTLP/HTTP, not a browsable UI, but the same reverse-proxy shape as
    # every sibling swarm service: TLS terminates here, then plain http to
    # the co-located container over loopback (shared netns, like the store
    # this collector writes to).
    services.nginx.virtualHosts."${cfg.domain}" = (gatewayCfg.lib.tlsFor cfg.domain) // {
      listen = gatewayCfg.lib.listen;
      extraConfig = gatewayCfg.lib.securityHeaders;
      locations =
        # One name for the whole collector, and the hive is a path under
        # it. The alternative — a vhost per hive — needs a certificate,
        # a DNS name and a `localNames` entry per hive to express the
        # same routing the gateway already does for free.
        #
        # ⚠️ The trailing slash on both sides is load-bearing: it is what
        # strips `/<hive>` before the request reaches the receiver, which
        # serves `/v1/metrics` and knows nothing about hives. Without it
        # the receiver sees `/<hive>/v1/metrics` and answers 404 to a
        # request that authenticated perfectly.
        lib.optionalAttrs hiveAuth (
          lib.mapAttrs' (
            h: p: lib.nameValuePair "/${h}/" { proxyPass = "http://127.0.0.1:${toString p}/"; }
          ) hivePorts
        )
        // {
          "/" =
            if hiveAuth then
              # Not a proxy to a receiver that no longer exists. A closed
              # door answering 404 is the honest description of this
              # collector once ingest is per-hive: there is no
              # swarm-wide inbox any more.
              { return = "404"; }
            else
              { proxyPass = "http://127.0.0.1:${toString cfg.port}"; };
        };
    };

    # Turning on the identities this tier authenticates against, which the
    # option exists to allow: its own description names this module as the
    # second consumer, so the queue is not a prerequisite for authenticated
    # telemetry.
    services.hyperhive.swarm.authelia.oidc.hiveIdentities = lib.mkIf hiveAuth true;

    # The CA bind source is written at runtime by a host unit, so the
    # container has to start after it — otherwise nspawn sets up a mount
    # over a file that does not exist yet.
    systemd.services."container@${cfg.machine}" = lib.mkIf hiveAuth caTrust.containerOrdering;

    assertions = [
      {
        # The tier exists to hold the upstream credential and to write the
        # swarm's store. With neither, it is a process that receives
        # samples and drops them — which looks healthy and loses data.
        assertion = otelCfg.endpoint != "" || vmCfg.enable;
        message = ''
          services.hyperhive.swarm.otel.enable is true but this collector
          has nowhere to send what it receives:
          services.hyperhive.otel.endpoint is empty and
          services.hyperhive.swarm.victoriametrics.enable is false.

          Set the endpoint to export upstream, or enable the swarm's
          metrics store.
        '';
      }
      {
        # Without a roster there are no receivers at all, so this
        # collector would listen on nothing while looking configured.
        assertion = !hiveAuth || hyperhiveCfg.swarm.hives != { };
        message = ''
          services.hyperhive.swarm.otel.requireHiveIdentity is true but
          services.hyperhive.swarm.hives is empty: ingest is authenticated
          per hive, so an empty roster means this collector accepts
          nothing from anyone.

          List the swarm's hives, or set requireHiveIdentity = false to
          keep an unauthenticated receiver.
        '';
      }
      {
        # A port collision between two listeners on one host is a runtime
        # coin toss with nothing in any log — the failure this whole
        # comment budget exists to prevent. Checked against every port
        # reachable from here; a port some other module picks is not.
        assertion =
          let
            derived = lib.attrValues hivePorts;
            others = [
              cfg.port
              cfg.telemetryPort
              otelCfg.collector.port
            ]
            ++ lib.optional vmCfg.enable vmCfg.port;
            all = derived ++ others;
          in
          !hiveAuth || lib.length (lib.unique all) == lib.length all;
        message = ''
          services.hyperhive.swarm.otel: the per-hive receiver range
          starting at hivePortBase (${toString cfg.hivePortBase}, one port
          per hive in services.hyperhive.swarm.hives) overlaps another
          port on this host.

          Every swarm container shares the host's network namespace, so
          two listeners claiming one port is not a build failure — it is
          whichever process started first, silently. Move hivePortBase to
          a free range.
        '';
      }
    ];

    containers.${cfg.machine} = {
      autoStart = true;
      ephemeral = false;
      # Shared host netns, like every sibling swarm service: the hive tier
      # reaches this collector, and this collector reaches the metrics
      # store, without either crossing a network boundary that would need
      # its own trust material.
      privateNetwork = false;

      # The upstream credential is operator-provided and lives on the host.
      # Read-only, and only when one is configured — binding a path that
      # does not exist makes nixos-container refuse to start the container,
      # which is a stall several layers from its cause.
      bindMounts =
        lib.optionalAttrs (otelCfg.headersCredential != null) {
          ${otelCfg.headersCredential} = {
            hostPath = otelCfg.headersCredential;
            isReadOnly = true;
          };
        }
        # The public hive CA, read-only — only when something in here
        # actually verifies a swarm-service name.
        // lib.optionalAttrs hiveAuth caTrust.bindMount;

      config =
        { ... }:
        {
          # This tier is the one that resolves an operator-configured
          # hostname: `otel.endpoint` is an external URL, and reaching it
          # is the entire reason this container holds a credential. The
          # `/etc/resolv.conf` nixos-containers copies in is a snapshot
          # taken once at boot, so without this the upstream export
          # depends on the host's file having been right at that instant.
          imports = [
            (import ./swarm-container-resolver.nix {
              inherit (config.services.hyperhive.network) bridgeIp;
              dnsConsumers = [ "opentelemetry-collector.service" ];
            })
          ]
          # `SSL_CERT_FILE` REPLACES the trust store rather than adding to
          # it, so a failed assembly yields an empty pool and every TLS
          # call fails while the unit looks healthy. That is why this is
          # the shared helper — it carries the `Requires` and the
          # non-empty check — and not a local `cat`.
          ++ lib.optional hiveAuth (
            caTrust.trustBundle {
              inherit pkgs;
              name = cfg.machine;
              consumers = [ "opentelemetry-collector" ];
            }
          );

          system.stateVersion = config.system.stateVersion;
          networking.firewall.enable = false;
          # Keep the host-copied /etc/resolv.conf intact — same reasoning
          # as the sibling swarm containers.
          networking.resolvconf.enable = lib.mkForce false;

          services.opentelemetry-collector = {
            enable = true;
            package = pkgs.opentelemetry-collector-contrib;
            # Runs `otelcol validate` at build time. ⚠️ A parser, not a
            # wiring check: it accepts a receiver naming an absent
            # extension, and the collector then dies at startup. A green
            # build does not prove this config starts, never mind that a
            # sample arrives — which is why this module's gate pushes a
            # real sample through both tiers into the store.
            validateConfigFile = true;
            settings = {
              # One receiver per hive when ingest is authenticated, and
              # that multiplicity is forced rather than chosen. The `hive`
              # label has to come from something the sender cannot write,
              # and the only such thing here is WHICH RECEIVER accepted
              # the sample: a processor cannot read the token's claims
              # (`from_context` reads request metadata, and asking it for
              # an auth claim yields nothing — silently, with a healthy
              # startup), and one receiver holding many credentials never
              # reveals which one matched.
              receivers =
                if hiveAuth then
                  lib.mapAttrs' (
                    h: p:
                    lib.nameValuePair "otlp/${h}" {
                      protocols.http = {
                        endpoint = "127.0.0.1:${toString p}";
                        auth.authenticator = "oidc/${h}";
                      };
                    }
                  ) hivePorts
                else
                  { otlp.protocols.http.endpoint = "127.0.0.1:${toString cfg.port}"; };

              exporters =
                lib.optionalAttrs vmCfg.enable {
                  # `metrics_endpoint`, NOT `endpoint`: the latter is a
                  # base that otlphttp appends `/v1/metrics` to, while
                  # VictoriaMetrics serves OTLP at
                  # `/opentelemetry/api/v1/push`. With `endpoint` the
                  # collector answers 200 to its own clients and posts the
                  # samples to a path that does not exist. Measured
                  # end-to-end, not read — `state/probe-3265-collector-to-vm.sh`.
                  "otlphttp/victoriametrics".metrics_endpoint =
                    "http://127.0.0.1:${toString vmCfg.port}/opentelemetry/api/v1/push";
                }
                // lib.optionalAttrs (otelCfg.endpoint != "") {
                  ${if otelCfg.protocol == "grpc" then "otlp" else "otlphttp"} = {
                    endpoint = otelCfg.endpoint;
                  }
                  // lib.optionalAttrs (otelCfg.headersCredential != null) {
                    # Interpolated by the collector from its environment at
                    # runtime, never by nix: `EnvironmentFile` below is what
                    # puts it there, so the value is not read into the store.
                    headers.${otelCfg.collector.upstreamHeaderName} = "\${env:${otelCfg.collector.upstreamHeaderName}}";
                  }
                  // lib.optionalAttrs (otelCfg.protocol == "http/json") { encoding = "json"; };
                };

              # Moves this collector's self-metrics off the built-in
              # default of `localhost:8888`, which the hive tier holds.
              #
              # ⚠️ `metrics.address` is the spelling that looks right and
              # is REJECTED by this collector version — measured, not
              # read: `'migration.MetricsConfigV030' has invalid keys:
              # address`. `readers` is the schema it accepts, and the
              # difference is a startup failure rather than a warning.
              service.telemetry.metrics.readers = [
                {
                  pull.exporter.prometheus = {
                    host = "127.0.0.1";
                    port = cfg.telemetryPort;
                  };
                }
              ];

              # ⚠️ An extension that is configured but not listed here is
              # INERT — the collector starts clean and the receiver
              # naming it authenticates nothing. Derived from the same
              # attrset as the receivers so the two cannot disagree.
              service.extensions = lib.optionals hiveAuth (map (h: "oidc/${h}") (lib.attrNames hivePorts));

              # Fan-out, not a choice: with both configured the same
              # samples go upstream AND into the swarm's store. The store
              # is for looking at this swarm; the upstream is for whoever
              # aggregates across swarms, and neither replaces the other.
              # `exporterNames` is shared by every pipeline — where a
              # sample goes is a property of this tier, not of the hive
              # that sent it.
              service.pipelines =
                if hiveAuth then
                  lib.mapAttrs' (
                    h: _:
                    lib.nameValuePair "metrics/${h}" {
                      receivers = [ "otlp/${h}" ];
                      processors = [ "resource/${h}" ];
                      exporters = exporterNames;
                    }
                  ) hivePorts
                else
                  {
                    metrics = {
                      receivers = [ "otlp" ];
                      exporters = exporterNames;
                    };
                  };
            }
            // lib.optionalAttrs hiveAuth {
              extensions = lib.mapAttrs' (
                h: _:
                lib.nameValuePair "oidc/${h}" {
                  issuer_url = autheliaCfg.url;
                  # The audience this hive's client is registered to
                  # request, and the reason one hive's token is refused by
                  # another hive's receiver. Same expression authelia
                  # registers it under — a second spelling here would deny
                  # every hive, as a 401 that blames the token.
                  audience = "${autheliaCfg.hiveClientPrefix}${h}";
                  # ⚠️ `issuer_ca_path`. `issuer_ca_file`, `ca_file` and
                  # `tls.ca_file` are all INVALID KEYS for this extension
                  # — measured, and the failure is a startup error naming
                  # the key rather than anything about certificates.
                  issuer_ca_path = caBundle;
                }
              ) hivePorts;

              # `upsert`, not `insert`: a sender that stamps its own
              # `hive` must be OVERWRITTEN, not deferred to. This
              # processor is the whole attribution boundary — the value
              # is a constant per receiver, so it says which hive
              # authenticated, not which hive claimed to be sending.
              processors = lib.mapAttrs' (
                h: _:
                lib.nameValuePair "resource/${h}" {
                  attributes = [
                    {
                      key = "hive";
                      value = h;
                      action = "upsert";
                    }
                  ];
                }
              ) hivePorts;
            };
          };

          # The credential file is already `NAME=value`, systemd's
          # EnvironmentFile format — so the secret reaches the process as an
          # environment variable without being read by nix, written to the
          # store, or passed in argv.
          systemd.services.opentelemetry-collector.serviceConfig =
            lib.optionalAttrs (otelCfg.headersCredential != null)
              {
                EnvironmentFile = otelCfg.headersCredential;
              };
        };
    };
  };
}