feat(sdks): add Node and Java parity debt

crates/taskito-core/src/storage/cursor.rs

@@ -31,10 +31,48 @@ pub fn decode_cursor(cursor: &str) -> Result<(i64, &str)> {
     Ok((sort_key, id))
 }
 
+/// Build the next-page cursor from a returned page, or `None` when the page is
+/// the last one — a short page means the listing is exhausted, so callers can
+/// loop `while let Some(cursor)`. `key` reads the row's `(sort_key, id)`.
+///
+/// Lives here rather than in a binding: every SDK pages the same way, and the
+/// "fewer rows than limit ⇒ last page" rule is part of the cursor contract, not
+/// of any one shell.
+pub fn next_cursor<T>(rows: &[T], limit: i64, key: impl Fn(&T) -> (i64, &str)) -> Option<String> {
+    if (rows.len() as i64) < limit {
+        return None;
+    }
+    rows.last().map(|row| {
+        let (sort_key, id) = key(row);
+        encode_cursor(sort_key, id)
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    /// `(sort_key, id)` rows standing in for a listing's page.
+    fn rows(n: usize) -> Vec<(i64, String)> {
+        (0..n).map(|i| (i as i64, format!("id-{i}"))).collect()
+    }
+
+    #[test]
+    fn next_cursor_is_none_on_a_short_page() {
+        assert!(next_cursor(&rows(2), 5, |r| (r.0, &r.1)).is_none());
+    }
+
+    #[test]
+    fn next_cursor_points_at_the_last_row_of_a_full_page() {
+        let cursor = next_cursor(&rows(3), 3, |r| (r.0, &r.1)).expect("full page has a cursor");
+        assert_eq!(cursor, encode_cursor(2, "id-2"));
+    }
+
+    #[test]
+    fn next_cursor_is_none_on_an_empty_page() {
+        assert!(next_cursor::<(i64, String)>(&[], 5, |r| (r.0, &r.1)).is_none());
+    }
+
     #[test]
     fn round_trips() {
         let s = encode_cursor(1_720_000_000_123, "0190a1b2-c3d4-7e5f-8a9b-0c1d2e3f4a5b");

crates/taskito-java/src/convert.rs

@@ -359,6 +359,15 @@ pub struct JobFilter {
     pub offset: Option<i64>,
 }
 
+/// One page of a keyset-paginated listing, as the Java `Page<T>` decodes it.
+/// `next_cursor` is absent on the last page.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PageView<T> {
+    pub items: Vec<T>,
+    pub next_cursor: Option<String>,
+}
+
 /// Map a lowercase status string to the core's `i32` status code.
 pub fn status_code(status: &str) -> Option<i32> {
     match status {

crates/taskito-java/src/queue/inspect.rs

@@ -12,7 +12,7 @@ use taskito_core::Storage;
 use super::borrow_queue;
 use crate::convert::{
     status_code, to_json, CircuitBreakerView, JobErrorView, JobFilter, JobView, MetricView,
-    StatsView, WorkerView,
+    PageView, StatsView, WorkerView,
 };
 use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_optional_string, read_string};
@@ -116,6 +116,86 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listJobs<'
     })
 }
 
+/// `String listJobsAfter(long handle, String filterJson, String afterOrNull)` —
+/// keyset-paginated `listJobs`, ordered by created time. Returns a `Page`; its
+/// `nextCursor` is absent on the last page.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listJobsAfter<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    filter_json: JString<'local>,
+    after: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let raw = read_string(env, &filter_json)?;
+        let filter: JobFilter = crate::convert::parse_json(&raw, "job filter")?;
+        // Reject an unknown status so a typo fails loudly instead of matching all.
+        let status = match filter.status.as_deref() {
+            Some(s) => Some(
+                status_code(s).ok_or_else(|| BindingError::new(format!("unknown status '{s}'")))?,
+            ),
+            None => None,
+        };
+        let limit = filter.limit.unwrap_or(DEFAULT_LIMIT).max(0);
+        // A malformed cursor is a bad request, not a reason to silently restart
+        // from the first page.
+        let after = read_optional_string(env, &after)?;
+        let cursor = after
+            .as_deref()
+            .map(taskito_core::storage::cursor::decode_cursor)
+            .transpose()?;
+        let jobs = queue.storage.list_jobs_after(
+            status,
+            filter.queue.as_deref(),
+            filter.task.as_deref(),
+            limit,
+            cursor,
+            queue.namespace.as_deref(),
+        )?;
+        let next_cursor =
+            taskito_core::storage::cursor::next_cursor(&jobs, limit, |j| (j.created_at, &j.id));
+        let page = PageView {
+            items: jobs.iter().map(JobView::from).collect(),
+            next_cursor,
+        };
+        new_string(env, to_json(&page)?)
+    })
+}
+
+/// `String listArchivedAfter(long handle, long limit, String afterOrNull)` —
+/// keyset-paginated `listArchived`, ordered by completed time.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listArchivedAfter<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    limit: jlong,
+    after: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let limit = limit.max(0);
+        let after = read_optional_string(env, &after)?;
+        let cursor = after
+            .as_deref()
+            .map(taskito_core::storage::cursor::decode_cursor)
+            .transpose()?;
+        let jobs = queue.storage.list_archived_after(limit, cursor)?;
+        // Archived rows always carry `completed_at`; the fallback keeps the key
+        // non-null for a row written before it was set.
+        let next_cursor = taskito_core::storage::cursor::next_cursor(&jobs, limit, |j| {
+            (j.completed_at.unwrap_or(j.created_at), &j.id)
+        });
+        let page = PageView {
+            items: jobs.iter().map(JobView::from).collect(),
+            next_cursor,
+        };
+        new_string(env, to_json(&page)?)
+    })
+}
+
 /// `String jobErrors(long handle, String jobId)` — one entry per failed attempt.
 #[no_mangle]
 pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_jobErrors<'local>(

