#!/bin/bash
# ==============================================================================
# mail-stats - Postfix Mail Summary Tool (v5.0.2)
#
# Requires: bash >= 4.2, GNU coreutils/findutils, util-linux (flock),
#           gawk (recommended) or mawk 1.3.4 (verified by a self-test at start)
# Optional: gzip/xz/bzip2/zstd (rotated logs), journalctl, less
#
# 5.0.2: security fixes
#  - arithmetic injection via MAIL_STATS_* (command execution) fixed
#  - env hardening (TMPDIR/XDG_RUNTIME_DIR ignored as root, LESS*/GZIP/XZ/
#    ZSTD/AWK env cleared), root lock in private /run/mail-stats (0700)
#  - less runs with a clean environment (no history file, secure mode)
#  - filter length limit, regex validation under resource limits
#  - watch mode refuses symlinked log files, control chars stripped from
#    all parsed fields and error messages
# 5.0.1: Subject parsing fixed for locally submitted mail ("from local;")
#
# Changes vs 4.0.9:
#  - long queue IDs (enable_long_queue_ids=yes) supported
#  - date/time filters applied on output, correlation state is kept across
#    midnight and log rotation (files processed in mtime order)
#  - TSV internal format (no more "_" <-> " " round-trip)
#  - numeric --min-size comparison, applies to rejects as well
#  - content filter hops (127.0.0.1/::1 :10xxx) skipped precisely, LMTP kept,
#    "queued as" / amavis queued_as linking (client IP, received time, subject)
#  - separate CLIENT_IP and RELAY columns, IPv6 support
#  - filters passed via ENVIRON (awk -v mangled backslashes in -R regexes)
#  - locale-independent (LC_ALL=C), syslog / RFC3339 / journalctl short-iso
#  - no regex intervals (mawk-safe), parser self-test on startup
#  - no silent aborts: SIGPIPE-safe pipelines, parser exit codes reported,
#    resource limits applied to the parser only and reported when hit
#  - summary: final status per recipient, size counted once per queue ID
#  - LESSSECURE pager, lock in /run/lock or XDG_RUNTIME_DIR
# ==============================================================================

set -euo pipefail
IFS=$'\n\t'
export LC_ALL=C
export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
umask 077

# Environment hardening: variables that change behaviour of helper tools
unset -v AWKPATH AWKLIBPATH POSIXLY_CORRECT GAWK_READ_TIMEOUT GAWK_SOCK_RETRIES \
         GZIP XZ_OPT XZ_DEFAULTS ZSTD_CLEVEL ZSTD_NBTHREADS \
         LESS LESSOPEN LESSCLOSE LESSKEY LESSKEYIN LESSKEY_SYSTEM LESSHISTFILE \
         LESSEDIT LESSSECURE_ALLOW EDITOR VISUAL MS_RE
# As root never trust user-controlled directories (sudo -E / env_keep / SETENV)
if (( EUID == 0 )); then
    unset -v TMPDIR XDG_RUNTIME_DIR
fi

readonly VERSION="5.0.2"

die()  { local m="$*"; printf 'Error: %s\n' "${m//[[:cntrl:]]/?}" >&2; exit 1; }
warn() { local m="$*"; printf 'Warning: %s\n' "${m//[[:cntrl:]]/?}" >&2; }

# ==============================================================================
# 1. HELP
# ==============================================================================
show_help() {
    cat << EOF
mail-stats v${VERSION} - Postfix mail log summary

Usage: mail-stats [OPTIONS]

FILTERING OPTIONS (filters are combined with AND):
  -d, --date DATE       Day of the delivery event. Default: today.
                        Formats: "Sep 4" | "Sep 04" | "2026-09-04"
                        Special: -d "" searches ALL available dates.
  --time TIME           Time prefix of the delivery event.
                        Examples: --time "14:30" (minute) | --time "14:" (hour)
  -f, --from STRING     Sender address or domain (null sender is shown as <>).
  -t, --to STRING       Recipient address or domain.
  -a, --ip STRING       Client IP or relay (host/IP), full or partial.
  -s, --status STATUS   sent | deferred | bounced | expired | reject | greylist
  -i, --id QUEUE_ID     Queue ID (short or long). Matches also the original
                        queue ID before content filter re-injection.
                        Enables Detail View. -i NOQUEUE lists pre-queue rejects.
  --min-size BYTES      Only messages with size >= BYTES.
  -R, --regex           Treat -f, -t, -a, -s, -i as ERE (case-insensitive).

OUTPUT & RUN MODE OPTIONS:
  -l, --list NUMBER     Number of latest records (1-${MAX_LIMIT}). Default: 10.
  -S, --summary         Aggregate statistics instead of a list.
  -c, --csv             CSV output (formula-injection safe).
  -w, --watch           Live mode (tail -F / journalctl -f). Date filter and
                        -l are ignored.

MISC OPTIONS:
  --log-dir PATH        /var/log (default) or /var/log/mail.
  -h, --help            This help.

NOTES:
  - Date/time filters match the delivery (status=) line, not the receive time.
  - Subject is available only with header_checks WARN/INFO on ^Subject:.
  - List/detail views show every delivery attempt (deferred retries included);
    summary counts the final status per queue ID + recipient.

ENVIRONMENT (resource limits for the parser):
  MAIL_STATS_MEM_KB    (default 1048576)  ulimit -v
  MAIL_STATS_CPU_SEC   (default 300)      ulimit -t (not in watch mode)
  MAIL_STATS_FSIZE_KB  (default 1048576)  ulimit -f (not in watch mode)
EOF
    exit 0
}

