// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa_test

import (
	"context"
	"fmt"
	"math"
	"sort"
	"testing"
	"time"

	pilosa "github.com/molecula/featurebase/v3"
	"github.com/molecula/featurebase/v3/disco"
	"github.com/molecula/featurebase/v3/test"
	"github.com/stretchr/testify/require"
)

func TestExecutor_DeleteRecords(t *testing.T) {
	c := test.MustRunCluster(t, 1)
	indexName := c.Idx()
	defer c.Close()
	setup := func(t *testing.T, r *require.Assertions, c *test.Cluster) {
		t.Helper()
		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "setfield")
		c.ImportBits(t, indexName, "setfield", [][2]uint64{
			{0, 0},
			{0, 1},
			{0, ShardWidth + 2},
			{10, 2},
			{10, ShardWidth},
			{10, 2 * ShardWidth},
			{10, ShardWidth + 1},
			{20, ShardWidth},
		})
		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "bsi", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
		c.ImportIntID(t, indexName, "bsi", []test.IntID{
			{ID: 0, Val: 4},
			{ID: 2, Val: 8},
		})
		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "timefield", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0"))
		c.ImportBitsWithTimestamp(t, indexName, "timefield", [][2]uint64{
			{0, 0},
			{0, 1},
			{0, 1},
			{0, 1},
			{0, 1},
		}, []int64{
			time.Date(2020, time.January, 2, 15, 45, 0, 0, time.UTC).Unix(),
			time.Date(2019, time.January, 2, 16, 45, 0, 0, time.UTC).Unix(),
			time.Date(2019, time.January, 2, 16, 45, 0, 0, time.UTC).Unix(),
			time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix(),
			time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix(),
		})

	}
	setupBig := func(t *testing.T, r *require.Assertions, c *test.Cluster, Rows uint64) {
		t.Helper()
		fieldName := "setfield"
		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, fieldName)
		// we don't need to populate the whole thing, just enough to get a sample of it
		width := uint64(ShardWidth / 4)
		rows := make([][2]uint64, width*Rows)
		n := 0
		// populate rows with decreasing density
		for columnID := uint64(0); columnID < width; columnID++ {
			for rowID := uint64(0); rowID < Rows; rowID++ {
				if (columnID % (rowID + 1)) == 0 {
					rows[n] = [2]uint64{rowID, columnID}
					n++
				}
			}
		}
		c.ImportBits(t, indexName, "setfield", rows[:n])
	}

	setupKeys := func(t *testing.T, r *require.Assertions, c *test.Cluster) {
		t.Helper()
		c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "timefield", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0"))
		c.ImportTimeQuantumKey(t, indexName, "timefield", []test.TimeQuantumKey{
			{RowKey: "fish", ColKey: "one", Ts: time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix()},
			{RowKey: "fish", ColKey: "one", Ts: time.Date(2020, time.January, 2, 17, 45, 0, 0, time.UTC).Unix()},
			{RowKey: "fish", ColKey: "two", Ts: time.Date(2019, time.January, 3, 17, 45, 0, 0, time.UTC).Unix()},
		})
		c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "keystuff")
		c.ImportIDKey(t, indexName, "keystuff", []test.KeyID{
			{ID: 1, Key: "A"},
			{ID: 2, Key: "B"},
			{ID: 3, Key: "C"},
			{ID: 4, Key: "D"},
		})

	}
	setupOverlap := func(t *testing.T, r *require.Assertions, c *test.Cluster) {
		t.Helper()
		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "setfield")
		c.ImportBits(t, indexName, "setfield", [][2]uint64{
			{0, 0},
			{0, 1},
			{1, 1},
			{2, 1},
			{3, 1},
			{0, ShardWidth},
			{2, ShardWidth},
			{4, ShardWidth},
			{6, ShardWidth},
		})
	}
	tearDown := func(t *testing.T, require *require.Assertions, c *test.Cluster) {
		t.Helper()
		api := c.GetPrimary().API
		err := api.DeleteIndex(context.Background(), indexName)
		require.NoErrorf(err, "DeleteIndex %v", indexName)
	}
	require := require.New(t)
	t.Run("DeleteRecords", func(t *testing.T) {
		t.Run("DeleteRace", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			for i := 0; i < 100; i++ {
				// imagine that we start with bsi set to 3 in column 0.
				// We then execute two operations:
				// (1) we set bsi to 1 in column 0
				// (2) we delete everything with bsi > 2
				// No matter which order these happen in, we should see
				// bsi set to 1 in column 0.
				// If the set happens first, the delete doesn't touch it.
				// If the set happens second, the delete deletes the previous
				// value, then the set happens.
				// Let's find out...
				c.ImportIntID(t, indexName, "bsi", []test.IntID{
					{ID: 0, Val: 3},
				})
				ch := make(chan error)
				go func() {
					// Make sure we close the channel even if we're failing out of the test.
					defer close(ch)
					// We don't actually care whether the delete succeeds or fails...
					_ = c.Query(t, indexName, `Delete(Row(bsi>2))`)
					// we don't try to handle an error from that, at this time.
				}()
				c.ImportIntID(t, indexName, "bsi", []test.IntID{
					{ID: 0, Val: 1},
				})
				// wait for the async delete
				<-ch
				resp := c.Query(t, indexName, `Row(bsi<2)`)
				// we expect to always find 0 in this row
				row, ok := resp.Results[0].(*pilosa.Row)
				if !ok {
					t.Fatalf("expected row return")
				}
				cols := row.Columns()
				if len(cols) < 1 || cols[0] != 0 {
					t.Fatalf("expected columns including 0, got %d on try %d", cols, i)
				}
			}
		})
		t.Run("Delete", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Extract(All())`)
			m := resp.Results[0].(pilosa.ExtractedTable)
			before := convert(m.Columns)
			require.Equal([]uint64{0, 1, 2, ShardWidth, ShardWidth + 1, ShardWidth + 2, 2 * ShardWidth}, before, "these records are expected")
			resp = c.Query(t, indexName, fmt.Sprintf(`Delete(ConstRow(columns=[1,2,3,%v]))`, ShardWidth+1))
			require.NotNil(resp, "Response should not be nil")
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Extract(All())`)

			//Note none of the removed records should remain
			m = resp.Results[0].(pilosa.ExtractedTable)
			after := convert(m.Columns)
			require.Equal([]uint64{0, ShardWidth, ShardWidth + 2, 2 * ShardWidth}, after, "these records should be remaining")
		})
		t.Run("DeleteKey", func(t *testing.T) {
			setupKeys(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Extract(All())`)
			m := resp.Results[0].(pilosa.ExtractedTable)
			before := convertKey(m.Columns)
			sort.Strings(before)
			expected := []string{"A", "B", "C", "D", "one", "two"}
			require.Equal(expected, before, "these keyed records before")
			resp = c.Query(t, indexName, `Delete(ConstRow(columns=["A","one"]))`)
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Extract(All())`)
			m = resp.Results[0].(pilosa.ExtractedTable)
			after := convertKey(m.Columns)
			sort.Strings(after)
			require.Equal([]string{"B", "C", "D", "two"}, after, "these keyed records after delete")
			//validate that column keys got deleted
			node := c.GetNode(0)
			keys := []string{"A", "one"}
			res, err := node.API.FindIndexKeys(context.Background(), indexName, keys...)
			require.Nil(err)
			require.Empty(res)
		})
		t.Run("Delete Row", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Delete(Row(setfield=20))`)
			require.NotNil(resp, "Response should not be nil")
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Extract(All())`)

			//Note none of the removed records should remain
			m := resp.Results[0].(pilosa.ExtractedTable)
			after := convert(m.Columns)
			require.Equal([]uint64{0, 1, 2, ShardWidth + 1, ShardWidth + 2, 2 * ShardWidth}, after, "these records are expected")
		})
		t.Run("Delete Not Row", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Delete(Not(Row(setfield=20)))`)
			require.NotNil(resp, "Response should not be nil")
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Extract(All())`)

			//Note none of the removed records should remain
			m := resp.Results[0].(pilosa.ExtractedTable)
			after := convert(m.Columns)
			require.Equal([]uint64{ShardWidth}, after, "these records are expected")
		})
		t.Run("Delete All", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Count(All())`)
			//Note none of the removed records should remain
			before := resp.Results[0].(uint64)
			require.Equal(uint64(7), before, "these records are expected")

			resp = c.Query(t, indexName, `Delete(All())`)
			require.NotNil(resp, "Response should not be nil")
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Count(All())`)
			//Note none of the removed records should remain
			after := resp.Results[0].(uint64)
			require.Equal(uint64(0), after, "these records are expected")
		})
		t.Run("DeleteOverlap", func(t *testing.T) {
			setupOverlap(t, require, c)
			defer tearDown(t, require, c)
			resp := c.Query(t, indexName, `Extract(All())`)
			m := resp.Results[0].(pilosa.ExtractedTable)
			before := convert(m.Columns)
			require.Equal([]uint64{0, 1, ShardWidth}, before, "these records are expected")
			resp = c.Query(t, indexName, fmt.Sprintf(`Delete(ConstRow(columns=[%v]))`, ShardWidth))
			require.NotNil(resp, "Response should not be nil")
			require.NotEmpty(resp.Results)
			require.Equal(true, resp.Results[0], "Change should have happened")

			resp = c.Query(t, indexName, `Extract(All())`)

			//Note none of the removed records should remain
			m = resp.Results[0].(pilosa.ExtractedTable)
			after := convert(m.Columns)
			require.Equal([]uint64{0, 1}, after, "these records should be remaining")
		})

		// FB-1281: Delete() calls with an invalid bitmap filter would cause a panic.
		// This test validates that the error is correctly propagated to the caller.
		t.Run("DeleteWithBitmapError", func(t *testing.T) {
			setup(t, require, c)
			defer tearDown(t, require, c)
			_, err := c.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{Index: indexName, Query: `Delete(Row(setfield == 1))`})
			if err == nil || err.Error() != `executing: executeDelete: mapping on primary node: bsigroup not found` {
				t.Fatalf("unexpected error: %s", err)
			}
		})
	})
	t.Run("DeleteRecordsBigWithRestart", func(t *testing.T) {
		// restarting doesn't work correctly for a shared cluster
		c := test.MustRunUnsharedCluster(t, 1)
		defer c.Close()
		setupBig(t, require, c, 16)
		defer tearDown(t, require, c)
		node := c.GetNode(0)
		resp := c.Query(t, indexName, `Delete(Row(setfield=12))`)
		require.NotNil(resp, "Response should not be nil")
		require.NotEmpty(resp.Results)
		require.Equal(true, resp.Results[0], "Change should have happened")
		resp = c.Query(t, indexName, `Count(Row(setfield=12))`)
		require.NotNil(resp, "Response should not be nil")
		require.NotEmpty(resp.Results)
		require.Equal(uint64(0), resp.Results[0], "Should have removed")
		err := node.Reopen()
		require.NoError(err, "restart cluster DeleteRecordsBig")
		err = c.AwaitState(disco.ClusterStateNormal, 10*time.Second)
		require.NoError(err, "backToNormal")
	})
}

func convert(before []pilosa.ExtractedTableColumn) []uint64 {
	result := make([]uint64, 0)
	for _, i := range before {
		result = append(result, i.Column.ID)
	}
	return result
}
func convertKey(before []pilosa.ExtractedTableColumn) []string {
	result := make([]string, 0)
	for _, i := range before {
		result = append(result, i.Column.Key)
	}
	return result
}