117 lines
4 KiB
Go
117 lines
4 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
"github.com/tigorlazuardi/redmage/models"
|
|
"github.com/tigorlazuardi/redmage/pkg/errs"
|
|
"github.com/tigorlazuardi/redmage/pkg/log"
|
|
"github.com/tigorlazuardi/redmage/pkg/telemetry"
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/trace"
|
|
)
|
|
|
|
func (api *API) StartSubredditDownloadPubsub(messages <-chan *message.Message) {
|
|
for msg := range messages {
|
|
var subreddit *models.Subreddit
|
|
if err := json.Unmarshal(msg.Payload, &subreddit); err != nil {
|
|
log.New(context.Background()).Err(err).Error("failed to unmarshal json for download pubsub", "topic", downloadTopic)
|
|
return
|
|
}
|
|
log.New(context.Background()).Debug("received pubsub message",
|
|
"message", string(msg.Payload),
|
|
)
|
|
func(msg *message.Message, subreddit *models.Subreddit) {
|
|
defer msg.Ack()
|
|
var err error
|
|
ctx, span := tracer.Start(context.Background(), "Download Subreddit Pubsub")
|
|
defer func() {
|
|
if err != nil {
|
|
if _, err := api.ScheduleSet(ctx, ScheduleSetParams{
|
|
Subreddit: subreddit.Name,
|
|
Status: ScheduleStatusError,
|
|
ErrorMessage: err.Error(),
|
|
}); err != nil {
|
|
log.New(ctx).Err(err).Error("failed to set schedule status", "subreddit", subreddit.Name, "status", ScheduleStatusError.String())
|
|
}
|
|
} else {
|
|
if _, err := api.ScheduleSet(ctx, ScheduleSetParams{
|
|
Subreddit: subreddit.Name,
|
|
Status: ScheduleStatusStandby,
|
|
}); err != nil {
|
|
log.New(ctx).Err(err).Error("failed to set schedule status", "subreddit", subreddit.Name, "status", ScheduleStatusStandby.String())
|
|
}
|
|
}
|
|
telemetry.EndWithStatus(span, err)
|
|
}()
|
|
span.AddEvent("pubsub." + downloadTopic)
|
|
_, err = api.ScheduleSet(ctx, ScheduleSetParams{
|
|
Subreddit: subreddit.Name,
|
|
Status: ScheduleStatusDownloading,
|
|
})
|
|
if err != nil {
|
|
log.New(ctx).Err(err).Error("failed to set schedule status", "subreddit", subreddit.Name, "status", ScheduleStatusDownloading.String())
|
|
}
|
|
|
|
devices, err := models.Devices.Query(ctx, api.db, models.SelectWhere.Devices.Enable.EQ(1)).All()
|
|
if err != nil {
|
|
msg.Nack()
|
|
log.New(ctx).Err(err).Error("failed to query devices")
|
|
return
|
|
}
|
|
|
|
err = api.DownloadSubredditImages(ctx, subreddit, devices)
|
|
if err != nil {
|
|
if errs.HasCode(err, http.StatusTooManyRequests) {
|
|
msg.Nack()
|
|
}
|
|
if errs.Source(err).Error() == "database is locked" {
|
|
msg.Nack()
|
|
}
|
|
log.New(ctx).Err(err).Error("failed to download subreddit images", "subreddit", subreddit)
|
|
return
|
|
}
|
|
}(msg, subreddit)
|
|
}
|
|
}
|
|
|
|
type PubsubStartDownloadSubredditParams struct {
|
|
Subreddit string `json:"subreddit"`
|
|
}
|
|
|
|
func (api *API) PubsubStartDownloadSubreddit(ctx context.Context, params PubsubStartDownloadSubredditParams) (err error) {
|
|
ctx, span := tracer.Start(ctx, "*API.PubsubStartDownloadSubreddit", trace.WithAttributes(attribute.String("subreddit", params.Subreddit)))
|
|
defer span.End()
|
|
|
|
subreddit, err := models.Subreddits.Query(ctx, api.db, models.SelectWhere.Subreddits.Name.EQ(params.Subreddit)).One()
|
|
if err != nil {
|
|
if err.Error() == "sql: no rows in result set" {
|
|
return errs.Wrapw(err, "subreddit not registered", "params", params).Code(http.StatusNotFound)
|
|
}
|
|
return errs.Wrapw(err, "failed to verify subreddit existence", "params", params)
|
|
}
|
|
|
|
_, errSchedule := api.scheduleSet(ctx, api.db, ScheduleSetParams{
|
|
Subreddit: subreddit.Name,
|
|
Status: ScheduleStatusEnqueued,
|
|
})
|
|
if errSchedule != nil {
|
|
log.New(ctx).Err(errSchedule).Error("failed to set schedule status", "subreddit", subreddit.Name, "status", ScheduleStatusEnqueued.String())
|
|
}
|
|
|
|
payload, err := json.Marshal(subreddit)
|
|
if err != nil {
|
|
return errs.Wrapw(err, "failed to marshal subreddit")
|
|
}
|
|
|
|
err = api.publisher.Publish(downloadTopic, message.NewMessage(watermill.NewUUID(), payload))
|
|
if err != nil {
|
|
return errs.Wrapw(err, "failed to enqueue reddit download", "params", params)
|
|
}
|
|
return nil
|
|
}
|