# B2G log-parsing pipeline.
#
# Produces a uniform document for every log line with these facet fields:
#   service         — b2g-micro, b2g-usermgmt, kong, fe-b2g, ...  ("by container")
#   level           — error | warning | info                     ("issue type" #1)
#   category        — keycloak-sso | bridge | payment | web | ... ("issue type" #2)
#   @timestamp      — real event time (Asia/Muscat)              ("date/time")
#   msg             — the human-readable message (secrets scrubbed)
#   log_source      — appfile | container
#
# Input types (set by Filebeat, see ../filebeat.yml):
#   [log_source]=appfile, [log_format]=yii            -> Yii FileTarget lines
#   [log_source]=appfile, [log_format]=digitall-job   -> digitall free-text jobs
#   (no [log_source])                                 -> raw container stdout/stderr

input {
  beats {
    port => 5044
  }
}

filter {

  # ---- 1. Container-stream events: tag source + derive a clean service name ----
  #   Compose names containers <project>-<service>-<n> (e.g. b2gnew-b2g-micro-1).
  #   Strip the project prefix and the numeric suffix to recover "b2g-micro".
  if ![log_source] {
    mutate { add_field => { "log_source" => "container" } }
    if [container][name] {
      grok {
        match => { "[container][name]" => "^%{DATA:_proj}-(?<service>.+)-%{INT:_idx}$" }
        tag_on_failure => []
      }
      if ![service] {
        mutate { add_field => { "service" => "%{[container][name]}" } }
      }
    }
  }

  # Safety net: never index the ELK stack's own logs (Filebeat already excludes
  # them; this guards against a feedback loop if that filter is ever changed).
  if [service] in ["elasticsearch", "logstash", "kibana", "filebeat", "es-init"] {
    drop { }
  }

  # ---- 2. Derive service from the file path for the shared basic-backend input ----
  if [log_source] == "appfile" and ![service] {
    grok { match => { "[log][file][path]" => "/repo/(?<service>b2g_[^/]+)/" } }
    # Directory is b2g_micro; container name is b2g-micro. Normalize to match.
    mutate { gsub => ["service", "_", "-"] }
  }

  # ---- 3. De-dup for backend containers ----
  #   The backends mirror Yii app logs to stderr (for `docker logs` visibility),
  #   so those app lines ALSO arrive on the container stream — single- AND
  #   multi-line. The file input is the authoritative, fully-parsed copy, so on
  #   the container stream we keep ONLY genuine nginx lines and drop the rest:
  #     - nginx access log starts with a client IP  (^1.2.3.4 )
  #     - nginx error  log starts with YYYY/MM/DD    (^2026/07/21 )
  #   Everything else from a b2g-* backend (Yii mirror, stack-trace continuation
  #   lines, php-fpm lifecycle notices) is dropped here.
  if [log_source] == "container" and [service] =~ /^b2g-/ {
    if [message] =~ /^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3} / {
      mutate { add_field => { "category" => "access" } }
    } else if [message] =~ /^\d{4}\/\d{2}\/\d{2} / {
      mutate { add_field => { "category" => "nginx-error" } }
    } else {
      drop { }
    }
  }

  # ---- 4. Parse the Yii structured line ----
  #   2026-07-21 14:10:10 [192.168.1.58][140][-][info][yii\db\Command::query] message
  if [log_format] == "yii" {
    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:log_ts} \[%{DATA:client_ip}\]\[%{DATA:user_id}\]\[%{DATA:session_id}\]\[%{WORD:level}\]\[%{DATA:category}\] %{GREEDYDATA:msg}"
      }
      tag_on_failure => ["_yiigrokfail"]
    }
    if [log_ts] {
      date {
        match    => ["log_ts", "yyyy-MM-dd HH:mm:ss"]
        timezone => "Asia/Muscat"
        target   => "@timestamp"
      }
    }
  }

  # ---- 5. Parse the digitall free-text job line ----
  #   [2026-07-22 12:10:22] Starting Bridge API Push for Survey Score Pk: 80
  else if [log_format] == "digitall-job" {
    grok {
      match => { "message" => "\[%{TIMESTAMP_ISO8601:log_ts}\] %{GREEDYDATA:msg}" }
      tag_on_failure => ["_jobgrokfail"]
    }
    if [log_ts] {
      date {
        match    => ["log_ts", "yyyy-MM-dd HH:mm:ss"]
        timezone => "Asia/Muscat"
        target   => "@timestamp"
      }
    }
    mutate { add_field => { "category" => "integration-job" } }
  }

  # ---- 6. Web-tier container lines (nginx / Kong / php-fpm) ----
  else {
    mutate { add_field => { "msg" => "%{message}" } }
    if [message] =~ /(?i)\b(error|crit|emerg|alert|fatal)\b/ {
      mutate { add_field => { "level" => "error" } }
    } else if [message] =~ /(?i)\bwarn/ {
      mutate { add_field => { "level" => "warning" } }
    } else {
      mutate { add_field => { "level" => "info" } }
    }
    # Don't clobber a category already set in step 3 (access / nginx-error).
    if ![category] { mutate { add_field => { "category" => "web" } } }
  }

  # ---- 7. Normalize / defaults ----
  if ![msg] { mutate { add_field => { "msg" => "%{message}" } } }
  mutate { lowercase => ["level"] }
  if ![level]    or [level]    == "" { mutate { add_field => { "level" => "info" } } }
  if ![category] or [category] == "" { mutate { add_field => { "category" => "application" } } }
  if ![service]  or [service]  == "" { mutate { add_field => { "service" => "unknown" } } }

  # ---- 8. SECRET REDACTION (mandatory) ----
  # The Yii logVars dump leaked S3/DB creds into existing logs. Scrub any that
  # slip through before indexing. (logVars is also being set to [] at the app.)
  # Single-quoted Logstash strings => no config-level escape processing, so the
  # regex + \1\2 backreferences reach the engine verbatim. \x27 is an apostrophe.
  mutate {
    gsub => [
      'msg', '(?i)(S3_SECRET|S3_KEY|AWS_SECRET_ACCESS_KEY|AWS_ACCESS_KEY_ID|DB_PASSWORD|COOKIE_VALIDATION_KEY|cookieValidationKey|password)([">=: \x27\]]{1,6})[^\s,"\x27}\]]+', '\1\2***REDACTED***',
      'msg', 'AKIA[0-9A-Z]{16}', '***REDACTED-AWS-KEY***'
    ]
  }

  # ---- 9. Trim bulky/duplicate metadata ----
  mutate {
    remove_field => ["log_ts", "agent", "ecs", "host", "input", "event",
                     "session_id", "log_format", "@version", "_proj", "_idx",
                     "docker", "stream", "tags"]
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    # Native-realm auth (security is enabled on ES). Password from the container
    # env (set in the root .env, passed through docker-compose.yml).
    user     => "elastic"
    password => "${ELASTIC_PASSWORD}"
    # Daily index; the b2g-logs template attaches the ILM policy that deletes
    # old indices and the SLM policy snapshots them to S3 first.
    index       => "b2g-logs-%{+YYYY.MM.dd}"
    ilm_enabled => false
    # Logstash installs the index template itself at pipeline start, so the
    # keyword facet mappings exist BEFORE the first document is indexed — no
    # dependency on es-init running first. (es-init still owns ILM/S3/Kibana.)
    manage_template    => true
    template_name      => "b2g-logs"
    template           => "/usr/share/logstash/templates/b2g-logs.json"
    template_overwrite => true
  }
}