use std::io::{BufRead as _, BufReader, Write as _};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::time::Duration;
use crate::routes::{compile_binary, find_binary};
const BACKFILL_ENV: &str = "AUTUMN_SEARCH_BACKFILL";
const BACKFILL_PURGE_ENV: &str = "AUTUMN_SEARCH_BACKFILL_PURGE";
const BACKFILL_STARTED_MARKER: &str = "autumn-search: backfill starting";
const STARTUP_GRACE: Duration = Duration::from_secs(120);
pub struct ReindexOptions<'a> {
pub package: Option<&'a str>,
pub bin: Option<&'a str>,
pub index: Option<&'a str>,
pub profile: Option<&'a str>,
pub purge: bool,
}
#[must_use]
pub fn backfill_target(index: Option<&str>) -> String {
index.unwrap_or("all").to_owned()
}
#[must_use]
pub fn is_started_marker(line: &str) -> bool {
line.starts_with(BACKFILL_STARTED_MARKER)
}
fn clear_competing_one_shot_env(command: &mut Command) {
for var in [
"AUTUMN_BUILD_STATIC",
"AUTUMN_DUMP_ROUTES",
"AUTUMN_DUMP_CACHE_COHERENCE",
"AUTUMN_DUMP_DATA_FLOW",
"AUTUMN_DUMP_AGENT_AUTHORITY",
"AUTUMN_DUMP_GRAPH",
"AUTUMN_DUMP_JOBS",
"AUTUMN_LIST_TASKS",
"AUTUMN_RUN_TASK",
"AUTUMN_MIGRATE",
"AUTUMN_RETENTION_DRY_RUN",
"AUTUMN_DB_RETENTION",
"AUTUMN_REPLAY_CAPSULE",
] {
command.env_remove(var);
}
}
fn apply_purge_env(command: &mut Command, purge: bool) {
if purge {
command.env(BACKFILL_PURGE_ENV, "1");
} else {
command.env_remove(BACKFILL_PURGE_ENV);
}
}
fn apply_profile_env(command: &mut Command, profile: Option<&str>) {
if let Some(profile) = profile {
command.env("AUTUMN_ENV", profile);
} else {
command.env_remove("AUTUMN_ENV");
command.env_remove("AUTUMN_PROFILE");
}
}
pub fn run(opts: &ReindexOptions<'_>) {
eprintln!("\u{1F342} autumn search reindex\n");
match opts.profile {
Some(profile) => eprintln!(" profile: {profile}\n"),
None => eprintln!(
" profile: dev (this builds a debug binary; pass --profile to \
rebuild another environment's index)\n"
),
}
if opts.purge {
eprintln!(
" purge: each index is CLEARED before it is rebuilt — \
searches return nothing until the rebuild finishes\n"
);
}
compile_binary(opts.package, opts.bin);
let binary = find_binary(opts.package, opts.bin);
let mut command = Command::new(&binary);
clear_competing_one_shot_env(&mut command);
command.env(BACKFILL_ENV, backfill_target(opts.index));
apply_purge_env(&mut command, opts.purge);
apply_profile_env(&mut command, opts.profile);
let mut child = command
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()
.unwrap_or_else(|e| {
eprintln!("\u{2717} Failed to run {}: {e}", binary.display());
std::process::exit(1);
});
let stdout = child.stdout.take().expect("stdout was piped");
let (tx, rx) = mpsc::channel::<()>();
let forwarder = std::thread::spawn(move || {
let mut announced = false;
for line in BufReader::new(stdout).lines().map_while(Result::ok) {
println!("{line}");
let _ = std::io::stdout().flush();
if !announced && is_started_marker(&line) {
announced = true;
let _ = tx.send(());
}
}
});
if rx.recv_timeout(STARTUP_GRACE).is_err() {
let _ = child.kill();
let _ = child.wait();
drop(forwarder);
eprintln!(
"\n\u{2717} The application did not start a search backfill within {}s.\n\
\n\
The most likely cause is that it does not install `SearchPlugin`, so nothing\n\
consumed {BACKFILL_ENV} and the process went on to serve HTTP instead.\n\
Add the plugin to the app builder:\n\
\n\
.plugin(SearchPlugin::new().postgres().index::<YourModel>())\n\
\n\
If the app is simply slow to boot (migrations, a cold pool), re-run once it is warm.",
STARTUP_GRACE.as_secs()
);
std::process::exit(1);
}
let status = child.wait().unwrap_or_else(|e| {
eprintln!("\u{2717} Failed to wait for {}: {e}", binary.display());
std::process::exit(1);
});
let _ = forwarder.join();
if !status.success() {
eprintln!(
"\u{2717} Reindex failed (exit status {status}). \
Is the database reachable?"
);
std::process::exit(status.code().unwrap_or(1));
}
eprintln!("\n\u{2714} Reindex complete");
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_profile_is_forwarded_as_the_selector_the_child_reads_first() {
let opts = ReindexOptions {
package: None,
bin: None,
index: None,
profile: Some("prod"),
purge: false,
};
assert_eq!(opts.profile, Some("prod"));
assert_eq!(
autumn_web::config::normalize_profile_name("prod").as_deref(),
Some("prod"),
"the forwarded value must be one core recognizes"
);
}
fn set_envs(command: &mut Command, keys: impl IntoIterator<Item = &'static str>) {
for key in keys {
command.env(key, "1");
}
}
fn still_set(command: &Command, keys: &[&str]) -> Vec<String> {
command
.get_envs()
.filter_map(|(key, value)| value.map(|_| key))
.filter_map(|key| key.to_str().map(str::to_owned))
.filter(|key| keys.contains(&key.as_str()))
.collect()
}
#[test]
fn clear_competing_one_shot_env_removes_every_var_checked_before_server_startup() {
let competing_vars = [
"AUTUMN_BUILD_STATIC",
"AUTUMN_DUMP_ROUTES",
"AUTUMN_DUMP_CACHE_COHERENCE",
"AUTUMN_DUMP_DATA_FLOW",
"AUTUMN_DUMP_AGENT_AUTHORITY",
"AUTUMN_DUMP_GRAPH",
"AUTUMN_DUMP_JOBS",
"AUTUMN_LIST_TASKS",
"AUTUMN_RUN_TASK",
"AUTUMN_MIGRATE",
"AUTUMN_RETENTION_DRY_RUN",
"AUTUMN_DB_RETENTION",
"AUTUMN_REPLAY_CAPSULE",
];
let mut command = Command::new("true");
set_envs(&mut command, competing_vars);
clear_competing_one_shot_env(&mut command);
let remaining = still_set(&command, &competing_vars);
assert!(
remaining.is_empty(),
"these competing one-shot vars survived: {remaining:?}"
);
}
#[test]
fn an_inherited_framework_retention_purge_cannot_hijack_a_reindex() {
let mut command = Command::new("true");
command.env("AUTUMN_DB_RETENTION", "purge");
clear_competing_one_shot_env(&mut command);
assert!(
still_set(&command, &["AUTUMN_DB_RETENTION"]).is_empty(),
"an inherited AUTUMN_DB_RETENTION=purge must never survive into a reindex"
);
}
#[test]
fn apply_purge_env_clears_an_inherited_purge_flag_when_purge_is_not_requested() {
let mut command = Command::new("true");
command.env(BACKFILL_PURGE_ENV, "1");
apply_purge_env(&mut command, false);
assert!(
still_set(&command, &[BACKFILL_PURGE_ENV]).is_empty(),
"an inherited purge flag must not survive a purge-less reindex"
);
}
#[test]
fn apply_purge_env_sets_the_flag_when_purge_is_requested() {
let mut command = Command::new("true");
apply_purge_env(&mut command, true);
assert_eq!(
still_set(&command, &[BACKFILL_PURGE_ENV]),
[BACKFILL_PURGE_ENV]
);
}
#[test]
fn apply_profile_env_clears_inherited_profile_selectors_for_default_reindex() {
let mut command = Command::new("true");
command.env("AUTUMN_ENV", "prod");
command.env("AUTUMN_PROFILE", "prod");
apply_profile_env(&mut command, None);
assert!(
still_set(&command, &["AUTUMN_ENV", "AUTUMN_PROFILE"]).is_empty(),
"an inherited profile selector must not survive an unset --profile"
);
}
#[test]
fn apply_profile_env_forwards_an_explicit_profile_as_autumn_env() {
let mut command = Command::new("true");
apply_profile_env(&mut command, Some("prod"));
assert_eq!(still_set(&command, &["AUTUMN_ENV"]), ["AUTUMN_ENV"]);
}
#[test]
fn no_index_means_every_index() {
assert_eq!(backfill_target(None), "all");
}
#[test]
fn a_named_index_is_passed_through_verbatim() {
assert_eq!(backfill_target(Some("articles")), "articles");
}
#[test]
fn the_env_var_names_match_the_plugin_contract() {
assert_eq!(BACKFILL_ENV, "AUTUMN_SEARCH_BACKFILL");
assert_eq!(BACKFILL_PURGE_ENV, "AUTUMN_SEARCH_BACKFILL_PURGE");
assert_eq!(BACKFILL_STARTED_MARKER, "autumn-search: backfill starting");
}
#[test]
fn the_start_marker_is_recognised_with_its_target_suffix() {
assert!(is_started_marker("autumn-search: backfill starting all"));
assert!(is_started_marker(
"autumn-search: backfill starting articles"
));
assert!(!is_started_marker(
"autumn-search: reindexed articles (3 documents)"
));
assert!(!is_started_marker("Listening on http://0.0.0.0:3000"));
assert!(!is_started_marker(""));
}
#[test]
fn the_startup_grace_bounds_boot_not_the_backfill() {
assert!(STARTUP_GRACE >= Duration::from_secs(60));
}
}