///|
/// A Datalog relation: a set of tuples with delta tracking for fixpoint evaluation.
///
/// `Relation[T]` holds typed `current` and `delta` sets via `Ref`. The Runtime
/// sees only type-erased closures on `RelationData`.
///
/// - `insert()` adds to frontier delta outside fixpoint, and to staged delta during fixpoint
/// - `contains()` checks current (the materialized post-drain set)
/// - `iter()` iterates current; records a dependency for pull verification
/// - `delta_iter()` iterates delta; used by rule bodies
pub(all) struct Relation[T] {
  priv rt : Runtime
  priv cell_id : CellId
  priv current : Ref[@hashset.HashSet[T]]
  priv delta : Ref[@hashset.HashSet[T]]
  priv staged_delta : Ref[@hashset.HashSet[T]]
}

///|
/// Creates a new relation. Enables `Relation(rt)` /
/// `Relation(rt, label="...")` call sites.
pub fn[T : Hash + Eq] Relation::Relation(
  rt : Runtime,
  label? : String,
) -> Relation[T] {
  let idx = rt.datalog.relations.length()
  let cell_id = rt.alloc_cell_id(Relation(idx))
  let current : Ref[@hashset.HashSet[T]] = { val: @hashset.HashSet([]) }
  let delta : Ref[@hashset.HashSet[T]] = { val: @hashset.HashSet([]) }
  let staged_delta : Ref[@hashset.HashSet[T]] = { val: @hashset.HashSet([]) }
  let current_len_before_fixpoint : Ref[Int] = { val: 0 }
  let data : RelationData = {
    meta: {
      cell_id,
      label,
      changed_at: Revision::initial(),
      durability: Low,
      subscribers: @hashset.HashSet([]),
      push_reachable_count: 0,
    },
    drain_delta: () => {
      for item in delta.val {
        current.val.add(item)
      }
    },
    is_delta_empty: () => delta.val.is_empty(),
    promote_staged_delta: () => {
      let previous_frontier = delta.val
      delta.val = staged_delta.val
      staged_delta.val = previous_frontier
      // Reuse the old frontier allocation as next iteration's staging buffer.
      staged_delta.val.clear()
    },
    is_staged_delta_empty: () => staged_delta.val.is_empty(),
    begin_fixpoint: () => current_len_before_fixpoint.val = current.val.length(),
    finish_fixpoint_changed: () => {
      current.val.length() != current_len_before_fixpoint.val
    },
  }
  rt.datalog.relations.push(data)
  let ops : &CellOps = rt.datalog.relations[idx]
  rt.core.cell_ops.push(ops)
  let lifecycle : &CellLifecycle = rt.datalog.relations[idx]
  rt.cell_lifecycle.push(lifecycle)
  { rt, cell_id, current, delta, staged_delta }
}

///|
/// Deprecated alias of the `Relation::Relation` constructor.
#deprecated("Use the constructor form `Relation(rt)` (`Relation::Relation`) instead.")
pub fn[T : Hash + Eq] Relation::new(
  rt : Runtime,
  label? : String,
) -> Relation[T] {
  Relation(rt, label?)
}

///|
/// Returns the CellId for this relation.
pub fn[T] Relation::id(self : Relation[T]) -> CellId {
  self.cell_id
}

///|
/// Inserts a fact into the delta set.
///
/// Outside `fixpoint()`, inserts go to the current frontier delta.
/// During `fixpoint()`, inserts go to the staged delta for the next iteration.
///
/// Returns `true` if the fact was newly added, `false` if it already exists.
pub fn[T : Hash + Eq] Relation::insert(self : Relation[T], value : T) -> Bool {
  guard !self.rt.is_cell_disposed(self.cell_id) else {
    abort("Relation::insert called on a disposed relation")
  }
  if self.current.val.contains(value) {
    return false
  }
  if self.delta.val.contains(value) {
    return false
  }
  if self.rt.core.phase is InFixpoint {
    if self.staged_delta.val.contains(value) {
      return false
    }
    self.staged_delta.val.add(value)
  } else {
    self.delta.val.add(value)
  }
  true
}

///|
/// Checks whether a fact exists in the current (materialized) set.
///
/// Facts in delta are NOT visible via `contains` until after `fixpoint()` drains them.
/// Like `iter()`, this records a dependency for pull verification.
pub fn[T : Hash + Eq] Relation::contains(self : Relation[T], value : T) -> Bool {
  guard !self.rt.is_cell_disposed(self.cell_id) else {
    abort("Relation::contains called on a disposed relation")
  }
  self.record_read_dependency()
  self.current.val.contains(value)
}

///|
/// Iterates over the current (materialized) set.
///
/// Records a dependency so pull memos that call `iter()` automatically
/// re-verify when the relation changes after a `fixpoint()`.
pub fn[T] Relation::iter(self : Relation[T]) -> Iter[T] {
  guard !self.rt.is_cell_disposed(self.cell_id) else {
    abort("Relation::iter called on a disposed relation")
  }
  self.record_read_dependency()
  self.current.val.iter()
}

///|
/// Records dependency for relation reads and rejects cross-runtime reads while
/// a memo/reaction computation is active.
fn[T] Relation::record_read_dependency(self : Relation[T]) -> Unit {
  self.rt.check_cross_runtime(self.rt.core.runtime_id, "Relation")
  Tracker::record_dependency(self.rt, self.cell_id)
}

///|
/// Disposes this relation, clearing its fact sets and marking it as Disposed.
///
/// A live rule pins every declared input and output relation. Disposal aborts
/// until those rules are disposed. Repeated disposal is a no-op after the
/// relation is disposed; all current and delta reads abort after disposal.
pub fn[T] Relation::dispose(self : Relation[T]) -> Unit {
  self.rt.dispose_cell(self.cell_id)
  self.current.val.clear()
  self.delta.val.clear()
  self.staged_delta.val.clear()
}

///|
/// Returns true if this relation has been disposed.
pub fn[T] Relation::is_disposed(self : Relation[T]) -> Bool {
  self.rt.is_cell_disposed(self.cell_id)
}

///|
/// Iterates over the delta set (new facts not yet drained to current).
///
/// Used by rule bodies to read only the new facts produced in the previous
/// fixpoint iteration. All relation reads abort after disposal.
pub fn[T] Relation::delta_iter(self : Relation[T]) -> Iter[T] {
  guard !self.rt.is_cell_disposed(self.cell_id) else {
    abort("Relation::delta_iter called on a disposed relation")
  }
  self.delta.val.iter()
}