// Copyright (C) MongoDB, Inc. 2014-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

package mongodump

import (
	"context"
	"io"
	"os"
	"path/filepath"
	"testing"

	"github.com/mongodb/mongo-tools/common/bsonutil"
	"github.com/mongodb/mongo-tools/common/db"
	"github.com/mongodb/mongo-tools/common/failpoint"
	"github.com/mongodb/mongo-tools/common/log"
	"github.com/mongodb/mongo-tools/common/testtype"
	"github.com/mongodb/mongo-tools/common/testutil"
	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	"go.mongodb.org/mongo-driver/v2/mongo/options"
)

func TestErrorOnImportCollection(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.UnitTestType)
	rawOp, err := os.ReadFile("./testdata/importCollection.bson")
	require.NoError(t, err)

	err = oplogDocumentValidator(rawOp)
	require.Error(t, err)
	assert.EqualError(t, err, "cannot dump with oplog while importCollection occurs")
}

func TestErrorOnViewlessTimeseriesConversion(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.UnitTestType)

	rawOp, err := bson.Marshal(db.Oplog{
		Timestamp: bson.Timestamp{T: 1, I: 1},
		Version:   2,
		Operation: "c",
		Namespace: "test.$cmd",
		Object: bson.D{
			{"upgradeDowngradeViewlessTimeseries", "foo_ts"},
			{"isUpgrade", true},
		},
	})
	require.NoError(t, err)

	err = oplogDocumentValidator(rawOp)
	assert.EqualError(
		t,
		err,
		"cannot dump with oplog while a timeseries format conversion caused by an FCV "+
			"change between 8.x and 9.0 occurs",
		"a dump whose oplog window spans a viewless timeseries conversion is rejected",
	)
}

// TestOplogDumpVectoredInsertsOplog tests dumping oplogs that are from vectored inserts.
// They have a special oplog format.
func TestOplogDumpVectoredInsertsOplog(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	// Oplog is not available in a standalone topology.
	testtype.SkipUnlessTestType(t, testtype.ReplSetTestType)

	log.SetWriter(io.Discard)

	session, err := testutil.GetBareSession()
	if err != nil {
		t.Fatalf("Failed to get session: %v", err)
	}
	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "8.0"); err != nil || cmp < 0 {
		if err != nil {
			t.Errorf("error getting FCV: %v", err)
		}
		t.Skipf("Requires server with FCV 8.0 or later; found %v", fcv)
	}

	ctx := t.Context()

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = ""
	md.OutputOptions.Oplog = true
	md.OutputOptions.Out = "vectored_inserts"

	require.NoError(t, md.Init())

	// Pause mongodump right after it captures the oplog start time so that the
	// test can create oplog entries that are guaranteed to fall inside the dump
	// window, then let it resume.
	require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
	defer failpoint.DefaultManager.Reset()

	//nolint:errcheck
	defer tearDownMongoDumpTestData(t)

	dumpErrCh := make(chan error, 1)
	go func() { dumpErrCh <- md.Dump() }()

	fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
	require.True(t, ok, "PauseUntilResumed failpoint should be enabled")
	require.NoError(t, fp.Reached(context.TODO()))
	require.NoError(t, vectoredInsert(ctx))
	fp.Signal()

	require.NoError(t, <-dumpErrCh)

	path, err := os.Getwd()
	require.NoError(t, err)

	dumpDir := filepath.FromSlash(filepath.Join(path, "vectored_inserts"))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	oplogFilePath := filepath.FromSlash(filepath.Join(dumpDir, "oplog.bson"))
	require.True(t, fileDirExists(dumpDir))
	require.True(t, fileDirExists(dumpDBDir))
	require.True(t, fileDirExists(oplogFilePath))

	defer os.RemoveAll(dumpDir)

	oplogFile, err := os.Open(oplogFilePath)
	require.NoError(t, err)
	defer oplogFile.Close()

	// The vectored insert happens during the dump window, so its oplog entry is
	// somewhere in the dumped oplog but not necessarily the first entry. Scan all
	// entries and assert that one of them is the vectored (multiOp) insert.
	bsonSrc := db.NewDecodedBSONSource(db.NewBufferlessBSONSource(oplogFile))
	foundMultiOpInsert := false
	var entry bson.M
	for bsonSrc.Next(&entry) {
		if entry["multiOpType"] == int32(1) {
			foundMultiOpInsert = true
			break
		}
	}
	require.NoError(t, bsonSrc.Err())
	require.True(
		t,
		foundMultiOpInsert,
		"oplog dump should contain a vectored insert entry with multiOpType == 1",
	)
}

