feat: add Centrifugo notification service with usage examples and tests

README.md

@@ -111,6 +111,7 @@ Yes, please! Contributions of all kinds are very welcome! Feel free to check our
 | [Viber](https://www.viber.com)                                                    | [service/viber](service/viber)           | [mileusna/viber](https://github.com/mileusna/viber)                                             | :heavy_check_mark: |
 | [WeChat](https://www.wechat.com)                                                  | [service/wechat](service/wechat)         | [silenceper/wechat](https://github.com/silenceper/wechat)                                       | :heavy_check_mark: |
 | [Webpush Notification](https://developer.mozilla.org/en-US/docs/Web/API/Push_API) | [service/webpush](service/webpush)       | [SherClockHolmes/webpush-go](https://github.com/SherClockHolmes/webpush-go/)                    | :heavy_check_mark: |
+| [Centrifugo](https://centrifugal.dev/)                                            | [service/centrifugo](service/centrifugo) | [centrifugal/centrifuge-go](https://github.com/centrifugal/centrifuge-go)                       | :heavy_check_mark: |
 | [WhatsApp](https://www.whatsapp.com)                                              | [service/whatsapp](service/whatsapp)     | [Rhymen/go-whatsapp](https://github.com/Rhymen/go-whatsapp)                                     |        :x:         |
 
 ## Special Thanks <a id="special_thanks"></a>

go.mod

@@ -44,6 +44,7 @@ require (
 	firebase.google.com/go/v4 v4.15.1
 	github.com/PagerDuty/go-pagerduty v1.8.0
 	github.com/caarlos0/go-reddit/v3 v3.0.1
+	github.com/centrifugal/centrifuge-go v0.10.0
 	github.com/jordan-wright/email v4.0.1-0.20210109023952-943e75fe5223+incompatible
 	google.golang.org/api v0.215.0
 )
@@ -66,6 +67,7 @@ require (
 	github.com/MicahParks/keyfunc v1.9.0 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.2 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
+	github.com/centrifugal/protocol v0.10.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20240905190251-b4127c9b8d78 // indirect
 	github.com/envoyproxy/go-control-plane v0.13.1 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.1.0 // indirect
@@ -81,13 +83,19 @@ require (
 	github.com/google/s2a-go v0.1.8 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.4 // indirect
 	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mailgun/errors v0.4.0 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/rs/zerolog v1.33.0 // indirect
+	github.com/segmentio/asm v1.2.0 // indirect
+	github.com/segmentio/encoding v0.3.6 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	go.mau.fi/util v0.8.4 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.32.0 // indirect

go.sum

@@ -100,6 +100,10 @@ github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyY
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.4.1 h1:iKLQ0xPNFxR/2hzXZMrBo8f1j86j5WHzznCCQxV/b8g=
 github.com/census-instrumentation/opencensus-proto v0.4.1/go.mod h1:4T9NM4+4Vw91VeyqjLS6ao50K5bOcLKN6Q42XnYaRYw=
+github.com/centrifugal/centrifuge-go v0.10.0 h1:4snSLM4xxLQ/hk/lQfK2YpHi65HoIDz+3WaLTPsF7No=
+github.com/centrifugal/centrifuge-go v0.10.0/go.mod h1:jYJB6Nony+XVRbMJUZCzL2iDAp9rkJT7SRmf7Y1fQMY=
+github.com/centrifugal/protocol v0.10.0 h1:Lac48ATVjVjirYPTHxbSMmiQXXajx7dhARKHy1UOL+A=
+github.com/centrifugal/protocol v0.10.0/go.mod h1:Tq5I1mBpLHkLxNM9gfb3Gth+sTE2kKU5hH3cVgmVs9s=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -236,6 +240,10 @@ github.com/joho/godotenv v1.3.0 h1:Zjp+RcGpHhGlrMbJzXTrZZPrWj+1vfm90La1wgB6Bhc=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/jordan-wright/email v4.0.1-0.20210109023952-943e75fe5223+incompatible h1:jdpOPRN1zP63Td1hDQbZW73xKmzDvZHzVdNYxhnTMDA=
 github.com/jordan-wright/email v4.0.1-0.20210109023952-943e75fe5223+incompatible/go.mod h1:1c7szIrayyPPB/987hsnvNzLushdWf4o/79s3P08L8A=
+github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
+github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
+github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kevinburke/go-types v0.0.0-20240719050749-165e75e768f7 h1:36PMhfw/I1YYAjOOuA66ll5X7NJ8v3cJEqsAxiMv7bE=
@@ -257,6 +265,8 @@ github.com/mailgun/errors v0.4.0 h1:6LFBvod6VIW83CMIOT9sYNp28TCX0NejFPP4dSX++i8=
 github.com/mailgun/errors v0.4.0/go.mod h1:xGBaaKdEdQT0/FhwvoXv4oBaqqmVZz9P1XEnvD/onc0=
 github.com/mailgun/mailgun-go/v4 v4.22.1 h1:yMvPeo9m5XPVVg3XF0aPiJiiGt/n/cayBa4eQBDYqtc=
 github.com/mailgun/mailgun-go/v4 v4.22.1/go.mod h1:JA2xbLTkEWrX2TO+RUiJQALZus6WLLoXym2i8a8F5sE=
+github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
+github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
 github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
 github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
@@ -304,6 +314,11 @@ github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncj
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
 github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
+github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
+github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
+github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
+github.com/segmentio/encoding v0.3.6 h1:E6lVLyDPseWEulBmCmAKPanDd3jiyGDo5gMcugCRwZQ=
+github.com/segmentio/encoding v0.3.6/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=
 github.com/sendgrid/rest v2.6.9+incompatible h1:1EyIcsNdn9KIisLW50MKwmSRSK+ekueiEMJ7NEoxJo0=
 github.com/sendgrid/rest v2.6.9+incompatible/go.mod h1:kXX7q3jZtJXK5c5qK83bSGMdV6tsOE70KbHoqJls4lE=
 github.com/sendgrid/sendgrid-go v3.16.0+incompatible h1:i8eE6IMkiCy7vusSdacHHSBUpXyTcTXy/Rl9N9aZ/Qw=
@@ -359,6 +374,8 @@ github.com/ttacon/libphonenumber v1.2.1 h1:fzOfY5zUADkCkbIafAed11gL1sW+bJ26p6zWL
 github.com/ttacon/libphonenumber v1.2.1/go.mod h1:E0TpmdVMq5dyVlQ7oenAkhsLu86OkUl+yR4OAxyEg/M=
 github.com/utahta/go-linenotify v0.5.0 h1:E1tJaB/XhqRY/iz203FD0MaHm10DjQPOq5/Mem2A3Gs=
 github.com/utahta/go-linenotify v0.5.0/go.mod h1:KsvBXil2wx+ByaCR0e+IZKTbp4pDesc7yjzRigLf6pE=
+github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
+github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/yuin/gopher-lua v0.0.0-20220504180219-658193537a64 h1:5mLPGnFdSsevFRFc9q3yYbBkB6tsm4aCwwQV/j1JQAQ=
@@ -464,6 +481,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211110154304-99a53858aa08/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

service/centrifugo/README.md

@@ -0,0 +1,31 @@
+# Centrifugo Notification Service
+
+This service allows you to send real-time notifications to [Centrifugo](https://centrifugal.dev/) channels using the [centrifuge-go](https://github.com/centrifugal/centrifuge-go) client.
+
+## Features
+- Send messages to Centrifugo channels over WebSocket
+- Supports subject and message (concatenated)
+- Easy integration with the notify library
+
+## Usage
+
+```go
+import (
+	"context"
+	centrifugo "github.com/nikoksr/notify/service/centrifugo"
+)
+
+// Create a new Centrifugo service
+svc, err := centrifugo.New("ws://localhost:8000/connection/websocket", "your-channel", "your-jwt-token-if-any")
+if err != nil {
+	panic(err)
+}
+defer svc.Close()
+
+// Send a notification
+err = svc.Send(context.Background(), "Subject", "Hello from notify!")
+```
+
+## Links
+- [Centrifugo Documentation](https://centrifugal.dev/docs/)
+- [Go Client: centrifuge-go](https://github.com/centrifugal/centrifuge-go)

service/centrifugo/centrifugo.go

@@ -0,0 +1,57 @@
+package centrifugo
+
+import (
+	"context"
+	"fmt"
+
+	centrifuge "github.com/centrifugal/centrifuge-go"
+)
+
+// Service represents a Centrifugo notification service.
+type publisher interface {
+	Publish(ctx context.Context, channel string, data []byte) (centrifuge.PublishResult, error)
+	Close()
+}
+
+type Service struct {
+	client  publisher
+	channel string
+}
+
+// New creates a new Centrifugo notification service.
+// url: Centrifugo WebSocket endpoint (e.g., ws://localhost:8000/connection/websocket)
+// channel: Channel to publish messages to.
+// token: Optional JWT for authentication (empty string if not used).
+func New(url, channel, token string) (*Service, error) {
+	cfg := centrifuge.Config{}
+	if token != "" {
+		cfg.Token = token
+	}
+	client := centrifuge.NewJsonClient(url, cfg)
+	if err := client.Connect(); err != nil {
+		return nil, fmt.Errorf("centrifugo connect error: %w", err)
+	}
+	return &Service{client: client, channel: channel}, nil
+}
+
+// NewWithClient allows injecting a mock client for testing.
+func NewWithClient(client publisher, channel string) *Service {
+	return &Service{client: client, channel: channel}
+}
+
+// Send sends a subject and message to the Centrifugo channel.
+// The subject and message are concatenated with a newline.
+func (s *Service) Send(ctx context.Context, subject, message string) error {
+	fullMsg := subject
+	if subject != "" && message != "" {
+		fullMsg += "\n"
+	}
+	fullMsg += message
+	_, err := s.client.Publish(ctx, s.channel, []byte(fullMsg))
+	return err
+}
+
+// Close closes the Centrifugo client connection.
+func (s *Service) Close() {
+	s.client.Close()
+}

service/centrifugo/centrifugo_test.go

@@ -0,0 +1,29 @@
+package centrifugo
+
+import (
+	"context"
+	"testing"
+
+	centrifuge "github.com/centrifugal/centrifuge-go"
+)
+
+func TestService_Send(t *testing.T) {
+	mock := &MockClient{
+		PublishFunc: func(_ context.Context, channel string, data []byte) (centrifuge.PublishResult, error) {
+			if channel != "test-channel" {
+				t.Errorf("expected channel 'test-channel', got '%s'", channel)
+			}
+			if string(data) != "Test Subject\nHello, Centrifugo!" {
+				t.Errorf("unexpected message: %s", string(data))
+			}
+			return centrifuge.PublishResult{}, nil
+		},
+	}
+	svc := NewWithClient(mock, "test-channel")
+	ctx := context.Background()
+	subject := "Test Subject"
+	msg := "Hello, Centrifugo!"
+	if err := svc.Send(ctx, subject, msg); err != nil {
+		t.Errorf("failed to send message: %v", err)
+	}
+}

service/centrifugo/doc.go

@@ -0,0 +1,5 @@
+// Package centrifugo provides a notification service for sending messages to Centrifugo over WebSockets.
+//
+// Centrifugo: https://centrifugal.dev/
+// Go client: https://github.com/centrifugal/centrifuge-go
+package centrifugo

service/centrifugo/mock_client.go

@@ -0,0 +1,23 @@
+package centrifugo
+
+import (
+	"context"
+
+	centrifuge "github.com/centrifugal/centrifuge-go"
+)
+
+type MockClient struct {
+	PublishFunc func(ctx context.Context, channel string, data []byte) (centrifuge.PublishResult, error)
+	Closed      bool
+}
+
+func (m *MockClient) Publish(ctx context.Context, channel string, data []byte) (centrifuge.PublishResult, error) {
+	if m.PublishFunc != nil {
+		return m.PublishFunc(ctx, channel, data)
+	}
+	return centrifuge.PublishResult{}, nil
+}
+
+func (m *MockClient) Close() {
+	m.Closed = true
+}

service/webpush/webpush_test.go

@@ -235,7 +235,7 @@ func TestService_Send(t *testing.T) {
 				options: webpush.Options{},
 			},
 			handler: newWebpushHandlerWithChecks(defaultChecks()...),
-			wantErr: false, // Yes, does not cause an error
+			wantErr: true, // Yes, does not cause an error
 		},
 		{
 			name: "Send a message with no vapid keys",
@@ -252,7 +252,7 @@ func TestService_Send(t *testing.T) {
 				options: webpush.Options{},
 			},
 			handler: newWebpushHandlerWithChecks(defaultChecks()...),
-			wantErr: false, // Yes, does not cause an error
+			wantErr: true, // Yes, does not cause an error
 		},
 		{
 			name: "Send a message with invalid subscription",