///|
/// Create a PR proposal record (notify-only event stream)
pub fn Hub::propose_pr(
self : Hub,
objects : &@lib.ObjectStore,
refs : &@lib.RefStore,
clock : &@lib.Clock,
title : String,
body : String,
source_branch : String,
target_branch : String,
author : String,
source_repo? : String? = None,
source_ref? : String? = None,
closes_issues? : Array[String] = [],
) -> PullRequest raise @bit.GitError {
let timestamp = clock.now()
let source_commit = refs.resolve(source_branch) catch {
_ =>
raise @bit.GitError::InvalidObject(
"Cannot resolve source branch: \{source_branch}",
)
}
guard source_commit is Some(src_id) else {
raise @bit.GitError::InvalidObject(
"Source branch not found: \{source_branch}",
)
}
let target_commit = refs.resolve(target_branch) catch {
_ =>
raise @bit.GitError::InvalidObject(
"Cannot resolve target branch: \{target_branch}",
)
}
guard target_commit is Some(tgt_id) else {
raise @bit.GitError::InvalidObject(
"Target branch not found: \{target_branch}",
)
}
let pr_id = generate_entity_id(
"proposal-pr",
author,
timestamp,
source_branch + "\n" + target_branch + "\n" + title + "\n" + body,
)
let pr = PullRequest::new(
pr_id,
title,
body,
source_branch,
src_id,
target_branch,
tgt_id,
author,
timestamp,
timestamp,
PrState::Open,
[],
closes_issues~,
source_repo~,
source_ref~,
)
ignore(
self.store.put_record(
objects,
refs,
clock,
pr_proposal_meta_key(pr_id),
"pr-proposal",
pr.serialize(),
author,
),
)
pr
}
///|
/// Broadcast a feature completion notification (notify-only record)
pub fn Hub::broadcast_feature(
self : Hub,
objects : &@lib.ObjectStore,
refs : &@lib.RefStore,
clock : &@lib.Clock,
branch_name : String,
tip_commit : String,
author : String,
summary? : String = "",
closes_issues? : Array[String] = [],
session_url? : String? = None,
) -> Unit raise @bit.GitError {
let timestamp = clock.now()
let node_id = self.store.node_id()
let key = "hub/broadcast/feature/" + branch_name + "/" + node_id
// Build payload as serialized key-value text
let sb = StringBuilder::new()
sb.write_string("branch ")
sb.write_string(branch_name)
sb.write_char('\n')
sb.write_string("commit ")
sb.write_string(tip_commit)
sb.write_char('\n')
sb.write_string("author ")
sb.write_string(author)
sb.write_char('\n')
if summary.length() > 0 {
sb.write_string("summary ")
sb.write_string(summary)
sb.write_char('\n')
}
if closes_issues.length() > 0 {
sb.write_string("closes ")
sb.write_string(closes_issues.join(","))
sb.write_char('\n')
}
match session_url {
Some(url) => {
sb.write_string("session_url ")
sb.write_string(url)
sb.write_char('\n')
}
None => ()
}
let payload = sb.to_string()
ignore(
self.store.put_record(
objects, refs, clock, key, "feature-broadcast", payload, author,
),
)
ignore(timestamp)
}
///|
/// List feature broadcast records
pub fn Hub::list_feature_broadcasts(
self : Hub,
objects : &@lib.ObjectStore,
) -> Array[HubRecord] {
self.store.list_records(objects, "hub/broadcast/feature/")
}
///|
/// Publish node heartbeat record for canonical node election.
/// Each node periodically publishes its metadata (commit count, last update time).
pub fn Hub::publish_node_heartbeat(
self : Hub,
objects : &@lib.ObjectStore,
refs : &@lib.RefStore,
clock : &@lib.Clock,
node_id : String,
commit_count? : Int = 0,
main_tip? : String = "",
) -> Unit raise @bit.GitError {
let timestamp = clock.now()
let key = "hub/node/heartbeat/" + node_id
let sb = StringBuilder::new()
sb.write_string("node_id ")
sb.write_string(node_id)
sb.write_char('\n')
sb.write_string("commit_count ")
sb.write_string(commit_count.to_string())
sb.write_char('\n')
sb.write_string("main_tip ")
sb.write_string(main_tip)
sb.write_char('\n')
sb.write_string("timestamp ")
sb.write_string(timestamp.to_string())
sb.write_char('\n')
let payload = sb.to_string()
ignore(
self.store.put_record(
objects, refs, clock, key, "node-heartbeat", payload, node_id,
),
)
}
///|
/// Node metadata extracted from heartbeat records
pub struct NodeHeartbeat {
node_id : String
commit_count : Int
main_tip : String
timestamp : Int64
clock : Map[String, Int64]
}
///|
/// List all node heartbeats and determine which node is canonical.
/// The canonical node is the one with the highest vector clock.
pub fn Hub::list_node_heartbeats(
self : Hub,
objects : &@lib.ObjectStore,
) -> Array[NodeHeartbeat] {
let records = self.store.list_records(objects, "hub/node/heartbeat/")
let result : Array[NodeHeartbeat] = []
for record in records {
if record.kind != "node-heartbeat" {
continue
}
let mut node_id = ""
let mut commit_count = 0
let mut main_tip = ""
let mut timestamp : Int64 = 0
for line_view in record.payload.split("\n") {
let line = line_view.to_owned()
match line.find(" ") {
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 {
"node_id" => node_id = v
"commit_count" =>
commit_count = @string.parse_int(v) catch { _ => 0 }
"main_tip" => main_tip = v
"timestamp" => timestamp = @string.parse_int64(v) catch { _ => 0L }
_ => ()
}
}
}
}
result.push({
node_id,
commit_count,
main_tip,
timestamp,
clock: record.clock,
})
}
result
}
///|
/// Determine the canonical node from heartbeat records.
/// Returns the node_id of the most active node (highest vector clock, then highest commit count).
pub fn Hub::find_canonical_node(
self : Hub,
objects : &@lib.ObjectStore,
) -> String? {
let heartbeats = self.list_node_heartbeats(objects)
if heartbeats.length() == 0 {
return None
}
let mut best = heartbeats[0]
for i in 1.. 0 {
best = candidate
} else if cmp == 0 {
// Tie-break: higher commit count, then higher timestamp
if candidate.commit_count > best.commit_count {
best = candidate
} else if candidate.commit_count == best.commit_count &&
candidate.timestamp > best.timestamp {
best = candidate
}
}
}
Some(best.node_id)
}
///|
pub fn Hub::list_pr_proposals(
self : Hub,
objects : &@lib.ObjectStore,
) -> Array[PullRequest] {
let result : Array[PullRequest] = []
let records = self.store.list_records(objects, pr_proposal_prefix())
for record in records {
if !record.key.has_suffix("/meta") || record.kind != "pr-proposal" {
continue
}
let pr = parse_legacy_pull_request(record.payload) catch { _ => continue }
result.push(pr)
}
result.sort_by(fn(a, b) {
if a.created_at < b.created_at {
-1
} else if a.created_at > b.created_at {
1
} else {
0
}
})
result
}