baileytownsend.dev
tangled.org trending bluesky account
wip
+49
-2
+14
bot/migrations/20250914170500_create_stars.sql
+14
bot/migrations/20250914170500_create_stars.sql
···
1
+
-- Create the stars table
2
+
CREATE TABLE IF NOT EXISTS stars (
3
+
createdAt TEXT NOT NULL, -- ISO8601 timestamp string (from record.created_at)
4
+
did TEXT NOT NULL,
5
+
rkey TEXT NOT NULL,
6
+
subject TEXT NOT NULL
7
+
);
8
+
9
+
-- Indexes for faster lookup
10
+
CREATE INDEX IF NOT EXISTS idx_stars_rkey ON stars(rkey);
11
+
CREATE INDEX IF NOT EXISTS idx_stars_subject ON stars(subject);
12
+
13
+
-- Optional composite unique constraint to prevent duplicate entries per DID/rkey
14
+
CREATE UNIQUE INDEX IF NOT EXISTS uq_stars_did_rkey ON stars(did, rkey);
+35
-2
bot/src/main.rs
+35
-2
bot/src/main.rs
···
35
35
let msg_rx = jetstream.get_msg_rx();
36
36
let reconnect_tx = jetstream.get_reconnect_tx();
37
37
38
+
// Read configuration from environment
39
+
let timeframe_hours: i64 = std::env::var("TIMEFRAME")
40
+
.ok()
41
+
.and_then(|v| v.parse::<i64>().ok())
42
+
.filter(|v| *v > 0)
43
+
.unwrap_or(1);
44
+
let star_threshold: i64 = std::env::var("STAR_THRESHOLD")
45
+
.ok()
46
+
.and_then(|v| v.parse::<i64>().ok())
47
+
.filter(|v| *v > 0)
48
+
.unwrap_or(10);
49
+
38
50
// Ingestor for the star collection
39
51
let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
40
52
ingestors.insert(
41
53
atproto_api::sh::tangled::feed::Star::NSID.to_string(),
42
-
Box::new(StarIngestor { pool: pool.clone() }),
54
+
Box::new(StarIngestor { pool: pool.clone(), timeframe_hours, star_threshold }),
43
55
);
44
56
let ingestors = Arc::new(ingestors);
45
57
···
61
73
62
74
struct StarIngestor {
63
75
pool: SqlitePool,
76
+
timeframe_hours: i64,
77
+
star_threshold: i64,
64
78
}
65
79
66
80
#[async_trait::async_trait]
···
72
86
if let Some(record) = &commit.record {
73
87
let rec = serde_json::from_value::<atproto_api::sh::tangled::feed::star::RecordData>(record.clone())?;
74
88
// Insert or ignore duplicate per did+rkey
75
-
let _ = sqlx::query(
89
+
let result = sqlx::query(
76
90
"INSERT OR IGNORE INTO stars(createdAt, did, rkey, subject) VALUES(?, ?, ?, ?)"
77
91
)
78
92
.bind(rec.created_at.as_str())
···
81
95
.bind(&rec.subject)
82
96
.execute(&self.pool)
83
97
.await?;
98
+
99
+
// Only check threshold if we actually inserted a new row
100
+
if result.rows_affected() > 0 {
101
+
let offset = format!("-{} hours", self.timeframe_hours);
102
+
// Count stars in the last timeframe_hours using RFC3339 string comparison
103
+
let (count_in_window,): (i64,) = sqlx::query_as(
104
+
"SELECT COUNT(*) as cnt FROM stars WHERE createdAt >= strftime('%Y-%m-%dT%H:%M:%fZ','now', ?)"
105
+
)
106
+
.bind(&offset)
107
+
.fetch_one(&self.pool)
108
+
.await?;
109
+
110
+
if count_in_window >= self.star_threshold {
111
+
println!(
112
+
"Star threshold met: {} stars in the last {} hour(s) (threshold: {})",
113
+
count_in_window, self.timeframe_hours, self.star_threshold
114
+
);
115
+
}
116
+
}
84
117
}
85
118
}
86
119
Operation::Delete => {