# ==============================================================================
# 2. DEPENDENCY CHECK
# ==============================================================================
for cmd in cat cut date env find flock head mktemp sort tail uniq wc; do
    command -v "$cmd" > /dev/null 2>&1 || die "Required command '$cmd' is not installed."
done

if command -v gawk > /dev/null 2>&1; then
    AWK="gawk"; IS_GAWK=1
elif command -v awk > /dev/null 2>&1; then
    AWK="awk"; IS_GAWK=0
else
    die "awk is not installed (gawk recommended)."
fi
readonly AWK IS_GAWK

# ==============================================================================
# 3. ARGUMENT PARSING & VALIDATION
# ==============================================================================
readonly MAX_LIMIT=1000
LIMIT=10
FROM_FILTER=""; TO_FILTER=""; ID_FILTER=""; IP_FILTER=""; STATUS_FILTER=""
TIME_FILTER=""; DATE_FILTER=""; LOG_DIR="/var/log"; MIN_SIZE=0
DATE_SET=0; CSV_MODE=0; SUMMARY_MODE=0; WATCH_MODE=0; REGEX_MODE=0

need_arg()  { (( $2 >= 2 )) || die "Option '$1' requires an argument."; }
need_uint() { [[ $2 =~ ^[0-9]{1,18}$ ]] || die "Option '$1' requires a non-negative integer, got '$2'."; }

while (( $# > 0 )); do
    case $1 in
        -h|--help)    show_help ;;
        -l|--list)    need_arg "$1" $#; need_uint "$1" "$2"; LIMIT=$((10#$2)); shift ;;
        -f|--from)    need_arg "$1" $#; FROM_FILTER=$2; shift ;;
        -t|--to)      need_arg "$1" $#; TO_FILTER=$2; shift ;;
        -i|--id)      need_arg "$1" $#; ID_FILTER=$2; shift ;;
        -a|--ip)      need_arg "$1" $#; IP_FILTER=$2; shift ;;
        -s|--status)  need_arg "$1" $#; STATUS_FILTER=$2; shift ;;
        --time)       need_arg "$1" $#; TIME_FILTER=$2; shift ;;
        --log-dir)    need_arg "$1" $#; LOG_DIR=$2; shift ;;
        --min-size)   need_arg "$1" $#; need_uint "$1" "$2"; MIN_SIZE=$((10#$2)); shift ;;
        -d|--date)    need_arg "$1" $#; DATE_FILTER=$2; DATE_SET=1; shift ;;
        -S|--summary) SUMMARY_MODE=1 ;;
        -c|--csv)     CSV_MODE=1 ;;
        -w|--watch)   WATCH_MODE=1 ;;
        -R|--regex)   REGEX_MODE=1 ;;
        *)            die "Unknown option: $1 (see --help)" ;;
    esac
    shift
done

(( LIMIT < 1 )) && LIMIT=1
(( LIMIT > MAX_LIMIT )) && LIMIT=$MAX_LIMIT

if (( WATCH_MODE && SUMMARY_MODE )); then die "--summary cannot be combined with --watch."; fi
if (( CSV_MODE && SUMMARY_MODE )); then die "--summary cannot be combined with --csv."; fi

LOG_DIR=${LOG_DIR%/}
if [[ $LOG_DIR != "/var/log" && $LOG_DIR != "/var/log/mail" ]]; then
    die "Unsafe log directory specified ($LOG_DIR). Only /var/log or /var/log/mail are allowed."
fi
[[ -d $LOG_DIR ]] || die "Directory $LOG_DIR does not exist."

# Date -> YYYY-MM-DD (syslog dates without year: a future date means last year)
DAY_ISO=""
TODAY=$(date +%F)
if (( DATE_SET == 0 )); then
    DAY_ISO=$TODAY
