monorepo/emseries/src/series.rs

/*
Copyright 2020-2023, Savanni D'Gerinel <savanni@luminescent-dreams.com>

This file is part of the Luminescent Dreams Tools.

Luminescent Dreams Tools is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

Luminescent Dreams Tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with Lumeto. If not, see <https://www.gnu.org/licenses/>.
*/

extern crate serde;
extern crate serde_json;
extern crate uuid;

use serde::de::DeserializeOwned;
use serde::ser::Serialize;
use std::cmp::Ordering;
use std::collections::HashMap;
use std::convert::TryFrom;
use std::fs::File;
use std::fs::OpenOptions;
use std::io::{BufRead, BufReader, LineWriter, Write};
use std::iter::Iterator;

use criteria::Criteria;
use types::{EmseriesReadError, EmseriesWriteError, Record, RecordId, Recordable};

// A RecordOnDisk, a private data structure, is useful for handling all of the on-disk
// representations of a record. Unlike [Record], this one can accept an empty data value to
// represent that the data may have been deleted. This is not made public because, so far as the
// user is concerned, any record in the system must have data associated with it.
#[derive(Clone, Deserialize, Serialize)]
struct RecordOnDisk<T: Clone + Recordable> {
    id: RecordId,
    data: Option<T>,
}

/*
impl<T> FromStr for RecordOnDisk<T>
where
    T: Clone + Recordable + DeserializeOwned + Serialize,
{
    type Err = EmseriesReadError;

    fn from_str(line: &str) -> Result<Self, Self::Err> {
        serde_json::from_str(line).map_err(EmseriesReadError::JSONParseError)
    }
}
*/

impl<T: Clone + Recordable> TryFrom<RecordOnDisk<T>> for Record<T> {
    type Error = EmseriesReadError;

    fn try_from(disk_record: RecordOnDisk<T>) -> Result<Self, Self::Error> {
        match disk_record.data {
            Some(data) => Ok(Record {
                id: disk_record.id,
                data,
            }),
            None => Err(Self::Error::RecordDeleted(disk_record.id)),
        }
    }
}

/// An open time series database.
///
/// Any given database can store only one data type, T. The data type must be determined when the
/// database is opened.
pub struct Series<T: Clone + Recordable + DeserializeOwned + Serialize> {
    //path: String,
    writer: LineWriter<File>,
    records: HashMap<RecordId, Record<T>>,
}

impl<T> Series<T>
where
    T: Clone + Recordable + DeserializeOwned + Serialize,
{
    /// Open a time series database at the specified path. `path` is the full path and filename for
    /// the database.
    pub fn open<P: AsRef<std::path::Path>>(path: P) -> Result<Series<T>, EmseriesReadError> {
        let f = OpenOptions::new()
            .read(true)
            .append(true)
            .create(true)
            .open(path)
            .map_err(EmseriesReadError::IOError)?;

        let records = Series::load_file(&f)?;

        let writer = LineWriter::new(f);

        Ok(Series {
            //path: String::from(path),
            writer,
            records,
        })
    }

    /// Load a file and return all of the records in it.
    fn load_file(f: &File) -> Result<HashMap<RecordId, Record<T>>, EmseriesReadError> {
        let mut records: HashMap<RecordId, Record<T>> = HashMap::new();
        let reader = BufReader::new(f);
        for line in reader.lines() {
            match line {
                Ok(line_) => {
                    match serde_json::from_str::<RecordOnDisk<T>>(line_.as_ref())
                        .map_err(EmseriesReadError::JSONParseError)
                        .and_then(Record::try_from)
                    {
                        Ok(record) => records.insert(record.id, record.clone()),
                        Err(EmseriesReadError::RecordDeleted(id)) => records.remove(&id),
                        Err(err) => return Err(err),
                    };
                }
                Err(err) => return Err(EmseriesReadError::IOError(err)),
            }
        }
        Ok(records)
    }

    /// Put a new record into the database. A unique id will be assigned to the record and
    /// returned.
    pub fn put(&mut self, entry: T) -> Result<RecordId, EmseriesWriteError> {
        let id = RecordId::default();
        let record = Record { id, data: entry };
        self.update(record)?;
        Ok(id)
    }

    /// Update an existing record. The [RecordId] of the record passed into this function must match
    /// the [RecordId] of a record already in the database.
    pub fn update(&mut self, record: Record<T>) -> Result<(), EmseriesWriteError> {
        self.records.insert(record.id, record.clone());
        let write_res = match serde_json::to_string(&RecordOnDisk {
            id: record.id,
            data: Some(record.data),
        }) {
            Ok(rec_str) => self
                .writer
                .write_fmt(format_args!("{}\n", rec_str.as_str()))
                .map_err(EmseriesWriteError::IOError),
            Err(err) => Err(EmseriesWriteError::JSONWriteError(err)),
        };

        match write_res {
            Ok(_) => Ok(()),
            Err(err) => Err(err),
        }
    }

    /// Delete a record from the database
    ///
    /// Future note: while this deletes a record from the view, it only adds an entry to the
    /// database that indicates `data: null`. If record histories ever become important, the record
    /// and its entire history (including this delete) will still be available.
    pub fn delete(&mut self, uuid: &RecordId) -> Result<(), EmseriesWriteError> {
        if !self.records.contains_key(uuid) {
            return Ok(());
        };
        self.records.remove(uuid);

        let rec: RecordOnDisk<T> = RecordOnDisk {
            id: *uuid,
            data: None,
        };
        match serde_json::to_string(&rec) {
            Ok(rec_str) => self
                .writer
                .write_fmt(format_args!("{}\n", rec_str.as_str()))
                .map_err(EmseriesWriteError::IOError),
            Err(err) => Err(EmseriesWriteError::JSONWriteError(err)),
        }
    }

    /// Get all of the records in the database.
    pub fn records(&self) -> impl Iterator<Item = &Record<T>> {
        self.records.values()
    }

    /*  The point of having Search is so that a lot of internal optimizations can happen once the
     *  data sets start getting large. */
    /// Perform a search on the records in a database, based on the given criteria.
    pub fn search<'s>(
        &'s self,
        criteria: impl Criteria + 's,
    ) -> impl Iterator<Item = &'s Record<T>> + 's {
        self.records().filter(move |&tr| criteria.apply(&tr.data))
    }

    /// Perform a search and sort the resulting records based on the comparison.
    pub fn search_sorted<'s, C, CMP>(&'s self, criteria: C, compare: CMP) -> Vec<&'s Record<T>>
    where
        C: Criteria + 's,
        CMP: FnMut(&&Record<T>, &&Record<T>) -> Ordering,
    {
        let search_iter = self.search(criteria);
        let mut records: Vec<&Record<T>> = search_iter.collect();
        records.sort_by(compare);
        records
    }

    /// Get an exact record from the database based on unique id.
    pub fn get(&self, uuid: &RecordId) -> Option<Record<T>> {
        self.records.get(uuid).cloned()
    }

    /*
    pub fn remove(&self, uuid: RecordId) -> Result<(), EmseriesError> {
        unimplemented!()
    }
    */
}