aphomie_db_client/
lib.rs

1use anyhow::bail;
2use bms_dtos::TrendDataRaw;
3use chrono::{DateTime, NaiveDateTime, TimeDelta, Utc};
4use serde::{Deserialize, Serialize, ser::SerializeStruct};
5use tokio_postgres::{Client, Config, NoTls, Row, ToStatement, types::ToSql};
6
7type Result<T> = core::result::Result<T, anyhow::Error>;
8
9pub struct AphomieDbClient {
10    db_client: Client,
11}
12
13impl AphomieDbClient {
14    pub async fn new(host: Option<&str>, port: Option<u16>) -> Result<Self> {
15        let host = if let Some(v) = host { v } else { "localhost" };
16
17        let port = if let Some(v) = port { v } else { 5432 };
18
19        let mut config = Config::new();
20        config
21            .user("dat7")
22            .password("dat7")
23            .dbname("dat7")
24            .host(host)
25            .port(port);
26
27        let (db_client, connection) = config.connect(NoTls).await?;
28
29        tokio::spawn(async move {
30            if let Err(e) = connection.await {
31                eprintln!("connection error: {}", e);
32            }
33        });
34
35        Ok(Self { db_client })
36    }
37
38    pub async fn query<T>(&self, statement: &T, params: &[&(dyn ToSql + Sync)]) -> Result<Vec<Row>>
39    where
40        T: ?Sized + ToStatement,
41    {
42        self.db_client
43            .query(statement, params)
44            .await
45            .map_err(|e| anyhow::anyhow!("failed performing query: {:?}", e))
46    }
47
48    pub async fn insert_metadata(&self, metadata: &MetadataDb) -> Result<Vec<MetadataDb>> {
49        insert_metadata(
50            &self.db_client,
51            &metadata.log_id,
52            &metadata.source,
53            &metadata.unit,
54            &metadata.alias,
55        )
56        .await
57        .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
58    }
59
60    pub async fn get_metadata_by_source(&self, source: &str) -> Result<Vec<MetadataDb>> {
61        get_metadata_by_source(&self.db_client, source)
62            .await
63            .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
64    }
65
66    pub async fn get_latest_trend_like_source_and_alias(
67        &self,
68        source: &str,
69        alias: &str,
70    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
71        get_latest_trend_like_source_and_alias(&self.db_client, source, alias)
72            .await
73            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
74    }
75
76    pub async fn get_latest_trend_like_source(
77        &self,
78        source: &str,
79    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
80        get_latest_trend_like_source(&self.db_client, source)
81            .await
82            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
83    }
84
85    pub async fn get_all_metadata(&self) -> Result<Vec<MetadataDb>> {
86        get_all_metadata(&self.db_client)
87            .await
88            .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
89    }
90
91    pub async fn insert_trend_data(&self, trend_data: &TrendDataDb) -> Result<Vec<TrendDataDb>> {
92        insert_trend_data(
93            &self.db_client,
94            &trend_data.log_id,
95            &trend_data.timestamp,
96            &trend_data.value,
97        )
98        .await
99        .map(|v| v.iter().map(|r| TrendDataDb::from(r)).collect())
100    }
101
102    pub async fn get_trend_data_by_period(
103        &self,
104        start: &DateTime<Utc>,
105        end: &DateTime<Utc>,
106    ) -> Result<Vec<TrendDataDb>> {
107        get_trend_data_by_period(&self.db_client, start, end)
108            .await
109            .map(|v| v.iter().map(|r| TrendDataDb::from(r)).collect())
110    }
111
112    pub async fn get_trend_data_by_source_offset_limit(
113        &self,
114        source: &str,
115        offset: i64,
116        limit: i64,
117    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
118        get_trend_data_by_source_offset_limit(&self.db_client, source, offset, limit)
119            .await
120            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
121    }
122
123    pub async fn get_metadata_join_trend_data_offset_limit(
124        &self,
125        offset: i64,
126        limit: i64,
127    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
128        get_metadata_join_trend_data_offset_limit(&self.db_client, offset, limit)
129            .await
130            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
131    }
132
133    pub async fn get_latest_trend_data(&self) -> Result<Option<TrendDataDb>> {
134        match get_latest_trend_data(&self.db_client).await.map(|v| {
135            v.iter()
136                .map(|e| TrendDataDb::from(e))
137                .collect::<Vec<TrendDataDb>>()
138        }) {
139            Ok(v) => Ok(v.get(0).map(|e| e.clone())),
140            Err(e) => bail!(e),
141        }
142    }
143
144    pub async fn get_latest_trend_by_log_id(&self, log_id: i64) -> Result<Option<TrendDataDb>> {
145        get_latest_trend_by_log_id(&self.db_client, log_id)
146            .await
147            .map(|v| {
148                v.iter()
149                    .map(|r| TrendDataDb::from(r))
150                    .collect::<Vec<TrendDataDb>>()
151            })
152            .map(|e| e.get(0).map(|e| e.clone()))
153    }
154
155    pub async fn get_metadata_by_log_id(&self, log_id: i64) -> Result<Option<MetadataDb>> {
156        get_metadata_by_id(&self.db_client, log_id)
157            .await
158            .map(|v| {
159                v.iter()
160                    .map(|r| MetadataDb::from(r))
161                    .collect::<Vec<MetadataDb>>()
162            })
163            .map(|e| e.get(0).map(|e| e.clone()))
164    }
165
166    pub async fn get_trends_like_source_and_alias_in_dt_range_count(
167        &self,
168        source: &str,
169        alias: &str,
170        start: &DateTime<Utc>,
171        end: &DateTime<Utc>,
172        offset: i64,
173        limit: i64,
174    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
175        get_trends_like_source_and_alias_count(
176            &self.db_client,
177            source,
178            alias,
179            start,
180            end,
181            offset,
182            limit,
183        )
184        .await
185        .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
186    }
187}
188
189async fn insert_metadata(
190    client: &Client,
191    log_id: &i64,
192    source: &str,
193    unit: &str,
194    alias: &Option<String>,
195) -> Result<Vec<Row>> {
196    let stmt =
197        "INSERT INTO metadata (log_id, source, unit, alias) VALUES ($1, $2, $3, $4) RETURNING *;";
198
199    match client.query(stmt, &[log_id, &source, &unit, &alias]).await {
200        Ok(v) => Ok(v),
201        Err(e) => {
202            bail!("error querying database: {e}");
203        }
204    }
205}
206
207async fn get_metadata_by_source(client: &Client, source: &str) -> Result<Vec<Row>> {
208    let stmt = "SELECT * FROM metadata WHERE source = $1;";
209
210    match client.query(stmt, &[&source]).await {
211        Ok(v) => Ok(v),
212        Err(e) => {
213            bail!("error querying database: {e}");
214        }
215    }
216}
217
218async fn get_latest_trend_like_source_and_alias(
219    client: &Client,
220    source: &str,
221    alias: &str,
222) -> Result<Vec<Row>> {
223    let stmt = "SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value FROM metadata m JOIN trend_data t on t.log_id = m.log_id WHERE m.source LIKE $1 AND m.alias = $2 ORDER BY t.timestamp DESC LIMIT 1;";
224
225    match client.query(stmt, &[&source, &alias]).await {
226        Ok(v) => Ok(v),
227        Err(e) => {
228            bail!("error querying database: {e}");
229        }
230    }
231}
232
233async fn get_latest_trend_like_source(client: &Client, source: &str) -> Result<Vec<Row>> {
234    let stmt = "SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value FROM metadata m JOIN trend_data t on t.log_id = m.log_id WHERE m.source LIKE $1 ORDER BY t.timestamp DESC LIMIT 1;";
235
236    match client.query(stmt, &[&source]).await {
237        Ok(v) => Ok(v),
238        Err(e) => {
239            bail!("error querying database: {e}");
240        }
241    }
242}
243
244async fn get_metadata_by_id(client: &Client, log_id: i64) -> Result<Vec<Row>> {
245    let stmt = "SELECT * FROM metadata WHERE log_id = $1;";
246
247    match client.query(stmt, &[&log_id]).await {
248        Ok(v) => Ok(v),
249        Err(e) => {
250            bail!("error querying database: {e}");
251        }
252    }
253}
254
255async fn get_latest_trend_by_log_id(client: &Client, log_id: i64) -> Result<Vec<Row>> {
256    let stmt = "SELECT * FROM trend_data WHERE log_id = $1 ORDER BY timestamp DESC LIMIT 1;";
257
258    match client.query(stmt, &[&log_id]).await {
259        Ok(v) => Ok(v),
260        Err(e) => {
261            bail!("error querying database: {e}");
262        }
263    }
264}
265
266async fn get_all_metadata(client: &Client) -> Result<Vec<Row>> {
267    let stmt = "SELECT * FROM metadata;";
268
269    match client.query(stmt, &[]).await {
270        Ok(v) => Ok(v),
271        Err(e) => {
272            bail!("error querying database: {e}");
273        }
274    }
275}
276
277async fn get_trend_data_by_period(
278    client: &Client,
279    start: &DateTime<Utc>,
280    end: &DateTime<Utc>,
281) -> Result<Vec<Row>> {
282    let stmt = "SELECT * FROM trend_data WHERE timestamp <= $1 AND timestamp >= $2;";
283
284    match client.query(stmt, &[&end, &start]).await {
285        Ok(v) => Ok(v),
286        Err(e) => {
287            bail!("error querying database: {e}");
288        }
289    }
290}
291
292async fn get_trend_data_by_source_offset_limit(
293    client: &Client,
294    source: &str,
295    offset: i64,
296    limit: i64,
297) -> Result<Vec<Row>> {
298    let stmt = "SELECT t.log_id, t.timestamp, t.value, m.source, m.unit FROM metadata m JOIN trend_data t ON m.log_id = t.log_id WHERE source LIKE $1 ORDER BY t.log_id, t.timestamp OFFSET $2 LIMIT $3;";
299
300    match client.query(stmt, &[&source, &offset, &limit]).await {
301        Ok(v) => Ok(v),
302        Err(e) => {
303            bail!("error querying database: {e}");
304        }
305    }
306}
307
308async fn get_metadata_join_trend_data_offset_limit(
309    client: &Client,
310    offset: i64,
311    limit: i64,
312) -> Result<Vec<Row>> {
313    let stmt = "SELECT t.log_id, t.timestamp, t.value, m.source, m.unit FROM metadata m JOIN trend_data t ON m.log_id = t.log_id ORDER BY t.timestamp, t.log_id OFFSET $1 LIMIT $2;";
314
315    match client.query(stmt, &[&offset, &limit]).await {
316        Ok(v) => Ok(v),
317        Err(e) => {
318            bail!("error querying database: {e}");
319        }
320    }
321}
322
323async fn get_latest_trend_data(client: &Client) -> Result<Vec<Row>> {
324    let stmt = "SELECT * FROM trend_data ORDER BY timestamp DESC LIMIT 1";
325
326    match client.query(stmt, &[]).await {
327        Ok(v) => Ok(v),
328        Err(e) => {
329            bail!("error querying database: {e}");
330        }
331    }
332}
333
334async fn insert_trend_data(
335    client: &Client,
336    log_id: &i64,
337    timestamp: &DateTime<Utc>,
338    value: &Option<f64>,
339) -> Result<Vec<Row>> {
340    let stmt = "INSERT INTO trend_data (log_id, timestamp, value) VALUES ($1, $2, $3) RETURNING *;";
341
342    match client.query(stmt, &[log_id, &timestamp, &value]).await {
343        Ok(v) => Ok(v),
344        Err(e) => {
345            bail!("error querying database: {e}");
346        }
347    }
348}
349
350async fn get_trends_like_source_and_alias_count(
351    client: &Client,
352    source: &str,
353    alias: &str,
354    start: &DateTime<Utc>,
355    end: &DateTime<Utc>,
356    offset: i64,
357    limit: i64,
358) -> Result<Vec<Row>> {
359    let stmt = "
360SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value 
361FROM metadata m JOIN trend_data t on t.log_id = m.log_id 
362WHERE 1=1
363    AND m.source LIKE $1 
364    AND m.alias LIKE $2
365    AND t.timestamp >= $3
366    AND t.timestamp < $4
367ORDER BY t.timestamp OFFSET $5 LIMIT $6;
368";
369
370    match client
371        .query(stmt, &[&source, &alias, start, end, &offset, &limit])
372        .await
373    {
374        Ok(v) => Ok(v),
375        Err(e) => {
376            bail!("error querying database: {e}");
377        }
378    }
379}
380
381#[derive(Debug, Clone)]
382pub struct MetadataTrendDataJoinDb {
383    pub log_id: i64,
384    pub source: String,
385    pub unit: String,
386    pub timestamp: DateTime<Utc>,
387    pub value: Option<f64>,
388    pub alias: Option<String>,
389}
390
391#[derive(Debug, Clone)]
392pub struct TrendDataDb {
393    pub log_id: i64,
394    pub timestamp: DateTime<Utc>,
395    pub value: Option<f64>,
396}
397
398impl Serialize for MetadataTrendDataJoinDb {
399    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
400    where
401        S: serde::Serializer,
402    {
403        let mut s = serializer.serialize_struct("TrendDataDb", 3)?;
404        s.serialize_field("log_id", &self.log_id)?;
405        s.serialize_field("timestamp", &self.timestamp.to_rfc3339())?;
406        s.serialize_field("value", &self.value)?;
407        s.serialize_field("source", &self.source)?;
408        s.serialize_field("unit", &self.unit)?;
409        s.serialize_field("alias", &self.alias)?;
410        s.end()
411    }
412}
413
414impl Serialize for TrendDataDb {
415    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
416    where
417        S: serde::Serializer,
418    {
419        let mut s = serializer.serialize_struct("TrendDataDb", 3)?;
420        s.serialize_field("log_id", &self.log_id)?;
421        s.serialize_field("timestamp", &self.timestamp.to_rfc3339())?;
422        s.serialize_field("value", &self.value)?;
423        s.end()
424    }
425}
426
427#[derive(Clone, Serialize, Deserialize, Debug)]
428pub struct MetadataDb {
429    pub log_id: i64,
430    pub source: String,
431    pub unit: String,
432    pub alias: Option<String>,
433}
434
435impl From<&Row> for MetadataTrendDataJoinDb {
436    fn from(value: &Row) -> Self {
437        let log_id: i64 = value.get("log_id");
438        let source: String = value.get("source");
439        let unit: String = value.get("unit");
440        let timestamp: DateTime<Utc> = value.get("timestamp");
441        let alias: Option<String> = value.get("alias");
442        let value: Option<f64> = value.get("value");
443
444        MetadataTrendDataJoinDb {
445            value,
446            log_id,
447            source,
448            unit,
449            timestamp,
450            alias,
451        }
452    }
453}
454
455impl From<&Row> for MetadataDb {
456    fn from(value: &Row) -> MetadataDb {
457        let log_id: i64 = value.get("log_id");
458        let source: String = value.get("source");
459        let unit: String = value.get("unit");
460        let alias: Option<String> = value.get("alias");
461
462        MetadataDb {
463            log_id,
464            source,
465            unit,
466            alias,
467        }
468    }
469}
470
471impl TryFrom<&TrendDataRaw> for TrendDataDb {
472    type Error = anyhow::Error;
473
474    fn try_from(value: &TrendDataRaw) -> core::result::Result<Self, Self::Error> {
475        let naive_timestamp = NaiveDateTime::parse_from_str(&value.timestamp, "%Y-%m-%d %H:%M:%S");
476        if let Err(e) = naive_timestamp {
477            let timestamp = &value.timestamp;
478            bail!("error parsing {timestamp}: {e}");
479        }
480        let mut naive_timestamp = naive_timestamp.unwrap();
481        match value.timestamp_tzinfo {
482            248 => {
483                // CET
484                naive_timestamp = naive_timestamp - TimeDelta::hours(1);
485            }
486            241 => {
487                // CEST
488                naive_timestamp = naive_timestamp - TimeDelta::hours(2);
489            }
490            _ => {
491                let timezone = value.timestamp_tzinfo;
492                bail!("did not recognize timestamp timezone {timezone}");
493            }
494        }
495
496        let utc_time: DateTime<Utc> = naive_timestamp.and_local_timezone(Utc).unwrap();
497
498        Ok(TrendDataDb {
499            log_id: value.externallogid,
500            timestamp: utc_time,
501            value: value.value,
502        })
503    }
504}
505
506impl From<&Row> for TrendDataDb {
507    fn from(row: &Row) -> TrendDataDb {
508        let log_id: i64 = row.get("log_id");
509        let timestamp: DateTime<Utc> = row.get("timestamp");
510        let value: Option<f64> = row.get("value");
511
512        TrendDataDb {
513            log_id,
514            timestamp,
515            value,
516        }
517    }
518}
aphomie_db_client/lib.rs

aphomie_db_client/
lib.rs
