# Vector: collects from the two taps in LOGGING.md and ships to VictoriaLogs.
#
#   docker_logs  every container's stdout/stderr, with the vanguard.service / vanguard.category labels
#                we put on each one, so filtering by category needs no parsing
#   journald     our own systemd units (dumpd, the sweep, trash purge, deploy pending, vgdeband)
#
# TRADE-OFF, chosen deliberately (CommunityDiscordMedia.md Phase 13): docker_logs needs the Docker API, so this
# container holds the socket and therefore root-equivalent access to the host — the same exposure act_runner
# already has. Accepted "for now". The upgrade paths, in order of preference, are the journald log driver for
# every container (one tap, no socket, labels survive as journal fields) or a read-only socket proxy in front
# of this container. Neither changes anything downstream of here.
#
# Every event ends up with: service, category, source_kind, message, level where it could be determined.

data_dir: /var/lib/vector

sources:
  containers:
    type: docker_logs
    docker_host: unix:///var/run/docker.sock
    exclude_containers: [vector]        # never ship our own output: it would feed itself
  units:
    type: journald
    current_boot_only: false
    journal_directory: /var/log/journal
    include_units: [dumpd, normalize-sweep, trash-purge, deploy-pending, vgdeband]

transforms:
  # ---- one shape, whichever tap it came from
  normalised:
    type: remap
    inputs: [containers, units]
    source: |
      if exists(.container_name) {
        .service = .label."vanguard.service" || .container_name
        .category = .label."vanguard.category" || "infra"
        .source_kind = "container"
        del(.label); del(.container_id); del(.image); del(.container_created_at); del(.stream)
      } else {
        unit = ._SYSTEMD_UNIT || .unit || "host"
        .service = replace(string!(unit), ".service", "")
        .category = if .service == "dumpd" { "stream" } else { "media" }
        .source_kind = "unit"
        if !exists(.message) && exists(.MESSAGE) { .message = del(.MESSAGE) }
        # the journal carries a lot of bookkeeping; keep what a query or a human needs
        del(.PRIORITY); del(.SYSLOG_IDENTIFIER); del(.SYSLOG_FACILITY); del(._TRANSPORT)
        del(._PID); del(._UID); del(._GID); del(._COMM); del(._EXE); del(._CMDLINE)
        del(._CAP_EFFECTIVE); del(._SELINUX_CONTEXT); del(._AUDIT_SESSION); del(._AUDIT_LOGINUID)
        del(._SYSTEMD_CGROUP); del(._SYSTEMD_SLICE); del(._SYSTEMD_INVOCATION_ID)
        del(._BOOT_ID); del(._MACHINE_ID); del(._SOURCE_REALTIME_TIMESTAMP); del(._SYSTEMD_UNIT)
      }

  # ---- parse the shapes LOGGING.md documents; anything else stays a plain message
  parsed:
    type: remap
    inputs: [normalised]
    source: |
      text = string(.message) ?? ""
      if starts_with(text, "{") {
        # caddy and authentik emit JSON
        fields, err = parse_json(text)
        if err == null && is_object(fields) {
          .json = fields
          .level = get(object!(fields), ["level"]) ?? get(object!(fields), ["severity"]) ?? null
          .message = get(object!(fields), ["msg"]) ?? get(object!(fields), ["event"]) ?? text
        }
      } else if match(text, r'(^|\s)(ts|level|component|event)=') {
        # our own scripts, and syncthing's key=value tail
        fields, err = parse_logfmt(text)
        if err == null {
          .kv = fields
          .level = get(object!(fields), ["level"]) ?? null
          .event = get(object!(fields), ["event"]) ?? null
        }
      } else if match(text, r'^\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} ') {
        # mediamtx: 2026/11/03 05:00:00 INF [component] message
        fields, err = parse_regex(text, r'^\S+ \S+ (?P<lvl>\w+) (?P<rest>.*)$')
        if err == null { .level = fields.lvl; .message = fields.rest }
      } else if match(text, r'^\[\d{2}:\d{2}:\d{2}') {
        # jellyfin: [00:00:00.000] [LVL] [thread] Logger: message   (stack traces arrive as their own events)
        fields, err = parse_regex(text, r'^\[[\d:.]+\] \[(?P<lvl>\w+)\] \[\d+\] (?P<rest>.*)$')
        if err == null { .level = fields.lvl; .message = fields.rest }
      }
      if exists(.level) { .level = downcase(string!(.level)) }

sinks:
  victorialogs:
    type: elasticsearch
    inputs: [parsed]
    endpoints: ["http://victorialogs:9428/insert/elasticsearch/"]
    mode: bulk
    api_version: v8
    compression: gzip
    healthcheck:
      enabled: false
    query:
      _msg_field: message
      _time_field: timestamp
      _stream_fields: service,category,source_kind