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

feat: add bounded Ticket and Objective query pages

Browse Source
This commit is contained in:
Hare
2026-08-18 03:08:30 +09:00
parent 71cd58f868
commit bc835b8503
7 changed files with 849 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ticket/src/lib.rs
+185 -2
View File
@@ -93,6 +93,26 @@ fn io_err(path: impl Into<PathBuf>, source: io::Error) -> TicketError {
}
}
fn read_ticket_summary_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TicketSummary> {
let workflow_state = row.get::<_, String>(7)?;
Ok(TicketSummary {
id: row.get(0)?,
slug: row.get(1)?,
title: row.get(2)?,
status: ExtensibleTicketStatus::from(row.get::<_, String>(3)?.as_str()),
kind: row.get(4)?,
priority: row.get(5)?,
labels: Vec::new(),
readiness: row.get(6)?,
workflow_state: TicketWorkflowState::parse(&workflow_state)
.unwrap_or(TicketWorkflowState::Planning),
workflow_state_explicit: row.get::<_, i64>(8)? != 0,
queued_by: row.get(9)?,
queued_at: row.get(10)?,
updated_at: row.get(11)?,
})
}
fn sqlite_err(error: impl std::fmt::Display) -> TicketError {
TicketError::Sqlite(error.to_string())
}
@@ -1404,6 +1424,26 @@ pub struct SqliteTicketListProjection {
pub items: Vec<SqliteTicketListItem>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SqliteTicketListCursor {
pub updated_at: Option<String>,
pub ticket_id: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SqliteTicketListPageQuery {
pub states: Vec<TicketWorkflowState>,
pub limit: usize,
pub after: Option<SqliteTicketListCursor>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SqliteTicketListPage {
pub items: Vec<SqliteTicketListItem>,
pub has_more: bool,
pub next: Option<SqliteTicketListCursor>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TicketInvalidRecord {
pub label: String,
@@ -2354,6 +2394,89 @@ impl SqliteTicketBackend {
})
}
/// Lists one stable keyset-paginated Workspace summary page.
///
/// Filtering, ordering, and `limit + 1` are applied by SQLite before the bounded blocker
/// hydration query. The returned cursor is storage data; callers must wrap it in their own
/// opaque, query-bound transport cursor.
pub fn list_workspace_projection_page(
&self,
query: SqliteTicketListPageQuery,
) -> Result<SqliteTicketListPage> {
self.with_read(|conn| {
let states = serde_json::to_string(
&query
.states
.iter()
.map(|state| state.as_str())
.collect::<Vec<_>>(),
)
.map_err(|error| TicketError::Sqlite(error.to_string()))?;
let cursor_updated_at = query
.after
.as_ref()
.and_then(|cursor| cursor.updated_at.clone());
let cursor_id = query
.after
.as_ref()
.map(|cursor| cursor.ticket_id.as_str());
let fetch_limit = query.limit.saturating_add(1);
let mut statement = conn
.prepare(
"SELECT ticket_id, slug, title, status, kind, priority, readiness,
workflow_state, workflow_state_explicit, queued_by, queued_at, updated_at
FROM typed_tickets AS ticket
WHERE workspace_id = ?1
AND (json_array_length(?2) = 0 OR EXISTS (
SELECT 1 FROM json_each(?2) AS state
WHERE state.value = ticket.workflow_state
))
AND (?3 IS NULL OR COALESCE(ticket.updated_at, '') < COALESCE(?4, '')
OR (COALESCE(ticket.updated_at, '') = COALESCE(?4, '')
AND ticket.ticket_id > ?3))
ORDER BY ticket.updated_at DESC, ticket.ticket_id ASC
LIMIT ?5",
)
.map_err(sqlite_err)?;
let rows = statement
.query_map(
params![
self.workspace_id,
states,
cursor_id,
cursor_updated_at,
i64::try_from(fetch_limit).unwrap_or(i64::MAX)
],
read_ticket_summary_row,
)
.map_err(sqlite_err)?;
let mut summaries = rows
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(sqlite_err)?;
let has_more = summaries.len() > query.limit;
summaries.truncate(query.limit);
let next = has_more.then(|| {
let summary = summaries.last().expect("non-empty page with continuation");
SqliteTicketListCursor {
updated_at: summary.updated_at.clone(),
ticket_id: summary.id.clone(),
}
});
let blockers = self.list_workspace_blockers(conn, &summaries)?;
Ok(SqliteTicketListPage {
items: summaries
.into_iter()
.map(|summary| SqliteTicketListItem {
relation_blockers: blockers.get(&summary.id).cloned().unwrap_or_default(),
summary,
})
.collect(),
has_more,
next,
})
})
}
fn list_workspace_summaries(
&self,
conn: &Connection,
@@ -6745,6 +6868,66 @@ state: planning
assert!(!debug.contains("full-artifact-marker"));
}
#[test]
fn sqlite_workspace_projection_page_uses_stable_keyset_and_state_filter() {
let tmp = TempDir::new().unwrap();
let db_path = tmp.path().join("workspace.db");
let backend = SqliteTicketBackend::open(&db_path, "workspace-test").unwrap();
let mut ids = Vec::new();
for (title, state, updated_at) in [
("Newest", TicketWorkflowState::Ready, "2026-08-12T03:00:00Z"),
("Middle", TicketWorkflowState::Ready, "2026-08-12T02:00:00Z"),
(
"Oldest",
TicketWorkflowState::Planning,
"2026-08-12T01:00:00Z",
),
] {
let mut input = NewTicket::new(title);
input.workflow_state = Some(state);
let ticket = backend.create(input).unwrap();
Connection::open(&db_path)
.unwrap()
.execute(
"UPDATE typed_tickets SET updated_at=?3 WHERE workspace_id=?1 AND ticket_id=?2",
params!["workspace-test", ticket.id, updated_at],
)
.unwrap();
ids.push(ticket.id);
}
let first = backend
.list_workspace_projection_page(SqliteTicketListPageQuery {
states: vec![TicketWorkflowState::Ready],
limit: 1,
after: None,
})
.unwrap();
assert_eq!(first.items[0].summary.id, ids[0]);
assert!(first.has_more);
let mut inserted = NewTicket::new("Inserted");
inserted.workflow_state = Some(TicketWorkflowState::Ready);
let inserted = backend.create(inserted).unwrap();
Connection::open(&db_path)
.unwrap()
.execute(
"UPDATE typed_tickets SET updated_at='2026-08-12T04:00:00Z' WHERE workspace_id=?1 AND ticket_id=?2",
params!["workspace-test", inserted.id],
)
.unwrap();
let second = backend
.list_workspace_projection_page(SqliteTicketListPageQuery {
states: vec![TicketWorkflowState::Ready],
limit: 1,
after: first.next,
})
.unwrap();
assert_eq!(second.items[0].summary.id, ids[1]);
assert!(!second.has_more);
}
#[test]
fn sqlite_workspace_projection_sql_shape_is_constant_for_ticket_count() {
let source = include_str!("lib.rs");
@@ -6756,8 +6939,8 @@ state: planning
.map(|offset| start + offset)
.expect("following method");
let projection_source = &source[start..end];
assert_eq!(projection_source.matches("self.with_read(").count(), 1);
assert_eq!(projection_source.matches(".prepare(").count(), 2);
assert_eq!(projection_source.matches("self.with_read(").count(), 2);
assert_eq!(projection_source.matches(".prepare(").count(), 3);
let item_loop = projection_source
.split("items: summaries")
.nth(1)
crates/ticket/src/sqlite_schema.rs
+34 -6
View File
@@ -7,7 +7,7 @@ use crate::{Result, TicketError, sqlite_err};
const MIGRATION_TABLE: &str = "ticket_schema_migrations";
const MAX_SCHEMA_DIAGNOSTICS: usize = 32;
pub const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 3;
pub const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 4;
#[derive(Clone, Copy)]
struct Migration {
@@ -32,6 +32,11 @@ const MIGRATIONS: &[Migration] = &[
name: "convert_legacy_reviews_to_comments",
apply: retire_legacy_ticket_review_events,
},
Migration {
version: 4,
name: "add_ticket_query_indexes",
apply: add_ticket_query_indexes,
},
];
#[derive(Clone, Copy)]
@@ -507,6 +512,29 @@ fn retire_legacy_ticket_review_events(connection: &Connection) -> Result<()> {
.map_err(sqlite_err)
}
fn add_ticket_query_indexes(connection: &Connection) -> Result<()> {
connection
.execute_batch(
r#"
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_state_updated
ON typed_tickets(workspace_id, workflow_state, updated_at DESC, ticket_id);
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_updated
ON typed_tickets(workspace_id, updated_at DESC, ticket_id);
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_created
ON typed_tickets(workspace_id, created_at DESC, ticket_id);
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_title
ON typed_tickets(workspace_id, title COLLATE NOCASE, ticket_id);
CREATE INDEX IF NOT EXISTS typed_ticket_events_workspace_kind_ticket
ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
CREATE INDEX IF NOT EXISTS typed_ticket_relations_workspace_source_kind
ON typed_ticket_relations(workspace_id, ticket_id, kind, target);
CREATE INDEX IF NOT EXISTS typed_ticket_relations_workspace_target_kind
ON typed_ticket_relations(workspace_id, target, kind, ticket_id);
"#,
)
.map_err(sqlite_err)
}
fn add_column_if_missing(
connection: &Connection,
table: &str,
@@ -841,10 +869,10 @@ mod tests {
verify_sqlite_ticket_schema(&connection).unwrap();
let versions = load_applied_migrations(&connection).unwrap();
assert_eq!(versions.len(), 3);
assert_eq!(versions.len(), 4);
assert_eq!(
versions.get(&LATEST_SQLITE_TICKET_SCHEMA_VERSION),
Some(&"convert_legacy_reviews_to_comments".to_string())
Some(&"add_ticket_query_indexes".to_string())
);
}
@@ -989,7 +1017,7 @@ mod tests {
.to_string()
.contains("unsupported Ticket schema migration version 99")
);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 4);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 5);
}
#[test]
@@ -1090,7 +1118,7 @@ mod tests {
connection.execute("INSERT INTO typed_ticket_events (workspace_id,ticket_id,event_index,kind,author,at,status,heading,body) VALUES ('workspace-1','ticket-1',0,'review','reviewer','2026-08-11T00:00:00Z','approve','Review','legacy evidence')",[]).unwrap();
connection.execute("INSERT INTO typed_ticket_event_attributes (workspace_id,ticket_id,event_index,key,value) VALUES ('workspace-1','ticket-1',0,'result','approve')",[]).unwrap();
connection
.execute("DELETE FROM ticket_schema_migrations WHERE version=3", [])
.execute("DELETE FROM ticket_schema_migrations WHERE version>=3", [])
.unwrap();
migrate_sqlite_ticket_schema(&connection).unwrap();
let (kind,status,heading,body):(String,Option<String>,Option<String>,Option<String>)=connection.query_row("SELECT kind,status,heading,body FROM typed_ticket_events WHERE workspace_id='workspace-1' AND ticket_id='ticket-1' AND event_index=0",[],|row|Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?))).unwrap();
@@ -1129,6 +1157,6 @@ mod tests {
let connection = Connection::open(database).unwrap();
verify_sqlite_ticket_schema(&connection).unwrap();
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 3);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 4);
}
}