360 lines
10 KiB
Go
360 lines
10 KiB
Go
|
package repository
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"strings"
|
||
|
|
|
||
|
|
app "haixun-backend/internal/library/errors"
|
||
|
|
"haixun-backend/internal/library/errors/code"
|
||
|
|
"haixun-backend/internal/model/publish_queue/domain/entity"
|
||
|
|
domrepo "haixun-backend/internal/model/publish_queue/domain/repository"
|
||
|
|
|
||
|
|
"go.mongodb.org/mongo-driver/bson"
|
||
|
|
"go.mongodb.org/mongo-driver/mongo"
|
||
|
|
"go.mongodb.org/mongo-driver/mongo/options"
|
||
|
|
)
|
||
|
|
|
||
|
|
type mongoRepository struct {
|
||
|
|
collection *mongo.Collection
|
||
|
|
}
|
||
|
|
|
||
|
|
func NewMongoRepository(db *mongo.Database) domrepo.Repository {
|
||
|
|
if db == nil {
|
||
|
|
return &mongoRepository{}
|
||
|
|
}
|
||
|
|
return &mongoRepository{collection: db.Collection(entity.CollectionName)}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) EnsureIndexes(ctx context.Context) error {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
_, err := r.collection.Indexes().CreateMany(ctx, []mongo.IndexModel{
|
||
|
|
{
|
||
|
|
Keys: bson.D{
|
||
|
|
{Key: "tenant_id", Value: 1},
|
||
|
|
{Key: "owner_uid", Value: 1},
|
||
|
|
{Key: "account_id", Value: 1},
|
||
|
|
{Key: "status", Value: 1},
|
||
|
|
{Key: "scheduled_at", Value: 1},
|
||
|
|
},
|
||
|
|
},
|
||
|
|
})
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
func actorFilter(tenantID, ownerUID, accountID string) bson.M {
|
||
|
|
filter := bson.M{
|
||
|
|
"tenant_id": strings.TrimSpace(tenantID),
|
||
|
|
"owner_uid": strings.TrimSpace(ownerUID),
|
||
|
|
}
|
||
|
|
if accountID = strings.TrimSpace(accountID); accountID != "" {
|
||
|
|
filter["account_id"] = accountID
|
||
|
|
}
|
||
|
|
return filter
|
||
|
|
}
|
||
|
|
|
||
|
|
func normalizePage(page, pageSize int) (int, int) {
|
||
|
|
if page <= 0 {
|
||
|
|
page = 1
|
||
|
|
}
|
||
|
|
if pageSize <= 0 {
|
||
|
|
pageSize = 10
|
||
|
|
}
|
||
|
|
if pageSize > 50 {
|
||
|
|
pageSize = 50
|
||
|
|
}
|
||
|
|
return page, pageSize
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) Create(ctx context.Context, item *entity.QueueItem) error {
|
||
|
|
if r.collection == nil {
|
||
|
|
return app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
if item == nil {
|
||
|
|
return app.For(code.ThreadsAccount).InputMissingRequired("queue item is required")
|
||
|
|
}
|
||
|
|
_, err := r.collection.InsertOne(ctx, item)
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) Get(ctx context.Context, tenantID, ownerUID, accountID, queueID string) (*entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
queueID = strings.TrimSpace(queueID)
|
||
|
|
if queueID == "" {
|
||
|
|
return nil, app.For(code.ThreadsAccount).InputMissingRequired("queue id is required")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["_id"] = queueID
|
||
|
|
var out entity.QueueItem
|
||
|
|
err := r.collection.FindOne(ctx, filter).Decode(&out)
|
||
|
|
if err != nil {
|
||
|
|
if err == mongo.ErrNoDocuments {
|
||
|
|
return nil, app.For(code.ThreadsAccount).ResNotFound("publish queue item not found")
|
||
|
|
}
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return &out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) ListByIDs(ctx context.Context, tenantID, ownerUID string, ids []string) ([]entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
trimmed := make([]string, 0, len(ids))
|
||
|
|
for _, id := range ids {
|
||
|
|
id = strings.TrimSpace(id)
|
||
|
|
if id != "" {
|
||
|
|
trimmed = append(trimmed, id)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if len(trimmed) == 0 {
|
||
|
|
return nil, nil
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, "")
|
||
|
|
filter["_id"] = bson.M{"$in": trimmed}
|
||
|
|
cur, err := r.collection.Find(ctx, filter)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer cur.Close(ctx)
|
||
|
|
var out []entity.QueueItem
|
||
|
|
if err := cur.All(ctx, &out); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) List(ctx context.Context, filter domrepo.ListFilter, page, pageSize int) (*domrepo.ListResult, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
page, pageSize = normalizePage(page, pageSize)
|
||
|
|
mongoFilter := actorFilter(filter.TenantID, filter.OwnerUID, filter.AccountID)
|
||
|
|
if status := strings.TrimSpace(filter.Status); status != "" {
|
||
|
|
mongoFilter["status"] = status
|
||
|
|
}
|
||
|
|
if filter.StartAt > 0 || filter.EndAt > 0 {
|
||
|
|
rangeFilter := bson.M{}
|
||
|
|
if filter.StartAt > 0 {
|
||
|
|
rangeFilter["$gte"] = filter.StartAt
|
||
|
|
}
|
||
|
|
if filter.EndAt > 0 {
|
||
|
|
rangeFilter["$lte"] = filter.EndAt
|
||
|
|
}
|
||
|
|
mongoFilter["scheduled_at"] = rangeFilter
|
||
|
|
}
|
||
|
|
total, err := r.collection.CountDocuments(ctx, mongoFilter)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
opts := options.Find().
|
||
|
|
SetSort(bson.D{{Key: "scheduled_at", Value: -1}}).
|
||
|
|
SetSkip(int64((page - 1) * pageSize)).
|
||
|
|
SetLimit(int64(pageSize))
|
||
|
|
cur, err := r.collection.Find(ctx, mongoFilter, opts)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer cur.Close(ctx)
|
||
|
|
var items []entity.QueueItem
|
||
|
|
if err := cur.All(ctx, &items); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return &domrepo.ListResult{Items: items, Total: total}, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) ListDue(ctx context.Context, now int64, limit int) ([]entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
if limit <= 0 {
|
||
|
|
limit = 20
|
||
|
|
}
|
||
|
|
filter := bson.M{
|
||
|
|
"status": entity.StatusScheduled,
|
||
|
|
"scheduled_at": bson.M{"$lte": now},
|
||
|
|
}
|
||
|
|
opts := options.Find().
|
||
|
|
SetSort(bson.D{{Key: "scheduled_at", Value: 1}}).
|
||
|
|
SetLimit(int64(limit))
|
||
|
|
cur, err := r.collection.Find(ctx, filter, opts)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer cur.Close(ctx)
|
||
|
|
var items []entity.QueueItem
|
||
|
|
if err := cur.All(ctx, &items); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return items, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) ListMissedCandidates(ctx context.Context, before int64, limit int) ([]entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
if limit <= 0 {
|
||
|
|
limit = 20
|
||
|
|
}
|
||
|
|
filter := bson.M{
|
||
|
|
"status": entity.StatusScheduled,
|
||
|
|
"scheduled_at": bson.M{"$lte": before},
|
||
|
|
}
|
||
|
|
opts := options.Find().
|
||
|
|
SetSort(bson.D{{Key: "scheduled_at", Value: 1}}).
|
||
|
|
SetLimit(int64(limit))
|
||
|
|
cur, err := r.collection.Find(ctx, filter, opts)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer cur.Close(ctx)
|
||
|
|
var items []entity.QueueItem
|
||
|
|
if err := cur.All(ctx, &items); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return items, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) UpdateIfStatus(ctx context.Context, item *entity.QueueItem, allowed []string) (*entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
if item == nil || item.ID == "" {
|
||
|
|
return nil, app.For(code.ThreadsAccount).InputMissingRequired("queue item is required")
|
||
|
|
}
|
||
|
|
filter := bson.M{
|
||
|
|
"_id": item.ID,
|
||
|
|
"status": bson.M{"$in": allowed},
|
||
|
|
}
|
||
|
|
res, err := r.collection.ReplaceOne(ctx, filter, item)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if res.MatchedCount == 0 {
|
||
|
|
return nil, app.For(code.ThreadsAccount).ResInvalidState("queue item status changed; update rejected")
|
||
|
|
}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) Update(ctx context.Context, item *entity.QueueItem) (*entity.QueueItem, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
if item == nil || item.ID == "" {
|
||
|
|
return nil, app.For(code.ThreadsAccount).InputMissingRequired("queue item is required")
|
||
|
|
}
|
||
|
|
filter := bson.M{"_id": item.ID}
|
||
|
|
res, err := r.collection.ReplaceOne(ctx, filter, item)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if res.MatchedCount == 0 {
|
||
|
|
return nil, app.For(code.ThreadsAccount).ResNotFound("publish queue item not found")
|
||
|
|
}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) Delete(ctx context.Context, tenantID, ownerUID, accountID, queueID string) error {
|
||
|
|
if r.collection == nil {
|
||
|
|
return app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
queueID = strings.TrimSpace(queueID)
|
||
|
|
if queueID == "" {
|
||
|
|
return app.For(code.ThreadsAccount).InputMissingRequired("queue id is required")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["_id"] = queueID
|
||
|
|
res, err := r.collection.DeleteOne(ctx, filter)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if res.DeletedCount == 0 {
|
||
|
|
return app.For(code.ThreadsAccount).ResNotFound("publish queue item not found")
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) CountByStatus(ctx context.Context, tenantID, ownerUID, accountID, status string) (int64, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return 0, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["status"] = strings.TrimSpace(status)
|
||
|
|
return r.collection.CountDocuments(ctx, filter)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) CountPublishedSince(ctx context.Context, tenantID, ownerUID, accountID string, since int64) (int64, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return 0, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["status"] = entity.StatusPublished
|
||
|
|
filter["published_at"] = bson.M{"$gte": since}
|
||
|
|
return r.collection.CountDocuments(ctx, filter)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) LastPublishedAt(ctx context.Context, tenantID, ownerUID, accountID string) (int64, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return 0, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["status"] = entity.StatusPublished
|
||
|
|
opts := options.FindOne().SetSort(bson.D{{Key: "published_at", Value: -1}})
|
||
|
|
var out entity.QueueItem
|
||
|
|
err := r.collection.FindOne(ctx, filter, opts).Decode(&out)
|
||
|
|
if err != nil {
|
||
|
|
if err == mongo.ErrNoDocuments {
|
||
|
|
return 0, nil
|
||
|
|
}
|
||
|
|
return 0, err
|
||
|
|
}
|
||
|
|
return out.PublishedAt, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) ListMediaIDsByAccount(ctx context.Context, tenantID, ownerUID, accountID string) ([]string, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
filter := actorFilter(tenantID, ownerUID, accountID)
|
||
|
|
filter["media_id"] = bson.M{"$ne": ""}
|
||
|
|
cur, err := r.collection.Find(ctx, filter, options.Find().SetProjection(bson.M{"media_id": 1}))
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer cur.Close(ctx)
|
||
|
|
seen := map[string]struct{}{}
|
||
|
|
var ids []string
|
||
|
|
for cur.Next(ctx) {
|
||
|
|
var row struct {
|
||
|
|
MediaID string `bson:"media_id"`
|
||
|
|
}
|
||
|
|
if err := cur.Decode(&row); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if id := strings.TrimSpace(row.MediaID); id != "" {
|
||
|
|
if _, ok := seen[id]; !ok {
|
||
|
|
seen[id] = struct{}{}
|
||
|
|
ids = append(ids, id)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if err := cur.Err(); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return ids, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *mongoRepository) DeleteByAccount(ctx context.Context, tenantID, ownerUID, accountID string) (int64, error) {
|
||
|
|
if r.collection == nil {
|
||
|
|
return 0, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
|
||
|
|
}
|
||
|
|
res, err := r.collection.DeleteMany(ctx, actorFilter(tenantID, ownerUID, accountID))
|
||
|
|
if err != nil {
|
||
|
|
return 0, err
|
||
|
|
}
|
||
|
|
return res.DeletedCount, nil
|
||
|
|
}
|