diff --git a/pkg/coredata/migrations/20260623T100000Z.sql b/pkg/coredata/migrations/20260623T100000Z.sql new file mode 100644 index 000000000..40d877937 --- /dev/null +++ b/pkg/coredata/migrations/20260623T100000Z.sql @@ -0,0 +1,15 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +ALTER TABLE slack_messages ADD COLUMN processing_started_at TIMESTAMP WITH TIME ZONE; diff --git a/pkg/coredata/slack_message.go b/pkg/coredata/slack_message.go index 392328851..34e5179b6 100644 --- a/pkg/coredata/slack_message.go +++ b/pkg/coredata/slack_message.go @@ -42,6 +42,7 @@ type ( CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` SentAt *time.Time `db:"sent_at"` + ProcessingStartedAt *time.Time `db:"processing_started_at"` Error *string `db:"error"` } @@ -181,7 +182,21 @@ func (s *SlackMessage) LoadNextUnsentForUpdate( conn pg.Tx, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE sent_at IS NULL AND error IS NULL ORDER BY created_at ASC @@ -213,7 +228,21 @@ func (s *SlackMessage) LoadNextInitalUnsentForUpdate( conn pg.Tx, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE sent_at IS NULL AND error IS NULL AND id = initial_slack_message_id ORDER BY created_at ASC @@ -258,6 +287,7 @@ SELECT sm.created_at, sm.updated_at, sm.sent_at, + sm.processing_started_at, sm.error FROM slack_messages sm INNER JOIN slack_messages original ON sm.initial_slack_message_id = original.id @@ -290,6 +320,84 @@ FOR UPDATE OF sm return nil } +func (s *SlackMessage) LoadNextClaimableForUpdateSkipLocked( + ctx context.Context, + conn pg.Tx, +) error { + q := ` +SELECT + sm.id, + sm.organization_id, + sm.type, + sm.body, + COALESCE(sm.message_ts, original.message_ts) as message_ts, + COALESCE(sm.channel_id, original.channel_id) as channel_id, + sm.requester_email, + sm.metadata, + sm.initial_slack_message_id, + sm.created_at, + sm.updated_at, + sm.sent_at, + sm.processing_started_at, + sm.error +FROM slack_messages sm +LEFT JOIN slack_messages original ON sm.initial_slack_message_id = original.id +WHERE sm.sent_at IS NULL + AND sm.error IS NULL + AND sm.processing_started_at IS NULL + AND ( + sm.id = sm.initial_slack_message_id + OR (original.sent_at IS NOT NULL AND original.error IS NULL) + ) +ORDER BY sm.created_at ASC +LIMIT 1 +FOR UPDATE OF sm SKIP LOCKED + ` + + rows, err := conn.Query(ctx, q) + if err != nil { + return fmt.Errorf("cannot query slack messages: %w", err) + } + + message, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[SlackMessage]) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrNoUnsentSlackMessage{} + } + + return fmt.Errorf("cannot collect slack message: %w", err) + } + + *s = message + + return nil +} + +func ResetStaleProcessingSlackMessages( + ctx context.Context, + conn pg.Querier, + staleAfter time.Duration, +) error { + q := ` +UPDATE slack_messages +SET processing_started_at = NULL, updated_at = now() +WHERE sent_at IS NULL + AND error IS NULL + AND processing_started_at IS NOT NULL + AND processing_started_at < @stale_threshold + ` + + args := pgx.StrictNamedArgs{ + "stale_threshold": time.Now().Add(-staleAfter), + } + + if _, err := conn.Exec(ctx, q, args); err != nil { + return fmt.Errorf("cannot reset stale processing slack messages: %w", err) + } + + return nil +} + func (s *SlackMessage) LoadInitialByChannelAndTS( ctx context.Context, conn pg.Querier, @@ -298,7 +406,21 @@ func (s *SlackMessage) LoadInitialByChannelAndTS( messageTS string, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE message_ts = @message_ts AND channel_id = @channel_id AND id = initial_slack_message_id AND %s LIMIT 1 @@ -338,15 +460,16 @@ func (s *SlackMessage) Update( ) error { q := ` UPDATE slack_messages -SET sent_at = @sent_at, updated_at = @updated_at, error = @error +SET sent_at = @sent_at, processing_started_at = @processing_started_at, updated_at = @updated_at, error = @error WHERE id = @id AND %s ` args := pgx.StrictNamedArgs{ - "id": s.ID, - "sent_at": s.SentAt, - "updated_at": s.UpdatedAt, - "error": s.Error, + "id": s.ID, + "sent_at": s.SentAt, + "processing_started_at": s.ProcessingStartedAt, + "updated_at": s.UpdatedAt, + "error": s.Error, } q = fmt.Sprintf(q, scope.SQLFragment()) @@ -402,7 +525,21 @@ func (s *SlackMessage) LoadById( slackMessageID gid.GID, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE id = @id AND %s @@ -442,7 +579,21 @@ func (s *SlackMessage) LoadLatestByInitialMessageID( initialSlackMessageID gid.GID, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE %s AND initial_slack_message_id = @initial_slack_message_id @@ -486,7 +637,21 @@ func (s *SlackMessage) LoadLatestByRequesterEmailAndType( since time.Time, ) error { q := ` -SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, metadata, initial_slack_message_id, created_at, updated_at, sent_at, error +SELECT + id, + organization_id, + type, + body, + message_ts, + channel_id, + requester_email, + metadata, + initial_slack_message_id, + created_at, + updated_at, + sent_at, + processing_started_at, + error FROM slack_messages WHERE %s AND organization_id = @organization_id diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index 39c1dcb1d..b53844fc0 100644 --- a/pkg/probod/probod.go +++ b/pkg/probod/probod.go @@ -701,14 +701,19 @@ func (impl *Implm) Run( ) slackSenderCtx, stopSlackSender := context.WithCancel(context.Background()) - slackSender := slack.NewSender(pgClient, l.Named("slack-sender"), encryptionKey, slack.Config{ - Interval: time.Duration(impl.cfg.Notifications.Slack.SenderInterval) * time.Second, - }) + slackSendingWorker := slack.NewSendingWorker( + pgClient, + l.Named("slack-sending-worker"), + encryptionKey, + nil, + worker.WithInterval(time.Duration(impl.cfg.Notifications.Slack.SenderInterval)*time.Second), + worker.WithMaxConcurrency(1), + ) wg.Go( func() { - if err := slackSender.Run(slackSenderCtx); err != nil { - cancel(fmt.Errorf("slack sender crashed: %w", err)) + if err := slackSendingWorker.Run(slackSenderCtx); err != nil { + cancel(fmt.Errorf("slack sending worker crashed: %w", err)) } }, ) diff --git a/pkg/slack/sender.go b/pkg/slack/sender.go deleted file mode 100644 index 53cda8b5f..000000000 --- a/pkg/slack/sender.go +++ /dev/null @@ -1,317 +0,0 @@ -// Copyright (c) 2025-2026 Probo Inc . -// -// Permission to use, copy, modify, and/or distribute this software for any -// purpose with or without fee is hereby granted, provided that the above -// copyright notice and this permission notice appear in all copies. -// -// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH -// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY -// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, -// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM -// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR -// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR -// PERFORMANCE OF THIS SOFTWARE. - -package slack - -import ( - "context" - "errors" - "fmt" - "time" - - "go.gearno.de/kit/log" - "go.gearno.de/kit/pg" - "go.probo.inc/probo/pkg/connector" - "go.probo.inc/probo/pkg/coredata" - "go.probo.inc/probo/pkg/crypto/cipher" -) - -type ( - Sender struct { - pg *pg.Client - logger *log.Logger - encryptionKey cipher.EncryptionKey - interval time.Duration - } - - Config struct { - Interval time.Duration - } -) - -func NewSender(pg *pg.Client, logger *log.Logger, encryptionKey cipher.EncryptionKey, cfg Config) *Sender { - return &Sender{ - pg: pg, - logger: logger, - encryptionKey: encryptionKey, - interval: cfg.Interval, - } -} - -func (s *Sender) Run(ctx context.Context) error { -LOOP: - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(s.interval): - ctx := context.Background() - if err := s.batchSendMessages(ctx); err != nil { - 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 - } -} - -func (s *Sender) batchSendMessages(ctx context.Context) error { - for { - err := s.pg.WithTx( - ctx, - func(ctx context.Context, tx pg.Tx) (err error) { - message := &coredata.SlackMessage{} - - defer func() { - if r := recover(); r != nil { - panicErr := fmt.Sprintf("panic recovered: %v", r) - message.Error = &panicErr - message.UpdatedAt = time.Now() - - scope := coredata.NewScope(message.ID.TenantID()) - if updateErr := message.Update(ctx, tx, scope); updateErr != nil { - s.logger.ErrorCtx(ctx, "cannot update slack message after panic", log.Error(updateErr)) - } - - s.logger.ErrorCtx(ctx, "panic while sending slack message", log.String("error", panicErr), log.String("message_id", message.ID.String())) - - err = fmt.Errorf("panic recovered: %v", r) - } - }() - - err = message.LoadNextInitalUnsentForUpdate(ctx, tx) - if err != nil { - return err - } - - scope := coredata.NewScope(message.ID.TenantID()) - channelID, messageTS, sendErr := s.sendMessage(ctx, tx, message) - - now := time.Now() - message.UpdatedAt = now - - if channelID != nil && messageTS != nil { - message.ChannelID = channelID - message.MessageTS = messageTS - message.UpdatedAt = now - - if err := message.UpdateChannelAndTSByInitialMessageID(ctx, tx, scope, message.ID, *channelID, *messageTS, now); err != nil { - return fmt.Errorf("cannot update all messages with initial message id: %w", err) - } - } - - if sendErr != nil { - errorMsg := sendErr.Error() - message.Error = &errorMsg - - if err := message.Update(ctx, tx, scope); err != nil { - return fmt.Errorf("cannot update slack message with error: %w", err) - } - - s.logger.ErrorCtx(ctx, "error sending slack message", log.Error(sendErr), log.String("message_id", message.ID.String())) - - return nil - } - - message.SentAt = &now - - if err := message.Update(ctx, tx, scope); err != nil { - return fmt.Errorf("cannot update slack message: %w", err) - } - - return nil - }, - ) - - if errors.Is(err, coredata.ErrNoUnsentSlackMessage{}) { - return nil - } - - if err != nil { - return err - } - } -} - -func (s *Sender) sendMessage(ctx context.Context, tx pg.Querier, message *coredata.SlackMessage) (*string, *string, error) { - tenantID := message.ID.TenantID() - scope := coredata.NewScope(tenantID) - - var c coredata.Connector - if err := c.LoadOneByOrganizationIDAndProvider( - ctx, - tx, - scope, - s.encryptionKey, - message.OrganizationID, - coredata.ConnectorProviderSlack, - ); err != nil { - if errors.Is(err, coredata.ErrResourceNotFound) { - return nil, nil, fmt.Errorf("cannot send slack message: no connector configured for organization") - } - - return nil, nil, fmt.Errorf("cannot send slack message: %w", err) - } - - if c.Connection == nil { - return nil, nil, fmt.Errorf("cannot send slack message: connector has nil connection") - } - - slackConn, ok := c.Connection.(*connector.SlackConnection) - if !ok { - return nil, nil, fmt.Errorf("cannot send slack message: unexpected connection type %T", c.Connection) - } - - if slackConn.Settings.ChannelID == "" { - return nil, nil, fmt.Errorf("cannot send slack message: connector %s has no channel ID", c.ID) - } - - if slackConn.AccessToken == "" { - return nil, nil, fmt.Errorf("cannot send slack message: 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, "cannot 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, "cannot post message to Slack", log.Error(err)) - return nil, nil, fmt.Errorf("cannot 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(ctx context.Context, tx pg.Tx) (err error) { - updateMessage := &coredata.SlackMessage{} - - defer func() { - if r := recover(); r != nil { - panicErr := fmt.Sprintf("panic recovered: %v", r) - updateMessage.Error = &panicErr - updateMessage.UpdatedAt = time.Now() - - scope := coredata.NewScope(updateMessage.ID.TenantID()) - if updateErr := updateMessage.Update(ctx, tx, scope); updateErr != nil { - s.logger.ErrorCtx(ctx, "cannot update slack message after panic", log.Error(updateErr)) - } - - s.logger.ErrorCtx(ctx, "panic while updating slack message", log.String("error", panicErr), log.String("message_id", updateMessage.ID.String())) - - err = fmt.Errorf("panic recovered: %v", r) - } - }() - - err = updateMessage.LoadNextUpdateUnsentForUpdate(ctx, tx) - if err != nil { - return err - } - - scope := coredata.NewScope(updateMessage.ID.TenantID()) - updateErr := s.updateMessage(ctx, tx, updateMessage) - - now := time.Now() - updateMessage.UpdatedAt = now - - if updateErr != nil { - errorMsg := updateErr.Error() - updateMessage.Error = &errorMsg - updateMessage.UpdatedAt = time.Now() - - if err := updateMessage.Update(ctx, tx, scope); err != nil { - return fmt.Errorf("cannot update slack message with error: %w", err) - } - - s.logger.ErrorCtx(ctx, "error updating slack message", log.Error(updateErr), log.String("message_id", updateMessage.ID.String())) - - return nil - } - - updateMessage.SentAt = &now - - if err := updateMessage.Update(ctx, tx, scope); err != nil { - return fmt.Errorf("cannot update slack message: %w", err) - } - - return nil - }, - ) - - if errors.Is(err, coredata.ErrNoUnsentSlackMessage{}) { - return nil - } - - if err != nil { - return err - } - } -} - -func (s *Sender) updateMessage(ctx context.Context, tx pg.Querier, updateMessage *coredata.SlackMessage) error { - if updateMessage.ChannelID == nil || updateMessage.MessageTS == nil { - return fmt.Errorf("cannot update slack message: missing channel ID or message TS") - } - - tenantID := updateMessage.ID.TenantID() - scope := coredata.NewScope(tenantID) - - var c coredata.Connector - if err := c.LoadOneByOrganizationIDAndProvider( - ctx, - tx, - scope, - s.encryptionKey, - updateMessage.OrganizationID, - coredata.ConnectorProviderSlack, - ); err != nil { - if errors.Is(err, coredata.ErrResourceNotFound) { - return fmt.Errorf("cannot update slack message: no connector configured for organization") - } - - return fmt.Errorf("cannot update slack message: %w", err) - } - - if c.Connection == nil { - return fmt.Errorf("cannot update slack message: connector has nil connection") - } - - slackConn, ok := c.Connection.(*connector.SlackConnection) - if !ok { - return fmt.Errorf("cannot update slack message: unexpected connection type %T", c.Connection) - } - - if slackConn.AccessToken == "" { - return fmt.Errorf("cannot update slack message: connector %s has no access token", c.ID) - } - - client := NewClient(s.logger) - - if err := client.UpdateMessage(ctx, slackConn.AccessToken, *updateMessage.ChannelID, *updateMessage.MessageTS, updateMessage.Body); err != nil { - s.logger.ErrorCtx(ctx, "cannot update message on Slack", log.Error(err)) - return fmt.Errorf("cannot update message on Slack: %w", err) - } - - return nil -} diff --git a/pkg/slack/worker.go b/pkg/slack/worker.go new file mode 100644 index 000000000..d5a16b34c --- /dev/null +++ b/pkg/slack/worker.go @@ -0,0 +1,268 @@ +// Copyright (c) 2025-2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package slack + +import ( + "context" + "errors" + "fmt" + "time" + + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.gearno.de/kit/worker" + "go.probo.inc/probo/pkg/connector" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/crypto/cipher" +) + +type ( + sendingHandler struct { + pg *pg.Client + logger *log.Logger + encryptionKey cipher.EncryptionKey + staleAfter time.Duration + } + + SendingWorkerOption func(*sendingHandler) +) + +var ( + _ worker.Handler[coredata.SlackMessage] = (*sendingHandler)(nil) + _ worker.StaleRecoverer = (*sendingHandler)(nil) +) + +func WithSendingWorkerStaleAfter(d time.Duration) SendingWorkerOption { + return func(h *sendingHandler) { h.staleAfter = d } +} + +func NewSendingWorker( + pgClient *pg.Client, + logger *log.Logger, + encryptionKey cipher.EncryptionKey, + handlerOpts []SendingWorkerOption, + workerOpts ...worker.Option, +) *worker.Worker[coredata.SlackMessage] { + h := &sendingHandler{ + pg: pgClient, + logger: logger, + encryptionKey: encryptionKey, + staleAfter: 5 * time.Minute, + } + + for _, opt := range handlerOpts { + opt(h) + } + + return worker.New( + "slack-sending-worker", + h, + logger, + workerOpts..., + ) +} + +func (h *sendingHandler) Claim(ctx context.Context) (coredata.SlackMessage, error) { + var message coredata.SlackMessage + + if err := h.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + if err := message.LoadNextClaimableForUpdateSkipLocked(ctx, tx); err != nil { + return err + } + + now := time.Now() + message.ProcessingStartedAt = &now + message.UpdatedAt = now + + scope := coredata.NewScope(message.ID.TenantID()) + if err := message.Update(ctx, tx, scope); err != nil { + return fmt.Errorf("cannot mark slack message as processing: %w", err) + } + + return nil + }, + ); err != nil { + if errors.Is(err, coredata.ErrNoUnsentSlackMessage{}) { + return coredata.SlackMessage{}, worker.ErrNoTask + } + + return coredata.SlackMessage{}, err + } + + return message, nil +} + +func (h *sendingHandler) Process(ctx context.Context, message coredata.SlackMessage) error { + isInitial := message.ID == message.InitialSlackMessageID + + var ( + channelID *string + messageTS *string + sendErr error + ) + + if isInitial { + channelID, messageTS, sendErr = h.sendMessage(ctx, &message) + } else { + sendErr = h.updateMessage(ctx, &message) + } + + scope := coredata.NewScope(message.ID.TenantID()) + + if commitErr := h.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + now := time.Now() + message.UpdatedAt = now + message.ProcessingStartedAt = nil + + if sendErr != nil { + errorMsg := sendErr.Error() + message.Error = &errorMsg + + if err := message.Update(ctx, tx, scope); err != nil { + return fmt.Errorf("cannot update slack message with error: %w", err) + } + + return nil + } + + if isInitial && channelID != nil && messageTS != nil { + message.ChannelID = channelID + message.MessageTS = messageTS + + if err := message.UpdateChannelAndTSByInitialMessageID(ctx, tx, scope, message.ID, *channelID, *messageTS, now); err != nil { + return fmt.Errorf("cannot update all messages with initial message id: %w", err) + } + } + + message.SentAt = &now + + if err := message.Update(ctx, tx, scope); err != nil { + return fmt.Errorf("cannot update slack message: %w", err) + } + + return nil + }, + ); commitErr != nil { + return commitErr + } + + if sendErr != nil { + h.logger.ErrorCtx(ctx, "error processing slack message", log.Error(sendErr), log.String("message_id", message.ID.String())) + return sendErr + } + + return nil +} + +func (h *sendingHandler) RecoverStale(ctx context.Context) error { + return h.pg.WithConn( + ctx, + func(ctx context.Context, conn pg.Querier) error { + return coredata.ResetStaleProcessingSlackMessages(ctx, conn, h.staleAfter) + }, + ) +} + +func (h *sendingHandler) loadSlackConnection(ctx context.Context, message *coredata.SlackMessage) (*connector.SlackConnection, error) { + scope := coredata.NewScope(message.ID.TenantID()) + + var c coredata.Connector + + if err := h.pg.WithConn( + ctx, + func(ctx context.Context, conn pg.Querier) error { + return c.LoadOneByOrganizationIDAndProvider( + ctx, + conn, + scope, + h.encryptionKey, + message.OrganizationID, + coredata.ConnectorProviderSlack, + ) + }, + ); err != nil { + if errors.Is(err, coredata.ErrResourceNotFound) { + return nil, fmt.Errorf("no connector configured for organization") + } + + return nil, err + } + + if c.Connection == nil { + return nil, fmt.Errorf("connector has nil connection") + } + + slackConn, ok := c.Connection.(*connector.SlackConnection) + if !ok { + return nil, fmt.Errorf("unexpected connection type %T", c.Connection) + } + + if slackConn.AccessToken == "" { + return nil, fmt.Errorf("connector %s has no access token", c.ID) + } + + return slackConn, nil +} + +func (h *sendingHandler) sendMessage(ctx context.Context, message *coredata.SlackMessage) (*string, *string, error) { + slackConn, err := h.loadSlackConnection(ctx, message) + if err != nil { + return nil, nil, fmt.Errorf("cannot send slack message: %w", err) + } + + if slackConn.Settings.ChannelID == "" { + return nil, nil, fmt.Errorf("cannot send slack message: connector has no channel ID") + } + + client := NewClient(h.logger) + + if message.Type == coredata.SlackMessageTypeWelcome { + if err := client.JoinChannel(ctx, slackConn.AccessToken, slackConn.Settings.ChannelID); err != nil { + h.logger.ErrorCtx(ctx, "cannot join Slack channel", log.Error(err)) + } + } + + slackResp, err := client.CreateMessage(ctx, slackConn.AccessToken, slackConn.Settings.ChannelID, message.Body) + if err != nil { + h.logger.ErrorCtx(ctx, "cannot post message to Slack", log.Error(err)) + return nil, nil, fmt.Errorf("cannot post message to Slack: %w", err) + } + + return &slackResp.Channel, &slackResp.TS, nil +} + +func (h *sendingHandler) updateMessage(ctx context.Context, message *coredata.SlackMessage) error { + if message.ChannelID == nil || message.MessageTS == nil { + return fmt.Errorf("cannot update slack message: missing channel ID or message TS") + } + + slackConn, err := h.loadSlackConnection(ctx, message) + if err != nil { + return fmt.Errorf("cannot update slack message: %w", err) + } + + client := NewClient(h.logger) + + if err := client.UpdateMessage(ctx, slackConn.AccessToken, *message.ChannelID, *message.MessageTS, message.Body); err != nil { + h.logger.ErrorCtx(ctx, "cannot update message on Slack", log.Error(err)) + return fmt.Errorf("cannot update message on Slack: %w", err) + } + + return nil +}