Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Support multiple merge requests per ticket

Browse Source
This commit is contained in:
Hare
2026-09-29 01:08:46 +09:00
parent ac7a6027bd
commit 590abf3142
26 changed files with 1539 additions and 723 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/merge-request/src/lib.rs
+377 -46
View File
@@ -97,6 +97,8 @@ pub struct ReviewRequestedEvent {
#[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))]
pub sequence: u64,
pub subject_ref: String,
#[serde(default)]
pub ticket_item_revision: String,
pub requested_by: WorkerIdentity,
pub reviewer: WorkerIdentity,
pub created_at: DateTime<Utc>,
@@ -108,6 +110,8 @@ pub struct ReviewEvent {
pub sequence: u64,
pub request_event_id: String,
pub subject_ref: String,
#[serde(default)]
pub ticket_item_revision: String,
pub decision: ReviewDecision,
pub body: String,
pub findings: Vec<ReviewFinding>,
@@ -259,7 +263,9 @@ pub struct OpenMergeRequest {
}
#[derive(Debug, Clone)]
pub struct RequestMergeRequestReview {
pub merge_request_id: String,
pub ticket_id: String,
pub ticket_item_revision: String,
pub subject_ref: String,
pub child_session_id: String,
pub capability_token: String,
@@ -282,11 +288,14 @@ pub struct RegisterReviewerChildSession {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewSubmissionAuthorization {
pub workspace_id: String,
pub merge_request_id: String,
pub ticket_id: String,
pub subject_ref: String,
}
#[derive(Debug, Clone)]
pub struct SubmitMergeRequestReview {
pub merge_request_id: String,
pub ticket_id: String,
pub current_subject_ref: String,
pub capability_token: String,
@@ -297,6 +306,7 @@ pub struct SubmitMergeRequestReview {
}
#[derive(Debug, Clone)]
pub struct RevokeMergeRequestReview {
pub merge_request_id: String,
pub ticket_id: String,
pub review_event_id: String,
pub reason: String,
@@ -306,6 +316,7 @@ pub struct RevokeMergeRequestReview {
#[derive(Debug, Clone)]
pub struct RepairSelectorFrom {
pub workspace_id: String,
pub merge_request_id: String,
pub ticket_id: String,
pub selector_from: String,
pub resolved_subject_ref: String,
@@ -315,6 +326,7 @@ pub struct RepairSelectorFrom {
}
#[derive(Debug, Clone)]
pub struct ReadinessCheck {
pub merge_request_id: String,
pub ticket_id: String,
pub current_subject_ref: Option<String>,
pub auth: MergeRequestAuth,
@@ -330,6 +342,7 @@ pub struct ReadinessReport {
}
#[derive(Debug, Clone)]
pub struct CompleteMergeRequest {
pub merge_request_id: String,
pub ticket_id: String,
pub operation_id: String,
pub approval_event_id: String,
@@ -341,6 +354,24 @@ pub struct CompleteMergeRequest {
pub auth: MergeRequestAuth,
pub now: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct CompleteTicket {
pub ticket_id: String,
pub operation_id: String,
pub item_revision: String,
pub merge_request_ids: Vec<String>,
pub auth: MergeRequestAuth,
pub now: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct TicketCompletionEvent {
pub operation_id: String,
pub ticket_id: String,
pub item_revision: String,
pub merge_request_ids: Vec<String>,
pub completed_by: WorkerIdentity,
pub created_at: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CurrentAssignment {
pub assignment_id: String,
@@ -421,13 +452,14 @@ impl MergeRequestStore {
"selectors must differ".into(),
));
}
let merge_request_id = i.merge_request_id.clone();
self.assigned(&i.auth, &i.ticket_id, &i.repository_id)?;
let mut c = self.lock()?;
let t = c.transaction()?;
let conflict:bool=t.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_ticket_relations rel JOIN merge_requests mr ON mr.workspace_id=rel.workspace_id AND mr.merge_request_id=rel.merge_request_id WHERE rel.workspace_id=?1 AND rel.ticket_id=?2 AND mr.state='open')",params![i.auth.workspace_id,i.ticket_id],|r|r.get(0))?;
let conflict:bool=t.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_ticket_relations rel JOIN merge_requests mr ON mr.workspace_id=rel.workspace_id AND mr.merge_request_id=rel.merge_request_id WHERE rel.workspace_id=?1 AND rel.ticket_id=?2 AND mr.repository_id=?3 AND mr.state='open')",params![i.auth.workspace_id,i.ticket_id,i.repository_id],|r|r.get(0))?;
if conflict {
return Err(MergeRequestError::Conflict(
"Ticket already has an open Merge Request; use ShowMergeRequest and advance the existing selector_from with a normal non-force push instead of opening a replacement Merge Request or adding a revision".into(),
"Ticket already has an open Merge Request for this repository; advance that Merge Request's existing selector_from with a normal non-force push instead of opening a replacement".into(),
));
}
let now = i.now.to_rfc3339();
@@ -469,7 +501,7 @@ impl MergeRequestStore {
}
t.commit()?;
drop(c);
self.get(&i.auth.workspace_id, &i.ticket_id)
self.get_by_id(&i.auth.workspace_id, &merge_request_id)
}
pub fn register_reviewer_child_session(
&self,
@@ -498,7 +530,12 @@ impl MergeRequestStore {
i: RequestMergeRequestReview,
) -> Result<RequestedMergeRequestReview, MergeRequestError> {
nonempty("subject_ref", &i.subject_ref)?;
let mr = self.get(&i.auth.workspace_id, &i.ticket_id)?;
let mr = self.get_for_operation(&i.auth.workspace_id, &i.merge_request_id, &i.ticket_id)?;
self.require_current_ticket_revision(
&i.auth.workspace_id,
&i.ticket_id,
&i.ticket_item_revision,
)?;
self.assigned(&i.auth, &i.ticket_id, &mr.repository_id)?;
if mr.state != MergeRequestState::Open || mr.selector_from.is_none() {
return Err(MergeRequestError::Conflict(
@@ -522,6 +559,7 @@ impl MergeRequestStore {
event_id: Uuid::now_v7().to_string(),
sequence: next_seq(&t, &mr.workspace_id, &mr.merge_request_id)?,
subject_ref: i.subject_ref,
ticket_item_revision: i.ticket_item_revision,
requested_by: WorkerIdentity {
runtime_id: i.auth.runtime_id,
worker_id: i.auth.worker_id,
@@ -548,25 +586,27 @@ impl MergeRequestStore {
}
pub fn authorize_review_submission(
&self,
ticket_id: &str,
merge_request_id: &str,
capability_token: &str,
) -> Result<ReviewSubmissionAuthorization, MergeRequestError> {
let connection = self.lock()?;
connection
.query_row(
"SELECT g.workspace_id,g.subject_ref
"SELECT g.workspace_id,g.merge_request_id,rel.ticket_id,g.subject_ref
FROM merge_request_review_grants g
JOIN merge_request_ticket_relations rel
ON rel.workspace_id=g.workspace_id AND rel.merge_request_id=g.merge_request_id
JOIN merge_requests mr
ON mr.workspace_id=g.workspace_id AND mr.merge_request_id=g.merge_request_id
WHERE g.capability_token=?1 AND rel.ticket_id=?2
WHERE g.capability_token=?1 AND g.merge_request_id=?2
AND g.status='issued' AND mr.state='open'",
params![capability_token, ticket_id],
params![capability_token, merge_request_id],
|row| {
Ok(ReviewSubmissionAuthorization {
workspace_id: row.get(0)?,
subject_ref: row.get(1)?,
merge_request_id: row.get(1)?,
ticket_id: row.get(2)?,
subject_ref: row.get(3)?,
})
},
)
@@ -594,8 +634,9 @@ impl MergeRequestStore {
JOIN merge_requests mr
ON mr.workspace_id=g.workspace_id AND mr.merge_request_id=g.merge_request_id
WHERE g.capability_token=?1 AND rel.ticket_id=?2
AND g.merge_request_id=?3
AND g.status='issued' AND mr.state='open'",
params![i.capability_token, i.ticket_id],
params![i.capability_token, i.ticket_id, i.merge_request_id],
|r| {
Ok((
r.get(0)?,
@@ -613,6 +654,16 @@ impl MergeRequestStore {
"review grant invalid".into(),
));
};
let ticket_item_revision = load_mr(&t, &ws, &mr)?
.and_then(|request| {
request.thread.into_iter().find_map(|event| match event {
MergeRequestThreadEvent::ReviewRequested(event) if event.event_id == req => {
Some(event.ticket_item_revision)
}
_ => None,
})
})
.ok_or_else(|| MergeRequestError::Corrupt("review request event missing".into()))?;
if subject != i.current_subject_ref {
let reason = format!(
"selector_from moved from requested subject {subject} to current subject {}; fresh review of the exact current source ref is required",
@@ -638,6 +689,7 @@ impl MergeRequestStore {
sequence: next_seq(&t, &ws, &mr)?,
request_event_id: req,
subject_ref: subject,
ticket_item_revision,
decision: i.decision,
body: i.body,
findings: i.findings,
@@ -657,7 +709,7 @@ impl MergeRequestStore {
i: RevokeMergeRequestReview,
) -> Result<ReviewRevokedEvent, MergeRequestError> {
bounded_body("revocation reason", &i.reason)?;
let mr = self.get(&i.auth.workspace_id, &i.ticket_id)?;
let mr = self.get_for_operation(&i.auth.workspace_id, &i.merge_request_id, &i.ticket_id)?;
self.assigned(&i.auth, &i.ticket_id, &mr.repository_id)?;
let r = mr
.thread
@@ -695,7 +747,7 @@ impl MergeRequestStore {
Ok(e)
}
pub fn readiness(&self, i: ReadinessCheck) -> Result<ReadinessReport, MergeRequestError> {
let mr = self.get(&i.auth.workspace_id, &i.ticket_id)?;
let mr = self.get_for_operation(&i.auth.workspace_id, &i.merge_request_id, &i.ticket_id)?;
let review = i
.current_subject_ref
.as_deref()
@@ -742,7 +794,7 @@ impl MergeRequestStore {
})
}
pub fn validate_completion(&self, i: &CompleteMergeRequest) -> Result<(), MergeRequestError> {
let mr = self.get(&i.auth.workspace_id, &i.ticket_id)?;
let mr = self.get_for_operation(&i.auth.workspace_id, &i.merge_request_id, &i.ticket_id)?;
self.repo(&i.auth, &mr.repository_id)?;
if let Some(existing) = mr.thread.iter().find_map(|event| match event {
MergeRequestThreadEvent::Merge(value) if value.operation_id == i.operation_id => {
@@ -808,7 +860,8 @@ impl MergeRequestStore {
self.validate_completion(&i)?;
let mut connection = self.lock()?;
let transaction = connection.transaction()?;
let mr = load_mr_for_ticket(&transaction, &i.auth.workspace_id, &i.ticket_id)?
let mr = load_mr(&transaction, &i.auth.workspace_id, &i.merge_request_id)?
.filter(|mr| mr.ticket_ids.iter().any(|id| id == &i.ticket_id))
.ok_or(MergeRequestError::NotFound)?;
if let Some(existing) = mr.thread.iter().find_map(|event| match event {
@@ -872,22 +925,6 @@ impl MergeRequestStore {
"Ticket must be inprogress".into(),
));
}
transaction.execute(
"UPDATE typed_tickets
SET workflow_state='done',workflow_state_explicit=1,updated_at=?3
WHERE workspace_id=?1 AND ticket_id=?2 AND workflow_state='inprogress'",
params![mr.workspace_id, i.ticket_id, i.now.to_rfc3339()],
)?;
let released_assignment = transaction.execute(
"DELETE FROM ticket_current_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3",
params![mr.workspace_id, i.ticket_id, i.auth.assignment_id],
)?;
if released_assignment != 1 {
return Err(MergeRequestError::Unauthorized(
"completion assignment changed while closing Ticket".into(),
));
}
let issued_grants = {
let mut statement = transaction.prepare(
"SELECT request_event_id,subject_ref,capability_token
@@ -965,13 +1002,195 @@ impl MergeRequestStore {
Ok(event)
}
pub fn complete_ticket(
&self,
mut input: CompleteTicket,
) -> Result<TicketCompletionEvent, MergeRequestError> {
nonempty("operation_id", &input.operation_id)?;
nonempty("item_revision", &input.item_revision)?;
if input.merge_request_ids.is_empty() {
return Err(MergeRequestError::NotReady(
"Ticket completion requires at least one linked Merge Request result".into(),
));
}
input.merge_request_ids.sort();
if input
.merge_request_ids
.windows(2)
.any(|ids| ids[0] == ids[1])
{
return Err(MergeRequestError::Validation(
"merge_request_ids must not contain duplicates".into(),
));
}
let mut connection = self.lock()?;
let transaction = connection.transaction()?;
let existing: Option<String> = transaction
.query_row(
"SELECT payload.value
FROM typed_ticket_event_attributes operation
JOIN typed_ticket_event_attributes payload
ON payload.workspace_id=operation.workspace_id
AND payload.ticket_id=operation.ticket_id
AND payload.event_index=operation.event_index
AND payload.key='ticket_completion'
WHERE operation.workspace_id=?1 AND operation.ticket_id=?2
AND operation.key='operation_id' AND operation.value=?3",
params![input.auth.workspace_id, input.ticket_id, input.operation_id],
|row| row.get(0),
)
.optional()?;
if let Some(payload) = existing {
let event: TicketCompletionEvent = json(&payload)?;
if event.item_revision == input.item_revision
&& event.merge_request_ids == input.merge_request_ids
&& event.completed_by == input.auth.actor()
{
return Ok(event);
}
return Err(MergeRequestError::Conflict(
"Ticket completion operation fingerprint mismatch".into(),
));
}
let current_revision =
current_ticket_revision(&transaction, &input.auth.workspace_id, &input.ticket_id)?
.ok_or(MergeRequestError::NotFound)?;
if current_revision != input.item_revision {
return Err(MergeRequestError::Conflict(format!(
"Ticket item revision changed from `{}` to `{current_revision}`; repeat requirement review before completion",
input.item_revision
)));
}
let state: String = transaction.query_row(
"SELECT workflow_state FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2",
params![input.auth.workspace_id, input.ticket_id],
|row| row.get(0),
)?;
if state != "inprogress" {
return Err(MergeRequestError::Conflict(
"Ticket must be inprogress".into(),
));
}
let assignment_is_current: bool = transaction.query_row(
"SELECT EXISTS(
SELECT 1 FROM ticket_current_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3
)",
params![
input.auth.workspace_id,
input.ticket_id,
input.auth.assignment_id
],
|row| row.get(0),
)?;
if !assignment_is_current {
return Err(MergeRequestError::Unauthorized(
"Ticket completion assignment changed before commit".into(),
));
}
let actual_ids = {
let mut statement = transaction.prepare(
"SELECT merge_request_id FROM merge_request_ticket_relations
WHERE workspace_id=?1 AND ticket_id=?2 AND relation_kind='implements'
ORDER BY merge_request_id",
)?;
statement
.query_map(params![input.auth.workspace_id, input.ticket_id], |row| {
row.get::<_, String>(0)
})?
.collect::<Result<Vec<_>, _>>()?
};
if actual_ids != input.merge_request_ids {
return Err(MergeRequestError::Conflict(
"linked Merge Request set changed; refresh Ticket completion evidence".into(),
));
}
for merge_request_id in &actual_ids {
let request = load_mr(&transaction, &input.auth.workspace_id, merge_request_id)?
.ok_or(MergeRequestError::NotFound)?;
if request.state != MergeRequestState::Merged {
return Err(MergeRequestError::NotReady(format!(
"Merge Request `{merge_request_id}` has no merged result"
)));
}
let merge = request
.thread
.iter()
.rev()
.find_map(|event| match event {
MergeRequestThreadEvent::Merge(event) => Some(event),
_ => None,
})
.ok_or_else(|| {
MergeRequestError::Corrupt(format!(
"merged Merge Request `{merge_request_id}` has no MergeResult"
))
})?;
let review = request
.effective_review(&merge.approved_source_ref)
.filter(|review| review.event_id == merge.approval_event_id)
.ok_or_else(|| {
MergeRequestError::NotReady(format!(
"Merge Request `{merge_request_id}` approval is no longer effective"
))
})?;
if review.decision != ReviewDecision::Approve
|| review.ticket_item_revision != input.item_revision
{
return Err(MergeRequestError::NotReady(format!(
"Merge Request `{merge_request_id}` lacks approval bound to the current Ticket revision"
)));
}
}
let updated = transaction.execute(
"UPDATE typed_tickets
SET workflow_state='done',workflow_state_explicit=1,updated_at=?3
WHERE workspace_id=?1 AND ticket_id=?2 AND workflow_state='inprogress'",
params![
input.auth.workspace_id,
input.ticket_id,
input.now.to_rfc3339()
],
)?;
if updated != 1 {
return Err(MergeRequestError::Conflict(
"Ticket state changed during completion".into(),
));
}
let released = transaction.execute(
"DELETE FROM ticket_current_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3",
params![
input.auth.workspace_id,
input.ticket_id,
input.auth.assignment_id
],
)?;
if released != 1 {
return Err(MergeRequestError::Unauthorized(
"Ticket completion assignment changed while releasing it".into(),
));
}
let event = TicketCompletionEvent {
operation_id: input.operation_id,
ticket_id: input.ticket_id.clone(),
item_revision: input.item_revision,
merge_request_ids: input.merge_request_ids,
completed_by: input.auth.actor(),
created_at: input.now,
};
ticket_completion_event(&transaction, &input.auth.workspace_id, &event)?;
transaction.commit()?;
Ok(event)
}
pub fn repair_selector_from(
&self,
i: RepairSelectorFrom,
) -> Result<MergeRequest, MergeRequestError> {
nonempty("selector_from", &i.selector_from)?;
nonempty("resolved_subject_ref", &i.resolved_subject_ref)?;
let mr = self.get(&i.workspace_id, &i.ticket_id)?;
let mr = self.get_for_operation(&i.workspace_id, &i.merge_request_id, &i.ticket_id)?;
if mr.selector_from.is_some() {
return Err(MergeRequestError::Conflict(
"selector_from is immutable after it is set".into(),
@@ -1011,7 +1230,7 @@ impl MergeRequestStore {
)?;
t.commit()?;
drop(c);
self.get(&i.workspace_id, &i.ticket_id)
self.get_by_id(&i.workspace_id, &i.merge_request_id)
}
pub fn get(&self, ws: &str, ticket: &str) -> Result<MergeRequest, MergeRequestError> {
let c = self.lock()?;
@@ -1021,6 +1240,21 @@ impl MergeRequestStore {
None => Err(MergeRequestError::NotFound),
}
}
fn get_for_operation(
&self,
workspace_id: &str,
merge_request_id: &str,
ticket_id: &str,
) -> Result<MergeRequest, MergeRequestError> {
let mr = self.get_by_id(workspace_id, merge_request_id)?;
if !mr.ticket_ids.iter().any(|id| id == ticket_id) {
return Err(MergeRequestError::Unauthorized(
"Merge Request is not linked to the addressed Ticket".into(),
));
}
Ok(mr)
}
pub fn get_by_id(
&self,
workspace_id: &str,
@@ -1029,6 +1263,23 @@ impl MergeRequestStore {
let c = self.lock()?;
load_mr(&c, workspace_id, merge_request_id)?.ok_or(MergeRequestError::NotFound)
}
pub fn list_for_ticket(
&self,
workspace_id: &str,
ticket_id: &str,
) -> Result<Vec<MergeRequest>, MergeRequestError> {
Ok(self
.list(
workspace_id,
&MergeRequestListQuery {
ticket_id: Some(ticket_id.to_string()),
limit: 100,
..Default::default()
},
)?
.items)
}
pub fn list(
&self,
workspace_id: &str,
@@ -1125,6 +1376,24 @@ impl MergeRequestStore {
limit.clamp(1, 200),
)
}
fn require_current_ticket_revision(
&self,
workspace_id: &str,
ticket_id: &str,
expected: &str,
) -> Result<(), MergeRequestError> {
nonempty("ticket_item_revision", expected)?;
let connection = self.lock()?;
let current = current_ticket_revision(&connection, workspace_id, ticket_id)?
.ok_or(MergeRequestError::NotFound)?;
if current != expected {
return Err(MergeRequestError::Conflict(format!(
"Ticket item revision changed from `{expected}` to `{current}`; review the current requirements"
)));
}
Ok(())
}
fn assigned(&self, a: &MergeRequestAuth, t: &str, r: &str) -> Result<(), MergeRequestError> {
self.repo(a, r)?;
let x = self
@@ -1248,29 +1517,39 @@ fn insert_event<T: Serialize>(
)?;
Ok(())
}
fn load_mr_for_ticket(
fn current_ticket_revision(
connection: &Connection,
workspace_id: &str,
ticket_id: &str,
) -> Result<Option<MergeRequest>, MergeRequestError> {
let merge_request_id: Option<String> = connection
) -> Result<Option<String>, MergeRequestError> {
let event_revision = connection
.query_row(
"SELECT rel.merge_request_id
FROM merge_request_ticket_relations rel
JOIN merge_requests mr
ON mr.workspace_id=rel.workspace_id
AND mr.merge_request_id=rel.merge_request_id
WHERE rel.workspace_id=?1 AND rel.ticket_id=?2
ORDER BY CASE mr.state WHEN 'open' THEN 0 ELSE 1 END,mr.created_at DESC
"SELECT attribute.value
FROM typed_ticket_events event
JOIN typed_ticket_event_attributes attribute
ON attribute.workspace_id=event.workspace_id
AND attribute.ticket_id=event.ticket_id
AND attribute.event_index=event.event_index
AND attribute.key='event_id'
WHERE event.workspace_id=?1 AND event.ticket_id=?2
AND event.kind IN ('create','item_edit')
ORDER BY event.event_index DESC
LIMIT 1",
params![workspace_id, ticket_id],
|row| row.get(0),
)
.optional()?;
match merge_request_id {
Some(id) => load_mr(connection, workspace_id, &id),
None => Ok(None),
if event_revision.is_some() {
return Ok(event_revision);
}
connection
.query_row(
"SELECT updated_at FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2",
params![workspace_id, ticket_id],
|row| row.get(0),
)
.optional()
.map_err(Into::into)
}
fn load_mr(c: &Connection, w: &str, m: &str) -> Result<Option<MergeRequest>, MergeRequestError> {
@@ -1341,9 +1620,10 @@ fn ticket_event(
) -> Result<(), MergeRequestError> {
let ticket = &mr.ticket_ids[0];
let n:i64=t.query_row("SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2",params![mr.workspace_id,ticket],|r|r.get(0))?;
t.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body)VALUES(?1,?2,?3,'state_changed',?4,?5,'inprogress','done','Merge Request completed',?6)",params![mr.workspace_id,ticket,n,format!("worker:{}:{}",e.merged_by.runtime_id,e.merged_by.worker_id),e.created_at.to_rfc3339(),format!("Approved source ref `{}` completed implementation.",e.approved_source_ref)])?;
t.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body)VALUES(?1,?2,?3,'comment',?4,?5,NULL,NULL,'Merge Request integrated',?6)",params![mr.workspace_id,ticket,n,format!("worker:{}:{}",e.merged_by.runtime_id,e.merged_by.worker_id),e.created_at.to_rfc3339(),format!("Merge Request `{}` integrated approved source ref `{}`; Ticket remains in progress until guarded Ticket completion.",mr.merge_request_id,e.approved_source_ref)])?;
for (k, v) in [
("implementation_assignment_id", a),
("merge_request_id", &mr.merge_request_id),
("approval_event_id", &e.approval_event_id),
("approved_source_ref", &e.approved_source_ref),
("operation_id", &e.operation_id),
@@ -1356,6 +1636,57 @@ fn ticket_event(
Ok(())
}
fn ticket_completion_event(
transaction: &Transaction<'_>,
workspace_id: &str,
event: &TicketCompletionEvent,
) -> Result<(), MergeRequestError> {
let event_index: i64 = transaction.query_row(
"SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2",
params![workspace_id, event.ticket_id],
|row| row.get(0),
)?;
transaction.execute(
"INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body)
VALUES(?1,?2,?3,'state_changed',?4,?5,'inprogress','done','Ticket requirements completed',?6)",
params![
workspace_id,
event.ticket_id,
event_index,
format!(
"worker:{}:{}",
event.completed_by.runtime_id, event.completed_by.worker_id
),
event.created_at.to_rfc3339(),
format!(
"Current Ticket revision `{}` was approved and completed with Merge Requests: {}.",
event.item_revision,
event.merge_request_ids.join(", ")
)
],
)?;
for (key, value) in [
("operation_id", event.operation_id.clone()),
("item_revision", event.item_revision.clone()),
(
"merge_request_ids",
serde_json::to_string(&event.merge_request_ids)
.map_err(|error| MergeRequestError::Operation(error.to_string()))?,
),
(
"ticket_completion",
serde_json::to_string(event)
.map_err(|error| MergeRequestError::Operation(error.to_string()))?,
),
] {
transaction.execute(
"INSERT INTO typed_ticket_event_attributes VALUES(?1,?2,?3,?4,?5)",
params![workspace_id, event.ticket_id, event_index, key, value],
)?;
}
Ok(())
}
pub fn migrate(c: &Connection) -> Result<(), MergeRequestError> {
match schema_state(c)? {
SchemaState::Fresh => fresh(c),
crates/merge-request/tests/store.rs
+256 -14
View File
@@ -12,7 +12,7 @@ impl AssignmentSource for Assignments {
struct Repositories;
impl RepositorySource for Repositories {
fn repository_belongs_to_workspace(&self, w: &str, r: &str) -> Result<bool, String> {
Ok(w == "W" && r == "R")
Ok(w == "W" && matches!(r, "R" | "R2"))
}
}
fn at(s: u32) -> chrono::DateTime<Utc> {
@@ -29,11 +29,17 @@ fn auth() -> MergeRequestAuth {
assignment_id: "A".into(),
}
}
fn auth_repo(repository_id: &str) -> MergeRequestAuth {
MergeRequestAuth {
repository_id: repository_id.into(),
..auth()
}
}
fn fixture() -> (tempfile::TempDir, MergeRequestStore) {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("db");
let c = Connection::open(&p).unwrap();
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t');").unwrap();
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t');").unwrap();
drop(c);
let a = Assignments(Arc::new(Mutex::new(CurrentAssignment {
assignment_id: "A".into(),
@@ -44,20 +50,33 @@ fn fixture() -> (tempfile::TempDir, MergeRequestStore) {
let s = MergeRequestStore::open(&p, Arc::new(a), Arc::new(Repositories)).unwrap();
(d, s)
}
fn open(s: &MergeRequestStore) {
fn open_for(s: &MergeRequestStore, merge_request_id: &str, repository_id: &str) {
s.open_merge_request(OpenMergeRequest {
merge_request_id: "MR".into(),
merge_request_id: merge_request_id.into(),
ticket_id: "T".into(),
repository_id: "R".into(),
selector_from: "work/t".into(),
repository_id: repository_id.into(),
selector_from: if merge_request_id == "MR" {
"work/t".into()
} else {
format!("work/{merge_request_id}")
},
selector_to: "develop".into(),
summary: "summary".into(),
auth: auth(),
auth: auth_repo(repository_id),
now: at(1),
})
.unwrap();
}
fn request(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewRequestedEvent {
fn open(s: &MergeRequestStore) {
open_for(s, "MR", "R");
}
fn request_for(
s: &MergeRequestStore,
merge_request_id: &str,
repository_id: &str,
subject: &str,
token: &str,
) -> ReviewRequestedEvent {
s.register_reviewer_child_session(RegisterReviewerChildSession {
workspace_id: "W".into(),
parent_runtime_id: "runtime".into(),
@@ -68,19 +87,31 @@ fn request(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewRequested
})
.unwrap();
s.request_review(RequestMergeRequestReview {
merge_request_id: merge_request_id.into(),
ticket_item_revision: "t".into(),
ticket_id: "T".into(),
subject_ref: subject.into(),
child_session_id: format!("child-{token}"),
capability_token: token.into(),
auth: auth(),
auth: auth_repo(repository_id),
now: at(3),
})
.unwrap()
.request_event
}
fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
request(s, subject, token);
fn request(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewRequestedEvent {
request_for(s, "MR", "R", subject, token)
}
fn approve_for(
s: &MergeRequestStore,
merge_request_id: &str,
repository_id: &str,
subject: &str,
token: &str,
) -> ReviewEvent {
request_for(s, merge_request_id, repository_id, subject, token);
s.submit_review(SubmitMergeRequestReview {
merge_request_id: merge_request_id.into(),
ticket_id: "T".into(),
current_subject_ref: subject.into(),
capability_token: token.into(),
@@ -91,6 +122,9 @@ fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
})
.unwrap()
}
fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
approve_for(s, "MR", "R", subject, token)
}
#[test]
fn review_submission_authorization_rejects_invalid_grants_before_side_effects() {
let (_d, store) = fixture();
@@ -98,13 +132,15 @@ fn review_submission_authorization_rejects_invalid_grants_before_side_effects()
request(&store, "published-source", "valid-token");
let invalid = store
.authorize_review_submission("T", "invalid-token")
.authorize_review_submission("MR", "invalid-token")
.unwrap_err();
assert!(matches!(invalid, MergeRequestError::Unauthorized(_)));
let authorized = store
.authorize_review_submission("T", "valid-token")
.authorize_review_submission("MR", "valid-token")
.unwrap();
assert_eq!(authorized.workspace_id, "W");
assert_eq!(authorized.merge_request_id, "MR");
assert_eq!(authorized.ticket_id, "T");
assert_eq!(authorized.subject_ref, "published-source");
}
@@ -115,6 +151,7 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
let review = approve(&s, "opaque-source-ref", "token");
let ready = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("opaque-source-ref".into()),
auth: auth(),
@@ -123,6 +160,7 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
assert!(ready.ready);
let merged = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: review.event_id,
@@ -150,9 +188,10 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
|row| row.get(0),
)
.unwrap();
assert!(!current_assignment);
assert!(current_assignment);
let replayed = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: merged.approval_event_id.clone(),
@@ -166,6 +205,41 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
})
.unwrap();
assert_eq!(replayed, merged);
let completed = s
.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-op".into(),
item_revision: "t".into(),
merge_request_ids: vec!["MR".into()],
auth: auth(),
now: at(7),
})
.unwrap();
assert_eq!(completed.merge_request_ids, vec!["MR"]);
let connection = Connection::open(d.path().join("db")).unwrap();
let (state, assignment_exists): (String, bool) = connection
.query_row(
"SELECT workflow_state, EXISTS(
SELECT 1 FROM ticket_current_worker_assignments
WHERE workspace_id='W' AND ticket_id='T'
) FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert_eq!(state, "done");
assert!(!assignment_exists);
let replayed_completion = s
.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-op".into(),
item_revision: "t".into(),
merge_request_ids: vec!["MR".into()],
auth: auth(),
now: at(8),
})
.unwrap();
assert_eq!(replayed_completion, completed);
let json = serde_json::to_string(&mr).unwrap();
for banned in [
"revision_id",
@@ -187,6 +261,7 @@ fn source_move_cancels_submission_and_old_approval_is_reusable_when_source_retur
request(&s, "source-b", "two");
assert!(
s.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "source-c".into(),
capability_token: "two".into(),
@@ -221,6 +296,7 @@ fn same_selector_source_advancement_requires_fresh_review_and_preserves_target_o
let stale = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source-2".into()),
auth: auth(),
@@ -244,6 +320,7 @@ fn same_selector_source_advancement_requires_fresh_review_and_preserves_target_o
let second = approve(&s, "source-2", "two");
let ready = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source-2".into()),
auth: auth(),
@@ -260,6 +337,7 @@ fn same_selector_source_advancement_requires_fresh_review_and_preserves_target_o
// integration evidence for the current target pair.
let merged = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
operation_id: "target-moved".into(),
ticket_id: "T".into(),
current_subject_ref: "source-2".into(),
@@ -283,6 +361,7 @@ fn review_revocation_invalidates_readiness() {
open(&s);
let review = approve(&s, "source", "one");
s.revoke_review(RevokeMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
review_event_id: review.event_id,
reason: "bad evidence".into(),
@@ -292,6 +371,7 @@ fn review_revocation_invalidates_readiness() {
.unwrap();
let r = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source".into()),
auth: auth(),
@@ -366,6 +446,7 @@ fn completion_rejects_superseded_approval_for_same_subject() {
request(&store, "subject", "changes");
store
.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "subject".into(),
capability_token: "changes".into(),
@@ -376,6 +457,7 @@ fn completion_rejects_superseded_approval_for_same_subject() {
})
.unwrap();
let result = store.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: old_approval.event_id,
@@ -398,6 +480,7 @@ fn completion_cancels_outstanding_grants_and_late_submit_fails() {
request(&store, "other-subject", "pending");
store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: approval.event_id,
@@ -411,6 +494,7 @@ fn completion_cancels_outstanding_grants_and_late_submit_fails() {
})
.unwrap();
let late = store.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "other-subject".into(),
capability_token: "pending".into(),
@@ -436,6 +520,7 @@ fn selector_repair_requires_and_accepts_an_approved_resolved_subject() {
.unwrap();
let repaired = store
.repair_selector_from(RepairSelectorFrom {
merge_request_id: "MR".into(),
workspace_id: "W".into(),
ticket_id: "T".into(),
selector_from: "restored-work".into(),
@@ -460,6 +545,7 @@ fn selector_repair_rejects_unapproved_resolved_subject() {
.execute("UPDATE merge_requests SET selector_from=NULL WHERE workspace_id='W' AND merge_request_id='MR'", [])
.unwrap();
let result = store.repair_selector_from(RepairSelectorFrom {
merge_request_id: "MR".into(),
workspace_id: "W".into(),
ticket_id: "T".into(),
selector_from: "wrong-work".into(),
@@ -552,6 +638,7 @@ fn transactional_completion_rejects_assignment_changed_in_control_plane_db() {
.execute("UPDATE ticket_current_worker_assignments SET assignment_id='B' WHERE workspace_id='W' AND ticket_id='T'", [])
.unwrap();
let result = store.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: approval.event_id,
@@ -574,3 +661,158 @@ fn transactional_completion_rejects_assignment_changed_in_control_plane_db() {
.unwrap();
assert_eq!(state, "inprogress");
}
#[test]
fn ticket_can_link_parallel_open_merge_requests_in_distinct_repositories() {
let (_dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
let linked = store.list_for_ticket("W", "T").unwrap();
assert_eq!(
linked
.iter()
.map(|request| request.merge_request_id.as_str())
.collect::<Vec<_>>(),
vec!["MR", "MR-2"]
);
let duplicate = store.open_merge_request(OpenMergeRequest {
merge_request_id: "MR-duplicate".into(),
ticket_id: "T".into(),
repository_id: "R2".into(),
selector_from: "work/duplicate".into(),
selector_to: "develop".into(),
summary: String::new(),
auth: auth_repo("R2"),
now: at(2),
});
assert!(matches!(duplicate, Err(MergeRequestError::Conflict(_))));
}
#[test]
fn mr_operations_and_review_capabilities_are_bound_to_explicit_identity() {
let (_dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
request_for(&store, "MR", "R", "subject-one", "token-one");
assert!(matches!(
store.authorize_review_submission("MR-2", "token-one"),
Err(MergeRequestError::Unauthorized(_))
));
assert!(matches!(
store.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR-2".into(),
ticket_id: "T".into(),
current_subject_ref: "subject-one".into(),
capability_token: "token-one".into(),
decision: ReviewDecision::Approve,
body: "wrong MR".into(),
findings: vec![],
now: at(4),
}),
Err(MergeRequestError::Unauthorized(_))
));
assert!(
!store
.get_by_id("W", "MR-2")
.unwrap()
.thread
.iter()
.any(|event| matches!(event, MergeRequestThreadEvent::Review(_)))
);
}
#[test]
fn partial_integration_retains_ticket_and_assignment_until_guarded_ticket_completion() {
let (dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
let first = approve_for(&store, "MR", "R", "subject-one", "token-one");
let second = approve_for(&store, "MR-2", "R2", "subject-two", "token-two");
store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge-one".into(),
approval_event_id: first.event_id,
current_subject_ref: "subject-one".into(),
target_ref_before: "target-one-before".into(),
target_ref_after: "target-one-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth_repo("R"),
now: at(5),
})
.unwrap();
let connection = Connection::open(dir.path().join("db")).unwrap();
let (state, assigned): (String, bool) = connection
.query_row(
"SELECT workflow_state, EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE workspace_id='W' AND ticket_id='T') FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert_eq!(state, "inprogress");
assert!(assigned);
assert!(matches!(
store.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-complete".into(),
item_revision: "t".into(),
merge_request_ids: vec!["MR".into(), "MR-2".into()],
auth: auth(),
now: at(6),
}),
Err(MergeRequestError::NotReady(_))
));
store
.complete(CompleteMergeRequest {
merge_request_id: "MR-2".into(),
ticket_id: "T".into(),
operation_id: "merge-two".into(),
approval_event_id: second.event_id,
current_subject_ref: "subject-two".into(),
target_ref_before: "target-two-before".into(),
target_ref_after: "target-two-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth_repo("R2"),
now: at(7),
})
.unwrap();
assert!(matches!(
store.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-complete".into(),
item_revision: "stale".into(),
merge_request_ids: vec!["MR".into(), "MR-2".into()],
auth: auth(),
now: at(8),
}),
Err(MergeRequestError::Conflict(_))
));
assert!(matches!(
store.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-complete".into(),
item_revision: "t".into(),
merge_request_ids: vec!["MR".into()],
auth: auth(),
now: at(8),
}),
Err(MergeRequestError::Conflict(_))
));
store
.complete_ticket(CompleteTicket {
ticket_id: "T".into(),
operation_id: "ticket-complete".into(),
item_revision: "t".into(),
merge_request_ids: vec!["MR-2".into(), "MR".into()],
auth: auth(),
now: at(9),
})
.unwrap();
}