Refactor slack messages

Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
2025-10-23 09:32:55 +02:00
parent 0073846d3b
commit b0f60d8de5
9 changed files with 611 additions and 529 deletions
--- a/pkg/coredata/slack_message.go
+++ b/pkg/coredata/slack_message.go
@@ -28,26 +28,34 @@ import (

 type (
 	SlackMessage struct {
-		ID             gid.GID          `db:"id"`
-		OrganizationID gid.GID          `db:"organization_id"`
-		Type           SlackMessageType `db:"type"`
-		Body           map[string]any   `db:"body"`
-		MessageTS      *string          `db:"message_ts"`
-		ChannelID      *string          `db:"channel_id"`
-		RequesterEmail *string          `db:"requester_email"`
-		CreatedAt      time.Time        `db:"created_at"`
-		UpdatedAt      time.Time        `db:"updated_at"`
-		SentAt         *time.Time       `db:"sent_at"`
-		Error          *string          `db:"error"`
+		ID                    gid.GID          `db:"id"`
+		OrganizationID        gid.GID          `db:"organization_id"`
+		Type                  SlackMessageType `db:"type"`
+		Body                  map[string]any   `db:"body"`
+		MessageTS             *string          `db:"message_ts"`
+		ChannelID             *string          `db:"channel_id"`
+		RequesterEmail        *string          `db:"requester_email"`
+		Metadata              map[string]any   `db:"metadata"`
+		InitialSlackMessageID gid.GID          `db:"initial_slack_message_id"`
+		CreatedAt             time.Time        `db:"created_at"`
+		UpdatedAt             time.Time        `db:"updated_at"`
+		SentAt                *time.Time       `db:"sent_at"`
+		Error                 *string          `db:"error"`
 	}

 	ErrNoUnsentSlackMessage struct{}
+
+	ErrSlackMessageNotFound struct{}
 )

 func (e ErrNoUnsentSlackMessage) Error() string {
 	return "no unsent slack message found"
 }

+func (e ErrSlackMessageNotFound) Error() string {
+	return "slack message not found"
+}
+
 func NewSlackMessage(
 	scope Scoper,
 	organizationID gid.GID,
@@ -56,14 +64,16 @@ func NewSlackMessage(
 	requesterEmail *string,
 ) *SlackMessage {
 	now := time.Now()
+	id := gid.New(scope.GetTenantID(), SlackMessageEntityType)
 	return &SlackMessage{
-		ID:             gid.New(scope.GetTenantID(), SlackMessageEntityType),
-		OrganizationID: organizationID,
-		Type:           messageType,
-		Body:           body,
-		RequesterEmail: requesterEmail,
-		CreatedAt:      now,
-		UpdatedAt:      now,
+		ID:                    id,
+		OrganizationID:        organizationID,
+		Type:                  messageType,
+		Body:                  body,
+		RequesterEmail:        requesterEmail,
+		InitialSlackMessageID: id,
+		CreatedAt:             now,
+		UpdatedAt:             now,
 	}
 }

@@ -73,19 +83,43 @@ func (s *SlackMessage) Insert(
 	scope Scoper,
 ) error {
 	q := `
-INSERT INTO slack_messages (id, tenant_id, organization_id, type, body, requester_email, created_at, updated_at)
-VALUES (@id, @tenant_id, @organization_id, @type, @body, @requester_email, @created_at, @updated_at)
+		INSERT INTO slack_messages (
+			id,
+			tenant_id,
+			organization_id,
+			type,
+			body,
+			requester_email,
+			metadata,
+			initial_slack_message_id,
+			created_at,
+			updated_at
+		)
+		VALUES (
+			@id,
+			@tenant_id,
+			@organization_id,
+			@type,
+			@body,
+			@requester_email,
+			@metadata,
+			@initial_slack_message_id,
+			@created_at,
+			@updated_at
+		)
 	`

 	args := pgx.StrictNamedArgs{
-		"id":              s.ID,
-		"tenant_id":       scope.GetTenantID(),
-		"organization_id": s.OrganizationID,
-		"type":            s.Type,
-		"body":            s.Body,
-		"requester_email": s.RequesterEmail,
-		"created_at":      s.CreatedAt,
-		"updated_at":      s.UpdatedAt,
+		"id":                       s.ID,
+		"tenant_id":                scope.GetTenantID(),
+		"organization_id":          s.OrganizationID,
+		"type":                     s.Type,
+		"body":                     s.Body,
+		"requester_email":          s.RequesterEmail,
+		"metadata":                 s.Metadata,
+		"initial_slack_message_id": s.InitialSlackMessageID,
+		"created_at":               s.CreatedAt,
+		"updated_at":               s.UpdatedAt,
 	}

 	_, err := conn.Exec(ctx, q, args)
@@ -101,7 +135,7 @@ func (s *SlackMessage) LoadNextUnsentForUpdate(
 	conn pg.Conn,
 ) error {
 	q := `
-SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, 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, error
 FROM slack_messages
 WHERE sent_at IS NULL AND error IS NULL
 ORDER BY created_at ASC
@@ -128,17 +162,99 @@ FOR UPDATE
 	return nil
 }

-// This is used for Slack webhook verification where we don't know the tenant yet
-func (s *SlackMessage) LoadByChannelAndTSUnscoped(
+func (s *SlackMessage) LoadNextInitalUnsentForUpdate(
 	ctx context.Context,
 	conn pg.Conn,
+) 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
+FROM slack_messages
+WHERE sent_at IS NULL AND error IS NULL AND id = initial_slack_message_id
+ORDER BY created_at ASC
+LIMIT 1
+FOR UPDATE
+	`
+
+	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 (s *SlackMessage) LoadNextUpdateUnsentForUpdate(
+	ctx context.Context,
+	conn pg.Conn,
+) 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.error
+FROM slack_messages sm
+INNER 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.id != sm.initial_slack_message_id
+	AND original.sent_at IS NOT NULL
+	AND original.error IS NULL
+ORDER BY sm.created_at ASC
+LIMIT 1
+FOR UPDATE OF sm
+	`
+
+	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 (s *SlackMessage) LoadInitialByChannelAndTS(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
 	channelID string,
 	messageTS string,
 ) error {
 	q := `
-SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, 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, error
 FROM slack_messages
-WHERE message_ts = @message_ts AND channel_id = @channel_id
+WHERE message_ts = @message_ts AND channel_id = @channel_id AND id = initial_slack_message_id AND %s
 LIMIT 1
 	`

@@ -146,6 +262,9 @@ LIMIT 1
 		"message_ts": messageTS,
 		"channel_id": channelID,
 	}
+	maps.Copy(args, scope.SQLArguments())
+
+	q = fmt.Sprintf(q, scope.SQLFragment())

 	rows, err := conn.Query(ctx, q, args)
 	if err != nil {
@@ -168,23 +287,25 @@ LIMIT 1
 func (s *SlackMessage) Update(
 	ctx context.Context,
 	conn pg.Conn,
+	scope Scoper,
 ) error {
 	q := `
 UPDATE slack_messages
-SET body = @body, sent_at = @sent_at, updated_at = @updated_at, error = @error, message_ts = @message_ts, channel_id = @channel_id
-WHERE id = @id
+SET sent_at = @sent_at, updated_at = @updated_at, error = @error
+WHERE id = @id AND %s
 	`

 	args := pgx.StrictNamedArgs{
 		"id":         s.ID,
-		"body":       s.Body,
 		"sent_at":    s.SentAt,
 		"updated_at": s.UpdatedAt,
 		"error":      s.Error,
-		"message_ts": s.MessageTS,
-		"channel_id": s.ChannelID,
 	}

+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	maps.Copy(args, scope.SQLArguments())
+
 	_, err := conn.Exec(ctx, q, args)
 	if err != nil {
 		return fmt.Errorf("cannot update slack message: %w", err)
@@ -193,6 +314,40 @@ WHERE id = @id
 	return nil
 }

+func (s *SlackMessage) UpdateChannelAndTSByInitialMessageID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	initialSlackMessageID gid.GID,
+	channelID string,
+	messageTS string,
+	updatedAt time.Time,
+) error {
+	q := `
+UPDATE slack_messages
+SET channel_id = @channel_id, message_ts = @message_ts, updated_at = @updated_at
+WHERE initial_slack_message_id = @initial_slack_message_id AND %s
+	`
+
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"initial_slack_message_id": initialSlackMessageID,
+		"channel_id":               channelID,
+		"message_ts":               messageTS,
+		"updated_at":               updatedAt,
+	}
+
+	maps.Copy(args, scope.SQLArguments())
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update slack messages with initial message id: %w", err)
+	}
+
+	return nil
+}
+
 func (s *SlackMessage) LoadById(
 	ctx context.Context,
 	conn pg.Conn,
@@ -200,13 +355,15 @@ func (s *SlackMessage) LoadById(
 	slackMessageID gid.GID,
 ) error {
 	q := `
-SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, 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, error
 FROM slack_messages
 WHERE id = @id
 AND %s
 LIMIT 1
 	`

+	q = fmt.Sprintf(q, scope.SQLFragment())
+
 	args := pgx.StrictNamedArgs{
 		"id": slackMessageID,
 	}
@@ -230,6 +387,46 @@ LIMIT 1
 	return nil
 }

+func (s *SlackMessage) LoadLatestByInitialMessageID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	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
+FROM slack_messages
+WHERE %s
+	AND initial_slack_message_id = @initial_slack_message_id
+ORDER BY created_at DESC
+LIMIT 1
+	`
+
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"initial_slack_message_id": initialSlackMessageID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query slack message: %w", err)
+	}
+
+	message, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[SlackMessage])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrSlackMessageNotFound{}
+		}
+		return err
+	}
+
+	*s = message
+
+	return nil
+}
+
 func (s *SlackMessage) LoadLatestByRequesterEmailAndType(
 	ctx context.Context,
 	conn pg.Conn,
@@ -240,7 +437,7 @@ func (s *SlackMessage) LoadLatestByRequesterEmailAndType(
 	since time.Time,
 ) error {
 	q := `
-SELECT id, organization_id, type, body, message_ts, channel_id, requester_email, 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, error
 FROM slack_messages
 WHERE %s
 	AND organization_id = @organization_id
@@ -268,6 +465,9 @@ LIMIT 1

 	message, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[SlackMessage])
 	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrSlackMessageNotFound{}
+		}
 		return err
 	}