130 lines
3.5 KiB
Go
130 lines
3.5 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"github.com/robfig/cron/v3"
|
|
"github.com/stephenafamo/bob"
|
|
"github.com/teivah/broadcast"
|
|
"github.com/tigorlazuardi/redmage/api/bmessage"
|
|
"github.com/tigorlazuardi/redmage/api/reddit"
|
|
"github.com/tigorlazuardi/redmage/config"
|
|
"github.com/tigorlazuardi/redmage/models"
|
|
"github.com/tigorlazuardi/redmage/pkg/errs"
|
|
"github.com/tigorlazuardi/redmage/pkg/log"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
watermillSql "github.com/ThreeDotsLabs/watermill-sql/v3/pkg/sql"
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
watermillSqlite "github.com/walterwanderley/watermill-sqlite"
|
|
)
|
|
|
|
type API struct {
|
|
db bob.Executor
|
|
|
|
scheduler *cron.Cron
|
|
scheduleMap map[cron.EntryID]*models.Subreddit
|
|
|
|
downloadBroadcast *broadcast.Relay[bmessage.ImageDownloadMessage]
|
|
|
|
config *config.Config
|
|
|
|
imageSemaphore chan struct{}
|
|
subredditSemaphore chan struct{}
|
|
|
|
reddit *reddit.Reddit
|
|
|
|
subscriber message.Subscriber
|
|
publisher message.Publisher
|
|
}
|
|
|
|
type Dependencies struct {
|
|
DB *sql.DB
|
|
PubsubDB *sql.DB
|
|
Config *config.Config
|
|
Reddit *reddit.Reddit
|
|
}
|
|
|
|
const downloadTopic = "subreddit_download"
|
|
|
|
var watermillLogger = &log.WatermillLogger{}
|
|
|
|
func New(deps Dependencies) *API {
|
|
ackDeadline := deps.Config.Duration("pubsub.ack.deadline")
|
|
subscriber, err := watermillSql.NewSubscriber(deps.PubsubDB, watermillSql.SubscriberConfig{
|
|
ConsumerGroup: "redmage",
|
|
AckDeadline: &ackDeadline,
|
|
SchemaAdapter: watermillSqlite.DefaultSQLiteSchema{},
|
|
OffsetsAdapter: watermillSqlite.DefaultSQLiteOffsetsAdapter{},
|
|
InitializeSchema: true,
|
|
}, watermillLogger)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
publisher, err := watermillSql.NewPublisher(deps.PubsubDB, watermillSql.PublisherConfig{
|
|
SchemaAdapter: watermillSqlite.DefaultSQLiteSchema{},
|
|
AutoInitializeSchema: true,
|
|
}, watermillLogger)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
ch, err := subscriber.Subscribe(context.Background(), downloadTopic)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
api := &API{
|
|
db: bob.New(deps.DB),
|
|
scheduler: cron.New(),
|
|
scheduleMap: make(map[cron.EntryID]*models.Subreddit, 8),
|
|
downloadBroadcast: broadcast.NewRelay[bmessage.ImageDownloadMessage](),
|
|
config: deps.Config,
|
|
imageSemaphore: make(chan struct{}, deps.Config.Int("download.concurrency.images")),
|
|
subredditSemaphore: make(chan struct{}, deps.Config.Int("download.concurrency.subreddits")),
|
|
reddit: deps.Reddit,
|
|
subscriber: subscriber,
|
|
publisher: publisher,
|
|
}
|
|
|
|
if err := api.StartScheduler(context.Background()); err != nil {
|
|
panic(err)
|
|
}
|
|
go api.StartSubredditDownloadPubsub(ch)
|
|
return api
|
|
}
|
|
|
|
func (api *API) StartScheduler(ctx context.Context) error {
|
|
subreddits, err := models.Subreddits.Query(ctx, api.db, models.SelectWhere.Subreddits.EnableSchedule.EQ(1)).All()
|
|
if err != nil {
|
|
return errs.Wrapw(err, "failed to get all subreddits")
|
|
}
|
|
|
|
for _, subreddit := range subreddits {
|
|
err := api.scheduleSubreddit(subreddit)
|
|
if err != nil {
|
|
log.New(ctx).Err(err).Error(
|
|
fmt.Sprintf("failed to start scheduler for subreddit '%s'", subreddit.Name),
|
|
"subreddit", subreddit,
|
|
)
|
|
continue
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (api *API) scheduleSubreddit(subreddit *models.Subreddit) error {
|
|
id, err := api.scheduler.AddFunc(subreddit.Schedule, func() {
|
|
payload, _ := json.Marshal(subreddit)
|
|
_ = api.publisher.Publish(downloadTopic, message.NewMessage(watermill.NewUUID(), payload))
|
|
})
|
|
if err != nil {
|
|
return errs.Wrap(err)
|
|
}
|
|
api.scheduleMap[id] = subreddit
|
|
|
|
return nil
|
|
}
|