fakecloud_timestream/
state.rs

1//! Account-partitioned, serializable state for Amazon Timestream.
2//!
3//! Everything a Timestream caller can create lives here: databases, tables,
4//! the ingested data points (a bounded in-memory buffer per table so `Query`
5//! can read them back), scheduled queries, batch-load tasks, the per-account
6//! query settings, and in-flight query ids (so `CancelQuery` resolves). Config
7//! blobs (`RetentionProperties`, `MagneticStoreWriteProperties`, `Schema`, a
8//! scheduled query's schedule/notification/target config) are stored as their
9//! already-output-valid wire JSON so a `Describe*` echoes exactly what its
10//! `Create*` persisted.
11//!
12//! Databases are keyed by name; tables and their record buffers by
13//! `"{database}\u{1}{table}"` (see [`crate::shared::table_key`]); scheduled
14//! queries and batch-load tasks by their minted id/ARN. Tags live in an
15//! ARN-keyed side map so `ListTagsForResource` has a single source of truth.
16
17use std::collections::BTreeMap;
18use std::sync::Arc;
19
20use parking_lot::RwLock;
21use serde::{Deserialize, Serialize};
22use serde_json::Value;
23
24use fakecloud_core::multi_account::{AccountState, MultiAccountState};
25
26pub const TIMESTREAM_SNAPSHOT_SCHEMA_VERSION: u32 = 1;
27
28/// The maximum number of ingested points retained per table. Timestream itself
29/// keeps data by retention window; fakecloud keeps a bounded ring so the
30/// process footprint stays flat under sustained `WriteRecords`.
31pub const MAX_RECORDS_PER_TABLE: usize = 100_000;
32
33/// A Timestream database.
34#[derive(Debug, Clone, Serialize, Deserialize)]
35pub struct Database {
36    pub name: String,
37    pub arn: String,
38    #[serde(default)]
39    pub kms_key_id: Option<String>,
40    pub table_count: i64,
41    pub creation_time: f64,
42    pub last_updated_time: f64,
43}
44
45/// A Timestream table.
46#[derive(Debug, Clone, Serialize, Deserialize)]
47pub struct Table {
48    pub name: String,
49    pub database_name: String,
50    pub arn: String,
51    /// `ACTIVE` / `DELETING` / `RESTORING`.
52    pub status: String,
53    /// The `RetentionProperties` wire object, stored verbatim.
54    #[serde(default)]
55    pub retention_properties: Value,
56    /// The `MagneticStoreWriteProperties` wire object, stored verbatim.
57    #[serde(default)]
58    pub magnetic_store_write_properties: Option<Value>,
59    /// The `Schema` wire object (composite partition key), stored verbatim.
60    #[serde(default)]
61    pub schema: Value,
62    pub creation_time: f64,
63    pub last_updated_time: f64,
64}
65
66/// One ingested Timestream data point, normalized to nanoseconds so the query
67/// engine can time-filter and format it uniformly regardless of the write-time
68/// `TimeUnit`.
69#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct StoredRecord {
71    /// Epoch time in nanoseconds.
72    pub time_nanos: i128,
73    /// Dimension name -> value, in insertion order.
74    pub dimensions: Vec<(String, String)>,
75    pub measure_name: String,
76    pub measure_value: String,
77    /// `DOUBLE` / `BIGINT` / `VARCHAR` / `BOOLEAN` / `TIMESTAMP` / `MULTI`.
78    pub measure_value_type: String,
79}
80
81/// A scheduled query.
82#[derive(Debug, Clone, Serialize, Deserialize)]
83pub struct ScheduledQuery {
84    pub arn: String,
85    pub name: String,
86    pub query_string: String,
87    /// `ENABLED` / `DISABLED`.
88    pub state: String,
89    pub creation_time: f64,
90    pub schedule_configuration: Value,
91    pub notification_configuration: Value,
92    #[serde(default)]
93    pub target_configuration: Option<Value>,
94    #[serde(default)]
95    pub scheduled_query_execution_role_arn: Option<String>,
96    #[serde(default)]
97    pub kms_key_id: Option<String>,
98    #[serde(default)]
99    pub error_report_configuration: Option<Value>,
100    #[serde(default)]
101    pub previous_invocation_time: Option<f64>,
102    #[serde(default)]
103    pub last_run_status: Option<String>,
104}
105
106/// A batch-load task.
107#[derive(Debug, Clone, Serialize, Deserialize)]
108pub struct BatchLoadTask {
109    pub task_id: String,
110    /// `CREATED` / `IN_PROGRESS` / `FAILED` / `SUCCEEDED` / `PROGRESS_STOPPED`
111    /// / `PENDING_RESUME`.
112    pub status: String,
113    pub target_database_name: String,
114    pub target_table_name: String,
115    pub data_source_configuration: Value,
116    pub report_configuration: Value,
117    #[serde(default)]
118    pub data_model_configuration: Option<Value>,
119    #[serde(default)]
120    pub record_version: Option<i64>,
121    pub creation_time: f64,
122    pub last_updated_time: f64,
123    pub resumable_until: f64,
124}
125
126/// Per-account query settings.
127#[derive(Debug, Clone, Serialize, Deserialize)]
128pub struct AccountSettings {
129    #[serde(default)]
130    pub max_query_tcu: Option<i64>,
131    /// `BYTES_SCANNED` / `COMPUTE_UNITS`.
132    pub query_pricing_model: String,
133    #[serde(default)]
134    pub query_compute: Option<Value>,
135}
136
137impl Default for AccountSettings {
138    fn default() -> Self {
139        Self {
140            max_query_tcu: None,
141            query_pricing_model: "BYTES_SCANNED".to_string(),
142            query_compute: None,
143        }
144    }
145}
146
147/// Per-account Amazon Timestream state.
148#[derive(Debug, Clone, Default, Serialize, Deserialize)]
149pub struct TimestreamData {
150    #[serde(default)]
151    pub account_id: String,
152    #[serde(default)]
153    pub region: String,
154
155    /// Databases keyed by name.
156    #[serde(default)]
157    pub databases: BTreeMap<String, Database>,
158    /// Tables keyed by `"{database}\u{1}{table}"`.
159    #[serde(default)]
160    pub tables: BTreeMap<String, Table>,
161    /// Ingested points keyed by `"{database}\u{1}{table}"`.
162    #[serde(default)]
163    pub records: BTreeMap<String, Vec<StoredRecord>>,
164    /// Scheduled queries keyed by ARN.
165    #[serde(default)]
166    pub scheduled_queries: BTreeMap<String, ScheduledQuery>,
167    /// Batch-load tasks keyed by task id.
168    #[serde(default)]
169    pub batch_load_tasks: BTreeMap<String, BatchLoadTask>,
170    /// In-flight query ids (set membership; `CancelQuery` removes).
171    #[serde(default)]
172    pub active_queries: BTreeMap<String, f64>,
173    /// Per-account query settings.
174    #[serde(default)]
175    pub account_settings: AccountSettings,
176    /// Resource tags keyed by ARN.
177    #[serde(default)]
178    pub tags: BTreeMap<String, BTreeMap<String, String>>,
179}
180
181impl AccountState for TimestreamData {
182    fn new_for_account(account_id: &str, region: &str, _endpoint: &str) -> Self {
183        Self {
184            account_id: account_id.to_string(),
185            region: region.to_string(),
186            ..Default::default()
187        }
188    }
189}
190
191pub type SharedTimestreamState = Arc<RwLock<MultiAccountState<TimestreamData>>>;
192
193#[derive(Debug, Serialize, Deserialize)]
194pub struct TimestreamSnapshot {
195    pub schema_version: u32,
196    pub accounts: MultiAccountState<TimestreamData>,
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    #[test]
204    fn account_state_seeds_ids() {
205        let d = TimestreamData::new_for_account("000000000000", "us-east-1", "");
206        assert_eq!(d.account_id, "000000000000");
207        assert_eq!(d.region, "us-east-1");
208        assert!(d.databases.is_empty());
209        assert_eq!(d.account_settings.query_pricing_model, "BYTES_SCANNED");
210    }
211}
fakecloud_timestream/state.rs

fakecloud_timestream/
state.rs
