65fc9e517a
- Added new API endpoints for sending, rescheduling, and canceling scheduled outbox messages. - Implemented outbox processing logic to handle attachments and manage message statuses. - Introduced a dead-letter strategy for failed outbox messages, enhancing reliability. - Updated database schema to support new outbox statuses and dead-letter entries. - Enhanced unit tests for outbox functionalities, ensuring robust error handling and validation. - Improved attachment handling in the outbox processor to support inline and regular attachments.
74 lines
2.0 KiB
Go
74 lines
2.0 KiB
Go
package mail
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
var ErrInvalidOutboxStatus = errors.New("invalid outbox status")
|
|
|
|
func (s *Service) SendOutboxNow(ctx context.Context, userID, outboxID string) (string, error) {
|
|
var status string
|
|
err := s.db.QueryRow(ctx, `
|
|
UPDATE outbox SET status = 'queued', scheduled_at = NULL, updated_at = NOW()
|
|
WHERE id = $1 AND user_id = $2 AND status = 'scheduled'
|
|
RETURNING status
|
|
`, outboxID, userID).Scan(&status)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", s.outboxActionError(ctx, userID, outboxID)
|
|
}
|
|
return "", err
|
|
}
|
|
return status, nil
|
|
}
|
|
|
|
func (s *Service) RescheduleOutbox(ctx context.Context, userID, outboxID string, scheduledAt time.Time) (string, error) {
|
|
var status string
|
|
err := s.db.QueryRow(ctx, `
|
|
UPDATE outbox SET status = 'scheduled', scheduled_at = $3, updated_at = NOW()
|
|
WHERE id = $1 AND user_id = $2 AND status IN ('scheduled', 'queued')
|
|
RETURNING status
|
|
`, outboxID, userID, scheduledAt).Scan(&status)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", s.outboxActionError(ctx, userID, outboxID)
|
|
}
|
|
return "", err
|
|
}
|
|
return status, nil
|
|
}
|
|
|
|
func (s *Service) CancelScheduledOutbox(ctx context.Context, userID, outboxID string) (string, error) {
|
|
var status string
|
|
err := s.db.QueryRow(ctx, `
|
|
UPDATE outbox SET status = 'cancelled', scheduled_at = NULL, updated_at = NOW()
|
|
WHERE id = $1 AND user_id = $2 AND status = 'scheduled'
|
|
RETURNING status
|
|
`, outboxID, userID).Scan(&status)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", s.outboxActionError(ctx, userID, outboxID)
|
|
}
|
|
return "", err
|
|
}
|
|
return status, nil
|
|
}
|
|
|
|
func (s *Service) outboxActionError(ctx context.Context, userID, outboxID string) error {
|
|
var status string
|
|
err := s.db.QueryRow(ctx, `
|
|
SELECT status FROM outbox WHERE id = $1 AND user_id = $2
|
|
`, outboxID, userID).Scan(&status)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return ErrInvalidOutboxStatus
|
|
}
|