thread-master/backend/internal/model/mention_inbox/repository/mongo.go

package repository

import (
	"context"

	app "haixun-backend/internal/library/errors"
	"haixun-backend/internal/library/errors/code"
	"haixun-backend/internal/model/mention_inbox/domain/entity"
	domrepo "haixun-backend/internal/model/mention_inbox/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: "media_id", Value: 1},
			},
			Options: options.Index().SetUnique(true),
		},
		{
			Keys: bson.D{
				{Key: "tenant_id", Value: 1},
				{Key: "owner_uid", Value: 1},
				{Key: "account_id", Value: 1},
				{Key: "synced_at", Value: -1},
			},
		},
		{
			Keys: bson.D{
				{Key: "tenant_id", Value: 1},
				{Key: "owner_uid", Value: 1},
				{Key: "account_id", Value: 1},
				{Key: "timestamp", Value: -1},
			},
		},
	})
	return err
}

func (r *mongoRepository) Upsert(ctx context.Context, item *entity.Mention) (*entity.Mention, error) {
	if r.collection == nil {
		return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
	}
	if item == nil {
		return nil, app.For(code.ThreadsAccount).InputMissingRequired("mention is required")
	}
	filter := actorMediaFilter(item.TenantID, item.OwnerUID, item.AccountID, item.MediaID)
	opts := options.Replace().SetUpsert(true)
	if _, err := r.collection.ReplaceOne(ctx, filter, item, opts); err != nil {
		return nil, err
	}
	return item, nil
}

func (r *mongoRepository) Get(ctx context.Context, tenantID, ownerUID, accountID, mediaID string) (*entity.Mention, error) {
	if r.collection == nil {
		return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
	}
	var out entity.Mention
	err := r.collection.FindOne(ctx, actorMediaFilter(tenantID, ownerUID, accountID, mediaID)).Decode(&out)
	if err != nil {
		if err == mongo.ErrNoDocuments {
			return nil, app.For(code.ThreadsAccount).ResNotFound("mention inbox item not found")
		}
		return nil, err
	}
	return &out, nil
}

func normalizeMentionPage(page, pageSize int) (int, int) {
	if page <= 0 {
		page = 1
	}
	if pageSize <= 0 {
		pageSize = 10
	}
	if pageSize > 50 {
		pageSize = 50
	}
	return page, pageSize
}

func accountFilter(tenantID, ownerUID, accountID string) bson.M {
	return bson.M{
		"tenant_id":  tenantID,
		"owner_uid":  ownerUID,
		"account_id": accountID,
	}
}

func (r *mongoRepository) CountByAccount(ctx context.Context, tenantID, ownerUID, accountID string) (int64, error) {
	if r.collection == nil {
		return 0, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
	}
	return r.collection.CountDocuments(ctx, accountFilter(tenantID, ownerUID, accountID))
}

func (r *mongoRepository) CountReplyReadyByAccount(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 := accountFilter(tenantID, ownerUID, accountID)
	filter["reply_ready"] = true
	return r.collection.CountDocuments(ctx, filter)
}

func (r *mongoRepository) ListByAccount(ctx context.Context, tenantID, ownerUID, accountID string, page, pageSize int) (*domrepo.ListResult, error) {
	if r.collection == nil {
		return nil, app.For(code.ThreadsAccount).DBUnavailable("Mongo is not configured")
	}
	page, pageSize = normalizeMentionPage(page, pageSize)
	filter := accountFilter(tenantID, ownerUID, accountID)
	total, err := r.collection.CountDocuments(ctx, filter)
	if err != nil {
		return nil, err
	}
	sort := bson.D{{Key: "timestamp", Value: -1}, {Key: "synced_at", Value: -1}}
	opts := options.Find().
		SetSort(sort).
		SetSkip(int64((page - 1) * pageSize)).
		SetLimit(int64(pageSize))
	cur, err := r.collection.Find(ctx, filter, opts)
	if err != nil {
		return nil, err
	}
	defer cur.Close(ctx)
	out := make([]entity.Mention, 0, pageSize)
	for cur.Next(ctx) {
		var item entity.Mention
		if err := cur.Decode(&item); err != nil {
			return nil, err
		}
		out = append(out, item)
	}
	if err := cur.Err(); err != nil {
		return nil, err
	}
	var newest entity.Mention
	newestErr := r.collection.FindOne(ctx, filter, options.FindOne().SetSort(bson.D{{Key: "synced_at", Value: -1}})).Decode(&newest)
	newestSyncedAt := int64(0)
	if newestErr == nil {
		newestSyncedAt = newest.SyncedAt
	}
	readyTotal, err := r.CountReplyReadyByAccount(ctx, tenantID, ownerUID, accountID)
	if err != nil {
		return nil, err
	}
	return &domrepo.ListResult{
		Items:          out,
		Total:          total,
		ReadyTotal:     readyTotal,
		NewestSyncedAt: newestSyncedAt,
	}, nil
}

func actorMediaFilter(tenantID, ownerUID, accountID, mediaID string) bson.M {
	return bson.M{
		"tenant_id":  tenantID,
		"owner_uid":  ownerUID,
		"account_id": accountID,
		"media_id":   mediaID,
	}
}