533 lines
15 KiB
Go
533 lines
15 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
|
|
app "haixun-backend/internal/library/errors"
|
|
"haixun-backend/internal/library/errors/code"
|
|
"haixun-backend/internal/model/copy_draft/domain/entity"
|
|
domrepo "haixun-backend/internal/model/copy_draft/domain/repository"
|
|
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
"go.mongodb.org/mongo-driver/mongo"
|
|
"go.mongodb.org/mongo-driver/mongo/options"
|
|
"go.mongodb.org/mongo-driver/mongo/writeconcern"
|
|
)
|
|
|
|
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: "persona_id", Value: 1},
|
|
{Key: "create_at", Value: -1},
|
|
},
|
|
},
|
|
{
|
|
Keys: bson.D{
|
|
{Key: "tenant_id", Value: 1},
|
|
{Key: "owner_uid", Value: 1},
|
|
{Key: "persona_id", Value: 1},
|
|
{Key: "content_plan_id", Value: 1},
|
|
{Key: "create_at", Value: -1},
|
|
},
|
|
},
|
|
{
|
|
Keys: bson.D{
|
|
{Key: "tenant_id", Value: 1},
|
|
{Key: "owner_uid", Value: 1},
|
|
{Key: "persona_id", Value: 1},
|
|
{Key: "copy_mission_id", Value: 1},
|
|
{Key: "sort_order", Value: 1},
|
|
},
|
|
},
|
|
})
|
|
return err
|
|
}
|
|
|
|
func personaFilter(tenantID, ownerUID, personaID string) bson.M {
|
|
return bson.M{
|
|
"tenant_id": strings.TrimSpace(tenantID),
|
|
"owner_uid": strings.TrimSpace(ownerUID),
|
|
"persona_id": strings.TrimSpace(personaID),
|
|
}
|
|
}
|
|
|
|
func (r *mongoRepository) CreateMany(ctx context.Context, drafts []*entity.CopyDraft) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
if len(drafts) == 0 {
|
|
return nil
|
|
}
|
|
docs := make([]any, 0, len(drafts))
|
|
for _, draft := range drafts {
|
|
if draft == nil {
|
|
continue
|
|
}
|
|
docs = append(docs, draft)
|
|
}
|
|
if len(docs) == 0 {
|
|
return nil
|
|
}
|
|
_, err := r.collection.InsertMany(ctx, docs)
|
|
return err
|
|
}
|
|
|
|
func (r *mongoRepository) Create(ctx context.Context, draft *entity.CopyDraft) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
if draft == nil {
|
|
return app.For(code.Persona).InputMissingRequired("draft is required")
|
|
}
|
|
_, err := r.collection.InsertOne(ctx, draft)
|
|
return err
|
|
}
|
|
|
|
func inboxStatusFilter(status string) bson.M {
|
|
switch strings.TrimSpace(status) {
|
|
case "draft":
|
|
return bson.M{
|
|
"$and": []bson.M{
|
|
{"$or": []bson.M{{"publish_queue_id": ""}, {"publish_queue_id": bson.M{"$exists": false}}}},
|
|
{"$or": []bson.M{{"published_at": 0}, {"published_at": bson.M{"$exists": false}}}},
|
|
},
|
|
}
|
|
case "scheduled":
|
|
return bson.M{
|
|
"publish_queue_id": bson.M{"$ne": ""},
|
|
"$or": []bson.M{{"published_at": 0}, {"published_at": bson.M{"$exists": false}}},
|
|
}
|
|
case "published":
|
|
return bson.M{"published_at": bson.M{"$gt": 0}}
|
|
default:
|
|
return bson.M{}
|
|
}
|
|
}
|
|
|
|
func normalizeInboxPage(page, pageSize int) (int, int) {
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
if pageSize <= 0 {
|
|
pageSize = 12
|
|
}
|
|
if pageSize > 100 {
|
|
pageSize = 100
|
|
}
|
|
return page, pageSize
|
|
}
|
|
|
|
func (r *mongoRepository) ListInbox(ctx context.Context, filter domrepo.InboxFilter) (*domrepo.InboxListResult, error) {
|
|
if r.collection == nil {
|
|
return nil, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
page, pageSize := normalizeInboxPage(filter.Page, filter.PageSize)
|
|
mongoFilter := personaFilter(filter.TenantID, filter.OwnerUID, filter.PersonaID)
|
|
if extra := inboxStatusFilter(filter.Status); len(extra) > 0 {
|
|
for k, v := range extra {
|
|
mongoFilter[k] = v
|
|
}
|
|
}
|
|
if filter.RangeStart > 0 || filter.RangeEnd > 0 {
|
|
rangeClauses := make([]bson.M, 0, 3)
|
|
createRange := bson.M{}
|
|
if filter.RangeStart > 0 {
|
|
createRange["$gte"] = filter.RangeStart
|
|
}
|
|
if filter.RangeEnd > 0 {
|
|
createRange["$lte"] = filter.RangeEnd
|
|
}
|
|
if len(createRange) > 0 {
|
|
rangeClauses = append(rangeClauses, bson.M{"create_at": createRange})
|
|
}
|
|
publishedRange := bson.M{}
|
|
if filter.RangeStart > 0 {
|
|
publishedRange["$gte"] = filter.RangeStart
|
|
}
|
|
if filter.RangeEnd > 0 {
|
|
publishedRange["$lte"] = filter.RangeEnd
|
|
}
|
|
if len(publishedRange) > 0 {
|
|
rangeClauses = append(rangeClauses, bson.M{"published_at": publishedRange})
|
|
}
|
|
if len(rangeClauses) > 0 {
|
|
mongoFilter["$or"] = rangeClauses
|
|
}
|
|
}
|
|
total, err := r.collection.CountDocuments(ctx, mongoFilter)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
opts := options.Find().
|
|
SetSort(bson.D{{Key: "create_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 out []entity.CopyDraft
|
|
if err := cur.All(ctx, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
return &domrepo.InboxListResult{Items: out, Total: total}, nil
|
|
}
|
|
|
|
func (r *mongoRepository) List(ctx context.Context, tenantID, ownerUID, personaID string, limit int) ([]entity.CopyDraft, error) {
|
|
if r.collection == nil {
|
|
return nil, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
if limit > 200 {
|
|
limit = 200
|
|
}
|
|
opts := options.Find().
|
|
SetSort(bson.D{{Key: "sort_order", Value: 1}, {Key: "create_at", Value: -1}}).
|
|
SetLimit(int64(limit))
|
|
cur, err := r.collection.Find(ctx, personaFilter(tenantID, ownerUID, personaID), opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer cur.Close(ctx)
|
|
var out []entity.CopyDraft
|
|
if err := cur.All(ctx, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (r *mongoRepository) Get(ctx context.Context, tenantID, ownerUID, personaID, draftID string) (*entity.CopyDraft, error) {
|
|
if r.collection == nil {
|
|
return nil, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
draftID = strings.TrimSpace(draftID)
|
|
if draftID == "" {
|
|
return nil, app.For(code.Persona).InputMissingRequired("draft_id is required")
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["_id"] = draftID
|
|
var out entity.CopyDraft
|
|
err := r.collection.FindOne(ctx, filter).Decode(&out)
|
|
if err != nil {
|
|
if err == mongo.ErrNoDocuments {
|
|
return nil, app.For(code.Persona).ResNotFound("copy draft not found")
|
|
}
|
|
return nil, err
|
|
}
|
|
return &out, nil
|
|
}
|
|
|
|
func (r *mongoRepository) Update(ctx context.Context, tenantID, ownerUID, personaID, draftID string, patch map[string]interface{}) (*entity.CopyDraft, error) {
|
|
if r.collection == nil {
|
|
return nil, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
if len(patch) == 0 {
|
|
return nil, app.For(code.Persona).InputMissingRequired("patch is required")
|
|
}
|
|
draftID = strings.TrimSpace(draftID)
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["_id"] = draftID
|
|
opts := options.FindOneAndUpdate().SetReturnDocument(options.After)
|
|
var out entity.CopyDraft
|
|
err := r.collection.FindOneAndUpdate(ctx, filter, bson.M{"$set": patch}, opts).Decode(&out)
|
|
if err != nil {
|
|
if err == mongo.ErrNoDocuments {
|
|
return nil, app.For(code.Persona).ResNotFound("copy draft not found")
|
|
}
|
|
return nil, err
|
|
}
|
|
return &out, nil
|
|
}
|
|
|
|
func (r *mongoRepository) Delete(ctx context.Context, tenantID, ownerUID, personaID, draftID string) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
draftID = strings.TrimSpace(draftID)
|
|
if draftID == "" {
|
|
return app.For(code.Persona).InputMissingRequired("draft_id is required")
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["_id"] = draftID
|
|
res, err := r.collection.DeleteOne(ctx, filter)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if res.DeletedCount == 0 {
|
|
return app.For(code.Persona).ResNotFound("copy draft not found")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *mongoRepository) DeleteMany(
|
|
ctx context.Context,
|
|
tenantID, ownerUID, personaID, missionID, draftType string,
|
|
draftIDs []string,
|
|
) (int64, error) {
|
|
if r.collection == nil {
|
|
return 0, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
missionID = strings.TrimSpace(missionID)
|
|
if missionID == "" {
|
|
return 0, app.For(code.Persona).InputMissingRequired("copy_mission_id is required")
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["copy_mission_id"] = missionID
|
|
if draftType = strings.TrimSpace(draftType); draftType != "" {
|
|
filter["draft_type"] = draftType
|
|
}
|
|
ids := make([]string, 0, len(draftIDs))
|
|
for _, id := range draftIDs {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" {
|
|
continue
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
if len(ids) > 0 {
|
|
filter["_id"] = bson.M{"$in": ids}
|
|
}
|
|
res, err := r.collection.DeleteMany(ctx, filter)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return res.DeletedCount, nil
|
|
}
|
|
|
|
func (r *mongoRepository) DeleteByMission(ctx context.Context, tenantID, ownerUID, personaID, missionID string) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
missionID = strings.TrimSpace(missionID)
|
|
if missionID == "" {
|
|
return app.For(code.Persona).InputMissingRequired("copy_mission_id is required")
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["copy_mission_id"] = missionID
|
|
_, err := r.collection.DeleteMany(ctx, filter)
|
|
return err
|
|
}
|
|
|
|
func (r *mongoRepository) DeleteByMissionAndType(ctx context.Context, tenantID, ownerUID, personaID, missionID, draftType string) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
missionID = strings.TrimSpace(missionID)
|
|
draftType = strings.TrimSpace(draftType)
|
|
if missionID == "" || draftType == "" {
|
|
return app.For(code.Persona).InputMissingRequired("copy_mission_id and draft_type are required")
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["copy_mission_id"] = missionID
|
|
filter["draft_type"] = draftType
|
|
_, err := r.collection.DeleteMany(ctx, filter)
|
|
return err
|
|
}
|
|
|
|
func matrixDraftFilter(tenantID, ownerUID, personaID, missionID string) bson.M {
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["copy_mission_id"] = strings.TrimSpace(missionID)
|
|
filter["draft_type"] = entity.DraftTypeMatrix
|
|
return filter
|
|
}
|
|
|
|
func (r *mongoRepository) ReplaceMissionMatrix(
|
|
ctx context.Context,
|
|
tenantID, ownerUID, personaID, missionID string,
|
|
drafts []*entity.CopyDraft,
|
|
) error {
|
|
if r.collection == nil {
|
|
return app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
missionID = strings.TrimSpace(missionID)
|
|
if missionID == "" {
|
|
return app.For(code.Persona).InputMissingRequired("copy_mission_id is required")
|
|
}
|
|
if err := r.replaceMissionMatrixWithTransaction(ctx, tenantID, ownerUID, personaID, missionID, drafts); err == nil {
|
|
return nil
|
|
}
|
|
return r.replaceMissionMatrixInsertFirst(ctx, tenantID, ownerUID, personaID, missionID, drafts)
|
|
}
|
|
|
|
func (r *mongoRepository) replaceMissionMatrixWithTransaction(
|
|
ctx context.Context,
|
|
tenantID, ownerUID, personaID, missionID string,
|
|
drafts []*entity.CopyDraft,
|
|
) error {
|
|
client := r.collection.Database().Client()
|
|
session, err := client.StartSession()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer session.EndSession(ctx)
|
|
|
|
wc := writeconcern.Majority()
|
|
txnOpts := options.Transaction().SetWriteConcern(wc)
|
|
_, err = session.WithTransaction(ctx, func(sessCtx mongo.SessionContext) (any, error) {
|
|
filter := matrixDraftFilter(tenantID, ownerUID, personaID, missionID)
|
|
if _, err := r.collection.DeleteMany(sessCtx, filter); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(drafts) == 0 {
|
|
return nil, nil
|
|
}
|
|
docs := make([]any, 0, len(drafts))
|
|
for _, draft := range drafts {
|
|
if draft == nil {
|
|
continue
|
|
}
|
|
docs = append(docs, draft)
|
|
}
|
|
if len(docs) == 0 {
|
|
return nil, nil
|
|
}
|
|
if _, err := r.collection.InsertMany(sessCtx, docs); err != nil {
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}, txnOpts)
|
|
return err
|
|
}
|
|
|
|
func (r *mongoRepository) replaceMissionMatrixInsertFirst(
|
|
ctx context.Context,
|
|
tenantID, ownerUID, personaID, missionID string,
|
|
drafts []*entity.CopyDraft,
|
|
) error {
|
|
batchID := ""
|
|
if len(drafts) > 0 {
|
|
batchID = strings.TrimSpace(drafts[0].MatrixBatchID)
|
|
}
|
|
if batchID == "" {
|
|
filter := matrixDraftFilter(tenantID, ownerUID, personaID, missionID)
|
|
if _, err := r.collection.DeleteMany(ctx, filter); err != nil {
|
|
return err
|
|
}
|
|
if len(drafts) == 0 {
|
|
return nil
|
|
}
|
|
return r.CreateMany(ctx, drafts)
|
|
}
|
|
if err := r.CreateMany(ctx, drafts); err != nil {
|
|
return err
|
|
}
|
|
filter := matrixDraftFilter(tenantID, ownerUID, personaID, missionID)
|
|
filter["matrix_batch_id"] = bson.M{"$ne": batchID}
|
|
_, err := r.collection.DeleteMany(ctx, filter)
|
|
return err
|
|
}
|
|
|
|
func pendingIdleDraftFilter(tenantID, ownerUID, personaID string) bson.M {
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["$and"] = []bson.M{
|
|
{"$or": []bson.M{{"publish_queue_id": ""}, {"publish_queue_id": bson.M{"$exists": false}}}},
|
|
{"$or": []bson.M{{"published_at": 0}, {"published_at": bson.M{"$exists": false}}}},
|
|
{"$or": []bson.M{{"copy_mission_id": ""}, {"copy_mission_id": bson.M{"$exists": false}}}},
|
|
}
|
|
return filter
|
|
}
|
|
|
|
func (r *mongoRepository) PruneOldestPendingBeyond(ctx context.Context, tenantID, ownerUID, personaID string, keep int) (int64, error) {
|
|
if r.collection == nil {
|
|
return 0, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
if keep <= 0 {
|
|
keep = 40
|
|
}
|
|
filter := pendingIdleDraftFilter(tenantID, ownerUID, personaID)
|
|
total, err := r.collection.CountDocuments(ctx, filter)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
excess := total - int64(keep)
|
|
if excess <= 0 {
|
|
return 0, nil
|
|
}
|
|
opts := options.Find().
|
|
SetSort(bson.D{{Key: "create_at", Value: 1}}).
|
|
SetLimit(excess).
|
|
SetProjection(bson.M{"_id": 1})
|
|
cur, err := r.collection.Find(ctx, filter, opts)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer cur.Close(ctx)
|
|
ids := make([]string, 0, excess)
|
|
for cur.Next(ctx) {
|
|
var row struct {
|
|
ID string `bson:"_id"`
|
|
}
|
|
if err := cur.Decode(&row); err != nil {
|
|
return 0, err
|
|
}
|
|
if row.ID != "" {
|
|
ids = append(ids, row.ID)
|
|
}
|
|
}
|
|
if err := cur.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
if len(ids) == 0 {
|
|
return 0, nil
|
|
}
|
|
deleteFilter := personaFilter(tenantID, ownerUID, personaID)
|
|
deleteFilter["_id"] = bson.M{"$in": ids}
|
|
res, err := r.collection.DeleteMany(ctx, deleteFilter)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return res.DeletedCount, nil
|
|
}
|
|
|
|
func (r *mongoRepository) ListByMission(ctx context.Context, tenantID, ownerUID, personaID, missionID string, limit int) ([]entity.CopyDraft, error) {
|
|
if r.collection == nil {
|
|
return nil, app.For(code.Persona).DBUnavailable("Mongo is not configured")
|
|
}
|
|
missionID = strings.TrimSpace(missionID)
|
|
if missionID == "" {
|
|
return nil, app.For(code.Persona).InputMissingRequired("copy_mission_id is required")
|
|
}
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
if limit > 200 {
|
|
limit = 200
|
|
}
|
|
filter := personaFilter(tenantID, ownerUID, personaID)
|
|
filter["copy_mission_id"] = missionID
|
|
opts := options.Find().
|
|
SetSort(bson.D{{Key: "sort_order", Value: 1}, {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.CopyDraft
|
|
if err := cur.All(ctx, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
return out, nil
|
|
}
|