crates/taskito-node/src/config.rs

@@ -88,6 +88,28 @@ pub struct JobFilter {
     pub offset: Option<i64>,
 }
 
+/// Filter for [`crate::queue::JsQueue::list_jobs_filtered`]: everything
+/// [`JobFilter`] matches on, plus substring and created-at-range predicates.
+/// All fields optional.
+#[napi(object)]
+#[derive(Default)]
+pub struct DetailedJobFilter {
+    /// Lowercase status (`"pending"`, `"running"`, `"complete"`, `"failed"`, `"dead"`, `"cancelled"`).
+    pub status: Option<String>,
+    pub queue: Option<String>,
+    pub task: Option<String>,
+    /// Substring matched against the job's metadata.
+    pub metadata_like: Option<String>,
+    /// Substring matched against the job's error.
+    pub error_like: Option<String>,
+    /// Only jobs created at or after this Unix-millisecond timestamp.
+    pub created_after: Option<i64>,
+    /// Only jobs created at or before this Unix-millisecond timestamp.
+    pub created_before: Option<i64>,
+    pub limit: Option<i64>,
+    pub offset: Option<i64>,
+}
+
 /// Circuit-breaker config for a task: trip after `threshold` failures within
 /// `windowMs`, stay open for `cooldownMs`, then probe before fully closing.
 #[napi(object)]

crates/taskito-node/src/convert/job.rs

@@ -115,3 +115,15 @@ pub fn job_to_js(job: Job) -> JsJob {
         namespace: job.namespace,
     }
 }
+
+/// One page of jobs plus the cursor for the next. napi has no generics, so each
+/// item type needs its own page struct; the TS shell declares a `Page<T>` they
+/// all satisfy structurally.
+///
+/// `next_cursor` is `null` on the last page — a short page means the listing is
+/// exhausted — so callers can loop until it is null.
+#[napi(object)]
+pub struct JsJobPage {
+    pub items: Vec<JsJob>,
+    pub next_cursor: Option<String>,
+}

crates/taskito-node/src/convert/mod.rs

@@ -13,7 +13,7 @@ mod task_config;
 #[cfg(feature = "workflows")]
 mod workflow;
 
-pub use job::{build_new_job, job_to_js, JsJob, JsTaskInvocation};
+pub use job::{build_new_job, job_to_js, JsJob, JsJobPage, JsTaskInvocation};
 pub(crate) use job::{DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS};
 pub use lock::{lock_info_to_js, JsLockInfo};
 pub use log::{log_to_js, JsTaskLog};
@@ -25,7 +25,7 @@ pub use periodic::{periodic_to_js, JsPeriodicTask};
 pub use pubsub::{subscription_to_js, JsSubscription};
 pub use stats::{
     dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js,
-    JsDeadJob, JsJobError, JsMetric, JsStats, JsWorkerRow,
+    JsDeadJob, JsDeadJobPage, JsJobError, JsMetric, JsStats, JsWorkerRow,
 };
 pub use task_config::{queue_config, task_config};
 #[cfg(feature = "workflows")]

crates/taskito-node/src/convert/stats.rs

@@ -145,3 +145,11 @@ pub fn status_code(status: &str) -> Option<i32> {
         _ => None,
     }
 }
+
+/// One page of dead-letter entries plus the cursor for the next. See
+/// [`crate::convert::JsJobPage`] for the contract.
+#[napi(object)]
+pub struct JsDeadJobPage {
+    pub items: Vec<JsDeadJob>,
+    pub next_cursor: Option<String>,
+}

crates/taskito-node/src/queue/admin.rs

@@ -10,7 +10,7 @@ use taskito_core::job::{now_millis, NewJob};
 use taskito_core::Storage;
 
 use super::JsQueue;
-use crate::convert::{dead_job_to_js, JsDeadJob};
+use crate::convert::{dead_job_to_js, JsDeadJob, JsDeadJobPage};
 use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err};
 
 const DEFAULT_LIMIT: i64 = 50;
@@ -35,6 +35,38 @@ impl JsQueue {
         .map_err(join_to_napi_err)?
     }
 
+    /// Keyset-paginated [`JsQueue::dead_letters`], ordered by failed time. Pass
+    /// a page's `nextCursor` back as `after`; `null` means the last page.
+    #[napi]
+    pub async fn dead_letters_after(
+        &self,
+        limit: Option<i64>,
+        after: Option<String>,
+    ) -> Result<JsDeadJobPage> {
+        let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            // A malformed cursor is a bad request, not a reason to silently
+            // restart from the first page.
+            let cursor = after
+                .as_deref()
+                .map(taskito_core::storage::cursor::decode_cursor)
+                .transpose()
+                .map_err(to_napi_err)?;
+            let dead = storage
+                .list_dead_after(limit, cursor)
+                .map_err(to_napi_err)?;
+            let next_cursor =
+                taskito_core::storage::cursor::next_cursor(&dead, limit, |d| (d.failed_at, &d.id));
+            Ok(JsDeadJobPage {
+                items: dead.into_iter().map(dead_job_to_js).collect(),
+                next_cursor,
+            })
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
+
     /// List dead-letter entries for a single task (paginated, newest first).
     #[napi]
     pub async fn dead_letters_by_task(