///| Hub notes storage (ObjectStore + RefStore based)
///
/// ## Vector Clock Merge Semantics
///
/// - Clock merge: take max of each node's counter
/// - Same-key conflict: higher timestamp wins; if equal, lexicographic node name wins
/// - Tombstone: deleted=true marks logical deletion; compaction is not yet implemented

///|
let hub_notes_ns : String = "bit-hub"

///|
let hub_notes_ref : String = "refs/notes/" + hub_notes_ns

///|
let work_item_meta_prefix_value : String = "hub/work-item/"

///|
let canonical_work_item_record_kind_value : String = "work-item"

///|
pub fn hub_notes_ref_name() -> String {
  hub_notes_ref
}

///|
pub fn hub_notes_ns_name() -> String {
  hub_notes_ns
}

///|
pub fn work_item_meta_prefix() -> String {
  work_item_meta_prefix_value
}

///|
pub fn work_item_meta_key(work_item_id : String) -> String {
  work_item_meta_prefix_value + work_item_id + "/meta"
}

///|
pub fn pr_proposal_prefix() -> String {
  "hub/proposal/pr/"
}

///|
pub fn pr_proposal_meta_key(pr_id : String) -> String {
  pr_proposal_prefix() + pr_id + "/meta"
}

///|
pub fn pr_comment_prefix(pr_id : String) -> String {
  "hub/pr/" + pr_id + "/comment/"
}

///|
pub fn pr_comment_key(pr_id : String, comment_id : String) -> String {
  pr_comment_prefix(pr_id) + comment_id
}

///|
pub fn pr_review_prefix(pr_id : String) -> String {
  "hub/pr/" + pr_id + "/review/"
}

///|
pub fn pr_review_key(pr_id : String, review_id : String) -> String {
  pr_review_prefix(pr_id) + review_id
}

///|
pub fn issue_comment_prefix(issue_id : String) -> String {
  "hub/issue/" + issue_id + "/comment/"
}

///|
pub fn issue_comment_key(issue_id : String, comment_id : String) -> String {
  issue_comment_prefix(issue_id) + comment_id
}

///|
pub fn canonical_work_item_record_kind() -> String {
  canonical_work_item_record_kind_value
}

///|
pub fn is_work_item_record_kind(kind : String) -> Bool {
  kind == canonical_work_item_record_kind_value
}

///|
let sync_link_record_kind_value : String = "sync-link"

///|
pub fn sync_link_record_kind() -> String {
  sync_link_record_kind_value
}

///|
pub fn is_sync_link_record_kind(kind : String) -> Bool {
  kind == sync_link_record_kind_value
}

///|
pub fn sync_link_key(local_id : String, provider : String) -> String {
  "hub/sync-link/" + local_id + "/" + provider
}

///|
pub struct HubRecord {
  version : Int
  key : String
  kind : String
  clock : Map[String, Int64]
  timestamp : Int64
  node : String
  deleted : Bool
  signature : String?
  payload : String
}

///|
pub(all) enum RecordTriggerClass {
  Full
  NotifyOnly
} derive(Eq, Debug)

///|
pub impl Show for RecordTriggerClass with fn output(self, logger) {
  match self {
    Full => logger.write_string("Full")
    NotifyOnly => logger.write_string("NotifyOnly")
  }
}

///|
pub fn classify_record_trigger(record : HubRecord) -> RecordTriggerClass {
  if record.kind == "pr-proposal" ||
    record.key.has_prefix("hub/proposal/") ||
    record.kind == "feature-broadcast" ||
    record.key.has_prefix("hub/broadcast/") {
    RecordTriggerClass::NotifyOnly
  } else {
    RecordTriggerClass::Full
  }
}

///|
pub fn record_notify_topic(record : HubRecord) -> String? {
  if classify_record_trigger(record) == RecordTriggerClass::Full {
    return None
  }
  match record.kind {
    "pr-proposal" => Some("hub.pr.proposal")
    "feature-broadcast" => Some("hub.feature.broadcast")
    _ => Some("hub.notify")
  }
}

