///|
/// Compaction and garbage collection.
///
/// The image is rewritten in full on every commit, so its size is the price
/// of each commit. Compaction folds a prefix of the entries into a snapshot
/// object and drops them from the image, keeping that price bounded.

///|
/// What a fold produced: the snapshot bytes, and the payload keys the
/// snapshot still needs.
///
/// Reporting the retained payloads is the caller's job because only the
/// caller knows which of the folded entries' payloads survived the fold — a
/// Git repack, for instance, replaces several packs with one.
pub(all) struct Fold {
  body : Bytes
  retained_payloads : Array[String]
}

///|
/// Fold entries up to and including `through_seq` into a new snapshot.
///
/// `fold` receives the previous snapshot (empty on the first compaction) and
/// the entries being folded, and returns the new snapshot. It must be a pure
/// function of those inputs: it may run again after a lost race.
///
/// Returns the sequence the log is now folded through.
pub async fn[S : @objstore.ObjectStore] Wal::compact(
  self : Wal[S],
  through_seq : Int64,
  fold : async (Bytes, Array[WalEntry]) -> Fold raise @bit.GitError,
) -> CommitOutcome raise @bit.GitError {
  for _ in 0.. ()
      NotApplied =>
        raise @bit.GitError::IoError(
          "snapshot write refused for \{snapshot_id}",
        )
      Unknown => {
        // The snapshot may or may not be there. Retrying is safe because the
        // key is derived from the content.
        self.invalidate()
        continue
      }
    }
    let next_image : WalImage = {
      version: WAL_FORMAT_VERSION,
      snapshot: snapshot_id,
      snapshot_seq: through_seq,
      snapshot_payloads: folded.retained_payloads,
      entries: remaining,
      settings: self.image.settings,
    }
    let condition = if self.etag == "" {
      @objstore.PutCondition::IfNotExists
    } else {
      @objstore.PutCondition::IfMatch(self.etag)
    }
    match self.store.put(self.image_key(), encode_image(next_image), condition) {
      Applied(new_etag) => {
        self.image = next_image
        self.etag = new_etag
        self.loaded = true
        return CommitOutcome::Committed(through_seq)
      }
      NotApplied => self.invalidate()
      Unknown => self.invalidate()
    }
  }
  CommitOutcome::Contended
}

///|
/// Read the current snapshot, or empty bytes when nothing has been folded.
pub async fn[S : @objstore.ObjectStore] Wal::read_snapshot(
  self : Wal[S],
) -> Bytes raise @bit.GitError {
  if self.image.snapshot == "" {
    return Bytes::new(0)
  }
  let key = self.snapshot_key(self.image.snapshot)
  let (body, _) = self.store.get(key).unwrap(key)
  body
}

///|
/// Keys that a sweep would delete: payloads and snapshots no longer
/// referenced by the image.
///
/// Separated from the deletion itself so a maintainer can report a sweep
/// before performing one, and so the decision is testable without destroying
/// anything.
pub async fn[S : @objstore.ObjectStore] Wal::unreferenced_keys(
  self : Wal[S],
) -> Array[String] raise @bit.GitError {
  self.sync()
  let live : Map[String, Bool] = Map([])
  for id in self.image.live_payloads() {
    live[self.payload_key(id)] = true
  }
  if self.image.snapshot != "" {
    live[self.snapshot_key(self.image.snapshot)] = true
  }
  let out : Array[String] = []
  for prefix in ["\{self.prefix}/payloads/", "\{self.prefix}/snapshots/"] {
    for entry in @objstore.list_all(self.store, prefix) {
      if !live.contains(entry.key) {
        out.push(entry.key)
      }
    }
  }
  out
}

///|
/// Delete unreferenced payloads and snapshots.
///
/// **Not safe to run against a log with writers in flight.** A payload is
/// written before the entry that references it lands, so between those two
/// moments it looks exactly like garbage. waltier makes the same restriction
/// for the same reason: collection is an offline operation.
///
/// Returns the keys removed.
pub async fn[S : @objstore.ObjectStore] Wal::sweep_offline(
  self : Wal[S],
) -> Array[String] raise @bit.GitError {
  let doomed = self.unreferenced_keys()
  for key in doomed {
    self.store.delete(key)
  }
  doomed
}