valmojr/CamFinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactored redis component to use a centralized postgres database with coordinator instead of descentralized sqlite shitty databases

Browse Source
This commit is contained in:
Valmo
2026-06-30 18:07:47 -03:00
parent bae8712f3b
commit e81211c88e
10 changed files with 854 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/redis_rtsp.rs
+527 -147
View File
@@ -1,6 +1,7 @@
use crate::{
config::Config,
terminal::{ScreenState, TerminalUi},
db::Database,
terminal::{RedisWorkerScreenState, ScreenState, TerminalUi},
};
use anyhow::{anyhow, Context, Result};
use std::{
@@ -20,6 +21,9 @@ use tokio::{
const DEFAULT_DONE_SUFFIX: &str = ":done";
const DEFAULT_SENTINEL: &str = "__CAMFINDER_DONE__";
const GLOBAL_PROGRESS_REFRESH: Duration = Duration::from_secs(5);
const UI_REFRESH: Duration = Duration::from_millis(250);
const RESULT_QUEUE_SUFFIX: &str = ":results";
pub async fn run_redis_coordinator(
config: Config,
@@ -34,8 +38,22 @@ pub async fn run_redis_coordinator(
));
}
let database = Database::open(&config.database_url).await?;
let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX);
clear_queue(&config.redis_url, &config.redis_queue_name, &done_key).await?;
let result_queue = result_queue_name(&config.redis_queue_name);
let total_key = total_ips_key(&config.redis_queue_name);
let completed_key = completed_ips_key(&config.redis_queue_name);
clear_queue(
&config.redis_url,
&config.redis_queue_name,
&result_queue,
&done_key,
&total_key,
&completed_key,
)
.await?;
set_redis_key(&config.redis_url, &total_key, ips.len()).await?;
set_redis_key(&config.redis_url, &completed_key, 0).await?;
enqueue_ips(
&config.redis_url,
&config.redis_queue_name,
@@ -63,30 +81,74 @@ pub async fn run_redis_coordinator(
);
}
consume_result_queue(
&config.redis_url,
&config.redis_queue_name,
&result_queue,
&completed_key,
ips.len(),
&database,
)
.await?;
Ok(())
}
pub async fn run_redis_worker(
config: Config,
_ui: Arc<TerminalUi>,
ui: Arc<TerminalUi>,
interrupted: Arc<AtomicBool>,
) -> Result<()> {
ensure_output_dirs(&config.snapshot_dir, &config.sqlite_path)?;
init_sqlite(&config.sqlite_path).await?;
ensure_snapshot_dir(&config.snapshot_dir)?;
let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX);
let result_queue = result_queue_name(&config.redis_queue_name);
let total_key = total_ips_key(&config.redis_queue_name);
let completed_key = completed_ips_key(&config.redis_queue_name);
ensure_total_ips_key(&config.redis_url, &total_key, &config.redis_input_csv).await?;
let started_at = Instant::now();
let mut processed = 0usize;
let mut captured = 0usize;
let paths = normalize_paths(&config.rtsp_paths);
let credentials = load_rtsp_credentials(Path::new("/app/main_ids.json"))?;
let worker_name = format!("{}-{}", config.worker_name, std::process::id());
let mut global_stats = read_global_progress(
&config.redis_url,
&config.redis_queue_name,
&total_key,
&completed_key,
)
.await
.unwrap_or_default();
let mut last_global_refresh = Instant::now();
let mut last_ui_render = Instant::now()
.checked_sub(UI_REFRESH)
.unwrap_or_else(Instant::now);
println!(
"worker pronto para {} paths, {} credenciais e fila {}",
paths.len(),
credentials.len(),
config.redis_queue_name
);
if ui.is_enabled() {
render_worker_progress(
&ui,
&worker_name,
&config.redis_queue_name,
"aguardando bloco",
"",
"iniciando",
processed,
captured,
&global_stats,
0,
0,
started_at,
);
} else {
println!(
"worker pronto para {} paths, {} credenciais e fila {}",
paths.len(),
credentials.len(),
config.redis_queue_name
);
}
loop {
if interrupted.load(Ordering::SeqCst) {
@@ -96,7 +158,14 @@ pub async fn run_redis_worker(
match pop_queue_item(&config.redis_url, &config.redis_queue_name).await? {
Some(item) if item == DEFAULT_SENTINEL => break,
Some(block) => {
for ip in parse_ip_block(&block) {
let block_ips = parse_ip_block(&block);
let block_total = block_ips
.len()
.saturating_mul(paths.len())
.saturating_mul(credentials.len());
let mut block_done = 0usize;
for ip in block_ips {
for rtsp_path in &paths {
if interrupted.load(Ordering::SeqCst) {
break;
@@ -154,24 +223,69 @@ pub async fn run_redis_worker(
)
};
insert_sqlite_result(
&config.sqlite_path,
checked_at,
&ip,
&rtsp_path,
&attempt,
status,
screenshot_path.as_deref(),
error.as_deref(),
push_fetch_result(
&config.redis_url,
&result_queue,
&RtspFetchResult {
checked_at,
target_ip: ip.clone(),
target_port: config.port,
rtsp_path: rtsp_path.clone(),
rtsp_url: attempt.clone(),
screenshot_path: screenshot_path.clone(),
status,
error,
},
)
.await?;
processed += 1;
println!(
"{} -> {}",
redact_rtsp_url(&attempt),
screenshot_path.unwrap_or_else(|| "sem imagem".to_string())
);
block_done = block_done.saturating_add(1);
let output_label =
screenshot_path.unwrap_or_else(|| "sem imagem".to_string());
if ui.is_enabled() {
let now = Instant::now();
if now.duration_since(last_global_refresh)
>= GLOBAL_PROGRESS_REFRESH
{
if let Ok(stats) = read_global_progress(
&config.redis_url,
&config.redis_queue_name,
&total_key,
&completed_key,
)
.await
{
global_stats = stats;
}
last_global_refresh = now;
}
if now.duration_since(last_ui_render) >= UI_REFRESH {
render_worker_progress(
&ui,
&worker_name,
&config.redis_queue_name,
&ip,
rtsp_path,
&format!(
"{} -> {}",
redact_rtsp_url(&attempt),
output_label
),
processed,
captured,
&global_stats,
block_done,
block_total,
started_at,
);
last_ui_render = now;
}
} else {
println!("{} -> {}", redact_rtsp_url(&attempt), output_label);
}
if attempt_captured {
captured += 1;
@@ -188,9 +302,45 @@ pub async fn run_redis_worker(
continue;
}
}
increment_redis_key(&config.redis_url, &completed_key).await?;
global_stats.completed_ips = global_stats.completed_ips.saturating_add(1);
}
if ui.is_enabled() {
render_worker_progress(
&ui,
&worker_name,
&config.redis_queue_name,
"aguardando bloco",
"",
"bloco finalizado",
processed,
captured,
&global_stats,
block_total,
block_total,
started_at,
);
}
}
None => {
if ui.is_enabled() {
render_worker_progress(
&ui,
&worker_name,
&config.redis_queue_name,
"fila vazia",
"",
"aguardando done",
processed,
captured,
&global_stats,
0,
0,
started_at,
);
}
if redis_key_exists(&config.redis_url, &done_key).await? {
break;
}
@@ -200,10 +350,28 @@ pub async fn run_redis_worker(
}
let elapsed = started_at.elapsed().as_secs_f64();
println!(
"worker finalizado em {:.2}s, tentativas {}, capturas {}",
elapsed, processed, captured
);
if ui.is_enabled() {
render_worker_progress(
&ui,
&worker_name,
&config.redis_queue_name,
"finalizado",
"",
&format!("tentativas {}, capturas {}", processed, captured),
processed,
captured,
&global_stats,
1,
1,
started_at,
);
ui.finish();
} else {
println!(
"worker finalizado em {:.2}s, tentativas {}, capturas {}",
elapsed, processed, captured
);
}
Ok(())
}
@@ -365,6 +533,25 @@ struct RtspProbe {
error: Option<String>,
}
#[derive(Debug, Clone, Default)]
struct GlobalProgress {
completed_ips: usize,
total_ips: usize,
queue_remaining_ips: Option<usize>,
}
#[derive(Debug, Clone)]
struct RtspFetchResult {
checked_at: i64,
target_ip: String,
target_port: u16,
rtsp_path: String,
rtsp_url: String,
screenshot_path: Option<String>,
status: String,
error: Option<String>,
}
impl RtspProbe {
fn is_connectable(&self) -> bool {
matches!(
@@ -423,8 +610,26 @@ async fn redis_key_exists(redis_url: &str, key: &str) -> Result<bool> {
Ok(output.trim() == "1")
}
async fn clear_queue(redis_url: &str, queue_name: &str, done_key: &str) -> Result<()> {
let _ = redis_cli(redis_url, &["DEL", queue_name, done_key]).await?;
async fn clear_queue(
redis_url: &str,
queue_name: &str,
result_queue: &str,
done_key: &str,
total_key: &str,
completed_key: &str,
) -> Result<()> {
let _ = redis_cli(
redis_url,
&[
"DEL",
queue_name,
result_queue,
done_key,
total_key,
completed_key,
],
)
.await?;
Ok(())
}
@@ -433,6 +638,294 @@ async fn mark_done(redis_url: &str, done_key: &str) -> Result<()> {
Ok(())
}
async fn set_redis_key(redis_url: &str, key: &str, value: usize) -> Result<()> {
let value = value.to_string();
let _ = redis_cli(redis_url, &["SET", key, &value]).await?;
Ok(())
}
async fn set_redis_key_if_missing(redis_url: &str, key: &str, value: usize) -> Result<()> {
let value = value.to_string();
let _ = redis_cli(redis_url, &["SETNX", key, &value]).await?;
Ok(())
}
async fn increment_redis_key(redis_url: &str, key: &str) -> Result<()> {
let _ = redis_cli(redis_url, &["INCR", key]).await?;
Ok(())
}
async fn ensure_total_ips_key(redis_url: &str, total_key: &str, input_csv: &str) -> Result<()> {
let total = load_ip_list(input_csv)?.len();
set_redis_key_if_missing(redis_url, total_key, total).await
}
async fn read_global_progress(
redis_url: &str,
queue_name: &str,
total_key: &str,
completed_key: &str,
) -> Result<GlobalProgress> {
let script = r#"
local queue = KEYS[1]
local total_key = KEYS[2]
local completed_key = KEYS[3]
local total = tonumber(redis.call('GET', total_key) or '0')
local completed = tonumber(redis.call('GET', completed_key) or '0')
local n = redis.call('LLEN', queue)
local remaining = 0
for i = 0, n - 1 do
local item = redis.call('LINDEX', queue, i)
if item ~= '__CAMFINDER_DONE__' then
for _ in string.gmatch(item, '[^\n]+') do
remaining = remaining + 1
end
end
end
return {completed, total, remaining}
"#;
let output = redis_cli(
redis_url,
&[
"--raw",
"EVAL",
script,
"3",
queue_name,
total_key,
completed_key,
],
)
.await?;
let values = output
.lines()
.filter_map(|line| line.trim().parse::<usize>().ok())
.collect::<Vec<_>>();
Ok(GlobalProgress {
completed_ips: values.get(0).copied().unwrap_or_default(),
total_ips: values.get(1).copied().unwrap_or_default(),
queue_remaining_ips: values.get(2).copied(),
})
}
fn total_ips_key(queue_name: &str) -> String {
format!("{queue_name}:total_ips")
}
fn completed_ips_key(queue_name: &str) -> String {
format!("{queue_name}:completed_ips")
}
fn result_queue_name(queue_name: &str) -> String {
format!("{queue_name}{RESULT_QUEUE_SUFFIX}")
}
async fn push_fetch_result(
redis_url: &str,
result_queue: &str,
result: &RtspFetchResult,
) -> Result<()> {
let serialized = serialize_fetch_result(result);
let _ = redis_cli(redis_url, &["RPUSH", result_queue, &serialized]).await?;
Ok(())
}
async fn pop_fetch_result(redis_url: &str, result_queue: &str) -> Result<Option<RtspFetchResult>> {
let output = redis_cli(redis_url, &["--raw", "LPOP", result_queue]).await?;
let value = output.trim_end_matches('\n');
if value.is_empty() {
return Ok(None);
}
deserialize_fetch_result(value).map(Some)
}
async fn consume_result_queue(
redis_url: &str,
queue_name: &str,
result_queue: &str,
completed_key: &str,
total_ips: usize,
database: &Database,
) -> Result<()> {
let mut inserted = 0usize;
loop {
let mut drained_any = false;
while let Some(result) = pop_fetch_result(redis_url, result_queue).await? {
database
.insert_rtsp_fetch_result(
result.checked_at,
&result.target_ip,
result.target_port,
&result.rtsp_path,
&result.rtsp_url,
result.screenshot_path.as_deref(),
&result.status,
result.error.as_deref(),
)
.await?;
inserted += 1;
drained_any = true;
}
let completed = redis_usize(redis_url, completed_key)
.await?
.unwrap_or_default();
let pending_results = redis_list_len(redis_url, result_queue).await?;
if completed >= total_ips && pending_results == 0 {
println!(
"writer finalizado: {} resultados gravados em Postgres para {}",
inserted, queue_name
);
break;
}
if !drained_any {
sleep(Duration::from_millis(500)).await;
}
}
Ok(())
}
async fn redis_usize(redis_url: &str, key: &str) -> Result<Option<usize>> {
let output = redis_cli(redis_url, &["--raw", "GET", key]).await?;
let value = output.trim();
if value.is_empty() {
return Ok(None);
}
value
.parse::<usize>()
.map(Some)
.with_context(|| format!("invalid Redis integer for {key}: {value}"))
}
async fn redis_list_len(redis_url: &str, key: &str) -> Result<usize> {
let output = redis_cli(redis_url, &["--raw", "LLEN", key]).await?;
output
.trim()
.parse::<usize>()
.with_context(|| format!("invalid Redis LLEN for {key}"))
}
fn serialize_fetch_result(result: &RtspFetchResult) -> String {
[
result.checked_at.to_string(),
result.target_port.to_string(),
encode_field(&result.target_ip),
encode_field(&result.rtsp_path),
encode_field(&result.rtsp_url),
encode_optional_field(result.screenshot_path.as_deref()),
encode_field(&result.status),
encode_optional_field(result.error.as_deref()),
]
.join("\t")
}
fn deserialize_fetch_result(value: &str) -> Result<RtspFetchResult> {
let parts = value.split('\t').collect::<Vec<_>>();
if parts.len() != 8 {
return Err(anyhow!(
"invalid RTSP fetch result field count: expected 8, got {}",
parts.len()
));
}
Ok(RtspFetchResult {
checked_at: parts[0]
.parse::<i64>()
.with_context(|| format!("invalid checked_at: {}", parts[0]))?,
target_port: parts[1]
.parse::<u16>()
.with_context(|| format!("invalid target_port: {}", parts[1]))?,
target_ip: decode_field(parts[2])?,
rtsp_path: decode_field(parts[3])?,
rtsp_url: decode_field(parts[4])?,
screenshot_path: decode_optional_field(parts[5])?,
status: decode_field(parts[6])?,
error: decode_optional_field(parts[7])?,
})
}
fn encode_optional_field(value: Option<&str>) -> String {
value.map(encode_field).unwrap_or_default()
}
fn decode_optional_field(value: &str) -> Result<Option<String>> {
if value.is_empty() {
Ok(None)
} else {
decode_field(value).map(Some)
}
}
fn encode_field(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.as_bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'.' | b'-' | b'_' | b'/' | b':' | b'@' => {
encoded.push(*byte as char)
}
_ => {
let _ = write!(&mut encoded, "%{byte:02X}");
}
}
}
encoded
}
fn decode_field(value: &str) -> Result<String> {
let bytes = value.as_bytes();
let mut decoded = Vec::with_capacity(bytes.len());
let mut index = 0usize;
while index < bytes.len() {
if bytes[index] == b'%' {
if index + 2 >= bytes.len() {
return Err(anyhow!("invalid percent escape in Redis result field"));
}
let hex = std::str::from_utf8(&bytes[index + 1..index + 3])
.context("invalid percent escape UTF-8")?;
let value = u8::from_str_radix(hex, 16)
.with_context(|| format!("invalid percent escape: %{hex}"))?;
decoded.push(value);
index += 3;
} else {
decoded.push(bytes[index]);
index += 1;
}
}
String::from_utf8(decoded).context("decoded Redis result field is not valid UTF-8")
}
fn render_worker_progress(
ui: &TerminalUi,
worker_name: &str,
queue_name: &str,
current_ip: &str,
current_path: &str,
current_status: &str,
processed: usize,
captured: usize,
global_stats: &GlobalProgress,
block_done: usize,
block_total: usize,
started_at: Instant,
) {
ui.render_redis_worker(&RedisWorkerScreenState {
worker_name: worker_name.to_string(),
queue_name: queue_name.to_string(),
current_ip: current_ip.to_string(),
current_path: current_path.to_string(),
current_status: current_status.to_string(),
processed,
captured,
global_done: global_stats.completed_ips,
global_total: global_stats.total_ips,
queue_remaining: global_stats.queue_remaining_ips,
block_done,
block_total,
elapsed: started_at.elapsed(),
});
}
async fn enqueue_ips(
redis_url: &str,
queue_name: &str,
@@ -496,103 +989,6 @@ async fn redis_cli(redis_url: &str, args: &[&str]) -> Result<String> {
Ok(String::from_utf8_lossy(&output.stdout).to_string())
}
async fn init_sqlite(sqlite_path: &Path) -> Result<()> {
let sql = r#"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA busy_timeout = 5000;
CREATE TABLE IF NOT EXISTS rtsp_feed_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
checked_at INTEGER NOT NULL,
target_ip TEXT NOT NULL,
rtsp_path TEXT NOT NULL,
rtsp_url TEXT NOT NULL,
screenshot_path TEXT,
status TEXT NOT NULL,
error TEXT
);
"#;
sqlite_exec(sqlite_path, sql).await
}
async fn insert_sqlite_result(
sqlite_path: &Path,
checked_at: i64,
target_ip: &str,
rtsp_path: &str,
rtsp_url: &str,
status: String,
screenshot_path: Option<&str>,
error: Option<&str>,
) -> Result<()> {
let sql = format!(
"INSERT INTO rtsp_feed_results (checked_at, target_ip, rtsp_path, rtsp_url, screenshot_path, status, error) VALUES ({checked_at}, {target_ip}, {rtsp_path}, {rtsp_url}, {screenshot_path}, {status}, {error});",
checked_at = checked_at,
target_ip = sql_quote(target_ip),
rtsp_path = sql_quote(rtsp_path),
rtsp_url = sql_quote(rtsp_url),
screenshot_path = sql_nullable_quote(screenshot_path),
status = sql_quote(&status),
error = sql_nullable_quote(error),
);
sqlite_exec_with_retry(sqlite_path, &sql).await
}
async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> {
let output = timeout(Duration::from_secs(20), async {
let mut command = Command::new("sqlite3");
command
.kill_on_drop(true)
.arg("-cmd")
.arg("PRAGMA busy_timeout = 5000;")
.arg("-cmd")
.arg("PRAGMA journal_mode = WAL;")
.arg("-cmd")
.arg("PRAGMA synchronous = NORMAL;")
.arg(sqlite_path)
.arg(sql);
command.output().await
})
.await
.context("sqlite3 timeout")??;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
return Err(anyhow!(
"sqlite3 failed{}",
if stderr.is_empty() {
String::new()
} else {
format!(": {stderr}")
}
));
}
Ok(())
}
async fn sqlite_exec_with_retry(sqlite_path: &Path, sql: &str) -> Result<()> {
let mut last_err: Option<anyhow::Error> = None;
for attempt in 0..5 {
match sqlite_exec(sqlite_path, sql).await {
Ok(()) => return Ok(()),
Err(err) if is_sqlite_locked(&err) && attempt < 4 => {
last_err = Some(err);
sleep(Duration::from_millis(200 * (attempt + 1) as u64)).await;
}
Err(err) => return Err(err),
}
}
Err(last_err.unwrap_or_else(|| anyhow!("sqlite3 failed after retries")))
}
fn is_sqlite_locked(err: &anyhow::Error) -> bool {
let text = err.to_string().to_ascii_lowercase();
text.contains("database is locked") || text.contains("busy")
}
fn load_ip_list(input: &str) -> Result<Vec<String>> {
let content = match fs::read_to_string(input) {
Ok(content) => content,
@@ -656,25 +1052,9 @@ fn build_snapshot_path(base_dir: &Path, ip: &str, rtsp_path: &str, checked_at: i
base_dir.join(format!("{ip_component}_{checked_at}_{path_component}.png"))
}
fn ensure_output_dirs(snapshot_dir: &Path, sqlite_path: &Path) -> Result<()> {
fn ensure_snapshot_dir(snapshot_dir: &Path) -> Result<()> {
fs::create_dir_all(snapshot_dir)
.with_context(|| format!("failed to create snapshot dir {}", snapshot_dir.display()))?;
if let Some(parent) = sqlite_path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("failed to create sqlite dir {}", parent.display()))?;
}
Ok(())
}
fn sql_quote(value: &str) -> String {
format!("'{}'", value.replace('\'', "''"))
}
fn sql_nullable_quote(value: Option<&str>) -> String {
match value {
Some(value) if !value.is_empty() => sql_quote(value),
_ => "NULL".to_string(),
}
.with_context(|| format!("failed to create snapshot dir {}", snapshot_dir.display()))
}
fn strip_quotes(value: &str) -> &str {