FranklinWH-Toolbox/service/repositories/mongo/dataDownloadRepo.go

package mongo

import (
	"context"
	"errors"
	"fmt"
	"time"

	"fwh-toolbox-server/database"
	"fwh-toolbox-server/domain"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

const (
	ddTasks        = "data_download_tasks"
	ddItems        = "data_download_items"
	ddDocuments    = "data_download_documents"
	ddTaskDocs     = "data_download_task_documents"
	ddRenderSlots  = "data_download_render_slots"
	ddArtifacts    = "data_download_artifacts"
	ddPackages     = "data_download_package_jobs"
	ddFaultEntries = "data_download_fault_record_entries"
	ddFileAttempts = "data_download_file_attempts"
	ddFaultCalls   = "data_download_fault_calls"
)

func InitDataDownloadIndexes(ctx context.Context) error {
	indexes := map[string][]mongo.IndexModel{
		ddTasks: {
			{Keys: bson.D{{Key: "ownerId", Value: 1}, {Key: "createdAt", Value: -1}}},
			{Keys: bson.D{{Key: "status", Value: 1}, {Key: "updatedAt", Value: 1}}},
		},
		ddItems: {
			{Keys: bson.D{{Key: "taskId", Value: 1}, {Key: "itemKey", Value: 1}}, Options: options.Index().SetUnique(true)},
			{Keys: bson.D{{Key: "taskId", Value: 1}, {Key: "status", Value: 1}}},
		},
		ddFileAttempts: {
			{Keys: bson.D{{Key: "taskId", Value: 1}, {Key: "itemId", Value: 1}, {Key: "sourceFileId", Value: 1}}, Options: options.Index().SetUnique(true)},
			{Keys: bson.D{{Key: "taskId", Value: 1}, {Key: "status", Value: 1}}},
		},
		ddFaultCalls: {
			{Keys: bson.D{{Key: "expiresAt", Value: 1}}, Options: options.Index().SetExpireAfterSeconds(0)},
		},
		ddDocuments: {
			{Keys: bson.D{{Key: "requestKey", Value: 1}}, Options: options.Index().SetUnique(true)},
			{Keys: bson.D{{Key: "kind", Value: 1}, {Key: "siteSn", Value: 1}, {Key: "date", Value: 1}}},
		},
		ddFaultEntries: {
			{Keys: bson.D{{Key: "documentId", Value: 1}, {Key: "sourceFileId", Value: 1}}, Options: options.Index().SetUnique(true)},
			{Keys: bson.D{{Key: "documentId", Value: 1}, {Key: "fileName", Value: 1}}},
		},
		ddTaskDocs: {
			{Keys: bson.D{{Key: "taskId", Value: 1}, {Key: "itemKey", Value: 1}, {Key: "documentId", Value: 1}}, Options: options.Index().SetUnique(true)},
			{Keys: bson.D{{Key: "documentId", Value: 1}}},
		},
		ddRenderSlots: {
			{Keys: bson.D{{Key: "documentId", Value: 1}, {Key: "format", Value: 1}}, Options: options.Index().SetUnique(true)},
		},
		ddArtifacts: {
			{Keys: bson.D{{Key: "documentId", Value: 1}, {Key: "format", Value: 1}, {Key: "contentGeneration", Value: 1}}},
			{Keys: bson.D{{Key: "expiresAt", Value: 1}}},
		},
		ddPackages: {
			{Keys: bson.D{{Key: "ownerId", Value: 1}, {Key: "createdAt", Value: -1}}},
			{Keys: bson.D{{Key: "taskIds", Value: 1}}},
			{Keys: bson.D{{Key: "purgeAt", Value: 1}}},
		},
	}
	for collection, models := range indexes {
		if _, err := database.GetCollection(collection).Indexes().CreateMany(ctx, models); err != nil {
			return fmt.Errorf("create %s indexes: %w", collection, err)
		}
	}
	return nil
}

func InsertDataDownloadTask(ctx context.Context, task *domain.DataDownloadTask) error {
	result, err := database.GetCollection(ddTasks).InsertOne(ctx, task)
	if err == nil {
		task.ID = result.InsertedID.(primitive.ObjectID)
	}
	return err
}

func GetDataDownloadTask(ctx context.Context, id primitive.ObjectID) (*domain.DataDownloadTask, error) {
	var task domain.DataDownloadTask
	err := database.GetCollection(ddTasks).FindOne(ctx, bson.M{"_id": id}).Decode(&task)
	return &task, err
}

func ListDataDownloadTasks(ctx context.Context, ownerID string, mineOnly bool, description, sn, status string, createdFrom, createdTo *time.Time, page, pageSize int64) ([]domain.DataDownloadTask, int64, error) {
	filter := bson.M{}
	if mineOnly {
		filter["ownerId"] = ownerID
	}
	if description != "" {
		filter["description"] = primitive.Regex{Pattern: regexpQuote(description), Options: "i"}
	}
	if sn != "" {
		filter["sns"] = sn
	}
	if status != "" {
		filter["status"] = status
	}
	if createdFrom != nil || createdTo != nil {
		rangeFilter := bson.M{}
		if createdFrom != nil {
			rangeFilter["$gte"] = *createdFrom
		}
		if createdTo != nil {
			rangeFilter["$lte"] = *createdTo
		}
		filter["createdAt"] = rangeFilter
	}
	coll := database.GetCollection(ddTasks)
	total, err := coll.CountDocuments(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	cursor, err := coll.Find(ctx, filter, options.Find().SetSort(bson.D{{Key: "createdAt", Value: -1}}).SetSkip((page-1)*pageSize).SetLimit(pageSize))
	if err != nil {
		return nil, 0, err
	}
	defer cursor.Close(ctx)
	tasks := make([]domain.DataDownloadTask, 0)
	err = cursor.All(ctx, &tasks)
	return tasks, total, err
}

func regexpQuote(value string) string {
	quoted := ""
	for _, r := range value {
		if r == '\\' || r == '.' || r == '+' || r == '*' || r == '?' || r == '(' || r == ')' || r == '[' || r == ']' || r == '{' || r == '}' || r == '^' || r == '$' || r == '|' {
			quoted += "\\"
		}
		quoted += string(r)
	}
	return quoted
}

func UpdateDataDownloadTaskStatus(ctx context.Context, id primitive.ObjectID, status domain.DataDownloadStatus) error {
	update := bson.M{"$set": bson.M{"status": status, "updatedAt": time.Now().UTC()}}
	if status == domain.DataDownloadSucceeded || status == domain.DataDownloadPartial || status == domain.DataDownloadFailed || status == domain.DataDownloadError || status == domain.DataDownloadCancelled {
		now := time.Now().UTC()
		update["$set"].(bson.M)["finishedAt"] = now
	}
	result, err := database.GetCollection(ddTasks).UpdateByID(ctx, id, update)
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func InsertDataDownloadItems(ctx context.Context, items []domain.DataDownloadItem) error {
	if len(items) == 0 {
		return nil
	}
	models := make([]mongo.WriteModel, len(items))
	keys := make([]string, len(items))
	for i := range items {
		keys[i] = items[i].ItemKey
		models[i] = mongo.NewUpdateOneModel().
			SetFilter(bson.M{"taskId": items[i].TaskID, "itemKey": items[i].ItemKey}).
			SetUpdate(bson.M{"$setOnInsert": items[i]}).
			SetUpsert(true)
	}
	coll := database.GetCollection(ddItems)
	if _, err := coll.BulkWrite(ctx, models, options.BulkWrite().SetOrdered(false)); err != nil {
		return err
	}
	cursor, err := coll.Find(ctx, bson.M{"taskId": items[0].TaskID, "itemKey": bson.M{"$in": keys}})
	if err != nil {
		return err
	}
	defer cursor.Close(ctx)
	var persisted []domain.DataDownloadItem
	if err := cursor.All(ctx, &persisted); err != nil {
		return err
	}
	byKey := make(map[string]primitive.ObjectID, len(persisted))
	for i := range persisted {
		byKey[persisted[i].ItemKey] = persisted[i].ID
	}
	for i := range items {
		id, ok := byKey[items[i].ItemKey]
		if !ok {
			return fmt.Errorf("data-download item %q was not persisted", items[i].ItemKey)
		}
		items[i].ID = id
	}
	return nil
}

func GetDataDownloadItem(ctx context.Context, id primitive.ObjectID) (*domain.DataDownloadItem, error) {
	var item domain.DataDownloadItem
	err := database.GetCollection(ddItems).FindOne(ctx, bson.M{"_id": id}).Decode(&item)
	return &item, err
}

func ListDataDownloadItems(ctx context.Context, taskID primitive.ObjectID) ([]domain.DataDownloadItem, error) {
	cursor, err := database.GetCollection(ddItems).Find(ctx, bson.M{"taskId": taskID}, options.Find().SetSort(bson.D{{Key: "itemKey", Value: 1}}))
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	items := make([]domain.DataDownloadItem, 0)
	err = cursor.All(ctx, &items)
	return items, err
}

func UpdateDataDownloadItem(ctx context.Context, id primitive.ObjectID, status domain.DataDownloadStatus, stage, reason string) error {
	update := bson.M{"$set": bson.M{"status": status, "stage": stage, "failureReason": reason, "updatedAt": time.Now().UTC()}}
	if status == domain.DataDownloadRunning {
		update["$inc"] = bson.M{"attempt": 1}
	}
	result, err := database.GetCollection(ddItems).UpdateByID(ctx, id, update)
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func UpdateDataDownloadFaultCall(ctx context.Context, id primitive.ObjectID, requestedAt *time.Time, requestedCount int, recordID int64, status *int) error {
	set := bson.M{"updatedAt": time.Now().UTC()}
	if requestedAt != nil {
		set["callRequestedAt"] = *requestedAt
	}
	if requestedCount > 0 {
		set["callRequestedCount"] = requestedCount
	}
	if recordID != 0 {
		set["callRecordId"] = recordID
	}
	if status != nil {
		set["callRecordStatus"] = *status
	}
	result, err := database.GetCollection(ddItems).UpdateByID(ctx, id, bson.M{"$set": set})
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func ResetDataDownloadFaultCall(ctx context.Context, id primitive.ObjectID) error {
	result, err := database.GetCollection(ddItems).UpdateByID(ctx, id, bson.M{
		"$unset": bson.M{"callRequestedAt": "", "callRequestedCount": "", "callRecordId": "", "callRecordStatus": ""},
		"$set":   bson.M{"updatedAt": time.Now().UTC()},
	})
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

// TryOwnDataDownloadFaultCall keeps the same logical owner across delayed
// queue executions. A different item cannot start another call for the same
// document until the owner releases it or the original 2-hour window expires.
func TryOwnDataDownloadFaultCall(ctx context.Context, documentID string, itemID primitive.ObjectID, requestedAt time.Time) (bool, error) {
	now := time.Now().UTC()
	expiresAt := requestedAt.Add(2*time.Hour + 10*time.Minute)
	filter := bson.M{"_id": documentID, "$or": bson.A{bson.M{"itemId": itemID}, bson.M{"expiresAt": bson.M{"$lte": now}}}}
	update := bson.M{"$set": bson.M{"itemId": itemID, "requestedAt": requestedAt, "expiresAt": expiresAt, "updatedAt": now}}
	result, err := database.GetCollection(ddFaultCalls).UpdateOne(ctx, filter, update, options.Update().SetUpsert(true))
	if mongo.IsDuplicateKeyError(err) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return result.MatchedCount == 1 || result.UpsertedCount == 1, nil
}

func OtherDataDownloadFaultCallActive(ctx context.Context, documentID string, itemID primitive.ObjectID) (bool, error) {
	var owner struct {
		ItemID    primitive.ObjectID `bson:"itemId"`
		ExpiresAt time.Time          `bson:"expiresAt"`
	}
	err := database.GetCollection(ddFaultCalls).FindOne(ctx, bson.M{"_id": documentID}).Decode(&owner)
	if errors.Is(err, mongo.ErrNoDocuments) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return owner.ItemID != itemID && owner.ExpiresAt.After(time.Now().UTC()), nil
}

func ReleaseDataDownloadFaultCall(ctx context.Context, documentID string, itemID primitive.ObjectID) error {
	_, err := database.GetCollection(ddFaultCalls).DeleteOne(ctx, bson.M{"_id": documentID, "itemId": itemID})
	return err
}

func DeferDataDownloadItem(ctx context.Context, id primitive.ObjectID) error {
	result, err := database.GetCollection(ddItems).UpdateByID(ctx, id, bson.M{
		"$set": bson.M{"status": domain.DataDownloadPending, "stage": "waiting", "failureReason": "", "updatedAt": time.Now().UTC()},
		"$inc": bson.M{"attempt": -1},
	})
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func RecountDataDownloadTask(ctx context.Context, taskID primitive.ObjectID) error {
	pipeline := mongo.Pipeline{
		{{Key: "$match", Value: bson.M{"taskId": taskID}}},
		{{Key: "$group", Value: bson.M{"_id": "$status", "count": bson.M{"$sum": 1}}}},
	}
	cursor, err := database.GetCollection(ddItems).Aggregate(ctx, pipeline)
	if err != nil {
		return err
	}
	defer cursor.Close(ctx)
	type count struct {
		ID    domain.DataDownloadStatus `bson:"_id"`
		Count int64                     `bson:"count"`
	}
	var counts []count
	if err := cursor.All(ctx, &counts); err != nil {
		return err
	}
	set := bson.M{"updatedAt": time.Now().UTC(), "succeeded": int64(0), "failed": int64(0), "empty": int64(0)}
	var total int64
	for _, c := range counts {
		total += c.Count
		switch c.ID {
		case domain.DataDownloadItemSucceeded, domain.DataDownloadSkipped, domain.DataDownloadPartial:
			set["succeeded"] = set["succeeded"].(int64) + c.Count
		case domain.DataDownloadFailed:
			set["failed"] = c.Count
		}
	}
	empty, err := database.GetCollection(ddTaskDocs).CountDocuments(ctx, bson.M{"taskId": taskID, "empty": true})
	if err != nil {
		return err
	}
	set["empty"] = empty
	set["total"] = total
	_, err = database.GetCollection(ddTasks).UpdateByID(ctx, taskID, bson.M{"$set": set})
	return err
}

func UpsertDataDownloadDocument(ctx context.Context, document *domain.DataDownloadDocument) error {
	_, err := database.GetCollection(ddDocuments).ReplaceOne(ctx, bson.M{"_id": document.ID}, document, options.Replace().SetUpsert(true))
	return err
}

func GetDataDownloadDocument(ctx context.Context, id string) (*domain.DataDownloadDocument, error) {
	var document domain.DataDownloadDocument
	err := database.GetCollection(ddDocuments).FindOne(ctx, bson.M{"_id": id}).Decode(&document)
	document.HasBinary = document.BinaryObjectKey != "" || document.FileCount > 0
	return &document, err
}

func ListDataDownloadDocuments(ctx context.Context, page, pageSize int64) ([]domain.DataDownloadDocument, int64, error) {
	coll := database.GetCollection(ddDocuments)
	total, err := coll.CountDocuments(ctx, bson.M{})
	if err != nil {
		return nil, 0, err
	}
	cursor, err := coll.Find(ctx, bson.M{}, options.Find().SetSort(bson.D{{Key: "updatedAt", Value: -1}}).SetSkip((page-1)*pageSize).SetLimit(pageSize))
	if err != nil {
		return nil, 0, err
	}
	defer cursor.Close(ctx)
	documents := make([]domain.DataDownloadDocument, 0)
	err = cursor.All(ctx, &documents)
	for i := range documents {
		documents[i].HasBinary = documents[i].BinaryObjectKey != "" || documents[i].FileCount > 0
	}
	return documents, total, err
}

func CountDataDownloadDocumentReferences(ctx context.Context, documentID string) (int64, error) {
	return database.GetCollection(ddTaskDocs).CountDocuments(ctx, bson.M{"documentId": documentID})
}

func ListDataDownloadArtifacts(ctx context.Context, documentID string) ([]domain.DataDownloadArtifact, error) {
	cursor, err := database.GetCollection(ddArtifacts).Find(ctx, bson.M{"documentId": documentID})
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	artifacts := make([]domain.DataDownloadArtifact, 0)
	err = cursor.All(ctx, &artifacts)
	return artifacts, err
}

func DeleteDataDownloadDocumentMetadata(ctx context.Context, documentID string) error {
	if _, err := database.GetCollection(ddRenderSlots).DeleteMany(ctx, bson.M{"documentId": documentID}); err != nil {
		return err
	}
	if _, err := database.GetCollection(ddArtifacts).DeleteMany(ctx, bson.M{"documentId": documentID}); err != nil {
		return err
	}
	if _, err := database.GetCollection(ddFaultEntries).DeleteMany(ctx, bson.M{"documentId": documentID}); err != nil {
		return err
	}
	result, err := database.GetCollection(ddDocuments).DeleteOne(ctx, bson.M{"_id": documentID})
	if err == nil && result.DeletedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func LinkDataDownloadDocument(ctx context.Context, link *domain.DataDownloadTaskDocument) error {
	_, err := database.GetCollection(ddTaskDocs).InsertOne(ctx, link)
	if mongo.IsDuplicateKeyError(err) {
		return nil
	}
	return err
}

func ListDataDownloadTaskDocuments(ctx context.Context, taskID primitive.ObjectID) ([]domain.DataDownloadTaskDocument, error) {
	cursor, err := database.GetCollection(ddTaskDocs).Find(ctx, bson.M{"taskId": taskID}, options.Find().SetSort(bson.D{{Key: "logicalPath", Value: 1}}))
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	links := make([]domain.DataDownloadTaskDocument, 0)
	err = cursor.All(ctx, &links)
	return links, err
}

func DetachDataDownloadRenderSlots(ctx context.Context, documentID string) error {
	_, err := database.GetCollection(ddRenderSlots).UpdateMany(ctx, bson.M{"documentId": documentID}, bson.M{"$unset": bson.M{"artifactId": ""}, "$set": bson.M{"updatedAt": time.Now().UTC()}})
	return err
}

func GetDataDownloadRenderSlot(ctx context.Context, documentID, format string) (*domain.DataDownloadRenderSlot, error) {
	var slot domain.DataDownloadRenderSlot
	err := database.GetCollection(ddRenderSlots).FindOne(ctx, bson.M{"documentId": documentID, "format": format}).Decode(&slot)
	return &slot, err
}

func GetDataDownloadArtifact(ctx context.Context, id primitive.ObjectID) (*domain.DataDownloadArtifact, error) {
	var artifact domain.DataDownloadArtifact
	err := database.GetCollection(ddArtifacts).FindOne(ctx, bson.M{"_id": id}).Decode(&artifact)
	return &artifact, err
}

func PublishDataDownloadArtifact(ctx context.Context, artifact *domain.DataDownloadArtifact) error {
	result, err := database.GetCollection(ddArtifacts).InsertOne(ctx, artifact)
	if err != nil {
		return err
	}
	artifact.ID = result.InsertedID.(primitive.ObjectID)
	_, err = database.GetCollection(ddRenderSlots).UpdateOne(ctx,
		bson.M{"documentId": artifact.DocumentID, "format": artifact.Format},
		bson.M{"$set": bson.M{"contentGeneration": artifact.ContentGeneration, "artifactId": artifact.ID, "updatedAt": time.Now().UTC()}},
		options.Update().SetUpsert(true))
	return err
}

func UpdateDataDownloadDescription(ctx context.Context, id primitive.ObjectID, description string) error {
	result, err := database.GetCollection(ddTasks).UpdateByID(ctx, id, bson.M{"$set": bson.M{"description": description, "updatedAt": time.Now().UTC()}})
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func DeleteDataDownloadTask(ctx context.Context, id primitive.ObjectID) error {
	if _, err := database.GetCollection(ddTaskDocs).DeleteMany(ctx, bson.M{"taskId": id}); err != nil {
		return err
	}
	if _, err := database.GetCollection(ddItems).DeleteMany(ctx, bson.M{"taskId": id}); err != nil {
		return err
	}
	if _, err := database.GetCollection(ddFileAttempts).DeleteMany(ctx, bson.M{"taskId": id}); err != nil {
		return err
	}
	result, err := database.GetCollection(ddTasks).DeleteOne(ctx, bson.M{"_id": id})
	if err == nil && result.DeletedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func GetDataDownloadFileAttempt(ctx context.Context, taskID, itemID primitive.ObjectID, sourceFileID string) (*domain.DataDownloadFileAttempt, error) {
	var attempt domain.DataDownloadFileAttempt
	err := database.GetCollection(ddFileAttempts).FindOne(ctx, bson.M{"taskId": taskID, "itemId": itemID, "sourceFileId": sourceFileID}).Decode(&attempt)
	return &attempt, err
}

func RecordDataDownloadFileAttempt(ctx context.Context, attempt *domain.DataDownloadFileAttempt) error {
	now := time.Now().UTC()
	set := bson.M{
		"documentId": attempt.DocumentID, "fileName": attempt.FileName, "status": attempt.Status,
		"entryId": attempt.EntryID, "failureStage": attempt.FailureStage,
		"failureReason": attempt.FailureReason, "updatedAt": now,
	}
	_, err := database.GetCollection(ddFileAttempts).UpdateOne(ctx,
		bson.M{"taskId": attempt.TaskID, "itemId": attempt.ItemID, "sourceFileId": attempt.SourceFileID},
		bson.M{"$set": set, "$setOnInsert": bson.M{"createdAt": now}, "$inc": bson.M{"attemptCount": 1}},
		options.Update().SetUpsert(true))
	return err
}

func ListDataDownloadFileAttempts(ctx context.Context, taskID primitive.ObjectID) ([]domain.DataDownloadFileAttempt, error) {
	cursor, err := database.GetCollection(ddFileAttempts).Find(ctx, bson.M{"taskId": taskID}, options.Find().SetSort(bson.D{{Key: "itemId", Value: 1}, {Key: "sourceFileId", Value: 1}}))
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	attempts := make([]domain.DataDownloadFileAttempt, 0)
	err = cursor.All(ctx, &attempts)
	return attempts, err
}

func DeleteDataDownloadFileAttempt(ctx context.Context, taskID, itemID primitive.ObjectID, sourceFileID string) error {
	_, err := database.GetCollection(ddFileAttempts).DeleteOne(ctx, bson.M{
		"taskId": taskID, "itemId": itemID, "sourceFileId": sourceFileID,
	})
	return err
}

func HasDataDownloadTaskDocument(ctx context.Context, taskID primitive.ObjectID, documentID string) (bool, error) {
	count, err := database.GetCollection(ddTaskDocs).CountDocuments(ctx, bson.M{"taskId": taskID, "documentId": documentID})
	return count > 0, err
}

func UpsertDataDownloadFaultRecordEntry(ctx context.Context, entry *domain.DataDownloadFaultRecordEntry) (string, error) {
	var existing domain.DataDownloadFaultRecordEntry
	err := database.GetCollection(ddFaultEntries).FindOne(ctx, bson.M{"documentId": entry.DocumentID, "sourceFileId": entry.SourceFileID}).Decode(&existing)
	oldObjectKey := ""
	if err == nil {
		entry.ID = existing.ID
		entry.CreatedAt = existing.CreatedAt
		oldObjectKey = existing.ObjectKey
	} else if !errors.Is(err, mongo.ErrNoDocuments) {
		return "", err
	}
	if entry.ID.IsZero() {
		entry.ID = primitive.NewObjectID()
	}
	if entry.CreatedAt.IsZero() {
		entry.CreatedAt = entry.UpdatedAt
	}
	_, err = database.GetCollection(ddFaultEntries).ReplaceOne(ctx,
		bson.M{"documentId": entry.DocumentID, "sourceFileId": entry.SourceFileID}, entry, options.Replace().SetUpsert(true))
	return oldObjectKey, err
}

func GetDataDownloadFaultRecordEntry(ctx context.Context, documentID, sourceFileID string) (*domain.DataDownloadFaultRecordEntry, error) {
	var entry domain.DataDownloadFaultRecordEntry
	err := database.GetCollection(ddFaultEntries).FindOne(ctx, bson.M{"documentId": documentID, "sourceFileId": sourceFileID}).Decode(&entry)
	return &entry, err
}

func GetDataDownloadFaultRecordEntryByID(ctx context.Context, id primitive.ObjectID) (*domain.DataDownloadFaultRecordEntry, error) {
	var entry domain.DataDownloadFaultRecordEntry
	err := database.GetCollection(ddFaultEntries).FindOne(ctx, bson.M{"_id": id}).Decode(&entry)
	return &entry, err
}

func ListDataDownloadFaultRecordEntries(ctx context.Context, documentID string) ([]domain.DataDownloadFaultRecordEntry, error) {
	cursor, err := database.GetCollection(ddFaultEntries).Find(ctx, bson.M{"documentId": documentID}, options.Find().SetSort(bson.D{{Key: "fileName", Value: 1}, {Key: "sourceFileId", Value: 1}}))
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	entries := make([]domain.DataDownloadFaultRecordEntry, 0)
	err = cursor.All(ctx, &entries)
	return entries, err
}

func InsertDataDownloadPackageJob(ctx context.Context, job *domain.DataDownloadPackageJob) error {
	result, err := database.GetCollection(ddPackages).InsertOne(ctx, job)
	if err == nil {
		job.ID = result.InsertedID.(primitive.ObjectID)
	}
	return err
}

func GetDataDownloadPackageJob(ctx context.Context, id primitive.ObjectID) (*domain.DataDownloadPackageJob, error) {
	var job domain.DataDownloadPackageJob
	err := database.GetCollection(ddPackages).FindOne(ctx, bson.M{"_id": id}).Decode(&job)
	return &job, err
}

func ListDataDownloadPackageJobs(ctx context.Context, ownerID string, mineOnly bool, page, pageSize int64) ([]domain.DataDownloadPackageJob, int64, error) {
	filter := bson.M{}
	if mineOnly {
		filter["ownerId"] = ownerID
	}
	coll := database.GetCollection(ddPackages)
	total, err := coll.CountDocuments(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	cursor, err := coll.Find(ctx, filter, options.Find().SetSort(bson.D{{Key: "createdAt", Value: -1}}).SetSkip((page-1)*pageSize).SetLimit(pageSize))
	if err != nil {
		return nil, 0, err
	}
	defer cursor.Close(ctx)
	jobs := make([]domain.DataDownloadPackageJob, 0)
	err = cursor.All(ctx, &jobs)
	return jobs, total, err
}

func UpdateDataDownloadPackageJob(ctx context.Context, id primitive.ObjectID, set bson.M) error {
	set["updatedAt"] = time.Now().UTC()
	result, err := database.GetCollection(ddPackages).UpdateByID(ctx, id, bson.M{"$set": set})
	if err == nil && result.MatchedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func DeleteDataDownloadPackageJob(ctx context.Context, id primitive.ObjectID) error {
	result, err := database.GetCollection(ddPackages).DeleteOne(ctx, bson.M{"_id": id})
	if err == nil && result.DeletedCount != 1 {
		return mongo.ErrNoDocuments
	}
	return err
}

func FindExpiredDataDownloadPackageJobs(ctx context.Context, now time.Time) ([]domain.DataDownloadPackageJob, error) {
	cursor, err := database.GetCollection(ddPackages).Find(ctx, bson.M{"status": domain.DataDownloadSucceeded, "expiresAt": bson.M{"$lte": now}})
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	jobs := make([]domain.DataDownloadPackageJob, 0)
	err = cursor.All(ctx, &jobs)
	return jobs, err
}

func PurgeExpiredDataDownloadPackageJobs(ctx context.Context, now time.Time) error {
	_, err := database.GetCollection(ddPackages).DeleteMany(ctx, bson.M{"status": "expired", "purgeAt": bson.M{"$lte": now}})
	return err
}

func FindExpiredDataDownloadArtifacts(ctx context.Context, now time.Time) ([]domain.DataDownloadArtifact, error) {
	cursor, err := database.GetCollection(ddArtifacts).Find(ctx, bson.M{"expiresAt": bson.M{"$lte": now}})
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)
	artifacts := make([]domain.DataDownloadArtifact, 0)
	err = cursor.All(ctx, &artifacts)
	return artifacts, err
}

func DetachDataDownloadArtifact(ctx context.Context, artifact domain.DataDownloadArtifact) error {
	_, err := database.GetCollection(ddRenderSlots).UpdateMany(ctx, bson.M{"artifactId": artifact.ID}, bson.M{"$unset": bson.M{"artifactId": ""}, "$set": bson.M{"updatedAt": time.Now().UTC()}})
	return err
}

func DeleteDataDownloadArtifact(ctx context.Context, artifact domain.DataDownloadArtifact) error {
	_, err := database.GetCollection(ddArtifacts).DeleteOne(ctx, bson.M{"_id": artifact.ID})
	return err
}