aleeve.dev
Scalable and distributed custom feed generator, ott - on that topic
Key like records as well
+5
-4
README.md
+5
-4
README.md
···
9
## Install dependencies
10
11
```shell
12
+
# Install k8s tooling
13
+
brew install kind helm skaffold
14
15
+
# Install fvm and fluvio cli
16
+
curl -fsS https://hub.infinyon.cloud/install/install.sh | bash
17
18
+
# Install rust
19
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
20
```
21
+5
connectors/likes-config.yaml
+5
connectors/likes-config.yaml
+17
-15
crates/ott-ingest/src/main.rs
+17
-15
crates/ott-ingest/src/main.rs
···
32
#[derive(Debug, Deserialize, Clone)]
33
struct Like {
34
did: String,
35
-
#[serde(rename = "commit.record.subject.uri")]
36
uri: String,
37
}
38
···
89
});
90
},
91
Some(Ok(record)) = like_stream.next() => {
92
-
let like: Like = serde_json::from_slice(record.value()).unwrap();
93
-
lcc.entry(like.uri)
94
-
.and_compute_with(|maybe_entry| {
95
-
if let Some(entry) = maybe_entry {
96
-
let mut post = entry.into_value();
97
-
if post.count < 2 {
98
-
post.count +=1;
99
-
warn!("Incread counter for {:#?}", post);
100
-
Op::Put(post)
101
} else {
102
-
Op::Remove
103
}
104
-
} else {
105
-
Op::Nop // Skip as post is out of cache
106
-
}
107
-
});
108
}
109
}
110
}
···
32
#[derive(Debug, Deserialize, Clone)]
33
struct Like {
34
did: String,
35
uri: String,
36
}
37
···
88
});
89
},
90
Some(Ok(record)) = like_stream.next() => {
91
+
if let Ok(like) = serde_json::from_slice::<Like>(record.value()) {
92
+
lcc.entry(like.uri)
93
+
.and_compute_with(|maybe_entry| {
94
+
if let Some(entry) = maybe_entry {
95
+
let mut post = entry.into_value();
96
+
if post.count < 20 {
97
+
post.count +=1;
98
+
warn!("Incread counter for {:#?}", post);
99
+
Op::Put(post)
100
+
} else {
101
+
Op::Remove
102
+
}
103
} else {
104
+
Op::Nop // Skip as post is out of cache
105
}
106
+
});
107
+
} else {
108
+
warn!("Failed deserializing, likely not like commit");
109
+
};
110
}
111
}
112
}
+35
-13
smart-modules/construct-post-uri/src/lib.rs
+35
-13
smart-modules/construct-post-uri/src/lib.rs
···
23
}
24
25
fn get_uri(obj: &Map<String, Value>) -> Result<String> {
26
-
let did = obj
27
-
.get("did")
28
.and_then(|v| v.as_str())
29
-
.ok_or(eyre!("did missing or not a string"))?;
30
31
-
let commit = obj.get("commit").ok_or(eyre!("commit missing"))?;
32
33
-
let collection = commit
34
-
.get("collection")
35
-
.and_then(|v| v.as_str())
36
-
.ok_or(eyre!("commit.collection missing or not a string"))?;
37
38
-
let rkey = commit
39
-
.get("rkey")
40
-
.and_then(|v| v.as_str())
41
-
.ok_or(eyre!("commit.rkey missing or not a string"))?;
42
43
-
Ok(format!("at://{did}/{collection}/{rkey}"))
44
}
···
23
}
24
25
fn get_uri(obj: &Map<String, Value>) -> Result<String> {
26
+
let collection = obj
27
+
.get("commit")
28
+
.and_then(|v| v.get("collection"))
29
.and_then(|v| v.as_str())
30
+
.ok_or(eyre!("Missing commit.collection"))?;
31
+
match collection {
32
+
"app.bsky.feed.post" => {
33
+
let did = obj
34
+
.get("did")
35
+
.and_then(|v| v.as_str())
36
+
.ok_or(eyre!("did missing or not a string"))?;
37
38
+
let commit = obj.get("commit").ok_or(eyre!("commit missing"))?;
39
40
+
let collection = commit
41
+
.get("collection")
42
+
.and_then(|v| v.as_str())
43
+
.ok_or(eyre!("commit.collection missing or not a string"))?;
44
45
+
let rkey = commit
46
+
.get("rkey")
47
+
.and_then(|v| v.as_str())
48
+
.ok_or(eyre!("commit.rkey missing or not a string"))?;
49
50
+
Ok(format!("at://{did}/{collection}/{rkey}"))
51
+
},
52
+
"app.bsky.feed.like" => {
53
+
let uri = obj
54
+
.get("commit")
55
+
.and_then(|v| v.get("record"))
56
+
.and_then(|v| v.get("subject"))
57
+
.and_then(|v| v.get("uri"))
58
+
.and_then(|v| v.as_str())
59
+
.ok_or(eyre!("Likely not commit message"))?;
60
+
Ok(uri.to_string())
61
+
}
62
+
&_ => {
63
+
Err(eyre!("Not supported yet"))
64
+
}
65
+
}
66
}