rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add interactif slack message

Browse Source 
Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
This commit is contained in:
Sacha Al Himdani
2025-10-20 17:22:02 +02:00
parent de004ce8d7
commit 0073846d3b
23 changed files with 1833 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
pkg/slack/sender.go
View File

@@ -18,13 +18,11 @@ import (
"context"
"errors"
"fmt"
"net/http"
"time"
"github.com/getprobo/probo/pkg/connector"
"github.com/getprobo/probo/pkg/coredata"
"github.com/getprobo/probo/pkg/crypto/cipher"
"go.gearno.de/kit/httpclient"
"go.gearno.de/kit/log"
"go.gearno.de/kit/pg"
)
@@ -35,7 +33,6 @@ type (
logger *log.Logger
encryptionKey cipher.EncryptionKey
interval time.Duration
httpClient *http.Client
}
Config struct {
@@ -44,18 +41,11 @@ type (
)
func NewSender(pg *pg.Client, logger *log.Logger, encryptionKey cipher.EncryptionKey, cfg Config) *Sender {
httpClientOpts := []httpclient.Option{
httpclient.WithLogger(logger),
}
httpClient := httpclient.DefaultPooledClient(httpClientOpts...)
return &Sender{
pg: pg,
logger: logger,
encryptionKey: encryptionKey,
interval: cfg.Interval,
httpClient: httpClient,
}
}
@@ -70,6 +60,10 @@ LOOP:
s.logger.ErrorCtx(ctx, "cannot send slack message", log.Error(err))
}
if err := s.batchUpdateMessages(ctx); err != nil {
s.logger.ErrorCtx(ctx, "cannot update slack message", log.Error(err))
}
goto LOOP
}
}
@@ -101,7 +95,14 @@ func (s *Sender) batchSendMessages(ctx context.Context) error {
return err
}
if sendErr := s.sendMessage(ctx, tx, message); sendErr != nil {
channelID, messageTS, sendErr := s.sendMessage(ctx, tx, message)
message.ChannelID = channelID
message.MessageTS = messageTS
now := time.Now()
message.UpdatedAt = now
if sendErr != nil {
errorMsg := sendErr.Error()
message.Error = &errorMsg
message.UpdatedAt = time.Now()
@@ -114,9 +115,7 @@ func (s *Sender) batchSendMessages(ctx context.Context) error {
return nil
}
now := time.Now()
message.SentAt = &now
message.UpdatedAt = now
if err := message.Update(ctx, tx); err != nil {
return fmt.Errorf("cannot update slack message: %w", err)
@@ -136,7 +135,137 @@ func (s *Sender) batchSendMessages(ctx context.Context) error {
}
}
func (s *Sender) sendMessage(ctx context.Context, tx pg.Conn, message *coredata.SlackMessage) error {
func (s *Sender) sendMessage(ctx context.Context, tx pg.Conn, message *coredata.SlackMessage) (*string, *string, error) {
tenantID := message.ID.TenantID()
scope := coredata.NewScope(tenantID)
var connectors coredata.Connectors
if err := connectors.LoadAllByOrganizationIDProtocolAndProvider(
ctx,
tx,
scope,
message.OrganizationID,
coredata.ConnectorProtocolOAuth2,
coredata.ConnectorProviderSlack,
s.encryptionKey,
); err != nil {
return nil, nil, fmt.Errorf("cannot load slack connectors: %w", err)
}
if len(connectors) == 0 {
return nil, nil, fmt.Errorf("no slack connectors configured for organization")
}
c := connectors[0]
if c.Connection == nil {
return nil, nil, fmt.Errorf("slack connector has nil connection")
}
slackConn, ok := c.Connection.(*connector.SlackConnection)
if !ok {
return nil, nil, fmt.Errorf("slack connector must have SlackConnection type, got %T", c.Connection)
}
if slackConn.Settings.ChannelID == "" {
return nil, nil, fmt.Errorf("slack connector %s has no channel ID", c.ID)
}
if slackConn.AccessToken == "" {
return nil, nil, fmt.Errorf("slack connector %s has no access token", c.ID)
}
client := NewClient(s.logger)
if message.Type == coredata.SlackMessageTypeWelcome {
if err := client.JoinChannel(ctx, slackConn.AccessToken, slackConn.Settings.ChannelID); err != nil {
s.logger.ErrorCtx(ctx, "failed to join Slack channel", log.Error(err))
}
}
slackResp, err := client.CreateMessage(ctx, slackConn.AccessToken, slackConn.Settings.ChannelID, message.Body)
if err != nil {
s.logger.ErrorCtx(ctx, "failed to post message to Slack", log.Error(err))
return nil, nil, fmt.Errorf("failed to post message to Slack: %w", err)
}
return &slackResp.Channel, &slackResp.TS, nil
}
func (s *Sender) batchUpdateMessages(ctx context.Context) error {
for {
err := s.pg.WithTx(
ctx,
func(tx pg.Conn) (err error) {
update := &coredata.SlackMessageUpdate{}
defer func() {
if r := recover(); r != nil {
panicErr := fmt.Sprintf("panic recovered: %v", r)
update.Error = &panicErr
update.UpdatedAt = time.Now()
if updateErr := update.Update(ctx, tx); updateErr != nil {
s.logger.ErrorCtx(ctx, "cannot update slack message update after panic", log.Error(updateErr))
}
s.logger.ErrorCtx(ctx, "panic while updating slack message", log.String("error", panicErr), log.String("update_id", update.ID.String()))
err = fmt.Errorf("panic recovered: %v", r)
}
}()
err = update.LoadNextUnsentForUpdate(ctx, tx)
if err != nil {
return err
}
message := &coredata.SlackMessage{}
if err := message.LoadById(ctx, tx, coredata.NewScope(update.SlackMessageID.TenantID()), update.SlackMessageID); err != nil {
return fmt.Errorf("cannot load slack message: %w", err)
}
updateErr := s.updateMessage(ctx, tx, message, update)
now := time.Now()
update.UpdatedAt = now
if updateErr != nil {
errorMsg := updateErr.Error()
update.Error = &errorMsg
update.UpdatedAt = time.Now()
if err := update.Update(ctx, tx); err != nil {
return fmt.Errorf("cannot update slack message update with error: %w", err)
}
s.logger.ErrorCtx(ctx, "error updating slack message", log.Error(updateErr), log.String("update_id", update.ID.String()))
return nil
}
update.SentAt = &now
if err := update.Update(ctx, tx); err != nil {
return fmt.Errorf("cannot update slack message update: %w", err)
}
return nil
},
)
if errors.Is(err, coredata.ErrNoUnsentSlackMessageUpdate{}) {
return nil
}
if err != nil {
return err
}
}
}
func (s *Sender) updateMessage(ctx context.Context, tx pg.Conn, message *coredata.SlackMessage, update *coredata.SlackMessageUpdate) error {
if message.ChannelID == nil || message.MessageTS == nil {
return fmt.Errorf("slack message has no channel ID or message TS")
}
tenantID := message.ID.TenantID()
scope := coredata.NewScope(tenantID)
@@ -157,20 +286,25 @@ func (s *Sender) sendMessage(ctx context.Context, tx pg.Conn, message *coredata.
return fmt.Errorf("no slack connectors configured for organization")
}
for _, c := range connectors {
slackConn, ok := c.Connection.(*connector.SlackConnection)
if !ok {
return fmt.Errorf("slack connector must have SlackConnection type")
}
c := connectors[0]
if c.Connection == nil {
return fmt.Errorf("slack connector has nil connection")
}
if slackConn.Settings.WebhookURL == "" {
return fmt.Errorf("slack connector %s has no webhook URL", c.ID)
}
slackConn, ok := c.Connection.(*connector.SlackConnection)
if !ok {
return fmt.Errorf("slack connector must have SlackConnection type, got %T", c.Connection)
}
client := NewClient(slackConn.Settings.WebhookURL, s.httpClient)
if err := client.PostMessage(ctx, message.Body); err != nil {
return fmt.Errorf("failed to post message to Slack: %w", err)
}
if slackConn.AccessToken == "" {
return fmt.Errorf("slack connector %s has no access token", c.ID)
}
client := NewClient(s.logger)
if err := client.UpdateMessage(ctx, slackConn.AccessToken, *message.ChannelID, *message.MessageTS, update.Body); err != nil {
s.logger.ErrorCtx(ctx, "failed to update message on Slack", log.Error(err))
return fmt.Errorf("failed to update message on Slack: %w", err)
}
return nil