feat: implement task event system for progress tracking and reporting

api/factory.go

@@ -20,6 +20,7 @@ import (
 	"github.com/krau/SaveAny-Bot/pkg/aria2"
 	"github.com/krau/SaveAny-Bot/pkg/enums/tasktype"
 	"github.com/krau/SaveAny-Bot/pkg/parser"
+	"github.com/krau/SaveAny-Bot/pkg/taskevent"
 	"github.com/krau/SaveAny-Bot/pkg/telegraph"
 	"github.com/krau/SaveAny-Bot/storage"
 	"github.com/rs/xid"
@@ -68,9 +69,14 @@ func (f *TaskFactory) CreateTask(req *CreateTaskRequest) (*CreateTaskResponse, e
 
 func (f *TaskFactory) registerAndEnqueueTask(task core.Executable, taskType tasktype.TaskType, storageName, path, webhook string) error {
 	taskID := task.TaskID()
-	RegisterTask(taskID, string(taskType), storageName, path, task.Title(), webhook)
+	info := RegisterTask(taskID, string(taskType), storageName, path, task.Title(), webhook)
 
-	err := core.AddTask(f.ctx, NewExecutableWrapper(task))
+	// Inject the progress sink into the context so the task's Emit calls update
+	// the API store (and fire the webhook on terminal states) without the task
+	// knowing about the API.
+	taskCtx := taskevent.WithSink(f.ctx, info)
+
+	err := core.AddTask(taskCtx, task)
 	if err != nil {
 		DeleteTask(taskID)
 		return fmt.Errorf("failed to add task: %w", err)

api/handlers.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/krau/SaveAny-Bot/core"
 	"github.com/krau/SaveAny-Bot/pkg/enums/tasktype"
@@ -117,7 +118,7 @@ func (h *Handlers) CancelTaskHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// 取消任务
+	// Cancel the task; the terminal status is set via the task event stream.
 	if err := core.CancelTask(r.Context(), taskID); err != nil {
 		WriteError(w, http.StatusInternalServerError, "cancel_failed", "failed to cancel task: "+err.Error())
 		return
@@ -184,27 +185,45 @@ func extractTaskIDFromPath(path string) string {
 	return parts[3]
 }
 
-// convertTaskProgressToResponse 将任务进度转换为响应格式
+// convertTaskProgressToResponse renders a task's current state, computing
+// percent and speed from the snapshot taken under the task's mutex.
 func convertTaskProgressToResponse(task *TaskProgressInfo) TaskInfoResponse {
+	status, total, downloaded, totalFiles, downloadedFiles, startedAt, errMsg, updatedAt := task.snapshot()
+
 	resp := TaskInfoResponse{
 		TaskID:    task.TaskID,
 		Type:      tasktype.TaskType(task.Type),
-		Status:    task.Status,
+		Status:    status,
 		Title:     task.Title,
 		Storage:   task.Storage,
 		Path:      task.Path,
-		Error:     task.Error,
+		Error:     errMsg,
 		CreatedAt: task.CreatedAt,
-		UpdatedAt: task.UpdatedAt,
+		UpdatedAt: updatedAt,
+	}
+
+	var percent float64
+	var speedMBPS float64
+	if total > 0 {
+		percent = float64(downloaded) * 100 / float64(total)
+	} else if totalFiles > 0 {
+		percent = float64(downloadedFiles) * 100 / float64(totalFiles)
+	}
+	if !startedAt.IsZero() {
+		elapsed := time.Since(startedAt).Seconds()
+		if elapsed > 0 {
+			speedMBPS = float64(downloaded) / elapsed / (1024 * 1024)
+		}
 	}
 
-	// 计算进度
-	if task.TotalBytes > 0 {
-		percent := float64(task.DownloadedBytes) * 100 / float64(task.TotalBytes)
+	if total > 0 || totalFiles > 0 {
 		resp.Progress = &TaskProgress{
-			TotalBytes:      task.TotalBytes,
-			DownloadedBytes: task.DownloadedBytes,
+			TotalBytes:      total,
+			DownloadedBytes: downloaded,
+			TotalFiles:      totalFiles,
+			DownloadedFiles: downloadedFiles,
 			Percent:         percent,
+			SpeedMBPS:       speedMBPS,
 		}
 	}
 

api/handlers_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/krau/SaveAny-Bot/pkg/enums/tasktype"
+	"github.com/krau/SaveAny-Bot/pkg/taskevent"
 )
 
 // setupTestServer creates a test server with handlers
@@ -403,32 +404,38 @@ func TestConcurrentProgressStore(t *testing.T) {
 
 // TestProgressTrackerConcurrentUpdates tests concurrent progress updates
 func TestProgressTrackerConcurrentUpdates(t *testing.T) {
-	tracker := NewProgressTracker("concurrent-progress", "directlinks", "local", "downloads", "Test", "")
-	tracker.OnStart(10000, 10)
+	info := RegisterTask("concurrent-progress", "directlinks", "local", "downloads", "Test", "")
+	info.Emit(taskevent.Event{TaskID: "concurrent-progress", Phase: taskevent.PhaseStart, TotalBytes: 10000})
 
 	var wg sync.WaitGroup
 	numGoroutines := 50
 	updatesPerGoroutine := 100
 
-	// Concurrent progress updates
+	// Concurrent progress updates via the Sink interface
 	for i := range numGoroutines {
 		wg.Add(1)
 		go func(id int) {
 			defer wg.Done()
 			for j := range updatesPerGoroutine {
-				tracker.OnProgress(int64(id*updatesPerGoroutine+j), j)
+				info.Emit(taskevent.Event{
+					TaskID:          "concurrent-progress",
+					Phase:           taskevent.PhaseProgress,
+					DownloadedBytes: int64(id*updatesPerGoroutine + j),
+					TotalBytes:      10000,
+				})
 			}
 		}(i)
 	}
 
 	wg.Wait()
 
-	info := tracker.GetInfo()
-	if info.Status != TaskStatusRunning {
-		t.Errorf("expected status Running after concurrent updates, got %s", info.Status)
+	status, _, downloaded, _, _, _, _, _ := info.snapshot()
+	if status != TaskStatusRunning {
+		t.Errorf("expected status Running after concurrent updates, got %s", status)
+	}
+	if downloaded <= 0 {
+		t.Errorf("expected downloaded bytes > 0 after concurrent updates, got %d", downloaded)
 	}
-	// Note: Due to race conditions in the simple implementation,
-	// we can't reliably check exact values without proper synchronization
 }
 
 // TestTaskFactoryValidation tests TaskFactory parameter validation
@@ -526,8 +533,7 @@ func TestEdgeCases(t *testing.T) {
 		{
 			name: "Progress tracker with empty webhook",
 			fn: func(t *testing.T) {
-				tracker := NewProgressTracker("test", "type", "storage", "path", "title", "")
-				info := tracker.GetInfo()
+				info := RegisterTask("test-empty-webhook", "type", "storage", "path", "title", "")
 				if info.Webhook != "" {
 					t.Error("expected empty webhook")
 				}