// The executable: turn argv into a catalog plus one query and dispatch to
// `write` or `exec`. The pieces it leans on live next door -- spill_dir.mbt for
// the run directory, glob.mbt for `--csv` expansion, usage.mbt for the texts.

///|
/// `--csv` globs expand to their matches; several matches scan into ONE table
/// named after the pattern (partitioned-dataset shape). `--parquet` stays
/// single-file. Each group is (table name, kind, files).
fn build_groups(
  paths : Array[(String, String)],
) -> Array[(String, String, Array[String])] raise {
  let groups : Array[(String, String, Array[String])] = []
  for pair in paths {
    if pair.0 == "csv" && pair.1.contains("*") {
      let hits = expand_glob(pair.1)
      if hits.length() == 0 {
        raise cli_fail("the --csv pattern \{pair.1} matched no files")
      }
      let name = if hits.length() == 1 {
        table_stem(hits[0])
      } else {
        let stem = pattern_stem(pair.1)
        // a bare "*.csv" keeps no name once the star and suffix go: the
        // group would register table "" and every query would quote it
        if stem == "" {
          raise cli_fail(
            "the --csv pattern \{pair.1} matched \{hits.length()} files and leaves no table name; give them a shared name prefix (part-*.csv) so they scan into a named table",
          )
        }
        stem
      }
      groups.push((name, "csv", hits))
    } else {
      groups.push((table_stem(pair.1), pair.0, [pair.1]))
    }
  }
  groups
}

///|
/// Lexical path identity, for refusing a `--out` that is one of the inputs:
/// "./t.csv", "t.csv" and "d/../t.csv" name the same file. This is not a
/// filesystem identity check (symlinks and hard links are not followed); it
/// covers the slip it exists for.
fn same_path(a : String, b : String) -> Bool {
  normalize_path(a) == normalize_path(b)
}

///|
fn normalize_path(path : String) -> String {
  let parts : Array[String] = []
  for view in path.split("/") {
    let seg = view.to_owned()
    if seg == "" || seg == "." {
      continue
    }
    if seg == ".." && parts.length() > 0 && parts[parts.length() - 1] != ".." {
      let _ = parts.pop()
      continue
    }
    parts.push(seg)
  }
  let joined = parts.join("/")
  if path.has_prefix("/") {
    "/" + joined
  } else {
    joined
  }
}

///|
/// Register every input as a table. A misspelled `--date-col` would otherwise
/// be a silent no-op and only surface later as a confusing int32-vs-date type
/// mismatch, so it is checked against the registered schemas.
fn register_inputs(
  groups : Array[(String, String, Array[String])],
  date_cols : Array[String],
) -> @catalog.Catalog raise {
  let cat = @catalog.Catalog::new()
  // --date-col has one job: type an unannotated integer epoch-day PARQUET
  // column as DATE. CSV columns get their type from inference, so with no
  // Parquet input the flag can only be a silent no-op
  if date_cols.length() > 0 {
    let mut parquet = false
    for (_, kind, _) in groups {
      if kind == "parquet" {
        parquet = true
      }
    }
    if !parquet {
      raise cli_fail(
        "--date-col types --parquet columns only; a CSV column gets its type from inference, so this flag would do nothing",
      )
    }
  }
  // a second file with the same stem must not silently replace the first:
  // `--csv a/data.csv --csv b/data.csv` used to register both under "data"
  // and answer from the last one only
  let seen : Array[(String, String)] = []
  for (name, kind, files) in groups {
    for prev in seen {
      if prev.0 == name {
        raise cli_fail(
          "two inputs register as table \"\{name}\" (\{prev.1} and \{files[0]}); rename one file or load them into separate tables",
        )
      }
    }
    seen.push((name, files[0]))
    let entry = match kind {
      "parquet" =>
        @pq.scan_file(files[0], name, date_cols) catch {
          err => raise cli_fail(err.to_string())
        }
      _ =>
        if files.length() > 1 {
          @csv.scan_files(files, name) catch {
            err => raise cli_fail(err.to_string())
          }
        } else {
          @csv.scan_file(files[0], name) catch {
            err => raise cli_fail(err.to_string())
          }
        }
    }
    cat.register(entry)
  }
  for dc in date_cols {
    let mut found = false
    for _, entry in cat.tables {
      for col in entry.schema.columns {
        if col.name == dc {
          found = true
        }
      }
    }
    if !found {
      raise cli_fail(
        "--date-col: no column named \"\{dc}\" in any scanned table",
      )
    }
  }
  cat
}

///|
/// argv as the subcommands expect it: args[0] is the launcher, args[1] the
/// subcommand, everything after it that subcommand's own arguments. js runs
/// the CLI through `node