use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
pub const CONFIG_KEY_PREFIX: &str = "config:";
pub const PENDING_ACTIONS_KEY: &str = "pending_actions";
pub const META_NONCE_KEY: &str = "__meta_nonce__";
pub const META_UPDATED_AT_KEY: &str = "__meta_updated_at__";
pub type DataMap = HashMap<String, String>;
pub fn config_key(widget_id: &str) -> String {
format!("{CONFIG_KEY_PREFIX}{widget_id}")
}
pub const INSTANCE_SEPARATOR: char = '#';
pub fn scoped_id(widget_id: &str, instance: Option<&str>) -> String {
match instance.filter(|i| !i.is_empty()) {
Some(i) => format!("{widget_id}{INSTANCE_SEPARATOR}{i}"),
None => widget_id.to_string(),
}
}
pub const DATA_KEY_PREFIX: &str = "data:";
pub const FETCHED_KEY_PREFIX: &str = "fetched:";
pub const INSTANCES_KEY_PREFIX: &str = "instances:";
pub fn data_key(widget_id: &str, instance: Option<&str>) -> String {
format!("{DATA_KEY_PREFIX}{}", scoped_id(widget_id, instance))
}
pub const STATE_KEY_PREFIX: &str = "state:";
pub const ACTIVITY_TEMPLATE_KEY_PREFIX: &str = "activity-template:";
pub const STATE_ACTION: &str = "$state";
pub fn state_event(env: &WidgetActionEnvelope) -> Option<serde_json::Value> {
use serde_json::Value;
if env.action != STATE_ACTION {
return None;
}
let p: Value = env
.payload
.as_deref()
.and_then(|p| serde_json::from_str(p).ok())
.unwrap_or_default();
Some(serde_json::json!({
"group": env.group,
"widgetId": p.get("widgetId").cloned().unwrap_or(Value::String(env.widget_id.clone())),
"instance": p.get("instance").cloned().unwrap_or(Value::Null),
"state": p.get("state").cloned().unwrap_or(Value::Null),
}))
}
pub fn state_key(widget_id: &str, instance: Option<&str>) -> String {
format!("{STATE_KEY_PREFIX}{}", scoped_id(widget_id, instance))
}
pub fn activity_template_key(name: &str) -> String {
format!("{ACTIVITY_TEMPLATE_KEY_PREFIX}{name}")
}
pub fn merged_state(
layers: &[Option<&serde_json::Value>],
) -> serde_json::Map<String, serde_json::Value> {
let mut out = serde_json::Map::new();
for layer in layers.iter().flatten() {
if let Some(m) = layer.as_object() {
out.extend(m.iter().map(|(k, v)| (k.clone(), v.clone())));
}
}
out
}
pub fn apply_state_patch(
layer: &mut serde_json::Map<String, serde_json::Value>,
current: &serde_json::Map<String, serde_json::Value>,
patch: &serde_json::Map<String, serde_json::Value>,
) {
use serde_json::Value;
for (k, v) in patch {
let now = current.get(k);
let step = |d: i64| -> Value {
match now {
Some(Value::Number(n)) => n
.as_i64()
.and_then(|i| i.checked_add(d))
.map(Value::from)
.unwrap_or_else(|| Value::from(n.as_f64().unwrap_or(0.0) + d as f64)),
_ => Value::from(d),
}
};
let next = match v {
Value::Null => {
layer.remove(k);
continue;
}
Value::String(op) if op == "$toggle" => {
Value::Bool(!matches!(now, Some(Value::Bool(true))))
}
Value::String(op) if op == "$inc" => step(1),
Value::String(op) if op == "$dec" => step(-1),
other => other.clone(),
};
layer.insert(k.clone(), next);
}
}
pub fn instances_key(widget_id: &str) -> String {
format!("{INSTANCES_KEY_PREFIX}{widget_id}")
}
pub fn lookup_keys(prefix: &str, widget_id: &str, instance: Option<&str>) -> Vec<String> {
let mut keys = Vec::with_capacity(2);
if instance.is_some_and(|i| !i.is_empty()) {
keys.push(format!("{prefix}{}", scoped_id(widget_id, instance)));
}
keys.push(format!("{prefix}{widget_id}"));
keys
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
pub struct WidgetInstanceOption {
pub id: String,
pub title: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ActiveWidget {
pub widget_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub instance: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub size: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_seen_ms: Option<u64>,
}
pub fn merge_active(live: Vec<ActiveWidget>, seen: Vec<ActiveWidget>) -> Vec<ActiveWidget> {
let mut out = live;
for s in seen {
match out
.iter_mut()
.find(|a| a.widget_id == s.widget_id && a.instance == s.instance)
{
Some(a) => {
a.last_seen_ms = a.last_seen_ms.or(s.last_seen_ms);
a.size = a.size.take().or(s.size);
}
None => out.push(s),
}
}
out.sort_by(|a, b| (&a.widget_id, &a.instance).cmp(&(&b.widget_id, &b.instance)));
out
}
pub fn merge_native_active(
live: Vec<ActiveWidget>,
seen: Vec<ActiveWidget>,
exact: bool,
) -> Vec<ActiveWidget> {
if exact {
let mut merged = merge_active(live.clone(), seen);
merged.retain(|a| {
live.iter()
.any(|l| l.widget_id == a.widget_id && l.instance == a.instance)
});
return merged;
}
let mut out: Vec<ActiveWidget> = Vec::new();
for l in &live {
let instances: Vec<&ActiveWidget> = seen
.iter()
.filter(|s| s.widget_id == l.widget_id && s.instance.is_some())
.collect();
if instances.is_empty() {
let receipt = seen.iter().find(|s| s.widget_id == l.widget_id);
out.push(ActiveWidget {
last_seen_ms: receipt.and_then(|r| r.last_seen_ms),
..l.clone()
});
} else {
for s in instances {
if !out.contains(s) {
out.push(s.clone());
}
}
}
}
out.sort_by(|a, b| {
(&a.widget_id, &a.instance, &a.size).cmp(&(&b.widget_id, &b.instance, &b.size))
});
out
}
pub const BOARD_WIDGETS_KEY: &str = "board_widgets";
pub fn parse_board_widgets(raw: &str) -> Vec<ActiveWidget> {
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Board {
widget_id: String,
instance: String,
}
serde_json::from_str::<Vec<Board>>(raw)
.unwrap_or_default()
.into_iter()
.map(|b| ActiveWidget {
widget_id: b.widget_id,
instance: Some(b.instance),
size: Some("medium".into()),
last_seen_ms: None,
})
.collect()
}
pub const ACTIVE_RECEIPT_MAX_AGE_MS: u64 = 48 * 60 * 60 * 1000;
pub fn active_from_receipts(
receipts: &[crate::receipt::WidgetRenderReceipt],
now_ms: u64,
max_age_ms: u64,
) -> Vec<ActiveWidget> {
let mut best: HashMap<(String, String), &crate::receipt::WidgetRenderReceipt> = HashMap::new();
for r in receipts {
if now_ms.saturating_sub(r.ts) > max_age_ms {
continue;
}
let key = (r.widget_id.clone(), r.instance.clone());
match best.get(&key) {
Some(prev) if prev.ts >= r.ts => {}
_ => {
best.insert(key, r);
}
}
}
let mut out: Vec<ActiveWidget> = best
.into_values()
.map(|r| ActiveWidget {
widget_id: r.widget_id.clone(),
instance: Some(r.instance.clone()).filter(|i| !i.is_empty()),
size: r.size.clone(),
last_seen_ms: Some(r.ts),
})
.collect();
out.sort_by(|a, b| (&a.widget_id, &a.instance).cmp(&(&b.widget_id, &b.instance)));
out
}
pub fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
pub fn map_nonce(map: &DataMap) -> u64 {
map.get(META_NONCE_KEY)
.and_then(|s| s.parse().ok())
.unwrap_or(0)
}
pub fn map_updated_at(map: &DataMap) -> u64 {
map.get(META_UPDATED_AT_KEY)
.and_then(|s| s.parse().ok())
.unwrap_or(0)
}
pub fn touch_meta(map: &mut DataMap) {
let next = map_nonce(map).saturating_add(1);
map.insert(META_NONCE_KEY.into(), next.to_string());
map.insert(META_UPDATED_AT_KEY.into(), now_ms().to_string());
}
pub fn touch_meta_above(map: &mut DataMap, floor: u64) {
let next = map_nonce(map)
.saturating_add(1)
.max(floor.saturating_add(1));
map.insert(META_NONCE_KEY.into(), next.to_string());
map.insert(META_UPDATED_AT_KEY.into(), now_ms().to_string());
}
pub fn map_has_config(map: &DataMap) -> bool {
map.keys().any(|k| k.starts_with(CONFIG_KEY_PREFIX))
}
pub fn pick_freshest(maps: impl IntoIterator<Item = DataMap>) -> DataMap {
let mut best: Option<DataMap> = None;
let mut best_has_config = false;
let mut best_nonce = 0u64;
let mut best_ts = 0u64;
for map in maps {
let has_config = map_has_config(&map);
let n = map_nonce(&map);
let t = map_updated_at(&map);
let better = match &best {
None => true,
Some(_) if has_config && !best_has_config => true,
Some(_) if !has_config && best_has_config => false,
Some(_) => n > best_nonce || (n == best_nonce && t > best_ts),
};
if better {
best_has_config = has_config;
best_nonce = n;
best_ts = t;
best = Some(map);
}
}
best.unwrap_or_default()
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct WidgetActionEnvelope {
pub action: String,
#[serde(default)]
pub payload: Option<String>,
pub ts: u64,
pub widget_id: String,
pub group: String,
}
impl WidgetActionEnvelope {
pub fn new(
action: impl Into<String>,
payload: Option<String>,
widget_id: impl Into<String>,
group: impl Into<String>,
) -> Self {
Self {
action: action.into(),
payload,
ts: now_ms(),
widget_id: widget_id.into(),
group: group.into(),
}
}
}
#[derive(Debug)]
pub struct ListenerGate {
ready: std::sync::atomic::AtomicBool,
started_ms: u64,
fallback_ms: u64,
}
impl ListenerGate {
pub const DEFAULT_FALLBACK_MS: u64 = 10_000;
pub fn new(now_ms: u64, fallback_ms: u64) -> Self {
Self {
ready: std::sync::atomic::AtomicBool::new(false),
started_ms: now_ms,
fallback_ms,
}
}
pub fn open(&self) {
self.ready.store(true, std::sync::atomic::Ordering::SeqCst);
}
pub fn remaining_ms(&self, now_ms: u64) -> u64 {
if self.is_open(now_ms) {
0
} else {
(self.started_ms + self.fallback_ms).saturating_sub(now_ms)
}
}
pub fn is_open(&self, now_ms: u64) -> bool {
self.ready.load(std::sync::atomic::Ordering::SeqCst)
|| now_ms.saturating_sub(self.started_ms) >= self.fallback_ms
}
}
pub fn parse_pending_actions(raw: Option<&str>) -> Vec<WidgetActionEnvelope> {
let Some(s) = raw.filter(|s| !s.is_empty()) else {
return Vec::new();
};
serde_json::from_str(s).unwrap_or_default()
}
pub fn pending_minus_drained(
disk_raw: Option<&str>,
drained: &[WidgetActionEnvelope],
) -> crate::Result<String> {
let rest: Vec<_> = parse_pending_actions(disk_raw)
.into_iter()
.filter(|a| !drained.contains(a))
.collect();
encode_pending_actions(&rest)
}
pub fn encode_pending_actions(actions: &[WidgetActionEnvelope]) -> crate::Result<String> {
Ok(serde_json::to_string(actions)?)
}
#[cfg(test)]
mod tests {
use super::*;
fn receipt(id: &str, inst: &str, ts: u64) -> crate::receipt::WidgetRenderReceipt {
crate::receipt::WidgetRenderReceipt {
widget_id: id.into(),
group: "g".into(),
instance: inst.into(),
nonce: 0,
size: Some("small".into()),
theme: None,
schema: 1,
source: "prefs".into(),
trigger: None,
rendered: vec![],
skipped: vec![],
ts,
}
}
fn aw(id: &str, inst: Option<&str>, size: Option<&str>, seen: Option<u64>) -> ActiveWidget {
ActiveWidget {
widget_id: id.into(),
instance: inst.map(Into::into),
size: size.map(Into::into),
last_seen_ms: seen,
}
}
#[test]
fn lookup_keys_scenario() {
assert_eq!(scoped_id("w", Some("")), "w");
assert_eq!(config_key("w"), "config:w");
assert_eq!(
lookup_keys(CONFIG_KEY_PREFIX, "w", Some("1")),
["config:w#1", "config:w"]
);
assert_eq!(lookup_keys(CONFIG_KEY_PREFIX, "w", Some("")), ["config:w"]);
assert_eq!(lookup_keys(FETCHED_KEY_PREFIX, "w", None), ["fetched:w"]);
}
#[test]
fn placement_listing_scenario() {
let rs = [
receipt("w", "2", 900),
receipt("w", "1", 500),
receipt("w", "1", 800),
receipt("w", "1", 700),
receipt("old", "", 99),
receipt("a", "", 1000),
];
let got = active_from_receipts(&rs, 1000, 900);
assert_eq!(
got,
[
ActiveWidget {
size: Some("small".into()),
..aw("a", None, None, Some(1000))
},
ActiveWidget {
size: Some("small".into()),
..aw("w", Some("1"), None, Some(800))
},
ActiveWidget {
size: Some("small".into()),
..aw("w", Some("2"), None, Some(900))
},
]
);
assert_eq!(
active_from_receipts(&[receipt("x", "", 100)], 1000, 900).len(),
1
);
let h = 60 * 60 * 1000;
let rs = [
receipt("fresh", "", 100 * h - 47 * h),
receipt("stale", "", 100 * h - 49 * h),
];
let got = active_from_receipts(&rs, 100 * h, ACTIVE_RECEIPT_MAX_AGE_MS);
assert_eq!(
got.iter().map(|a| a.widget_id.as_str()).collect::<Vec<_>>(),
["fresh"]
);
let got = merge_active(
vec![
aw("w", Some("win"), None, None),
aw("b", None, Some("large"), None),
],
vec![
aw("w", Some("win"), Some("small"), Some(5)),
aw("a", None, None, Some(1)),
aw("b", None, Some("small"), Some(2)),
],
);
assert_eq!(
got,
vec![
aw("a", None, None, Some(1)),
aw("b", None, Some("large"), Some(2)),
aw("w", Some("win"), Some("small"), Some(5)),
]
);
let got = merge_native_active(
vec![aw("w", Some("7"), None, None)],
vec![
aw("w", Some("7"), Some("small"), Some(3)),
aw("w", Some("8"), None, Some(4)),
],
true,
);
assert_eq!(got, vec![aw("w", Some("7"), Some("small"), Some(3))]);
let got = merge_native_active(
vec![
aw("w", None, Some("small"), None),
aw("w", None, Some("medium"), None),
aw("plain", None, Some("large"), None),
],
vec![
aw("w", Some("nyc"), Some("small"), Some(5)),
aw("w", Some("ldn"), Some("medium"), Some(6)),
aw("plain", None, Some("large"), Some(9)),
aw("gone", Some("x"), None, Some(1)),
],
false,
);
assert_eq!(
got,
vec![
aw("plain", None, Some("large"), Some(9)),
aw("w", Some("ldn"), Some("medium"), Some(6)),
aw("w", Some("nyc"), Some("small"), Some(5)),
]
);
assert_eq!(
merge_native_active(vec![aw("n", None, None, None)], vec![], false),
vec![aw("n", None, None, None)]
);
assert_eq!(
parse_board_widgets(r#"[{"widgetId":"default","instance":"B1"}]"#),
vec![aw("default", Some("B1"), Some("medium"), None)]
);
assert!(parse_board_widgets("nope").is_empty());
}
#[test]
fn transport_freshness_scenario() {
let m = |tag: &str, config: bool, nonce: u64, ts: u64| {
let mut d = DataMap::new();
d.insert("tag".into(), tag.into());
if config {
d.insert("config:w".into(), "{}".into());
}
d.insert(META_NONCE_KEY.into(), nonce.to_string());
d.insert(META_UPDATED_AT_KEY.into(), ts.to_string());
d
};
let pick = |maps: Vec<DataMap>| pick_freshest(maps)["tag"].clone();
let before = now_ms();
assert!(before > 1_600_000_000_000, "now_ms is Unix ms");
let mut live = m("live", true, 7, 0);
touch_meta(&mut live);
assert_eq!(map_nonce(&live), 8);
assert!(map_updated_at(&live) >= before);
assert_eq!(map_updated_at(&m("x", false, 0, 123)), 123);
assert_eq!(
(map_nonce(&DataMap::new()), map_updated_at(&DataMap::new())),
(0, 0)
);
assert!(map_has_config(&live) && !map_has_config(&m("p", false, 0, 0)));
let probe = m("probe", false, 99, u64::MAX);
assert_eq!(pick(vec![probe.clone(), live.clone()]), "live");
assert_eq!(pick(vec![live.clone(), probe]), "live");
let stale = m("stale", true, 65, 0);
assert_eq!(pick(vec![live.clone(), stale.clone()]), "stale");
touch_meta_above(&mut live, 65);
assert_eq!(map_nonce(&live), 66);
assert_eq!(pick(vec![stale, live]), "live");
assert_eq!(pick(vec![m("a", false, 1, 9), m("b", false, 2, 0)]), "b");
assert_eq!(pick(vec![m("a", true, 1, 9), m("b", true, 2, 0)]), "b");
assert_eq!(pick(vec![m("a", false, 2, 0), m("b", false, 1, 9)]), "a");
assert_eq!(pick(vec![m("a", false, 2, 1), m("b", false, 2, 5)]), "b");
assert_eq!(pick(vec![m("a", false, 2, 5), m("b", false, 2, 1)]), "a");
assert_eq!(pick(vec![m("a", false, 2, 5), m("b", false, 2, 5)]), "a");
assert!(pick_freshest(Vec::new()).is_empty());
}
#[test]
fn action_queue_scenario() {
let g = ListenerGate::new(1_000, 500);
assert!(!g.is_open(1_000));
assert!(!g.is_open(1_499));
assert!(g.is_open(1_500));
assert!(!g.is_open(0));
assert_eq!(g.remaining_ms(1_200), 300);
assert_eq!(g.remaining_ms(1_500), 0);
g.open();
assert_eq!(g.remaining_ms(1_000), 0);
assert!(g.is_open(1_000));
let a = WidgetActionEnvelope::new("a", None, "w", "g");
let b = WidgetActionEnvelope::new("b", Some("1".into()), "w", "g");
let disk = encode_pending_actions(&[a.clone(), b.clone()]).unwrap();
let left = pending_minus_drained(Some(&disk), &[a]).unwrap();
assert_eq!(parse_pending_actions(Some(&left)), vec![b]);
assert_eq!(pending_minus_drained(None, &[]).unwrap(), "[]");
assert!(parse_pending_actions(Some("junk")).is_empty());
}
}