2021-05-24 07:12:46 +00:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
|
|
|
"database/sql"
|
|
|
|
"errors"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/araddon/dateparse"
|
|
|
|
)
|
|
|
|
|
2021-07-13 15:23:10 +00:00
|
|
|
type queueItem struct {
|
|
|
|
id int
|
|
|
|
name string
|
|
|
|
content []byte
|
|
|
|
schedule *time.Time
|
|
|
|
}
|
|
|
|
|
2021-06-06 12:39:42 +00:00
|
|
|
func (db *database) enqueue(name string, content []byte, schedule time.Time) error {
|
2021-05-24 07:12:46 +00:00
|
|
|
if len(content) == 0 {
|
|
|
|
return errors.New("empty content")
|
|
|
|
}
|
2021-11-28 08:02:09 +00:00
|
|
|
_, err := db.exec(
|
|
|
|
"insert into queue (name, content, schedule) values (@name, @content, @schedule)",
|
|
|
|
sql.Named("name", name),
|
|
|
|
sql.Named("content", content),
|
|
|
|
sql.Named("schedule", schedule.UTC().Format(time.RFC3339Nano)),
|
|
|
|
)
|
2021-05-24 07:12:46 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-06-06 12:39:42 +00:00
|
|
|
func (db *database) reschedule(qi *queueItem, dur time.Duration) error {
|
2021-11-28 08:02:09 +00:00
|
|
|
_, err := db.exec(
|
|
|
|
"update queue set schedule = @schedule, content = @content where id = @id",
|
|
|
|
sql.Named("schedule", qi.schedule.Add(dur).UTC().Format(time.RFC3339Nano)),
|
|
|
|
sql.Named("content", qi.content),
|
|
|
|
sql.Named("id", qi.id),
|
|
|
|
)
|
2021-05-24 07:12:46 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-06-06 12:39:42 +00:00
|
|
|
func (db *database) dequeue(qi *queueItem) error {
|
|
|
|
_, err := db.exec("delete from queue where id = @id", sql.Named("id", qi.id))
|
2021-05-24 07:12:46 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-06-06 12:39:42 +00:00
|
|
|
func (db *database) peekQueue(name string) (*queueItem, error) {
|
2021-11-28 08:02:09 +00:00
|
|
|
row, err := db.queryRow(
|
|
|
|
"select id, name, content, schedule from queue where schedule <= @schedule and name = @name order by schedule asc limit 1",
|
|
|
|
sql.Named("name", name),
|
|
|
|
sql.Named("schedule", time.Now().UTC().Format(time.RFC3339Nano)),
|
|
|
|
)
|
2021-05-24 07:12:46 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
qi := &queueItem{}
|
|
|
|
var timeString string
|
|
|
|
if err = row.Scan(&qi.id, &qi.name, &qi.content, &timeString); err != nil {
|
2021-07-13 15:23:10 +00:00
|
|
|
if errors.Is(err, sql.ErrNoRows) {
|
2021-05-24 07:12:46 +00:00
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
return nil, err
|
|
|
|
}
|
2021-07-13 15:23:10 +00:00
|
|
|
t, err := dateparse.ParseIn(timeString, time.UTC)
|
2021-05-24 07:12:46 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
qi.schedule = &t
|
|
|
|
return qi, nil
|
|
|
|
}
|