feat: Implement Publisher

shared/plugin/definition/publisher.go

@@ -0,0 +1,16 @@
+package definition
+
+import "context"
+
+// Publisher defines the general publisher interface for messaging plugins.
+type Publisher interface {
+	// Publish sends a message (as a byte slice) using the underlying messaging system.
+	Publish(ctx context.Context, msg []byte) error
+
+	Close() error // Important for releasing resources
+}
+
+type PublisherProvider interface {
+	// New initializes a new publisher instance with the given configuration
+	New(ctx context.Context, config map[string]string) (Publisher, error)
+}

shared/plugin/implementation/publisher/cmd/plugin.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/beckn/beckn-onix/shared/plugin/definition"
+	"github.com/beckn/beckn-onix/shared/plugin/implementation/publisher"
+)
+
+// Returns error if required fields are missing.
+func validateConfig(config map[string]string) (*publisher.Config, error) {
+	if config == nil {
+		return nil, fmt.Errorf("config cannot be nil")
+	}
+
+	project, ok := config["project"]
+	if !ok || project == "" {
+		return nil, fmt.Errorf("project ID is required")
+	}
+
+	topic, ok := config["topic"]
+	if !ok || topic == "" {
+		return nil, fmt.Errorf("topic ID is required")
+	}
+
+	return &publisher.Config{
+		ProjectID: project,
+		TopicID:   topic,
+	}, nil
+}
+
+// PublisherProvider implements the definition.PublisherProvider interface.
+type PublisherProvider struct{}
+
+// New creates a new Publisher instance.
+func (p PublisherProvider) New(ctx context.Context, config map[string]string) (definition.Publisher, error) {
+	if ctx == nil {
+		return nil, fmt.Errorf("context cannot be nil")
+	}
+
+	cfg, err := validateConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("invalid config: %w", err)
+	}
+
+	return publisher.New(ctx, cfg)
+}
+
+// Provider is the exported symbol that the plugin manager will look for.
+var Provider definition.PublisherProvider = PublisherProvider{}

