4829321272

Build Type

github

Committed by GitHub

Commit Message

Merge b6d982211 into f2ab0e6ce

Pull Request Pull Request #1515: feat: Run migration only when necessary

Run Details

193 of 193 new or added lines in 11 files covered. (100.0%)

35565 of 45252 relevant lines covered (78.59%)

16462.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.73

/dozer-log/src/home_dir.rs

use std::path::{Path, PathBuf};

#[derive(Debug, Clone)]
pub struct HomeDir {
    api_dir: PathBuf,
    cache_dir: PathBuf,
    log_dir: PathBuf,
}

pub type Error = (PathBuf, std::io::Error);

impl HomeDir {
    pub fn new(home_dir: &Path, cache_dir: PathBuf) -> Self {
        let api_dir = home_dir.join("api");
        let log_dir = home_dir.join("pipeline").join("logs");
        Self {
            api_dir,
            cache_dir,
            log_dir,
        }
    }

    pub fn create_migration_dir_all(
        &self,
        endpoint_name: &str,
        migration_id: MigrationId,
    ) -> Result<MigrationPath, Error> {
        std::fs::create_dir_all(&self.cache_dir).map_err(|e| (self.cache_dir.clone(), e))?;

        let migration_path = self.get_migration_path(endpoint_name, migration_id);

        std::fs::create_dir_all(&migration_path.api_dir)
            .map_err(|e| (migration_path.api_dir.clone(), e))?;
        std::fs::create_dir_all(&migration_path.log_dir)
            .map_err(|e: std::io::Error| (migration_path.log_dir.clone(), e))?;

        Ok(migration_path)
    }

    pub fn find_latest_migration_path(
        &self,
        endpoint_name: &str,
    ) -> Result<Option<MigrationPath>, Error> {
        Ok(self
            .find_latest_migration_id(endpoint_name)?
            .map(|migration_id| self.get_migration_path(endpoint_name, migration_id)))
    }

    fn find_latest_migration_id(&self, endpoint_name: &str) -> Result<Option<MigrationId>, Error> {
        let api_dir = self.get_endpoint_api_dir(endpoint_name);
        let migration1 = find_latest_migration_id(&api_dir)?;
        let log_dir = self.get_endpoint_log_dir(endpoint_name);
        let migration2 = find_latest_migration_id(&log_dir)?;

        match (migration1, migration2) {
            (Some(migration1), Some(migration2)) => {
                if migration1.id > migration2.id {
                    Ok(Some(migration1))
                } else {
                    Ok(Some(migration2))
                }
            }
            (Some(migration1), None) => Ok(Some(migration1)),
            (None, Some(migration2)) => Ok(Some(migration2)),
            (None, None) => Ok(None),
        }
    }

    fn get_migration_path(&self, endpoint_name: &str, migration_id: MigrationId) -> MigrationPath {
        let api_dir = self
            .get_endpoint_api_dir(endpoint_name)
            .join(&migration_id.name);
        let descriptor_path = api_dir.join("file_descriptor_set.bin");
        let log_dir = self
            .get_endpoint_log_dir(endpoint_name)
            .join(&migration_id.name);
        let schema_path = log_dir.join("schema.json");
        let log_path = log_dir.join("log");
        MigrationPath {
            id: migration_id,
            api_dir,
            descriptor_path,
            log_dir,
            schema_path,
            log_path,
        }
    }

    fn get_endpoint_api_dir(&self, endpoint_name: &str) -> PathBuf {
        self.api_dir.join(endpoint_name)
    }

    fn get_endpoint_log_dir(&self, endpoint_name: &str) -> PathBuf {
        self.log_dir.join(endpoint_name)
    }
}

#[derive(Debug, Clone)]
pub struct MigrationId {
    id: u32,
    name: String,
}

impl MigrationId {
    fn from_id(id: u32) -> Self {
        Self {
            id,
            name: format!("v{id:04}"),
        }
    }