///|
pub fn HubRecord::new(
  key : String,
  kind : String,
  payload : String,
  node : String,
  timestamp : Int64,
  clock? : Map[String, Int64] = Map([]),
  deleted? : Bool = false,
  version? : Int = 1,
  signature? : String? = None,
) -> HubRecord {
  { version, key, kind, clock, timestamp, node, deleted, signature, payload }
}

///|
fn serialize_hub_record(record : HubRecord, signature : String?) -> String {
  let sb = StringBuilder::new()
  sb.write_string("version ")
  sb.write_string(record.version.to_string())
  sb.write_char('\n')
  sb.write_string("key ")
  sb.write_string(record.key)
  sb.write_char('\n')
  sb.write_string("kind ")
  sb.write_string(record.kind)
  sb.write_char('\n')
  sb.write_string("clock ")
  sb.write_string(clock_to_string(record.clock))
  sb.write_char('\n')
  sb.write_string("timestamp ")
  sb.write_string(record.timestamp.to_string())
  sb.write_char('\n')
  sb.write_string("node ")
  sb.write_string(record.node)
  sb.write_char('\n')
  sb.write_string("deleted ")
  sb.write_string(if record.deleted { "1" } else { "0" })
  sb.write_char('\n')
  match signature {
    Some(sig) => {
      sb.write_string("signature ")
      sb.write_string(sig)
      sb.write_char('\n')
    }
    None => ()
  }
  sb.write_char('\n')
  sb.write_string(record.payload)
  sb.to_string()
}

///|
fn serialize_hub_record_unsigned(record : HubRecord) -> String {
  serialize_hub_record(record, None)
}

///|
pub fn HubRecord::serialize(self : HubRecord) -> String {
  serialize_hub_record(self, self.signature)
}

///|
pub fn parse_hub_record(text : String) -> HubRecord {
  let mut version = 0
  let mut key = ""
  let mut kind = ""
  let mut clock = Map([])
  let mut timestamp : Int64 = 0
  let mut node = ""
  let mut deleted = false
  let mut signature : String? = None
  let payload_lines : Array[String] = []
  let mut in_body = false
  for line_view in text.split("\n") {
    let line = line_view.to_owned()
    if in_body {
      payload_lines.push(line)
      continue
    }
    if line.length() == 0 {
      in_body = true
      continue
    }
    let space = line.find(" ")
    match space {
      None => continue
      Some(idx) => {
        let k = String::unsafe_substring(line, start=0, end=idx)
        let v = String::unsafe_substring(line, start=idx + 1, end=line.length())
        match k {
          "version" => version = parse_int(v)
          "key" => key = v
          "kind" => kind = v
          "clock" => clock = parse_clock(v)
          "timestamp" => timestamp = parse_int64(v)
          "node" => node = v
          "deleted" => deleted = v == "1" || v == "true" || v == "yes"
          "signature" => signature = Some(v)
          _ => ()
        }
      }
    }
  }
  HubRecord::new(
    key,
    kind,
    payload_lines.join("\n"),
    node,
    timestamp,
    clock~,
    deleted~,
    version~,
    signature~,
  )
}

///|
fn parse_clock(s : String) -> Map[String, Int64] {
  let result : Map[String, Int64] = Map([])
  if s.length() == 0 {
    return result
  }
  for part_view in s.split(",") {
    let part = part_view.to_owned()
    if part.length() == 0 {
      continue
    }
    let eq = part.find("=")
    match eq {
      None => continue
      Some(idx) => {
        let key = String::unsafe_substring(part, start=0, end=idx)
        let value = String::unsafe_substring(
          part,
          start=idx + 1,
          end=part.length(),
        )
        let parsed = parse_int64(value)
        result[key] = parsed
      }
    }
  }
  result
}

///|
fn clock_to_string(clock : Map[String, Int64]) -> String {
  if clock.length() == 0 {
    return ""
  }
  let items = clock.to_array()
  items.sort_by((a, b) => String::compare(a.0, b.0))
  let parts : Array[String] = []
  for item in items {
    let (key, value) = item
    parts.push(key + "=" + value.to_string())
  }
  parts.join(",")
}

///|

