#!/usr/bin/env bash
# issued.live-convert.sh -- turn a daily issued.live CSV file into another format.
#
#   ./issued.live-convert.sh delta.domains.csv.zst              -> delta.domains.jsonl.zst
#   ./issued.live-convert.sh --to ndjson  delta.dns.csv.zst     -> same thing, explicit
#   ./issued.live-convert.sh --to parquet delta.dns.csv.zst     -> needs duckdb
#   ./issued.live-convert.sh --to csv     delta.dns.csv.zst     -> plain .csv, decompressed
#   ./issued.live-convert.sh --stdout --to ndjson f.csv.zst | your-loader
#
# We publish one format, CSV compressed with zstd, and this converts it to whatever you
# need. That is deliberate: it halves what you download and what we store, and the work
# happens on your machine where it costs a couple of minutes of CPU you already have.
#
# *** WHY YOU CANNOT JUST RUN csvtojson. ***
# Two columns are ARRAYS -- `values` on the dns and dns_vantage files, `sans` on the
# certificates file. CSV has no array type, so they are stored as a JSON array inside the
# cell, which a naive converter turns into a STRING containing "[...]" rather than a list.
# This script parses those back into real arrays, so the output matches the JSON the API
# returns for the same records.
#
# Everything streams. A 10 GB file is converted in constant memory; nothing is held whole.
set -euo pipefail

TO=ndjson
TO_STDOUT=0
OUT=""
usage() {
    sed -n '2,20p' "$0" | sed 's/^# \{0,1\}//'
    exit "${1:-0}"
}
while [ $# -gt 0 ]; do
    case "$1" in
        --to)     TO="${2:?--to needs a format}"; shift 2 ;;
        --out)    OUT="${2:?--out needs a path}"; shift 2 ;;
        --stdout) TO_STDOUT=1; shift ;;
        -h|--help) usage 0 ;;
        -*) echo "unknown option: $1" >&2; usage 2 ;;
        *)  break ;;
    esac
done
SRC="${1:?usage: $0 [--to ndjson|csv|parquet] <file.csv.zst>}"
[ -r "$SRC" ] || { echo "cannot read $SRC" >&2; exit 1; }

command -v zstd >/dev/null || { echo "zstd is required: apt install zstd / brew install zstd" >&2; exit 1; }

# Which columns hold arrays is decided by the DATASET, which is the middle part of the
# published name (delta.dns.csv.zst -> dns). Passing the wrong list would only mean a
# column stays a string, never wrong data, but get it right anyway.
base="$(basename "$SRC")"
dataset="$(echo "$base" | cut -d. -f2)"
# *** THE TYPE MAP IS PER DATASET, AND IT IS NOT GUESSWORK. ***
# CSV has no types: every cell is text. The API and the JSON files give `ttl` as a number
# and `wildcard` as a boolean, so a converter that emits everything as a string produces
# output that does not match -- verified against the published JSON, where this was the
# only difference.
#
# The lists below are derived from the published files, not inferred at runtime. Inferring
# would be worse than useless here: a domain like 1e5.com parses as a float, and a zone with
# a leading zero loses it. Anything not named below stays exactly the text we published.
case "$dataset" in
    dns)             ARRAYS="values"; NUMS="ttl";                                BOOLS="" ;;
    dns_vantage)     ARRAYS="values"; NUMS="ttl,changes";                        BOOLS="" ;;
    certificates)    ARRAYS="sans";   NUMS="key_bits,san_count,ct_logs_seen";    BOOLS="wildcard" ;;
    *)               ARRAYS="";       NUMS="";                                   BOOLS="" ;;
esac

stem="${base%.zst}"; stem="${stem%.csv}"
case "$TO" in
    ndjson|jsonl|json) DEFAULT_OUT="$stem.jsonl.zst" ;;
    csv)               DEFAULT_OUT="$stem.csv" ;;
    parquet)           DEFAULT_OUT="$stem.parquet" ;;
    *) echo "unknown --to format: $TO (want ndjson, csv or parquet)" >&2; exit 2 ;;
esac
[ -n "$OUT" ] || OUT="$DEFAULT_OUT"

if [ "$TO" = csv ]; then
    if [ "$TO_STDOUT" = 1 ]; then exec zstd -dc -- "$SRC"; fi
    zstd -dc -- "$SRC" > "$OUT"; echo "wrote $OUT" >&2; exit 0
fi

if [ "$TO" = parquet ]; then
    command -v duckdb >/dev/null || {
        echo "parquet needs duckdb (https://duckdb.org). Or use --to ndjson." >&2; exit 1; }
    tmp="$(mktemp -t issued-XXXXXX.csv)"; trap 'rm -f "$tmp"' EXIT
    zstd -dc -- "$SRC" > "$tmp"
    duckdb -c "COPY (SELECT * FROM read_csv_auto('$tmp', header=true)) TO '$OUT' (FORMAT parquet);"
    echo "wrote $OUT" >&2; exit 0
fi

# ---- ndjson -----------------------------------------------------------------------------
# python3's csv module streams and handles the doubled-quote escaping the files use.
# Fields are emitted as strings exactly as published: no type guessing, because a domain
# like 1e5.com would otherwise silently become a float.
convert() {
python3 -c '
import csv, json, sys
arrays = set(a for a in sys.argv[1].split(",") if a)
nums   = set(a for a in sys.argv[2].split(",") if a)
bools  = set(a for a in sys.argv[3].split(",") if a)
csv.field_size_limit(1 << 30)          # SAN lists and big value arrays exceed the default
r = csv.reader(sys.stdin)
try:
    header = next(r)
except StopIteration:
    sys.exit(0)
w = sys.stdout.write
for row in r:
    o = {}
    for k, v in zip(header, row):
        if k in arrays:
            try:
                v = json.loads(v) if v else []
            except ValueError:
                pass                    # keep the raw string rather than lose the value
        elif k in nums and v != "":
            try:
                v = int(v)
            except ValueError:
                try:
                    v = float(v)
                except ValueError:
                    pass                # an unparseable number stays as published
        elif k in bools and v != "":
            if v in ("true", "1"):    v = True
            elif v in ("false", "0"): v = False
        o[k] = v
    w(json.dumps(o, separators=(",", ":"), ensure_ascii=False))
    w("\n")
' "$ARRAYS" "$NUMS" "$BOOLS"
}

if [ "$TO_STDOUT" = 1 ]; then
    zstd -dc -- "$SRC" | convert
else
    # -T0 uses every core for the recompress; the conversion itself is one process.
    zstd -dc -- "$SRC" | convert | zstd -12 -T0 -q -o "$OUT" -f
    echo "wrote $OUT" >&2
fi