#!/usr/bin/env bash
# Stream local Telegram Lite archive batches into production without storing
# large JSONL packages on the VPS.

set -Eeuo pipefail

SOURCE_ROOT="${TELEGRAM_LITE_ARCHIVE_SOURCE_ROOT:-/Users/belilovsky/Downloads/Telegram Lite}"
WORK_DIR="${TELEGRAM_LITE_ARCHIVE_WORK_DIR:-/tmp/echo-telegram-lite-stream}"
REPORT_DIR="${TELEGRAM_LITE_ARCHIVE_REPORT_DIR:-$PWD/reports/telegram_lite_archive_stream}"
STATE_FILE="${TELEGRAM_LITE_ARCHIVE_STATE_FILE:-$REPORT_DIR/stream_state.tsv}"
REMOTE_HOST="${TELEGRAM_LITE_ARCHIVE_REMOTE_HOST:-root@62.72.32.112}"
REMOTE_APP_DIR="${TELEGRAM_LITE_ARCHIVE_REMOTE_APP_DIR:-/opt/echo-sounder}"
LIMIT_EXPORTS="${TELEGRAM_LITE_ARCHIVE_LIMIT_EXPORTS:-5}"
START_OFFSET="${TELEGRAM_LITE_ARCHIVE_START_OFFSET:-0}"
END_OFFSET="${TELEGRAM_LITE_ARCHIVE_END_OFFSET:-}"
IMPORT_BATCH_SIZE="${TELEGRAM_LITE_ARCHIVE_IMPORT_BATCH_SIZE:-500}"
TARGETS="${TELEGRAM_LITE_ARCHIVE_TARGETS:-echo,qazlake}"
PYTHON_BIN="${TELEGRAM_LITE_ARCHIVE_PYTHON:-}"
ROWS_PER_STREAM="${TELEGRAM_LITE_ARCHIVE_ROWS_PER_STREAM:-500}"
MAX_STREAM_CHUNKS="${TELEGRAM_LITE_ARCHIVE_MAX_STREAM_CHUNKS:-0}"
MAX_LOAD_1M="${TELEGRAM_LITE_ARCHIVE_MAX_LOAD_1M:-8.0}"

if [[ -z "$PYTHON_BIN" ]]; then
    if [[ -x ".venv/bin/python" ]]; then
        PYTHON_BIN=".venv/bin/python"
    else
        PYTHON_BIN="python3"
    fi
fi

mkdir -p "$WORK_DIR" "$REPORT_DIR"
touch "$STATE_FILE"

offset="$START_OFFSET"
while true; do
    if [[ -n "$END_OFFSET" && "$offset" -ge "$END_OFFSET" ]]; then
        break
    fi

    batch_jsonl="$WORK_DIR/archive_offset_${offset}.jsonl"
    batch_report="$REPORT_DIR/archive_offset_${offset}.json"
    "$PYTHON_BIN" scripts/telegram_lite_archive_intake.py \
        --source-root "$SOURCE_ROOT" \
        --offset "$offset" \
        --limit-exports "$LIMIT_EXPORTS" \
        --dump-jsonl "$batch_jsonl" \
        --report "$batch_report" \
        --rescan-processed

    row_count="$(wc -l < "$batch_jsonl" | tr -d ' ')"
    if [[ "$row_count" == "0" ]]; then
        echo "telegram_lite_stream done offset=$offset rows=0"
        break
    fi

    chunk_dir="$WORK_DIR/chunks_offset_${offset}_$(date +%s)_$$"
    mkdir -p "$chunk_dir"
    split -l "$ROWS_PER_STREAM" -d -a 5 "$batch_jsonl" "$chunk_dir/chunk_"

    chunk_index=0
    processed_chunks=0
    paused=0
    for chunk_file in "$chunk_dir"/chunk_*; do
        [[ -f "$chunk_file" ]] || continue
        if [[ "$MAX_STREAM_CHUNKS" -gt 0 && "$processed_chunks" -ge "$MAX_STREAM_CHUNKS" ]]; then
            echo "telegram_lite_stream paused offset=$offset max_chunks=$MAX_STREAM_CHUNKS"
            paused=1
            break
        fi
        state_key="${offset}:${chunk_index}"
        if grep -qxF "$state_key" "$STATE_FILE"; then
            chunk_index=$((chunk_index + 1))
            continue
        fi
        remote_load="$(ssh "$REMOTE_HOST" "cut -d' ' -f1 /proc/loadavg" || echo 99)"
        if ! awk -v load_value="$remote_load" -v max_value="$MAX_LOAD_1M" 'BEGIN {exit !(load_value <= max_value)}'; then
            echo "telegram_lite_stream paused load_high load_1m=$remote_load max=$MAX_LOAD_1M offset=$offset chunk=$chunk_index"
            paused=1
            break
        fi
        IFS=',' read -r -a target_list <<< "$TARGETS"
        for target in "${target_list[@]}"; do
            target="$(echo "$target" | xargs)"
            [[ -z "$target" ]] && continue
            cat "$chunk_file" | ssh "$REMOTE_HOST" \
                "cd '$REMOTE_APP_DIR' && docker compose exec -T web env PYTHONDONTWRITEBYTECODE=1 python scripts/telegram_lite_archive_intake.py --input-jsonl - --apply --target '$target' --skip-target-summary --import-batch-size '$IMPORT_BATCH_SIZE' --report reports/telegram_lite_archive_stream_offset_${offset}_chunk_${chunk_index}.${target}.json"
        done
        echo "$state_key" >> "$STATE_FILE"
        processed_chunks=$((processed_chunks + 1))
        chunk_index=$((chunk_index + 1))
    done

    echo "telegram_lite_stream imported offset=$offset rows=$row_count chunks=$processed_chunks targets=$TARGETS"
    if [[ "$paused" == "1" ]]; then
        break
    fi
    offset=$((offset + LIMIT_EXPORTS))
done