func vectoredInsert(ctx context.Context) error {
	client, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	if sessionErr := client.UseSessionWithOptions(
		ctx,
		options.Session().SetCausalConsistency(false),
		func(sessionContext context.Context) error {
			docs := []any{
				bson.D{{"_id", 100}, {"a", 1}},
				bson.D{{"_id", 200}, {"a", 2}},
			}
			_, insertErr := client.Database(testDB).Collection(testCollectionNames[0]).InsertMany(ctx, docs)
			if insertErr != nil {
				return insertErr
			}

			return nil
		},
	); sessionErr != nil {
		return sessionErr
	}

	return nil
}

func TestOplogDumpCollModIndexUniqueness(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	// Oplog is not available in a standalone topology.
	testtype.SkipUnlessTestType(t, testtype.ReplSetTestType)

	ctx := t.Context()

	session, err := testutil.GetBareSession()
	if err != nil {
		t.Fatalf("Failed to get session: %v", err)
	}
	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "6.0"); err != nil || cmp < 0 {
		if err != nil {
			t.Errorf("error getting FCV: %v", err)
		}
		t.Skipf("Requires server with FCV 6.0 or later; found %v", fcv)
	}

	testCollName := testCollectionNames[0]

	err = session.Database(testDB).Collection(testCollName).Drop(ctx)
	require.NoError(t, err)

	err = session.Database(testDB).CreateCollection(ctx, testCollName)
	require.NoError(t, err)
	//nolint:errcheck
	defer session.Database(testDB).Collection(testCollName).Drop(ctx)

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = ""
	md.OutputOptions.Oplog = true
	md.OutputOptions.Out = "collMod_indexUniqueness"

	require.NoError(t, md.Init())

	// Pause mongodump right after it captures the oplog start time so that the
	// test can create oplog entries during the dump window, then let it resume.
	require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
	defer failpoint.DefaultManager.Reset()

	//nolint:errcheck
	defer tearDownMongoDumpTestData(t)

	dumpErrCh := make(chan error, 1)
	go func() { dumpErrCh <- md.Dump() }()

	fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
	require.True(t, ok, "PauseUntilResumed failpoint should be enabled")
	require.NoError(t, fp.Reached(context.TODO()))
	require.NoError(t, createIndexesAndCollModIndexUniqueness(ctx))
	fp.Signal()

	require.NoError(t, <-dumpErrCh)

	path, err := os.Getwd()
	require.NoError(t, err)

	dumpDir := filepath.FromSlash(filepath.Join(path, "collMod_indexUniqueness"))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	oplogFilePath := filepath.FromSlash(filepath.Join(dumpDir, "oplog.bson"))
	require.True(t, fileDirExists(dumpDir))
	require.True(t, fileDirExists(dumpDBDir))
	require.True(t, fileDirExists(oplogFilePath))

	defer os.RemoveAll(dumpDir)

	oplogFile, err := os.Open(oplogFilePath)
	require.NoError(t, err)
	defer oplogFile.Close()

	bsonSrc := db.NewDecodedBSONSource(db.NewBufferlessBSONSource(oplogFile))
	prepareUniqueTrueCount := 0
	prepareUniqueFalseCount := 0
	forceNonUniqueCount := 0

	var oplog db.Oplog
	for bsonSrc.Next(&oplog) {
		require.NoError(t, bsonSrc.Err())

		if oplog.Namespace == testDB+".$cmd" {
			indexDoc, ok := bsonutil.ToMap(oplog.Object)["index"].(bson.D)
			indexDocMap := bsonutil.ToMap(indexDoc)
			if ok {
				if indexDocMap["prepareUnique"] == true {
					prepareUniqueTrueCount++
				} else if indexDocMap["prepareUnique"] == false {
					prepareUniqueFalseCount++
				} else if indexDocMap["forceNonUnique"] == true {
					forceNonUniqueCount++
				}
			}
		}
	}
	require.NoError(t, oplogFile.Close())
	require.Equal(t, 8, prepareUniqueTrueCount)
	require.Equal(t, 4, prepareUniqueFalseCount)
	require.Equal(t, 4, forceNonUniqueCount)
}