    fn from_name(name: &str) -> Option<Self> {
        let id = name.strip_prefix('v').and_then(|s| s.parse::<u32>().ok())?;
        Some(Self {
            id,
            name: name.to_string(),
        })
    }

    pub fn first() -> Self {
        Self::from_id(1)
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn next(&self) -> Self {
        Self::from_id(self.id + 1)
    }
}

fn find_latest_migration_id(dir: &Path) -> Result<Option<MigrationId>, Error> {
    if !dir.exists() {
        return Ok(None);
    }

    let mut result = None;
    for entry in std::fs::read_dir(dir).map_err(|e| (dir.to_path_buf(), e))? {
        let entry = entry.map_err(|e| (dir.to_path_buf(), e))?;
        if entry.path().is_dir() {
            if let Some(file_name) = entry.file_name().to_str() {
                if let Some(migration) = MigrationId::from_name(file_name) {
                    if let Some(MigrationId { id, .. }) = result {
                        if migration.id > id {
                            result = Some(migration);
                        }
                    } else {
                        result = Some(migration);
                    }
                }
            }
        }
    }
    Ok(result)
}

#[derive(Debug, Clone)]
pub struct MigrationPath {
    pub id: MigrationId,
    pub api_dir: PathBuf,
    pub descriptor_path: PathBuf,
    log_dir: PathBuf,
    pub schema_path: PathBuf,
    pub log_path: PathBuf,
}

1	use std::path::{Path, PathBuf};
2
3	#[derive(Debug, Clone)]	×
4	pub struct HomeDir {
5	api_dir: PathBuf,
6	cache_dir: PathBuf,
7	log_dir: PathBuf,
8	}
9
10	pub type Error = (PathBuf, std::io::Error);
11
12	impl HomeDir {
13	pub fn new(home_dir: &Path, cache_dir: PathBuf) -> Self {	48✔
14	let api_dir = home_dir.join("api");	48✔
15	let log_dir = home_dir.join("pipeline").join("logs");	48✔
16	Self {	48✔
17	api_dir,	48✔
18	cache_dir,	48✔
19	log_dir,	48✔
20	}	48✔
21	}	48✔
22
23	pub fn create_migration_dir_all(
24	&self,
25	endpoint_name: &str,
26	migration_id: MigrationId,
27	) -> Result<MigrationPath, Error> {
28	std::fs::create_dir_all(&self.cache_dir).map_err(\|e\| (self.cache_dir.clone(), e))?;	18✔
29
30	let migration_path = self.get_migration_path(endpoint_name, migration_id);	18✔
31		18✔
32	std::fs::create_dir_all(&migration_path.api_dir)	18✔
33	.map_err(\|e\| (migration_path.api_dir.clone(), e))?;	18✔
34	std::fs::create_dir_all(&migration_path.log_dir)	18✔
35	.map_err(\|e: std::io::Error\| (migration_path.log_dir.clone(), e))?;	18✔
36
37	Ok(migration_path)	18✔
38	}	18✔
39
40	pub fn find_latest_migration_path(	54✔
41	&self,	54✔
42	endpoint_name: &str,	54✔
43	) -> Result<Option<MigrationPath>, Error> {	54✔
44	Ok(self	54✔
45	.find_latest_migration_id(endpoint_name)?	54✔
46	.map(\|migration_id\| self.get_migration_path(endpoint_name, migration_id)))	54✔
47	}	54✔
48
49	fn find_latest_migration_id(&self, endpoint_name: &str) -> Result<Option<MigrationId>, Error> {	54✔
50	let api_dir = self.get_endpoint_api_dir(endpoint_name);	54✔
51	let migration1 = find_latest_migration_id(&api_dir)?;	54✔
52	let log_dir = self.get_endpoint_log_dir(endpoint_name);	54✔
53	let migration2 = find_latest_migration_id(&log_dir)?;	54✔
54
55	match (migration1, migration2) {	54✔
56	(Some(migration1), Some(migration2)) => {	36✔
57	if migration1.id > migration2.id {	36✔
58	Ok(Some(migration1))	×
59	} else {
60	Ok(Some(migration2))	36✔
61	}
62	}
63	(Some(migration1), None) => Ok(Some(migration1)),	×
64	(None, Some(migration2)) => Ok(Some(migration2)),	×
65	(None, None) => Ok(None),	18✔
66	}
67	}	54✔
68
69	fn get_migration_path(&self, endpoint_name: &str, migration_id: MigrationId) -> MigrationPath {	54✔
70	let api_dir = self	54✔
71	.get_endpoint_api_dir(endpoint_name)	54✔
72	.join(&migration_id.name);	54✔
73	let descriptor_path = api_dir.join("file_descriptor_set.bin");	54✔
74	let log_dir = self	54✔
75	.get_endpoint_log_dir(endpoint_name)	54✔
76	.join(&migration_id.name);	54✔
77	let schema_path = log_dir.join("schema.json");	54✔
78	let log_path = log_dir.join("log");	54✔
79	MigrationPath {	54✔
80	id: migration_id,	54✔
81	api_dir,	54✔
82	descriptor_path,	54✔
83	log_dir,	54✔
84	schema_path,	54✔
85	log_path,	54✔
86	}	54✔
87	}	54✔
88
89	fn get_endpoint_api_dir(&self, endpoint_name: &str) -> PathBuf {	108✔
90	self.api_dir.join(endpoint_name)	108✔
91	}	108✔
92
93	fn get_endpoint_log_dir(&self, endpoint_name: &str) -> PathBuf {	108✔
94	self.log_dir.join(endpoint_name)	108✔
95	}	108✔
96	}
97
98	#[derive(Debug, Clone)]	×
99	pub struct MigrationId {
100	id: u32,
101	name: String,
102	}
103
104	impl MigrationId {
105	fn from_id(id: u32) -> Self {	18✔
106	Self {	18✔
107	id,	18✔
108	name: format!("v{id:04}"),	18✔
109	}	18✔
110	}	18✔
111
112	fn from_name(name: &str) -> Option<Self> {	72✔
113	let id = name.strip_prefix('v').and_then(\|s\| s.parse::<u32>().ok())?;	72✔
114	Some(Self {	72✔
115	id,	72✔
116	name: name.to_string(),	72✔
117	})	72✔
118	}	72✔
119
120	pub fn first() -> Self {	18✔
121	Self::from_id(1)	18✔
122	}	18✔
123
124	pub fn name(&self) -> &str {	18✔
125	&self.name	18✔
126	}	18✔
127
128	pub fn next(&self) -> Self {	×
129	Self::from_id(self.id + 1)	×
130	}	×
131	}
132
133	fn find_latest_migration_id(dir: &Path) -> Result<Option<MigrationId>, Error> {	108✔
134	if !dir.exists() {	108✔
135	return Ok(None);	36✔
136	}	72✔
137		72✔
138	let mut result = None;	72✔
139	for entry in std::fs::read_dir(dir).map_err(\|e\| (dir.to_path_buf(), e))? {	72✔
140	let entry = entry.map_err(\|e\| (dir.to_path_buf(), e))?;	72✔
141	if entry.path().is_dir() {	72✔
142	if let Some(file_name) = entry.file_name().to_str() {	72✔
143	if let Some(migration) = MigrationId::from_name(file_name) {	72✔
144	if let Some(MigrationId { id, .. }) = result {	72✔
145	if migration.id > id {	×
146	result = Some(migration);	×
147	}	×
148	} else {	72✔
149	result = Some(migration);	72✔
150	}	72✔
151	}	×
152	}	×
153	}	×
154	}
155	Ok(result)	72✔
156	}	108✔
157
158	#[derive(Debug, Clone)]	×
159	pub struct MigrationPath {
160	pub id: MigrationId,
161	pub api_dir: PathBuf,
162	pub descriptor_path: PathBuf,
163	log_dir: PathBuf,
164	pub schema_path: PathBuf,
165	pub log_path: PathBuf,
166	}

getdozer / dozer / 4829321272

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous