///|
/// Per-cell state for AccumulatorCommitHook between before_recompute and
/// after_success/after_abort. Snapshot of prev contributions + staged
/// touched slots and synthetic reads.
priv struct RecomputeState {
  prev_contributions : Array[@incr_types.AccumulatorId]
  mut touched : @hashset.HashSet[@incr_types.AccumulatorId]?
  mut reads : @hashmap.HashMap[
    (@incr_types.AccumulatorId, CellId),
    @incr_types.Revision,
  ]?
}

///|
/// MemoCommitPhase implementor owning per-cell accumulator state across a
/// memo recompute. `active` is keyed by the recomputing `CellId`; the hook
/// pair owns each entry's lifecycle: `before_recompute(c)` installs the
/// entry (or `ensure_for_cell` installs it lazily from a push path when
/// `before_recompute` was a no-op), and exactly one of `after_success(c)`
/// or `after_abort(c)` removes it. Hooks fire serially on the commit hot
/// path; implementors must not re-enter graph reads or user code.
priv struct AccumulatorCommitHook {
  active : @hashmap.HashMap[CellId, RecomputeState]
}

///|
fn AccumulatorCommitHook::new() -> AccumulatorCommitHook {
  { active: @hashmap.HashMap([]) }
}

///|
/// Returns-or-creates the entry for the given recomputing cell. Used by the
/// three accumulator push paths so they can lazily install a hook entry when
/// `before_recompute` was skipped (e.g. when no accumulator existed at
/// recompute-start time but one was registered mid-recompute).
fn AccumulatorCommitHook::ensure_for_cell(
  self : AccumulatorCommitHook,
  cell_id : CellId,
) -> RecomputeState {
  match self.active.get(cell_id) {
    Some(state) => state
    None => {
      let state : RecomputeState = {
        prev_contributions: [],
        touched: None,
        reads: None,
      }
      self.active.set(cell_id, state)
      state
    }
  }
}

///|
fn RecomputeState::ensure_touched(
  self : RecomputeState,
) -> @hashset.HashSet[@incr_types.AccumulatorId] {
  match self.touched {
    Some(s) => s
    None => {
      let s = @hashset.HashSet([])
      self.touched = Some(s)
      s
    }
  }
}

///|
fn RecomputeState::ensure_reads(
  self : RecomputeState,
) -> @hashmap.HashMap[(@incr_types.AccumulatorId, CellId), @incr_types.Revision] {
  match self.reads {
    Some(m) => m
    None => {
      let m = @hashmap.HashMap([])
      self.reads = Some(m)
      m
    }
  }
}

///|
/// before_recompute: snapshot prev_contributions buffers (clear so this
/// run starts empty); create active entry keyed by recomputing cell.
///
/// Fast-path: skip when no accumulators have ever been registered.
/// `accumulator_slots` is monotonic (disposed slots stay; no reuse), so
/// `is_empty()` means no slot has ever existed — accumulator_contributions
/// is necessarily empty too, the snapshot loop is a no-op, and any push
/// during this recompute will use `ensure_for_cell` to lazy-create the
/// entry on demand.
impl MemoCommitPhase for AccumulatorCommitHook with fn before_recompute(
  self,
  rt,
  cell_id,
) {
  if rt.accumulator_slots.is_empty() {
    return
  }
  let prev : Array[@incr_types.AccumulatorId] = match
    rt.accumulator_contributions.get(cell_id) {
    Some(s) => s.to_array()
    None => []
  }
  for slot_id in prev {
    let slot = rt.accumulator_slots[slot_id.id]
    if !slot.disposed {
      (slot.snapshot_and_clear)(cell_id)
    }
  }
  self.active.set(cell_id, {
    prev_contributions: prev,
    touched: None,
    reads: None,
  })
}

///|
/// after_abort: restore prev snapshots; clear new-run buffer for
/// touched-not-prev slots (preserves push_revised_at).
impl MemoCommitPhase for AccumulatorCommitHook with fn after_abort(
  self,
  rt,
  cell_id,
  _e,
) {
  // No entry → before_recompute skipped (fast-path) AND no push happened
  // during this aborted run. Nothing to restore or clean up.
  let state = match self.active.get(cell_id) {
    Some(s) => s
    None => return
  }
  self.active.remove(cell_id)
  let touched : Array[@incr_types.AccumulatorId] = match state.touched {
    Some(s) => s.to_array()
    None => []
  }
  for slot_id in state.prev_contributions {
    let slot = rt.accumulator_slots[slot_id.id]
    if !slot.disposed {
      (slot.restore_buffer)(cell_id)
    }
  }
  for slot_id in touched {
    let mut in_prev = false
    for p in state.prev_contributions {
      if p == slot_id {
        in_prev = true
        break
      }
    }
    if in_prev {
      continue
    }
    let slot = rt.accumulator_slots[slot_id.id]
    if !slot.disposed {
      (slot.clear_new_run_buffer)(cell_id)
    }
  }
}

///|
/// after_success: atomic per-slot finalize over prev ∪ touched; commit
/// staged reads + rebuilt contributions.
impl MemoCommitPhase for AccumulatorCommitHook with fn after_success(
  self,
  rt,
  cell_id,
) {
  // No entry → before_recompute skipped (fast-path) AND no push/read
  // happened during this run. Nothing to finalize or commit.
  let state = match self.active.get(cell_id) {
    Some(s) => s
    None => return
  }
  self.active.remove(cell_id)
  let cell = rt.get_memo_data(cell_id)
  let all_slots : @hashset.HashSet[@incr_types.AccumulatorId] = @hashset.HashSet([],
  )
  for s in state.prev_contributions {
    all_slots.add(s)
  }
  match state.touched {
    Some(t) =>
      for s in t {
        all_slots.add(s)
      }
    None => ()
  }
  let current_revision = rt.core.revision.current_revision
  let new_contributions : @hashset.HashSet[@incr_types.AccumulatorId] = @hashset.HashSet([],
  )
  for slot_id in all_slots {
    let slot = rt.accumulator_slots[slot_id.id]
    if slot.disposed {
      continue
    }
    (slot.finalize_memo)(cell_id, current_revision)
    if (slot.has_buffer_for)(cell_id) {
      new_contributions.add(slot_id)
    }
  }
  cell.accumulator_reads.clear()
  match state.reads {
    Some(reads) =>
      for k, v in reads {
        cell.accumulator_reads.set(k, v)
      }
    None => ()
  }
  if new_contributions.is_empty() {
    rt.accumulator_contributions.remove(cell_id)
  } else {
    rt.accumulator_contributions.set(cell_id, new_contributions)
  }
}