func createIndexesAndCollModIndexUniqueness(ctx context.Context) error {
	client, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	testCollName := testCollectionNames[0]

	indexes := []mongo.IndexModel{
		{
			Keys: bson.D{{"a", 1}},
		},
		{
			Keys:    bson.D{{"b", 1}},
			Options: options.Index().SetHidden(true),
		},
		{
			Keys:    bson.D{{"c", 1}},
			Options: options.Index().SetExpireAfterSeconds(100000),
		},
		{
			Keys:    bson.D{{"d", 1}},
			Options: options.Index().SetExpireAfterSeconds(100000).SetHidden(true),
		},
	}

	_, err = client.Database(testDB).Collection(testCollName).Indexes().CreateMany(
		ctx,
		indexes,
	)
	if err != nil {
		return err
	}

	for _, index := range indexes {
		for _, prepareUnique := range []bool{true, false, true} {
			res := client.Database(testDB).RunCommand(
				ctx,
				bson.D{
					{"collMod", testCollName},
					{"index", bson.D{
						{"keyPattern", index.Keys},
						{"prepareUnique", prepareUnique},
					}},
				},
			)
			if res.Err() != nil {
				return res.Err()
			}
		}

		for _, option := range []string{"unique", "forceNonUnique"} {
			res := client.Database(testDB).RunCommand(
				ctx,
				bson.D{
					{"collMod", testCollName},
					{"index", bson.D{
						{"keyPattern", index.Keys},
						{option, true},
					}},
				},
			)
			if res.Err() != nil {
				return res.Err()
			}
		}
	}

	return nil
}

// This test is here so that we can regenerate the oplog.bson file for the mongorestore test
// `TestOplogRestoreBypassDocumentValidation`.
func TestOplogDumpBypassDocumentValidation(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	// Oplog is not available in a standalone topology.
	testtype.SkipUnlessTestType(t, testtype.ReplSetTestType)

	ctx := t.Context()

	session, err := testutil.GetBareSession()
	if err != nil {
		t.Fatalf("Failed to get session: %v", err)
	}

	testCollName := testCollectionNames[0]

	require.NoError(t, session.Database(testDB).Collection(testCollName).Drop(ctx))
	//nolint:errcheck
	defer session.Database(testDB).Collection(testCollName).Drop(ctx)

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = ""
	md.OutputOptions.Oplog = true
	md.OutputOptions.Out = "oplog_bypassDocumentValidation"

	require.NoError(t, md.Init())

	// Pause mongodump right after it captures the oplog start time so that the
	// test can create oplog entries during the dump window, then let it resume.
	require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
	defer failpoint.DefaultManager.Reset()

	//nolint:errcheck
	defer tearDownMongoDumpTestData(t)

	dumpErrCh := make(chan error, 1)
	go func() { dumpErrCh <- md.Dump() }()

	fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
	require.True(t, ok, "PauseUntilResumed failpoint should be enabled")
	require.NoError(t, fp.Reached(context.TODO()))
	createCollectionWithValidatorAndInsertBypassValidation(ctx, t)
	fp.Signal()

	require.NoError(t, <-dumpErrCh)

	path, err := os.Getwd()
	require.NoError(t, err)

	dumpDir := filepath.FromSlash(filepath.Join(path, "oplog_bypassDocumentValidation"))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	oplogFilePath := filepath.FromSlash(filepath.Join(dumpDir, "oplog.bson"))
	require.True(t, fileDirExists(dumpDir))
	require.True(t, fileDirExists(dumpDBDir))
	require.True(t, fileDirExists(oplogFilePath))

	defer os.RemoveAll(dumpDir)

	oplogFile, err := os.Open(oplogFilePath)
	require.NoError(t, err)
	defer oplogFile.Close()

	bsonSrc := db.NewDecodedBSONSource(db.NewBufferlessBSONSource(oplogFile))

	createCount := 0
	insertCount := 0
	var oplog db.Oplog
	for bsonSrc.Next(&oplog) {
		require.NoError(t, bsonSrc.Err())

		objMap := bsonutil.ToMap(oplog.Object)
		if oplog.Namespace == "mongodump_test_db.$cmd" {
			if objMap["create"] != nil {
				createCount++
				assert.NotEmpty(t, objMap["validator"], "create oplog has validator option")
			}
		}

		if oplog.Namespace == "mongodump_test_db.coll1" {
			if oplog.Operation == "i" {
				insertCount++
			}
		}
	}
	assert.Equal(t, 1, createCount)
	assert.Equal(t, 3, insertCount)
}

