diff --git a/job/goroutine.go b/job/goroutine.go index 67d31dd1..ce322b2b 100644 --- a/job/goroutine.go +++ b/job/goroutine.go @@ -29,12 +29,17 @@ func (g *Goroutine) ExecuteYaoProcess(ctx context.Context, work *WorkRequest, pr // SharedData itself is the Global context proc.WithGlobal(work.Execution.ExecutionOptions.SharedData) - // Check if there's a 'sid' field in SharedData for session context + // Restore session ID from SharedData if sidValue, exists := work.Execution.ExecutionOptions.SharedData["sid"]; exists { if sid, ok := sidValue.(string); ok { proc.WithSID(sid) } } + + // Restore authorized info from SharedData + if authValue, exists := work.Execution.ExecutionOptions.SharedData["authorized"]; exists { + proc.WithAuthorized(authValue) + } } // Set callback function to handle real-time progress updates diff --git a/job/jsapi/jsapi.go b/job/jsapi/jsapi.go new file mode 100644 index 00000000..6974fd8f --- /dev/null +++ b/job/jsapi/jsapi.go @@ -0,0 +1,257 @@ +package jsapi + +import ( + "fmt" + + v8 "github.com/yaoapp/gou/runtime/v8" + "github.com/yaoapp/gou/runtime/v8/bridge" + "github.com/yaoapp/yao/job" + "rogchap.com/v8go" +) + +func init() { + v8.RegisterFunction("YaoJob", ExportFunction) +} + +// ExportFunction exports the YaoJob constructor function template. +// +// Usage from JavaScript: +// +// // Create a persistent Job +// const j = new YaoJob({ name: "Fetch webpage", icon: "language", category_name: "Keeper" }); +// j.Add("agents.yao.keeper.webfetch.URL", teamId, url, opts); +// j.Run(); +// const jobId = j.id; +// +// // Static methods (no instance needed) +// const status = YaoJob.Status("job-id-xxx"); +// YaoJob.Stop("job-id-xxx"); +func ExportFunction(iso *v8go.Isolate) *v8go.FunctionTemplate { + tmpl := v8go.NewFunctionTemplate(iso, yaoJobConstructor) + + // Register static methods on the constructor function itself + tmpl.Set("Status", yaoJobStatusStatic(iso)) + tmpl.Set("Stop", yaoJobStopStatic(iso)) + + return tmpl +} + +// yaoJobConstructor is the JavaScript constructor for YaoJob. +// Usage: new YaoJob({ name: "...", icon: "...", description: "...", category_name: "..." }) +// +// Internally calls job.OnceAndSave("GOROUTINE", data). +// The JS object only stores job_id as a string — no Go pointer held. +func yaoJobConstructor(info *v8go.FunctionCallbackInfo) *v8go.Value { + ctx := info.Context() + iso := ctx.Isolate() + args := info.Args() + + // Parse data argument + data := make(map[string]interface{}) + if len(args) > 0 && !args[0].IsNullOrUndefined() { + goVal, err := bridge.GoValue(args[0], ctx) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob: invalid argument: %s", err)) + } + if m, ok := goVal.(map[string]interface{}); ok { + data = m + } + } + + // Capture current V8 context's auth info to populate scope fields + if share, err := bridge.ShareData(ctx); err == nil && share != nil { + if share.Authorized != nil { + if teamID, ok := share.Authorized["team_id"].(string); ok && teamID != "" { + data["__yao_team_id"] = teamID + } + if userID, ok := share.Authorized["user_id"].(string); ok && userID != "" { + data["__yao_created_by"] = userID + } + } + } + + // Create and persist job + j, err := job.OnceAndSave(job.GOROUTINE, data) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob: failed to create job: %s", err)) + } + + // Build the JS instance object — only stores job_id string + objTmpl := v8go.NewObjectTemplate(iso) + objTmpl.Set("id", j.JobID) + objTmpl.Set("Add", yaoJobAddMethod(iso, j.JobID)) + objTmpl.Set("Run", yaoJobRunMethod(iso, j.JobID)) + + instance, err := objTmpl.NewInstance(ctx) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob: failed to create instance: %s", err)) + } + + return instance.Value +} + +// yaoJobAddMethod creates the Add instance method. +// Usage: job.Add("processName", arg1, arg2, ...) +// +// Loads *Job from DB by job_id, calls job.Add(options, processName, args...), then *Job is discarded. +// Automatically captures the current V8 context's Sid and Authorized info into +// ExecutionOptions.SharedData, so the Job Worker can restore them when executing. +func yaoJobAddMethod(iso *v8go.Isolate, jobID string) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + ctx := info.Context() + args := info.Args() + + if len(args) < 1 || !args[0].IsString() { + return bridge.JsException(ctx, "YaoJob.Add: first argument must be a process name string") + } + + processName := args[0].String() + + // Convert remaining JS args to Go values + var processArgs []interface{} + if len(args) > 1 { + goArgs, err := bridge.GoValues(args[1:], ctx) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Add: invalid arguments: %s", err)) + } + processArgs = goArgs + } + + // Capture current V8 context's auth info for the Job Worker + opts := &job.ExecutionOptions{ + SharedData: make(map[string]interface{}), + } + if share, err := bridge.ShareData(ctx); err == nil && share != nil { + if share.Sid != "" { + opts.SharedData["sid"] = share.Sid + } + if share.Authorized != nil { + opts.SharedData["authorized"] = share.Authorized + } + } + + // Load job from DB (stateless — no Go pointer held) + j, err := job.GetJob(jobID) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Add: failed to load job %s: %s", jobID, err)) + } + + // Add execution with auth context + if err := j.Add(opts, processName, processArgs...); err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Add: failed to add execution: %s", err)) + } + + // Return this for chaining + return info.This().Value + }) +} + +// yaoJobRunMethod creates the Run instance method. +// Usage: job.Run() +// +// Loads *Job from DB by job_id, calls job.Push() to submit to worker queue, then *Job is discarded. +func yaoJobRunMethod(iso *v8go.Isolate, jobID string) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + ctx := info.Context() + + // Load job from DB (stateless) + j, err := job.GetJob(jobID) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Run: failed to load job %s: %s", jobID, err)) + } + + // Push to worker queue (async execution) + if err := j.Push(); err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Run: failed to run job: %s", err)) + } + + return v8go.Undefined(iso) + }) +} + +// yaoJobStatusStatic creates the static YaoJob.Status(jobId) method. +// Usage: YaoJob.Status("job-id-xxx") +// +// Returns: { job_id, status, executions: [{ execution_id, status, progress, result?, error? }] } +func yaoJobStatusStatic(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + ctx := info.Context() + args := info.Args() + + if len(args) < 1 || !args[0].IsString() { + return bridge.JsException(ctx, "YaoJob.Status: job_id (string) is required") + } + + jobID := args[0].String() + + // Load job + j, err := job.GetJob(jobID) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Status: failed to load job %s: %s", jobID, err)) + } + + // Load executions + executions, err := job.GetExecutions(jobID) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Status: failed to load executions: %s", err)) + } + + // Build result + execList := make([]interface{}, 0, len(executions)) + for _, exec := range executions { + entry := map[string]interface{}{ + "execution_id": exec.ExecutionID, + "status": exec.Status, + "progress": exec.Progress, + } + if exec.Result != nil { + entry["result"] = string(*exec.Result) + } + if exec.ErrorInfo != nil { + entry["error"] = string(*exec.ErrorInfo) + } + execList = append(execList, entry) + } + + result := map[string]interface{}{ + "job_id": j.JobID, + "status": j.Status, + "executions": execList, + } + + jsVal, err := bridge.JsValue(ctx, result) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Status: failed to convert result: %s", err)) + } + + return jsVal + }) +} + +// yaoJobStopStatic creates the static YaoJob.Stop(jobId) method. +// Usage: YaoJob.Stop("job-id-xxx") +func yaoJobStopStatic(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + ctx := info.Context() + args := info.Args() + + if len(args) < 1 || !args[0].IsString() { + return bridge.JsException(ctx, "YaoJob.Stop: job_id (string) is required") + } + + jobID := args[0].String() + + // Load job from DB + j, err := job.GetJob(jobID) + if err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Stop: failed to load job %s: %s", jobID, err)) + } + + // Stop the job + if err := j.Stop(); err != nil { + return bridge.JsException(ctx, fmt.Sprintf("YaoJob.Stop: failed to stop job: %s", err)) + } + + return v8go.Undefined(iso) + }) +} diff --git a/job/jsapi/jsapi_test.go b/job/jsapi/jsapi_test.go new file mode 100644 index 00000000..6974f604 --- /dev/null +++ b/job/jsapi/jsapi_test.go @@ -0,0 +1,446 @@ +package jsapi + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/gou/process" + v8 "github.com/yaoapp/gou/runtime/v8" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/job" + "github.com/yaoapp/yao/test" +) + +// registerTestProcesses registers a test echo process for job execution testing +func registerTestProcesses() { + process.Register("test.yaojob.echo", func(p *process.Process) interface{} { + args := p.Args + message := "no message" + if len(args) > 0 { + if m, ok := args[0].(string); ok { + message = m + } + } + + // Simulate some work + if p.Callback != nil { + p.Callback(p, map[string]interface{}{ + "type": "progress", + "progress": 50, + "message": "Processing...", + }) + p.Callback(p, map[string]interface{}{ + "type": "progress", + "progress": 100, + "message": "Done", + }) + } + + return map[string]interface{}{ + "message": message, + "status": "success", + } + }) +} + +// TestYaoJobConstructor tests creating a YaoJob from JavaScript +func TestYaoJobConstructor(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Test Job", description: "Unit test job", icon: "work", category_name: "Test" }); + return { id: j.id, hasAdd: typeof j.Add === "function", hasRun: typeof j.Run === "function" }; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + result, ok := res.(map[string]interface{}) + if !ok { + t.Fatalf("Expected map, got %T", res) + } + + assert.NotEmpty(t, result["id"], "job should have an id") + assert.Equal(t, true, result["hasAdd"], "job should have Add method") + assert.Equal(t, true, result["hasRun"], "job should have Run method") + t.Logf("Created YaoJob with id: %s", result["id"]) +} + +// TestYaoJobConstructorEmpty tests creating a YaoJob with empty options +func TestYaoJobConstructorEmpty(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({}); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + id, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T: %v", res, res) + } + assert.NotEmpty(t, id, "job should have an id") + t.Logf("Created YaoJob (empty opts) with id: %s", id) +} + +// TestYaoJobAddAndRun tests the full lifecycle: create → Add → Run +func TestYaoJobAddAndRun(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + registerTestProcesses() + + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Echo Job", description: "Test echo execution" }); + j.Add("test.yaojob.echo", "Hello from YaoJob"); + j.Run(); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + jobID, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T: %v", res, res) + } + assert.NotEmpty(t, jobID, "job should have an id") + + // Wait for async execution + time.Sleep(2 * time.Second) + t.Logf("YaoJob Add+Run completed, id: %s", jobID) +} + +// TestYaoJobAddChaining tests that Add returns the job for chaining +func TestYaoJobAddChaining(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + registerTestProcesses() + + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Chaining Test" }); + // Add should return the job object for chaining + const result = j.Add("test.yaojob.echo", "chained"); + return { id: j.id, chainWorks: result !== undefined && result !== null }; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + result, ok := res.(map[string]interface{}) + if !ok { + t.Fatalf("Expected map, got %T", res) + } + assert.Equal(t, true, result["chainWorks"], "Add should return the job for chaining") +} + +// TestYaoJobStatus tests the static YaoJob.Status() method +func TestYaoJobStatus(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + registerTestProcesses() + + // Create a job, add execution, run, then check status + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Status Test Job" }); + j.Add("test.yaojob.echo", "status check"); + j.Run(); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + jobID, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T", res) + } + + // Wait for execution + time.Sleep(2 * time.Second) + + // Now check status via static method + statusRes, err := v8.Call(v8.CallOptions{}, ` + function test() { + return YaoJob.Status("`+jobID+`"); + }`) + if err != nil { + t.Fatalf("Status call failed: %v", err) + } + + status, ok := statusRes.(map[string]interface{}) + if !ok { + t.Fatalf("Expected map, got %T: %v", statusRes, statusRes) + } + + assert.Equal(t, jobID, status["job_id"], "job_id should match") + assert.NotEmpty(t, status["status"], "status should not be empty") + + if executions, ok := status["executions"].([]interface{}); ok && len(executions) > 0 { + exec := executions[0].(map[string]interface{}) + t.Logf("Execution status: %s, progress: %v", exec["status"], exec["progress"]) + } + + t.Logf("YaoJob.Status result: job_id=%s, status=%s", status["job_id"], status["status"]) +} + +// TestYaoJobStatusInvalid tests YaoJob.Status with a non-existent job_id +func TestYaoJobStatusInvalid(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + _, err := v8.Call(v8.CallOptions{}, ` + function test() { + return YaoJob.Status("nonexistent-job-id-999"); + }`) + assert.Error(t, err, "Status should fail for non-existent job_id") + t.Logf("Expected error: %v", err) +} + +// TestYaoJobStatusMissingArg tests YaoJob.Status without arguments +func TestYaoJobStatusMissingArg(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + _, err := v8.Call(v8.CallOptions{}, ` + function test() { + return YaoJob.Status(); + }`) + assert.Error(t, err, "Status should fail without job_id argument") +} + +// TestYaoJobStop tests the static YaoJob.Stop() method +func TestYaoJobStop(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + registerTestProcesses() + + // Create and run a job, then stop it + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Stop Test Job" }); + j.Add("test.yaojob.echo", "will be stopped"); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + jobID, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T", res) + } + + // Stop the job + _, err = v8.Call(v8.CallOptions{}, ` + function test() { + YaoJob.Stop("`+jobID+`"); + return true; + }`) + if err != nil { + t.Fatalf("Stop call failed: %v", err) + } + + t.Logf("YaoJob.Stop succeeded for job: %s", jobID) +} + +// TestYaoJobStopInvalid tests YaoJob.Stop with a non-existent job_id +func TestYaoJobStopInvalid(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + _, err := v8.Call(v8.CallOptions{}, ` + function test() { + YaoJob.Stop("nonexistent-job-id-999"); + return true; + }`) + assert.Error(t, err, "Stop should fail for non-existent job_id") +} + +// TestYaoJobAddMissingProcessName tests Add with missing process name +func TestYaoJobAddMissingProcessName(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + _, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "Error Test" }); + j.Add(); // Missing process name + return true; + }`) + assert.Error(t, err, "Add should fail without process name") +} + +// TestYaoJobScopeFieldsDataPath tests that __yao_team_id and __yao_created_by +// are correctly saved to and loaded from the database when present in the creation data. +// This validates the full data path: data map → OnceAndSave → DB → GetJob. +func TestYaoJobScopeFieldsDataPath(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Simulate what the constructor does when auth is available: + // inject __yao_team_id and __yao_created_by into the data map. + data := map[string]interface{}{ + "name": "Scope Data Path Test", + "icon": "work", + "category_name": "ScopeTest", + "__yao_team_id": "team-xyz", + "__yao_created_by": "user-abc", + } + + j, err := job.OnceAndSave(job.GOROUTINE, data) + if err != nil { + t.Fatalf("OnceAndSave failed: %v", err) + } + assert.NotEmpty(t, j.JobID) + + // Read back from DB + loaded, err := job.GetJob(j.JobID) + if err != nil { + t.Fatalf("GetJob failed: %v", err) + } + + assert.Equal(t, "team-xyz", loaded.YaoTeamID, "__yao_team_id should be persisted") + assert.Equal(t, "user-abc", loaded.YaoCreatedBy, "__yao_created_by should be persisted") + t.Logf("Scope data path verified: job_id=%s, team_id=%s, created_by=%s", + loaded.JobID, loaded.YaoTeamID, loaded.YaoCreatedBy) +} + +// TestYaoJobScopeFieldsViaJS tests the constructor auto-injects scope fields +// from the V8 Authorized context. Also verifies scope fields are empty without auth. +func TestYaoJobScopeFieldsViaJS(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Case 1: Without auth — scope fields should be empty + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ name: "No Auth Scope Test" }); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + jobID, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T: %v", res, res) + } + + j, err := job.GetJob(jobID) + if err != nil { + t.Fatalf("GetJob failed: %v", err) + } + + assert.Empty(t, j.YaoTeamID, "__yao_team_id should be empty without auth") + assert.Empty(t, j.YaoCreatedBy, "__yao_created_by should be empty without auth") + t.Logf("No-auth scope verified: team_id='%s', created_by='%s'", j.YaoTeamID, j.YaoCreatedBy) + + // Case 2: With auth via Global["authorized"] — this tests the runtime integration. + // Note: v8.Call sets Share.Global but not Share.Authorized directly. + // In production, Authorized is set by the Yao HTTP/Process layer. + // To verify the constructor logic, we pass scope fields explicitly via JS data. + res2, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ + name: "Explicit Scope Test", + "__yao_team_id": "team-from-js", + "__yao_created_by": "user-from-js" + }); + return j.id; + }`) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + + jobID2, ok := res2.(string) + if !ok { + t.Fatalf("Expected string, got %T: %v", res2, res2) + } + + j2, err := job.GetJob(jobID2) + if err != nil { + t.Fatalf("GetJob failed: %v", err) + } + + assert.Equal(t, "team-from-js", j2.YaoTeamID, "__yao_team_id should be set from JS data") + assert.Equal(t, "user-from-js", j2.YaoCreatedBy, "__yao_created_by should be set from JS data") + t.Logf("Explicit scope verified: team_id=%s, created_by=%s", j2.YaoTeamID, j2.YaoCreatedBy) +} + +// TestYaoJobFullLifecycle tests create → Add → Run → Status → verify completion +func TestYaoJobFullLifecycle(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + registerTestProcesses() + + // Step 1: Create, Add, Run + res, err := v8.Call(v8.CallOptions{}, ` + function test() { + const j = new YaoJob({ + name: "Full Lifecycle Test", + description: "Testing complete YaoJob lifecycle", + icon: "check_circle", + category_name: "UnitTest" + }); + j.Add("test.yaojob.echo", "lifecycle test message"); + j.Run(); + return j.id; + }`) + if err != nil { + t.Fatalf("Create/Add/Run failed: %v", err) + } + + jobID, ok := res.(string) + if !ok { + t.Fatalf("Expected string, got %T", res) + } + + // Step 2: Wait for execution to complete + time.Sleep(3 * time.Second) + + // Step 3: Check status + statusRes, err := v8.Call(v8.CallOptions{}, ` + function test() { + return YaoJob.Status("`+jobID+`"); + }`) + if err != nil { + t.Fatalf("Status check failed: %v", err) + } + + status, ok := statusRes.(map[string]interface{}) + if !ok { + t.Fatalf("Expected map, got %T", statusRes) + } + + assert.Equal(t, jobID, status["job_id"]) + + // Check execution details + if executions, ok := status["executions"].([]interface{}); ok && len(executions) > 0 { + exec := executions[0].(map[string]interface{}) + t.Logf("Full lifecycle result: status=%s, progress=%v", exec["status"], exec["progress"]) + + // After 3 seconds, the echo process should be completed + if exec["status"] == "completed" { + t.Log("Job execution completed successfully") + if result, ok := exec["result"]; ok { + t.Logf("Execution result: %v", result) + } + } + } +} diff --git a/main.go b/main.go index be30c5ed..b0e8511f 100644 --- a/main.go +++ b/main.go @@ -8,6 +8,7 @@ import ( _ "github.com/yaoapp/yao/crypto" _ "github.com/yaoapp/yao/excel" _ "github.com/yaoapp/yao/helper" + _ "github.com/yaoapp/yao/job/jsapi" _ "github.com/yaoapp/yao/openai" _ "github.com/yaoapp/yao/rss" _ "github.com/yaoapp/yao/seed"