86 lines
2.6 KiB
Go
86 lines
2.6 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 {
|
|
log.New(context.Background()).Debug("received pubsub message",
|
|
"message", msg,
|
|
"len", len(api.subredditSemaphore),
|
|
"cap", cap(api.subredditSemaphore),
|
|
"download.concurrency.subreddits", api.config.Int("download.concurrency.subreddits"),
|
|
)
|
|
api.subredditSemaphore <- struct{}{}
|
|
go func(msg *message.Message) {
|
|
defer func() {
|
|
msg.Ack()
|
|
<-api.subredditSemaphore
|
|
}()
|
|
var (
|
|
err error
|
|
subreddit *models.Subreddit
|
|
)
|
|
ctx, span := tracer.Start(context.Background(), "Download Subreddit Pubsub")
|
|
defer func() { telemetry.EndWithStatus(span, err) }()
|
|
span.AddEvent("pubsub." + downloadTopic)
|
|
|
|
err = json.Unmarshal(msg.Payload, &subreddit)
|
|
if err != nil {
|
|
log.New(ctx).Err(err).Error("failed to unmarshal json for download pubsub", "topic", downloadTopic)
|
|
return
|
|
}
|
|
|
|
devices, err := models.Devices.Query(ctx, api.db).All()
|
|
if err != nil {
|
|
log.New(ctx).Err(err).Error("failed to query devices")
|
|
return
|
|
}
|
|
|
|
err = api.DownloadSubredditImages(ctx, subreddit, devices)
|
|
if err != nil {
|
|
log.New(ctx).Err(err).Error("failed to download subreddit images", "subreddit", subreddit)
|
|
return
|
|
}
|
|
}(msg)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
payload, _ := json.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
|
|
}
|