///|
/// Compare two vector clocks. Returns 1 if a > b, -1 if b > a, 0 if concurrent.
pub fn compare_vector_clocks(
  a : Map[String, Int64],
  b : Map[String, Int64],
) -> Int {
  compare_clock(a, b)
}

///|
fn compare_clock(a : Map[String, Int64], b : Map[String, Int64]) -> Int {
  let mut a_gt = false
  let mut b_gt = false
  // Walk a's entries; b-only entries are handled by the second pass.
  for key, va in a {
    let vb = b.get(key).unwrap_or(0)
    if va > vb {
      a_gt = true
    } else if vb > va {
      b_gt = true
    }
  }
  // b-only keys: a's missing value is treated as 0.
  for key, vb in b {
    if a.contains(key) {
      continue
    }
    if vb > 0 {
      b_gt = true
    } else if vb < 0 {
      a_gt = true
    }
  }
  if a_gt && !b_gt {
    1
  } else if b_gt && !a_gt {
    -1
  } else {
    0
  }
}

///|
fn record_is_newer(a : HubRecord, b : HubRecord) -> Bool {
  let cmp = compare_clock(a.clock, b.clock)
  if cmp > 0 {
    return true
  }
  if cmp < 0 {
    return false
  }
  if a.timestamp != b.timestamp {
    return a.timestamp > b.timestamp
  }
  String::compare(a.node, b.node) >= 0
}

///|
pub enum RecordMergePolicy {
  Lww
  TombstoneWins
}

///|
pub fn parse_record_merge_policy(value : String) -> RecordMergePolicy? {
  match value.to_lower() {
    "lww" => Some(RecordMergePolicy::Lww)
    "tombstone-wins" | "deleted-wins" => Some(RecordMergePolicy::TombstoneWins)
    _ => None
  }
}

///|
pub fn default_record_merge_policy() -> RecordMergePolicy {
  RecordMergePolicy::Lww
}

///|
fn select_merge_winner(
  ours : HubRecord,
  theirs : HubRecord,
  conflict_policy : RecordMergePolicy,
) -> HubRecord {
  match conflict_policy {
    RecordMergePolicy::TombstoneWins =>
      if ours.deleted != theirs.deleted {
        if theirs.deleted {
          theirs
        } else {
          ours
        }
      } else if record_is_newer(theirs, ours) {
        theirs
      } else {
        ours
      }
    RecordMergePolicy::Lww =>
      if record_is_newer(theirs, ours) {
        theirs
      } else {
        ours
      }
  }
}

///|
fn is_valid_record(record : HubRecord) -> Bool {
  record.key.length() > 0 && record.kind.length() > 0
}

///|
fn compute_record_signature(record : HubRecord, signing_key : String) -> String {
  let state = @bit.Sha1State::new()
  state.update_string(serialize_hub_record_unsigned(record))
  state.update_string("\n--signing-key--\n")
  state.update_string(signing_key)
  state.finish().to_hex()
}

///|
fn verify_record_signature(record : HubRecord, signing_key : String) -> Bool {
  match record.signature {
    Some(sig) => sig == compute_record_signature(record, signing_key)
    None => false
  }
}

///|
fn sign_record(record : HubRecord, signing_key : String) -> HubRecord {
  let signature = compute_record_signature(record, signing_key)
  HubRecord::new(
    record.key,
    record.kind,
    record.payload,
    record.node,
    record.timestamp,
    clock=record.clock,
    deleted=record.deleted,
    version=record.version,
    signature=Some(signature),
  )
}

///|
pub fn merge_record_bytes(ours : Bytes, theirs : Bytes) -> Bytes {
  merge_record_bytes_with_policy(ours, theirs, RecordMergePolicy::Lww)
}

///|
pub fn merge_record_bytes_with_policy(
  ours : Bytes,
  theirs : Bytes,
  conflict_policy : RecordMergePolicy,
) -> Bytes {
  let ours_text = @utf8.decode_lossy(ours[:])
  let theirs_text = @utf8.decode_lossy(theirs[:])
  let ours_record = parse_hub_record(ours_text)
  let theirs_record = parse_hub_record(theirs_text)
  if !(ours_record |> is_valid_record) && is_valid_record(theirs_record) {
    return theirs
  }
  if !(theirs_record |> is_valid_record) {
    return ours
  }
  let winner = select_merge_winner(ours_record, theirs_record, conflict_policy)
  @utf8.encode(winner.serialize())
}

