Updates

Author: mateeullahmalik

sdk/action/client.go

@@ -12,24 +12,29 @@ import (
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
 )
 
+// Client defines the interface for action operations
 type Client interface {
 	StartCascade(ctx context.Context, fileHash string, actionID string, filePath string, signedData string) (string, error)
-
+	DeleteTask(ctx context.Context, taskID string) error
 	GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool)
-
-	SubscribeToEvents(eventType event.EventType, handler event.Handler)
-
-	SubscribeToAllEvents(handler event.Handler)
+	SubscribeToEvents(ctx context.Context, eventType event.EventType, handler event.Handler)
+	SubscribeToAllEvents(ctx context.Context, handler event.Handler)
 }
 
+// ClientImpl implements the Client interface
 type ClientImpl struct {
 	config      config.Config
 	taskManager task.Manager
 	logger      log.Logger
 	keyring     keyring.Keyring
 }
 
+// Verify interface compliance at compile time
+var _ Client = (*ClientImpl)(nil)
+
+// NewClient creates a new action client
 func NewClient(
+	ctx context.Context,
 	config config.Config,
 	logger log.Logger,
 	keyring keyring.Keyring,
@@ -38,7 +43,7 @@ func NewClient(
 		logger = log.NewNoopLogger()
 	}
 
-	taskManager, err := task.NewManager(config, logger, keyring)
+	taskManager, err := task.NewManager(ctx, config, logger, keyring)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create task manager: %w", err)
 	}
@@ -51,69 +56,89 @@ func NewClient(
 	}, nil
 }
 
