feat: expose merge requests as workspace resources

crates/merge-request/src/lib.rs

@@ -27,6 +27,14 @@ pub enum MergeRequestState {
     Closed,
 }
 impl MergeRequestState {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Open => "open",
+            Self::Merged => "merged",
+            Self::Closed => "closed",
+        }
+    }
+
     fn parse(v: &str) -> Result<Self, MergeRequestError> {
         match v {
             "draft" | "open" => Ok(Self::Open),
@@ -216,6 +224,23 @@ impl MergeRequest {
     }
 }
 
+#[derive(Clone, Debug, Default)]
+pub struct MergeRequestListQuery {
+    pub state: Option<MergeRequestState>,
+    pub repository_id: Option<String>,
+    pub ticket_id: Option<String>,
+    pub selector_from: Option<String>,
+    pub selector_to: Option<String>,
+    pub cursor: Option<String>,
+    pub limit: usize,
+}
+
+#[derive(Clone, Debug)]
+pub struct MergeRequestListPage {
+    pub items: Vec<MergeRequest>,
+    pub next_cursor: Option<String>,
+}
+
 #[derive(Debug, Clone)]
 pub struct OpenMergeRequest {
     pub merge_request_id: String,
@@ -935,6 +960,80 @@ impl MergeRequestStore {
             None => Err(MergeRequestError::NotFound),
         }
     }
+    pub fn get_by_id(
+        &self,
+        workspace_id: &str,
+        merge_request_id: &str,
+    ) -> Result<MergeRequest, MergeRequestError> {
+        let c = self.lock()?;
+        load_mr(&c, workspace_id, merge_request_id)?.ok_or(MergeRequestError::NotFound)
+    }
+    pub fn list(
+        &self,
+        workspace_id: &str,
+        query: &MergeRequestListQuery,
+    ) -> Result<MergeRequestListPage, MergeRequestError> {
+        let c = self.lock()?;
+        let limit = query.limit.clamp(1, 100);
+        let cursor_position = match query.cursor.as_deref() {
+            Some(cursor) => Some(
+                c.query_row(
+                    "SELECT updated_at,merge_request_id FROM merge_requests WHERE workspace_id=?1 AND merge_request_id=?2",
+                    params![workspace_id, cursor],
+                    |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
+                )
+                .optional()?
+                .ok_or_else(|| MergeRequestError::Validation("invalid merge request cursor".into()))?,
+            ),
+            None => None,
+        };
+        let cursor_updated_at = cursor_position
+            .as_ref()
+            .map(|(updated_at, _)| updated_at.as_str());
+        let cursor_id = cursor_position.as_ref().map(|(_, id)| id.as_str());
+        let state = query.state.map(MergeRequestState::as_str);
+        let mut statement = c.prepare(
+            "SELECT mr.merge_request_id
+             FROM merge_requests mr
+             WHERE mr.workspace_id=?1
+               AND (?2 IS NULL OR mr.state=?2)
+               AND (?3 IS NULL OR mr.repository_id=?3)
+               AND (?4 IS NULL OR mr.selector_from=?4)
+               AND (?5 IS NULL OR mr.selector_to=?5)
+               AND (?6 IS NULL OR EXISTS (
+                    SELECT 1 FROM merge_request_ticket_relations relation
+                    WHERE relation.workspace_id=mr.workspace_id
+                      AND relation.merge_request_id=mr.merge_request_id
+                      AND relation.ticket_id=?6
+               ))
+               AND (?7 IS NULL OR mr.updated_at<?7 OR (mr.updated_at=?7 AND mr.merge_request_id>?8))
+             ORDER BY mr.updated_at DESC,mr.merge_request_id ASC
+             LIMIT ?9",
+        )?;
+        let rows = statement.query_map(
+            params![
+                workspace_id,
+                state,
+                query.repository_id.as_deref(),
+                query.selector_from.as_deref(),
+                query.selector_to.as_deref(),
+                query.ticket_id.as_deref(),
+                cursor_updated_at,
+                cursor_id,
+                (limit + 1) as i64,
+            ],
+            |row| row.get::<_, String>(0),
+        )?;
+        let mut ids = rows.collect::<Result<Vec<_>, _>>()?;
+        let has_more = ids.len() > limit;
+        ids.truncate(limit);
+        let next_cursor = has_more.then(|| ids.last().cloned()).flatten();
+        let items = ids
+            .iter()
+            .map(|id| load_mr(&c, workspace_id, id)?.ok_or(MergeRequestError::NotFound))
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(MergeRequestListPage { items, next_cursor })
+    }
     pub fn thread_page(
         &self,
         ws: &str,
@@ -946,6 +1045,25 @@ impl MergeRequestStore {
         let c = self.lock()?;
         load_thread(&c, ws, &mr.merge_request_id, after, limit.clamp(1, 200))
     }
+    pub fn thread_page_by_id(
+        &self,
+        workspace_id: &str,
+        merge_request_id: &str,
+        after: Option<u64>,
+        limit: usize,
+    ) -> Result<Vec<MergeRequestThreadEvent>, MergeRequestError> {
+        let c = self.lock()?;
+        if load_mr(&c, workspace_id, merge_request_id)?.is_none() {
+            return Err(MergeRequestError::NotFound);
+        }
+        load_thread(
+            &c,
+            workspace_id,
+            merge_request_id,
+            after,
+            limit.clamp(1, 200),
+        )
+    }
     fn assigned(&self, a: &MergeRequestAuth, t: &str, r: &str) -> Result<(), MergeRequestError> {
         self.repo(a, r)?;
         let x = self

crates/merge-request/tests/store.rs

@@ -436,6 +436,75 @@ fn selector_repair_rejects_unapproved_resolved_subject() {
     assert!(matches!(result, Err(MergeRequestError::NotReady(_))));
 }
 
+#[test]
+fn first_class_list_and_detail_are_workspace_scoped_and_cursor_bounded() {
+    let (dir, store) = fixture();
+    open(&store);
+    Connection::open(dir.path().join("db"))
+        .unwrap()
+        .execute(
+            "UPDATE merge_requests SET state='closed' WHERE workspace_id='W' AND merge_request_id='MR'",
+            [],
+        )
+        .unwrap();
+    store
+        .open_merge_request(OpenMergeRequest {
+            merge_request_id: "MR-2".into(),
+            ticket_id: "T".into(),
+            repository_id: "R".into(),
+            selector_from: "work/t-2".into(),
+            selector_to: "develop".into(),
+            summary: "second".into(),
+            auth: auth(),
+            now: at(8),
+        })
+        .unwrap();
+
+    let first = store
+        .list(
+            "W",
+            &MergeRequestListQuery {
+                ticket_id: Some("T".into()),
+                limit: 1,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    assert_eq!(first.items[0].merge_request_id, "MR-2");
+    assert_eq!(first.next_cursor.as_deref(), Some("MR-2"));
+
+    let second = store
+        .list(
+            "W",
+            &MergeRequestListQuery {
+                ticket_id: Some("T".into()),
+                cursor: first.next_cursor,
+                limit: 1,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    assert_eq!(second.items[0].merge_request_id, "MR");
+    assert!(second.next_cursor.is_none());
+
+    let closed = store
+        .list(
+            "W",
+            &MergeRequestListQuery {
+                state: Some(MergeRequestState::Closed),
+                limit: 10,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    assert_eq!(closed.items.len(), 1);
+    assert_eq!(store.get_by_id("W", "MR").unwrap().merge_request_id, "MR");
+    assert!(matches!(
+        store.get_by_id("other", "MR"),
+        Err(MergeRequestError::NotFound)
+    ));
+}
+
 #[test]
 fn transactional_completion_rejects_assignment_changed_in_control_plane_db() {
     let (dir, store) = fixture();

crates/workspace-server/src/authority.rs

@@ -1589,7 +1589,7 @@ fn ticket_evidence_event(sequence: usize, event: &TicketEvent) -> TicketEvidence
     }
 }
 
-fn merge_request_summary(
+pub(crate) fn merge_request_summary(
     request: MergeRequest,
     current_subject_ref: Option<String>,
 ) -> TicketMergeRequestSummary {

crates/workspace-server/src/records.rs

@@ -285,6 +285,21 @@ pub struct TicketMergeRequestSummary {
     pub review_excerpt: Option<String>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub struct MergeRequestListItem {
+    pub summary: TicketMergeRequestSummary,
+    pub ticket_ids: Vec<String>,
+    pub thread_event_count: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub struct MergeRequestListResponse {
+    pub items: Vec<MergeRequestListItem>,
+    pub next_cursor: Option<String>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 pub struct TicketEvidenceSummary {
@@ -474,6 +489,8 @@ pub fn ticket_api_typescript() -> String {
         TicketEvidenceEvent::decl(&config),
         TicketAssignmentSummary::decl(&config),
         TicketMergeRequestSummary::decl(&config),
+        MergeRequestListItem::decl(&config),
+        MergeRequestListResponse::decl(&config),
         TicketEvidenceSummary::decl(&config),
         TicketQueryRequest::decl(&config),
         TicketQueryItem::decl(&config),

crates/workspace-server/src/server.rs

@@ -65,7 +65,7 @@ use crate::auth::{
 };
 use crate::authority::{
     MemoryAuthority, ObjectiveAuthority, ObjectiveCreateInput, ObjectiveEditInput,
-    SqliteWorkspaceAuthority, TicketAuthority, TicketMergeRevisionSource,
+    SqliteWorkspaceAuthority, TicketAuthority, TicketMergeRevisionSource, merge_request_summary,
 };
 use crate::companion::{
     CompanionCancelRequest, CompanionConsole, CompanionMessageRequest, CompanionMessageResponse,
@@ -97,8 +97,9 @@ use crate::observation::{
 };
 use crate::profile_settings::UpdateWorkspaceMetadataRequest;
 use crate::records::{
-    ObjectiveDetail, ObjectiveQueryRequest, ObjectiveQueryResponse, ObjectiveShowRequest,
-    ProjectRecordList, TicketDetail, TicketQueryRequest, TicketQueryResponse, TicketShowRequest,
+    MergeRequestListItem, MergeRequestListResponse, ObjectiveDetail, ObjectiveQueryRequest,
+    ObjectiveQueryResponse, ObjectiveShowRequest, ProjectRecordList, TicketDetail,
+    TicketQueryRequest, TicketQueryResponse, TicketShowRequest,
 };
 use crate::repositories::{
     ConfiguredRepository, RepositoryListProjection, RepositoryLogRead, RepositoryLookupError,
@@ -1729,9 +1730,17 @@ pub fn build_router(api: WorkspaceApi) -> Router {
             "/api/w/{workspace_id}/tickets/{id}/workflow/queue",
             post(scoped_queue_ticket_record),
         )
+        .route(
+            "/api/w/{workspace_id}/merge-requests",
+            get(scoped_list_merge_requests),
+        )
+        .route(
+            "/api/w/{workspace_id}/merge-requests/{merge_request_id}",
+            get(scoped_show_merge_request),
+        )
         .route(
             "/api/w/{workspace_id}/tickets/{id}/merge-request",
-            get(scoped_show_merge_request).post(scoped_open_merge_request),
+            post(scoped_open_merge_request),
         )
         .route(
             "/api/w/{workspace_id}/tickets/{id}/merge-request/readiness",
@@ -4401,39 +4410,172 @@ fn resolve_workspace_ticket_reference(
         .ok_or_else(|| Error::Ticket(ticket::TicketError::NotFound(reference.to_string())).into())
 }
 
+#[derive(Debug, serde::Deserialize)]
+struct MergeRequestListHttpQuery {
+    state: Option<String>,
+    repository_id: Option<String>,
+    ticket_ref: Option<String>,
+    selector_from: Option<String>,
+    selector_to: Option<String>,
+    cursor: Option<String>,
+    limit: Option<usize>,
+}
+
+#[derive(Debug, serde::Serialize)]
+struct MergeRequestRefResponse {
+    status: String,
+    #[serde(rename = "ref")]
+    revision_ref: Option<String>,
+    observed_at: String,
+}
+
+#[derive(Debug, serde::Serialize)]
+struct MergeRequestLinkedTicketResponse {
+    ticket_id: String,
+    key: Option<String>,
+}
+
+#[derive(Debug, serde::Serialize)]
+struct MergeRequestDetailResponse {
+    #[serde(flatten)]
+    merge_request: merge_request::MergeRequest,
+    source: MergeRequestRefResponse,
+    target: MergeRequestRefResponse,
+    linked_tickets: Vec<MergeRequestLinkedTicketResponse>,
+}
+
+async fn scoped_list_merge_requests(
+    State(api): State<WorkspaceApi>,
+    AxumPath(workspace_id): AxumPath<String>,
+    Query(query): Query<MergeRequestListHttpQuery>,
+) -> ApiResult<Json<MergeRequestListResponse>> {
+    let workspace_id = parse_workspace_id(&workspace_id)?;
+    require_workspace_access(&workspace_id, &api)?;
+    let ticket_id = query
+        .ticket_ref
+        .as_deref()
+        .map(|reference| resolve_workspace_ticket_reference(&api, &workspace_id, reference))
+        .transpose()?;
+    let state = query
+        .state
+        .as_deref()
+        .map(|state| match state {
+            "open" => Ok(merge_request::MergeRequestState::Open),
+            "merged" => Ok(merge_request::MergeRequestState::Merged),
+            "closed" => Ok(merge_request::MergeRequestState::Closed),
+            _ => Err(settings_bad_request(
+                "invalid_merge_request_state",
+                "state must be one of open, merged, or closed",
+            )),
+        })
+        .transpose()?;
+    let store = merge_request_store(&api, &workspace_id)?;
+    let page = store.list(
+        &workspace_id,
+        &merge_request::MergeRequestListQuery {
+            state,
+            repository_id: query.repository_id,
+            ticket_id,
+            selector_from: query.selector_from,
+            selector_to: query.selector_to,
+            cursor: query.cursor,
+            limit: query.limit.unwrap_or(50),
+        },
+    )?;
+    let reader = api.repository_reader();
+    let items = page
+        .items
+        .into_iter()
+        .map(|merge_request| {
+            let current_subject_ref = merge_request.selector_from.as_deref().and_then(|selector| {
+                reader
+                    .observe_merge_target(&merge_request.repository_id, Some(selector))
+                    .ok()
+                    .map(|observation| observation.commit)
+            });
+            let ticket_ids = merge_request.ticket_ids.clone();
+            let thread_event_count = merge_request.thread.len();
+            MergeRequestListItem {
+                summary: merge_request_summary(merge_request, current_subject_ref),
+                ticket_ids,
+                thread_event_count,
+            }
+        })
+        .collect();
+    Ok(Json(MergeRequestListResponse {
+        items,
+        next_cursor: page.next_cursor,
+    }))
+}
+
 async fn scoped_show_merge_request(
     State(api): State<WorkspaceApi>,
-    AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
-) -> ApiResult<Json<serde_json::Value>> {
+    AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>,
+    Query(query): Query<MergeRequestThreadQuery>,
+) -> ApiResult<Json<MergeRequestDetailResponse>> {
     let workspace_id = parse_workspace_id(&workspace_id)?;
-    let ticket_id = resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
+    require_workspace_access(&workspace_id, &api)?;
     let store = merge_request_store(&api, &workspace_id)?;
-    let mut mr = store.get(&workspace_id, &ticket_id)?;
-    mr.thread = store.thread_page(&workspace_id, &ticket_id, None, 100)?;
+    let mut mr = store.get_by_id(&workspace_id, &merge_request_id)?;
+    mr.thread = store.thread_page_by_id(
+        &workspace_id,
+        &merge_request_id,
+        query.after,
+        query.limit.unwrap_or(100),
+    )?;
     let reader = api.repository_reader();
     let observed_at = Utc::now().to_rfc3339();
     let source = match mr.selector_from.as_deref() {
         Some(selector) => match reader.observe_merge_target(&mr.repository_id, Some(selector)) {
-            Ok(value) => {
-                serde_json::json!({"status":"known","ref":value.commit,"observed_at":observed_at})
-            }
-            Err(_) => serde_json::json!({"status":"unknown","observed_at":observed_at}),
+            Ok(value) => MergeRequestRefResponse {
+                status: "known".into(),
+                revision_ref: Some(value.commit),
+                observed_at: observed_at.clone(),
+            },
+            Err(_) => MergeRequestRefResponse {
+                status: "unknown".into(),
+                revision_ref: None,
+                observed_at: observed_at.clone(),
+            },
+        },
+        None => MergeRequestRefResponse {
+            status: "requires_repair".into(),
+            revision_ref: None,
+            observed_at: observed_at.clone(),
         },
-        None => serde_json::json!({"status":"requires_repair","observed_at":observed_at}),
     };
     let target = match reader.observe_merge_target(&mr.repository_id, Some(&mr.selector_to)) {
-        Ok(value) => {
-            serde_json::json!({"status":"known","ref":value.commit,"observed_at":observed_at})
-        }
-        Err(_) => serde_json::json!({"status":"unknown","observed_at":observed_at}),
+        Ok(value) => MergeRequestRefResponse {
+            status: "known".into(),
+            revision_ref: Some(value.commit),
+            observed_at,
+        },
+        Err(_) => MergeRequestRefResponse {
+            status: "unknown".into(),
+            revision_ref: None,
+            observed_at,
+        },
     };
-    let mut response =
-        serde_json::to_value(mr).map_err(|error| Error::InvalidInput(error.to_string()))?;
-    if let Some(object) = response.as_object_mut() {
-        object.insert("source".into(), source);
-        object.insert("target".into(), target);
-    }
-    Ok(Json(response))
+    let linked_tickets = mr
+        .ticket_ids
+        .iter()
+        .map(|ticket_id| {
+            Ok(MergeRequestLinkedTicketResponse {
+                ticket_id: ticket_id.clone(),
+                key: api.store.resource_human_key(
+                    &workspace_id,
+                    WorkspaceResourceKind::Ticket,
+                    ticket_id,
+                )?,
+            })
+        })
+        .collect::<Result<Vec<_>>>()?;
+    Ok(Json(MergeRequestDetailResponse {
+        merge_request: mr,
+        source,
+        target,
+        linked_tickets,
+    }))
 }
 
 async fn scoped_merge_request_readiness(
@@ -17849,6 +17991,67 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn merge_request_reads_use_first_class_workspace_resources() {
+        let dir = tempfile::tempdir().unwrap();
+        let app = build_router(test_api(dir.path()).await);
+
+        let collection = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri(format!("/api/w/{TEST_WORKSPACE_ID}/merge-requests"))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(collection.status(), StatusCode::OK);
+        let body = to_bytes(collection.into_body(), usize::MAX).await.unwrap();
+        let body: Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(body["items"], json!([]));
+        assert!(body["next_cursor"].is_null());
+
+        let missing = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri(format!("/api/w/{TEST_WORKSPACE_ID}/merge-requests/missing"))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(missing.status(), StatusCode::NOT_FOUND);
+
+        let nested = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri(format!(
+                        "/api/w/{TEST_WORKSPACE_ID}/tickets/T-1/merge-request"
+                    ))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(nested.status(), StatusCode::METHOD_NOT_ALLOWED);
+
+        let invalid_filter = app
+            .oneshot(
+                Request::builder()
+                    .uri(format!(
+                        "/api/w/{TEST_WORKSPACE_ID}/merge-requests?state=unknown"
+                    ))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(invalid_filter.status(), StatusCode::BAD_REQUEST);
+    }
+
     #[tokio::test]
     async fn ticket_rest_search_requires_worker_source_identity_and_rpc_route_is_removed() {
         let dir = tempfile::tempdir().unwrap();