use crate::tui::api::{self, ApiCall, PublishSpec, Severity, RECEIVE_HOLD_SECS, RECEIVE_MAX};
use crate::tui::editor::{Editor, Field, Outcome, Purpose};
use crate::tui::effect::Effect;
use crate::tui::keymap::Action;
use crate::tui::model::{
BodyView, Confirm, Modal, Model, Pending, Received, TopicMonitor, Typed, MAX_RECEIVED,
};
use crate::tui::screens::cluster::moved;
use crate::tui::update::run;
use super::refuse;
pub mod format;
pub mod view;
const KEPT_BODY: usize = 4096;
const KEPT_RAW: usize = 16 * 1024;
fn monitor(m: &Model) -> Option<&TopicMonitor> {
m.topics.monitor.as_ref()
}
pub fn is_held(row: &Received, now: u64) -> bool {
!row.acked && row.held_until.is_some_and(|until| until > now)
}
pub fn gate(m: &Model, action: Action) -> Option<&'static str> {
let Some(me) = &m.me.data else {
return Some("gate.me-unknown");
};
let mon = monitor(m)?;
if m.busy.is_some() {
return Some("gate.busy");
}
match action {
Action::MsgPublish if mon.dead_letter => Some("gate.topic-read-only"),
Action::MsgPublish if !me.can("publish") => Some("gate.need-publish"),
Action::MsgReceive
| Action::MsgAck
| Action::MsgAckAll
| Action::MsgRelease
| Action::MsgVisibility
if mon.distributor =>
{
Some("gate.topic-no-receive")
}
Action::MsgReceive
| Action::MsgAck
| Action::MsgAckAll
| Action::MsgRelease
| Action::MsgVisibility
if !me.can("consume") =>
{
Some("gate.need-consume")
}
Action::MsgAck | Action::MsgRelease | Action::MsgVisibility
if selected_held(m).is_none() =>
{
Some("gate.no-held-message")
}
Action::MsgAckAll if !mon.messages.rows.iter().any(|r| is_held(r, m.now)) => {
Some("gate.no-held-message")
}
Action::MsgPrune if mon.distributor => Some("gate.topic-no-receive"),
Action::MsgPrune if !me.can("manage") => Some("gate.need-manage"),
_ => None,
}
}
fn selected_held(m: &Model) -> Option<&Received> {
let mon = monitor(m)?;
mon.messages
.rows
.get(mon.messages.selected)
.filter(|row| is_held(row, m.now))
}
pub fn on_action(m: &mut Model, action: Action) -> Vec<Effect> {
match action {
Action::Up => step(m, -1),
Action::Down => step(m, 1),
Action::PageUp => step(m, -10),
Action::PageDown => step(m, 10),
Action::First => step(m, i32::MIN / 2),
Action::Last => step(m, i32::MAX / 2),
Action::Enter => {
if let Some(mon) = &mut m.topics.monitor {
if !mon.messages.rows.is_empty() {
mon.messages.last = None;
m.modal = Some(Modal::Message);
}
}
Vec::new()
}
Action::MsgPublish
| Action::MsgReceive
| Action::MsgAck
| Action::MsgAckAll
| Action::MsgRelease
| Action::MsgPrune => {
if let Some(key) = gate(m, action) {
return refuse(m, key);
}
carry_out(m, action)
}
_ => Vec::new(),
}
}
fn carry_out(m: &mut Model, action: Action) -> Vec<Effect> {
let Some(mon) = monitor(m) else {
return Vec::new();
};
let topic = mon.name.clone();
match action {
Action::MsgPublish => {
open_publish(m, topic);
Vec::new()
}
Action::MsgReceive => {
open_receive(m, topic);
Vec::new()
}
Action::MsgPrune => {
let typed = Typed {
expected: topic.clone(),
input: String::new(),
};
confirm(m, Pending::Prune(topic), Some(typed));
Vec::new()
}
Action::MsgAck => match selected_held(m).map(|r| (r.id.clone(), r.receipt.clone())) {
Some((id, receipt)) => run(m, ApiCall::Ack(topic, id, receipt)),
None => Vec::new(),
},
Action::MsgRelease => match selected_held(m).map(|r| (r.id.clone(), r.receipt.clone())) {
Some((id, receipt)) => run(m, ApiCall::Visibility(topic, id, receipt, 0)),
None => Vec::new(),
},
Action::MsgAckAll => ack_all(m, &topic),
_ => Vec::new(),
}
}
fn confirm(m: &mut Model, pending: Pending, typed: Option<Typed>) {
m.modal = Some(Modal::Confirm(Confirm {
pending,
typed,
ok_focused: false,
}));
}
fn ack_all(m: &mut Model, topic: &str) -> Vec<Effect> {
let now = m.now;
let calls: Vec<ApiCall> = monitor(m)
.map(|mon| {
mon.messages
.rows
.iter()
.filter(|row| is_held(row, now))
.map(|row| ApiCall::Ack(topic.to_owned(), row.id.clone(), row.receipt.clone()))
.collect()
})
.unwrap_or_default();
calls
.into_iter()
.map(|call| Effect::Call {
id: m.id(),
generation: m.generation,
call,
})
.collect()
}
fn step(m: &mut Model, delta: i32) -> Vec<Effect> {
if let Some(mon) = &mut m.topics.monitor {
let rows = mon.messages.rows.len();
if rows > 0 {
mon.messages.selected = moved(mon.messages.selected, delta, rows);
}
}
Vec::new()
}
fn open_publish(m: &mut Model, topic: String) {
let fields = vec![
Field::area("body", "editor.field.body", ""),
Field::text(
"content-type",
"editor.field.content-type",
"application/json",
),
Field::check("base64", "editor.field.base64", false),
Field::text("group", "editor.field.group", ""),
Field::text("dedup", "editor.field.dedup", ""),
];
m.modal = Some(Modal::Editor(Editor::new(
Purpose::MessagePublish { topic },
fields,
)));
}
fn open_receive(m: &mut Model, topic: String) {
let fields = vec![
Field::text("max", "editor.field.max", &RECEIVE_MAX.to_string()),
Field::text("hold", "editor.field.hold", &RECEIVE_HOLD_SECS.to_string()),
];
m.modal = Some(Modal::Editor(Editor::new(
Purpose::MessageReceive { topic },
fields,
)));
}
fn build_receive(editor: &Editor, topic: &str) -> Result<ApiCall, (&'static str, &'static str)> {
let max = match editor.text("max").parse::<u32>() {
Ok(n) if n > 0 => n,
_ => return Err(("max", "editor.error.max")),
};
match editor.text("hold").parse::<u32>() {
Ok(secs) if secs > 0 => Ok(ApiCall::Receive(topic.to_owned(), secs, max)),
_ => Err(("hold", "editor.error.hold")),
}
}
fn open_visibility(m: &mut Model) {
let Some(mon) = monitor(m) else { return };
let Some(row) = selected_held(m) else { return };
let purpose = Purpose::MessageVisibility {
topic: mon.name.clone(),
id: row.id.clone(),
receipt: row.receipt.clone(),
};
let fields = vec![Field::text(
"seconds",
"editor.field.seconds",
&RECEIVE_HOLD_SECS.to_string(),
)];
m.modal = Some(Modal::Editor(Editor::new(purpose, fields)));
}
fn build_visibility(
editor: &Editor,
topic: &str,
id: &str,
receipt: &str,
) -> Result<ApiCall, (&'static str, &'static str)> {
match editor.text("seconds").parse::<u32>() {
Ok(secs) => Ok(ApiCall::Visibility(
topic.to_owned(),
id.to_owned(),
receipt.to_owned(),
secs,
)),
Err(_) => Err(("seconds", "editor.error.seconds")),
}
}
fn set_view(m: &mut Model, wanted: BodyView) -> Vec<Effect> {
if let Some(mon) = &mut m.topics.monitor {
mon.messages.view = if mon.messages.view == wanted {
BodyView::Sent
} else {
wanted
};
}
Vec::new()
}
pub fn detail_key(m: &mut Model, event: ratatui::crossterm::event::KeyEvent) -> Vec<Effect> {
use ratatui::crossterm::event::KeyCode;
let action = match event.code {
KeyCode::Esc | KeyCode::Enter | KeyCode::Char('q') => {
m.modal = None;
return Vec::new();
}
KeyCode::Char('f') => return set_view(m, BodyView::Formatted),
KeyCode::Char('b') => return set_view(m, BodyView::Base64),
KeyCode::Char('v') => Action::MsgVisibility,
KeyCode::Delete | KeyCode::Char('d' | 'a') => Action::MsgAck,
_ => return Vec::new(),
};
if let Some(key) = gate(m, action) {
return refuse(m, key);
}
if action == Action::MsgVisibility {
open_visibility(m);
return Vec::new();
}
carry_out(m, Action::MsgAck)
}
fn build_publish(editor: &Editor, topic: &str) -> Result<ApiCall, (&'static str, &'static str)> {
let body = editor.field("body").map_or("", |f| f.value.as_str());
if body.trim().is_empty() {
return Err(("body", "editor.error.body"));
}
let bytes = if editor.checked("base64") {
let compact: String = body.split_whitespace().collect();
match taba_client::base64::decode(&compact) {
Some(bytes) if !bytes.is_empty() => bytes,
_ => return Err(("body", "editor.error.base64")),
}
} else {
body.as_bytes().to_vec()
};
let content_type = editor.text("content-type");
if content_type.is_empty() {
return Err(("content-type", "editor.error.content-type"));
}
let optional = |id: &str| {
Some(editor.text(id))
.filter(|s| !s.is_empty())
.map(str::to_owned)
};
Ok(ApiCall::Publish(PublishSpec {
topic: topic.to_owned(),
body: bytes,
content_type: content_type.to_owned(),
group_id: optional("group"),
deduplication_id: optional("dedup"),
}))
}
pub fn editor_key(m: &mut Model, mut editor: Editor, outcome: Outcome) -> Vec<Effect> {
match outcome {
Outcome::Cancel => {
m.modal = None;
Vec::new()
}
Outcome::Submit => {
let built = match editor.purpose.clone() {
Purpose::MessagePublish { topic } => build_publish(&editor, &topic),
Purpose::MessageReceive { topic } => build_receive(&editor, &topic),
Purpose::MessageVisibility { topic, id, receipt } => {
build_visibility(&editor, &topic, &id, &receipt)
}
_ => return Vec::new(),
};
match built {
Err((field, key)) => {
editor.error = Some(m.tx().t(key));
editor.focus_on(field);
m.modal = Some(Modal::Editor(editor));
Vec::new()
}
Ok(call) => {
editor.busy = true;
editor.error = None;
m.modal = Some(Modal::Editor(editor));
run(m, call)
}
}
}
Outcome::Test | Outcome::None => {
m.modal = Some(Modal::Editor(editor));
Vec::new()
}
}
}
fn say(m: &mut Model, severity: Severity, text: String) {
if let Some(mon) = &mut m.topics.monitor {
mon.messages.last = Some((severity, text.clone()));
}
}
pub fn published(m: &mut Model, text: &str) {
if matches!(m.modal, Some(Modal::Editor(_))) {
m.modal = None;
}
let tx = m.tx();
match api::parse_published(text) {
Ok(done) => {
let (severity, line) = if done.duplicate {
(Severity::Warning, tx.t("messages.published-duplicate"))
} else if let Some(matched) = done.matched {
(
Severity::Ok,
tx.tw("messages.published-matched", &[("n", &matched)]),
)
} else {
let id = done.id.unwrap_or_default();
(Severity::Ok, tx.tw("messages.published", &[("id", &id)]))
};
say(m, severity, line);
}
Err(e) => {
let line = tx.tw("notice.bad-reply", &[("error", &e)]);
say(m, Severity::Warning, line);
}
}
if let Some(mon) = &mut m.topics.monitor {
mon.feed.refresh_now();
}
}
pub fn received(m: &mut Model, topic: &str, hold: u32, text: &str) {
if matches!(m.modal, Some(Modal::Editor(_))) {
m.modal = None;
}
let tx = m.tx();
let now = m.now;
let messages = match api::parse_received(text) {
Ok(messages) => messages,
Err(e) => {
let line = tx.tw("notice.bad-reply", &[("error", &e)]);
return say(m, Severity::Warning, line);
}
};
let Some(mon) = &mut m.topics.monitor else {
return;
};
if mon.name != topic {
return;
}
let count = messages.len();
let fresh: Vec<Received> = messages
.into_iter()
.map(|message| Received {
size: message.body.len(),
body: String::from_utf8_lossy(&message.body)
.chars()
.take(KEPT_BODY)
.collect(),
raw: message.body.iter().take(KEPT_RAW).copied().collect(),
id: message.id,
group: message.group_id,
deliveries: message.delivery_count,
content_type: message.content_type,
receipt: message.receipt_token,
at: now,
held_until: Some(now + u64::from(hold) * 1000),
acked: false,
})
.collect();
let mut rows = fresh;
rows.truncate(MAX_RECEIVED);
mon.messages.selected = 0;
mon.messages.rows = rows;
mon.feed.refresh_now();
let line = if count == 0 {
tx.t("messages.nothing")
} else {
tx.tw("messages.received", &[("n", &count), ("secs", &hold)])
};
say(m, Severity::Ok, line);
}
pub fn acked(m: &mut Model, id: &str) {
if matches!(m.modal, Some(Modal::Message)) {
m.modal = None;
}
if let Some(mon) = &mut m.topics.monitor {
if let Some(row) = mon.messages.rows.iter_mut().find(|r| r.id == id) {
row.acked = true;
row.held_until = None;
}
mon.feed.refresh_now();
}
let line = m.tx().tw("messages.acked", &[("id", &short(id))]);
say(m, Severity::Ok, line);
}
pub fn visibility_changed(m: &mut Model, id: &str, secs: u32) {
if matches!(m.modal, Some(Modal::Message | Modal::Editor(_))) {
m.modal = None;
}
if secs == 0 {
return released(m, id);
}
let now = m.now;
if let Some(mon) = &mut m.topics.monitor {
if let Some(row) = mon.messages.rows.iter_mut().find(|r| r.id == id) {
row.held_until = Some(now + u64::from(secs) * 1000);
}
mon.feed.refresh_now();
}
let line = m.tx().tw(
"messages.visibility",
&[("id", &short(id)), ("secs", &secs)],
);
say(m, Severity::Ok, line);
}
pub fn released(m: &mut Model, id: &str) {
if let Some(mon) = &mut m.topics.monitor {
if let Some(row) = mon.messages.rows.iter_mut().find(|r| r.id == id) {
row.held_until = None;
}
mon.feed.refresh_now();
}
let line = m.tx().tw("messages.released", &[("id", &short(id))]);
say(m, Severity::Ok, line);
}
pub fn refused(m: &mut Model, id: &str, message: &str) {
if let Some(mon) = &mut m.topics.monitor {
if let Some(row) = mon.messages.rows.iter_mut().find(|r| r.id == id) {
row.held_until = None;
}
}
let line = m.tx().tw(
"messages.refused",
&[("id", &short(id)), ("error", &message)],
);
say(m, Severity::Warning, line);
}
pub fn pruned(m: &mut Model, text: &str) {
let tx = m.tx();
match api::parse_pruned(text) {
Ok(removed) => {
if let Some(mon) = &mut m.topics.monitor {
for row in &mut mon.messages.rows {
row.acked = true;
row.held_until = None;
}
mon.feed.refresh_now();
}
let shown = crate::tui::i18n::group(m.lang, removed);
let line = tx.tw("messages.pruned", &[("n", &shown)]);
say(m, Severity::Ok, line);
}
Err(e) => {
let line = tx.tw("notice.bad-reply", &[("error", &e)]);
say(m, Severity::Warning, line);
}
}
m.topics.stats.refresh_now();
}
pub fn short(id: &str) -> String {
crate::tui::model::short_id(id)
}
#[cfg(test)]
mod tests;