Kruto

Satont · Dec 23, 2023 · d1a63da · d1a63da
1 parent a11a828
commit d1a63da
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 6 deletions.
diff --git a/internal/messagesender/temporal/impl.go b/internal/messagesender/temporal/impl.go
@@ -2,19 +2,66 @@ package temporal
 
 import (
 	"context"
+	"fmt"
+	"time"
 
+	"github.com/google/uuid"
 	"github.com/satont/twitch-notifier/internal/messagesender"
+	"github.com/satont/twitch-notifier/pkg/logger"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
+	"go.uber.org/fx"
 )
 
-func NewTemporal() *Temporal {
-	return &Temporal{}
+type TemporalOpts struct {
+	fx.In
+
+	Workflow *Workflow
+	Logger   logger.Logger
+}
+
+func NewTemporal(opts TemporalOpts) (*Temporal, error) {
+	cl, err := client.Dial(client.Options{})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Temporal{
+		client:   cl,
+		workflow: opts.Workflow,
+		logger:   opts.Logger,
+	}, nil
 }
 
+const queueName = "message-sender"
+
 type Temporal struct {
+	client   client.Client
+	workflow *Workflow
+	logger   logger.Logger
 }
 
 var _ messagesender.MessageSender = (*Temporal)(nil)
 
-func (m *Temporal) SendMessageTelegram(ctx context.Context, opts messagesender.TelegramOpts) error {
+func (c *Temporal) SendMessageTelegram(ctx context.Context, opts messagesender.TelegramOpts) error {
+	workflowOptions := client.StartWorkflowOptions{
+		ID:        fmt.Sprintf("MSG: Telegram to %s #%s", uuid.NewString(), opts.ServiceChatID),
+		TaskQueue: queueName,
+		RetryPolicy: &temporal.RetryPolicy{
+			MaximumAttempts: 5,
+			InitialInterval: 10 * time.Second,
+		},
+	}
+
+	we, err := c.client.ExecuteWorkflow(ctx, workflowOptions, c.workflow.SendTelegram, thumbnailUrl)
+	if err != nil {
+		return err
+	}
+
+	err = we.Get(ctx, nil)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
diff --git a/internal/messagesender/temporal/workflow.go b/internal/messagesender/temporal/workflow.go
@@ -2,12 +2,15 @@ package temporal
 
 import (
 	"github.com/satont/twitch-notifier/internal/messagesender"
+	"github.com/satont/twitch-notifier/pkg/logger"
 	"go.temporal.io/sdk/workflow"
 	"go.uber.org/fx"
 )
 
 type WorkflowOpts struct {
 	fx.In
+
+	Logger logger.Logger
 }
 
 func NewWorkflow(opts WorkflowOpts) *Workflow {
@@ -17,4 +20,6 @@ func NewWorkflow(opts WorkflowOpts) *Workflow {
 type Workflow struct {
 }
 
-func (c *Workflow) SendTelegram(ctx workflow.Context, opts messagesender.TelegramOpts)
+func (c *Workflow) SendTelegram(ctx workflow.Context, opts messagesender.TelegramOpts) error {
+	return nil
+}
diff --git a/internal/thumbnailchecker/temporal/impl.go b/internal/thumbnailchecker/temporal/impl.go
@@ -61,7 +61,7 @@ func (c *Temporal) ValidateThumbnail(ctx context.Context, thumbnailUrl string) e
 		return err
 	}
 
-	err = we.Get(context.Background(), nil)
+	err = we.Get(ctx, nil)
 	if err != nil {
 		return err
 	}

diff --git a/internal/thumbnailchecker/temporal/worker.go b/internal/thumbnailchecker/temporal/worker.go
@@ -20,7 +20,6 @@ type WorkerOpts struct {
 }
 
 func NewWorker(opts WorkerOpts) error {
-	// The client and worker are heavyweight objects that should be created once per process.
 	temporalClient, err := client.Dial(
 		client.Options{
 			Logger: log.NewStructuredLogger(opts.Logger.GetSlog()),