1use super::{
8 NnsProposalCollectionState, collection::validate_collection_state, model::NnsProposalRow,
9};
10use crate::nns::governance::{NnsGovernanceCollectionStatus, NnsGovernanceSourceProvenance};
11use serde::Serialize;
12use std::collections::{BTreeMap, HashSet};
13use thiserror::Error as ThisError;
14
15pub const NNS_PROPOSAL_ACTIVITY_REPORT_SCHEMA_VERSION: u32 = 1;
17
18const SECONDS_PER_DAY: u64 = 86_400;
19
20#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
27pub struct NnsProposalActivityRequest {
28 pub from_proposal_timestamp_seconds: Option<u64>,
30 pub until_proposal_timestamp_seconds: Option<u64>,
32}
33
34#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
41pub struct NnsProposalTopicCount {
42 pub topic: i32,
44 pub proposal_count: u64,
46}
47
48#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
55pub struct NnsProposalStatusCount {
56 pub status: i32,
58 pub proposal_count: u64,
60}
61
62#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
69pub struct NnsProposalRewardStatusCount {
70 pub reward_status: i32,
72 pub proposal_count: u64,
74}
75
76#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
83pub struct NnsProposalDayCount {
84 pub day_start_timestamp_seconds: u64,
86 pub proposal_count: u64,
88}
89
90#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
97pub struct NnsProposalActivityReport {
98 #[serde(flatten)]
100 pub context: crate::nns::governance::NnsGovernanceAnalyticsContext,
101 pub collected_proposal_count: u64,
103 pub from_proposal_timestamp_seconds: Option<u64>,
105 pub until_proposal_timestamp_seconds: Option<u64>,
107 pub included_proposal_count: u64,
109 pub excluded_before_from_count: u64,
111 pub excluded_at_or_after_until_count: u64,
113 pub earliest_included_proposal_timestamp_seconds: Option<u64>,
115 pub latest_included_proposal_timestamp_seconds: Option<u64>,
117 pub topic_counts: Vec<NnsProposalTopicCount>,
119 pub status_counts: Vec<NnsProposalStatusCount>,
121 pub reward_status_counts: Vec<NnsProposalRewardStatusCount>,
123 pub day_counts: Vec<NnsProposalDayCount>,
125}
126
127#[derive(Debug, ThisError)]
134pub enum NnsProposalActivityError {
135 #[error("invalid NNS proposal collection state for activity projection: {reason}")]
137 InvalidCollectionState {
138 reason: String,
140 },
141
142 #[error("NNS proposal activity requires a complete collection; state is {status}")]
144 CollectionNotComplete {
145 status: NnsGovernanceCollectionStatus,
147 },
148
149 #[error(
151 "invalid NNS proposal activity time window: from {from_proposal_timestamp_seconds} must be below until {until_proposal_timestamp_seconds}"
152 )]
153 InvalidTimeWindow {
154 from_proposal_timestamp_seconds: u64,
156 until_proposal_timestamp_seconds: u64,
158 },
159
160 #[error(
162 "NNS proposal activity received {actual} rows; complete collection accounts for {expected}"
163 )]
164 ProposalCountMismatch {
165 expected: u64,
167 actual: u64,
169 },
170
171 #[error("NNS proposal activity received a row without a proposal id")]
173 MissingProposalId,
174
175 #[error("NNS proposal activity received proposal id zero")]
177 ZeroProposalId,
178
179 #[error("NNS proposal activity received duplicate proposal id {proposal_id}")]
181 DuplicateProposalId {
182 proposal_id: u64,
184 },
185
186 #[error("NNS proposal {proposal_id} has proposal timestamp zero")]
188 ZeroProposalTimestamp {
189 proposal_id: u64,
191 },
192
193 #[error("NNS proposal activity accounting overflow while updating {field}")]
195 AccountingOverflow {
196 field: &'static str,
198 },
199}
200
201pub fn build_nns_proposal_activity_report(
203 request: &NnsProposalActivityRequest,
204 collection: &NnsProposalCollectionState,
205 proposals: &[NnsProposalRow],
206) -> Result<NnsProposalActivityReport, NnsProposalActivityError> {
207 validate_collection_state(collection).map_err(|error| {
208 NnsProposalActivityError::InvalidCollectionState {
209 reason: error.to_string(),
210 }
211 })?;
212 if !collection.is_complete() {
213 return Err(NnsProposalActivityError::CollectionNotComplete {
214 status: collection.status(),
215 });
216 }
217 validate_time_window(request)?;
218
219 let expected = collection.proposals_fetched();
220 let actual = u64::try_from(proposals.len()).map_err(|_| {
221 NnsProposalActivityError::AccountingOverflow {
222 field: "supplied_proposal_count",
223 }
224 })?;
225 if actual != expected {
226 return Err(NnsProposalActivityError::ProposalCountMismatch { expected, actual });
227 }
228
229 let mut proposal_ids = HashSet::with_capacity(proposals.len());
230 aggregate_activity(request, collection, proposals, |proposal| {
231 validate_proposal_row(proposal, &mut proposal_ids)
232 })
233}
234
235#[cfg(feature = "nns-host")]
236pub(super) fn build_activity_from_cache(
237 request: &NnsProposalActivityRequest,
238 cache: &super::cache::AdmittedProposalCache,
239) -> Result<NnsProposalActivityReport, NnsProposalActivityError> {
240 aggregate_activity(
242 request,
243 cache.collection_state(),
244 cache.rows(),
245 |proposal| {
246 let proposal_id = proposal
247 .proposal_id
248 .ok_or(NnsProposalActivityError::MissingProposalId)?;
249 validate_proposal_timestamp(proposal, proposal_id)
250 },
251 )
252}
253
254fn aggregate_activity(
255 request: &NnsProposalActivityRequest,
256 collection: &NnsProposalCollectionState,
257 proposals: &[NnsProposalRow],
258 mut validate_row: impl FnMut(&NnsProposalRow) -> Result<(), NnsProposalActivityError>,
259) -> Result<NnsProposalActivityReport, NnsProposalActivityError> {
260 let mut activity = ActivityAccumulator::default();
261 for proposal in proposals {
262 validate_row(proposal)?;
263 activity.observe(request, proposal)?;
264 }
265 let source = collection.source().cloned().ok_or_else(|| {
266 NnsProposalActivityError::InvalidCollectionState {
267 reason: "complete collection has no concrete source provenance".to_string(),
268 }
269 })?;
270 Ok(activity.into_report(request, collection, source))
271}
272
273#[derive(Default)]
274struct ActivityAccumulator {
275 topic_counts: BTreeMap<i32, u64>,
276 status_counts: BTreeMap<i32, u64>,
277 reward_status_counts: BTreeMap<i32, u64>,
278 day_counts: BTreeMap<u64, u64>,
279 included_proposal_count: u64,
280 excluded_before_from_count: u64,
281 excluded_at_or_after_until_count: u64,
282 included_timestamp_bounds: Option<(u64, u64)>,
283}
284
285impl ActivityAccumulator {
286 fn observe(
287 &mut self,
288 request: &NnsProposalActivityRequest,
289 proposal: &NnsProposalRow,
290 ) -> Result<(), NnsProposalActivityError> {
291 let timestamp = proposal.proposal_timestamp_seconds;
292 if request
293 .from_proposal_timestamp_seconds
294 .is_some_and(|from| timestamp < from)
295 {
296 return increment_count(
297 &mut self.excluded_before_from_count,
298 "excluded_before_from_count",
299 );
300 }
301 if request
302 .until_proposal_timestamp_seconds
303 .is_some_and(|until| timestamp >= until)
304 {
305 return increment_count(
306 &mut self.excluded_at_or_after_until_count,
307 "excluded_at_or_after_until_count",
308 );
309 }
310
311 increment_count(&mut self.included_proposal_count, "included_proposal_count")?;
312 increment_count(
313 self.topic_counts.entry(proposal.topic).or_default(),
314 "topic_count",
315 )?;
316 increment_count(
317 self.status_counts.entry(proposal.status).or_default(),
318 "status_count",
319 )?;
320 increment_count(
321 self.reward_status_counts
322 .entry(proposal.reward_status)
323 .or_default(),
324 "reward_status_count",
325 )?;
326 let day_start = timestamp - (timestamp % SECONDS_PER_DAY);
327 increment_count(self.day_counts.entry(day_start).or_default(), "day_count")?;
328 self.included_timestamp_bounds = Some(match self.included_timestamp_bounds {
329 Some((earliest, latest)) => (earliest.min(timestamp), latest.max(timestamp)),
330 None => (timestamp, timestamp),
331 });
332 Ok(())
333 }
334
335 fn into_report(
336 self,
337 request: &NnsProposalActivityRequest,
338 collection: &NnsProposalCollectionState,
339 source: NnsGovernanceSourceProvenance,
340 ) -> NnsProposalActivityReport {
341 NnsProposalActivityReport {
342 context: crate::nns::governance::NnsGovernanceAnalyticsContext {
343 schema_version: NNS_PROPOSAL_ACTIVITY_REPORT_SCHEMA_VERSION,
344 network: collection.network().to_string(),
345 governance_canister_id: collection.governance_canister_id().to_string(),
346 source,
347 collection_started_at: collection.started_at().to_string(),
348 collection_updated_at: collection.updated_at().to_string(),
349 collection_page_count: collection.pages_fetched(),
350 point_in_time_guaranteed: false,
351 },
352 collected_proposal_count: collection.proposals_fetched(),
353 from_proposal_timestamp_seconds: request.from_proposal_timestamp_seconds,
354 until_proposal_timestamp_seconds: request.until_proposal_timestamp_seconds,
355 included_proposal_count: self.included_proposal_count,
356 excluded_before_from_count: self.excluded_before_from_count,
357 excluded_at_or_after_until_count: self.excluded_at_or_after_until_count,
358 earliest_included_proposal_timestamp_seconds: self
359 .included_timestamp_bounds
360 .map(|bounds| bounds.0),
361 latest_included_proposal_timestamp_seconds: self
362 .included_timestamp_bounds
363 .map(|bounds| bounds.1),
364 topic_counts: self
365 .topic_counts
366 .into_iter()
367 .map(|(topic, proposal_count)| NnsProposalTopicCount {
368 topic,
369
370 proposal_count,
371 })
372 .collect(),
373 status_counts: self
374 .status_counts
375 .into_iter()
376 .map(|(status, proposal_count)| NnsProposalStatusCount {
377 status,
378
379 proposal_count,
380 })
381 .collect(),
382 reward_status_counts: self
383 .reward_status_counts
384 .into_iter()
385 .map(
386 |(reward_status, proposal_count)| NnsProposalRewardStatusCount {
387 reward_status,
388
389 proposal_count,
390 },
391 )
392 .collect(),
393 day_counts: self
394 .day_counts
395 .into_iter()
396 .map(
397 |(day_start_timestamp_seconds, proposal_count)| NnsProposalDayCount {
398 day_start_timestamp_seconds,
399 proposal_count,
400 },
401 )
402 .collect(),
403 }
404 }
405}
406
407pub(super) const fn validate_time_window(
408 request: &NnsProposalActivityRequest,
409) -> Result<(), NnsProposalActivityError> {
410 if let (Some(from), Some(until)) = (
411 request.from_proposal_timestamp_seconds,
412 request.until_proposal_timestamp_seconds,
413 ) && from >= until
414 {
415 return Err(NnsProposalActivityError::InvalidTimeWindow {
416 from_proposal_timestamp_seconds: from,
417 until_proposal_timestamp_seconds: until,
418 });
419 }
420 Ok(())
421}
422
423fn validate_proposal_row(
424 proposal: &NnsProposalRow,
425 proposal_ids: &mut HashSet<u64>,
426) -> Result<(), NnsProposalActivityError> {
427 let proposal_id = proposal
428 .proposal_id
429 .ok_or(NnsProposalActivityError::MissingProposalId)?;
430 if proposal_id == 0 {
431 return Err(NnsProposalActivityError::ZeroProposalId);
432 }
433 if !proposal_ids.insert(proposal_id) {
434 return Err(NnsProposalActivityError::DuplicateProposalId { proposal_id });
435 }
436 validate_proposal_timestamp(proposal, proposal_id)
437}
438
439const fn validate_proposal_timestamp(
440 proposal: &NnsProposalRow,
441 proposal_id: u64,
442) -> Result<(), NnsProposalActivityError> {
443 if proposal.proposal_timestamp_seconds == 0 {
444 return Err(NnsProposalActivityError::ZeroProposalTimestamp { proposal_id });
445 }
446
447 Ok(())
448}
449
450fn increment_count(count: &mut u64, field: &'static str) -> Result<(), NnsProposalActivityError> {
451 *count = count
452 .checked_add(1)
453 .ok_or(NnsProposalActivityError::AccountingOverflow { field })?;
454 Ok(())
455}
456
457#[cfg(test)]
458mod tests;