///|
/// Returns the strict majority required by a cluster size.
pub fn quorum_size(cluster_size : Int) -> Int {
  if cluster_size <= 0 {
    0
  } else {
    cluster_size / 2 + 1
  }
}

///|
/// Selects highest priority and resolves ties by lexicographically smaller id.
pub fn select_candidate(candidates : Array[Candidate]) -> Candidate? {
  let mut selected : Candidate? = None
  for candidate in candidates {
    if candidate.id != "" {
      match selected {
        None => selected = Some(candidate)
        Some(current) =>
          if candidate.priority > current.priority ||
            (
              candidate.priority == current.priority &&
              candidate.id < current.id
            ) {
            selected = Some(candidate)
          }
      }
    }
  }
  selected
}

///|
fn contains_string(values : Array[String], value : String) -> Bool {
  for item in values {
    if item == value {
      return true
    }
  }
  false
}

///|
fn find_tally(tallies : Array[VoteTally], candidate : String) -> Int {
  for index, tally in tallies {
    if tally.candidate == candidate {
      return index
    }
  }
  -1
}

///|
fn find_vote(votes : Array[ElectionVote], voter : String) -> ElectionVote? {
  for vote in votes {
    if vote.voter == voter {
      return Some(vote)
    }
  }
  None
}

///|
/// Validates a term's votes and returns deterministic quorum evidence.
pub fn certify_votes(
  votes : Array[ElectionVote],
  cluster_size : Int,
  term : Int,
) -> QuorumCertificate {
  let quorum = quorum_size(cluster_size)
  if quorum == 0 {
    return {
      valid: false,
      leader: "",
      term,
      quorum,
      voters: [],
      equivocations: [],
      message: "cluster size must be positive",
    }
  }
  if term < 0 {
    return {
      valid: false,
      leader: "",
      term,
      quorum,
      voters: [],
      equivocations: [],
      message: "term must not be negative",
    }
  }
  let seen_votes : Array[ElectionVote] = []
  let tallies : Array[VoteTally] = []
  let equivocations : Array[String] = []
  for vote in votes {
    if vote.voter == "" || vote.candidate == "" || vote.term != term {
      continue
    }
    match find_vote(seen_votes, vote.voter) {
      Some(previous) =>
        if previous.granted &&
          vote.granted &&
          previous.candidate != vote.candidate &&
          !contains_string(equivocations, vote.voter) {
          equivocations.push(vote.voter)
        }
      None => {
        seen_votes.push(vote)
        if vote.granted {
          let index = find_tally(tallies, vote.candidate)
          if index < 0 {
            tallies.push({ candidate: vote.candidate, voters: [vote.voter] })
          } else {
            tallies[index].voters.push(vote.voter)
          }
        }
      }
    }
  }
  if equivocations.length() > 0 {
    return {
      valid: false,
      leader: "",
      term,
      quorum,
      voters: [],
      equivocations,
      message: "equivocating voters detected",
    }
  }
  let mut winner : VoteTally? = None
  for tally in tallies {
    if tally.voters.length() >= quorum {
      match winner {
        None => winner = Some(tally)
        Some(current) =>
          if tally.voters.length() > current.voters.length() ||
            (
              tally.voters.length() == current.voters.length() &&
              tally.candidate < current.candidate
            ) {
            winner = Some(tally)
          }
      }
    }
  }
  match winner {
    Some(tally) =>
      {
        valid: true,
        leader: tally.candidate,
        term,
        quorum,
        voters: tally.voters,
        equivocations: [],
        message: "quorum certificate valid",
      }
    None =>
      {
        valid: false,
        leader: "",
        term,
        quorum,
        voters: [],
        equivocations: [],
        message: "no candidate reached quorum",
      }
  }
}

///|
/// Acquires a leadership lease only when a quorum certificate is valid.
pub fn LeaseTable::acquire_leadership(
  self : LeaseTable,
  resource : String,
  certificate : QuorumCertificate,
  ttl : Int,
  now : Int,
) -> TableDecision {
  if certificate.valid {
    self.apply(resource, Acquire(certificate.leader, ttl, None), now)
  } else {
    let state = match self.get(resource) {
      Some(state) => state
      None => LeaseState::new(resource)
    }
    let decision : LeaseDecision = {
      state,
      accepted: false,
      kind: Rejected,
      lease: state.lease,
      event: None,
      message: "valid quorum certificate required",
    }
    {
      table: { ..self, rejected_commands: self.rejected_commands + 1 },
      decision,
    }
  }
}