pds.dad
What if there was a feed with more than Bluesky lexicons and it was personalized to you?
news.modelo.social
Got the cache going
author
Bailey Townsend
date
(May 12, 2026, 10:29 AM -0700)
commit
c3901580
c3901580ac617d0e76082ffae28bd57836650cad
parent
17866e84
17866e84b0d89ec9619025b5f4d8d59c79ae5e27
+204
-5
+19
-1
TODO.md
+19
-1
TODO.md
···
1
1
- Bring in the svelte project
2
2
- Set it up
3
3
- try the getFeedSkelton
4
-
- Make sure claude is setup with tailwindcss and daisyuui's mcp
5
4
- Make a cache wrapper for fjall
6
5
- Bring in sqlx
6
+
7
+
8
+
# How Feeds work
9
+
- https://selfhosted.social/xrpc/app.bsky.feed.getFeed?feed=at%3A%2F%2Fdid%3Aplc%3A3guzzweuqraryl3rdkimjamk%2Fapp.bsky.feed.generator%2Ffor-you&cursor=tMX2oQeGpgLc1gGIqAPkUszdAdgSgrwD9EFmroABjkeGPfB2%2FBD2rAHYHej7AY6RAcwB1DfgC8a3AsYo%2FPIBtkioLJSzAdJIqjE%3D_21600&limit=30
10
+
11
+
12
+
- getPref to get feeds (can hard code for now)
13
+
- getFeedGenerator to get details about the feed
14
+
15
+
16
+
//What the end feed looks like
17
+
https://api.graze.social/xrpc/app.bsky.feed.getFeedSkeleton?feed=at://did:plc:rnpkyqnmsw4ipey6eotbdnnf/app.bsky.feed.generator/Cornbread
18
+
19
+
20
+
# What happens when someone logs in
21
+
- When the user logs in for the first time we pull their repo and get everyone they follow
22
+
- Should probably show a "loading" spinner while we fetch the feed
23
+
- When we see a record go by that the user follows we add it to the sqlx database
24
+
- when the user loads the feed we load in 2 feeds, a bsky one they select and one from our system which is mixed
+1
api/Cargo.lock
+1
api/Cargo.lock
+4
-1
api/Cargo.toml
+4
-1
api/Cargo.toml
···
11
11
fjall = "3.1.4"
12
12
futures = "0.3"
13
13
jacquard = "0.11.0"
14
-
jacquard-api = { version = "0.11.1", features = ["default", "app_bsky", "streaming"] }
14
+
jacquard-api = { version = "0.11.1", features = ["default", "app_bsky", "streaming", "sh_tangled"] }
15
15
jacquard-axum = "0.11.0"
16
16
jacquard-common = { version = "0.11.0", features = ["websocket", "streaming", "reqwest-client"] }
17
17
jacquard-derive = "0.11.0"
···
27
27
tokio-util = { version = "0.7", features = ["rt"] }
28
28
tower-http = { version = "0.6.8", features = ["cors"] }
29
29
url = "2"
30
+
31
+
[dev-dependencies]
32
+
tempfile = "3"
+180
-3
api/src/storage/cache.rs
+180
-3
api/src/storage/cache.rs
···
1
+
use fjall::compaction::filter::{
2
+
CompactionFilter, CompactionFilterResult, Context, Factory, ItemAccessor, Verdict,
3
+
};
4
+
use jacquard::deps::chrono::{self, DateTime, Utc};
5
+
use serde::{Deserialize, Serialize, de::DeserializeOwned};
1
6
use std::{path::Path, sync::Arc};
2
7
3
-
use crate::storage::error::StorageResult;
8
+
use crate::storage::error::{StorageError, StorageResult};
9
+
10
+
const CACHE_KEYSPACE: &str = "cache";
4
11
5
12
pub struct Cache {
6
13
pub(crate) database: fjall::Database,
···
11
18
pub type CacheRef = Arc<Cache>;
12
19
13
20
/// Open (or create) the fjall database at `path` and return a shared handle.
21
+
///
22
+
/// Registers a TTL compaction filter on the `cache` keyspace so expired
23
+
/// entries are reclaimed during background compactions.
14
24
pub fn open(path: &Path) -> StorageResult<CacheRef> {
15
-
let database = fjall::Database::builder(path).open()?;
16
-
let cache_keyspace = database.keyspace("cache", fjall::KeyspaceCreateOptions::default)?;
25
+
let database = fjall::Database::builder(path)
26
+
.with_compaction_filter_factories(Arc::new(|name| {
27
+
(name == CACHE_KEYSPACE).then(|| Arc::new(CacheFilterFactory) as Arc<dyn Factory>)
28
+
}))
29
+
.open()?;
30
+
let cache_keyspace =
31
+
database.keyspace(CACHE_KEYSPACE, fjall::KeyspaceCreateOptions::default)?;
17
32
Ok(Arc::new(Cache {
18
33
database,
19
34
cache_keyspace,
20
35
}))
21
36
}
37
+
38
+
impl Cache {
39
+
/// Insert `value` under `key`, expiring `ttl` from now.
40
+
pub fn insert<K: AsRef<[u8]>, T: Serialize>(
41
+
&self,
42
+
key: K,
43
+
value: &T,
44
+
ttl: chrono::Duration,
45
+
) -> StorageResult<()> {
46
+
let data = postcard::to_allocvec(value)?;
47
+
let entry = CacheEntry {
48
+
expires_at: Utc::now() + ttl,
49
+
data,
50
+
};
51
+
let bytes = postcard::to_allocvec(&entry)?;
52
+
self.cache_keyspace.insert(key.as_ref(), bytes)?;
53
+
Ok(())
54
+
}
55
+
56
+
/// Fetch and decode `key`. Returns `Ok(None)` if missing or expired;
57
+
/// expired entries are removed eagerly on read so callers don't see
58
+
/// stale values between compactions.
59
+
pub fn get<K: AsRef<[u8]>, T: DeserializeOwned>(&self, key: K) -> StorageResult<Option<T>> {
60
+
let Some(raw) = self.cache_keyspace.get(key.as_ref())? else {
61
+
return Ok(None);
62
+
};
63
+
let entry: CacheEntry = postcard::from_bytes(&raw).map_err(|_| StorageError::Corrupt {
64
+
key: String::from_utf8_lossy(key.as_ref()).into_owned(),
65
+
reason: "cache wrapper postcard decode failed",
66
+
})?;
67
+
if Utc::now() >= entry.expires_at {
68
+
self.cache_keyspace.remove(key.as_ref())?;
69
+
return Ok(None);
70
+
}
71
+
let value = postcard::from_bytes::<T>(&entry.data)?;
72
+
Ok(Some(value))
73
+
}
74
+
75
+
pub fn remove<K: AsRef<[u8]>>(&self, key: K) -> StorageResult<()> {
76
+
self.cache_keyspace.remove(key.as_ref())?;
77
+
Ok(())
78
+
}
79
+
}
80
+
81
+
#[derive(Debug, Deserialize, Serialize)]
82
+
struct CacheEntry {
83
+
expires_at: DateTime<Utc>,
84
+
data: Vec<u8>,
85
+
}
86
+
87
+
struct CacheFilter;
88
+
89
+
impl CompactionFilter for CacheFilter {
90
+
fn filter_item(&mut self, item: ItemAccessor<'_>, _ctx: &Context) -> CompactionFilterResult {
91
+
let bytes = item.value()?;
92
+
93
+
let Ok(entry) = postcard::from_bytes::<CacheEntry>(&bytes) else {
94
+
return Ok(Verdict::Keep);
95
+
};
96
+
97
+
if Utc::now() >= entry.expires_at {
98
+
Ok(Verdict::Remove)
99
+
} else {
100
+
Ok(Verdict::Keep)
101
+
}
102
+
}
103
+
}
104
+
105
+
struct CacheFilterFactory;
106
+
107
+
impl Factory for CacheFilterFactory {
108
+
fn name(&self) -> &str {
109
+
"cache-ttl"
110
+
}
111
+
112
+
fn make_filter(&self, _ctx: &Context) -> Box<dyn CompactionFilter> {
113
+
Box::new(CacheFilter)
114
+
}
115
+
}
116
+
117
+
#[cfg(test)]
118
+
mod tests {
119
+
120
+
use super::*;
121
+
use std::{thread::sleep, time::Duration};
122
+
123
+
#[derive(Debug, Clone, Serialize, Deserialize)]
124
+
struct ATest {
125
+
test: String,
126
+
}
127
+
128
+
#[test]
129
+
fn ttl_eviction_on_read() {
130
+
let dir = tempfile::tempdir().expect("tempdir");
131
+
let cache = open(dir.path()).expect("open");
132
+
133
+
cache
134
+
.insert(
135
+
"greeting",
136
+
&"hello".to_string(),
137
+
chrono::Duration::milliseconds(50),
138
+
)
139
+
.expect("insert");
140
+
141
+
let immediate: Option<String> = cache.get("greeting").expect("get");
142
+
assert_eq!(immediate.as_deref(), Some("hello"));
143
+
144
+
sleep(Duration::from_millis(120));
145
+
146
+
let expired: Option<String> = cache.get("greeting").expect("get after expiry");
147
+
assert!(expired.is_none(), "expected entry to be evicted after TTL");
148
+
149
+
assert!(
150
+
cache
151
+
.cache_keyspace
152
+
.get("greeting")
153
+
.expect("raw get")
154
+
.is_none(),
155
+
"read-path eviction should have removed the underlying key",
156
+
);
157
+
158
+
let test = ATest {
159
+
test: "hello".to_string(),
160
+
};
161
+
162
+
// let serialized = serde_json::to_string(&test).expect("serialize");
163
+
cache
164
+
.insert("test", &test, chrono::Duration::milliseconds(50))
165
+
.expect("insert");
166
+
167
+
let retrieved: Option<ATest> = cache.get("test").expect("get");
168
+
println!("{retrieved:?}");
169
+
assert!(retrieved.is_some(), "expected entry to be retrieved");
170
+
171
+
let tangled_repo = sh_tangled::repo::Repo {
172
+
created_at: jacquard_common::types::string::Datetime::now(),
173
+
description: None,
174
+
knot: "Test".into(),
175
+
labels: None,
176
+
name: "Test".into(),
177
+
source: None,
178
+
spindle: None,
179
+
topics: None,
180
+
website: None,
181
+
extra_data: None,
182
+
};
183
+
let serialized = serde_json::to_string(&tangled_repo).expect("serialize");
184
+
cache
185
+
.insert(
186
+
"tangled_repo",
187
+
&serialized,
188
+
chrono::Duration::milliseconds(50),
189
+
)
190
+
.expect("insert");
191
+
192
+
let retrieved: Option<String> = cache.get("tangled_repo").expect("get");
193
+
let idk = retrieved.unwrap();
194
+
let deserialized: Option<sh_tangled::repo::Repo> = serde_json::from_str(&idk).ok();
195
+
println!("{deserialized:?}");
196
+
// assert!(retrieved.is_some(), "expected entry to be retrieved");
197
+
}
198
+
}