func createCollectionWithValidatorAndInsertBypassValidation(ctx context.Context, t *testing.T) {
	client, err := testutil.GetBareSession()
	require.NoError(t, err)

	testCollName := testCollectionNames[0]

	err = client.Database(testDB).
		CreateCollection(ctx, testCollName, options.CreateCollection().SetValidator(bson.D{
			{"$jsonSchema", bson.D{
				{"bsonType", "object"},
				{"title", "Test validator"},
				{"properties", bson.D{
					{"testField", bson.D{
						{"enum", []string{"valid"}},
					}},
				}},
			}},
		}))
	require.NoError(t, err)

	invalidDocs := []bson.D{
		{{"testField", "invalid"}},
		{{"testField", "also_invalid"}},
	}

	for _, doc := range invalidDocs {
		_, err = client.Database(testDB).
			Collection(testCollName).
			InsertOne(
				ctx,
				doc,
				options.InsertOne().SetBypassDocumentValidation(true),
			)
		require.NoError(t, err)
	}

	validDoc := bson.D{{"testField", "valid"}}
	_, err = client.Database(testDB).
		Collection(testCollName).
		InsertOne(ctx, validDoc)
	require.NoError(t, err)

	// Verify that the validator is set correctly.
	for _, doc := range invalidDocs {
		_, err = client.Database(testDB).Collection(testCollName).InsertOne(ctx, doc)
		require.Error(t, err)
	}
}

// This test is here so that we can regenerate the oplog.bson file for the mongorestore test
// `TestOplogRestoreCollModTTLIndex`.
func TestOplogDumpCollModTTL(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	// Oplog is not available in a standalone topology.
	testtype.SkipUnlessTestType(t, testtype.ReplSetTestType)
	testutil.SkipIfFCVLessThan(t, "6.0", "collMod TTL is not supported")

	ctx := t.Context()

	session, err := testutil.GetBareSession()
	require.NoError(t, err)

	testCollName := testCollectionNames[0]
	err = session.Database(testDB).Collection(testCollName).Drop(ctx)
	require.NoError(t, err)

	err = session.Database(testDB).CreateCollection(ctx, testCollName)
	require.NoError(t, err)
	//nolint:errcheck
	defer session.Database(testDB).Collection(testCollName).Drop(ctx)

	idx := mongo.IndexModel{
		Keys: bson.D{{"a", 1}},
	}

	_, err = session.Database(testDB).Collection(testCollName).Indexes().CreateOne(ctx, idx)
	require.NoError(t, err)

	_, err = session.Database(testDB).Collection(testCollName).InsertMany(
		ctx,
		lo.RepeatBy(10, func(i int) bson.D {
			return bson.D{{"a", i}}
		}),
	)
	require.NoError(t, err)

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = ""
	md.OutputOptions.Oplog = true
	md.OutputOptions.Out = "collMod_ttl_index"

	require.NoError(t, md.Init())

	// Pause mongodump right after it captures the oplog start time so that the
	// test can create oplog entries during the dump window, then let it resume.
	require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
	defer failpoint.DefaultManager.Reset()

	//nolint:errcheck
	defer tearDownMongoDumpTestData(t)

	dumpErrCh := make(chan error, 1)
	go func() { dumpErrCh <- md.Dump() }()

	fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
	require.True(t, ok, "PauseUntilResumed failpoint should be enabled")
	require.NoError(t, fp.Reached(context.TODO()))
	convertIndexToTTL(ctx, t)
	fp.Signal()

	require.NoError(t, <-dumpErrCh)

	path, err := os.Getwd()
	require.NoError(t, err)

	dumpDir := filepath.FromSlash(filepath.Join(path, md.OutputOptions.Out))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	oplogFilePath := filepath.FromSlash(filepath.Join(dumpDir, "oplog.bson"))
	require.True(t, fileDirExists(dumpDir))
	require.True(t, fileDirExists(dumpDBDir))
	require.True(t, fileDirExists(oplogFilePath))

	defer os.RemoveAll(dumpDir)

	oplogFile, err := os.Open(oplogFilePath)
	require.NoError(t, err)
	defer oplogFile.Close()

	bsonSrc := db.NewDecodedBSONSource(db.NewBufferlessBSONSource(oplogFile))
	var oplog db.Oplog
	var ttl int64
	for bsonSrc.Next(&oplog) {
		require.NoError(t, bsonSrc.Err())
		if oplog.Namespace == testDB+".$cmd" && oplog.Object != nil {
			indexObj, ok := bsonutil.ToMap(oplog.Object)["index"].(bson.D)
			if !ok {
				continue
			}

			ttl, ok = bsonutil.ToMap(indexObj)["expireAfterSeconds"].(int64)
			if !ok {
				continue
			}
		}
	}
	require.EqualValues(t, 1000, ttl)
}

