From 30c1f8cf47ecb1b0930f100196681bb93f927533 Mon Sep 17 00:00:00 2001 From: Deluan Date: Thu, 26 Feb 2026 15:10:23 -0500 Subject: [PATCH] feat(plugins): define TaskWorker capability for task execution callbacks --- plugins/capabilities/taskworker.go | 33 +++++++ plugins/capabilities/taskworker.yaml | 45 ++++++++++ plugins/pdk/go/taskworker/taskworker.go | 86 ++++++++++++++++++ plugins/pdk/go/taskworker/taskworker_stub.go | 48 ++++++++++ .../pdk/rust/nd-pdk-capabilities/src/lib.rs | 1 + .../nd-pdk-capabilities/src/taskworker.rs | 87 +++++++++++++++++++ 6 files changed, 300 insertions(+) create mode 100644 plugins/capabilities/taskworker.go create mode 100644 plugins/capabilities/taskworker.yaml create mode 100644 plugins/pdk/go/taskworker/taskworker.go create mode 100644 plugins/pdk/go/taskworker/taskworker_stub.go create mode 100644 plugins/pdk/rust/nd-pdk-capabilities/src/taskworker.rs diff --git a/plugins/capabilities/taskworker.go b/plugins/capabilities/taskworker.go new file mode 100644 index 000000000..8fc2dc9fa --- /dev/null +++ b/plugins/capabilities/taskworker.go @@ -0,0 +1,33 @@ +package capabilities + +// TaskWorker provides task execution handling. +// This capability allows plugins to receive callbacks when their queued tasks +// are ready to execute. Plugins that use the taskqueue host service must +// implement this capability. +// +//nd:capability name=taskworker +type TaskWorker interface { + // OnTaskExecute is called when a queued task is ready to run. + // Return an error to trigger retry (if retries are configured). + //nd:export name=nd_task_execute + OnTaskExecute(TaskExecuteRequest) (TaskExecuteResponse, error) +} + +// TaskExecuteRequest is the request provided when a task is ready to execute. +type TaskExecuteRequest struct { + // QueueName is the name of the queue this task belongs to. + QueueName string `json:"queueName"` + // TaskID is the unique identifier for this task. + TaskID string `json:"taskId"` + // Payload is the opaque data provided when the task was enqueued. + Payload []byte `json:"payload"` + // Attempt is the current attempt number (1-based: first attempt = 1). + Attempt int32 `json:"attempt"` +} + +// TaskExecuteResponse is the response from task execution. +type TaskExecuteResponse struct { + // Error, if non-empty, indicates the task failed. The task will be retried + // if retries are configured and attempts remain. + Error string `json:"error,omitempty"` +} diff --git a/plugins/capabilities/taskworker.yaml b/plugins/capabilities/taskworker.yaml new file mode 100644 index 000000000..760ca420b --- /dev/null +++ b/plugins/capabilities/taskworker.yaml @@ -0,0 +1,45 @@ +version: v1-draft +exports: + nd_task_execute: + description: |- + OnTaskExecute is called when a queued task is ready to run. + Return an error to trigger retry (if retries are configured). + input: + $ref: '#/components/schemas/TaskExecuteRequest' + contentType: application/json + output: + $ref: '#/components/schemas/TaskExecuteResponse' + contentType: application/json +components: + schemas: + TaskExecuteRequest: + description: TaskExecuteRequest is the request provided when a task is ready to execute. + properties: + queueName: + type: string + description: QueueName is the name of the queue this task belongs to. + taskId: + type: string + description: TaskID is the unique identifier for this task. + payload: + type: array + description: Payload is the opaque data provided when the task was enqueued. + items: + type: object + attempt: + type: integer + format: int32 + description: 'Attempt is the current attempt number (1-based: first attempt = 1).' + required: + - queueName + - taskId + - payload + - attempt + TaskExecuteResponse: + description: TaskExecuteResponse is the response from task execution. + properties: + error: + type: string + description: |- + Error, if non-empty, indicates the task failed. The task will be retried + if retries are configured and attempts remain. diff --git a/plugins/pdk/go/taskworker/taskworker.go b/plugins/pdk/go/taskworker/taskworker.go new file mode 100644 index 000000000..da7f76b62 --- /dev/null +++ b/plugins/pdk/go/taskworker/taskworker.go @@ -0,0 +1,86 @@ +// Code generated by ndpgen. DO NOT EDIT. +// +// This file contains export wrappers for the TaskWorker capability. +// It is intended for use in Navidrome plugins built with TinyGo. +// +//go:build wasip1 + +package taskworker + +import ( + "github.com/navidrome/navidrome/plugins/pdk/go/pdk" +) + +// TaskExecuteRequest is the request provided when a task is ready to execute. +type TaskExecuteRequest struct { + // QueueName is the name of the queue this task belongs to. + QueueName string `json:"queueName"` + // TaskID is the unique identifier for this task. + TaskID string `json:"taskId"` + // Payload is the opaque data provided when the task was enqueued. + Payload []byte `json:"payload"` + // Attempt is the current attempt number (1-based: first attempt = 1). + Attempt int32 `json:"attempt"` +} + +// TaskExecuteResponse is the response from task execution. +type TaskExecuteResponse struct { + // Error, if non-empty, indicates the task failed. The task will be retried + // if retries are configured and attempts remain. + Error string `json:"error,omitempty"` +} + +// TaskWorker is the marker interface for taskworker plugins. +// Implement one or more of the provider interfaces below. +// TaskWorker provides task execution handling. +// This capability allows plugins to receive callbacks when their queued tasks +// are ready to execute. Plugins that use the taskqueue host service must +// implement this capability. +type TaskWorker interface{} + +// TaskExecuteProvider provides the OnTaskExecute function. +type TaskExecuteProvider interface { + OnTaskExecute(TaskExecuteRequest) (TaskExecuteResponse, error) +} // Internal implementation holders +var ( + taskExecuteImpl func(TaskExecuteRequest) (TaskExecuteResponse, error) +) + +// Register registers a taskworker implementation. +// The implementation is checked for optional provider interfaces. +func Register(impl TaskWorker) { + if p, ok := impl.(TaskExecuteProvider); ok { + taskExecuteImpl = p.OnTaskExecute + } +} + +// NotImplementedCode is the standard return code for unimplemented functions. +// The host recognizes this and skips the plugin gracefully. +const NotImplementedCode int32 = -2 + +//go:wasmexport nd_task_execute +func _NdTaskExecute() int32 { + if taskExecuteImpl == nil { + // Return standard code - host will skip this plugin gracefully + return NotImplementedCode + } + + var input TaskExecuteRequest + if err := pdk.InputJSON(&input); err != nil { + pdk.SetError(err) + return -1 + } + + output, err := taskExecuteImpl(input) + if err != nil { + pdk.SetError(err) + return -1 + } + + if err := pdk.OutputJSON(output); err != nil { + pdk.SetError(err) + return -1 + } + + return 0 +} diff --git a/plugins/pdk/go/taskworker/taskworker_stub.go b/plugins/pdk/go/taskworker/taskworker_stub.go new file mode 100644 index 000000000..9944f2d1f --- /dev/null +++ b/plugins/pdk/go/taskworker/taskworker_stub.go @@ -0,0 +1,48 @@ +// Code generated by ndpgen. DO NOT EDIT. +// +// This file provides stub implementations for non-WASM platforms. +// It allows Go plugins to compile and run tests outside of WASM, +// but the actual functionality is only available in WASM builds. +// +//go:build !wasip1 + +package taskworker + +// TaskExecuteRequest is the request provided when a task is ready to execute. +type TaskExecuteRequest struct { + // QueueName is the name of the queue this task belongs to. + QueueName string `json:"queueName"` + // TaskID is the unique identifier for this task. + TaskID string `json:"taskId"` + // Payload is the opaque data provided when the task was enqueued. + Payload []byte `json:"payload"` + // Attempt is the current attempt number (1-based: first attempt = 1). + Attempt int32 `json:"attempt"` +} + +// TaskExecuteResponse is the response from task execution. +type TaskExecuteResponse struct { + // Error, if non-empty, indicates the task failed. The task will be retried + // if retries are configured and attempts remain. + Error string `json:"error,omitempty"` +} + +// TaskWorker is the marker interface for taskworker plugins. +// Implement one or more of the provider interfaces below. +// TaskWorker provides task execution handling. +// This capability allows plugins to receive callbacks when their queued tasks +// are ready to execute. Plugins that use the taskqueue host service must +// implement this capability. +type TaskWorker interface{} + +// TaskExecuteProvider provides the OnTaskExecute function. +type TaskExecuteProvider interface { + OnTaskExecute(TaskExecuteRequest) (TaskExecuteResponse, error) +} + +// NotImplementedCode is the standard return code for unimplemented functions. +const NotImplementedCode int32 = -2 + +// Register is a no-op on non-WASM platforms. +// This stub allows code to compile outside of WASM. +func Register(_ TaskWorker) {} diff --git a/plugins/pdk/rust/nd-pdk-capabilities/src/lib.rs b/plugins/pdk/rust/nd-pdk-capabilities/src/lib.rs index 0f0daf80f..06c2c5c0d 100644 --- a/plugins/pdk/rust/nd-pdk-capabilities/src/lib.rs +++ b/plugins/pdk/rust/nd-pdk-capabilities/src/lib.rs @@ -9,4 +9,5 @@ pub mod lifecycle; pub mod metadata; pub mod scheduler; pub mod scrobbler; +pub mod taskworker; pub mod websocket; diff --git a/plugins/pdk/rust/nd-pdk-capabilities/src/taskworker.rs b/plugins/pdk/rust/nd-pdk-capabilities/src/taskworker.rs new file mode 100644 index 000000000..961d44b85 --- /dev/null +++ b/plugins/pdk/rust/nd-pdk-capabilities/src/taskworker.rs @@ -0,0 +1,87 @@ +// Code generated by ndpgen. DO NOT EDIT. +// +// This file contains export wrappers for the TaskWorker capability. +// It is intended for use in Navidrome plugins built with extism-pdk. + +use serde::{Deserialize, Serialize}; + +// Helper functions for skip_serializing_if with numeric types +#[allow(dead_code)] +fn is_zero_i32(value: &i32) -> bool { *value == 0 } +#[allow(dead_code)] +fn is_zero_u32(value: &u32) -> bool { *value == 0 } +#[allow(dead_code)] +fn is_zero_i64(value: &i64) -> bool { *value == 0 } +#[allow(dead_code)] +fn is_zero_u64(value: &u64) -> bool { *value == 0 } +#[allow(dead_code)] +fn is_zero_f32(value: &f32) -> bool { *value == 0.0 } +#[allow(dead_code)] +fn is_zero_f64(value: &f64) -> bool { *value == 0.0 } +/// TaskExecuteRequest is the request provided when a task is ready to execute. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskExecuteRequest { + /// QueueName is the name of the queue this task belongs to. + #[serde(default)] + pub queue_name: String, + /// TaskID is the unique identifier for this task. + #[serde(default)] + pub task_id: String, + /// Payload is the opaque data provided when the task was enqueued. + #[serde(default)] + pub payload: Vec, + /// Attempt is the current attempt number (1-based: first attempt = 1). + #[serde(default)] + pub attempt: i32, +} +/// TaskExecuteResponse is the response from task execution. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskExecuteResponse { + /// Error, if non-empty, indicates the task failed. The task will be retried + /// if retries are configured and attempts remain. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub error: String, +} + +/// Error represents an error from a capability method. +#[derive(Debug)] +pub struct Error { + pub message: String, +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.message) + } +} + +impl std::error::Error for Error {} + +impl Error { + pub fn new(message: impl Into) -> Self { + Self { message: message.into() } + } +} + +/// TaskExecuteProvider provides the OnTaskExecute function. +pub trait TaskExecuteProvider { + fn on_task_execute(&self, req: TaskExecuteRequest) -> Result; +} + +/// Register the on_task_execute export. +/// This macro generates the WASM export function for this method. +#[macro_export] +macro_rules! register_taskworker_task_execute { + ($plugin_type:ty) => { + #[extism_pdk::plugin_fn] + pub fn nd_task_execute( + req: extism_pdk::Json<$crate::taskworker::TaskExecuteRequest> + ) -> extism_pdk::FnResult> { + let plugin = <$plugin_type>::default(); + let result = $crate::taskworker::TaskExecuteProvider::on_task_execute(&plugin, req.into_inner())?; + Ok(extism_pdk::Json(result)) + } + }; +}