1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
use abi::repo::{create, delete_record, list_records};
use abi::{AtUri, Call, Datetime, Did, Fan, Handle, LexiconType, On, Run, Step, entries};
use serde::{Deserialize, Serialize};
use serde_json::json;
const COLLECTION: &str = "xyz.statusphere.status";
const CLEAR_LIMIT: usize = 10;
#[derive(Deserialize)]
struct Args {
did: Did,
now: String,
#[serde(default)]
status: Option<String>,
}
#[derive(Serialize, PartialEq)]
struct Status {
status: String,
#[serde(rename = "createdAt")]
created_at: Datetime,
}
impl LexiconType for Status {
fn lexicon_type() -> &'static str {
COLLECTION
}
}
fn validate_status(status: Option<&str>) -> Result<String, String> {
let status = status.unwrap_or_default();
if status.is_empty() || status.len() > 32 {
return Err("status must be 1 to 32 bytes".into());
}
if status.chars().any(char::is_control) {
return Err("status must not contain control characters".into());
}
Ok(status.to_string())
}
abi::call_ids! {
#[derive(Clone, Copy)]
enum Id {
List = "list",
Create = "create",
Delete(usize) = "delete",
}
}
fn list_statuses(did: &Did, limit: usize) -> Call {
list_records(did.as_str(), COLLECTION, limit, None)
}
#[derive(Default, Deserialize, Serialize)]
struct SetStatus {
status: String,
}
// list the latest status -> create one unless it already matches -> report the uri.
fn set_status_handler(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<Args, SetStatus, Id>::new(run)?;
let args = &step.args;
match on {
On::Start => {
step.state.status = validate_status(args.status.as_deref())?;
Datetime::new(args.now.clone()).map_err(|e| format!("invalid now: {e}"))?;
step.out.call(Id::List, list_statuses(&args.did, 1));
}
On::Result(Id::List, result) => {
let status = validate_status(Some(&step.state.status))?;
let latest = &result?["records"][0];
if latest["value"]["status"].as_str() == Some(status.as_str()) {
step.out.finish(json!({ "status": status, "unchanged": true, "uri": latest["uri"] }));
return step.handle();
}
let now = Datetime::new(args.now.clone()).map_err(|e| format!("invalid now: {e}"))?;
let created_at = Datetime::try_from(now.to_utc()).map_err(|e| format!("invalid now: {e}"))?;
step.out.call(Id::Create, create(args.did.as_str(), Status { status, created_at }));
}
On::Result(Id::Create, result) => step.out.finish(json!({ "status": step.state.status, "unchanged": false, "uri": result?["uri"] })),
_ => {}
}
step.handle()
}
#[derive(Default, Deserialize, Serialize)]
struct ClearStatus {
deletes: Fan,
deleted: u64,
failed: u64,
}
// list up to CLEAR_LIMIT statuses -> delete each (results arrive one by one) -> count.
fn clear_status_handler(run: Run<Args>) -> Result<Handle, String> {
let (mut step, on) = Step::<Args, ClearStatus, Id>::new(run)?;
let did = step.args.did.as_str();
match on {
On::Start => step.out.call(Id::List, list_statuses(&step.args.did, CLEAR_LIMIT)),
On::Result(Id::List, result) => {
let output = result?;
let rkeys: Vec<String> = output["records"].as_array().into_iter().flatten().filter_map(|record| Some(record["uri"].as_str()?.parse::<AtUri>().ok()?.rkey)).take(CLEAR_LIMIT).collect();
if rkeys.is_empty() {
step.out.finish(json!({ "deleted": 0 }));
return step.handle();
}
let deletes = rkeys.iter().enumerate().map(|(i, rkey)| (Id::Delete(i), delete_record(&AtUri::new(did, COLLECTION, rkey))));
step.out.fan_out(&mut step.state.deletes, deletes);
}
On::Result(id @ Id::Delete(_), result) => {
let s = &mut step.state;
if result.is_ok() { s.deleted += 1 } else { s.failed += 1 }
if s.deletes.land(&id, result).is_some() {
step.out.finish(json!({ "deleted": s.deleted, "failed": s.failed }));
}
}
_ => {}
}
step.handle()
}
entries!(
set_status => set_status_handler,
clear_status => clear_status_handler
);