shared/plugin/implementation/publisher/cmd/plugin_test.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/beckn/beckn-onix/shared/plugin/definition"
+)
+
+// MockPublisher is a mock implementation of the definition.Publisher interface for testing.
+type MockPublisher struct{}
+
+func (m *MockPublisher) Publish(ctx context.Context, msg []byte) error {
+	return nil
+}
+
+func (m *MockPublisher) Close() error {
+	return nil
+}
+
+// TestValidateConfig tests the validateConfig function.
+func TestValidateConfig(t *testing.T) {
+	tests := []struct {
+		name    string
+		config  map[string]string
+		wantErr bool
+	}{
+		{"Valid config", map[string]string{"project": "test-project", "topic": "test-topic"}, false},
+		{"Nil config", nil, true},
+		{"Missing project", map[string]string{"topic": "test-topic"}, true},
+		{"Missing topic", map[string]string{"project": "test-project"}, true},
+		{"Empty project", map[string]string{"project": "", "topic": "test-topic"}, true},
+		{"Empty topic", map[string]string{"project": "test-project", "topic": ""}, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := validateConfig(tt.config)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("validateConfig() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+// TestPublisherProviderNew tests the New method of PublisherProvider.
+func TestPublisherProviderNew(t *testing.T) {
+	tests := []struct {
+		name    string
+		ctx     context.Context
+		config  map[string]string
+		wantErr bool
+	}{
+		{"Nil context", nil, map[string]string{"project": "test-project", "topic": "test-topic"}, true},
+		{"Invalid config", context.Background(), map[string]string{"project": "", "topic": "test-topic"}, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			provider := PublisherProvider{}
+			_, err := provider.New(tt.ctx, tt.config)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("PublisherProvider.New() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+// TestProviderImplementation verifies that the Provider is correctly initialized.
+func TestProviderImplementation(t *testing.T) {
+	if _, ok := interface{}(Provider).(definition.PublisherProvider); !ok {
+		t.Errorf("Provider does not implement definition.PublisherProvider")
+	}
+}

shared/plugin/implementation/publisher/publisher.go

@@ -0,0 +1,95 @@
+package publisher
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
+
+	"cloud.google.com/go/pubsub"
+	"google.golang.org/api/option"
+)
+
+// Config holds the Pub/Sub configuration.
+type Config struct {
+	ProjectID string
+	TopicID   string
+}
+
+// Publisher is a concrete implementation of a Google Cloud Pub/Sub publisher.
+type Publisher struct {
+	client *pubsub.Client
+	topic  *pubsub.Topic
+	config *Config
+}
+
+var (
+	ErrProjectMissing = errors.New("missing required field 'Project'")
+	ErrTopicMissing   = errors.New("missing required field 'Topic'")
+	ErrEmptyConfig    = errors.New("empty config")
+)
+
+
+func validate(cfg *Config) error {
+	if cfg == nil {
+		return ErrEmptyConfig
+	}
+	if strings.TrimSpace(cfg.ProjectID) == "" {
+		return ErrProjectMissing
+	}
+	if strings.TrimSpace(cfg.TopicID) == "" {
+		return ErrTopicMissing
+	}
+	return nil
+}
+
+// New initializes a new Publisher instance.
+// It creates a real pubsub.Client and then calls NewWithClient.
+func New(ctx context.Context, config *Config, opts ...option.ClientOption) (*Publisher, error) {
+	if err := validate(config); err != nil {
+		return nil, err
+	}
+	client, err := pubsub.NewClient(ctx, config.ProjectID, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create pubsub client: %w", err)
+	}
+
+	topic := client.Topic(config.TopicID)
+	exists, err := topic.Exists(ctx)
+	if err != nil {
+		_ = client.Close()
+		return nil, fmt.Errorf("failed to check topic existence: %w", err)
+	}
+	if !exists {
+		_ = client.Close()
+		return nil, fmt.Errorf("topic %s does not exist", config.TopicID)
+	}
+	return &Publisher{
+		client: client,
+		topic:  topic,
+		config: config,
+	}, nil
+}
+
+
+// Publish sends a message to Google Cloud Pub/Sub.
+func (p *Publisher) Publish(ctx context.Context, msg []byte) error {
+	pubsubMsg := &pubsub.Message{
+		Data: msg,
+	}
+
+	result := p.topic.Publish(ctx, pubsubMsg)
+	id, err := result.Get(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to publish message: %w", err)
+	}
+
+	log.Printf("Published message with ID: %s\n", id)
+	return nil
+}
+
+// Close closes the underlying Pub/Sub client.
+func (p *Publisher) Close() error {
+	return p.client.Close()
+}

shared/plugin/implementation/publisher/publisher_test.go

@@ -0,0 +1,123 @@
+package publisher
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"cloud.google.com/go/pubsub"
+)
+
+// Helper to create a test publisher directly
+func createTestPublisher() *Publisher {
+	client, _ := pubsub.NewClient(context.Background(), "test-project")
+	topic := client.Topic("test-topic")
+	return &Publisher{
+		client: client,
+		topic:  topic,
+		config: &Config{ProjectID: "test-project", TopicID: "test-topic"},
+	}
+}
+
+// TestValidate tests the validate function
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name    string
+		config  *Config
+		wantErr error
+	}{
+		{
+			name:    "Valid config",
+			config:  &Config{ProjectID: "test-project", TopicID: "test-topic"},
+			wantErr: nil,
+		},
+		{
+			name:    "Nil config",
+			config:  nil,
+			wantErr: ErrEmptyConfig,
+		},
+		{
+			name:    "Empty project ID",
+			config:  &Config{ProjectID: "", TopicID: "test-topic"},
+			wantErr: ErrProjectMissing,
+		},
+		{
+			name:    "Whitespace project ID",
+			config:  &Config{ProjectID: "   ", TopicID: "test-topic"},
+			wantErr: ErrProjectMissing,
+		},
+		{
+			name:    "Empty topic ID",
+			config:  &Config{ProjectID: "test-project", TopicID: ""},
+			wantErr: ErrTopicMissing,
+		},
+		{
+			name:    "Whitespace topic ID",
+			config:  &Config{ProjectID: "test-project", TopicID: "  "},
+			wantErr: ErrTopicMissing,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validate(tt.config)
+			if !errors.Is(err, tt.wantErr) {
+				t.Errorf("validate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+// TestNew tests the New function with validation errors only
+// We can't easily test the pubsub client creation parts without complex mocks
+func TestNew(t *testing.T) {
+	tests := []struct {
+		name    string
+		ctx     context.Context
+		config  *Config
+		wantErr bool
+	}{
+		{
+			// Should fail validation
+			name:    "Empty project ID",
+			ctx:     context.Background(),
+			config:  &Config{ProjectID: "", TopicID: "test-topic"},
+			wantErr: true,
+		},
+		{
+			// Should fail validation
+			name:    "Empty topic ID",
+			ctx:     context.Background(),
+			config:  &Config{ProjectID: "test-project", TopicID: ""},
+			wantErr: true,
+		},
+		{
+			// Should fail due to nil context
+			name:    "Nil context",
+			ctx:     nil,
+			config:  &Config{ProjectID: "test-project", TopicID: "test-topic"},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := New(tt.ctx, tt.config)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("New() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+// TestPublish tests the behavior of the Publisher.Publish method
+// Since we can't easily mock the pubsub client and topic, we'll skip the actual test
+func TestPublish(t *testing.T) {
+	t.Skip("Requires real pubsub client or complex mocking")
+}
+
+// TestClose tests the behavior of the Publisher.Close method
+// Since we can't easily mock the pubsub client, we'll skip the actual test
+func TestClose(t *testing.T) {
+	t.Skip("Requires real pubsub client or complex mocking")
+}

shared/plugin/manager.go

@@ -15,6 +15,7 @@ type Config struct {
 	Root     string       `yaml:"root"`
 	Signer   PluginConfig `yaml:"signer"`
 	Verifier PluginConfig `yaml:"verifier"`
+	Publisher PluginConfig `yaml:"publisher"`
 }
 
 // PluginConfig represents configuration details for a plugin.
@@ -27,6 +28,7 @@ type PluginConfig struct {
 type Manager struct {
 	sp  definition.SignerProvider
 	vp  definition.VerifierProvider
+	pb  definition.PublisherProvider
 	cfg *Config
 }
 
@@ -42,13 +44,19 @@ func NewManager(ctx context.Context, cfg *Config) (*Manager, error) {
 		return nil, fmt.Errorf("failed to load signer plugin: %w", err)
 	}
 
+	// Load publisher plugin
+	pb, err := provider[definition.PublisherProvider](cfg.Root, cfg.Publisher.ID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load publisher plugin: %w", err)
+	}
+
 	// Load verifier plugin
 	vp, err := provider[definition.VerifierProvider](cfg.Root, cfg.Verifier.ID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to load Verifier plugin: %w", err)
 	}
 
-	return &Manager{sp: sp, vp: vp, cfg: cfg}, nil
+	return &Manager{sp: sp, vp: vp, pb: pb, cfg: cfg}, nil
 }
 
 // provider loads a plugin dynamically and retrieves its provider instance.
@@ -106,3 +114,16 @@ func (m *Manager) Verifier(ctx context.Context) (definition.Verifier, func() err
 	}
 	return Verifier, close, nil
 }
+
+// Publisher retrieves the publisher plugin instance.
+func (m *Manager) Publisher(ctx context.Context) (definition.Publisher, error) {
+	if m.pb == nil {
+		return nil, fmt.Errorf("publisher plugin provider not loaded")
+	}
+
+	publisher,  err := m.pb.New(ctx, m.cfg.Publisher.Config)
+	if err != nil {
+		return nil,  fmt.Errorf("failed to initialize publisher: %w", err)
+	}
+	return publisher,  nil
+}