package store import ( "context" "fmt" "git.kundeng.us/phoenix/textsender-models/tx0/message/scheduling" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) 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 { db *pgxpool.Pool } 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) VALUES ($1, $2, $3) RETURNING id, created ` return s.db.QueryRow(ctx, query, schedMsg.Scheduled, schedMsg.Status, schedMsg.UserId).Scan( &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 } }