tsk-34: Fetch next ready scheduled message (#38)

Closes #34

Reviewed-on: phoenix/textsender-api#38
Co-authored-by: phoenix <kundeng00@pm.me>
Co-committed-by: phoenix <kundeng00@pm.me>

cmd/api/main.go

@@ -64,10 +64,13 @@ func main() {
 	contactStore := store.NewContactStore(db.Pool)
 	messageStore := store.NewMessageStore(db.Pool)
 	schStore := store.NewScheduledMessageStore(db.Pool)
+	schMsgEventStore := store.NewScheduledMessageEventStore(db.Pool)
 
 	contactHandler := handler.NewContactHandler(contactStore)
 	messageHandler := handler.NewMessageHandler(messageStore)
 	scheduledMessageHandler := handler.NewScheduledMessageHandler(schStore)
+	scheduledMessageEventHandler := handler.NewScheduledMessageEventHandler(schMsgEventStore, schStore)
+	scheduledMessageStatusHandler := handler.NewScheduledMessageStatusHandler(schMsgEventStore, schStore)
 
 	router := chi.NewRouter()
 
@@ -81,6 +84,12 @@ func main() {
 	router.Handle(endpoint.ADD_MESSAGE, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(messageHandler.AddMessage)))
 	router.Handle(endpoint.GET_MESSAGE, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(messageHandler.GetMessage)))
 	router.Handle(endpoint.ScheduleMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.AddScheduledMessage)))
+	router.Handle(endpoint.AddEventToScheduledMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageEventHandler.AddScheduledMessageEvent)))
+	router.Method("GET", endpoint.GetScheduledMessageEventEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageEventHandler.GetScheduledMessageEvent)))
+	router.Method("DELETE", endpoint.DeleteScheduledMessageEventEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageEventHandler.DeleteScheduledMessageEvent)))
+	router.Method("PATCH", endpoint.UpdateScheduledMessageStatusEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageStatusHandler.UpdateStatus)))
+	router.Method("GET", endpoint.GetScheduledMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.GetScheduledMessage)))
+	router.Method("GET", endpoint.FetchNextScheduledMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.FetchNextMessage)))
 
 	// Start server
 	server := &http.Server{

cmd/api/main_test.go

@@ -4,44 +4,78 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"net/http"
 	"os"
 	"path"
 	"testing"
 
-	"github.com/gorilla/mux"
+	"github.com/go-chi/chi/v5"
 	"github.com/joho/godotenv"
 
 	"git.kundeng.us/phoenix/textsender-api/internal/config"
 	"git.kundeng.us/phoenix/textsender-api/internal/db"
 	"git.kundeng.us/phoenix/textsender-api/internal/handler"
 	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	mdlware "git.kundeng.us/phoenix/textsender-api/internal/middleware"
+	"git.kundeng.us/phoenix/textsender-api/internal/services"
 	"git.kundeng.us/phoenix/textsender-api/internal/store"
 )
 
-var testRouter *mux.Router
+var testRouter *chi.Mux
 
 func TestMain(m *testing.M) {
 	cfg := load()
 
-	db, err := db.NewDatabase(cfg.GetDBConnString())
+	database, err := db.NewDatabase(cfg.GetDBConnString())
 	if err != nil {
 		fmt.Println(err.Error())
 		panic("Failed to initialize database")
 	}
-	defer db.Close()
+	defer database.Close()
 
 	ctx := context.Background()
-	err = db.ResetDatabase(ctx)
+
-	if err != nil {
+	if cfg.ResetDB {
-		fmt.Println(err.Error())
+		if err = database.ResetDatabase(ctx); err != nil {
-		panic("Failed to initialize database")
+			fmt.Println(err.Error())
+			panic("Failed to initialize database")
+		}
+	} else {
+		if exists, err := db.TableExists(ctx, database.Pool, "contacts "); err == nil && !exists {
+			fmt.Println("Resetting database")
+			err = database.ResetDatabase(ctx)
+			if err != nil {
+				fmt.Printf("Error:%v", err)
+			}
+		} else {
+			fmt.Printf("Error:%v", err)
+		}
 	}
 
-	contactStore := store.NewContactStore(db.Pool)
+	jwtService := services.NewJWTService(cfg.JWTSecret)
-	contactHandler := handler.NewContactHandler(contactStore)
 
-	testRouter = mux.NewRouter()
+	contactStore := store.NewContactStore(database.Pool)
-	testRouter.HandleFunc(endpoint.ADD_CONTACT_ENDPOINT, contactHandler.AddContact).Methods("POST")
+	messageStore := store.NewMessageStore(db.Pool)
+	schStore := store.NewScheduledMessageStore(db.Pool)
+	schMsgEventStore := store.NewScheduledMessageEventStore(db.Pool)
+
+	contactHandler := handler.NewContactHandler(contactStore)
+	messageHandler := handler.NewMessageHandler(messageStore)
+	scheduledMessageHandler := handler.NewScheduledMessageHandler(schStore)
+	scheduledMessageEventHandler := handler.NewScheduledMessageEventHandler(schMsgEventStore, schStore)
+	scheduledMessageStatusHandler := handler.NewScheduledMessageStatusHandler(schMsgEventStore, schStore)
+
+	testRouter = chi.NewRouter()
+	testRouter.Handle(endpoint.ADD_CONTACT_ENDPOINT, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(contactHandler.AddContact)))
+	testRouter.Handle(endpoint.GET_CONTACT, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(contactHandler.GetContact)))
+	testRouter.Handle(endpoint.ADD_MESSAGE, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(messageHandler.AddMessage)))
+	testRouter.Handle(endpoint.GET_MESSAGE, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(messageHandler.GetMessage)))
+	testRouter.Handle(endpoint.ScheduleMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.AddScheduledMessage)))
+	testRouter.Method("GET", endpoint.GetScheduledMessageEventEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageEventHandler.GetScheduledMessageEvent)))
+	testRouter.Method("DELETE", endpoint.DeleteScheduledMessageEventEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageEventHandler.DeleteScheduledMessageEvent)))
+	testRouter.Method("PATCH", endpoint.UpdateScheduledMessageStatusEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageStatusHandler.UpdateStatus)))
+	testRouter.Method("GET", endpoint.GetScheduledMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.GetScheduledMessage)))
+	testRouter.Method("GET", endpoint.FetchNextScheduledMessageEndpoint, mdlware.AuthMiddleware(jwtService)(http.HandlerFunc(scheduledMessageHandler.FetchNextMessage)))
 
 	code := m.Run()
 	os.Exit(code)

go.mod

