package repository import ( "context" "strings" app "haixun-backend/internal/library/errors" "haixun-backend/internal/library/errors/code" "haixun-backend/internal/model/publish_queue_event/domain/entity" domrepo "haixun-backend/internal/model/publish_queue_event/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: "queue_id", Value: 1}, {Key: "create_at", Value: -1}}}, }) return err } func (r *mongoRepository) Create(ctx context.Context, event *entity.Event) error { if r.collection == nil { return app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured") } if event == nil { return app.For(code.ThreadsAccount).InputMissingRequired("event is required") } _, err := r.collection.InsertOne(ctx, event) return err } func (r *mongoRepository) ListByQueue(ctx context.Context, tenantID, ownerUID, accountID, queueID string, limit int) ([]entity.Event, error) { filter := actorFilter(tenantID, ownerUID, accountID) filter["queue_id"] = strings.TrimSpace(queueID) return r.list(ctx, filter, limit) } func (r *mongoRepository) ListRecentByAccount(ctx context.Context, tenantID, ownerUID, accountID string, limit int) ([]entity.Event, error) { return r.list(ctx, actorFilter(tenantID, ownerUID, accountID), limit) } func (r *mongoRepository) DeleteByAccount(ctx context.Context, tenantID, ownerUID, accountID string) error { if r.collection == nil { return app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured") } _, err := r.collection.DeleteMany(ctx, actorFilter(tenantID, ownerUID, accountID)) return err } func (r *mongoRepository) list(ctx context.Context, filter bson.M, limit int) ([]entity.Event, error) { if r.collection == nil { return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured") } if limit <= 0 { limit = 50 } if limit > 200 { limit = 200 } opts := options.Find().SetSort(bson.D{{Key: "create_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 out []entity.Event if err := cur.All(ctx, &out); err != nil { return nil, err } return out, nil } 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 }