Refactor storage and webhook code (#346)

functions/emailReceive/main.go

@@ -17,9 +17,9 @@ import (
 
 	"github.com/harryzcy/mailbox/internal/datasource/storage"
 	"github.com/harryzcy/mailbox/internal/env"
+	"github.com/harryzcy/mailbox/internal/hook"
 	"github.com/harryzcy/mailbox/internal/thread"
 	"github.com/harryzcy/mailbox/internal/util/format"
-	"github.com/harryzcy/mailbox/internal/webhook"
 )
 
 func main() {
@@ -103,26 +103,22 @@ func receiveEmail(ctx context.Context, ses events.SimpleEmailService) {
 		TimeReceived: format.FormatRFC3399(ses.Mail.Timestamp),
 	})
 
-	if storage.SQS.Enabled() {
-		err = storage.SQS.SendEmailReceipt(ctx, sqs.NewFromConfig(cfg), storage.EmailReceipt{
-			MessageID: ses.Mail.MessageID,
-			Timestamp: ses.Mail.Timestamp.UTC().Format(time.RFC3339),
-		})
-		if err != nil {
-			log.Printf("failed to send email receipt to SQS, %v\n", err)
-		}
+	err = hook.SendSQS(ctx, sqs.NewFromConfig(cfg), hook.EmailReceipt{
+		MessageID: ses.Mail.MessageID,
+		Timestamp: ses.Mail.Timestamp.UTC().Format(time.RFC3339),
+	})
+	if err != nil {
+		log.Printf("failed to send email receipt to SQS, %v\n", err)
 	}
 
-	if webhook.Enabled() {
-		err = webhook.SendWebhook(ctx, &webhook.Webhook{
-			Event:  webhook.EventEmail,
-			Action: webhook.ActionReceived,
-			Email: webhook.Email{
-				ID: ses.Mail.MessageID,
-			},
-		})
-		if err != nil {
-			log.Printf("failed to send webhook, %v\n", err)
-		}
+	err = hook.SendWebhook(ctx, &hook.Webhook{
+		Event:  hook.EventEmail,
+		Action: hook.ActionReceived,
+		Email: hook.Email{
+			ID: ses.Mail.MessageID,
+		},
+	})
+	if err != nil {
+		log.Printf("failed to send webhook, %v\n", err)
 	}
 }

internal/api/sqs.go

@@ -0,0 +1,13 @@
+package api
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+)
+
+// SQSSendMessageAPI defines set of API required by SendEmailReceipt and SendEmailNotification functions
+type SQSSendMessageAPI interface {
+	GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
+	SendMessage(ctx context.Context, params *sqs.SendMessageInput, optFns ...func(*sqs.Options)) (*sqs.SendMessageOutput, error)
+}

internal/hook/content.go

@@ -0,0 +1,29 @@
+package hook
+
+const (
+	EventEmail     = "email"
+	ActionReceived = "received"
+)
+
+// EmailReceipt contains information needed for an email receipt
+type EmailReceipt struct {
+	MessageID string
+	Timestamp string
+}
+
+// EmailNotification contains information needed for an email state change notification
+type EmailNotification struct {
+	Event     string `json:"event"`
+	MessageID string `json:"messageID"`
+	Timestamp string `json:"timestamp"`
+}
+
+type Webhook struct {
+	Event  string `json:"event"`
+	Action string `json:"action"`
+	Email  Email
+}
+
+type Email struct {
+	ID string `json:"id"` // message id
+}

internal/hook/sqs.go

@@ -0,0 +1,68 @@
+package hook
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/harryzcy/mailbox/internal/api"
+	"github.com/harryzcy/mailbox/internal/env"
+)
+
+// sqsEnabled returns true if SQS is enabled
+func sqsEnabled() bool {
+	return env.QueueName != ""
+}
+
+// SendEmailHandle sends an email receipt to SQS, if SQS is enabled.
+// Otherwise, it does nothing.
+func SendSQS(ctx context.Context, api api.SQSSendMessageAPI, input EmailReceipt) error {
+	if !sqsEnabled() {
+		return nil
+	}
+
+	fmt.Printf("Sending email receipt (MessageID: %s)\n", input.MessageID)
+	return sendSQSEmailNotification(ctx, api, EmailNotification{
+		Event:     "receive",
+		MessageID: input.MessageID,
+		Timestamp: input.Timestamp,
+	})
+}
+
+// sendSQSEmailNotification notifies about a change of state of an email, categorized by event.
+func sendSQSEmailNotification(ctx context.Context, api api.SQSSendMessageAPI, input EmailNotification) error {
+	result, err := api.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
+		QueueName: &env.QueueName,
+	})
+	if err != nil {
+		fmt.Println("Failed to get queue url")
+		return err
+	}
+
+	body, _ := json.Marshal(input)
+
+	resp, err := api.SendMessage(ctx, &sqs.SendMessageInput{
+		MessageAttributes: map[string]types.MessageAttributeValue{
+			"Event": {
+				DataType:    aws.String("String"),
+				StringValue: aws.String(input.Event),
+			},
+			"Timestamp": {
+				DataType:    aws.String("String"),
+				StringValue: aws.String(input.Timestamp),
+			},
+		},
+		MessageBody: aws.String(string(body)),
+		QueueUrl:    result.QueueUrl,
+	})
+	if err != nil {
+		fmt.Println("Failed to send message to SQS")
+		return err
+	}
+
+	fmt.Println("Sent message with ID: " + *resp.MessageId)
+	return nil
+}