implement action sdk

Author: mateeullahmalik

sdk/action/.gitkeep

@@ -0,0 +1,143 @@
+package action
+
+import (
+	"context"
+	"fmt"
+
+	"action/config"
+	"action/event"
+	"action/task"
+
+	"github.com/LumeraProtocol/supernode/pkg/lumera"
+)
+
+type ActionClient interface {
+	StartSense(ctx context.Context, fileHash string, actionID string, filePath string) (string, error)
+
+	StartCascade(ctx context.Context, fileHash string, actionID string, filePath string, signedData string) (string, error)
+
+	GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool)
+
+	SubscribeToEvents(eventType event.EventType, handler event.Handler)
+
+	SubscribeToAllEvents(handler event.Handler)
+}
+
+type ActionClientImpl struct {
+	lumeraClient lumera.Client
+	config       config.Config
+	taskManager  task.Manager
+}
+
+func NewActionClient(lumeraClient lumera.Client, config config.Config) (ActionClient, error) {
+	taskManager, err := task.NewManager(lumeraClient, config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create task manager: %w", err)
+	}
+
+	return &ActionClientImpl{
+		lumeraClient: lumeraClient,
+		config:       config,
+		taskManager:  taskManager,
+	}, nil
+}
+
+func (ac *ActionClientImpl) StartSense(
+	ctx context.Context,
+	fileHash string,
+	actionID string,
+	filePath string,
+) (string, error) {
+	ac.config.Logger.Debug(ctx, "Starting sense operation",
+		"fileHash", fileHash,
+		"actionID", actionID,
+		"filePath", filePath)
+
+	if fileHash == "" {
+		ac.config.Logger.Error(ctx, "Empty file hash provided")
+		return "", ErrEmptyFileHash
+	}
+	if actionID == "" {
+		ac.config.Logger.Error(ctx, "Empty action ID provided")
+		return "", ErrEmptyActionID
+	}
+	if filePath == "" {
+		ac.config.Logger.Error(ctx, "Empty file path provided")
+		return "", ErrEmptyFilePath
+	}
+
+	taskID, err := ac.taskManager.CreateSenseTask(ctx, fileHash, actionID, filePath)
+	if err != nil {
+		ac.config.Logger.Error(ctx, "Failed to create sense task", "error", err)
+		return "", fmt.Errorf("failed to create sense task: %w", err)
+	}
+
+	ac.config.Logger.Info(ctx, "Sense task created successfully", "taskID", taskID)
+	return taskID, nil
+}
+
+func (ac *ActionClientImpl) StartCascade(
+	ctx context.Context,
+	fileHash string,
+	actionID string,
+	filePath string,
+	signedData string,
+) (string, error) {
+	ac.config.Logger.Debug(ctx, "Starting cascade operation",
+		"fileHash", fileHash,
+		"actionID", actionID,
+		"filePath", filePath)
+
+	if fileHash == "" {
+		ac.config.Logger.Error(ctx, "Empty file hash provided")
+		return "", ErrEmptyFileHash
+	}
+	if actionID == "" {
+		ac.config.Logger.Error(ctx, "Empty action ID provided")
+		return "", ErrEmptyActionID
+	}
+	if filePath == "" {
+		ac.config.Logger.Error(ctx, "Empty file path provided")
+		return "", ErrEmptyFilePath
+	}
+
+	taskID, err := ac.taskManager.CreateCascadeTask(ctx, fileHash, actionID, filePath, signedData)
+	if err != nil {
+		ac.config.Logger.Error(ctx, "Failed to create cascade task", "error", err)
+		return "", fmt.Errorf("failed to create cascade task: %w", err)
+	}
+
+	ac.config.Logger.Info(ctx, "Cascade task created successfully", "taskID", taskID)
+	return taskID, nil
+}
+
+func (ac *ActionClientImpl) GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool) {
+	ac.config.Logger.Debug(ctx, "Getting task", "taskID", taskID)
+	task, found := ac.taskManager.GetTask(ctx, taskID)
+	if !found {
+		ac.config.Logger.Debug(ctx, "Task not found", "taskID", taskID)
+	} else {
+		ac.config.Logger.Debug(ctx, "Task found", "taskID", taskID, "status", task.Status)
+	}
+	return task, found
+}
+
+// SubscribeToEvents registers a handler for specific event types
+func (ac *ActionClientImpl) SubscribeToEvents(eventType event.EventType, handler event.Handler) {
+	ac.config.Logger.Debug(context.Background(), "Subscribing to events", "eventType", eventType)
+	if ac.config.EventBus != nil {
+		ac.config.EventBus.Subscribe(eventType, handler)
+	} else {
+		ac.config.Logger.Warn(context.Background(), "EventBus is nil, cannot subscribe to events")
+	}
+}
+
+// SubscribeToAllEvents registers a handler for all events
+func (ac *ActionClientImpl) SubscribeToAllEvents(handler event.Handler) {
+	ac.config.Logger.Debug(context.Background(), "Subscribing to all events")
+	if ac.config.EventBus != nil {
+		ac.config.EventBus.SubscribeAll(handler)
+	} else {
+		ac.config.Logger.Warn(context.Background(), "EventBus is nil, cannot subscribe to events")
+	}
+}

sdk/action/client.go

@@ -0,0 +1,51 @@
+package action
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	ErrEmptyFileHash = errors.New("file hash cannot be empty")
+	ErrEmptyActionID = errors.New("action ID cannot be empty")
+	ErrEmptyFilePath = errors.New("file path cannot be empty")
+	ErrNoValidAction = errors.New("no action found with the specified ID")
+	ErrInvalidAction = errors.New("action is not in a valid state")
+	ErrNoSupernodes  = errors.New("no valid supernodes available")
+	ErrTaskCreation  = errors.New("failed to create task")
+	ErrCommunication = errors.New("communication with supernode failed")
+)
+
+// SupernodeError represents an error related to supernode operations
+type SupernodeError struct {
+	NodeID  string
+	Message string
+	Err     error
+}
+
+// Error returns the error message
+func (e *SupernodeError) Error() string {
+	return fmt.Sprintf("supernode error (ID: %s): %s: %v", e.NodeID, e.Message, e.Err)
+}
+
+// Unwrap returns the underlying error
+func (e *SupernodeError) Unwrap() error {
+	return e.Err
+}
+
+// ActionError represents an error related to action operations
+type ActionError struct {
+	ActionID string
+	Message  string
+	Err      error
+}
+
+// Error returns the error message
+func (e *ActionError) Error() string {
+	return fmt.Sprintf("action error (ID: %s): %s: %v", e.ActionID, e.Message, e.Err)
+}
+
+// Unwrap returns the underlying error
+func (e *ActionError) Unwrap() error {
+	return e.Err
+}

sdk/action/client/check_health.go

@@ -0,0 +1,8 @@
+package action
+
+type TaskType string
+
+const (
+	TaskTypeSense   TaskType = "SENSE"
+	TaskTypeCascade TaskType = "CASCADE"
+)