-func (ac *ClientImpl) StartCascade(
+// StartCascade initiates a cascade operation
+func (c *ClientImpl) StartCascade(
 	ctx context.Context,
-	fileHash string,
-	actionID string,
-	filePath string,
-	signedData string,
+	fileHash string, // Hash of the file to process
+	actionID string, // ID of the action to perform
+	filePath string, // Path to the file on disk
+	signedData string, // Optional signed authorization data
 ) (string, error) {
-	ac.logger.Debug(ctx, "Starting cascade operation",
+	c.logger.Debug(ctx, "Starting cascade operation",
 		"fileHash", fileHash,
 		"actionID", actionID,
 		"filePath", filePath,
 	)
 
 	if fileHash == "" {
-		ac.logger.Error(ctx, "Empty file hash provided")
+		c.logger.Error(ctx, "Empty file hash provided")
 		return "", ErrEmptyFileHash
 	}
 	if actionID == "" {
-		ac.logger.Error(ctx, "Empty action ID provided")
+		c.logger.Error(ctx, "Empty action ID provided")
 		return "", ErrEmptyActionID
 	}
 	if filePath == "" {
-		ac.logger.Error(ctx, "Empty file path provided")
+		c.logger.Error(ctx, "Empty file path provided")
 		return "", ErrEmptyFilePath
 	}
 
-	taskID, err := ac.taskManager.CreateCascadeTask(ctx, fileHash, actionID, filePath, signedData)
+	taskID, err := c.taskManager.CreateCascadeTask(ctx, fileHash, actionID, filePath, signedData)
 	if err != nil {
-		ac.logger.Error(ctx, "Failed to create cascade task", "error", err)
+		c.logger.Error(ctx, "Failed to create cascade task", "error", err)
 		return "", fmt.Errorf("failed to create cascade task: %w", err)
 	}
 
-	ac.logger.Info(ctx, "Cascade task created successfully", "taskID", taskID)
+	c.logger.Info(ctx, "Cascade task created successfully", "taskID", taskID)
 	return taskID, nil
 }
 
-func (ac *ClientImpl) GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool) {
-	ac.logger.Debug(ctx, "Getting task", "taskID", taskID)
-	task, found := ac.taskManager.GetTask(ctx, taskID)
+// GetTask retrieves a task by its ID
+func (c *ClientImpl) GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool) {
+	c.logger.Debug(ctx, "Getting task", "taskID", taskID)
+	task, found := c.taskManager.GetTask(ctx, taskID)
 	if !found {
-		ac.logger.Debug(ctx, "Task not found", "taskID", taskID)
+		c.logger.Debug(ctx, "Task not found", "taskID", taskID)
 	} else {
-		ac.logger.Debug(ctx, "Task found", "taskID", taskID, "status", task.Status)
+		c.logger.Debug(ctx, "Task found", "taskID", taskID, "status", task.Status)
 	}
 	return task, found
 }
 
+// DeleteTask removes a task by its ID
+func (c *ClientImpl) DeleteTask(ctx context.Context, taskID string) error {
+	c.logger.Debug(ctx, "Deleting task", "taskID", taskID)
+	if taskID == "" {
+		c.logger.Error(ctx, "Empty task ID provided")
+		return fmt.Errorf("task ID cannot be empty")
+	}
+
+	err := c.taskManager.DeleteTask(ctx, taskID)
+	if err != nil {
+		c.logger.Error(ctx, "Failed to delete task", "taskID", taskID, "error", err)
+		return fmt.Errorf("failed to delete task: %w", err)
+	}
+
+	c.logger.Info(ctx, "Task deleted successfully", "taskID", taskID)
+	return nil
+}
+
 // SubscribeToEvents registers a handler for specific event types
-func (ac *ClientImpl) SubscribeToEvents(eventType event.EventType, handler event.Handler) {
-	ac.logger.Debug(context.Background(), "Subscribing to events via task manager", "eventType", eventType)
-	if ac.taskManager != nil {
-		ac.taskManager.SubscribeToEvents(eventType, handler)
+func (c *ClientImpl) SubscribeToEvents(ctx context.Context, eventType event.EventType, handler event.Handler) {
+	c.logger.Debug(ctx, "Subscribing to events via task manager", "eventType", eventType)
+	if c.taskManager != nil {
+		c.taskManager.SubscribeToEvents(ctx, eventType, handler)
 	} else {
-		ac.logger.Warn(context.Background(), "TaskManager is nil, cannot subscribe to events")
+		c.logger.Warn(ctx, "TaskManager is nil, cannot subscribe to events")
 	}
 }
 
 // SubscribeToAllEvents registers a handler for all events
-func (ac *ClientImpl) SubscribeToAllEvents(handler event.Handler) {
-	ac.logger.Debug(context.Background(), "Subscribing to all events via task manager")
-	if ac.taskManager != nil {
-		ac.taskManager.SubscribeToAllEvents(handler)
+func (c *ClientImpl) SubscribeToAllEvents(ctx context.Context, handler event.Handler) {
+	c.logger.Debug(ctx, "Subscribing to all events via task manager")
+	if c.taskManager != nil {
+		c.taskManager.SubscribeToAllEvents(ctx, handler)
 	} else {
-		ac.logger.Warn(context.Background(), "TaskManager is nil, cannot subscribe to all events")
+		c.logger.Warn(ctx, "TaskManager is nil, cannot subscribe to all events")
 	}
 }

sdk/adapters/lumera/adapter.go

@@ -41,6 +41,11 @@ func NewAdapter(
 		logger = log.NewNoopLogger()
 	}
 
+	logger.Debug(ctx, "Creating Lumera adapter",
+		"grpcAddr", config.GRPCAddr,
+		"chainID", config.ChainID,
+		"timeout", config.Timeout)
+
 	// Create client options from the config
 	options := []lumeraclient.Option{
 		lumeraclient.WithGRPCAddr(config.GRPCAddr),
@@ -61,9 +66,12 @@ func NewAdapter(
 	// Initialize the client
 	client, err := lumeraclient.NewClient(ctx, options...)
 	if err != nil {
+		logger.Error(ctx, "Failed to initialize Lumera client", "error", err)
 		return nil, fmt.Errorf("failed to initialize Lumera client: %w", err)
 	}
 
+	logger.Info(ctx, "Lumera adapter created successfully")
+
 	return &Adapter{
 		client: client,
 		logger: logger,

sdk/adapters/supernodeservice/adapter.go

@@ -14,11 +14,13 @@ type cascadeAdapter struct {
 	logger log.Logger
 }
 
-func NewCascadeAdapter(client cascade.CascadeServiceClient, logger log.Logger) CascadeServiceClient {
+func NewCascadeAdapter(ctx context.Context, client cascade.CascadeServiceClient, logger log.Logger) CascadeServiceClient {
 	if logger == nil {
 		logger = log.NewNoopLogger()
 	}
 
+	logger.Debug(ctx, "Creating cascade service adapter")
+
 	return &cascadeAdapter{
 		client: client,
 		logger: logger,

sdk/event/bus.go

@@ -8,7 +8,7 @@ import (
 )
 
 // Handler is a function that processes events
-type Handler func(Event)
+type Handler func(ctx context.Context, e Event)
 
 // Bus manages event subscriptions and dispatching
 type Bus struct {
@@ -21,7 +21,7 @@ type Bus struct {
 }
 
 // NewBus creates a new event bus
-func NewBus(logger log.Logger, maxWorkers int) *Bus {
+func NewBus(ctx context.Context, logger log.Logger, maxWorkers int) *Bus {
 	// Use default logger if nil is provided
 	if logger == nil {
 		logger = log.NewNoopLogger()
@@ -32,6 +32,8 @@ func NewBus(logger log.Logger, maxWorkers int) *Bus {
 		maxWorkers = 50
 	}
 
+	logger.Debug(ctx, "Creating new event bus", "maxWorkers", maxWorkers)
+
 	return &Bus{
 		subscribers: make(map[EventType][]Handler),
 		logger:      logger,
@@ -41,11 +43,10 @@ func NewBus(logger log.Logger, maxWorkers int) *Bus {
 }
 
 // Subscribe registers a handler for a specific event type
-func (b *Bus) Subscribe(eventType EventType, handler Handler) {
+func (b *Bus) Subscribe(ctx context.Context, eventType EventType, handler Handler) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	ctx := context.Background()
 	b.logger.Debug(ctx, "Subscribing handler to event type", "eventType", eventType)
 
 	if _, exists := b.subscribers[eventType]; !exists {
@@ -55,17 +56,16 @@ func (b *Bus) Subscribe(eventType EventType, handler Handler) {
 }
 
 // SubscribeAll registers a handler for all event types
-func (b *Bus) SubscribeAll(handler Handler) {
+func (b *Bus) SubscribeAll(ctx context.Context, handler Handler) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	ctx := context.Background()
 	b.logger.Debug(ctx, "Subscribing handler to all event types")
 	b.wildcardHandlers = append(b.wildcardHandlers, handler)
 }
 
 // safelyCallHandler executes a handler with proper panic recovery
-func (b *Bus) safelyCallHandler(handler Handler, event Event) {
+func (b *Bus) safelyCallHandler(ctx context.Context, handler Handler, event Event) {
 	// Acquire a worker slot
 	b.workerPool <- struct{}{}
 
@@ -77,7 +77,6 @@ func (b *Bus) safelyCallHandler(handler Handler, event Event) {
 			// Recover from panics
 			if r := recover(); r != nil {
 				stackTrace := debug.Stack()
-				ctx := context.Background()
 				b.logger.Error(ctx,
 					"Event handler panicked",
 					"error", r,
@@ -90,8 +89,8 @@ func (b *Bus) safelyCallHandler(handler Handler, event Event) {
 		// Create a deep copy of the event to avoid race conditions
 		eventCopy := copyEvent(event)
 
-		// Execute the handler
-		handler(eventCopy)
+		// Execute the handler with the provided context
+		handler(ctx, eventCopy)
 	}()
 }
 
@@ -115,11 +114,10 @@ func copyEvent(e Event) Event {
 }
 
 // Publish sends an event to all relevant subscribers
-func (b *Bus) Publish(event Event) {
+func (b *Bus) Publish(ctx context.Context, event Event) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
-	ctx := context.Background()
 	b.logger.Debug(ctx, "Publishing event",
 		"type", event.Type,
 		"taskID", event.TaskID,
@@ -132,7 +130,7 @@ func (b *Bus) Publish(event Event) {
 			"handlerCount", len(handlers))
 
 		for _, handler := range handlers {
-			b.safelyCallHandler(handler, event)
+			b.safelyCallHandler(ctx, handler, event)
 		}
 	}
 
@@ -142,13 +140,15 @@ func (b *Bus) Publish(event Event) {
 			"handlerCount", len(b.wildcardHandlers))
 
 		for _, handler := range b.wildcardHandlers {
-			b.safelyCallHandler(handler, event)
+			b.safelyCallHandler(ctx, handler, event)
 		}
 	}
 }
 
 // WaitForHandlers waits for all event handlers to complete
-func (b *Bus) WaitForHandlers() {
+func (b *Bus) WaitForHandlers(ctx context.Context) {
+	b.logger.Debug(ctx, "Waiting for all handlers to complete")
+
 	// Fill the worker pool to capacity (blocks until all workers are free)
 	for i := 0; i < b.maxWorkers; i++ {
 		b.workerPool <- struct{}{}
@@ -161,6 +161,7 @@ func (b *Bus) WaitForHandlers() {
 }
 
 // Close releases resources used by the event bus
-func (b *Bus) Close() {
-	b.WaitForHandlers()
+func (b *Bus) Close(ctx context.Context) {
+	b.logger.Debug(ctx, "Closing event bus")
+	b.WaitForHandlers(ctx)
 }

sdk/event/types.go

@@ -2,6 +2,7 @@ package event
 
 import (
 	"action/adapters/lumera"
+	"context"
 	"time"
 )
 
@@ -41,7 +42,7 @@ type SupernodeData struct {
 	Error     string           // Error message if applicable
 }
 
-func NewEvent(eventType EventType, taskID, taskType string, data map[string]interface{}) Event {
+func NewEvent(ctx context.Context, eventType EventType, taskID, taskType string, data map[string]interface{}) Event {
 	if data == nil {
 		data = make(map[string]interface{})
 	}

sdk/net/factory.go

@@ -27,10 +27,19 @@ type ClientFactory struct {
 
 // NewClientFactory creates a new client factory with the provided dependencies
 func NewClientFactory(
+	ctx context.Context,
 	logger log.Logger,
 	keyring keyring.Keyring,
 	config FactoryConfig,
 ) *ClientFactory {
+	if logger == nil {
+		logger = log.NewNoopLogger()
+	}
+
+	logger.Debug(ctx, "Creating supernode client factory",
+		"localAddress", config.LocalCosmosAddress,
+		"defaultPort", config.DefaultSupernodePort)
+
 	return &ClientFactory{
 		logger:        logger,
 		keyring:       keyring,

sdk/net/impl.go

@@ -17,20 +17,23 @@ import (
 	"google.golang.org/grpc/health/grpc_health_v1"
 )
 
+// supernodeClient implements the SupernodeClient interface
 type supernodeClient struct {
 	cascadeClient supernodeservice.CascadeServiceClient
 	healthClient  grpc_health_v1.HealthClient
 	conn          *grpc.ClientConn
 	logger        log.Logger
 }
 
+// Verify interface compliance at compile time
+var _ SupernodeClient = (*supernodeClient)(nil)
+
 // NewSupernodeClient creates a new supernode client
 func NewSupernodeClient(
 	ctx context.Context,
 	logger log.Logger,
 	keyring keyring.Keyring,
 	localCosmosAddress string,
-
 	targetSupernode lumera.Supernode,
 	clientOptions *client.ClientOptions,
 ) (SupernodeClient, error) {
@@ -80,6 +83,7 @@ func NewSupernodeClient(
 
 	// Create service clients
 	cascadeClient := supernodeservice.NewCascadeAdapter(
+		ctx,
 		cascade.NewCascadeServiceClient(conn),
 		logger,
 	)