@@ -3,7 +3,7 @@ module git.kundeng.us/phoenix/textsender-api
 go 1.25.3
 
 require (
-	git.kundeng.us/phoenix/textsender-models v0.0.5-3-ca38ce876f-556
+	git.kundeng.us/phoenix/textsender-models v0.0.6-4-e124bcfda8-556
 	github.com/go-chi/chi/v5 v5.2.3
 	github.com/golang-jwt/jwt/v5 v5.3.0
 	github.com/google/uuid v1.6.0

go.sum

@@ -1,5 +1,5 @@
-git.kundeng.us/phoenix/textsender-models v0.0.5-3-ca38ce876f-556 h1:dDgjre2FlukxeKZfJiP3gtIT4wyzpQ8glAYV9LpLtuU=
+git.kundeng.us/phoenix/textsender-models v0.0.6-4-e124bcfda8-556 h1:2shgOKmglCODswpCJeCuyPiUAph3wfc0Ep1z6b/ifis=
-git.kundeng.us/phoenix/textsender-models v0.0.5-3-ca38ce876f-556/go.mod h1:lx5MCnOgGgsdpwzrfi9uph5xmkeb6H8AuexUNGss2no=
+git.kundeng.us/phoenix/textsender-models v0.0.6-4-e124bcfda8-556/go.mod h1:lx5MCnOgGgsdpwzrfi9uph5xmkeb6H8AuexUNGss2no=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

internal/handler/contact_test.go

@@ -15,6 +15,7 @@ import (
 
 	"git.kundeng.us/phoenix/textsender-api/internal/db"
 	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	"git.kundeng.us/phoenix/textsender-api/internal/store/mock"
 )
 
 type Request struct {
@@ -23,7 +24,7 @@ type Request struct {
 }
 
 func TestCreateContactWithMock(t *testing.T) {
-	mockstore := NewMockContactStore()
+	mockstore := mock.NewMockContactStore()
 	handler := NewContactHandler(mockstore)
 
 	testUserId := uuid.New()
@@ -47,7 +48,7 @@ func TestCreateContactWithMock(t *testing.T) {
 }
 
 func TestGetContactWithMock(t *testing.T) {
-	mockstore := NewMockContactStore()
+	mockstore := mock.NewMockContactStore()
 	testUserId := uuid.New()
 
 	testCon := contact.Contact{PhoneNumber: "+12335403383", UserId: testUserId}

internal/handler/endpoint/endpoint.go

@@ -6,3 +6,9 @@ const GET_MESSAGE = "/api/v1/message"
 const GET_CONTACT = "/api/v1/contact"
 const ADD_CONTACT_ENDPOINT = "/api/v1/contact/new"
 const ScheduleMessageEndpoint = "/api/v1/schedule/message"
+const GetScheduledMessageEndpoint = "/api/v1/schedule/message"
+const AddEventToScheduledMessageEndpoint = "/api/v1/schedule/message/event"
+const GetScheduledMessageEventEndpoint = "/api/v1/schedule/message/event"
+const DeleteScheduledMessageEventEndpoint = "/api/v1/schedule/message/event/{id}"
+const UpdateScheduledMessageStatusEndpoint = "/api/v1/schedule/message/status/update"
+const FetchNextScheduledMessageEndpoint = "/api/v1/schedule/message/fetch"

internal/handler/helper.go

@@ -1,8 +1,11 @@
 package handler
 
-import "encoding/json"
+import (
-import "log"
+	"encoding/json"
-import "net/http"
+	"fmt"
+	"log"
+	"net/http"
+)
 
 func ExtractFromRequest(r *http.Request, reqItem interface{}) error {
 	err := json.NewDecoder(r.Body).Decode(&reqItem)
@@ -21,3 +24,18 @@ func RespondWithJSON(w http.ResponseWriter, statusCode int, data interface{}) {
 		log.Printf("Error encoding JSON: %v", err)
 	}
 }
+
+// Gets the query parameter from the URL
+func ParseQueryParams(r *http.Request, query string) (*string, error) {
+	queryParams := r.URL.Query()
+	if _, exists := queryParams[query]; exists {
+		value := queryParams.Get(query)
+		if len(value) == 0 {
+			return nil, fmt.Errorf("Value of query parameter is empty")
+		} else {
+			return &value, nil
+		}
+	} else {
+		return nil, fmt.Errorf("Could not find query %s", query)
+	}
+}

internal/handler/message_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	"git.kundeng.us/phoenix/textsender-api/internal/store/mock"
 )
 
 type CreateMessageRequest struct {
@@ -21,7 +22,7 @@ type CreateMessageRequest struct {
 }
 
 func TestCreateMessageWithMock(t *testing.T) {
-	mockStore := NewMockMessageStore()
+	mockStore := mock.NewMockMessageStore()
 	handler := NewMessageHandler(mockStore)
 
 	testUserId := uuid.New()
@@ -45,7 +46,7 @@ func TestCreateMessageWithMock(t *testing.T) {
 }
 
 func TestGetMessageWithMock(t *testing.T) {
-	mockstore := NewMockMessageStore()
+	mockstore := mock.NewMockMessageStore()
 	testUserId := uuid.New()
 
 	testCon := message.Message{Content: "Who is the one that benefits?", UserId: testUserId}

internal/handler/schedule_message_status.go

@@ -0,0 +1,154 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/google/uuid"
+
+	"git.kundeng.us/phoenix/textsender-api/internal/store"
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+)
+
+type RequestScheduledMessageStatus struct {
+	ScheduledMessageId uuid.UUID `json:"scheduled_message_id"`
+	Status             string    `json:"status"`
+}
+
+type ScheduledMessageChange struct {
+	OldStatus        string                      `json:"old_status"`
+	ScheduledMessage scheduling.ScheduledMessage `json:"scheduled_message"`
+}
+
+type ScheduledMessageStatusResponse struct {
+	Message string                   `json:"message"`
+	Data    []ScheduledMessageChange `json:"data"`
+}
+
+type ScheduledMessageStatusHandler struct {
+	ScheduledMessageEventStore store.ScheduledMessageEventStore
+	ScheduledMessageStore      store.ScheduledMessageStore
+}
+
+func NewScheduledMessageStatusHandler(str store.ScheduledMessageEventStore, schStore store.ScheduledMessageStore) *ScheduledMessageStatusHandler {
+	return &ScheduledMessageStatusHandler{ScheduledMessageEventStore: str, ScheduledMessageStore: schStore}
+}
+
+func (s *ScheduledMessageStatusHandler) UpdateStatus(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPatch {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req RequestScheduledMessageStatus
+	if err := ExtractFromRequest(r, &req); err != nil {
+		http.Error(w, "Invalid JSON: "+err.Error(), http.StatusBadRequest)
+	}
+	defer r.Body.Close()
+
+	var statusCode int
+	var resp ScheduledMessageStatusResponse
+
+	if req.ScheduledMessageId == uuid.Nil {
+		statusCode = http.StatusBadRequest
+		resp.Message = "Scheduled messaged Id is nil"
+	} else if len(req.Status) == 0 {
+		statusCode = http.StatusBadRequest
+		resp.Message = "No status provided"
+	} else {
+		ctx := r.Context()
+
+		if schMsg, err := s.ScheduledMessageStore.Get(ctx, req.ScheduledMessageId); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			if schMsg.Status == scheduling.Done {
+				statusCode = http.StatusNotFound
+				resp.Message = "Scheduled message has already completed"
+			} else {
+				switch req.Status {
+				case scheduling.Pending:
+					{
+						if schMsg.Status == scheduling.Processing {
+							statusCode = http.StatusBadRequest
+							resp.Message = "Message is currently processing"
+						} else {
+							if returnedStatus, err := s.ScheduledMessageStore.UpdateStatus(ctx, schMsg.Id, req.Status); err != nil {
+								statusCode = http.StatusInternalServerError
+								resp.Message = err.Error()
+							} else {
+								chg := initScheduledMessageChange(schMsg, schMsg.Status, *returnedStatus)
+								statusCode = http.StatusOK
+								resp.Message = "Successful"
+								resp.Data = append(resp.Data, chg)
+							}
+						}
+					}
+				case scheduling.Ready:
+					{
+						if events, err := s.ScheduledMessageEventStore.GetWithScheduleMessageId(ctx, schMsg.Id); err != nil {
+							statusCode = http.StatusInternalServerError
+							resp.Message = err.Error()
+						} else {
+							oldStatus := schMsg.Status
+							if schMsg.Status != scheduling.Processing && schMsg.Status != scheduling.Ready {
+								// Update status
+								if len(events) > 0 {
+									if returnedStatus, err := s.ScheduledMessageStore.UpdateStatus(ctx, schMsg.Id, req.Status); err != nil {
+										statusCode = http.StatusInternalServerError
+										resp.Message = err.Error()
+									} else {
+										chg := initScheduledMessageChange(schMsg, oldStatus, *returnedStatus)
+										statusCode = http.StatusOK
+										resp.Message = "Successful"
+										resp.Data = append(resp.Data, chg)
+									}
+								} else {
+									statusCode = http.StatusInternalServerError
+									resp.Message = "Not enough scheduled message events"
+								}
+							} else {
+								if schMsg.Status == scheduling.Processing {
+									statusCode = http.StatusBadRequest
+									resp.Message = "Scheduled messages are processing"
+								} else {
+									statusCode = http.StatusNotModified
+									resp.Message = "Status is already set"
+								}
+							}
+						}
+					}
+				case scheduling.Processing, scheduling.Done:
+					{
+						if events, err := s.ScheduledMessageEventStore.GetWithScheduleMessageId(ctx, schMsg.Id); err != nil {
+							statusCode = http.StatusInternalServerError
+							resp.Message = err.Error()
+						} else {
+							if len(events) > 0 {
+								if returnedStatus, err := s.ScheduledMessageStore.UpdateStatus(ctx, schMsg.Id, req.Status); err != nil {
+									statusCode = http.StatusInternalServerError
+									resp.Message = err.Error()
+								} else {
+									chg := initScheduledMessageChange(schMsg, schMsg.Status, *returnedStatus)
+									statusCode = http.StatusOK
+									resp.Message = "Successful"
+									resp.Data = append(resp.Data, chg)
+								}
+							} else {
+								statusCode = http.StatusInternalServerError
+								resp.Message = "Not enough scheduled message events"
+							}
+						}
+					}
+				}
+			}
+		}
+
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}
+
+func initScheduledMessageChange(schMsg *scheduling.ScheduledMessage, oldStatus, newStatus string) ScheduledMessageChange {
+	schMsg.Status = newStatus
+	return ScheduledMessageChange{OldStatus: oldStatus, ScheduledMessage: *schMsg}
+}

internal/handler/schedule_message_status_test.go

@@ -0,0 +1,73 @@
+package handler
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+
+	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	"git.kundeng.us/phoenix/textsender-api/internal/store/mock"
+)
+
+func TestUpdateScheduledMessageStatusWithMock(t *testing.T) {
+	now := time.Now()
+
+	schMsgEventStore := mock.NewMockScheduledMessageEventStore()
+	contactStore := mock.NewMockContactStore()
+	messageStore := mock.NewMockMessageStore()
+	schMsgStore := mock.NewMockScheduledMessageStore()
+
+	recipientId := uuid.New()
+	messageId := uuid.New()
+	scheduledMessageId := uuid.New()
+	testUserId := uuid.New()
+
+	con := testContact(recipientId, testUserId)
+	msg := testMessage(messageId, testUserId)
+	schMsg := testScheduledMessage(scheduledMessageId, testUserId, now)
+	event := testScheduledMessageEvent(msg.Id, con.Id, schMsg.Id)
+
+	ctx := t.Context()
+
+	if err := contactStore.CreateContact(ctx, &con); err != nil {
+		assert.NoError(t, err, "Error creating contact: %v", err)
+	} else if err = messageStore.CreateMessage(ctx, &msg); err != nil {
+		assert.NoError(t, err, "Error creating message: %v", err)
+	} else if err = schMsgStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message: %v", err)
+	} else if err = schMsgEventStore.CreateScheduledMessageEvent(ctx, &event); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message event: %v", err)
+	}
+
+	testReq := RequestScheduledMessageStatus{}
+	testReq.Status = scheduling.Ready
+	testReq.ScheduledMessageId = schMsg.Id
+	jsonValue, _ := json.Marshal(testReq)
+	jsonBody := strings.NewReader(string(jsonValue))
+
+	req, _ := http.NewRequest("PATCH", endpoint.UpdateScheduledMessageStatusEndpoint, jsonBody)
+	rr := httptest.NewRecorder()
+
+	handler := NewScheduledMessageStatusHandler(schMsgEventStore, schMsgStore)
+	handler.UpdateStatus(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	var response ScheduledMessageStatusResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error creating event %v", err)
+
+	assert.NotEmpty(t, response.Data, "Data should not be empty")
+
+	changes := response.Data[0]
+	assert.NotEqual(t, changes.OldStatus, changes.ScheduledMessage.Status, "The status should not match")
+	assert.Equal(t, scheduling.Pending, changes.OldStatus, "The Old status does not match Old %s New %s", changes.OldStatus, scheduling.Pending)
+	assert.Equal(t, scheduling.Ready, changes.ScheduledMessage.Status, "Status has not been updated")
+}

internal/handler/scheduled_message.go

@@ -24,6 +24,16 @@ type AddScheduledMessageResponse struct {
 	Data    []scheduling.ScheduledMessage `json:"data"`
 }
 
+type GetScheduledMessageResponse struct {
+	Message string                        `json:"message"`
+	Data    []scheduling.ScheduledMessage `json:"data"`
+}
+
+type FetchNextMessageResponse struct {
+	Message string                        `json:"message"`
+	Data    []scheduling.ScheduledMessage `json:"data"`
+}
+
 type ScheduledMessageHandler struct {
 	ScheduledMessageStore store.ScheduledMessageStore
 }
@@ -32,7 +42,7 @@ func NewScheduledMessageHandler(str store.ScheduledMessageStore) *ScheduledMessa
 	return &ScheduledMessageHandler{ScheduledMessageStore: str}
 }
 
-func (c *ScheduledMessageHandler) AddScheduledMessage(w http.ResponseWriter, r *http.Request) {
+func (s *ScheduledMessageHandler) AddScheduledMessage(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 		return
@@ -55,10 +65,10 @@ func (c *ScheduledMessageHandler) AddScheduledMessage(w http.ResponseWriter, r *
 	} else {
 		ctx := r.Context()
 
-		if validStatus, err := isStatusValid(&scheduledMessage); err == nil {
+		if validStatus, err := IsStatusValid(&scheduledMessage); err == nil {
 			if validStatus {
 				if valid, err := isScheduledTimeValid(&scheduledMessage); err == nil && valid {
-					if err = c.ScheduledMessageStore.CreateScheduledMessage(ctx, &scheduledMessage); err == nil {
+					if err = s.ScheduledMessageStore.CreateScheduledMessage(ctx, &scheduledMessage); err == nil {
 						statusCode = http.StatusCreated
 						resp.Data = append(resp.Data, scheduledMessage)
 						resp.Message = "Successful"
@@ -83,6 +93,92 @@ func (c *ScheduledMessageHandler) AddScheduledMessage(w http.ResponseWriter, r *
 	RespondWithJSON(w, statusCode, &resp)
 }
 
+func (s *ScheduledMessageHandler) GetScheduledMessage(w http.ResponseWriter, r *http.Request) {
+	var id, userId uuid.UUID
+
+	if idParam, err := ParseQueryParams(r, "id"); err == nil {
+		var err error
+		id, err = uuid.Parse(*idParam)
+		if err != nil {
+			http.Error(w, "Error parsing Id", http.StatusBadRequest)
+			return
+		}
+	}
+	if userIdParam, err := ParseQueryParams(r, "user_id"); err == nil {
+		userId, err = uuid.Parse(*userIdParam)
+		if err != nil {
+			http.Error(w, "Error parsing Id", http.StatusBadRequest)
+			return
+		}
+	}
+
+	var resp GetScheduledMessageResponse
+	var statusCode int
+	ctx := r.Context()
+
+	if userId == uuid.Nil && id == uuid.Nil {
+		http.Error(w, "Invalid query parameters", http.StatusBadRequest)
+		return
+	} else if userId != uuid.Nil && id != uuid.Nil {
+		if schMsgs, err := s.ScheduledMessageStore.GetWithUserId(ctx, userId); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			for _, schMsg := range schMsgs {
+				if schMsg.Id == id {
+					resp.Data = append(resp.Data, *schMsg)
+				}
+			}
+			statusCode = http.StatusOK
+			resp.Message = "Successful"
+		}
+	} else if id != uuid.Nil {
+		if schMsg, err := s.ScheduledMessageStore.Get(ctx, id); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			statusCode = http.StatusOK
+			resp.Message = "Successful"
+			resp.Data = append(resp.Data, *schMsg)
+		}
+	} else {
+		if schMsgs, err := s.ScheduledMessageStore.GetWithUserId(ctx, userId); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			statusCode = http.StatusOK
+			resp.Message = "Successful"
+			for _, schMsg := range schMsgs {
+				resp.Data = append(resp.Data, *schMsg)
+			}
+		}
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}
+
+func (s *ScheduledMessageHandler) FetchNextMessage(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	var resp FetchNextMessageResponse
+	var statusCode int
+
+	if schMsg, err := s.ScheduledMessageStore.FetchNextScheduledMessage(ctx); err != nil {
+		statusCode = http.StatusInternalServerError
+		resp.Message = err.Error()
+	} else {
+		if schMsg != nil {
+			statusCode = http.StatusOK
+			resp.Message = "Successful"
+			resp.Data = append(resp.Data, *schMsg)
+		} else {
+			statusCode = http.StatusNotFound
+			resp.Message = "No item"
+		}
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}
+
 func isScheduledTimeValid(schMsg *scheduling.ScheduledMessage) (bool, error) {
 	now := time.Now()
 	timeCutOff := now.Add(-5 * time.Minute)
@@ -99,7 +195,7 @@ func isScheduledTimeValid(schMsg *scheduling.ScheduledMessage) (bool, error) {
 	}
 }
 
-func isStatusValid(schMsg *scheduling.ScheduledMessage) (bool, error) {
+func IsStatusValid(schMsg *scheduling.ScheduledMessage) (bool, error) {
 	schMsg.Status = strings.ToUpper(schMsg.Status)
 	if schMsg.Status == scheduling.Pending {
 		return true, nil

internal/handler/scheduled_message_event.go

@@ -0,0 +1,208 @@
+package handler
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/google/uuid"
+
+	"git.kundeng.us/phoenix/textsender-api/internal/store"
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+)
+
+type RequestAddScheduledMessageEvent struct {
+	RecipientId        uuid.UUID `json:"recipient_id"`
+	MessageId          uuid.UUID `json:"message_id"`
+	ScheduledMessageId uuid.UUID `json:"scheduled_message_id"`
+}
+
+type AddScheduledMessageEventResponse struct {
+	Message string                             `json:"message"`
+	Data    []scheduling.ScheduledMessageEvent `json:"data"`
+}
+
+type GetScheduledMessageEventResponse struct {
+	Message string                             `json:"message"`
+	Data    []scheduling.ScheduledMessageEvent `json:"data"`
+}
+
+type DeleteScheduledMessageEventResponse struct {
+	Message string                             `json:"message"`
+	Data    []scheduling.ScheduledMessageEvent `json:"data"`
+}
+
+type ScheduledMessageEventHandler struct {
+	ScheduledMessageEventStore store.ScheduledMessageEventStore
+	ScheduledMessageStore      store.ScheduledMessageStore
+}
+
+func NewScheduledMessageEventHandler(str store.ScheduledMessageEventStore, schStore store.ScheduledMessageStore) *ScheduledMessageEventHandler {
+	return &ScheduledMessageEventHandler{ScheduledMessageEventStore: str, ScheduledMessageStore: schStore}
+}
+
+func (s *ScheduledMessageEventHandler) AddScheduledMessageEvent(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req RequestAddScheduledMessageEvent
+	if err := ExtractFromRequest(r, &req); err != nil {
+		http.Error(w, "Invalid JSON: "+err.Error(), http.StatusBadRequest)
+	}
+	defer r.Body.Close()
+
+	event := scheduling.ScheduledMessageEvent{RecipientId: req.RecipientId, MessageId: req.MessageId, ScheduledMessageId: req.ScheduledMessageId}
+
+	var statusCode int
+	var resp AddScheduledMessageEventResponse
+
+	if event.RecipientId == uuid.Nil {
+		statusCode = http.StatusBadRequest
+		resp.Message = "Recipient Id is nil"
+	} else if event.MessageId == uuid.Nil {
+		statusCode = http.StatusBadRequest
+		resp.Message = "Message Id is nil"
+	} else if event.ScheduledMessageId == uuid.Nil {
+		statusCode = http.StatusBadRequest
+		resp.Message = "Scheduled Message Id is nil"
+	} else {
+		ctx := r.Context()
+
+		if schMsg, err := s.ScheduledMessageStore.Get(ctx, event.ScheduledMessageId); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			if schMsg.Status == scheduling.Pending {
+				if exists, err := s.ScheduledMessageEventStore.Exists(ctx, &event); err == nil {
+					if exists {
+						statusCode = http.StatusBadRequest
+						resp.Message = "Event already present"
+					} else {
+						if err = s.ScheduledMessageEventStore.CreateScheduledMessageEvent(ctx, &event); err == nil {
+							statusCode = http.StatusCreated
+							resp.Message = "Successful"
+							resp.Data = append(resp.Data, event)
+						} else {
+							statusCode = http.StatusInternalServerError
+							resp.Message = err.Error()
+						}
+					}
+				} else {
+					statusCode = http.StatusInternalServerError
+					resp.Message = err.Error()
+				}
+			} else {
+				statusCode = http.StatusBadRequest
+				resp.Message = "Scheduled message cannot be modified due to the status"
+			}
+		}
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}
+
+func (s *ScheduledMessageEventHandler) GetScheduledMessageEvent(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Metnot allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var id, scheduledMessageId uuid.UUID
+	if idParam, err := ParseQueryParams(r, "id"); err == nil {
+		id, err = uuid.Parse(*idParam)
+	}
+	if scheduledMessageIdParam, err := ParseQueryParams(r, "scheduled_message_id"); err == nil {
+		scheduledMessageId, err = uuid.Parse(*scheduledMessageIdParam)
+	}
+
+	var statusCode int
+	var resp GetScheduledMessageEventResponse
+	ctx := r.Context()
+
+	if id == uuid.Nil && scheduledMessageId == uuid.Nil {
+		statusCode = http.StatusBadRequest
+		resp.Message = "Query parameters missing"
+	} else if id != uuid.Nil {
+		if event, err := s.ScheduledMessageEventStore.Get(ctx, id); err != nil {
+			resp.Message = err.Error()
+			statusCode = http.StatusInternalServerError
+		} else {
+			if event != nil {
+				resp.Message = "Successful"
+				statusCode = http.StatusOK
+				resp.Data = append(resp.Data, *event)
+			} else {
+				statusCode = http.StatusNotFound
+				resp.Message = "Scheduled message event not found"
+			}
+		}
+	} else {
+		if events, err := s.ScheduledMessageEventStore.GetWithScheduleMessageId(ctx, scheduledMessageId); err != nil {
+			statusCode = http.StatusInternalServerError
+			resp.Message = err.Error()
+		} else {
+			statusCode = http.StatusOK
+			resp.Message = "Successful"
+			for _, event := range events {
+				resp.Data = append(resp.Data, *event)
+			}
+		}
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}
+
+func (s *ScheduledMessageEventHandler) DeleteScheduledMessageEvent(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodDelete {
+		http.Error(w, "Metnot allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	id := chi.URLParam(r, "id")
+	if len(id) == 0 {
+		pathParts := strings.Split(r.URL.Path, "/")
+		if len(pathParts) < 7 {
+			http.Error(w, "Id not provided", http.StatusBadRequest)
+			return
+		} else {
+			id = pathParts[6]
+		}
+	}
+
+	var resp DeleteScheduledMessageEventResponse
+	var statusCode int
+
+	if parsedId, err := uuid.Parse(id); err != nil {
+		resp.Message = err.Error()
+		statusCode = http.StatusBadRequest
+	} else {
+		ctx := r.Context()
+		if event, err := s.ScheduledMessageEventStore.Get(ctx, parsedId); err != nil {
+			resp.Message = err.Error()
+			statusCode = http.StatusInternalServerError
+		} else {
+			if schMsg, err := s.ScheduledMessageStore.Get(ctx, event.ScheduledMessageId); err != nil {
+				resp.Message = err.Error()
+				statusCode = http.StatusInternalServerError
+			} else {
+				if schMsg.Status == scheduling.Pending {
+					if err = s.ScheduledMessageEventStore.Delete(ctx, event.Id); err != nil {
+						resp.Message = err.Error()
+						statusCode = http.StatusInternalServerError
+					} else {
+						resp.Message = "Successful"
+						resp.Data = append(resp.Data, *event)
+						statusCode = http.StatusOK
+					}
+				} else {
+					statusCode = http.StatusBadRequest
+					resp.Message = "Invalid status"
+				}
+			}
+		}
+	}
+
+	RespondWithJSON(w, statusCode, &resp)
+}

internal/handler/scheduled_message_event_test.go

@@ -0,0 +1,206 @@
+package handler
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"git.kundeng.us/phoenix/textsender-models/pkg/contact"
+	"git.kundeng.us/phoenix/textsender-models/pkg/message"
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+
+	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	"git.kundeng.us/phoenix/textsender-api/internal/store/mock"
+)
+
+var (
+	recipientId        = uuid.New()
+	messageId          = uuid.New()
+	scheduledMessageId = uuid.New()
+	testUserId         = uuid.New()
+)
+
+type CreateScheduledMessageEventRequest struct {
+	RecipientId        uuid.UUID `json:"recipient_id"`
+	MessageId          uuid.UUID `json:"message_id"`
+	ScheduledMessageId uuid.UUID `json:"scheduled_message_id"`
+}
+
+func TestCreateScheduledMessageEventWithMock(t *testing.T) {
+	now := time.Now()
+
+	mockStore := mock.NewMockScheduledMessageEventStore()
+	contactStore := mock.NewMockContactStore()
+	messageStore := mock.NewMockMessageStore()
+	schMsgStore := mock.NewMockScheduledMessageStore()
+
+	handler := NewScheduledMessageEventHandler(mockStore, schMsgStore)
+
+	recipientId := uuid.New()
+	messageId := uuid.New()
+	scheduledMessageId := uuid.New()
+	testUserId := uuid.New()
+
+	con := testContact(recipientId, testUserId)
+	msg := testMessage(messageId, testUserId)
+	schMsg := testScheduledMessage(scheduledMessageId, testUserId, now)
+
+	ctx := t.Context()
+
+	if err := contactStore.CreateContact(ctx, &con); err != nil {
+		assert.NoError(t, err, "Error creating contact: %v", err)
+	} else if err = messageStore.CreateMessage(ctx, &msg); err != nil {
+		assert.NoError(t, err, "Error creating message: %v", err)
+	} else if err = schMsgStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message: %v", err)
+	}
+
+	testReq := CreateScheduledMessageEventRequest{RecipientId: recipientId, MessageId: messageId, ScheduledMessageId: scheduledMessageId}
+	jsonValue, _ := json.Marshal(testReq)
+
+	req, _ := http.NewRequest("POST", endpoint.AddEventToScheduledMessageEndpoint, strings.NewReader(string(jsonValue)))
+	rr := httptest.NewRecorder()
+
+	handler.AddScheduledMessageEvent(rr, req)
+
+	assert.Equal(t, http.StatusCreated, rr.Code)
+
+	var response AddScheduledMessageEventResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error creating event %v", err)
+
+	assert.NotEmpty(t, response.Data, "No event created")
+
+	var event scheduling.ScheduledMessageEvent
+	event = response.Data[0]
+
+	assert.NotEmpty(t, event.Created, "Created date should not be empty")
+}
+
+func TestGetScheduledMessageEventWithMock(t *testing.T) {
+	now := time.Now()
+
+	contactStore := mock.NewMockContactStore()
+	messageStore := mock.NewMockMessageStore()
+	schMsgStore := mock.NewMockScheduledMessageStore()
+	schMsgEventStore := mock.NewMockScheduledMessageEventStore()
+	handler := NewScheduledMessageEventHandler(schMsgEventStore, schMsgStore)
+
+	recipientId := uuid.New()
+	messageId := uuid.New()
+	scheduledMessageId := uuid.New()
+	testUserId := uuid.New()
+
+	con := testContact(recipientId, testUserId)
+	msg := testMessage(messageId, testUserId)
+	schMsg := testScheduledMessage(scheduledMessageId, testUserId, now)
+	event := testScheduledMessageEvent(msg.Id, con.Id, schMsg.Id)
+
+	ctx := t.Context()
+
+	if err := contactStore.CreateContact(ctx, &con); err != nil {
+		assert.NoError(t, err, "Error creating contact: %v", err)
+	} else if err = messageStore.CreateMessage(ctx, &msg); err != nil {
+		assert.NoError(t, err, "Error creating message: %v", err)
+	} else if err = schMsgStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message: %v", err)
+	} else if err = schMsgEventStore.CreateScheduledMessageEvent(ctx, &event); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message event: %v", err)
+	}
+
+	url := fmt.Sprintf("%s?id=%s", endpoint.GetScheduledMessageEventEndpoint, event.Id.String())
+	req, _ := http.NewRequest("GET", url, nil)
+	rr := httptest.NewRecorder()
+
+	handler.GetScheduledMessageEvent(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	var response GetScheduledMessageEventResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error creating event %v", err)
+
+	assert.NotEmpty(t, response.Data, "No event created")
+
+	msgEvent := response.Data[0]
+
+	assert.NotEmpty(t, msgEvent.Created, "Created date should not be empty")
+}
+
+func TestDeleteScheduledMessageEventWithMock(t *testing.T) {
+	now := time.Now()
+	contactStore := mock.NewMockContactStore()
+	messageStore := mock.NewMockMessageStore()
+	schMsgStore := mock.NewMockScheduledMessageStore()
+	schMsgEventStore := mock.NewMockScheduledMessageEventStore()
+	handler := NewScheduledMessageEventHandler(schMsgEventStore, schMsgStore)
+
+	con := testContact(recipientId, testUserId)
+	msg := testMessage(messageId, testUserId)
+	schMsg := testScheduledMessage(scheduledMessageId, testUserId, now)
+	event := testScheduledMessageEvent(msg.Id, con.Id, schMsg.Id)
+
+	ctx := t.Context()
+
+	if err := contactStore.CreateContact(ctx, &con); err != nil {
+		assert.NoError(t, err, "Error creating contact: %v", err)
+	} else if err = messageStore.CreateMessage(ctx, &msg); err != nil {
+		assert.NoError(t, err, "Error creating message: %v", err)
+	} else if err = schMsgStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message: %v", err)
+	} else if err = schMsgEventStore.CreateScheduledMessageEvent(ctx, &event); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message event: %v", err)
+	}
+
+	endpointValue := strings.Replace(endpoint.DeleteScheduledMessageEventEndpoint, "{id}", event.Id.String(), 1)
+	req, _ := http.NewRequest("DELETE", endpointValue, nil)
+	rr := httptest.NewRecorder()
+
+	handler.DeleteScheduledMessageEvent(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	var response DeleteScheduledMessageEventResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error creating event %v", err)
+
+	assert.NotEmpty(t, response.Data, "No event created")
+
+	msgEvent := response.Data[0]
+
+	assert.NotEmpty(t, msgEvent.Created, "Created date should not be empty")
+}
+
+func testContact(id uuid.UUID, userId uuid.UUID) contact.Contact {
+	if id == uuid.Nil {
+		return contact.Contact{Id: uuid.New(), PhoneNumber: "+10123456789", UserId: userId}
+	} else {
+		return contact.Contact{Id: id, PhoneNumber: "+10123456789", UserId: userId}
+	}
+}
+
+func testMessage(id uuid.UUID, userId uuid.UUID) message.Message {
+	if id == uuid.Nil {
+		return message.Message{Id: uuid.New(), Content: "Oh how the mighty have fallen", UserId: userId}
+	} else {
+		return message.Message{Id: id, Content: "Oh how the mighty have fallen", UserId: userId}
+	}
+}
+
+func testScheduledMessage(id uuid.UUID, userId uuid.UUID, now time.Time) scheduling.ScheduledMessage {
+	if id == uuid.Nil {
+		return scheduling.ScheduledMessage{Id: uuid.New(), UserId: userId, Scheduled: now.Add(20 * time.Minute), Status: scheduling.Pending}
+	} else {
+		return scheduling.ScheduledMessage{Id: id, UserId: userId, Scheduled: now.Add(20 * time.Minute), Status: scheduling.Pending}
+	}
+}
+
+func testScheduledMessageEvent(messageId, recipientId, scheduledMessageId uuid.UUID) scheduling.ScheduledMessageEvent {
+	return scheduling.ScheduledMessageEvent{MessageId: messageId, RecipientId: recipientId, ScheduledMessageId: scheduledMessageId}
+}

internal/handler/scheduled_message_test.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -13,6 +14,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"git.kundeng.us/phoenix/textsender-api/internal/handler/endpoint"
+	"git.kundeng.us/phoenix/textsender-api/internal/store/mock"
 )
 
 type CreateScheduledMessageRequest struct {
@@ -23,12 +25,11 @@ type CreateScheduledMessageRequest struct {
 
 func TestCreateScheduledMessageWithMock(t *testing.T) {
 	now := time.Now()
-	mockStore := NewMockScheduledMessageStore()
+	mockStore := mock.NewMockScheduledMessageStore()
 	handler := NewScheduledMessageHandler(mockStore)
 
 	testUserId := uuid.New()
-	scheduled := now.Add(5 * time.Minute)
+	testBody := testCreateScheduledMessageRequest(testUserId, now)
-	testBody := CreateScheduledMessageRequest{Status: scheduling.Pending, UserId: testUserId, Scheduled: scheduled}
 	jsonValue, _ := json.Marshal(testBody)
 
 	req, _ := http.NewRequest("POST", endpoint.ScheduleMessageEndpoint, strings.NewReader(string(jsonValue)))
@@ -46,3 +47,89 @@ func TestCreateScheduledMessageWithMock(t *testing.T) {
 
 	assert.NotNil(t, response.Data[0].Id, "Id should not be nil")
 }
+
+func TestGetScheduledMessageWithMock(t *testing.T) {
+	now := time.Now()
+	mockStore := mock.NewMockScheduledMessageStore()
+
+	testUserId := uuid.New()
+	testBody := testCreateScheduledMessageRequest(testUserId, now)
+
+	ctx := t.Context()
+	schMsg := scheduling.ScheduledMessage{}
+	schMsg.Scheduled = testBody.Scheduled
+	schMsg.Status = testBody.Status
+	schMsg.UserId = testBody.UserId
+	if err := mockStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error Creating message %v", err)
+	}
+
+	url := fmt.Sprintf("%s?id=%s&user_id=%s", endpoint.GetScheduledMessageEndpoint, schMsg.Id, schMsg.UserId)
+	req, _ := http.NewRequest("GET", url, nil)
+	rr := httptest.NewRecorder()
+
+	handler := NewScheduledMessageHandler(mockStore)
+	handler.GetScheduledMessage(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	var response GetScheduledMessageResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error parsing response %v", err)
+}
+
+func TestFetchScheduledMessageWithMock(t *testing.T) {
+	now := time.Now()
+	mockStore := mock.NewMockScheduledMessageStore()
+	contactStore := mock.NewMockContactStore()
+	messageStore := mock.NewMockMessageStore()
+	schMsgEventStore := mock.NewMockScheduledMessageEventStore()
+
+	recipientId := uuid.New()
+	messageId := uuid.New()
+	scheduledMessageId := uuid.New()
+	testUserId := uuid.New()
+
+	con := testContact(recipientId, testUserId)
+	msg := testMessage(messageId, testUserId)
+	schMsg := testScheduledMessage(scheduledMessageId, testUserId, now)
+	event := testScheduledMessageEvent(msg.Id, con.Id, schMsg.Id)
+
+	ctx := t.Context()
+
+	if err := contactStore.CreateContact(ctx, &con); err != nil {
+		assert.NoError(t, err, "Error creating contact: %v", err)
+	} else if err = messageStore.CreateMessage(ctx, &msg); err != nil {
+		assert.NoError(t, err, "Error creating message: %v", err)
+	} else if err = mockStore.CreateScheduledMessage(ctx, &schMsg); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message: %v", err)
+	} else if err = schMsgEventStore.CreateScheduledMessageEvent(ctx, &event); err != nil {
+		assert.NoError(t, err, "Error creating scheduled message event: %v", err)
+	} else if newStatus, err := mockStore.UpdateStatus(ctx, schMsg.Id, scheduling.Ready); err != nil {
+		assert.NoError(t, err, "Error updating status: %v", err)
+	} else {
+		assert.Equal(t, *newStatus, scheduling.Ready)
+	}
+
+	req, _ := http.NewRequest("GET", endpoint.FetchNextScheduledMessageEndpoint, nil)
+	rr := httptest.NewRecorder()
+
+	handler := NewScheduledMessageHandler(mockStore)
+	handler.FetchNextMessage(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	var response FetchNextMessageResponse
+	err := json.Unmarshal(rr.Body.Bytes(), &response)
+	assert.NoError(t, err, "Error fetching scheduled message: %v", err)
+
+	assert.NotEmpty(t, response.Data, "No data")
+
+	fetchedSchMsg := response.Data[0]
+	assert.Equal(t, scheduling.Processing, fetchedSchMsg.Status, "The statuses do not match")
+}
+
+func testCreateScheduledMessageRequest(userId uuid.UUID, now time.Time) CreateScheduledMessageRequest {
+	scheduled := now.Add(5 * time.Minute)
+	return CreateScheduledMessageRequest{Status: scheduling.Pending, UserId: userId, Scheduled: scheduled}
+}

internal/store/mock/mock_store.go

@@ -1,4 +1,4 @@
-package handler
+package mock
 
 import (
 	"context"
@@ -9,7 +9,6 @@ import (
 
 	"git.kundeng.us/phoenix/textsender-models/pkg/contact"
 	"git.kundeng.us/phoenix/textsender-models/pkg/message"
-	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
 )
 
 type Key struct {
@@ -201,40 +200,3 @@ func (m *MockMessageStore) MessageExists(ctx context.Context, msg *message.Messa
 
 	return exists, nil
 }
-
-type ScheduledMessageKey struct {
-	UserId uuid.UUID
-}
-
-type MockScheduledMessageStore struct {
-	ScheduledMessages      map[uuid.UUID]*scheduling.ScheduledMessage
-	ScheduledMessagesByKey map[ScheduledMessageKey]*scheduling.ScheduledMessage
-	mu                     sync.RWMutex
-	Error                  error // Optional: simulate errors
-}
-
-func NewMockScheduledMessageStore() *MockScheduledMessageStore {
-	return &MockScheduledMessageStore{
-		ScheduledMessages:      make(map[uuid.UUID]*scheduling.ScheduledMessage),
-		ScheduledMessagesByKey: make(map[ScheduledMessageKey]*scheduling.ScheduledMessage),
-	}
-}
-
-func (m *MockScheduledMessageStore) CreateScheduledMessage(ctx context.Context, schedMsg *scheduling.ScheduledMessage) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if m.Error != nil {
-		return m.Error
-	}
-
-	if schedMsg.Id == uuid.Nil {
-		schedMsg.Id = uuid.New()
-	}
-
-	key := ScheduledMessageKey{UserId: schedMsg.UserId}
-
-	m.ScheduledMessages[schedMsg.Id] = schedMsg
-	m.ScheduledMessagesByKey[key] = schedMsg
-	return nil
-}

internal/store/mock/scheduled_message_event_store.go

@@ -0,0 +1,160 @@
+package mock
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+)
+
+type ScheduledMessageEventKey struct {
+	RecipientId        uuid.UUID
+	MessageId          uuid.UUID
+	ScheduledMessageId uuid.UUID
+}
+
+type MockScheduledMessageEventStore struct {
+	ScheduledMessageEvents      map[uuid.UUID]*scheduling.ScheduledMessageEvent
+	ScheduledMessageEventsByKey map[ScheduledMessageEventKey]*scheduling.ScheduledMessageEvent
+	mu                          sync.RWMutex
+	Error                       error // Optional: simulate errors
+}
+
+func NewMockScheduledMessageEventStore() *MockScheduledMessageEventStore {
+	return &MockScheduledMessageEventStore{
+		ScheduledMessageEvents:      make(map[uuid.UUID]*scheduling.ScheduledMessageEvent),
+		ScheduledMessageEventsByKey: make(map[ScheduledMessageEventKey]*scheduling.ScheduledMessageEvent),
+	}
+}
+
+func (m *MockScheduledMessageEventStore) Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessageEvent, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	if id == uuid.Nil {
+		return nil, fmt.Errorf("Id is nil")
+	}
+
+	if _, exists := m.ScheduledMessageEvents[id]; exists {
+		return m.ScheduledMessageEvents[id], nil
+	} else {
+		return nil, fmt.Errorf("Not found")
+	}
+}
+
+func (m *MockScheduledMessageEventStore) GetWithScheduleMessageId(ctx context.Context, schMsgId uuid.UUID) ([]*scheduling.ScheduledMessageEvent, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	if schMsgId == uuid.Nil {
+		return nil, fmt.Errorf("Scheduled message Id is nil")
+	}
+
+	var events []*scheduling.ScheduledMessageEvent
+
+	for _, event := range m.ScheduledMessageEvents {
+		if event.ScheduledMessageId == schMsgId {
+			events = append(events, event)
+		}
+	}
+
+	if len(events) > 0 {
+		return events, nil
+	} else {
+		return nil, nil
+	}
+}
+
+func (m *MockScheduledMessageEventStore) Delete(ctx context.Context, id uuid.UUID) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return m.Error
+	}
+
+	if id == uuid.Nil {
+		return fmt.Errorf("Id is nil")
+	}
+
+	if _, exists := m.ScheduledMessageEvents[id]; exists {
+		originalAmount := len(m.ScheduledMessageEvents)
+		copiedEvents := make(map[uuid.UUID]*scheduling.ScheduledMessageEvent)
+		copiedEventsKey := make(map[ScheduledMessageEventKey]*scheduling.ScheduledMessageEvent)
+
+		for i, schMsgEvent := range m.ScheduledMessageEvents {
+			key := ScheduledMessageEventKey{RecipientId: schMsgEvent.RecipientId, MessageId: schMsgEvent.MessageId, ScheduledMessageId: schMsgEvent.ScheduledMessageId}
+			if schMsgEvent.Id != id {
+				copiedEvents[i] = schMsgEvent
+				copiedEventsKey[key] = schMsgEvent
+			}
+		}
+
+		if originalAmount > 1 {
+			if len(copiedEvents) > 0 && len(copiedEventsKey) > 0 {
+				m.ScheduledMessageEvents = copiedEvents
+				m.ScheduledMessageEventsByKey = copiedEventsKey
+				return nil
+			} else {
+				return fmt.Errorf("Not removed")
+			}
+		} else {
+			m.ScheduledMessageEvents = make(map[uuid.UUID]*scheduling.ScheduledMessageEvent)
+			m.ScheduledMessageEventsByKey = make(map[ScheduledMessageEventKey]*scheduling.ScheduledMessageEvent)
+
+			return nil
+		}
+	} else {
+		return fmt.Errorf("Not found")
+	}
+}
+
+func (m *MockScheduledMessageEventStore) CreateScheduledMessageEvent(ctx context.Context, event *scheduling.ScheduledMessageEvent) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return m.Error
+	}
+
+	if event.Id == uuid.Nil {
+		event.Id = uuid.New()
+	}
+
+	key := ScheduledMessageEventKey{RecipientId: event.RecipientId, MessageId: event.MessageId, ScheduledMessageId: event.ScheduledMessageId}
+
+	if _, exists := m.ScheduledMessageEventsByKey[key]; exists {
+		return fmt.Errorf("Already exists")
+	} else {
+		event.Created = time.Now()
+		m.ScheduledMessageEvents[event.Id] = event
+		m.ScheduledMessageEventsByKey[key] = event
+		return nil
+	}
+}
+
+func (m *MockScheduledMessageEventStore) Exists(ctx context.Context, event *scheduling.ScheduledMessageEvent) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return false, m.Error
+	}
+
+	key := ScheduledMessageEventKey{RecipientId: event.RecipientId, MessageId: event.MessageId, ScheduledMessageId: event.ScheduledMessageId}
+	_, exists := m.ScheduledMessageEventsByKey[key]
+
+	return exists, nil
+}

internal/store/mock/scheduled_message_store.go

@@ -0,0 +1,142 @@
+package mock
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/google/uuid"
+
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+)
+
+type ScheduledMessageKey struct {
+	UserId uuid.UUID
+}
+
+type MockScheduledMessageStore struct {
+	ScheduledMessages      map[uuid.UUID]*scheduling.ScheduledMessage
+	ScheduledMessagesByKey map[ScheduledMessageKey]*scheduling.ScheduledMessage
+	mu                     sync.RWMutex
+	Error                  error // Optional: simulate errors
+}
+
+func NewMockScheduledMessageStore() *MockScheduledMessageStore {
+	return &MockScheduledMessageStore{
+		ScheduledMessages:      make(map[uuid.UUID]*scheduling.ScheduledMessage),
+		ScheduledMessagesByKey: make(map[ScheduledMessageKey]*scheduling.ScheduledMessage),
+	}
+}
+
+func (m *MockScheduledMessageStore) Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessage, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	if id == uuid.Nil {
+		return nil, fmt.Errorf("Id is nil")
+	} else {
+		if schMsg := m.ScheduledMessages[id]; schMsg != nil {
+			return schMsg, nil
+		} else {
+			return nil, fmt.Errorf("Scheduled message does not exist")
+		}
+	}
+}
+
+func (m *MockScheduledMessageStore) FetchNextScheduledMessage(ctx context.Context) (*scheduling.ScheduledMessage, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	var schMsg *scheduling.ScheduledMessage
+	for _, msg := range m.ScheduledMessages {
+		if msg.Status == scheduling.Ready {
+			msg.Status = scheduling.Processing
+			schMsg = msg
+			break
+		}
+	}
+
+	if schMsg == nil {
+		return nil, fmt.Errorf("No scheduled message is ready")
+	} else {
+		return schMsg, nil
+	}
+}
+
+func (m *MockScheduledMessageStore) GetWithUserId(ctx context.Context, userId uuid.UUID) ([]*scheduling.ScheduledMessage, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	if userId == uuid.Nil {
+		return nil, fmt.Errorf("User Id is nil")
+	} else {
+		var schMsgs []*scheduling.ScheduledMessage
+		for _, schMsg := range m.ScheduledMessages {
+			if schMsg.UserId == userId {
+				schMsgs = append(schMsgs, schMsg)
+			}
+		}
+
+		if len(schMsgs) == 0 {
+			return nil, nil
+		} else {
+			return schMsgs, nil
+		}
+	}
+}
+
+func (m *MockScheduledMessageStore) CreateScheduledMessage(ctx context.Context, schedMsg *scheduling.ScheduledMessage) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return m.Error
+	}
+
+	if schedMsg.Id == uuid.Nil {
+		schedMsg.Id = uuid.New()
+	}
+
+	key := ScheduledMessageKey{UserId: schedMsg.UserId}
+
+	m.ScheduledMessages[schedMsg.Id] = schedMsg
+	m.ScheduledMessagesByKey[key] = schedMsg
+	return nil
+}
+
+func (m *MockScheduledMessageStore) UpdateStatus(ctx context.Context, id uuid.UUID, updatedStatus string) (*string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.Error != nil {
+		return nil, m.Error
+	}
+
+	if id == uuid.Nil {
+		return nil, fmt.Errorf("Id is nil")
+	} else {
+		if schMsg := m.ScheduledMessages[id]; schMsg != nil {
+			key := ScheduledMessageKey{UserId: schMsg.UserId}
+
+			schMsg.Status = updatedStatus
+			m.ScheduledMessages[schMsg.Id] = schMsg
+			m.ScheduledMessagesByKey[key] = schMsg
+
+			return &updatedStatus, nil
+		} else {
+			return nil, fmt.Errorf("Scheduled message does not exist")
+		}
+	}
+}

internal/store/scheduled_message_event_store.go

@@ -0,0 +1,114 @@
+package store
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
+)
+
+type ScheduledMessageEventStore interface {
+	Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessageEvent, error)
+	GetWithScheduleMessageId(ctx context.Context, schMsgId uuid.UUID) ([]*scheduling.ScheduledMessageEvent, error)
+	Delete(ctx context.Context, id uuid.UUID) error
+	CreateScheduledMessageEvent(ctx context.Context, event *scheduling.ScheduledMessageEvent) error
+	Exists(ctx context.Context, event *scheduling.ScheduledMessageEvent) (bool, error)
+}
+
+type PGScheduledMessageEventStore struct {
+	db *pgxpool.Pool
+}
+
+func NewScheduledMessageEventStore(db *pgxpool.Pool) *PGScheduledMessageEventStore {
+	return &PGScheduledMessageEventStore{db: db}
+}
+
+func (s *PGScheduledMessageEventStore) Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessageEvent, error) {
+	query := `SELECT id, recipient_id, message_id, scheduled_message_id, created FROM scheduled_message_events WHERE id = $1`
+
+	var event scheduling.ScheduledMessageEvent
+	err := s.db.QueryRow(ctx, query, id).Scan(
+		&event.Id, &event.RecipientId, &event.MessageId, &event.ScheduledMessageId, &event.Created,
+	)
+
+	if err == pgx.ErrNoRows {
+		return nil, nil
+	} else if err != nil {
+		return nil, fmt.Errorf("getting scheduled message event by ID: %w", err)
+	} else {
+		return &event, nil
+	}
+}
+
+func (s *PGScheduledMessageEventStore) GetWithScheduleMessageId(ctx context.Context, schMsgId uuid.UUID) ([]*scheduling.ScheduledMessageEvent, error) {
+	var events []*scheduling.ScheduledMessageEvent
+	query := `SELECT id, recipient_id, message_id, scheduled_message_id, created FROM scheduled_message_events WHERE scheduled_message_id = $1
+	`
+
+	rows, err := s.db.Query(ctx, query, schMsgId)
+
+	if err != nil {
+		return nil, fmt.Errorf("Error retrieving rows from scheduled_message_Events: %w", err)
+	}
+
+	for rows.Next() {
+		var event scheduling.ScheduledMessageEvent
+
+		if err := rows.Scan(&event.Id, &event.RecipientId, &event.MessageId, &event.ScheduledMessageId, &event.Created); err != nil {
+			return nil, fmt.Errorf("Error fetching data from row: %w", err)
+		} else {
+			events = append(events, &event)
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return events, nil
+}
+
+func (s *PGScheduledMessageEventStore) Delete(ctx context.Context, id uuid.UUID) error {
+	query := `
+		DELETE FROM scheduled_message_events WHERE id = $1
+	`
+
+	commandTag, err := s.db.Exec(ctx, query, id)
+	if err != nil {
+		return fmt.Errorf("error deleting event: %w", err)
+	}
+
+	if commandTag.RowsAffected() == 0 {
+		return fmt.Errorf("no event found with id %d", id)
+	} else {
+		return nil
+	}
+}
+
+func (s *PGScheduledMessageEventStore) CreateScheduledMessageEvent(ctx context.Context, event *scheduling.ScheduledMessageEvent) error {
+	query := `
+		INSERT INTO scheduled_message_events (recipient_id, message_id, scheduled_message_id)
+		VALUES ($1, $2, $3)
+		RETURNING id, created
+	`
+
+	return s.db.QueryRow(ctx, query, event.RecipientId, event.MessageId, event.ScheduledMessageId).Scan(
+		&event.Id, &event.Created,
+	)
+}
+
+func (s *PGScheduledMessageEventStore) Exists(ctx context.Context, event *scheduling.ScheduledMessageEvent) (bool, error) {
+	query := `SELECT EXISTS(SELECT 1 FROM scheduled_message_events WHERE scheduled_message_id = $1 AND recipient_id = $2)`
+
+	var exists bool
+	err := s.db.QueryRow(ctx, query, event.ScheduledMessageId, event.RecipientId).Scan(&exists)
+	if err != nil {
+		return false, fmt.Errorf("Scheduled message event does not exist: %w", err)
+	} else {
+		return exists, nil
+	}
+}

internal/store/scheduled_message_store.go

@@ -2,17 +2,21 @@ package store
 
 import (
 	"context"
-	// "fmt"
+	"fmt"
 
-	// "github.com/google/uuid"
+	"github.com/google/uuid"
-	// "github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"git.kundeng.us/phoenix/textsender-models/pkg/message/scheduling"
 )
 
 type ScheduledMessageStore interface {
+	Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessage, error)
+	FetchNextScheduledMessage(ctx context.Context) (*scheduling.ScheduledMessage, error)
+	GetWithUserId(ctx context.Context, userId uuid.UUID) ([]*scheduling.ScheduledMessage, error)
 	CreateScheduledMessage(ctx context.Context, schedMsg *scheduling.ScheduledMessage) error
+	UpdateStatus(ctx context.Context, id uuid.UUID, updatedStatus string) (*string, error)
 }
 
 type PGScheduledMessageStore struct {
@@ -23,6 +27,79 @@ func NewScheduledMessageStore(db *pgxpool.Pool) *PGScheduledMessageStore {
 	return &PGScheduledMessageStore{db: db}
 }
 
+func (s *PGScheduledMessageStore) Get(ctx context.Context, id uuid.UUID) (*scheduling.ScheduledMessage, error) {
+	query := `
+		SELECT id, scheduled, created, status, user_id FROM scheduled_messages WHERE id = $1
+	`
+
+	var schMsg scheduling.ScheduledMessage
+	err := s.db.QueryRow(ctx, query, id).Scan(
+		&schMsg.Id, &schMsg.Scheduled, &schMsg.Created, &schMsg.Status, &schMsg.UserId,
+	)
+
+	if err == pgx.ErrNoRows {
+		return nil, fmt.Errorf("No rows")
+	} else if err != nil {
+		return nil, fmt.Errorf("Getting scheduled message: %w", err)
+	}
+
+	return &schMsg, nil
+}
+
+func (s *PGScheduledMessageStore) FetchNextScheduledMessage(ctx context.Context) (*scheduling.ScheduledMessage, error) {
+	query := `
+		UPDATE scheduled_messages
+		SET status = $1
+		WHERE id = (
+			SELECT id FROM scheduled_messages
+			WHERE status = $2
+			ORDER BY id
+			FOR UPDATE SKIP LOCKED
+			LIMIT 1
+		)
+		RETURNING id, scheduled, created, status, user_id
+	`
+	var schMsg scheduling.ScheduledMessage
+	err := s.db.QueryRow(ctx, query, scheduling.Processing, scheduling.Ready).Scan(
+		&schMsg.Id, &schMsg.Scheduled, &schMsg.Created, &schMsg.Status, &schMsg.UserId,
+	)
+
+	if err == pgx.ErrNoRows {
+		return nil, nil
+	} else if err != nil {
+		return nil, fmt.Errorf("Getting scheduled message: %w", err)
+	} else {
+		return &schMsg, nil
+	}
+}
+
+func (s *PGScheduledMessageStore) GetWithUserId(ctx context.Context, userId uuid.UUID) ([]*scheduling.ScheduledMessage, error) {
+	query := `
+		SELECT id, scheduled, created, status, user_id FROM scheduled_messages WHERE user_id = $1
+	`
+	rows, err := s.db.Query(ctx, query, userId)
+	if err != nil {
+		return nil, fmt.Errorf("Error retrieving rows from scheduled_messages: %w", err)
+	}
+
+	var schMsgs []*scheduling.ScheduledMessage
+
+	for rows.Next() {
+		var schMsg scheduling.ScheduledMessage
+		if err := rows.Scan(&schMsg.Id, &schMsg.Scheduled, &schMsg.Created, &schMsg.Status, &schMsg.UserId); err != nil {
+			return nil, fmt.Errorf("Error fetching data from row: %w", err)
+		} else {
+			schMsgs = append(schMsgs, &schMsg)
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	} else {
+		return schMsgs, nil
+	}
+}
+
 func (s *PGScheduledMessageStore) CreateScheduledMessage(ctx context.Context, schedMsg *scheduling.ScheduledMessage) error {
 	query := `
 		INSERT INTO scheduled_messages (scheduled, status, user_id)
@@ -34,3 +111,18 @@ func (s *PGScheduledMessageStore) CreateScheduledMessage(ctx context.Context, sc
 		&schedMsg.Id, &schedMsg.Created,
 	)
 }
+
+func (s *PGScheduledMessageStore) UpdateStatus(ctx context.Context, id uuid.UUID, updatedStatus string) (*string, error) {
+	query := `
+		UPDATE scheduled_messages set status = $1 WHERE id = $2
+		RETURNING status
+	`
+
+	var returnedStatus string
+
+	if err := s.db.QueryRow(ctx, query, updatedStatus, id).Scan(&returnedStatus); err != nil {
+		return nil, err
+	} else {
+		return &returnedStatus, nil
+	}
+}

migrations/schema.sql

@@ -2,22 +2,30 @@ CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
 DROP TABLE IF EXISTS contacts CASCADE;
 
-CREATE TABLE contacts (
+CREATE TABLE IF NOT EXISTS contacts (
     id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
     phone_number TEXT NOT NULL,
     user_id UUID NOT NULL
 );
 
-CREATE TABLE messages (
+CREATE TABLE IF NOT EXISTS messages (
     id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
     content TEXT NOT NULL,
     user_id UUID NOT NULL
 );
 
-CREATE TABLE scheduled_messages (
+CREATE TABLE IF NOT EXISTS scheduled_messages (
     id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
     scheduled timestamptz NOT NULL,
     created timestamptz DEFAULT now(),
     status TEXT CHECK (status IN ('PENDING', 'READY', 'PROCESSING', 'DONE')),
     user_id UUID NOT NULL
 );
+
+CREATE TABLE IF NOT EXISTS scheduled_message_events (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    recipient_id UUID NOT NULL,
+    message_id UUID NOT NULL,
+    scheduled_message_id UUID NOT NULL,
+    created timestamptz DEFAULT now()
+)