#!/usr/bin/env bash
# load-consumer-fetch.sh — Pull-based consumer (fetch) that polls and logs throughput.
#
# Polls `fafb-q fetch` in a loop, tracking offset progression. Logs throughput
# every LOG_INTERVAL seconds. Silent when no messages are received.
#
# Usage:
#   ./scripts/load-consumer-fetch.sh
#   POLL_INTERVAL_MS=100 FETCH_BATCH=2000 ./scripts/load-consumer-fetch.sh
#
# Requires: fafb-q CLI in PATH (or set FAFB_BIN).

set -euo pipefail

# ── Configurable variables ───────────────────────────────────────────────────
FAFB_BIN="${FAFB_BIN:-fafb-q}"                    # Path to fafb-q binary
SERVER="${SERVER:-localhost:9876}"                 # TCP server address
API_KEY="${API_KEY:-}"                             # API key (required)
NAMESPACE="${NAMESPACE:-default}"                  # Namespace
TOPIC="${TOPIC:-load-test}"                        # Topic name
FROM_OFFSET="${FROM_OFFSET:-}"                    # Start offset (empty = tail/latest)
FETCH_BATCH="${FETCH_BATCH:-2000}"                # Messages per fetch request
POLL_INTERVAL_MS="${POLL_INTERVAL_MS:-50}"         # Milliseconds between polls (when caught up)
PROFILE="${PROFILE:-}"                            # Config profile (optional)
TLS="${TLS:-false}"                               # Enable TLS (true/false)
TLS_CA="${TLS_CA:-}"                              # CA cert path (optional)
LOG_INTERVAL="${LOG_INTERVAL:-5}"                 # Stats logging interval in seconds
# ─────────────────────────────────────────────────────────────────────────────

if [[ -z "$API_KEY" && -z "$PROFILE" ]]; then
    echo "ERROR: API_KEY or PROFILE must be set" >&2; exit 1
fi
command -v "$FAFB_BIN" &>/dev/null || { echo "ERROR: $FAFB_BIN not found" >&2; exit 1; }

FLAGS=()
if [[ -n "$PROFILE" ]]; then FLAGS+=(--profile "$PROFILE")
else FLAGS+=(--server "$SERVER" --api-key "$API_KEY"); fi
[[ "$TLS" == "true" ]] && FLAGS+=(--tls)
[[ -n "$TLS_CA" ]] && FLAGS+=(--tls-ca "$TLS_CA")

FULL_TOPIC="${NAMESPACE}/${TOPIC}"
POLL_SECS=$(awk "BEGIN {printf \"%.3f\", ${POLL_INTERVAL_MS}/1000}")

echo "═══════════════════════════════════════════════════"
echo " fafb-q Fetch Consumer (pull-based)"
echo "═══════════════════════════════════════════════════"
echo " Server:       ${SERVER}"
echo " Topic:        ${FULL_TOPIC}"
echo " Start offset: ${FROM_OFFSET:-tail (latest)}"
echo " Batch size:   ${FETCH_BATCH}"
echo " Poll interval: ${POLL_INTERVAL_MS}ms (when caught up)"
echo "═══════════════════════════════════════════════════"
echo ""

NEXT_OFFSET="${FROM_OFFSET}"
TOTAL_RECEIVED=0
INTERVAL_RECEIVED=0
START_TIME=$(date +%s)
LAST_LOG_TIME=$START_TIME
BATCH_COUNT=0

trap 'echo ""; echo "Fetch consumer stopped. Total: ${TOTAL_RECEIVED} msgs"; exit 0' INT TERM

# Discover head offset if none specified
if [[ -z "$NEXT_OFFSET" ]]; then
    echo "No offset specified — starting from tail (new messages only)"
    TAIL_OUTPUT=$("$FAFB_BIN" fetch "$FULL_TOPIC" "${FLAGS[@]}" --tail 1 --json 2>&1 || echo "")
    if [[ -n "$TAIL_OUTPUT" ]] && echo "$TAIL_OUTPUT" | head -1 | grep -q '"offset"'; then
        NEXT_OFFSET=$(echo "$TAIL_OUTPUT" | head -1 | sed 's/.*"offset":\([0-9]*\).*/\1/')
        NEXT_OFFSET=$((NEXT_OFFSET + 1))
        echo "Discovered head offset: ${NEXT_OFFSET}"
    else
        echo "Topic empty or not created — polling from offset 0"
        NEXT_OFFSET=0
    fi
fi

while true; do
    # Fetch uses --json for reliable line counting (one JSON object per message).
    # Output is discarded after counting — we only care about throughput.
    OUTPUT=$("$FAFB_BIN" fetch "$FULL_TOPIC" \
        "${FLAGS[@]}" \
        --from "$NEXT_OFFSET" \
        --count "$FETCH_BATCH" \
        --json 2>/dev/null || echo "")

    if [[ -n "$OUTPUT" ]]; then
        # Use --json so each message is exactly one line. Count with wc -l.
        BATCH_COUNT=$(printf '%s\n' "$OUTPUT" | wc -l)
        BATCH_COUNT=$((BATCH_COUNT + 0))

        # Extract next_offset from the last line if it's a metadata line,
        # otherwise advance by batch count.
        if [[ $BATCH_COUNT -gt 0 ]]; then
            TOTAL_RECEIVED=$((TOTAL_RECEIVED + BATCH_COUNT))
            INTERVAL_RECEIVED=$((INTERVAL_RECEIVED + BATCH_COUNT))
            NEXT_OFFSET=$((NEXT_OFFSET + BATCH_COUNT))
        fi
    else
        BATCH_COUNT=0
    fi

    # Log stats at interval
    NOW=$(date +%s)
    SINCE_LAST_LOG=$((NOW - LAST_LOG_TIME))

    if [[ $SINCE_LAST_LOG -ge $LOG_INTERVAL ]]; then
        if [[ $INTERVAL_RECEIVED -gt 0 ]]; then
            ELAPSED=$((NOW - START_TIME))
            RATE=$((INTERVAL_RECEIVED / SINCE_LAST_LOG))
            echo "[${ELAPSED}s] fetched ${INTERVAL_RECEIVED} msgs in ${SINCE_LAST_LOG}s (${RATE} msgs/sec, offset: ${NEXT_OFFSET}, total: ${TOTAL_RECEIVED})"
        fi
        INTERVAL_RECEIVED=0
        LAST_LOG_TIME=$NOW
    fi

    # Full batch = more data available, fetch immediately. Partial = caught up, sleep.
    if [[ $BATCH_COUNT -lt $FETCH_BATCH ]]; then
        sleep "$POLL_SECS"
    fi
done