///|
pub struct HubStore {
  node_id : String
  signing_key : String?
  require_signed : Bool
  entries : Map[String, @bit.ObjectId] // entry_name -> blob_id
  mut head : @bit.ObjectId? // current notes commit
}

///|
pub fn HubStore::load(
  objects : &@lib.ObjectStore,
  refs : &@lib.RefStore,
  node_id? : String = "local",
  signing_key? : String? = None,
  require_signed? : Bool = false,
) -> HubStore raise @bit.GitError {
  let node = node_id
  let key = signing_key
  let require_signed_val = require_signed
  let commit = refs.resolve(hub_notes_ref)
  let entries : Map[String, @bit.ObjectId] = Map([])
  let head = match commit {
    Some(cid) => {
      let obj = objects.get(cid)
      guard obj is Some(commit_obj) else { None }
      let info = @bit.parse_commit(commit_obj.data)
      let tree_obj = objects.get(info.tree)
      guard tree_obj is Some(tobj) else { None }
      let tree_entries = @bit.parse_tree(tobj.data)
      for entry in tree_entries {
        entries[entry.name] = entry.id
      }
      Some(cid)
    }
    None => None
  }
  {
    node_id: node,
    signing_key: key,
    require_signed: require_signed_val,
    entries,
    head,
  }
}

///|
pub fn HubStore::node_id(self : HubStore) -> String {
  self.node_id
}

///|
pub fn HubStore::signing_key(self : HubStore) -> String? {
  self.signing_key
}

///|
pub fn HubStore::require_signed(self : HubStore) -> Bool {
  self.require_signed
}

///|
fn verify_record_by_policy(store : HubStore, record : HubRecord) -> Bool {
  match store.signing_key {
    Some(key) =>
      match record.signature {
        Some(_) => verify_record_signature(record, key)
        None => !store.require_signed
      }
    None => !store.require_signed
  }
}

///|
fn ensure_signature_ready_for_write(
  store : HubStore,
) -> Unit raise @bit.GitError {
  if store.require_signed && store.signing_key is None {
    raise @bit.GitError::InvalidObject(
      "Hub signing is required but signing key is not configured",
    )
  }
}

///|
pub fn HubStore::get_record(
  self : HubStore,
  objects : &@lib.ObjectStore,
  key : String,
) -> HubRecord? {
  let target_id = key_to_target_id(key)
  let entry_name = target_id.to_hex()
  let blob_id = self.entries.get(entry_name)
  match blob_id {
    None => None
    Some(bid) => {
      let obj = objects.get(bid) catch { _ => None }
      match obj {
        None => None
        Some(blob_obj) => {
          let text = @utf8.decode_lossy(blob_obj.data[:])
          let record = parse_hub_record(text)
          if !verify_record_by_policy(self, record) {
            None
          } else if record.deleted {
            None
          } else {
            Some(record)
          }
        }
      }
    }
  }
}

///|
pub fn HubStore::put_record(
  self : HubStore,
  objects : &@lib.ObjectStore,
  refs : &@lib.RefStore,
  clock : &@lib.Clock,
  key : String,
  kind : String,
  payload : String,
  node : String,
) -> HubRecord raise @bit.GitError {
  ensure_signature_ready_for_write(self)
  let timestamp = clock.now()
  let existing = self.get_record(objects, key)
  let base_clock = match existing {
    Some(r) => r.clock
    None => Map([])
  }
  let next_clock = increment_clock(base_clock, node)
  let unsigned_record = HubRecord::new(
    key,
    kind,
    payload,
    node,
    timestamp,
    clock=next_clock,
    deleted=false,
  )
  let record = match self.signing_key {
    Some(sign_key) => sign_record(unsigned_record, sign_key)
    None => unsigned_record
  }
  let target_id = key_to_target_id(key)
  // Store the key blob
  let key_bytes = @utf8.encode(key)
  ignore(objects.put(@bit.ObjectType::Blob, key_bytes))
  // Store the record blob
  let record_text = record.serialize()
  let record_bytes = @utf8.encode(record_text)
  let record_blob_id = objects.put(@bit.ObjectType::Blob, record_bytes)
  self.entries[target_id.to_hex()] = record_blob_id
  let _ = commit_notes(self, objects, refs, "Update " + kind, timestamp)
  record
}

