Redmage/api/api.go

101 lines
2.5 KiB
Go
Raw Permalink Normal View History

2024-04-08 23:18:09 +07:00
package api
import (
2024-04-09 21:49:23 +07:00
"context"
"database/sql"
"sync"
2024-04-08 23:18:09 +07:00
2024-04-24 13:01:13 +07:00
"github.com/stephenafamo/bob"
2024-04-09 21:49:23 +07:00
"github.com/teivah/broadcast"
2024-04-10 22:38:19 +07:00
"github.com/tigorlazuardi/redmage/api/bmessage"
2024-06-10 14:52:53 +07:00
"github.com/tigorlazuardi/redmage/api/events"
2024-04-14 00:32:55 +07:00
"github.com/tigorlazuardi/redmage/api/reddit"
2024-05-14 12:08:34 +07:00
"github.com/tigorlazuardi/redmage/api/scheduler"
"github.com/tigorlazuardi/redmage/config"
2024-04-09 21:49:23 +07:00
"github.com/tigorlazuardi/redmage/pkg/log"
2024-05-14 12:08:34 +07:00
"go.opentelemetry.io/otel/attribute"
2024-04-25 12:31:20 +07:00
"github.com/ThreeDotsLabs/watermill/message"
2024-04-08 23:18:09 +07:00
)
type API struct {
db bob.Executor
sqldb *sql.DB
2024-04-09 21:49:23 +07:00
2024-05-14 12:08:34 +07:00
scheduler *scheduler.Scheduler
2024-04-09 21:49:23 +07:00
2024-04-11 16:04:13 +07:00
downloadBroadcast *broadcast.Relay[bmessage.ImageDownloadMessage]
2024-06-10 14:52:53 +07:00
eventBroadcast *broadcast.Relay[events.Event]
config *config.Config
2024-04-14 00:32:55 +07:00
imageSemaphore chan struct{}
2024-04-14 00:32:55 +07:00
reddit *reddit.Reddit
2024-04-25 12:31:20 +07:00
subscriber message.Subscriber
publisher message.Publisher
mu *sync.Mutex
2024-04-14 00:32:55 +07:00
}
type Dependencies struct {
DB *sql.DB
Config *config.Config
Reddit *reddit.Reddit
Publisher message.Publisher
Subscriber message.Subscriber
2024-04-09 21:49:23 +07:00
}
2024-04-25 13:05:41 +07:00
const downloadTopic = "subreddit_download"
2024-04-14 00:32:55 +07:00
func New(deps Dependencies) *API {
ch, err := deps.Subscriber.Subscribe(context.Background(), downloadTopic)
2024-04-25 13:05:41 +07:00
if err != nil {
panic(err)
}
2024-05-14 12:08:34 +07:00
2024-04-25 13:05:41 +07:00
api := &API{
db: bob.New(deps.DB),
sqldb: deps.DB,
downloadBroadcast: broadcast.NewRelay[bmessage.ImageDownloadMessage](),
2024-06-10 14:52:53 +07:00
eventBroadcast: broadcast.NewRelay[events.Event](),
config: deps.Config,
imageSemaphore: make(chan struct{}, deps.Config.Int("download.concurrency.images")),
reddit: deps.Reddit,
subscriber: deps.Subscriber,
publisher: deps.Publisher,
mu: &sync.Mutex{},
2024-04-09 21:49:23 +07:00
}
2024-04-25 13:05:41 +07:00
2024-05-14 12:08:34 +07:00
api.scheduler = scheduler.New(api.scheduleRun)
if err := api.scheduler.Sync(context.Background(), api.db); err != nil {
panic(err)
}
2024-05-14 18:49:45 +07:00
go api.scheduler.Start()
2024-05-14 12:08:34 +07:00
2024-04-25 20:22:05 +07:00
go api.StartSubredditDownloadPubsub(ch)
2024-04-25 13:05:41 +07:00
return api
2024-04-09 21:49:23 +07:00
}
2024-05-14 12:08:34 +07:00
func (api *API) scheduleRun(subreddit string) {
ctx, cancel := context.WithTimeout(context.Background(), api.config.Duration("scheduler.timeout"))
defer cancel()
2024-04-09 21:49:23 +07:00
2024-05-14 12:08:34 +07:00
ctx, span := tracer.Start(ctx, "*API.scheduleRun")
defer span.End()
span.SetAttributes(attribute.String("subreddit", subreddit))
2024-04-09 21:49:23 +07:00
2024-05-14 12:08:34 +07:00
log.New(ctx).Info("api: schedule run", "subreddit", subreddit)
2024-04-12 01:32:06 +07:00
2024-05-14 12:08:34 +07:00
err := api.PubsubStartDownloadSubreddit(ctx, PubsubStartDownloadSubredditParams{Subreddit: subreddit})
2024-04-12 01:32:06 +07:00
if err != nil {
2024-05-14 12:08:34 +07:00
log.New(ctx).Err(err).Error("api: failed to start download subreddit", "subreddit", subreddit)
2024-04-12 01:32:06 +07:00
}
}
2024-06-10 14:52:53 +07:00
func (api *API) GetEventBroadcaster() *broadcast.Relay[events.Event] {
return api.eventBroadcast
}