elif [[ -n $DATE_FILTER ]]; then
    if [[ $DATE_FILTER =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then
        DAY_ISO=$(date -d "$DATE_FILTER" +%F 2> /dev/null) || die "Invalid date '$DATE_FILTER'."
    elif [[ $DATE_FILTER =~ ^([A-Za-z]{3})[[:space:]]+0?([0-9]{1,2})$ ]]; then
        DAY_ISO=$(date -d "${BASH_REMATCH[1]} ${BASH_REMATCH[2]}" +%F 2> /dev/null) || die "Invalid date '$DATE_FILTER'."
        if [[ $DAY_ISO > $TODAY ]]; then
            DAY_ISO=$(date -d "$DAY_ISO -1 year" +%F)
        fi
    else
        die "Invalid date format '$DATE_FILTER'. Use \"MMM D\" or \"YYYY-MM-DD\"."
    fi
fi
DAY_YMD=${DAY_ISO//-/}
readonly DAY_ISO DAY_YMD TODAY

if [[ -n $TIME_FILTER && ! $TIME_FILTER =~ ^([01][0-9]|2[0-3]):([0-5]([0-9](:([0-5][0-9]?)?)?)?)?$ ]]; then
    die "Invalid --time '$TIME_FILTER'. Use \"HH:\", \"HH:MM\" or \"HH:MM:SS\"."
fi

# Resource limits - validated as strings BEFORE any arithmetic use:
# bash (( )) evaluates variable contents, e.g. 'x[$(cmd)]' would execute cmd
: "${MAIL_STATS_MEM_KB:=1048576}" "${MAIL_STATS_CPU_SEC:=300}" "${MAIL_STATS_FSIZE_KB:=1048576}"
for lim in MAIL_STATS_MEM_KB MAIL_STATS_CPU_SEC MAIL_STATS_FSIZE_KB; do
    [[ ${!lim} =~ ^[1-9][0-9]{0,11}$ ]] || die "$lim must be a positive integer."
done
readonly MAIL_STATS_MEM_KB MAIL_STATS_CPU_SEC MAIL_STATS_FSIZE_KB

# Run a command with resource limits (parser/sort only, not journalctl/gzip)
run_limited() {
    (
        ulimit -v "$MAIL_STATS_MEM_KB"
        if (( ! WATCH_MODE )); then
            ulimit -t "$MAIL_STATS_CPU_SEC"
            ulimit -f "$MAIL_STATS_FSIZE_KB"
        fi
        exec "$@"
    )
}

readonly MAX_FILTER_LEN=256
for flt in "$FROM_FILTER" "$TO_FILTER" "$IP_FILTER" "$STATUS_FILTER" "$ID_FILTER"; do
    (( ${#flt} <= MAX_FILTER_LEN )) || die "Filter too long (max $MAX_FILTER_LEN characters)."
done

if (( REGEX_MODE )); then
    for re in "$FROM_FILTER" "$TO_FILTER" "$IP_FILTER" "$STATUS_FILTER" "$ID_FILTER"; do
        [[ -z $re ]] && continue
        MS_RE=$re WATCH_MODE=0 run_limited "$AWK" 'BEGIN { if ("" ~ ENVIRON["MS_RE"]) {} }' 2> /dev/null \
            || die "Invalid regular expression: $re"
    done
fi

DETAIL_VIEW=0
if [[ -n $ID_FILTER && $ID_FILTER != "NOQUEUE" ]] && (( ! CSV_MODE )); then
    DETAIL_VIEW=1
fi

USE_COLOR=0
if [[ -t 1 && -z ${NO_COLOR:-} ]] && (( ! CSV_MODE )); then
    USE_COLOR=1
fi
if (( USE_COLOR )); then C_Y=$'\033[1;33m'; C_H=$'\033[1;36m'; C_R=$'\033[0m'; else C_Y=""; C_H=""; C_R=""; fi


# ==============================================================================
# 4. LOCKING & TEMP FILES
# ==============================================================================
if (( EUID == 0 )); then
    # private root-only directory: no symlink/pre-created lock file tricks
    LOCK_DIR=/run/mail-stats
    [[ -L $LOCK_DIR ]] && die "$LOCK_DIR is a symlink, refusing to use it."
    if [[ ! -d $LOCK_DIR ]]; then
        mkdir -m 0700 -- "$LOCK_DIR" || die "Cannot create $LOCK_DIR."
    fi
    [[ -O $LOCK_DIR && ! -L $LOCK_DIR ]] || die "$LOCK_DIR is not owned by root, refusing to use it."
    chmod 0700 -- "$LOCK_DIR"
    LOCK_UID=${SUDO_UID:-0}
    [[ $LOCK_UID =~ ^[0-9]+$ ]] || LOCK_UID=0
else
    LOCK_DIR=${XDG_RUNTIME_DIR:-${TMPDIR:-/tmp}}
    LOCK_UID=$UID
fi

if (( WATCH_MODE )); then
    exec 9>> "$LOCK_DIR/mail-stats-watch-${LOCK_UID}.lock"
    flock -n 9 || die "Another watch instance is already running."
else
    exec 9>> "$LOCK_DIR/mail-stats-query-${LOCK_UID}.lock"
    flock -w 60 9 || die "Timed out waiting for another running query (lock)."

    TMP_DIR=$(mktemp -d "${TMPDIR:-/tmp}/mail-stats.XXXXXX")
    readonly TMP_DIR
    cleanup() { [[ -n ${TMP_DIR:-} && -d $TMP_DIR ]] && rm -rf -- "$TMP_DIR"; }
    trap cleanup EXIT
    readonly RAW="$TMP_DIR/raw.tsv" SORTED="$TMP_DIR/sorted.tsv" LATEST="$TMP_DIR/latest.tsv"
fi
trap 'exit 130' INT
trap 'exit 143' TERM


# ==============================================================================
# 5. LOG DISCOVERY & STREAMING
# ==============================================================================
CUR_Y=$(date +%Y); CUR_M=$(date +%m)
readonly CUR_Y CUR_M
declare -A MON_NUM=([Jan]=01 [Feb]=02 [Mar]=03 [Apr]=04 [May]=05 [Jun]=06
                    [Jul]=07 [Aug]=08 [Sep]=09 [Oct]=10 [Nov]=11 [Dec]=12)

log_cat() {
    case $1 in
        *.gz)  gzip -dc -- "$1" ;;
        *.xz)  xz -dc -- "$1" ;;
        *.bz2) bzip2 -dc -- "$1" ;;
        *.zst) zstd -dcq -- "$1" ;;
        *)     cat -- "$1" ;;
    esac
}

first_line() { { log_cat "$1" 2> /dev/null || true; } | head -n 1; }

# Prints YYYYMMDD of a log line (syslog or RFC3339), nothing if unknown
line_ymd() {
    local l=$1 y m
    if [[ $l =~ ^([0-9]{4})-([0-9]{2})-([0-9]{2})T ]]; then
        printf '%s%s%s' "${BASH_REMATCH[1]}" "${BASH_REMATCH[2]}" "${BASH_REMATCH[3]}"
    elif [[ $l =~ ^([A-Z][a-z]{2})[[:space:]]+([0-9]{1,2})[[:space:]] ]] && [[ -n ${MON_NUM[${BASH_REMATCH[1]}]:-} ]]; then
        m=${MON_NUM[${BASH_REMATCH[1]}]}
        y=$CUR_Y
        if (( 10#$m > 10#$CUR_M )); then y=$((CUR_Y - 1)); fi
        printf '%s%s%02d' "$y" "$m" "$((10#${BASH_REMATCH[2]}))"
    fi
}

LOG_FILES=(); LOG_MTIME_YMD=(); SELECTED=()

# Chronological order by mtime (gzip/logrotate preserve it)
discover_logs() {
    local ymd path
    while IFS=$'\t' read -r -d '' _ ymd path; do
        LOG_FILES+=("$path")
        LOG_MTIME_YMD+=("$ymd")
    done < <(find "$LOG_DIR" -maxdepth 1 -type f \( -name 'maillog*' -o -name 'mail.log*' \) \
                  -printf '%T@\t%TY%Tm%Td\t%p\0' 2> /dev/null | sort -z -n)
}

# A file covers <first line date, mtime date>. The preceding file is included
# as well so that messages received before rotation keep sender/IP/size.
select_logs() {
    local i start n=${#LOG_FILES[@]}
    local -a rel=()
    for (( i = 0; i <= n; i++ )); do rel[i]=0; done
    for (( i = 0; i < n; i++ )); do
        if [[ -z $DAY_YMD ]]; then rel[i]=1; continue; fi
        if [[ ${LOG_MTIME_YMD[i]} < $DAY_YMD ]]; then continue; fi
        start=$(line_ymd "$(first_line "${LOG_FILES[i]}")")
        if [[ -z $start ]] || ! [[ $start > $DAY_YMD ]]; then rel[i]=1; fi
    done
    for (( i = 0; i < n; i++ )); do
        if (( rel[i] || rel[i + 1] )); then SELECTED+=("${LOG_FILES[i]}"); fi
    done
}

stream_logs() {
    local f
    local -a jargs
    if (( ${#LOG_FILES[@]} == 0 )); then
        jargs=(SYSLOG_FACILITY=2 -o short-iso --no-pager -q)
        if [[ -n $DAY_ISO ]]; then
            jargs+=(--since "$(date -d "$DAY_ISO -1 day" +%F) 00:00:00"
                    --until "$(date -d "$DAY_ISO +1 day" +%F) 00:00:00")
        fi
        journalctl "${jargs[@]}" || warn "journalctl exited with an error; results may be incomplete."
        return 0
    fi
    (( ${#SELECTED[@]} == 0 )) && return 0
    for f in "${SELECTED[@]}"; do
        if [[ ! -r $f ]]; then warn "Cannot read $f, skipping."; continue; fi
        log_cat "$f" 2> /dev/null \
            || warn "Error reading $f (corrupted/truncated or missing decompressor); results may be incomplete."
    done
    return 0
}

stream_watch() {
    local active=""
    if [[ -f $LOG_DIR/maillog && ! -L $LOG_DIR/maillog ]]; then active=$LOG_DIR/maillog
    elif [[ -f $LOG_DIR/mail.log && ! -L $LOG_DIR/mail.log ]]; then active=$LOG_DIR/mail.log
    fi
    if [[ -n $active ]]; then
        printf 'Starting Live Watch Mode on: %s (Ctrl+C to stop)...\n' "$active" >&2
        tail -n 100 -F -- "$active"
    elif command -v journalctl > /dev/null 2>&1; then
        printf 'Starting Live Watch Mode via systemd journal (Ctrl+C to stop)...\n' >&2
        journalctl SYSLOG_FACILITY=2 -o short-iso -n 100 -f
    else
        die "No active log found to watch and journalctl is unavailable."
    fi
}

# ==============================================================================
# 6. PARSER (log -> TSV)
# ------------------------------------------------------------------------------
# TSV columns:
#   1 sort key (YYYYMMDDHHMMSS)  2 event time  3 received time  4 queue id
#   5 original queue id  6 client ip  7 relay  8 sender  9 recipient
#   10 status  11 size  12 dsn  13 subject  14 reason
# ==============================================================================
# shellcheck disable=SC2016
AWK_PARSE='
BEGIN {
    OFS = "\t"
    d_ymd   = ENVIRON["MS_DAY_YMD"]
    tm_filt = ENVIRON["MS_TIME"]
    f_filt  = ENVIRON["MS_FROM"];   lf_filt  = tolower(f_filt)
    t_filt  = ENVIRON["MS_TO"];     lt_filt  = tolower(t_filt)
    i_filt  = ENVIRON["MS_ID"];     li_filt  = tolower(i_filt)
    ip_filt = ENVIRON["MS_IP"];     lip_filt = tolower(ip_filt)
    st_filt = ENVIRON["MS_STATUS"]; lst_filt = tolower(st_filt)
    m_sz    = ENVIRON["MS_MIN_SIZE"] + 0
    r_mode  = ENVIRON["MS_REGEX"] + 0
    w_mode  = ENVIRON["MS_WATCH"] + 0
    is_gawk = ENVIRON["MS_GAWK"] + 0
    cur_y   = ENVIRON["MS_CUR_Y"] + 0
    cur_m   = ENVIRON["MS_CUR_M"] + 0
    n = split("Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec", mn, " ")
    for (i = 1; i <= n; i++) mon[mn[i]] = sprintf("%02d", i)
    # short (hex) or long (base52 without vowels) queue id
    # (no {n,m} intervals: broken/unsupported in some mawk builds)
    hx = "[0-9A-F]"; lg = "[0-9B-DF-HJ-NP-TV-Zb-df-hj-np-tv-z]"
    QID_RE = "^(" hx hx hx hx hx hx hx "*|" lg lg lg lg lg lg lg lg lg lg lg "*):$"
    MAX_STATE = 200000
    nstate = 0; npend = 0; warned = 0
}

function chk(val, filt, lfilt,   r) {
    if (filt == "") return 1
    if (r_mode) {
        if (is_gawk) { IGNORECASE = 1; r = (val ~ filt); IGNORECASE = 0; return r }
        return (tolower(val) ~ lfilt)
    }
    return (index(tolower(val), lfilt) > 0)
}

function clean(s, max) {
    gsub(/[[:cntrl:]]/, " ", s)
    if (max > 0 && length(s) > max) s = substr(s, 1, max) "..."
    if (s == "") s = "-"
    return s
}

function is_loop(ip) { return (ip ~ /^127\./ || ip == "::1" || ip == "ipv6:::1") }

function join(f, n,   i, s) { s = f[1]; for (i = 2; i <= n; i++) s = s OFS f[i]; return s }

function parse_ts() {
    if ($1 in mon) {
        if (NF < 6 || $2 !~ /^[0-9]+$/ || $3 !~ /^[0-9][0-9]:[0-9][0-9]:[0-9][0-9]$/) return 0
        l_mm = mon[$1]; l_dd = sprintf("%02d", $2); l_time = $3
        l_y = (l_mm + 0 > cur_m) ? cur_y - 1 : cur_y
        return 1
    }
    if ($1 ~ /^[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9]/) {
        l_y = substr($1, 1, 4); l_mm = substr($1, 6, 2); l_dd = substr($1, 9, 2); l_time = substr($1, 12, 8)
        return 1
    }
    return 0
}

function pass(f) {
    if (!w_mode && d_ymd != "" && substr(f[1], 1, 8) != d_ymd) return 0
    if (tm_filt != "" && index(substr(f[2], 12), tm_filt) != 1) return 0
    if (f[11] + 0 < m_sz) return 0
    if (!chk(f[8], f_filt, lf_filt) || !chk(f[9], t_filt, lt_filt) || !chk(f[10], st_filt, lst_filt)) return 0
    if (ip_filt != "" && !chk(f[6], ip_filt, lip_filt) && !chk(f[7], ip_filt, lip_filt)) return 0
    if (i_filt != "" && !chk(f[4], i_filt, li_filt) && !chk(f[5], i_filt, li_filt)) return 0
    return 1
}

function out(rec,   f) {
    split(rec, f, "\t")
    if (!pass(f)) return
    print rec
    if (w_mode) fflush()
}

function track(q) {
    if (q in seen) return
    if (nstate >= MAX_STATE) reset_state()
    seen[q] = 1; nstate++
}

function forget(q) {
    if (q in seen) { delete seen[q]; nstate-- }
    delete c_ip[q]; delete c_time[q]; delete snd[q]; delete subj[q]; delete sz[q]; delete orig[q]
}

function flush_pend(q, patch,   i, f, n, rec) {
    for (i = 1; i <= pn[q]; i++) {
        rec = pend[q, i]; delete pend[q, i]; npend--
        if (patch) {
            n = split(rec, f, "\t")
            if (q in c_time) f[3] = c_time[q]
            if (q in orig)   f[5] = orig[q]
            if (q in c_ip)   f[6] = c_ip[q]
            if (f[13] == "-" && (q in subj)) f[13] = subj[q]
            rec = join(f, n)
        }
        out(rec)
    }
    delete pn[q]
}

function flush_all_pend(   q, ks, k, i) {
    k = 0
    for (q in pn) ks[++k] = q
    for (i = 1; i <= k; i++) flush_pend(ks[i], 0)
    npend = 0
}

function reset_state() {
    flush_all_pend()
    split("", seen); split("", c_ip); split("", c_time); split("", snd)
    split("", subj); split("", sz); split("", orig)
    nstate = 0
    if (!warned++) print "Warning: correlation state limit (" MAX_STATE " queue IDs) reached and reset; some records may lack sender/IP/size." > "/dev/stderr"
}

# Content filter re-injection: carry origin data from src to dst queue id
function link(src, dst) {
    if (src == "" || dst == "" || src == dst) return
    track(dst)
    if (!(dst in orig)) orig[dst] = (src in orig) ? orig[src] : src
    if ((src in c_ip) && (!(dst in c_ip) || is_loop(c_ip[dst]))) c_ip[dst] = c_ip[src]
    if (src in c_time) c_time[dst] = c_time[src]
    if ((src in snd)  && !(dst in snd))  snd[dst]  = snd[src]
    if ((src in subj) && !(dst in subj)) subj[dst] = subj[src]
    if ((src in sz)   && !(dst in sz))   sz[dst]   = sz[src]
    if (dst in pn) flush_pend(dst, 1)
}

function emit(q, to, relay, st, dsn, reason,   rec) {
    rec = l_key OFS l_ts OFS ((q in c_time) ? c_time[q] : l_ts) OFS q
    rec = rec OFS ((q in orig) ? orig[q] : "-") OFS ((q in c_ip) ? c_ip[q] : "local") OFS relay
    rec = rec OFS ((q in snd) ? snd[q] : "-") OFS to OFS st OFS ((q in sz) ? sz[q] : 0)
    rec = rec OFS dsn OFS ((q in subj) ? subj[q] : "-") OFS reason
    # Batch mode: loopback client without a link yet may be a re-injected
    # message whose "queued as" line is logged later -> hold until linked/END
    if (!w_mode && !(q in orig) && (q in c_ip) && is_loop(c_ip[q])) {
        pend[q, ++pn[q]] = rec
        if (++npend > MAX_STATE) flush_all_pend()
        return
    }
    out(rec)
}

function handle_reject(q,   p, r, f, t, ip, ss, re, dsn, st, rec) {
    p = index($0, "reject: ")
    r = substr($0, p + 8)
    ip = (q != "NOQUEUE" && (q in c_ip)) ? c_ip[q] : "unknown"
    if (match(r, /^[A-Z-]+ from [^ ]*\[[^]]*\]: /)) {
        ss = substr(r, 1, RLENGTH); r = substr(r, RLENGTH + 1)
        sub(/\]: $/, "", ss); sub(/^.*\[/, "", ss); ip = clean(ss, 64)
    }
    f = "-"
    if (match($0, /from=<[^>]*>/)) { f = substr($0, RSTART + 6, RLENGTH - 7); f = (f == "") ? "<>" : clean(f, 0) }
    t = "-"
    if (match($0, / to=<[^>]*>/)) t = clean(substr($0, RSTART + 5, RLENGTH - 6), 0)
    p = index(r, "; from=<"); if (p > 0) r = substr(r, 1, p - 1)
    re = clean(r, 512)
    dsn = "-"; if (match(re, /[245]\.[0-9]+\.[0-9]+/)) dsn = substr(re, RSTART, RLENGTH)
    st = (re ~ /[Gg]reylist/) ? "GREYLIST" : "REJECT"
    rec = l_key OFS l_ts OFS ((q in c_time) ? c_time[q] : l_ts) OFS q OFS ((q in orig) ? orig[q] : "-")
    rec = rec OFS ip OFS "-" OFS f OFS t OFS st OFS ((q in sz) ? sz[q] : 0)
    rec = rec OFS dsn OFS ((q in subj) ? subj[q] : "-") OFS re
    out(rec)
}

function amavis(   q1, q2, ip) {
    if (!match($0, /Queue-ID: [0-9A-Za-z]+/)) return
    q1 = substr($0, RSTART + 10, RLENGTH - 10)
    if (!match($0, /queued_as: [0-9A-Za-z]+/)) return
    q2 = substr($0, RSTART + 11, RLENGTH - 11)
    link(q1, q2)
    if (match($0, /\[[0-9A-Fa-f:.]+\]/)) {
        ip = substr($0, RSTART + 1, RLENGTH - 2)
        if (!is_loop(ip) && (!(q2 in c_ip) || is_loop(c_ip[q2]))) {
            c_ip[q2] = ip
            if (q2 in pn) flush_pend(q2, 1)
        }
    }
}

{
    if (length($0) < 20 || !parse_ts()) next
    tt = l_time; gsub(/:/, "", tt)
    l_key = l_y l_mm l_dd tt
    l_ts  = l_y "-" l_mm "-" l_dd " " l_time

    if (index($0, "NOQUEUE: ") && $0 ~ /NOQUEUE: (milter-)?reject: /) { handle_reject("NOQUEUE"); next }

    if (index($0, "Queue-ID: ") && index($0, "queued_as: ") && $0 ~ / amavisd?\[[0-9]+\]: /) { amavis(); next }

    qid = ""
    for (i = 2; i <= 8 && i <= NF; i++) if ($i ~ QID_RE) { qid = substr($i, 1, length($i) - 1); break }
    if (qid == "") next
    track(qid)

    if (index($0, qid ": client=")) {
        if (match($0, /client=[^ ,]*\[[^]]*\]/)) {
            ip = substr($0, RSTART, RLENGTH); sub(/^[^[]*\[/, "", ip); sub(/\]$/, "", ip)
            if (!(qid in c_ip) || is_loop(c_ip[qid])) c_ip[qid] = clean(ip, 64)
        }
        if (!(qid in c_time)) c_time[qid] = l_ts
        next
    }

    if (index($0, qid ": removed")) { forget(qid); next }

    if (index($0, qid ": from=<")) {
        if (match($0, /from=<[^>]*>/)) { s = substr($0, RSTART + 6, RLENGTH - 7); snd[qid] = (s == "") ? "<>" : clean(s, 0) }
        if (match($0, /size=[0-9]+/)) sz[qid] = substr($0, RSTART + 5, RLENGTH - 5) + 0
        next
    }

    if (index($0, " header Subject: ")) {
        s = substr($0, index($0, " header Subject: ") + 17)
        # "from host[ip]; from=<..." (smtpd) or "from local; from=<..." (pickup)
        sub(/ from [^ ;]*; from=<.*$/, "", s)
        if (!(qid in subj)) subj[qid] = clean(s, 256)
    }

    if (index($0, qid ": reject: ") || index($0, qid ": milter-reject: ")) { handle_reject(qid); next }

    if (index($0, " to=<") && index($0, ", status=")) {
        # content filter hop (amavis & co.) - link to re-injected queue id, do not report
        if ($0 ~ /relay=[^ ,]*\[(127\.0\.0\.1|(ipv6:)?::1)\]:10[0-9][0-9][0-9],/) {
            if (match($0, /queued as [0-9A-Za-z]+/)) link(qid, substr($0, RSTART + 10, RLENGTH - 10))
            next
        }
        to = "-";     if (match($0, / to=<[^>]*>/)) to = clean(substr($0, RSTART + 5, RLENGTH - 6), 0)
        relay = "-";  if (match($0, /relay=[^ ,]+/)) { relay = substr($0, RSTART + 6, RLENGTH - 6); sub(/:[0-9]+$/, "", relay); relay = clean(relay, 255) }
        st = "-";     if (match($0, /status=[a-z]+/)) st = substr($0, RSTART + 7, RLENGTH - 7)
        dsn = "-";    if (match($0, /dsn=[0-9.]+/)) dsn = substr($0, RSTART + 4, RLENGTH - 4)
        reason = "-"
        if (match($0, /status=[a-z]+ \(.*\)/)) {
            reason = substr($0, RSTART, RLENGTH)
            sub(/^status=[a-z]+ \(/, "", reason); sub(/\)$/, "", reason)
            reason = clean(reason, 512)
        }
        emit(qid, to, relay, st, dsn, reason)
    }
}

END { flush_all_pend() }
'

# ==============================================================================
# 7. OUTPUT RENDERING (TSV -> list / detail / CSV)
# ==============================================================================
# shellcheck disable=SC2016
AWK_FORMAT='
BEGIN {
    FS = "\t"
    u_col = ENVIRON["MS_COLOR"] + 0; csv = ENVIRON["MS_CSV"] + 0; d_view = ENVIRON["MS_DETAIL"] + 0
    if (u_col) { B = "\033[1;34m"; R = "\033[0m"; G = "\033[1;32m"; RD = "\033[1;31m"; Y = "\033[1;33m"; GR = "\033[0;90m"; C = "\033[0;36m"; BO = "\033[1m" }
    if (csv) print "DATE_TIME,RECEIVED,QUEUE_ID,ORIG_QUEUE_ID,CLIENT_IP,RELAY,SENDER,RECIPIENT,STATUS,SIZE_BYTES,DSN,SUBJECT,REASON"
    else if (!d_view && u_col) printf "%s%-14s %-15s %-15s %-30s %-28s %-9s %s%s\n", BO, "TIME", "QUEUE_ID", "CLIENT_IP", "SENDER", "RECIPIENT", "STATUS", "DSN/REASON", R
    fflush()
}
function csv_q(s) { if (s ~ /^[=+@\-]/ && s != "-") s = "\047" s; gsub(/"/, "\"\"", s); return "\"" s "\"" }
function cut(s, m) { if (length(s) <= m) return s; return ".." substr(s, length(s) - m + 3) }
NF >= 14 {
    if (csv) {
        line = csv_q($2)
        for (i = 3; i <= 14; i++) line = line "," csv_q($i)
        print line; fflush(); next
    }
    scol = ($10 == "sent") ? G : (($10 == "deferred" || $10 == "GREYLIST") ? Y : RD)
    if (d_view) {
        printf "\n%s--- Queue ID: %s%s ---%s\n", B, $4, ($5 != "-" ? " (original: " $5 ")" : ""), R
        printf "Time:       %s\nReceived:   %s\nClient IP:  %s\nRelay:      %s\nFrom:       %s\nTo:         %s\n", $2, $3, $6, $7, $8, $9
        printf "Status:     %s%s%s (dsn %s)\nSize:       %s bytes\nSubject:    %s\nReason:     %s\n", scol, $10, R, $12, $11, $13, $14
    } else {
        sr = ($12 != "-") ? $12 : substr($14, 1, 20)
        printf "%-14s %s%-15s%s %s%-15s%s %-30s %-28s %s%-9s%s %s[%s]%s\n", substr($2, 6), ($4 == "NOQUEUE" ? RD : B), $4, R, GR, cut($6, 15), R, cut($8, 30), cut($9, 28), scol, $10, R, ($10 == "sent" ? C : scol), sr, R
    }
    fflush()
}
'

# Parser self-test (catches incompatible awk implementations early)
selftest_out=$(printf '%s\n' \
    'Sep  1 10:00:00 mx postfix/smtpd[1]: 4T3bV22xLnz1t: client=a.example[192.0.2.1]' \
    'Sep  1 10:00:01 mx postfix/qmgr[2]: 4T3bV22xLnz1t: from=<a@example.org>, size=1234, nrcpt=1 (queue active)' \
    'Sep  1 10:00:02 mx postfix/smtp[3]: 4T3bV22xLnz1t: to=<b@example.net>, relay=mx.example.net[192.0.2.2]:25, delay=1, dsn=2.0.0, status=sent (250 2.0.0 Ok)' \
    | MS_CUR_Y=2026 MS_CUR_M=9 "$AWK" "$AWK_PARSE" 2>&1) || true
if [[ $selftest_out != *$'\t4T3bV22xLnz1t\t-\t192.0.2.1\tmx.example.net[192.0.2.2]\ta@example.org\tb@example.net\tsent\t1234\t2.0.0\t-\t250 2.0.0 Ok' ]]; then
    die "Parser self-test failed with '$AWK' (incompatible awk implementation?). Install gawk."
fi
unset selftest_out

use_pager() {
    if [[ -t 1 ]] && (( ! CSV_MODE && ! WATCH_MODE )) && command -v less > /dev/null 2>&1; then
        # clean env + LESSSECURE: no !shell, |pipe, :e, v, LESSOPEN, history file
        env -i TERM="${TERM:-dumb}" LESSCHARSET=utf-8 LESSSECURE=1 LESSHISTFILE=- less -R -F -X
    else
        cat
    fi
}

# Exit on real errors, ignore SIGPIPE (e.g. user quit less early)
check_pipe() {
    local s
    for s in "$@"; do
        (( s == 0 || s == 141 )) || exit "$s"
    done
}

export MS_DAY_YMD="$DAY_YMD" MS_TIME="$TIME_FILTER" MS_FROM="$FROM_FILTER" MS_TO="$TO_FILTER" \
       MS_ID="$ID_FILTER" MS_IP="$IP_FILTER" MS_STATUS="$STATUS_FILTER" MS_MIN_SIZE="$MIN_SIZE" \
       MS_REGEX="$REGEX_MODE" MS_WATCH="$WATCH_MODE" MS_GAWK="$IS_GAWK" MS_CUR_Y="$CUR_Y" \
       MS_CUR_M="$((10#$CUR_M))" MS_COLOR="$USE_COLOR" MS_CSV="$CSV_MODE" MS_DETAIL="$DETAIL_VIEW"

# ==============================================================================
# 8. WATCH MODE
# ==============================================================================
if (( WATCH_MODE )); then
    set +e
    stream_watch | run_limited "$AWK" "$AWK_PARSE" | "$AWK" "$AWK_FORMAT"
    ps=("${PIPESTATUS[@]}")
    set -e
    (( ps[1] == 0 || ps[1] == 130 || ps[1] == 141 )) || die "Parser exited with status ${ps[1]} (memory limit MAIL_STATS_MEM_KB=$MAIL_STATS_MEM_KB?)."
    exit 0
fi

# ==============================================================================
# 9. BATCH PROCESSING
# ==============================================================================
discover_logs
if (( ${#LOG_FILES[@]} == 0 )); then
    command -v journalctl > /dev/null 2>&1 || die "No Postfix logs found in $LOG_DIR and journalctl is unavailable."
else
    select_logs
fi

set +e
stream_logs | run_limited "$AWK" "$AWK_PARSE" > "$RAW"
ps=("${PIPESTATUS[@]}")
set -e

case ${ps[1]} in
    0)   ;;
    152) die "Parser killed: CPU time limit exceeded (MAIL_STATS_CPU_SEC=$MAIL_STATS_CPU_SEC). Narrow the date range or raise the limit." ;;
    153) die "Parser killed: output size limit exceeded (MAIL_STATS_FSIZE_KB=$MAIL_STATS_FSIZE_KB)." ;;
    *)   die "Parser failed with status ${ps[1]} (memory limit MAIL_STATS_MEM_KB=$MAIL_STATS_MEM_KB?)." ;;
esac
(( ps[0] == 0 )) || warn "Log reader exited with status ${ps[0]}; results may be incomplete."

run_limited sort -s -t $'\t' -k1,1n -S 64M -- "$RAW" > "$SORTED"

# ---------------------------------------------------------------- summary ----
render_summary() (
    set -e
    total_events=$(wc -l < "$SORTED")
    total_final=$(wc -l < "$LATEST")

    top_n() {
        printf '\n%s%s%s\n' "$C_Y" "$1" "$C_R"
        "$AWK" -F'\t' "$3" "$LATEST" | sort | uniq -c | sort -k1,1nr \
            | "$AWK" -v m="$2" 'NR <= m { n = $1; sub(/^ *[0-9]+ /, ""); printf "  %7d  %s\n", n, $0 }'
    }

    printf '%s==========================================================%s\n' "$C_H" "$C_R"
    printf '%s SUMMARY: %s events, %s final per recipient%s\n' "$C_H" "$total_events" "$total_final" "$C_R"
    printf '%s==========================================================%s\n' "$C_H" "$C_R"

    if (( total_final == 0 )); then
        echo "No data matched your filters."
        exit 0
    fi

    top_n "STATUS BREAKDOWN (final status per recipient):" 10 '{ print $10 }'
    top_n "TOP 10 CLIENT IPs (per message):" 10 '$4 == "NOQUEUE" || !seen[$4]++ { print $6 }'
    top_n "TOP 10 SENDERS (per message):" 10 '$4 == "NOQUEUE" || !seen[$4]++ { print $8 }'
    top_n "TOP 10 RECIPIENT DOMAINS:" 10 '$9 != "-" { d = tolower($9); sub(/^.*@/, "", d); print d }'
    top_n "TOP 10 RELAYS:" 10 '$7 != "-" && $7 != "none" { print $7 }'

    printf '\n%sDATA VOLUME:%s\n' "$C_Y" "$C_R"
    "$AWK" -F'\t' '$4 != "NOQUEUE" && !($4 in s) { s[$4] = 1; sum += $11; n++ }
        END { printf "  Messages:   %d\n  Total size: %.2f MB (counted once per queue ID)\n", n, sum / 1048576 }' "$LATEST"
)

if (( SUMMARY_MODE )); then
    # last status per (queue id, recipient); NOQUEUE rejects counted individually
    run_limited "$AWK" -F'\t' '
        $4 == "NOQUEUE" { print; next }
        { k = $4 SUBSEP $9; if (!(k in last)) ord[++n] = k; last[k] = $0 }
        END { for (i = 1; i <= n; i++) print last[ord[i]] }' "$SORTED" > "$LATEST"
    set +e
    render_summary | use_pager
    ps=("${PIPESTATUS[@]}")
    set -e
    check_pipe "${ps[@]}"
    exit 0
fi

# ---------------------------------------------------------- list / detail ----
if [[ ! -s $SORTED ]] && (( ! CSV_MODE )); then
    if (( DETAIL_VIEW )); then
        echo "No records found for ID: $ID_FILTER"
    else
        echo "No records matched your filters."
    fi
    exit 0
fi

set +e
if (( DETAIL_VIEW )); then
    "$AWK" "$AWK_FORMAT" "$SORTED" | use_pager
else
    tail -n "$LIMIT" -- "$SORTED" | "$AWK" "$AWK_FORMAT" | use_pager
fi
ps=("${PIPESTATUS[@]}")
set -e
check_pipe "${ps[@]}"