pinto/storage/file_repository/
items.rs

1//! Backlog-item persistence for [`FileRepository`]: the [`BacklogItemRepository`]
2//! implementation and the item record reading/validation helpers it relies on.
3
4use super::{FileRepository, ItemRecord};
5use crate::backlog::{BacklogItem, ItemId};
6use crate::error::Error;
7use crate::error::Result;
8use crate::storage::atomic_write;
9use crate::storage::issued_ids::{max_number, record, record_many};
10use crate::storage::markdown::{from_markdown, to_markdown};
11use crate::storage::repository::BacklogItemRepository;
12use rayon::prelude::*;
13use std::collections::HashMap;
14use std::io;
15use std::path::{Path, PathBuf};
16use tokio::fs;
17use tokio::task::JoinSet;
18
19/// Keep concurrent file reads below common per-process descriptor limits while retaining async
20/// I/O parallelism. A bounded JoinSet is enough because parsing is performed separately by Rayon.
21const MAX_CONCURRENT_FILE_READS: usize = 64;
22
23impl BacklogItemRepository for FileRepository {
24    async fn save(&self, item: &BacklogItem) -> Result<()> {
25        let (_, archived) = self.read_all_item_records().await?;
26        let dir = self.tasks_dir();
27        fs::create_dir_all(&dir)
28            .await
29            .map_err(|e| Error::io(&dir, &e))?;
30        let path = self.path_for(&item.id)?;
31        if let Some((archive_path, _)) =
32            archived.iter().find(|(_, existing)| existing.id == item.id)
33        {
34            return Err(Error::parse(
35                &path,
36                format!(
37                    "cannot save item `{}`: archive file {} already exists; remove the archived copy before restoring the item",
38                    item.id,
39                    archive_path.display()
40                ),
41            ));
42        }
43        record(&self.root, &item.id).await?;
44        let text = to_markdown(item)?;
45        atomic_write(&path, &text).await
46    }
47
48    async fn load(&self, id: &ItemId) -> Result<BacklogItem> {
49        let (active, _) = self.read_all_item_records().await?;
50        active
51            .into_iter()
52            .find_map(|(_, item)| (item.id == *id).then_some(item))
53            .ok_or_else(|| Error::NotFound(id.clone()))
54    }
55
56    async fn list(&self) -> Result<Vec<BacklogItem>> {
57        let (active, _) = self.read_all_item_records().await?;
58        let mut items = active.into_iter().map(|(_, item)| item).collect::<Vec<_>>();
59
60        // Canonical backlog order (rank asc, ID tie-break) shared with every view.
61        items.sort_by(BacklogItem::backlog_cmp);
62        Ok(items)
63    }
64
65    async fn list_archived(&self) -> Result<Vec<BacklogItem>> {
66        let (_, archived) = self.read_all_item_records().await?;
67        let mut items = archived
68            .into_iter()
69            .map(|(_, item)| item)
70            .collect::<Vec<_>>();
71        items.sort_by(BacklogItem::backlog_cmp);
72        Ok(items)
73    }
74
75    async fn load_archived(&self, id: &ItemId) -> Result<BacklogItem> {
76        let (_, archived) = self.read_all_item_records().await?;
77        archived
78            .into_iter()
79            .find_map(|(_, item)| (item.id == *id).then_some(item))
80            .ok_or_else(|| Error::NotFound(id.clone()))
81    }
82
83    async fn delete(&self, id: &ItemId) -> Result<()> {
84        self.read_all_item_records().await?;
85        let path = self.path_for(id)?;
86        match fs::remove_file(&path).await {
87            Ok(()) => record(&self.root, id).await,
88            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(id.clone())),
89            Err(e) => Err(Error::io(&path, &e)),
90        }
91    }
92
93    async fn archive(&self, id: &ItemId) -> Result<PathBuf> {
94        let src = self.path_for(id)?;
95        let source_exists = fs::try_exists(&src)
96            .await
97            .map_err(|e| Error::io(&src, &e))?;
98        let dest = self.archive_path_for(id)?;
99        let destination_exists = fs::try_exists(&dest)
100            .await
101            .map_err(|e| Error::io(&dest, &e))?;
102        if source_exists && destination_exists {
103            return Err(Error::parse(
104                &dest,
105                format!(
106                    "cannot archive `{id}`: destination already exists at {}; remove the archived copy before retrying",
107                    dest.display()
108                ),
109            ));
110        }
111        self.read_all_item_records().await?;
112        if !source_exists {
113            return Err(Error::NotFound(id.clone()));
114        }
115        let archive_dir = self.archive_dir();
116        fs::create_dir_all(&archive_dir)
117            .await
118            .map_err(|e| Error::io(&archive_dir, &e))?;
119
120        // Rename after validation; map a source that disappears between validation and the move
121        // to `NotFound` without allowing an existing destination to be replaced.
122        match fs::rename(&src, &dest).await {
123            Ok(()) => {
124                record(&self.root, id).await?;
125                Ok(dest)
126            }
127            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(id.clone())),
128            Err(e) => Err(Error::io(&src, &e)),
129        }
130    }
131
132    async fn restore(&self, id: &ItemId) -> Result<()> {
133        let src = self.archive_path_for(id)?;
134        let source_exists = fs::try_exists(&src)
135            .await
136            .map_err(|e| Error::io(&src, &e))?;
137        if !source_exists {
138            return Err(Error::NotFound(id.clone()));
139        }
140
141        let dest = self.path_for(id)?;
142        let destination_exists = fs::try_exists(&dest)
143            .await
144            .map_err(|e| Error::io(&dest, &e))?;
145        if destination_exists {
146            return Err(Error::parse(
147                &dest,
148                format!(
149                    "cannot restore `{id}`: active item already exists at {}; remove or rename the active copy before retrying",
150                    dest.display()
151                ),
152            ));
153        }
154
155        // Validate both stores before moving the archive. The destination check above deliberately
156        // happens first so a collision is reported without letting duplicate IDs obscure it.
157        self.read_all_item_records().await?;
158        let tasks_dir = self.tasks_dir();
159        fs::create_dir_all(&tasks_dir)
160            .await
161            .map_err(|e| Error::io(&tasks_dir, &e))?;
162        match fs::rename(&src, &dest).await {
163            Ok(()) => Ok(()),
164            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(id.clone())),
165            Err(e) => Err(Error::io(&src, &e)),
166        }
167    }
168
169    async fn next_id(&self, prefix: &str) -> Result<ItemId> {
170        // IDs are never reused after issuance. The history file also covers physically deleted IDs
171        // and backend migrations; scanning both directories keeps older boards safe as well.
172        let mut max = max_number(&self.root, prefix).await?;
173        let (active, archived) = self.read_all_item_records().await?;
174        let existing_max = active
175            .iter()
176            .chain(archived.iter())
177            .map(|(_, item)| item.id.clone())
178            .filter(|id| id.prefix() == prefix)
179            .map(|id| id.number())
180            .max()
181            .unwrap_or(0);
182        max = max.max(existing_max);
183        let next = max
184            .checked_add(1)
185            .ok_or_else(|| Error::InvalidItemId(format!("{prefix}-{max}")))?;
186        ItemId::try_new(prefix, next)
187    }
188}
189
190impl FileRepository {
191    /// Save a validated batch without re-reading the growing task set for every item.
192    pub(crate) async fn save_batch(&self, items: &[BacklogItem]) -> Result<()> {
193        let (_, archived) = self.read_all_item_records().await?;
194        let dir = self.tasks_dir();
195        fs::create_dir_all(&dir)
196            .await
197            .map_err(|e| Error::io(&dir, &e))?;
198
199        for item in items {
200            let path = self.path_for(&item.id)?;
201            if let Some((archive_path, _)) =
202                archived.iter().find(|(_, existing)| existing.id == item.id)
203            {
204                return Err(Error::parse(
205                    &path,
206                    format!(
207                        "cannot save item `{}`: archive file {} already exists; remove the archived copy before restoring the item",
208                        item.id,
209                        archive_path.display()
210                    ),
211                ));
212            }
213        }
214
215        record_many(
216            &self.root,
217            &items.iter().map(|item| item.id.clone()).collect::<Vec<_>>(),
218        )
219        .await?;
220        for item in items {
221            let path = self.path_for(&item.id)?;
222            let text = to_markdown(item)?;
223            atomic_write(&path, &text).await?;
224        }
225        Ok(())
226    }
227
228    /// Read and validate every active and archived item before exposing the active set.
229    async fn read_all_item_records(&self) -> Result<(Vec<ItemRecord>, Vec<ItemRecord>)> {
230        let active = self.read_item_records(&self.tasks_dir()).await?;
231        let archived = self.read_item_records(&self.archive_dir()).await?;
232        Self::ensure_unique_item_ids(active.iter().chain(archived.iter()))?;
233        Ok((active, archived))
234    }
235
236    /// Read item files from one directory, validate their logical IDs, and retain their paths for
237    /// actionable corruption diagnostics and cross-directory collision checks.
238    async fn read_item_records(&self, dir: &Path) -> Result<Vec<ItemRecord>> {
239        let Some(paths) = self.markdown_paths(dir).await? else {
240            return Ok(Vec::new());
241        };
242
243        // Read all files concurrently; this is I/O-bound work.
244        let mut reads = JoinSet::new();
245        let mut contents = Vec::with_capacity(paths.len());
246        for path in paths {
247            if reads.len() >= MAX_CONCURRENT_FILE_READS
248                && let Some(joined) = reads.join_next().await
249            {
250                contents.push(joined.map_err(Error::task)??);
251            }
252            reads.spawn(async move {
253                fs::read_to_string(&path)
254                    .await
255                    .map_err(|e| Error::io(&path, &e))
256                    .map(|text| (path, text))
257            });
258        }
259        while let Some(joined) = reads.join_next().await {
260            contents.push(joined.map_err(Error::task)??);
261        }
262
263        // Parse frontmatter in parallel; this is CPU-bound work.
264        let records = contents
265            .into_par_iter()
266            .map(|(path, text)| from_markdown(&text, &path).map(|item| (path, item)))
267            .collect::<Result<Vec<_>>>()?;
268
269        // Report duplicate logical IDs before filename mismatches so the error names both records
270        // when a copied file creates an ambiguous lookup key.
271        Self::ensure_unique_item_ids(records.iter())?;
272        for (path, item) in &records {
273            Self::validate_item_filename(path, item)?;
274        }
275        Ok(records)
276    }
277
278    /// Reject two files that resolve to the same logical item ID.
279    fn ensure_unique_item_ids<'a>(records: impl IntoIterator<Item = &'a ItemRecord>) -> Result<()> {
280        let mut seen = HashMap::new();
281        for (path, item) in records {
282            if let Some(previous) = seen.insert(item.id.clone(), path.clone()) {
283                return Err(Error::parse(
284                    path,
285                    format!(
286                        "duplicate item ID `{}` in {} and {}; fix one frontmatter ID or rename one file",
287                        item.id,
288                        previous.display(),
289                        path.display()
290                    ),
291                ));
292            }
293        }
294        Ok(())
295    }
296
297    /// Ensure an item's filename stem and frontmatter ID describe the same record.
298    fn validate_item_filename(path: &Path, item: &BacklogItem) -> Result<()> {
299        let stem = path
300            .file_stem()
301            .and_then(|value| value.to_str())
302            .ok_or_else(|| {
303                Error::parse(
304                    path,
305                    "item filename must be a UTF-8 `<PREFIX>-<NUMBER>.md`; rename the file",
306                )
307            })?;
308        let filename_id = stem.parse::<ItemId>().map_err(|error| {
309            Error::parse(
310                path,
311                format!("invalid item filename `{stem}.md`: {error}; rename the file to `<ID>.md`"),
312            )
313        })?;
314        if filename_id != item.id {
315            return Err(Error::parse(
316                path,
317                format!(
318                    "filename ID `{filename_id}` does not match frontmatter ID `{}`; rename the file or fix its frontmatter",
319                    item.id
320                ),
321            ));
322        }
323        Ok(())
324    }
325}
pinto/storage/file_repository/items.rs

pinto/storage/file_repository/
items.rs
