Skip to main content

icechunk_format/
repo_info.rs

1//! Version info, branches, and tags for a repository.
2
3use itertools::Itertools as _;
4use serde::{Deserialize, Serialize};
5use std::{
6    borrow::Cow,
7    collections::{BTreeSet, HashMap},
8};
9use tracing::trace;
10
11use crate::{
12    IcechunkFormatError, IcechunkFormatErrorKind, IcechunkResult, SnapshotId,
13    flatbuffers::generated,
14    format_constants::SpecVersionBin,
15    lookup_index_by_key,
16    snapshot::{SnapshotInfo, SnapshotProperties},
17};
18
19use chrono::{DateTime, Utc};
20use flatbuffers::{UnionWIPOffset, VerifierOptions, WIPOffset};
21use icechunk_types::ICResultExt as _;
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
24pub enum RepoAvailability {
25    Online,
26    ReadOnly,
27    // Offline is defined in the flatbuffers, but we won't support it
28    // before better specs on how we want to use it
29}
30
31impl From<generated::RepoAvailability> for RepoAvailability {
32    fn from(value: generated::RepoAvailability) -> Self {
33        match value {
34            generated::RepoAvailability::Online => RepoAvailability::Online,
35            generated::RepoAvailability::ReadOnly => RepoAvailability::ReadOnly,
36            _ => RepoAvailability::Online,
37        }
38    }
39}
40
41impl From<RepoAvailability> for generated::RepoAvailability {
42    fn from(value: RepoAvailability) -> Self {
43        match value {
44            RepoAvailability::Online => generated::RepoAvailability::Online,
45            RepoAvailability::ReadOnly => generated::RepoAvailability::ReadOnly,
46        }
47    }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
51pub struct RepoStatus {
52    pub availability: RepoAvailability,
53    pub set_at: DateTime<Utc>,
54    pub limited_availability_reason: Option<String>,
55}
56
57impl TryFrom<generated::RepoStatus<'_>> for RepoStatus {
58    type Error = IcechunkFormatError;
59
60    fn try_from(fb_status: generated::RepoStatus<'_>) -> Result<Self, Self::Error> {
61        let ts: i64 = fb_status
62            .set_at()
63            .try_into()
64            .map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp)
65            .capture()?;
66        let set_at = DateTime::from_timestamp_micros(ts)
67            .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
68            .capture()?;
69        Ok(RepoStatus {
70            availability: fb_status.availability().into(),
71            set_at,
72            limited_availability_reason: fb_status
73                .limited_availability_reason()
74                .map(|s| s.to_string()),
75        })
76    }
77}
78
79impl RepoStatus {
80    pub fn error_msg(&self) -> String {
81        format!(
82            "Repo status is {0:?}, set at {1}, reason: {2:?}",
83            self.availability, self.set_at, self.limited_availability_reason
84        )
85    }
86}
87
88// TODO: should we not implement serialize and let the session fetch the repo info?
89#[derive(PartialEq, Serialize, Deserialize)]
90pub struct RepoInfo {
91    buffer: Vec<u8>,
92}
93
94impl std::fmt::Debug for RepoInfo {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        let tags = self.tags().map(Vec::from_iter).unwrap_or_default();
97        let tags =
98            tags.into_iter().map(|(name, snap)| format!("{name} -> {snap}")).join(", ");
99        let branches = self.branches().map(Vec::from_iter).unwrap_or_default();
100        let branches = branches
101            .into_iter()
102            .map(|(name, snap)| format!("{name} -> {snap}"))
103            .join(", ");
104        let snaps = self.all_snapshots().map(Vec::from_iter).unwrap_or_default();
105        let snaps = snaps
106            .into_iter()
107            .map(|ms| match ms {
108                Ok(snap) => format!(
109                    "{} -> {}",
110                    snap.id,
111                    snap.parent_id.map(|s| s.to_string()).unwrap_or_default()
112                ),
113                Err(_) => "#err".to_string(),
114            })
115            .join(", ");
116        // FIXME: add other fields
117        f.debug_struct("RepoInfo")
118            .field("tags", &tags)
119            .field("branches", &branches)
120            .field("snapshots", &snaps)
121            .finish_non_exhaustive()
122    }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
126pub enum UpdateType {
127    RepoInitializedUpdate,
128    RepoMigratedUpdate {
129        from_version: SpecVersionBin,
130        to_version: SpecVersionBin,
131    },
132    RepoStatusChangedUpdate {
133        status: RepoStatus,
134    },
135    ConfigChangedUpdate,
136    MetadataChangedUpdate,
137    TagCreatedUpdate {
138        name: String,
139    },
140    TagDeletedUpdate {
141        name: String,
142        previous_snap_id: SnapshotId,
143    },
144    BranchCreatedUpdate {
145        name: String,
146    },
147    BranchDeletedUpdate {
148        name: String,
149        previous_snap_id: SnapshotId,
150    },
151    BranchResetUpdate {
152        name: String,
153        previous_snap_id: SnapshotId,
154    },
155    NewCommitUpdate {
156        branch: String,
157        new_snap_id: SnapshotId,
158    },
159    CommitAmendedUpdate {
160        branch: String,
161        previous_snap_id: SnapshotId,
162        new_snap_id: SnapshotId,
163    },
164    NewDetachedSnapshotUpdate {
165        new_snap_id: SnapshotId,
166    },
167    GCRanUpdate,
168    ExpirationRanUpdate,
169    FeatureFlagChanged {
170        id: u16,
171        new_value: Option<bool>,
172    },
173}
174
175static ROOT_OPTIONS: VerifierOptions = VerifierOptions {
176    max_depth: 10,
177    max_tables: 5_000_000,
178    max_apparent_size: 1 << 31, // taken from the default
179    ignore_missing_null_terminator: true,
180};
181
182#[derive(Debug, Clone)]
183pub struct UpdateInfo<I> {
184    pub update_type: UpdateType,
185    pub update_time: DateTime<Utc>,
186    pub previous_updates: I,
187}
188
189type UpdateTuple<'a> = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>;
190
191type LatestUpdatesResult<'bldr> = IcechunkResult<(
192    WIPOffset<
193        flatbuffers::Vector<
194            'bldr,
195            flatbuffers::ForwardsUOffset<generated::Update<'bldr>>,
196        >,
197    >,
198    Option<WIPOffset<&'bldr str>>,
199)>;
200
201impl RepoInfo {
202    #[expect(clippy::too_many_arguments)]
203    pub fn new<
204        'a,
205        I: IntoIterator<Item = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>>,
206        EFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
207        DFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
208    >(
209        spec_version: SpecVersionBin,
210        tags: impl IntoIterator<Item = (&'a str, SnapshotId)>,
211        branches: impl IntoIterator<Item = (&'a str, SnapshotId)>,
212        deleted_tags: impl IntoIterator<Item = &'a str>,
213        snapshots: impl IntoIterator<Item = SnapshotInfo>,
214        metadata: &SnapshotProperties,
215        update: UpdateInfo<I>,
216        backup_path: Option<&'a str>,
217        num_updates_per_file: u16,
218        previous_info: Option<&'a str>,
219        config_bytes: Option<&[u8]>,
220        sorted_enabled_feature_flags: Option<EFFIt>,
221        sorted_disabled_feature_flags: Option<DFFIt>,
222        status: &RepoStatus,
223    ) -> IcechunkResult<Self> {
224        let mut snapshots: Vec<_> = snapshots.into_iter().collect();
225        snapshots.sort_by(|a, b| a.id.0.cmp(&b.id.0));
226        let tags = resolve_ref_iter(&snapshots, tags)?;
227        let branches = resolve_ref_iter(&snapshots, branches)?;
228        let mut deleted_tags: Vec<_> = deleted_tags.into_iter().collect();
229        deleted_tags.sort();
230        Self::from_parts(
231            spec_version,
232            tags,
233            branches,
234            deleted_tags,
235            snapshots,
236            metadata,
237            |builder| {
238                Self::mk_latest_updates(
239                    builder,
240                    update,
241                    backup_path,
242                    num_updates_per_file,
243                    previous_info,
244                )
245            },
246            config_bytes,
247            sorted_enabled_feature_flags,
248            sorted_disabled_feature_flags,
249            status,
250        )
251    }
252
253    #[expect(clippy::too_many_arguments)]
254    fn from_parts<
255        'a,
256        F: for<'bldr> FnOnce(
257            &mut flatbuffers::FlatBufferBuilder<'bldr>,
258        ) -> LatestUpdatesResult<'bldr>,
259        EFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
260        DFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
261    >(
262        spec_version: SpecVersionBin,
263        sorted_tags: impl IntoIterator<Item = (&'a str, u32)>,
264        sorted_branches: impl IntoIterator<Item = (&'a str, u32)>,
265        sorted_deleted_tags: impl IntoIterator<Item = &'a str>,
266        sorted_snapshots: impl IntoIterator<Item = SnapshotInfo>,
267        metadata: &SnapshotProperties,
268        build_updates: F,
269        config_bytes: Option<&[u8]>,
270        sorted_enabled_feature_flags: Option<EFFIt>,
271        sorted_disabled_feature_flags: Option<DFFIt>,
272        status: &RepoStatus,
273    ) -> IcechunkResult<Self> {
274        trace!("Creating new repo info from parts");
275        let mut builder = flatbuffers::FlatBufferBuilder::with_capacity(4_096);
276        let tags = sorted_tags
277            .into_iter()
278            .map(|(name, offset)| {
279                let args = generated::RefArgs {
280                    name: Some(builder.create_string(name)),
281                    snapshot_index: offset,
282                };
283                generated::Ref::create(&mut builder, &args)
284            })
285            .collect::<Vec<_>>();
286        let tags = builder.create_vector(&tags);
287
288        let mut main_found = false;
289        let branches = sorted_branches
290            .into_iter()
291            .map(|(name, offset)| {
292                if name == icechunk_types::DEFAULT_BRANCH {
293                    main_found = true;
294                }
295                let args = generated::RefArgs {
296                    name: Some(builder.create_string(name)),
297                    snapshot_index: offset,
298                };
299                generated::Ref::create(&mut builder, &args)
300            })
301            .collect::<Vec<_>>();
302        if !main_found {
303            return Err(IcechunkFormatErrorKind::BranchNotFound {
304                branch: icechunk_types::DEFAULT_BRANCH.to_string(),
305            })
306            .capture();
307        }
308
309        let branches = builder.create_vector(&branches);
310
311        let deleted_tags = sorted_deleted_tags
312            .into_iter()
313            .map(|name| builder.create_string(name))
314            .collect::<Vec<_>>();
315        let deleted_tags = builder.create_vector(&deleted_tags);
316
317        let snapshots: Vec<_> = sorted_snapshots.into_iter().collect();
318        debug_assert!(snapshots.is_sorted_by(|a, b| a.id.0 <= b.id.0));
319
320        let snapshot_index: HashMap<_, _> =
321            snapshots.iter().enumerate().map(|(ix, sn)| (sn.id.clone(), ix)).collect();
322
323        // TODO: we should check no loops
324        let snapshots: Vec<_> = snapshots
325            .iter()
326            .map(|snap| {
327                let id = &snap.id.0;
328                let id = generated::ObjectId12::new(id);
329                let parent_offset = match snap.parent_id.as_ref() {
330                    Some(parent_id) => {
331                        let index = snapshot_index
332                            .get(parent_id)
333                            .ok_or_else(|| IcechunkFormatErrorKind::SnapshotIdNotFound {
334                                snapshot_id: snap.id.clone(),
335                            })
336                            .capture()?;
337                        Ok(*index as i32)
338                    }
339                    None => Ok::<_, IcechunkFormatError>(-1),
340                }?;
341
342                let metadata_items: Vec<_> = snap
343                    .metadata
344                    .iter()
345                    .map(|(k, v)| {
346                        let name = builder.create_shared_string(k.as_str());
347                        let serialized =
348                            flexbuffers::to_vec(v).map_err(Box::new).capture()?;
349                        let value = builder.create_vector(serialized.as_slice());
350                        let item = generated::MetadataItem::create(
351                            &mut builder,
352                            &generated::MetadataItemArgs {
353                                name: Some(name),
354                                value: Some(value),
355                            },
356                        );
357                        Ok::<_, IcechunkFormatError>(item)
358                    })
359                    .try_collect()?;
360
361                let metadata = builder.create_vector(metadata_items.as_slice());
362                let args = generated::SnapshotInfoArgs {
363                    id: Some(&id),
364                    parent_offset,
365                    flushed_at: snap.flushed_at.timestamp_micros() as u64,
366                    message: Some(builder.create_string(snap.message.as_str())),
367                    metadata: Some(metadata),
368                };
369                Ok::<_, IcechunkFormatError>(generated::SnapshotInfo::create(
370                    &mut builder,
371                    &args,
372                ))
373            })
374            .try_collect()?;
375        let snapshots = builder.create_vector(&snapshots);
376
377        let limited_reason = status
378            .limited_availability_reason
379            .as_deref()
380            .map(|s| builder.create_string(s));
381        let status = generated::RepoStatus::create(
382            &mut builder,
383            &generated::RepoStatusArgs {
384                availability: status.availability.into(),
385                set_at: status.set_at.timestamp_micros() as u64,
386                limited_availability_reason: limited_reason,
387            },
388        );
389
390        let metadata_items: Vec<_> = metadata
391            .iter()
392            .map(|(k, v)| {
393                let name = builder.create_shared_string(k.as_str());
394                let serialized = flexbuffers::to_vec(v).map_err(Box::new).capture()?;
395                let value = builder.create_vector(serialized.as_slice());
396                let item = generated::MetadataItem::create(
397                    &mut builder,
398                    &generated::MetadataItemArgs { name: Some(name), value: Some(value) },
399                );
400                Ok::<_, IcechunkFormatError>(item)
401            })
402            .try_collect()?;
403
404        let metadata = builder.create_vector(metadata_items.as_slice());
405
406        let enabled_feature_flags =
407            sorted_enabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
408        let disabled_feature_flags =
409            sorted_disabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
410
411        let (latest_updates, repo_before_updates) = build_updates(&mut builder)?;
412
413        let config = config_bytes.map(|bytes| builder.create_vector(bytes));
414
415        // TODO: provide accessors for last_updated_at, status, metadata, etc.
416        let repo_args = generated::RepoArgs {
417            tags: Some(tags),
418            branches: Some(branches),
419            deleted_tags: Some(deleted_tags),
420            snapshots: Some(snapshots),
421            spec_version: spec_version as u8,
422            status: Some(status),
423            metadata: Some(metadata),
424            latest_updates: Some(latest_updates),
425            repo_before_updates,
426            config,
427            enabled_feature_flags,
428            disabled_feature_flags,
429            ..Default::default()
430        };
431        let repo = generated::Repo::create(&mut builder, &repo_args);
432        builder.finish(repo, Some("Ichk"));
433        let (mut buffer, offset) = builder.collapse();
434        buffer.drain(0..offset);
435        buffer.shrink_to_fit();
436        Ok(Self { buffer })
437    }
438
439    fn mk_latest_updates<
440        'bldr,
441        'a,
442        I: IntoIterator<Item = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>>,
443    >(
444        builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
445        update: UpdateInfo<I>,
446        backup_path: Option<&'a str>,
447        num_updates_per_file: u16,
448        previous_info: Option<&'a str>,
449    ) -> LatestUpdatesResult<'bldr> {
450        // replace the backup path in the last update, that must be None, by the new backup path
451        let mut previous_updates = update.previous_updates.into_iter();
452        let last_update = previous_updates.next().map(|maybe_data| {
453            maybe_data.map(|(ut, dt, path)| {
454                assert!(
455                    path.is_none(),
456                    "Invalid latest update iterator, last element has backup path"
457                );
458                (ut, dt, backup_path)
459            })
460        });
461        // A backup_path points to the previous repo info file. It is only meaningful
462        // when there are previous updates (you can't back up what doesn't exist).
463        // However, previous updates CAN exist without a backup_path: during migration,
464        // synthetic ops log entries are generated with no prior repo info file to
465        // reference. (This differs from RepoInitializedUpdate, which has neither
466        // previous updates nor a backup path.)
467        assert!(
468            backup_path.is_none() || last_update.is_some(),
469            "A backup path must not be provided without previous updates"
470        );
471
472        // Reject updates whose timestamp is not strictly newer than the top of the ops log
473        if let Some(Ok((_, latest_time, _))) = &last_update
474            && update.update_time <= *latest_time
475        {
476            return Err(IcechunkFormatErrorKind::InvalidUpdateTimestamp {
477                latest_time: *latest_time,
478                new_time: update.update_time,
479            })
480            .capture();
481        }
482
483        let new_updates: Box<dyn Iterator<Item = _>> =
484            if let Some(last_update) = last_update {
485                Box::new(
486                    [Ok((update.update_type, update.update_time, None)), last_update]
487                        .into_iter(),
488                )
489            } else {
490                Box::new([Ok((update.update_type, update.update_time, None))].into_iter())
491            };
492
493        let all_updates = new_updates.into_iter().chain(previous_updates);
494        // If we didn't overflow (all previous updates fit in the new file),
495        // preserve the old file's chain pointer so older history remains reachable.
496        let mut repo_before_updates = previous_info;
497
498        let num_updates = num_updates_per_file as usize;
499        let mut updates = Vec::new();
500        for maybe_data in all_updates {
501            let (u_type, u_time, file) = maybe_data?;
502
503            // Once we've reached the target file size, look for a valid overflow
504            // point: an entry whose backup_path we can use as the chain pointer.
505            // Entries without a backup_path (e.g. synthetic migration entries)
506            // can't serve as overflow — keep them in the file instead.
507            if updates.len() >= num_updates
508                && let Some(bp) = file
509            {
510                repo_before_updates = Some(bp);
511                break;
512            }
513
514            let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
515            let file = file.map(|file| builder.create_string(file));
516            updates.push(generated::Update::create(
517                builder,
518                &generated::UpdateArgs {
519                    update_type_type,
520                    update_type: Some(update_type),
521                    updated_at: u_time.timestamp_micros() as u64,
522                    backup_path: file,
523                },
524            ));
525        }
526
527        debug_assert!(!updates.is_empty(), "Must have at least one update in repo file");
528
529        let updates = builder.create_vector(&updates);
530        let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
531        Ok((updates, repo_before_updates))
532    }
533
534    fn mk_latest_updates_passthrough<'bldr, 'a>(
535        builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
536        updates: impl IntoIterator<Item = UpdateTuple<'a>>,
537        repo_before_updates: Option<&'a str>,
538    ) -> LatestUpdatesResult<'bldr> {
539        let mut fb_updates = Vec::new();
540        for maybe_data in updates {
541            let (u_type, u_time, file) = maybe_data?;
542            let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
543            let file = file.map(|file| builder.create_string(file));
544            fb_updates.push(generated::Update::create(
545                builder,
546                &generated::UpdateArgs {
547                    update_type_type,
548                    update_type: Some(update_type),
549                    updated_at: u_time.timestamp_micros() as u64,
550                    backup_path: file,
551                },
552            ));
553        }
554        debug_assert!(
555            !fb_updates.is_empty(),
556            "Must have at least one update in repo file"
557        );
558        let updates = builder.create_vector(&fb_updates);
559        let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
560        Ok((updates, repo_before_updates))
561    }
562
563    pub fn initial<C: Serialize>(
564        spec_version: SpecVersionBin,
565        snapshot: SnapshotInfo,
566        num_updates_per_file: u16,
567        config: Option<&C>,
568        update_time: Option<DateTime<Utc>>,
569    ) -> Self {
570        #[expect(clippy::expect_used)]
571        let config_bytes =
572            config.map(|c| flexbuffers::to_vec(c).expect("Cannot serialize config"));
573        // This method is basically constant, so it's OK to unwrap in it
574        #[expect(clippy::expect_used)]
575        Self::from_parts(
576            spec_version,
577            [],
578            [("main", 0)],
579            [],
580            [snapshot],
581            &Default::default(),
582            |builder| {
583                Self::mk_latest_updates(
584                    builder,
585                    UpdateInfo {
586                        update_type: UpdateType::RepoInitializedUpdate,
587                        update_time: update_time.unwrap_or_else(Utc::now),
588                        previous_updates: [],
589                    },
590                    None,
591                    num_updates_per_file,
592                    None,
593                )
594            },
595            config_bytes.as_deref(),
596            None::<std::iter::Empty<u16>>,
597            None::<std::iter::Empty<u16>>,
598            &RepoStatus {
599                availability: RepoAvailability::Online,
600                set_at: update_time.unwrap_or_else(Utc::now),
601                limited_availability_reason: None,
602            },
603        )
604        .expect("Cannot generate initial snapshot")
605    }
606
607    /// Read the raw config bytes from the `FlatBuffer` (for pass-through in mutations).
608    pub fn config_bytes_raw(&self) -> IcechunkResult<Option<Vec<u8>>> {
609        Ok(self.root()?.config().map(|v| v.bytes().to_vec()))
610    }
611
612    /// Read the repository configuration from the repo info.
613    /// Returns `None` for repos created before config was embedded,
614    /// or for repos using the default configuration.
615    pub fn config<C: serde::de::DeserializeOwned>(&self) -> IcechunkResult<Option<C>> {
616        match self.root()?.config() {
617            None => Ok(None),
618            Some(config_fb) => {
619                let config: C = flexbuffers::from_slice(config_fb.bytes())
620                    .map_err(Box::new)
621                    .capture()?;
622                Ok(Some(config))
623            }
624        }
625    }
626
627    pub fn metadata(&self) -> IcechunkResult<SnapshotProperties> {
628        self.root()?
629            .metadata()
630            .unwrap_or_default()
631            .iter()
632            .map(|item| {
633                let key = item.name().to_string();
634                let value = flexbuffers::from_slice(item.value().bytes())
635                    .map_err(Box::new)
636                    .capture()?;
637                Ok((key, value))
638            })
639            .try_collect()
640    }
641
642    pub fn status(&self) -> IcechunkResult<RepoStatus> {
643        let root = self.root()?;
644        let fb_status = root.status();
645        fb_status.try_into()
646    }
647
648    pub fn enabled_feature_flags(
649        &self,
650    ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
651    {
652        Ok(self.root()?.enabled_feature_flags().map(|v| v.iter()))
653    }
654
655    pub fn disabled_feature_flags(
656        &self,
657    ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
658    {
659        Ok(self.root()?.disabled_feature_flags().map(|v| v.iter()))
660    }
661
662    /// None means not set, use the default value
663    /// Some(true) means enabled
664    /// Some(false) means disabled
665    pub fn feature_flag_enabled(&self, id: u16) -> IcechunkResult<Option<bool>> {
666        let root = self.root()?;
667        if root
668            .enabled_feature_flags()
669            .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
670            .is_some()
671        {
672            return Ok(Some(true));
673        }
674        if root
675            .disabled_feature_flags()
676            .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
677            .is_some()
678        {
679            return Ok(Some(false));
680        }
681        Ok(None)
682    }
683
684    fn all_tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
685        Ok(self.root()?.tags().iter().map(|r| (r.name(), r.snapshot_index())))
686    }
687
688    fn all_branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
689        Ok(self.root()?.branches().iter().map(|r| (r.name(), r.snapshot_index())))
690    }
691
692    pub fn deleted_tags(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
693        Ok(self.root()?.deleted_tags().iter())
694    }
695
696    pub fn all_snapshots(
697        &self,
698    ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>>> {
699        let root = self.root()?;
700        Ok(root.snapshots().iter().map(move |snap| mk_snapshot_info(&root, &snap)))
701    }
702
703    /// Doesn't check the validity of `flag_id`
704    pub fn update_feature_flag(
705        &self,
706        spec_version: SpecVersionBin,
707        flag_id: u16,
708        enabled: Option<bool>,
709        previous_file: &str,
710        num_updates_per_file: u16,
711    ) -> IcechunkResult<Self> {
712        let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
713        let (eff, dff): (Option<Vec<_>>, Option<Vec<_>>) = match enabled {
714            Some(false) => {
715                let e = self
716                    .enabled_feature_flags()?
717                    .map(|it| it.filter(|x| *x != flag_id).collect());
718
719                let mut d: BTreeSet<_> = self
720                    .disabled_feature_flags()?
721                    .map(|it| it.collect())
722                    .unwrap_or_default();
723                d.insert(flag_id);
724                let d = d.into_iter().collect();
725                (e, Some(d))
726            }
727            Some(true) => {
728                let d = self
729                    .disabled_feature_flags()?
730                    .map(|it| it.filter(|x| *x != flag_id).collect());
731
732                let mut e: BTreeSet<_> = self
733                    .enabled_feature_flags()?
734                    .map(|it| it.collect())
735                    .unwrap_or_default();
736                e.insert(flag_id);
737                let e = e.into_iter().collect();
738                (Some(e), d)
739            }
740            None => {
741                let e = self
742                    .enabled_feature_flags()?
743                    .map(|it| it.filter(|x| *x != flag_id).collect());
744                let d = self
745                    .disabled_feature_flags()?
746                    .map(|it| it.filter(|x| *x != flag_id).collect());
747                (e, d)
748            }
749        };
750
751        let latest_updates = self.latest_updates()?;
752        let rbu = self.repo_before_updates()?;
753        Self::from_parts(
754            spec_version,
755            self.all_tags()?,
756            self.all_branches()?,
757            self.deleted_tags()?,
758            snaps,
759            &self.metadata()?,
760            |builder| {
761                Self::mk_latest_updates(
762                    builder,
763                    UpdateInfo {
764                        update_type: UpdateType::FeatureFlagChanged {
765                            id: flag_id,
766                            new_value: enabled,
767                        },
768                        update_time: Utc::now(),
769                        previous_updates: latest_updates,
770                    },
771                    Some(previous_file),
772                    num_updates_per_file,
773                    rbu,
774                )
775            },
776            self.config_bytes_raw()?.as_deref(),
777            eff.map(|it| it.into_iter()),
778            dff.map(|it| it.into_iter()),
779            &self.status()?,
780        )
781    }
782
783    #[expect(clippy::too_many_arguments)]
784    pub fn add_snapshot(
785        &self,
786        spec_version: SpecVersionBin,
787        snap: SnapshotInfo,
788        branch: Option<&str>,
789        update_type: UpdateType,
790        update_time: Option<DateTime<Utc>>, // for testing
791        previous_file: &str,
792        num_updates_per_file: u16,
793    ) -> IcechunkResult<Self> {
794        let mut snapshots: Vec<_> = self.all_snapshots()?.try_collect()?;
795        let new_index = match snapshots.binary_search_by_key(&&snap.id, |snap| &snap.id) {
796            Ok(_) => Err(IcechunkFormatErrorKind::DuplicateSnapshotId {
797                snapshot_id: snap.id.clone(),
798            })
799            .capture(),
800            Err(idx) => Ok(idx),
801        }?;
802
803        snapshots.insert(new_index, snap);
804
805        let tags = self.all_tags()?.map(|(name, idx)| {
806            if idx as usize >= new_index { (name, idx + 1) } else { (name, idx) }
807        });
808        let branches = self.all_branches()?.map(|(name, idx)| {
809            if Some(name) == branch {
810                (name, new_index as u32)
811            } else if idx as usize >= new_index {
812                (name, idx + 1)
813            } else {
814                (name, idx)
815            }
816        });
817
818        let latest_updates = self.latest_updates()?;
819        let rbu = self.repo_before_updates()?;
820        let res = Self::from_parts(
821            spec_version,
822            tags,
823            branches,
824            self.deleted_tags()?,
825            snapshots,
826            &self.metadata()?,
827            |builder| {
828                Self::mk_latest_updates(
829                    builder,
830                    UpdateInfo {
831                        update_type,
832                        update_time: update_time.unwrap_or_else(Utc::now),
833                        previous_updates: latest_updates,
834                    },
835                    Some(previous_file),
836                    num_updates_per_file,
837                    rbu,
838                )
839            },
840            self.config_bytes_raw()?.as_deref(),
841            self.enabled_feature_flags()?,
842            self.disabled_feature_flags()?,
843            &self.status()?,
844        )?;
845        Ok(res)
846    }
847
848    pub fn add_branch(
849        &self,
850        spec_version: SpecVersionBin,
851        name: &str,
852        snap: &SnapshotId,
853        previous_file: &str,
854        num_updates_per_file: u16,
855    ) -> IcechunkResult<Self> {
856        if let Ok(snapshot_id) = self.resolve_branch(name) {
857            return Err(IcechunkFormatErrorKind::BranchAlreadyExists {
858                branch: name.to_string(),
859                snapshot_id,
860            })
861            .capture();
862        }
863
864        match self.resolve_snapshot_index(snap)? {
865            Some(snap_idx) => {
866                let mut branches: Vec<_> = self.all_branches()?.collect();
867                branches.push((name, snap_idx as u32));
868                branches.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
869                let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
870                let latest_updates = self.latest_updates()?;
871                let rbu = self.repo_before_updates()?;
872                Ok(Self::from_parts(
873                    spec_version,
874                    self.all_tags()?,
875                    branches,
876                    self.deleted_tags()?,
877                    snaps,
878                    &self.metadata()?,
879                    |builder| {
880                        Self::mk_latest_updates(
881                            builder,
882                            UpdateInfo {
883                                update_type: UpdateType::BranchCreatedUpdate {
884                                    name: name.to_string(),
885                                },
886                                update_time: Utc::now(),
887                                previous_updates: latest_updates,
888                            },
889                            Some(previous_file),
890                            num_updates_per_file,
891                            rbu,
892                        )
893                    },
894                    self.config_bytes_raw()?.as_deref(),
895                    self.enabled_feature_flags()?,
896                    self.disabled_feature_flags()?,
897                    &self.status()?,
898                )?)
899            }
900            None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
901                snapshot_id: snap.clone(),
902            })
903            .capture(),
904        }
905    }
906
907    pub fn delete_branch(
908        &self,
909        spec_version: SpecVersionBin,
910        name: &str,
911        previous_file: &str,
912        num_updates_per_file: u16,
913    ) -> IcechunkResult<Self> {
914        match self.resolve_branch(name) {
915            Ok(previous_snap_id) => {
916                let mut branches: Vec<_> = self.all_branches()?.collect();
917                // retain preserves order
918                branches.retain(|(n, _)| n != &name);
919                let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
920                let latest_updates = self.latest_updates()?;
921                let rbu = self.repo_before_updates()?;
922                Self::from_parts(
923                    spec_version,
924                    self.all_tags()?,
925                    branches,
926                    self.deleted_tags()?,
927                    snaps,
928                    &self.metadata()?,
929                    |builder| {
930                        Self::mk_latest_updates(
931                            builder,
932                            UpdateInfo {
933                                update_type: UpdateType::BranchDeletedUpdate {
934                                    name: name.to_string(),
935                                    previous_snap_id,
936                                },
937                                update_time: Utc::now(),
938                                previous_updates: latest_updates,
939                            },
940                            Some(previous_file),
941                            num_updates_per_file,
942                            rbu,
943                        )
944                    },
945                    self.config_bytes_raw()?.as_deref(),
946                    self.enabled_feature_flags()?,
947                    self.disabled_feature_flags()?,
948                    &self.status()?,
949                )
950            }
951            Err(IcechunkFormatError {
952                kind: IcechunkFormatErrorKind::BranchNotFound { .. },
953                ..
954            }) => {
955                Err(IcechunkFormatErrorKind::BranchNotFound { branch: name.to_string() })
956                    .capture()
957            }
958            Err(err) => Err(err),
959        }
960    }
961
962    pub fn update_branch(
963        &self,
964        spec_version: SpecVersionBin,
965        name: &str,
966        new_snap: &SnapshotId,
967        previous_file: &str,
968        num_updates_per_file: u16,
969    ) -> IcechunkResult<Self> {
970        let previous_snap_id = self.resolve_branch(name)?;
971        match self.resolve_snapshot_index(new_snap)? {
972            Some(snap_idx) => {
973                let branches = self.all_branches()?.map(|(br, idx)| {
974                    if br == name { (br, snap_idx as u32) } else { (br, idx) }
975                });
976                let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
977                let latest_updates = self.latest_updates()?;
978                let rbu = self.repo_before_updates()?;
979                Ok(Self::from_parts(
980                    spec_version,
981                    self.all_tags()?,
982                    branches,
983                    self.deleted_tags()?,
984                    snaps,
985                    &self.metadata()?,
986                    |builder| {
987                        Self::mk_latest_updates(
988                            builder,
989                            UpdateInfo {
990                                update_type: UpdateType::BranchResetUpdate {
991                                    name: name.to_string(),
992                                    previous_snap_id,
993                                },
994                                update_time: Utc::now(),
995                                previous_updates: latest_updates,
996                            },
997                            Some(previous_file),
998                            num_updates_per_file,
999                            rbu,
1000                        )
1001                    },
1002                    self.config_bytes_raw()?.as_deref(),
1003                    self.enabled_feature_flags()?,
1004                    self.disabled_feature_flags()?,
1005                    &self.status()?,
1006                )?)
1007            }
1008            None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1009                snapshot_id: new_snap.clone(),
1010            })
1011            .capture(),
1012        }
1013    }
1014
1015    pub fn add_tag(
1016        &self,
1017        spec_version: SpecVersionBin,
1018        name: &str,
1019        snap: &SnapshotId,
1020        previous_file: &str,
1021        num_updates_per_file: u16,
1022    ) -> IcechunkResult<Self> {
1023        if self.resolve_tag(name).is_ok() {
1024            return Err(IcechunkFormatErrorKind::TagAlreadyExists {
1025                tag: name.to_string(),
1026            })
1027            .capture();
1028        }
1029        if self.tag_was_deleted(name)? {
1030            return Err(IcechunkFormatErrorKind::TagPreviouslyDeleted {
1031                tag: name.to_string(),
1032            })
1033            .capture();
1034        }
1035
1036        match self.resolve_snapshot_index(snap)? {
1037            Some(snap_idx) => {
1038                let mut tags: Vec<_> = self.all_tags()?.collect();
1039                tags.push((name, snap_idx as u32));
1040                tags.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
1041                let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1042                let latest_updates = self.latest_updates()?;
1043                let rbu = self.repo_before_updates()?;
1044                Ok(Self::from_parts(
1045                    spec_version,
1046                    tags,
1047                    self.all_branches()?,
1048                    self.deleted_tags()?,
1049                    snaps,
1050                    &self.metadata()?,
1051                    |builder| {
1052                        Self::mk_latest_updates(
1053                            builder,
1054                            UpdateInfo {
1055                                update_type: UpdateType::TagCreatedUpdate {
1056                                    name: name.to_string(),
1057                                },
1058                                update_time: Utc::now(),
1059                                previous_updates: latest_updates,
1060                            },
1061                            Some(previous_file),
1062                            num_updates_per_file,
1063                            rbu,
1064                        )
1065                    },
1066                    self.config_bytes_raw()?.as_deref(),
1067                    self.enabled_feature_flags()?,
1068                    self.disabled_feature_flags()?,
1069                    &self.status()?,
1070                )?)
1071            }
1072            None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1073                snapshot_id: snap.clone(),
1074            })
1075            .capture(),
1076        }
1077    }
1078
1079    pub fn delete_tag(
1080        &self,
1081        spec_version: SpecVersionBin,
1082        name: &str,
1083        previous_file: &str,
1084        num_updates_per_file: u16,
1085    ) -> IcechunkResult<Self> {
1086        match self.resolve_tag(name) {
1087            Ok(previous_snap_id) => {
1088                let mut tags: Vec<_> = self.all_tags()?.collect();
1089                // retain preserves order
1090                tags.retain(|(n, _)| n != &name);
1091
1092                let mut deleted_tags: BTreeSet<_> = self.deleted_tags()?.collect();
1093                debug_assert!(!deleted_tags.contains(name));
1094                deleted_tags.insert(name);
1095
1096                let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1097                let latest_updates = self.latest_updates()?;
1098                let rbu = self.repo_before_updates()?;
1099                Self::from_parts(
1100                    spec_version,
1101                    tags,
1102                    self.all_branches()?,
1103                    deleted_tags,
1104                    snaps,
1105                    &self.metadata()?,
1106                    |builder| {
1107                        Self::mk_latest_updates(
1108                            builder,
1109                            UpdateInfo {
1110                                update_type: UpdateType::TagDeletedUpdate {
1111                                    name: name.to_string(),
1112                                    previous_snap_id,
1113                                },
1114                                update_time: Utc::now(),
1115                                previous_updates: latest_updates,
1116                            },
1117                            Some(previous_file),
1118                            num_updates_per_file,
1119                            rbu,
1120                        )
1121                    },
1122                    self.config_bytes_raw()?.as_deref(),
1123                    self.enabled_feature_flags()?,
1124                    self.disabled_feature_flags()?,
1125                    &self.status()?,
1126                )
1127            }
1128            Err(IcechunkFormatError {
1129                kind: IcechunkFormatErrorKind::TagNotFound { .. },
1130                ..
1131            }) => Err(IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1132                .capture(),
1133            Err(err) => Err(err),
1134        }
1135    }
1136
1137    pub fn set_metadata(
1138        &self,
1139        spec_version: SpecVersionBin,
1140        metadata: &SnapshotProperties,
1141        previous_file: &str,
1142        num_updates_per_file: u16,
1143    ) -> IcechunkResult<Self> {
1144        let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1145        let latest_updates = self.latest_updates()?;
1146        let rbu = self.repo_before_updates()?;
1147        Self::from_parts(
1148            spec_version,
1149            self.all_tags()?,
1150            self.all_branches()?,
1151            self.deleted_tags()?,
1152            snaps,
1153            metadata,
1154            |builder| {
1155                Self::mk_latest_updates(
1156                    builder,
1157                    UpdateInfo {
1158                        update_type: UpdateType::MetadataChangedUpdate,
1159                        update_time: Utc::now(),
1160                        previous_updates: latest_updates,
1161                    },
1162                    Some(previous_file),
1163                    num_updates_per_file,
1164                    rbu,
1165                )
1166            },
1167            self.config_bytes_raw()?.as_deref(),
1168            self.enabled_feature_flags()?,
1169            self.disabled_feature_flags()?,
1170            &self.status()?,
1171        )
1172    }
1173
1174    /// Update the embedded configuration and record a `ConfigChangedUpdate` in the op log.
1175    pub fn set_config<C: Serialize>(
1176        &self,
1177        spec_version: SpecVersionBin,
1178        config: &C,
1179        previous_file: &str,
1180        num_updates_per_file: u16,
1181    ) -> IcechunkResult<Self> {
1182        let config_bytes = flexbuffers::to_vec(config).map_err(Box::new).capture()?;
1183        let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1184        let latest_updates = self.latest_updates()?;
1185        let rbu = self.repo_before_updates()?;
1186        Self::from_parts(
1187            spec_version,
1188            self.all_tags()?,
1189            self.all_branches()?,
1190            self.deleted_tags()?,
1191            snaps,
1192            &self.metadata()?,
1193            |builder| {
1194                Self::mk_latest_updates(
1195                    builder,
1196                    UpdateInfo {
1197                        update_type: UpdateType::ConfigChangedUpdate,
1198                        update_time: Utc::now(),
1199                        previous_updates: latest_updates,
1200                    },
1201                    Some(previous_file),
1202                    num_updates_per_file,
1203                    rbu,
1204                )
1205            },
1206            Some(config_bytes.as_slice()),
1207            self.enabled_feature_flags()?,
1208            self.disabled_feature_flags()?,
1209            &self.status()?,
1210        )
1211    }
1212
1213    /// Set the repository status.
1214    ///
1215    /// When `update_config` is `Some((previous_file, num_updates_per_file))`, a
1216    /// `RepoStatusChangedUpdate` entry is appended to the ops log (normal behavior).
1217    ///
1218    /// When `update_config` is `None`, the status is changed silently: the existing
1219    /// ops log entries are preserved as-is with no new entry added.
1220    pub fn set_status(
1221        &self,
1222        spec_version: SpecVersionBin,
1223        status: &RepoStatus,
1224        update_config: Option<(&str, u16)>,
1225    ) -> IcechunkResult<Self> {
1226        let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1227        let latest_updates = self.latest_updates()?;
1228        let rbu = self.repo_before_updates()?;
1229        if let Some((previous_file, num_updates_per_file)) = update_config {
1230            let status_update = status.clone();
1231            Self::from_parts(
1232                spec_version,
1233                self.all_tags()?,
1234                self.all_branches()?,
1235                self.deleted_tags()?,
1236                snaps,
1237                &self.metadata()?,
1238                |builder| {
1239                    Self::mk_latest_updates(
1240                        builder,
1241                        UpdateInfo {
1242                            update_type: UpdateType::RepoStatusChangedUpdate {
1243                                status: status_update,
1244                            },
1245                            update_time: Utc::now(),
1246                            previous_updates: latest_updates,
1247                        },
1248                        Some(previous_file),
1249                        num_updates_per_file,
1250                        rbu,
1251                    )
1252                },
1253                self.config_bytes_raw()?.as_deref(),
1254                self.enabled_feature_flags()?,
1255                self.disabled_feature_flags()?,
1256                status,
1257            )
1258        } else {
1259            Self::from_parts(
1260                spec_version,
1261                self.all_tags()?,
1262                self.all_branches()?,
1263                self.deleted_tags()?,
1264                snaps,
1265                &self.metadata()?,
1266                |builder| {
1267                    Self::mk_latest_updates_passthrough(builder, latest_updates, rbu)
1268                },
1269                self.config_bytes_raw()?.as_deref(),
1270                self.enabled_feature_flags()?,
1271                self.disabled_feature_flags()?,
1272                status,
1273            )
1274        }
1275    }
1276
1277    pub fn from_buffer(buffer: Vec<u8>) -> IcechunkResult<RepoInfo> {
1278        let _ = flatbuffers::root_with_opts::<generated::Repo<'_>>(
1279            &ROOT_OPTIONS,
1280            buffer.as_slice(),
1281        )
1282        .capture()?;
1283        Ok(RepoInfo { buffer })
1284    }
1285
1286    pub fn bytes(&self) -> &[u8] {
1287        self.buffer.as_slice()
1288    }
1289
1290    fn root(&self) -> IcechunkResult<generated::Repo<'_>> {
1291        flatbuffers::root::<generated::Repo<'_>>(&self.buffer).capture()
1292    }
1293
1294    pub fn tag_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1295        Ok(self.root()?.tags().iter().map(|r| r.name()))
1296    }
1297
1298    pub fn branch_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1299        Ok(self.root()?.branches().iter().map(|r| r.name()))
1300    }
1301
1302    pub fn tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1303        let root = self.root()?;
1304        Ok(self.all_tags()?.map(move |(name, idx)| {
1305            (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1306        }))
1307    }
1308
1309    pub fn branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1310        let root = self.root()?;
1311        Ok(self.all_branches()?.map(move |(name, idx)| {
1312            (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1313        }))
1314    }
1315
1316    pub fn resolve_tag(&self, name: &str) -> IcechunkResult<SnapshotId> {
1317        let root = self.root()?;
1318        let res = root
1319            .tags()
1320            .lookup_by_key(name, |r, key| r.name().cmp(key))
1321            .map(|r| {
1322                let index = r.snapshot_index();
1323                SnapshotId::new(root.snapshots().get(index as usize).id().0)
1324            })
1325            .ok_or_else(|| IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1326            .capture()?;
1327
1328        Ok(res)
1329    }
1330
1331    pub fn tag_was_deleted(&self, name: &str) -> IcechunkResult<bool> {
1332        let root = self.root()?;
1333        let res = root.deleted_tags().lookup_by_key(name, |name, key| name.cmp(key));
1334        Ok(res.is_some())
1335    }
1336
1337    pub fn resolve_branch(&self, name: &str) -> IcechunkResult<SnapshotId> {
1338        let root = self.root()?;
1339        let res = root
1340            .branches()
1341            .lookup_by_key(name, |r, key| r.name().cmp(key))
1342            .map(|r| {
1343                let index = r.snapshot_index();
1344                SnapshotId::new(root.snapshots().get(index as usize).id().0)
1345            })
1346            .ok_or_else(|| IcechunkFormatErrorKind::BranchNotFound {
1347                branch: name.to_string(),
1348            })
1349            .capture()?;
1350
1351        Ok(res)
1352    }
1353
1354    pub fn spec_version(&self) -> IcechunkResult<SpecVersionBin> {
1355        let raw = self.root()?.spec_version();
1356        raw.try_into()
1357            .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1358                found: raw,
1359                max_supported: SpecVersionBin::current() as u8,
1360            })
1361            .capture()
1362    }
1363
1364    pub fn latest_updates(
1365        &self,
1366    ) -> IcechunkResult<impl Iterator<Item = UpdateTuple<'_>>> {
1367        let res = self.root()?.latest_updates().iter().map(|up| self.update_to_tuple(up));
1368        Ok(res)
1369    }
1370
1371    fn update_to_tuple<'a>(
1372        &'a self,
1373        update: generated::Update<'a>,
1374    ) -> IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)> {
1375        let ty = self.mk_update_type(&update)?;
1376        let ts = timestamp_to_timestamp(update.updated_at())?;
1377        let bp = update.backup_path();
1378        Ok((ty, ts, bp))
1379    }
1380
1381    fn mk_update_type(
1382        &self,
1383        update: &generated::Update<'_>,
1384    ) -> IcechunkResult<UpdateType> {
1385        #[expect(clippy::unwrap_used)]
1386        match update.update_type_type() {
1387            generated::UpdateType::RepoInitializedUpdate => {
1388                Ok(UpdateType::RepoInitializedUpdate)
1389            }
1390            generated::UpdateType::RepoMigratedUpdate => {
1391                let up = update.update_type_as_repo_migrated_update().unwrap();
1392                let from_raw = up.from_version();
1393                let to_raw = up.to_version();
1394                Ok(UpdateType::RepoMigratedUpdate {
1395                    from_version: from_raw
1396                        .try_into()
1397                        .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1398                            found: from_raw,
1399                            max_supported: SpecVersionBin::current() as u8,
1400                        })
1401                        .capture()?,
1402                    to_version: to_raw
1403                        .try_into()
1404                        .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1405                            found: to_raw,
1406                            max_supported: SpecVersionBin::current() as u8,
1407                        })
1408                        .capture()?,
1409                })
1410            }
1411            generated::UpdateType::RepoStatusChangedUpdate => {
1412                let up = update.update_type_as_repo_status_changed_update().unwrap();
1413                let fb_status = up.status().unwrap();
1414                let status = fb_status.try_into()?;
1415
1416                Ok(UpdateType::RepoStatusChangedUpdate { status })
1417            }
1418            generated::UpdateType::ConfigChangedUpdate => {
1419                Ok(UpdateType::ConfigChangedUpdate)
1420            }
1421            generated::UpdateType::MetadataChangedUpdate => {
1422                Ok(UpdateType::MetadataChangedUpdate)
1423            }
1424            generated::UpdateType::TagCreatedUpdate => {
1425                let up = update.update_type_as_tag_created_update().unwrap();
1426                Ok(UpdateType::TagCreatedUpdate { name: up.name().to_string() })
1427            }
1428            generated::UpdateType::TagDeletedUpdate => {
1429                let up = update.update_type_as_tag_deleted_update().unwrap();
1430                let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1431                Ok(UpdateType::TagDeletedUpdate {
1432                    name: up.name().to_string(),
1433                    previous_snap_id,
1434                })
1435            }
1436            generated::UpdateType::BranchCreatedUpdate => {
1437                let up = update.update_type_as_branch_created_update().unwrap();
1438                Ok(UpdateType::BranchCreatedUpdate { name: up.name().to_string() })
1439            }
1440            generated::UpdateType::BranchDeletedUpdate => {
1441                let up = update.update_type_as_branch_deleted_update().unwrap();
1442                let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1443                Ok(UpdateType::BranchDeletedUpdate {
1444                    name: up.name().to_string(),
1445                    previous_snap_id,
1446                })
1447            }
1448            generated::UpdateType::BranchResetUpdate => {
1449                let up = update.update_type_as_branch_reset_update().unwrap();
1450                let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1451                Ok(UpdateType::BranchResetUpdate {
1452                    name: up.name().to_string(),
1453                    previous_snap_id,
1454                })
1455            }
1456            generated::UpdateType::NewCommitUpdate => {
1457                let up = update.update_type_as_new_commit_update().unwrap();
1458                let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1459                Ok(UpdateType::NewCommitUpdate {
1460                    branch: up.branch().to_string(),
1461                    new_snap_id,
1462                })
1463            }
1464            generated::UpdateType::CommitAmendedUpdate => {
1465                let up = update.update_type_as_commit_amended_update().unwrap();
1466                let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1467                let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1468                Ok(UpdateType::CommitAmendedUpdate {
1469                    branch: up.branch().to_string(),
1470                    previous_snap_id,
1471                    new_snap_id,
1472                })
1473            }
1474            generated::UpdateType::NewDetachedSnapshotUpdate => {
1475                let up = update.update_type_as_new_detached_snapshot_update().unwrap();
1476                let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1477                Ok(UpdateType::NewDetachedSnapshotUpdate { new_snap_id })
1478            }
1479            generated::UpdateType::GCRanUpdate => Ok(UpdateType::GCRanUpdate),
1480            generated::UpdateType::ExpirationRanUpdate => {
1481                Ok(UpdateType::ExpirationRanUpdate)
1482            }
1483            generated::UpdateType::FeatureFlagChangedUpdate => {
1484                let up = update.update_type_as_feature_flag_changed_update().unwrap();
1485                Ok(UpdateType::FeatureFlagChanged {
1486                    id: up.id(),
1487                    new_value: if up.is_set() { Some(up.new_value()) } else { None },
1488                })
1489            }
1490            _ => Err(IcechunkFormatErrorKind::InvalidFlatBuffer(
1491                flatbuffers::InvalidFlatbuffer::InconsistentUnion {
1492                    field: Cow::Borrowed("latest_update_type"),
1493                    field_type: Cow::Borrowed("UpdateType"),
1494                    error_trace: Default::default(),
1495                },
1496            ))
1497            .capture(),
1498        }
1499    }
1500
1501    pub fn ancestry<'a>(
1502        &'a self,
1503        snapshot: &SnapshotId,
1504    ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>> + Send + use<'a>>
1505    {
1506        let root = self.root()?;
1507        if let Some(start) = self.resolve_snapshot_index(snapshot)? {
1508            let mut index = Some(start as i32);
1509            let iter = std::iter::from_fn(move || {
1510                if let Some(ix) = index {
1511                    if ix >= 0 {
1512                        let snap = root.snapshots().get(ix as usize);
1513                        index = Some(snap.parent_offset());
1514                        Some(mk_snapshot_info(&root, &snap))
1515                    } else {
1516                        index = None;
1517                        None
1518                    }
1519                } else {
1520                    None
1521                }
1522            });
1523            Ok(iter)
1524        } else {
1525            Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1526                snapshot_id: snapshot.clone(),
1527            })
1528            .capture()
1529        }
1530    }
1531
1532    pub fn find_snapshot(&self, id: &SnapshotId) -> IcechunkResult<SnapshotInfo> {
1533        let mut anc = self.ancestry(id)?;
1534        #[expect(clippy::panic)]
1535        match anc.next() {
1536            Some(snap) => snap,
1537            // It's OK to panic here because ancestry already found the snapshot, and
1538            // it's always the first element of the ancestry
1539            None => panic!("Ancestry head snapshot not found"),
1540        }
1541    }
1542
1543    pub fn repo_before_updates(&self) -> IcechunkResult<Option<&str>> {
1544        Ok(self.root()?.repo_before_updates())
1545    }
1546
1547    fn resolve_snapshot_index(&self, id: &SnapshotId) -> IcechunkResult<Option<usize>> {
1548        Ok(lookup_index_by_key(self.root()?.snapshots(), &id.0, |snap, key| {
1549            snap.id().0.cmp(key)
1550        }))
1551    }
1552}
1553
1554fn resolve_ref_iter<'a>(
1555    sorted_snapshots: &[SnapshotInfo],
1556    it: impl IntoIterator<Item = (&'a str, SnapshotId)>,
1557) -> IcechunkResult<Vec<(&'a str, u32)>> {
1558    let mut res: Vec<_> = it
1559        .into_iter()
1560        .map(|(name, id)| {
1561            let idx = sorted_snapshots
1562                .binary_search_by_key(&&id.0, |snap| &snap.id.0)
1563                .map_err(|_| IcechunkFormatErrorKind::SnapshotIdNotFound {
1564                    snapshot_id: id.clone(),
1565                })
1566                .capture()? as u32;
1567            Ok::<_, IcechunkFormatError>((name, idx))
1568        })
1569        .try_collect()?;
1570    res.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
1571    Ok(res)
1572}
1573
1574fn timestamp_to_timestamp(ts: u64) -> IcechunkResult<DateTime<Utc>> {
1575    let ts: i64 =
1576        ts.try_into().map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp).capture()?;
1577    DateTime::from_timestamp_micros(ts)
1578        .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
1579        .capture()
1580}
1581
1582fn mk_snapshot_info(
1583    repo: &generated::Repo<'_>,
1584    snap: &generated::SnapshotInfo<'_>,
1585) -> IcechunkResult<SnapshotInfo> {
1586    let flushed_at = timestamp_to_timestamp(snap.flushed_at())?;
1587    let parent_id = if snap.parent_offset() >= 0 {
1588        let parent = repo.snapshots().get(snap.parent_offset() as usize).id();
1589        Some(parent)
1590    } else {
1591        None
1592    };
1593    let metadata = snap
1594        .metadata()
1595        .map(|items| {
1596            let items = items
1597                .iter()
1598                .map(|item| {
1599                    let name = item.name().to_string();
1600                    let value = flexbuffers::from_slice(item.value().bytes())
1601                        .map_err(Box::new)
1602                        .capture()?;
1603                    Ok::<_, IcechunkFormatError>((name, value))
1604                })
1605                .try_collect()?;
1606            Ok::<_, IcechunkFormatError>(items)
1607        })
1608        .transpose()?
1609        .unwrap_or_default();
1610
1611    Ok(SnapshotInfo {
1612        id: SnapshotId::new(snap.id().0),
1613        flushed_at,
1614        message: snap.message().to_string(),
1615        metadata,
1616        parent_id: parent_id.map(|buf| SnapshotId::new(buf.0)),
1617    })
1618}
1619
1620fn update_type_to_fb<'bldr>(
1621    builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
1622    update: &UpdateType,
1623) -> IcechunkResult<(generated::UpdateType, WIPOffset<UnionWIPOffset>)> {
1624    match update {
1625        UpdateType::RepoInitializedUpdate => Ok((
1626            generated::UpdateType::RepoInitializedUpdate,
1627            generated::RepoInitializedUpdate::create(
1628                builder,
1629                &generated::RepoInitializedUpdateArgs {},
1630            )
1631            .as_union_value(),
1632        )),
1633        UpdateType::RepoMigratedUpdate { from_version, to_version } => Ok((
1634            generated::UpdateType::RepoMigratedUpdate,
1635            generated::RepoMigratedUpdate::create(
1636                builder,
1637                &generated::RepoMigratedUpdateArgs {
1638                    from_version: *from_version as u8,
1639                    to_version: *to_version as u8,
1640                },
1641            )
1642            .as_union_value(),
1643        )),
1644        UpdateType::RepoStatusChangedUpdate { status } => {
1645            let limited_availability_reason = status
1646                .limited_availability_reason
1647                .as_ref()
1648                .map(|r| builder.create_string(r));
1649            let status = generated::RepoStatus::create(
1650                builder,
1651                &generated::RepoStatusArgs {
1652                    availability: status.availability.into(),
1653                    set_at: status.set_at.timestamp_micros() as u64,
1654                    limited_availability_reason,
1655                },
1656            );
1657
1658            Ok((
1659                generated::UpdateType::RepoStatusChangedUpdate,
1660                generated::RepoStatusChangedUpdate::create(
1661                    builder,
1662                    &generated::RepoStatusChangedUpdateArgs { status: Some(status) },
1663                )
1664                .as_union_value(),
1665            ))
1666        }
1667        UpdateType::ConfigChangedUpdate => Ok((
1668            generated::UpdateType::ConfigChangedUpdate,
1669            generated::ConfigChangedUpdate::create(
1670                builder,
1671                &generated::ConfigChangedUpdateArgs {},
1672            )
1673            .as_union_value(),
1674        )),
1675        UpdateType::MetadataChangedUpdate => Ok((
1676            generated::UpdateType::MetadataChangedUpdate,
1677            generated::MetadataChangedUpdate::create(
1678                builder,
1679                &generated::MetadataChangedUpdateArgs {},
1680            )
1681            .as_union_value(),
1682        )),
1683        UpdateType::TagCreatedUpdate { name } => {
1684            let name = Some(builder.create_string(name));
1685            Ok((
1686                generated::UpdateType::TagCreatedUpdate,
1687                generated::TagCreatedUpdate::create(
1688                    builder,
1689                    &generated::TagCreatedUpdateArgs { name },
1690                )
1691                .as_union_value(),
1692            ))
1693        }
1694        UpdateType::TagDeletedUpdate { name, previous_snap_id } => {
1695            let name = Some(builder.create_string(name));
1696            let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1697            let previous_snap_id = Some(&object_id12);
1698            Ok((
1699                generated::UpdateType::TagDeletedUpdate,
1700                generated::TagDeletedUpdate::create(
1701                    builder,
1702                    &generated::TagDeletedUpdateArgs { name, previous_snap_id },
1703                )
1704                .as_union_value(),
1705            ))
1706        }
1707        UpdateType::BranchCreatedUpdate { name } => {
1708            let name = Some(builder.create_string(name));
1709            Ok((
1710                generated::UpdateType::BranchCreatedUpdate,
1711                generated::BranchCreatedUpdate::create(
1712                    builder,
1713                    &generated::BranchCreatedUpdateArgs { name },
1714                )
1715                .as_union_value(),
1716            ))
1717        }
1718        UpdateType::BranchDeletedUpdate { name, previous_snap_id } => {
1719            let name = Some(builder.create_string(name));
1720            let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1721            let previous_snap_id = Some(&object_id12);
1722            Ok((
1723                generated::UpdateType::BranchDeletedUpdate,
1724                generated::BranchDeletedUpdate::create(
1725                    builder,
1726                    &generated::BranchDeletedUpdateArgs { name, previous_snap_id },
1727                )
1728                .as_union_value(),
1729            ))
1730        }
1731        UpdateType::BranchResetUpdate { name, previous_snap_id } => {
1732            let name = Some(builder.create_string(name));
1733            let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1734            let previous_snap_id = Some(&object_id12);
1735            Ok((
1736                generated::UpdateType::BranchResetUpdate,
1737                generated::BranchResetUpdate::create(
1738                    builder,
1739                    &generated::BranchResetUpdateArgs { name, previous_snap_id },
1740                )
1741                .as_union_value(),
1742            ))
1743        }
1744        UpdateType::NewCommitUpdate { branch, new_snap_id } => {
1745            let branch = Some(builder.create_string(branch));
1746            let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1747            let new_snap_id = Some(&object_id12);
1748            Ok((
1749                generated::UpdateType::NewCommitUpdate,
1750                generated::NewCommitUpdate::create(
1751                    builder,
1752                    &generated::NewCommitUpdateArgs { branch, new_snap_id },
1753                )
1754                .as_union_value(),
1755            ))
1756        }
1757        UpdateType::CommitAmendedUpdate { branch, previous_snap_id, new_snap_id } => {
1758            let branch = Some(builder.create_string(branch));
1759            let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1760            let previous_snap_id = Some(&object_id12);
1761            let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1762            let new_snap_id = Some(&object_id12);
1763            Ok((
1764                generated::UpdateType::CommitAmendedUpdate,
1765                generated::CommitAmendedUpdate::create(
1766                    builder,
1767                    &generated::CommitAmendedUpdateArgs {
1768                        branch,
1769                        previous_snap_id,
1770                        new_snap_id,
1771                    },
1772                )
1773                .as_union_value(),
1774            ))
1775        }
1776        UpdateType::NewDetachedSnapshotUpdate { new_snap_id } => {
1777            let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1778            let new_snap_id = Some(&object_id12);
1779            Ok((
1780                generated::UpdateType::NewDetachedSnapshotUpdate,
1781                generated::NewDetachedSnapshotUpdate::create(
1782                    builder,
1783                    &generated::NewDetachedSnapshotUpdateArgs { new_snap_id },
1784                )
1785                .as_union_value(),
1786            ))
1787        }
1788        UpdateType::GCRanUpdate => Ok((
1789            generated::UpdateType::GCRanUpdate,
1790            generated::GCRanUpdate::create(builder, &generated::GCRanUpdateArgs {})
1791                .as_union_value(),
1792        )),
1793        UpdateType::ExpirationRanUpdate => Ok((
1794            generated::UpdateType::ExpirationRanUpdate,
1795            generated::ExpirationRanUpdate::create(
1796                builder,
1797                &generated::ExpirationRanUpdateArgs {},
1798            )
1799            .as_union_value(),
1800        )),
1801        UpdateType::FeatureFlagChanged { id, new_value } => Ok((
1802            generated::UpdateType::FeatureFlagChangedUpdate,
1803            generated::FeatureFlagChangedUpdate::create(
1804                builder,
1805                &generated::FeatureFlagChangedUpdateArgs {
1806                    id: *id,
1807                    new_value: new_value.unwrap_or_default(),
1808                    is_set: new_value.is_some(),
1809                },
1810            )
1811            .as_union_value(),
1812        )),
1813    }
1814}
1815
1816#[cfg(test)]
1817mod tests {
1818
1819    use super::*;
1820    use crate::roundtrip_serialization_tests;
1821    use proptest::prelude::*;
1822    use std::collections::HashSet;
1823
1824    // Generates an instance of RepoInfo which may not deserialize to a valid repository
1825    fn potentially_invalid_repo_info() -> impl Strategy<Value = RepoInfo> {
1826        any::<Vec<u8>>().prop_map(|buffer| RepoInfo { buffer })
1827    }
1828
1829    roundtrip_serialization_tests!(
1830        serialize_and_deserialize_repo_info - potentially_invalid_repo_info
1831    );
1832
1833    #[test]
1834    fn test_add_snapshot() -> Result<(), Box<dyn std::error::Error>> {
1835        let id1 = SnapshotId::random();
1836        let snap1 = SnapshotInfo {
1837            id: id1.clone(),
1838            parent_id: None,
1839            // needs to be micro second rounded
1840            flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1841            message: "snap 1".to_string(),
1842            metadata: Default::default(),
1843        };
1844        let repo = RepoInfo::initial(
1845            SpecVersionBin::current(),
1846            snap1.clone(),
1847            100,
1848            None::<&()>,
1849            None,
1850        );
1851        assert_eq!(repo.all_snapshots()?.next().unwrap().unwrap(), snap1);
1852
1853        let id2 = SnapshotId::random();
1854        let snap2 = SnapshotInfo {
1855            id: id2.clone(),
1856            parent_id: Some(id1.clone()),
1857            flushed_at: DateTime::from_timestamp_micros(2_000_000).unwrap(),
1858            message: "snap 2".to_string(),
1859            ..snap1.clone()
1860        };
1861        let repo = repo.add_snapshot(
1862            SpecVersionBin::current(),
1863            snap2.clone(),
1864            Some("main"),
1865            UpdateType::NewCommitUpdate {
1866                branch: "main".to_string(),
1867                new_snap_id: snap2.id.clone(),
1868            },
1869            None,
1870            "foo/bar",
1871            100,
1872        )?;
1873        assert_eq!(&repo.resolve_branch("main")?, &snap2.id);
1874        assert_eq!(repo.repo_before_updates()?, None);
1875
1876        let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1877        assert_eq!(all, HashSet::from_iter([snap1.clone(), snap2.clone()]));
1878
1879        let anc: Vec<_> = repo.ancestry(&id1)?.try_collect()?;
1880        assert_eq!(anc, std::slice::from_ref(&snap1));
1881
1882        let anc: Vec<_> = repo.ancestry(&id2)?.try_collect()?;
1883        assert_eq!(anc, [snap2.clone(), snap1.clone()]);
1884
1885        assert!(repo.ancestry(&SnapshotId::random()).is_err());
1886
1887        let id3 = SnapshotId::random();
1888        let snap3 = SnapshotInfo {
1889            id: id3.clone(),
1890            parent_id: Some(id2.clone()),
1891            flushed_at: DateTime::from_timestamp_micros(3_000_000).unwrap(),
1892            message: "snap 3".to_string(),
1893            ..snap2.clone()
1894        };
1895        let repo = repo.add_snapshot(
1896            SpecVersionBin::current(),
1897            snap3.clone(),
1898            Some("main"),
1899            UpdateType::NewCommitUpdate {
1900                branch: "main".to_string(),
1901                new_snap_id: snap3.id.clone(),
1902            },
1903            None,
1904            "foo",
1905            100,
1906        )?;
1907        assert_eq!(&repo.resolve_branch("main")?, &snap3.id);
1908        let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1909        assert_eq!(
1910            all,
1911            HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1912        );
1913
1914        let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1915        assert_eq!(
1916            all,
1917            HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1918        );
1919
1920        let anc: Vec<_> = repo.ancestry(&id3)?.try_collect()?;
1921        assert_eq!(anc, [snap3.clone(), snap2.clone(), snap1.clone()]);
1922        Ok(())
1923    }
1924
1925    #[test]
1926    fn test_tags_and_branches() -> Result<(), Box<dyn std::error::Error>> {
1927        let id1 = SnapshotId::random();
1928        let snap1 = SnapshotInfo {
1929            id: id1.clone(),
1930            parent_id: None,
1931            // needs to be micro second rounded
1932            flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1933            message: "snap 1".to_string(),
1934            metadata: Default::default(),
1935        };
1936        let repo = RepoInfo::initial(
1937            SpecVersionBin::current(),
1938            snap1.clone(),
1939            100,
1940            None::<&()>,
1941            None,
1942        );
1943        let repo = repo.add_branch(SpecVersionBin::current(), "foo", &id1, "foo", 100)?;
1944        let repo = repo.add_branch(SpecVersionBin::current(), "bar", &id1, "bar", 100)?;
1945        assert!(matches!(
1946            repo.add_branch(
1947                SpecVersionBin::current(),
1948                "bad-snap",
1949                &SnapshotId::random(),
1950                "bad",
1951                100
1952            ),
1953            Err(IcechunkFormatError {
1954                kind: IcechunkFormatErrorKind::SnapshotIdNotFound { .. },
1955                ..
1956            })
1957        ));
1958        // cannot add existing
1959        assert!(matches!(
1960            repo.add_branch(SpecVersionBin::current(), "bar", &id1, "/foo/bar", 100),
1961            Err(IcechunkFormatError {
1962                kind: IcechunkFormatErrorKind::BranchAlreadyExists { .. },
1963                ..
1964            })
1965        ));
1966
1967        assert_eq!(
1968            repo.all_branches()?.collect::<HashSet<_>>(),
1969            [("main", 0), ("foo", 0), ("bar", 0)].into()
1970        );
1971
1972        let id2 = SnapshotId::random();
1973        let snap2 = SnapshotInfo {
1974            id: id2.clone(),
1975            parent_id: Some(id1.clone()),
1976            flushed_at: Utc::now(),
1977            message: "snap 2".to_string(),
1978            ..snap1.clone()
1979        };
1980        let repo = repo.add_snapshot(
1981            SpecVersionBin::current(),
1982            snap2.clone(),
1983            Some("main"),
1984            UpdateType::NewCommitUpdate {
1985                branch: "main".to_string(),
1986                new_snap_id: snap2.id.clone(),
1987            },
1988            None,
1989            "foo",
1990            100,
1991        )?;
1992        let repo =
1993            repo.add_branch(SpecVersionBin::current(), "baz", &id2, "/foo/bar", 100)?;
1994        assert_eq!(repo.resolve_branch("main")?, id2.clone());
1995        assert_eq!(repo.resolve_branch("foo")?, id1.clone());
1996        assert_eq!(repo.resolve_branch("bar")?, id1.clone());
1997        assert_eq!(repo.resolve_branch("baz")?, id2.clone());
1998
1999        let repo = repo.delete_branch(SpecVersionBin::current(), "bar", "bar", 100)?;
2000        assert!(repo.resolve_branch("bar").is_err());
2001        assert_eq!(
2002            repo.all_branches()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2003            ["main", "foo", "baz"].into()
2004        );
2005
2006        assert!(
2007            repo.delete_branch(SpecVersionBin::current(), "bad-branch", "bad", 100)
2008                .is_err()
2009        );
2010
2011        // tags
2012        let repo = repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1", 100)?;
2013        let repo = repo.add_tag(SpecVersionBin::current(), "tag2", &id2, "tag2", 100)?;
2014        assert!(
2015            repo.add_tag(
2016                SpecVersionBin::current(),
2017                "bad-snap",
2018                &SnapshotId::random(),
2019                "bad",
2020                100
2021            )
2022            .is_err()
2023        );
2024        assert!(
2025            repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1-again", 100)
2026                .is_err()
2027        );
2028        assert_eq!(repo.resolve_tag("tag1")?, id1.clone());
2029        assert_eq!(repo.resolve_tag("tag2")?, id2.clone());
2030        assert_eq!(
2031            repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2032            ["tag1", "tag2"].into()
2033        );
2034
2035        // delete tags
2036        let repo = repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3", 100)?;
2037        let repo =
2038            repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3", 100)?;
2039        assert_eq!(
2040            repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2041            ["tag1", "tag2"].into()
2042        );
2043        // cannot add deleted
2044        assert!(
2045            repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3-again", 100)
2046                .is_err()
2047        );
2048        // cannot delete deleted
2049        assert!(
2050            repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3-again", 100)
2051                .is_err()
2052        );
2053        assert_eq!(
2054            repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2055            ["tag1", "tag2"].into()
2056        );
2057        Ok(())
2058    }
2059
2060    #[test]
2061    fn test_repo_info_updates() -> Result<(), Box<dyn std::error::Error>> {
2062        let id1 = SnapshotId::random();
2063        let snap1 = SnapshotInfo {
2064            id: id1.clone(),
2065            parent_id: None,
2066            // needs to be micro second rounded
2067            flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2068            message: "snap 1".to_string(),
2069            metadata: Default::default(),
2070        };
2071
2072        let num_updates_per_file: u16 = 10;
2073        let n = num_updates_per_file as usize;
2074
2075        // check updates for a new repo
2076        let mut repo = RepoInfo::initial(
2077            SpecVersionBin::current(),
2078            snap1,
2079            num_updates_per_file,
2080            None::<&()>,
2081            None,
2082        );
2083        assert_eq!(repo.latest_updates()?.count(), 1);
2084        let (last_update, _, file) = repo.latest_updates()?.next().unwrap()?;
2085        assert!(file.is_none());
2086        assert_eq!(last_update, UpdateType::RepoInitializedUpdate);
2087        assert!(repo.repo_before_updates()?.is_none());
2088        // check updates after num_updates_per_file changes
2089        // fill the first page of updates by adding branches
2090        for i in 1..=(n - 1) {
2091            repo = repo.add_branch(
2092                SpecVersionBin::current(),
2093                i.to_string().as_str(),
2094                &id1,
2095                (i - 1).to_string().as_str(),
2096                num_updates_per_file,
2097            )?;
2098        }
2099
2100        assert_eq!(repo.latest_updates()?.count(), n);
2101        let updates = repo.latest_updates()?;
2102
2103        // check all other updates
2104        for (idx, update) in updates.enumerate() {
2105            let (update, _, file) = update?;
2106            if idx == n - 1 {
2107                assert_eq!(update, UpdateType::RepoInitializedUpdate);
2108                assert_eq!(file, Some("0"));
2109            } else {
2110                assert_eq!(
2111                    update,
2112                    UpdateType::BranchCreatedUpdate { name: (n - 1 - idx).to_string() }
2113                );
2114                if idx == 0 {
2115                    assert!(file.is_none());
2116                } else {
2117                    assert_eq!(file, Some((n - 1 - idx).to_string().as_str()));
2118                }
2119            }
2120        }
2121        assert!(repo.repo_before_updates()?.is_none());
2122
2123        // Now, if we add another change, it won't fit in the first "page" of repo updates
2124        repo = repo.add_tag(
2125            SpecVersionBin::current(),
2126            "tag",
2127            &id1,
2128            "first-branches",
2129            num_updates_per_file,
2130        )?;
2131        // the file only contains the first "page" worth of updates
2132        assert_eq!(repo.latest_updates()?.count(), n);
2133        // next file is the oldest change
2134        assert_eq!(repo.repo_before_updates()?, Some("0"));
2135        let mut updates = repo.latest_updates()?;
2136        // last change is the tag creation
2137        let (last_update, _, file) = updates.next().unwrap()?;
2138        assert_eq!(last_update, UpdateType::TagCreatedUpdate { name: "tag".to_string() });
2139        assert!(file.is_none());
2140
2141        // next change is a branch creation backed up to first-branches
2142        let (last_update, _, file) = updates.next().unwrap()?;
2143        assert_eq!(
2144            last_update,
2145            UpdateType::BranchCreatedUpdate { name: (n - 1).to_string() }
2146        );
2147        assert_eq!(file, Some("first-branches"));
2148
2149        // all other changes are branch creation (repo creation is in the next page)
2150        for (idx, update) in updates.enumerate() {
2151            let (update, _, file) = update?;
2152            assert_eq!(file, Some((n - 2 - idx).to_string().as_str()));
2153            assert_eq!(
2154                update,
2155                UpdateType::BranchCreatedUpdate { name: (n - 2 - idx).to_string() }
2156            );
2157        }
2158
2159        Ok(())
2160    }
2161
2162    #[test]
2163    fn test_update_timestamp_ordering_rejected() -> Result<(), Box<dyn std::error::Error>>
2164    {
2165        let flushed_at = DateTime::from_timestamp_micros(1_000_000).unwrap();
2166        let id1 = SnapshotId::random();
2167        let snap1 = SnapshotInfo {
2168            id: id1.clone(),
2169            parent_id: None,
2170            flushed_at,
2171            message: "snap 1".to_string(),
2172            metadata: Default::default(),
2173        };
2174        let repo = RepoInfo::initial(
2175            SpecVersionBin::current(),
2176            snap1,
2177            100,
2178            None::<&()>,
2179            Some(flushed_at),
2180        );
2181
2182        // Attempting add_snapshot with a timestamp equal to the top of the ops log
2183        // should fail
2184        let id2 = SnapshotId::random();
2185        let snap2 = SnapshotInfo {
2186            id: id2.clone(),
2187            parent_id: Some(id1.clone()),
2188            flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2189            message: "snap 2".to_string(),
2190            metadata: Default::default(),
2191        };
2192        let result = repo.add_snapshot(
2193            SpecVersionBin::current(),
2194            snap2,
2195            Some("main"),
2196            UpdateType::NewCommitUpdate {
2197                branch: "main".to_string(),
2198                new_snap_id: id2.clone(),
2199            },
2200            Some(flushed_at),
2201            "backup",
2202            100,
2203        );
2204        assert!(matches!(
2205            result,
2206            Err(IcechunkFormatError {
2207                kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2208                ..
2209            })
2210        ));
2211
2212        // Attempting add_snapshot with a timestamp older than the top of the ops log
2213        // should also fail
2214        let flushed_at = DateTime::from_timestamp_micros(500_000).unwrap();
2215        let id3 = SnapshotId::random();
2216        let snap3 = SnapshotInfo {
2217            id: id3.clone(),
2218            parent_id: Some(id1.clone()),
2219            flushed_at,
2220            message: "snap 3".to_string(),
2221            metadata: Default::default(),
2222        };
2223        let result = repo.add_snapshot(
2224            SpecVersionBin::current(),
2225            snap3,
2226            Some("main"),
2227            UpdateType::NewCommitUpdate {
2228                branch: "main".to_string(),
2229                new_snap_id: id3.clone(),
2230            },
2231            Some(flushed_at),
2232            "backup",
2233            100,
2234        );
2235        assert!(matches!(
2236            result,
2237            Err(IcechunkFormatError {
2238                kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2239                ..
2240            })
2241        ));
2242
2243        // Attempting add_snapshot with a strictly newer timestamp should succeed
2244        let flushed_at = DateTime::from_timestamp_micros(2_000_000).unwrap();
2245        let id4 = SnapshotId::random();
2246        let snap4 = SnapshotInfo {
2247            id: id4.clone(),
2248            parent_id: Some(id1.clone()),
2249            flushed_at,
2250            message: "snap 4".to_string(),
2251            metadata: Default::default(),
2252        };
2253        let result = repo.add_snapshot(
2254            SpecVersionBin::current(),
2255            snap4,
2256            Some("main"),
2257            UpdateType::NewCommitUpdate {
2258                branch: "main".to_string(),
2259                new_snap_id: id4.clone(),
2260            },
2261            Some(flushed_at),
2262            "backup",
2263            100,
2264        );
2265        assert!(result.is_ok());
2266
2267        Ok(())
2268    }
2269
2270    /// Verify that `set_status` with `update_config = None` (silent mode)
2271    /// changes the repo status without appending an entry to the ops log.
2272    #[test]
2273    fn test_set_status_silent() -> Result<(), Box<dyn std::error::Error>> {
2274        let id1 = SnapshotId::random();
2275        let snap1 = SnapshotInfo {
2276            id: id1.clone(),
2277            parent_id: None,
2278            flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2279            message: "snap 1".to_string(),
2280            metadata: Default::default(),
2281        };
2282
2283        let repo =
2284            RepoInfo::initial(SpecVersionBin::current(), snap1, 100, None::<&()>, None);
2285
2286        // Add a branch to have more than one ops log entry
2287        let repo =
2288            repo.add_branch(SpecVersionBin::current(), "dev", &id1, "backup0", 100)?;
2289
2290        let updates_before: Vec<_> =
2291            repo.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2292        let rbu_before = repo.repo_before_updates()?;
2293        let num_updates_before = updates_before.len();
2294
2295        // Normal set_status: should add a RepoStatusChangedUpdate entry
2296        let new_status = RepoStatus {
2297            availability: RepoAvailability::ReadOnly,
2298            set_at: Utc::now(),
2299            limited_availability_reason: Some("maintenance".to_string()),
2300        };
2301        let repo_normal = repo.set_status(
2302            SpecVersionBin::current(),
2303            &new_status,
2304            Some(("backup1", 100)),
2305        )?;
2306        let updates_after_normal: Vec<_> =
2307            repo_normal.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2308        assert_eq!(updates_after_normal.len(), num_updates_before + 1);
2309        assert!(matches!(
2310            &updates_after_normal[0].0,
2311            UpdateType::RepoStatusChangedUpdate { .. }
2312        ));
2313        let normal_status = repo_normal.status()?;
2314        assert_eq!(normal_status.availability, RepoAvailability::ReadOnly);
2315        assert_eq!(
2316            normal_status.limited_availability_reason.as_deref(),
2317            Some("maintenance")
2318        );
2319
2320        // Silent set_status: should NOT add an ops log entry
2321        let silent_status = RepoStatus {
2322            availability: RepoAvailability::ReadOnly,
2323            set_at: Utc::now(),
2324            limited_availability_reason: None,
2325        };
2326        let repo_silent =
2327            repo.set_status(SpecVersionBin::current(), &silent_status, None)?;
2328
2329        let updates_after_silent: Vec<_> =
2330            repo_silent.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2331        // Ops log unchanged: same number of entries, same content
2332        assert_eq!(updates_after_silent.len(), num_updates_before);
2333        for (before, after) in updates_before.iter().zip(updates_after_silent.iter()) {
2334            assert_eq!(before.0, after.0);
2335            assert_eq!(before.1, after.1);
2336            assert_eq!(before.2, after.2);
2337        }
2338        assert_eq!(repo_silent.repo_before_updates()?, rbu_before);
2339
2340        // But the status itself IS changed
2341        let silent_result_status = repo_silent.status()?;
2342        assert_eq!(silent_result_status.availability, RepoAvailability::ReadOnly);
2343        assert!(silent_result_status.limited_availability_reason.is_none());
2344
2345        Ok(())
2346    }
2347}