implement rate limiting

Author: AchoArnold

android/app/src/main/java/com/httpsms/SmsManagerService.kt

@@ -21,7 +21,7 @@ class SmsManagerService {
 
     fun sendMessage(context: Context, message: Message, sentIntent:PendingIntent, deliveryIntent: PendingIntent) {
         getSmsManager(context)
-            .sendTextMessage(message.contact, message.owner, message.content, sentIntent, deliveryIntent)
+            .sendTextMessage(message.contact, null, message.content, sentIntent, deliveryIntent)
     }
 
     @Suppress("DEPRECATION")

api/go.mod

@@ -3,6 +3,7 @@ module github.com/NdoleStudio/http-sms-manager
 go 1.18
 
 require (
+	cloud.google.com/go/cloudtasks v1.4.0
 	firebase.google.com/go v3.13.0+incompatible
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace v1.8.3
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
@@ -24,18 +25,20 @@ require (
 	go.opentelemetry.io/otel/exporters/jaeger v1.7.0
 	go.opentelemetry.io/otel/sdk v1.7.0
 	go.opentelemetry.io/otel/trace v1.7.0
-	google.golang.org/api v0.74.0
+	google.golang.org/api v0.85.0
+	google.golang.org/genproto v0.0.0-20220617124728-180714bec0ad
+	google.golang.org/protobuf v1.28.0
 	gorm.io/datatypes v1.0.6
 	gorm.io/driver/postgres v1.3.7
 	gorm.io/gorm v1.23.5
 )
 
 require (
-	cloud.google.com/go v0.100.2 // indirect
-	cloud.google.com/go/compute v1.5.0 // indirect
+	cloud.google.com/go v0.102.0 // indirect
+	cloud.google.com/go/compute v1.7.0 // indirect
 	cloud.google.com/go/firestore v1.6.1 // indirect
-	cloud.google.com/go/iam v0.1.1 // indirect
-	cloud.google.com/go/storage v1.21.0 // indirect
+	cloud.google.com/go/iam v0.3.0 // indirect
+	cloud.google.com/go/storage v1.22.1 // indirect
 	cloud.google.com/go/trace v1.2.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.32.3 // indirect
 	github.com/KyleBanks/depth v1.2.1 // indirect
@@ -51,8 +54,10 @@ require (
 	github.com/go-sql-driver/mysql v1.6.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/go-cmp v0.5.7 // indirect
-	github.com/googleapis/gax-go/v2 v2.2.0 // indirect
+	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.1.0 // indirect
+	github.com/googleapis/gax-go/v2 v2.4.0 // indirect
+	github.com/googleapis/go-type-adapters v1.0.0 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
 	github.com/jackc/pgconn v1.12.1 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect
@@ -83,16 +88,14 @@ require (
 	go.uber.org/zap v1.13.0 // indirect
 	golang.org/x/crypto v0.0.0-20220517005047-85d78b3ac167 // indirect
 	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 // indirect
-	golang.org/x/net v0.0.0-20220526153639-5463443f8c37 // indirect
-	golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/net v0.0.0-20220617184016-355a448f1bc9 // indirect
+	golang.org/x/oauth2 v0.0.0-20220608161450-d0670ef3b1eb // indirect
+	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.10 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220405205423-9d709892a2bf // indirect
-	google.golang.org/grpc v1.45.0 // indirect
-	google.golang.org/protobuf v1.28.0 // indirect
+	google.golang.org/grpc v1.47.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gorm.io/driver/mysql v1.3.2 // indirect
 )

api/go.sum

@@ -23,7 +23,7 @@ import (
 //
 // @securitydefinitions.apikey ApiKeyAuth
 // @in header
-// @name X-API-Key
+// @name x-api-Key
 func main() {
 	if len(os.Args) == 1 {
 		di.LoadEnv()

api/main.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"time"
 
+	cloudtasks "cloud.google.com/go/cloudtasks/apiv2"
+
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/propagation"
@@ -75,6 +77,8 @@ func NewContainer(projectID string) (container *Container) {
 
 	container.RegisterPhoneRoutes()
 
+	container.RegisterEventRoutes()
+
 	container.RegisterNotificationListeners()
 
 	// this has to be last since it registers the /* route
@@ -188,6 +192,10 @@ func (container *Container) DB() (db *gorm.DB) {
 		container.logger.Fatal(stacktrace.Propagate(err, fmt.Sprintf("cannot migrate %T", &entities.Phone{})))
 	}
 
+	if err = db.AutoMigrate(&entities.PhoneNotification{}); err != nil {
+		container.logger.Fatal(stacktrace.Propagate(err, fmt.Sprintf("cannot migrate %T", &entities.PhoneNotification{})))
+	}
+
 	return container.db
 }
 
@@ -213,6 +221,42 @@ func (container *Container) FirebaseAuthClient() (client *auth.Client) {
 	return authClient
 }
 
+// CloudTasksClient creates a new instance of cloudtasks.Client
+func (container *Container) CloudTasksClient() (client *cloudtasks.Client) {
+	container.logger.Debug(fmt.Sprintf("creating %T", client))
+
+	client, err := cloudtasks.NewClient(context.Background(), option.WithCredentialsJSON(container.FirebaseCredentials()))
+	if err != nil {
+		container.logger.Fatal(stacktrace.Propagate(err, "cannot initialize cloud tasks client"))
+	}
+
+	return client
+}
+
+// EventsQueueConfiguration creates a new instance of services.PushQueueConfig
+func (container *Container) EventsQueueConfiguration() (config services.PushQueueConfig) {
+	container.logger.Debug(fmt.Sprintf("creating %T", config))
+
+	return services.PushQueueConfig{
+		UserAPIKey:       os.Getenv("EVENTS_QUEUE_USER_API_KEY"),
+		Name:             os.Getenv("EVENTS_QUEUE_NAME"),
+		UserID:           entities.UserID(os.Getenv("EVENTS_QUEUE_USER_ID")),
+		ConsumerEndpoint: os.Getenv("EVENTS_QUEUE_ENDPOINT"),
+	}
+}
+
+// EventsQueue creates a new instance of services.PushQueue
+func (container *Container) EventsQueue() (queue services.PushQueue) {
+	container.logger.Debug("creating events services.PushQueue")
+
+	return services.NewGooglePushQueue(
+		container.Logger(),
+		container.Tracer(),
+		container.CloudTasksClient(),
+		container.EventsQueueConfiguration(),
+	)
+}
+
 // FirebaseMessagingClient creates a new instance of messaging.Client
 func (container *Container) FirebaseMessagingClient() (client *messaging.Client) {
 	container.logger.Debug(fmt.Sprintf("creating %T", client))
@@ -317,6 +361,8 @@ func (container *Container) EventDispatcher() (dispatcher *services.EventDispatc
 		container.Logger(),
 		container.Tracer(),
 		container.EventRepository(),
+		container.EventsQueue(),
+		container.EventsQueueConfiguration(),
 	)
 
 	container.eventDispatcher = dispatcher
@@ -343,6 +389,16 @@ func (container *Container) PhoneRepository() (repository repositories.PhoneRepo
 	)
 }
 
+// PhoneNotificationRepository creates a new instance of repositories.PhoneNotificationRepository
+func (container *Container) PhoneNotificationRepository() (repository repositories.PhoneNotificationRepository) {
+	container.logger.Debug("creating GORM repositories.PhoneNotificationRepository")
+	return repositories.NewGormPhoneNotificationRepository(
+		container.Logger(),
+		container.Tracer(),
+		container.DB(),
+	)
+}
+
 // MessageThreadRepository creates a new instance of repositories.MessageThreadRepository
 func (container *Container) MessageThreadRepository() (repository repositories.MessageThreadRepository) {
 	container.logger.Debug("creating GORM repositories.MessageThreadRepository")
@@ -446,6 +502,18 @@ func (container *Container) PhoneHandler() (handler *handlers.PhoneHandler) {
 	)
 }
 
+// EventsHandler creates a new instance of handlers.EventsHandler
+func (container *Container) EventsHandler() (handler *handlers.EventsHandler) {
+	container.logger.Debug(fmt.Sprintf("creating %T", handler))
+
+	return handlers.NewEventsHandler(
+		container.Logger(),
+		container.Tracer(),
+		container.EventsQueueConfiguration(),
+		container.EventDispatcher(),
+	)
+}
+
 // RegisterMessageListeners registers event listeners for listeners.MessageListener
 func (container *Container) RegisterMessageListeners() {
 	container.logger.Debug(fmt.Sprintf("registering listners for %T", listeners.MessageListener{}))
@@ -523,6 +591,8 @@ func (container *Container) NotificationService() (service *services.Notificatio
 		container.Tracer(),
 		container.FirebaseMessagingClient(),
 		container.PhoneRepository(),
+		container.PhoneNotificationRepository(),
+		container.EventDispatcher(),
 	)
 }
 
@@ -556,6 +626,12 @@ func (container *Container) RegisterUserRoutes() {
 	container.UserHandler().RegisterRoutes(container.AuthRouter())
 }
 
+// RegisterEventRoutes registers routes for the /events prefix
+func (container *Container) RegisterEventRoutes() {
+	container.logger.Debug(fmt.Sprintf("registering %T routes", &handlers.EventsHandler{}))
+	container.EventsHandler().RegisterRoutes(container.AuthRouter())
+}
+
 // RegisterSwaggerRoutes registers routes for swagger
 func (container *Container) RegisterSwaggerRoutes() {
 	container.logger.Debug(fmt.Sprintf("registering %T routes", &handlers.MessageHandler{}))

api/pkg/di/container.go

@@ -0,0 +1,18 @@
+package entities
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// Phone represents an android phone which has installed the http sms app
+type Phone struct {
+	ID                uuid.UUID `json:"id" gorm:"primaryKey;type:uuid;" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
+	UserID            UserID    `json:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
+	FcmToken          *string   `json:"fcm_token" example:"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzd....."`
+	PhoneNumber       string    `json:"phone_number" example:"+18005550199"`
+	MessagesPerMinute uint      `json:"messages_per_second" example:"1"`
+	CreatedAt         time.Time `json:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
+	UpdatedAt         time.Time `json:"updated_at" example:"2022-06-05T14:26:10.303278+03:00"`
+}

api/pkg/entities/Phone.go

@@ -0,0 +1,31 @@
+package entities
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+const (
+	// PhoneNotificationStatusPending is the status when a notification is scheduled to be sent
+	PhoneNotificationStatusPending = "pending"
+	// PhoneNotificationStatusSent is the status when a notification has been sent
+	PhoneNotificationStatusSent = "sent"
+	// PhoneNotificationStatusFailed is the status when a notification could not be sent.
+	PhoneNotificationStatusFailed = "failed"
+)
+
+// PhoneNotificationStatus is the status of a phone notification
+type PhoneNotificationStatus string
+
+// PhoneNotification represents an FCM notification to a mobile phone
+type PhoneNotification struct {
+	ID          uuid.UUID `json:"id" gorm:"primaryKey;type:uuid;"`
+	MessageID   uuid.UUID `json:"message_id"`
+	UserID      UserID    `json:"user_id"`
+	PhoneID     uuid.UUID `json:"phone_id"`
+	Status      string    `json:"status"`
+	ScheduledAt time.Time `json:"scheduled_at"`
+	CreatedAt   time.Time `json:"created_at"`
+	UpdatedAt   time.Time `json:"updated_at"`
+}

api/pkg/entities/phone.go

@@ -0,0 +1,21 @@
+package events
+
+import (
+	"time"
+
+	"github.com/NdoleStudio/http-sms-manager/pkg/entities"
+
+	"github.com/google/uuid"
+)
+
+// EventTypeMessageNotificationScheduled is emitted when a new message notification is scheduled
+const EventTypeMessageNotificationScheduled = "message.notification.scheduled"
+
+// MessageNotificationScheduledPayload is the payload of the MessageNotificationScheduledPayload event
+type MessageNotificationScheduledPayload struct {
+	MessageID      uuid.UUID       `json:"id"`
+	UserID         entities.UserID `json:"user_id"`
+	PhoneID        uuid.UUID       `json:"phone_id"`
+	ScheduledAt    time.Time       `json:"scheduled_at"`
+	NotificationID uuid.UUID       `json:"notification_id"`
+}

api/pkg/entities/phone_notification.go

@@ -0,0 +1,79 @@
+package handlers
+
+import (
+	"fmt"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+
+	"github.com/NdoleStudio/http-sms-manager/pkg/services"
+	"github.com/NdoleStudio/http-sms-manager/pkg/telemetry"
+	"github.com/davecgh/go-spew/spew"
+	"github.com/gofiber/fiber/v2"
+	"github.com/palantir/stacktrace"
+)
+
+// EventsHandler handles heartbeat http requests.
+type EventsHandler struct {
+	handler
+	logger      telemetry.Logger
+	tracer      telemetry.Tracer
+	queueConfig services.PushQueueConfig
+	service     *services.EventDispatcher
+}
+
+// NewEventsHandler creates a new EventsHandler
+func NewEventsHandler(
+	logger telemetry.Logger,
+	tracer telemetry.Tracer,
+	queueConfig services.PushQueueConfig,
+	service *services.EventDispatcher,
+) (h *EventsHandler) {
+	return &EventsHandler{
+		logger:      logger.WithService(fmt.Sprintf("%T", h)),
+		tracer:      tracer,
+		queueConfig: queueConfig,
+		service:     service,
+	}
+}
+
+// RegisterRoutes registers the routes for the MessageHandler
+func (h *EventsHandler) RegisterRoutes(router fiber.Router) {
+	router.Post("/events", h.Dispatch)
+}
+
+// Dispatch a cloud event
+// This is an internal API so no documentation provided
+func (h *EventsHandler) Dispatch(c *fiber.Ctx) error {
+	ctx, span := h.tracer.StartFromFiberCtx(c)
+	defer span.End()
+
+	ctxLogger := h.tracer.CtxLogger(h.logger, span)
+
+	var request cloudevents.Event
+	if err := c.BodyParser(&request); err != nil {
+		msg := fmt.Sprintf("cannot marshall params [%s] into %T", c.OriginalURL(), request)
+		ctxLogger.Warn(stacktrace.Propagate(err, msg))
+		return h.responseBadRequest(c, err)
+	}
+
+	if err := request.Validate(); err != nil {
+		msg := fmt.Sprintf("validation errors [%s], while dispatching event [%+#v]", spew.Sdump(err.Error()), request)
+		ctxLogger.Warn(stacktrace.NewError(msg))
+		return h.responseUnprocessableEntity(c, map[string][]string{"event": {err.Error()}}, "validation errors while dispatching event")
+	}
+
+	if h.userIDFomContext(c) != h.queueConfig.UserID {
+		msg := fmt.Sprintf("user with ID [%s], cannot dispatch event [%+#v]", h.userIDFomContext(c), request)
+		ctxLogger.Error(stacktrace.NewError(msg))
+		return h.responseForbidden(c)
+	}
+
+	err := h.service.Dispatch(ctx, request)
+	if err != nil {
+		msg := fmt.Sprintf("cannot dispatch event with ID [%s]", request.ID())
+		ctxLogger.Error(stacktrace.Propagate(err, msg))
+		return h.responseInternalServerError(c)
+	}
+
+	return h.responseNoContent(c, "event dispatched successfully")
+}