func convertIndexToTTL(ctx context.Context, t *testing.T) {
	client, err := testutil.GetBareSession()
	require.NoError(t, err)

	testCollName := testCollectionNames[0]

	res := client.Database(testDB).RunCommand(
		ctx,
		bson.D{
			{"collMod", testCollName},
			{"index", bson.D{
				{"keyPattern", bson.D{{"a", 1}}},
				{"expireAfterSeconds", 1000},
			}},
		},
	)
	require.NoError(t, res.Err())
}

// TestOplogRollover verifies that mongodump refuses to produce a dump when
// the oplog has rolled over during dumping and evicted the entry it captured
// its starting timestamp from.
//
// jstests/dump/oplog_rollover_test.js used to test this by racing a fixed
// 15-second sleep against a fixed amount of write load meant to overflow a
// deliberately tiny (2MB) oplog, and needed several rounds of "make this
// less racy" fixes over the years as host speed varied. That approach isn't
// available here anyway: the shared replica set used by Go integration
// tests can't be resized below 990MB (server-enforced), and forcing genuine
// eviction against that cluster turned out to depend on WiredTiger
// checkpoint/oldest-timestamp scheduling in ways that were empirically
// unpredictable even when forcing checkpoints and using majority write
// concern.
//
// Instead, this uses the PauseUntilResumed failpoint to block mongodump
// after it has captured its real starting timestamp, overwrites that
// timestamp with one guaranteed not to exist in the oplog, and resumes.
// mongodump then runs the exact same overflow-detection logic
// (checkOplogTimestampExists, in oplog_dump.go) it would in production,
// deterministically and without needing to write any data at all.
func TestOplogRollover(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	// Oplog is not available in a standalone topology.
	testtype.SkipUnlessTestType(t, testtype.ReplSetTestType)

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = ""
	md.OutputOptions.Oplog = true
	md.OutputOptions.Out = "oplog_rollover"
	require.NoError(t, md.Init())
	defer os.RemoveAll(md.OutputOptions.Out)

	require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
	defer failpoint.DefaultManager.Reset()

	dumpErrCh := make(chan error, 1)
	go func() {
		dumpErrCh <- md.Dump()
	}()

	fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
	require.True(t, ok, "PauseUntilResumed failpoint should be enabled")
	require.NoError(t, fp.Reached(context.TODO()))
	md.oplogStart = bson.Timestamp{T: 1, I: 1}
	fp.Signal()

	err = <-dumpErrCh
	require.ErrorContains(
		t,
		err,
		"oplog overflow: mongodump was unable to capture all new oplog entries during execution",
		"mongodump should crash when the oplog rolls over during dumping",
	)
}
