This commit is contained in:
Rogee
2025-01-15 15:55:49 +08:00
parent 9002862415
commit 13566cfa38
13 changed files with 1092 additions and 872 deletions

View File

@@ -0,0 +1,23 @@
package publishers
import (
"encoding/json"
"backend/app/events"
"git.ipao.vip/rogeecn/atom/contracts"
)
var _ contracts.EventPublisher = (*UserRegister)(nil)
type PostCreated struct {
ID int64 `json:"id"`
}
func (e *PostCreated) Marshal() ([]byte, error) {
return json.Marshal(e)
}
func (e *PostCreated) Topic() string {
return events.TopicPostCreated.String()
}

View File

@@ -0,0 +1,99 @@
package subscribers
import (
"context"
"encoding/json"
"backend/app/events"
"backend/app/events/publishers"
"backend/app/http/posts"
"backend/app/jobs"
"backend/database/fields"
"backend/providers/job"
"git.ipao.vip/rogeecn/atom/contracts"
"github.com/ThreeDotsLabs/watermill/message"
"github.com/pkg/errors"
"github.com/samber/lo"
"github.com/sirupsen/logrus"
)
var _ contracts.EventHandler = (*PostCreated)(nil)
// @provider(event)
type PostCreated struct {
log *logrus.Entry `inject:"false"`
postSvc *posts.Service
job *job.Job
}
func (e *PostCreated) Prepare() error {
e.log = logrus.WithField("module", "events.subscribers.user_register")
return nil
}
// PublishToTopic implements contracts.EventHandler.
func (e *PostCreated) PublishToTopic() string {
return events.TopicProcessed.String()
}
// Topic implements contracts.EventHandler.
func (e *PostCreated) Topic() string {
return events.TopicPostCreated.String()
}
// Handler implements contracts.EventHandler.
func (e *PostCreated) Handler(msg *message.Message) ([]*message.Message, error) {
var payload publishers.PostCreated
err := json.Unmarshal(msg.Payload, &payload)
if err != nil {
return nil, err
}
e.log.Infof("received event %s", msg.Payload)
post, err := e.postSvc.GetPostByID(context.Background(), payload.ID)
if err != nil {
return nil, errors.Wrapf(err, "failed to get user by id: %d", payload.ID)
}
video, ok := lo.Find(post.Assets.Data, func(asset fields.MediaAsset) bool {
return asset.Type == fields.MediaAssetTypeVideo
})
if !ok {
e.log.Warnf("post %d %s has no video asset", post.ID, post.Title)
return nil, nil
}
_ = video
job, err := e.job.Client()
if err != nil {
return nil, err
}
// cut video
_, err = job.Insert(context.Background(), jobs.VideoCut{
PostID: post.ID,
Hash: video.Hash,
TenantID: post.TenantID,
UserID: post.UserID,
}, nil)
if err != nil {
return nil, err
}
// extract audio
_, err = job.Insert(context.Background(), jobs.VideoExtractAudio{
PostID: post.ID,
Hash: video.Hash,
TenantID: post.TenantID,
UserID: post.UserID,
}, nil)
if err != nil {
return nil, err
}
return nil, nil
}

View File

@@ -18,6 +18,8 @@ const (
TopicProcessed Topic = "event:processed"
// TopicUserRegister is a Topic of type UserRegister.
TopicUserRegister Topic = "user:register"
// TopicPostCreated is a Topic of type PostCreated.
TopicPostCreated Topic = "post:created"
)
var ErrInvalidTopic = fmt.Errorf("not a valid Topic, try [%s]", strings.Join(_TopicNames, ", "))
@@ -25,6 +27,7 @@ var ErrInvalidTopic = fmt.Errorf("not a valid Topic, try [%s]", strings.Join(_To
var _TopicNames = []string{
string(TopicProcessed),
string(TopicUserRegister),
string(TopicPostCreated),
}
// TopicNames returns a list of possible string values of Topic.
@@ -39,6 +42,7 @@ func TopicValues() []Topic {
return []Topic{
TopicProcessed,
TopicUserRegister,
TopicPostCreated,
}
}
@@ -57,6 +61,7 @@ func (x Topic) IsValid() bool {
var _TopicValue = map[string]Topic{
"event:processed": TopicProcessed,
"user:register": TopicUserRegister,
"post:created": TopicPostCreated,
}
// ParseTopic attempts to convert a string to a Topic.

View File

@@ -5,6 +5,7 @@ package events
//
// Processed = "event:processed"
// UserRegister = "user:register"
// PostCreated = "post:created"
//
// )
type Topic string