///|
pub fn HubStore::delete_record(
  self : HubStore,
  objects : &@lib.ObjectStore,
  refs : &@lib.RefStore,
  clock : &@lib.Clock,
  key : String,
  kind : String,
  node : String,
) -> HubRecord raise @bit.GitError {
  ensure_signature_ready_for_write(self)
  let timestamp = clock.now()
  let existing = self.get_record(objects, key)
  let base_clock = match existing {
    Some(r) => r.clock
    None => Map([])
  }
  let next_clock = increment_clock(base_clock, node)
  let unsigned_record = HubRecord::new(
    key,
    kind,
    "",
    node,
    timestamp,
    clock=next_clock,
    deleted=true,
  )
  let record = match self.signing_key {
    Some(sign_key) => sign_record(unsigned_record, sign_key)
    None => unsigned_record
  }
  let target_id = key_to_target_id(key)
  // Store the key blob
  let key_bytes = @utf8.encode(key)
  ignore(objects.put(@bit.ObjectType::Blob, key_bytes))
  // Store the record blob
  let record_text = record.serialize()
  let record_bytes = @utf8.encode(record_text)
  let record_blob_id = objects.put(@bit.ObjectType::Blob, record_bytes)
  self.entries[target_id.to_hex()] = record_blob_id
  let _ = commit_notes(self, objects, refs, "Delete " + kind, timestamp)
  record
}

///|
pub fn HubStore::list_records(
  self : HubStore,
  objects : &@lib.ObjectStore,
  prefix : String,
  include_deleted? : Bool = false,
) -> Array[HubRecord] {
  let result : Array[HubRecord] = []
  let include_deleted_val = include_deleted
  for blob_id in self.entries.values() {
    let obj = objects.get(blob_id) catch { _ => None }
    match obj {
      None => ()
      Some(blob_obj) => {
        let text = @utf8.decode_lossy(blob_obj.data[:])
        let record = parse_hub_record(text)
        if !verify_record_by_policy(self, record) {
          continue
        }
        if !record.key.has_prefix(prefix) {
          continue
        }
        if !include_deleted_val && record.deleted {
          continue
        }
        result.push(record)
      }
    }
  }
  result
}

///|
fn increment_clock(
  clock : Map[String, Int64],
  node : String,
) -> Map[String, Int64] {
  let next = clock
  let current = next.get(node).unwrap_or(0)
  next[node] = current + 1
  next
}

///|
fn key_to_target_id(key : String) -> @bit.ObjectId {
  let (id, _compressed) = @bit.create_blob_string(key)
  id
}

///|
fn commit_notes(
  store : HubStore,
  objects : &@lib.ObjectStore,
  refs : &@lib.RefStore,
  message : String,
  timestamp : Int64,
) -> @bit.ObjectId raise @bit.GitError {
  // Build tree from entries
  let tree_entries : Array[@bit.TreeEntry] = Array::new(
    capacity=store.entries.length(),
  )
  for name, blob_id in store.entries {
    tree_entries.push(@bit.TreeEntry::new("100644", name, blob_id))
  }
  tree_entries.sort_by((a, b) => String::compare(a.name, b.name))
  let tree_bytes = @bit.serialize_tree(tree_entries)
  let tree_id = objects.put(@bit.ObjectType::Tree, tree_bytes)
  // Create commit
  let parents = match store.head {
    Some(p) => [p]
    None => []
  }
  let commit = @bit.Commit::new(
    tree_id,
    parents,
    "Hub ",
    timestamp,
    "+0000",
    "Hub ",
    timestamp,
    "+0000",
    message + "\n",
  )
  let commit_bytes = @bit.serialize_commit_content(commit)
  let commit_id = objects.put(@bit.ObjectType::Commit, commit_bytes)
  store.head = Some(commit_id)
  refs.update(hub_notes_ref, Some(commit_id))
  commit_id
}