// SetGlobalEnable is a convenience function for manipulating the GlobalConfig.
//
// It's meant to be called from admin handlers on your app to turn dscache
// functionality on or off in emergencies.
func SetGlobalEnable(c context.Context, memcacheEnabled bool) error {
	// always go to the default namespace
	c, err := info.Get(c).Namespace("")
	if err != nil {
		return err
	}
	return datastore.Get(c).RunInTransaction(func(c context.Context) error {
		ds := datastore.Get(c)
		cfg := &GlobalConfig{Enable: true}
		if err := ds.Get(cfg); err != nil && err != datastore.ErrNoSuchEntity {
			return err
		}
		if cfg.Enable == memcacheEnabled {
			return nil
		}
		cfg.Enable = memcacheEnabled
		if memcacheEnabled {
			// when going false -> true, wipe memcache.
			if err := memcache.Get(c).Flush(); err != nil {
				return err
			}
		}
		return ds.Put(cfg)
	}, nil)
}

func TestRaceNonConflictingPuts(t *testing.T) {
	t.Parallel()

	ds := datastore.Get(Use(context.Background()))

	num := int32(0)

	wg := sync.WaitGroup{}

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			err := ds.RunInTransaction(func(c context.Context) error {
				ds := datastore.Get(c)
				return ds.Put(pmap(
					"$kind", "Thing", Next,
					"Value", 100))
			}, nil)
			if err != nil {
				t.Fatal("error during transaction", err)
			}
			atomic.AddInt32(&num, 1)
		}()
	}
	wg.Wait()

	if num != 100 {
		t.Fatal("expected 100 runs, got", num)
	}
}

func TestRaceGetPut(t *testing.T) {
	t.Parallel()

	value := int32(0)
	num := int32(0)

	ds := datastore.Get(Use(context.Background()))

	wg := sync.WaitGroup{}

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			err := ds.RunInTransaction(func(c context.Context) error {
				atomic.AddInt32(&num, 1)

				ds := datastore.Get(c)

				obj := pmap("$key", ds.MakeKey("Obj", 1))
				if err := ds.Get(obj); err != nil && err != datastore.ErrNoSuchEntity {
					t.Fatal("error get", err)
				}
				cur := int64(0)
				if ps, ok := obj["Value"]; ok {
					cur = ps[0].Value().(int64)
				}

				cur++
				obj["Value"] = []datastore.Property{prop(cur)}

				return ds.Put(obj)
			}, &datastore.TransactionOptions{Attempts: 200})

			if err != nil {
				t.Fatal("error during transaction", err)
			}

			atomic.AddInt32(&value, 1)
		}()
	}
	wg.Wait()

	obj := pmap("$key", ds.MakeKey("Obj", 1))
	if ds.Get(obj) != nil {
		t.FailNow()
	}
	t.Logf("Ran %d inner functions", num)
	if int64(value) != obj["Value"][0].Value().(int64) {
		t.Fatalf("value wrong value %d v %d", value, obj["Value"][0].Value().(int64))
	}
}

func registerVisitor(ctx context.Context) (int, error) {
	counter := VisitorCounter{ID: "frontend"}
	err := datastore.Get(ctx).RunInTransaction(func(c context.Context) error {
		ds := datastore.Get(c)
		err := ds.Get(&counter)
		if err != nil && err != datastore.ErrNoSuchEntity {
			return err
		}
		counter.Count++
		return ds.Put(&counter)
	}, nil)
	return counter.Count, err
}

func TestBrokenFeatures(t *testing.T) {
	t.Parallel()

	e := errors.New("default err")

	Convey("BrokenFeatures", t, func() {
		c := memory.Use(context.Background())

		Convey("Can break ds", func() {
			Convey("without a default", func() {
				c, bf := FilterRDS(c, nil)
				ds := datastore.Get(c)
				vals := []datastore.PropertyMap{{
					"$key": {datastore.MkPropertyNI(ds.NewKey("Wut", "", 1, nil))},
				}}

				Convey("by specifying an error", func() {
					bf.BreakFeatures(e, "GetMulti", "PutMulti")
					So(ds.GetMulti(vals), ShouldEqual, e)

					Convey("and you can unbreak them as well", func() {
						bf.UnbreakFeatures("GetMulti")

						So(errors.SingleError(ds.GetMulti(vals)), ShouldEqual, datastore.ErrNoSuchEntity)

						Convey("no broken features at all is a shortcut", func() {
							bf.UnbreakFeatures("PutMulti")
							So(errors.SingleError(ds.GetMulti(vals)), ShouldEqual, datastore.ErrNoSuchEntity)
						})
					})
				})

				Convey("Not specifying an error gets you a generic error", func() {
					bf.BreakFeatures(nil, "GetMulti")
					So(ds.GetMulti(vals).Error(), ShouldContainSubstring, `feature "GetMulti" is broken`)
				})
			})

			Convey("with a default", func() {
				c, bf := FilterRDS(c, e)
				ds := datastore.Get(c)
				vals := []datastore.PropertyMap{{
					"$key": {datastore.MkPropertyNI(ds.NewKey("Wut", "", 1, nil))},
				}}
				bf.BreakFeatures(nil, "GetMulti")
				So(ds.GetMulti(vals), ShouldEqual, e)
			})
		})
	})
}

func TestDatastoreKinder(t *testing.T) {
	t.Parallel()

	Convey("Datastore keys", t, func() {
		c := Use(context.Background())
		ds := dsS.Get(c)
		So(ds, ShouldNotBeNil)

		Convey("implements DSNewKeyer", func() {
			Convey("NewKey", func() {
				key := ds.NewKey("nerd", "stringID", 0, nil)
				So(key, ShouldNotBeNil)
				So(key.Kind(), ShouldEqual, "nerd")
				So(key.StringID(), ShouldEqual, "stringID")
				So(key.IntID(), ShouldEqual, 0)
				So(key.Parent(), ShouldBeNil)
				So(key.AppID(), ShouldEqual, "dev~app")
				So(key.Namespace(), ShouldEqual, "")
				So(key.String(), ShouldEqual, "/nerd,stringID")
				So(key.Incomplete(), ShouldBeFalse)
				So(key.Valid(false, "dev~app", ""), ShouldBeTrue)
			})
		})

	})
}

func TestDatastoreQueries(t *testing.T) {
	Convey("Datastore Query suport", t, func() {
		c := Use(context.Background())
		ds := dsS.Get(c)
		So(ds, ShouldNotBeNil)

		Convey("can create good queries", func() {
			q := ds.NewQuery("Foo").Filter("farnsworth >", 20).KeysOnly().Limit(10).Offset(39)

			// normally you can only get cursors from inside of the memory
			// implementation, so this construction is just for testing.
			start := queryCursor(bjoin(
				mkNum(2),
				serialize.ToBytes(dsS.IndexColumn{Property: "farnsworth"}),
				serialize.ToBytes(dsS.IndexColumn{Property: "__key__"}),
				serialize.ToBytes(prop(200)),
				serialize.ToBytes(prop(ds.NewKey("Foo", "id", 0, nil)))))

			So(start.String(), ShouldEqual,
				`gYAAZzFdTeeb3d9zOxsAAF-v221Xy32_AIGHyIgAAUc32-AFabMAAA==`)

			end := queryCursor(bjoin(
				mkNum(2),
				serialize.ToBytes(dsS.IndexColumn{Property: "farnsworth"}),
				serialize.ToBytes(dsS.IndexColumn{Property: "__key__"}),
				serialize.ToBytes(prop(3000)),
				serialize.ToBytes(prop(ds.NewKey("Foo", "zeta", 0, nil)))))

			q = q.Start(start).End(end)
			So(q, ShouldNotBeNil)
			So(q.(*queryImpl).err, ShouldBeNil)
			rq, err := q.(*queryImpl).reduce("", false)
			So(rq, ShouldNotBeNil)
			So(err, ShouldBeNil)
		})

		Convey("ensures orders make sense", func() {
			q := ds.NewQuery("Cool")
			q = q.Filter("cat =", 19).Filter("bob =", 10).Order("bob").Order("bob")

			Convey("removes dups and equality orders", func() {
				q = q.Order("wat")
				qi := q.(*queryImpl)
				So(qi.err, ShouldBeNil)
				rq, err := qi.reduce("", false)
				So(err, ShouldBeNil)
				So(rq.suffixFormat, ShouldResemble, []dsS.IndexColumn{
					{Property: "wat"}, {Property: "__key__"}})
			})

			Convey("if we equality-filter on __key__, that's just silly", func() {
				q = q.Order("wat").Filter("__key__ =", ds.NewKey("Foo", "wat", 0, nil))
				_, err := q.(*queryImpl).reduce("", false)
				So(err, ShouldErrLike, "query equality filter on __key__ is silly")
			})

		})

	})
}

func ExampleFilterRDS() {
	// Set up your context using a base service implementation (memory or prod)
	c := memory.Use(context.Background())

	// Apply the counter.FilterRDS
	c, counter := FilterRDS(c)

	// functions use ds from the context like normal... they don't need to know
	// that there are any filters at all.
	someCalledFunc := func(c context.Context) {
		ds := datastore.Get(c)
		vals := []datastore.PropertyMap{{
			"FieldName": {datastore.MkProperty(100)},
			"$key":      {datastore.MkProperty(ds.NewKey("Kind", "", 1, nil))}},
		}
		if err := ds.PutMulti(vals); err != nil {
			panic(err)
		}
	}

	// Using the other function.
	someCalledFunc(c)
	someCalledFunc(c)

	// Then we can see what happened!
	fmt.Printf("%d\n", counter.PutMulti.Successes())
	// Output:
	// 2
}

func withTxnBuf(ctx context.Context, cb func(context.Context) error, opts *datastore.TransactionOptions) error {
	inf := info.Get(ctx)
	ns := inf.GetNamespace()

	parentState, _ := ctx.Value(dsTxnBufParent).(*txnBufState)
	roots := stringset.New(0)
	rootLimit := 1
	if opts != nil && opts.XG {
		rootLimit = XGTransactionGroupLimit
	}
	sizeBudget := DefaultSizeBudget
	if parentState != nil {
		// TODO(riannucci): this is a bit wonky since it means that a child
		// transaction declaring XG=true will only get to modify 25 groups IF
		// they're same groups affected by the parent transactions. So instead of
		// respecting opts.XG for inner transactions, we just dup everything from
		// the parent transaction.
		roots = parentState.roots.Dup()
		rootLimit = parentState.rootLimit

		sizeBudget = parentState.sizeBudget - parentState.entState.total
		if sizeBudget < DefaultSizeThreshold {
			return ErrTransactionTooLarge
		}
	}

	bufDS, err := memory.NewDatastore(inf.FullyQualifiedAppID(), ns)
	if err != nil {
		return err
	}

	state := &txnBufState{
		entState:   &sizeTracker{},
		bufDS:      bufDS.Raw(),
		roots:      roots,
		rootLimit:  rootLimit,
		ns:         ns,
		aid:        inf.AppID(),
		parentDS:   datastore.Get(context.WithValue(ctx, dsTxnBufHaveLock, true)).Raw(),
		sizeBudget: sizeBudget,
	}
	if err = cb(context.WithValue(ctx, dsTxnBufParent, state)); err != nil {
		return err
	}

	// no reason to unlock this ever. At this point it's toast.
	state.Lock()

	if parentState == nil {
		return commitToReal(state)
	}

	if err = parentState.canApplyLocked(state); err != nil {
		return err
	}

	parentState.commitLocked(state)
	return nil
}

// Add adds a value to the current counter, and returns the old+new values. It
// may cause a counter to come into existance.
func (Example) Add(c context.Context, r *AddReq) (rsp *AddRsp, err error) {
	rsp = &AddRsp{}

	c = prod.Use(c)
	err = dstore.Get(c).RunInTransaction(func(c context.Context) error {
		ds := dstore.Get(c)
		ctr := &Counter{Name: r.Name}
		if err := ds.Get(ctr); err != nil && err != dstore.ErrNoSuchEntity {
			return err
		}
		rsp.Prev = ctr.Val
		ctr.Val += r.Delta
		rsp.Cur = ctr.Val
		return ds.Put(ctr)
	}, nil)
	return
}

func mkds(data []*Foo) (under, over *count.DSCounter, ds datastore.Interface) {
	c := memory.UseWithAppID(context.Background(), "something~else")
	ds = datastore.Get(c)
	_, err := ds.AllocateIDs(ds.KeyForObj(data[0]), 100)
	if err != nil {
		panic(err)
	}
	if err := ds.PutMulti(data); err != nil {
		panic(err)
	}

	c, under = count.FilterRDS(c)
	c = FilterRDS(c)
	c, over = count.FilterRDS(c)
	ds = datastore.Get(c)
	return
}

func testGetMeta(c context.Context, k *dsS.Key) int64 {
	ds := dsS.Get(c)
	mg := &MetaGroup{Parent: k.Root()}
	if err := ds.Get(mg); err != nil {
		panic(err)
	}
	return mg.Version
}

func TestRace(t *testing.T) {
	t.Parallel()

	c := FilterRDS(memory.Use(context.Background()))
	ds := datastore.Get(c)

	wg := sync.WaitGroup{}
	for i := 0; i < 100; i++ {
		id := int64(i + 1)

		wg.Add(1)
		go func() {
			defer wg.Done()
			err := ds.RunInTransaction(func(c context.Context) error {
				ds := datastore.Get(c)

				for i := 0; i < 100; i++ {
					err := ds.RunInTransaction(func(c context.Context) error {
						ds := datastore.Get(c)

						ctr := &Counter{ID: id}
						if err := ds.Get(ctr); err != nil && err != datastore.ErrNoSuchEntity {
							t.Fatal("bad Get", err)
						}
						ctr.Value++
						return ds.Put(ctr)
					}, nil)
					if err != nil {
						t.Fatal("bad inner RIT", err)
					}
				}

				return nil
			}, nil)
			if err != nil {
				t.Fatal("bad outer RIT", err)
			}
		}()
	}
	wg.Wait()
}

func TestHuge(t *testing.T) {
	t.Parallel()

	Convey("inner txn too big allows outer txn", t, func() {
		_, _, ds := mkds(dataMultiRoot)

		So(ds.RunInTransaction(func(c context.Context) error {
			ds := datastore.Get(c)

			So(18, fooSetTo(ds), hugeField)

			So(ds.RunInTransaction(func(c context.Context) error {
				ds := datastore.Get(c)
				So(ds.PutMulti(hugeData), ShouldBeNil)
				return nil
			}, nil), ShouldErrLike, ErrTransactionTooLarge)

			return nil
		}, &datastore.TransactionOptions{XG: true}), ShouldBeNil)

		So(18, fooShouldHave(ds), hugeField)
	})

	Convey("outer txn too big prevents inner txn", t, func() {
		_, _, ds := mkds(dataMultiRoot)

		So(ds.RunInTransaction(func(c context.Context) error {
			ds := datastore.Get(c)

			So(ds.PutMulti(hugeData), ShouldBeNil)

			So(ds.RunInTransaction(func(c context.Context) error {
				panic("never!")
			}, nil), ShouldErrLike, ErrTransactionTooLarge)

			return nil
		}, &datastore.TransactionOptions{XG: true}), ShouldBeNil)

		So(1, fooShouldHave(ds), hugeField)
	})
}

// GetEntityGroupVersion returns the entity group version for the entity group
// containing root. If the entity group doesn't exist, this function will return
// zero and a nil error.
func GetEntityGroupVersion(c context.Context, root dstore.Key) (int64, error) {
	ds := dstore.Get(c)
	egm := &EntityGroupMeta{Parent: dskey.Root(root)}
	err := ds.Get(egm)
	ret := egm.Version
	if err == dstore.ErrNoSuchEntity {
		// this is OK for callers. The version of the entity group is effectively 0
		// in this case.
		err = nil
	}
	return ret, err
}

func TestCompoundIndexes(t *testing.T) {
	t.Parallel()

	idxKey := func(def dsS.IndexDefinition) string {
		So(def, ShouldNotBeNil)
		return "idx::" + string(serialize.ToBytes(*def.PrepForIdxTable()))
	}

	numItms := func(c *memCollection) uint64 {
		ret, _ := c.GetTotals()
		return ret
	}

	Convey("Test Compound indexes", t, func() {
		type Model struct {
			ID int64 `gae:"$id"`

			Field1 []string
			Field2 []int64
		}

		c := Use(context.Background())
		ds := dsS.Get(c)
		t := ds.Testable().(*dsImpl)
		head := t.data.head

		So(ds.Put(&Model{1, []string{"hello", "world"}, []int64{10, 11}}), ShouldBeNil)

		idx := dsS.IndexDefinition{
			Kind: "Model",
			SortBy: []dsS.IndexColumn{
				{Property: "Field2"},
			},
		}

		coll := head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 2)

		idx.SortBy[0].Property = "Field1"
		coll = head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 2)

		idx.SortBy = append(idx.SortBy, dsS.IndexColumn{Property: "Field1"})
		So(head.GetCollection(idxKey(idx)), ShouldBeNil)

		t.AddIndexes(&idx)
		coll = head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 4)
	})
}

// CAS does an atomic compare-and-swap on a counter.
func (Example) CAS(c context.Context, r *CASReq) (err error) {
	success := false
	c = prod.Use(c)
	err = dstore.Get(c).RunInTransaction(func(c context.Context) error {
		ds := dstore.Get(c)
		ctr := &Counter{Name: r.Name}
		if err := ds.Get(ctr); err != nil {
			return err
		}
		if ctr.Val == r.OldVal {
			success = true
			ctr.Val = r.NewVal
			return ds.Put(ctr)
		}
		success = false
		return nil
	}, nil)
	if err == nil && !success {
		err = endpoints.ConflictError
	}
	return
}

// NewDatastore creates a new standalone memory implementation of the datastore,
// suitable for embedding for doing in-memory data organization.
//
// It's configured by default with the following settings:
//   * AutoIndex(true)
//   * Consistent(true)
//   * DisableSpecialEntities(true)
//
// These settings can of course be changed by using the Testable() interface.
func NewDatastore(aid, ns string) (ds.Interface, error) {
	ctx := UseWithAppID(context.Background(), aid)
	ctx, err := info.Get(ctx).Namespace(ns)
	if err != nil {
		return nil, err
	}
	ret := ds.Get(ctx)
	t := ret.Testable()
	t.AutoIndex(true)
	t.Consistent(true)
	t.DisableSpecialEntities(true)
	return ret, nil
}

// High level test for regression in how zero time is stored,
// see https://codereview.chromium.org/1334043003/
func TestDefaultTimeField(t *testing.T) {
	t.Parallel()

	Convey("Default time.Time{} can be stored", t, func() {
		type Model struct {
			ID   int64 `gae:"$id"`
			Time time.Time
		}
		ds := dsS.Get(Use(context.Background()))
		m := Model{ID: 1}
		So(ds.Put(&m), ShouldBeNil)

		// Reset to something non zero to ensure zero is fetched.
		m.Time = time.Now().UTC()
		So(ds.Get(&m), ShouldBeNil)
		So(m.Time.IsZero(), ShouldBeTrue)
	})
}

func TestGetEntityGroupVersion(t *testing.T) {
	t.Parallel()

	Convey("GetEntityGroupVersion", t, func() {
		c := memory.Use(context.Background())
		c, fb := featureBreaker.FilterRDS(c, errors.New("INTERNAL_ERROR"))
		ds := dstore.Get(c)

		pm := dstore.PropertyMap{
			"$key": {dstore.MkPropertyNI(ds.NewKey("A", "", 0, nil))},
			"Val":  {dstore.MkProperty(10)},
		}
		So(ds.Put(pm), ShouldBeNil)
		aKey, ok := pm.GetMetaDefault("key", nil).(dstore.Key)
		So(ok, ShouldBeTrue)
		So(aKey, ShouldNotBeNil)

		v, err := GetEntityGroupVersion(c, aKey)
		So(err, ShouldBeNil)
		So(v, ShouldEqual, 1)

		So(ds.Delete(aKey), ShouldBeNil)

		v, err = GetEntityGroupVersion(c, ds.NewKey("madeUp", "thing", 0, aKey))
		So(err, ShouldBeNil)
		So(v, ShouldEqual, 2)

		v, err = GetEntityGroupVersion(c, ds.NewKey("madeUp", "thing", 0, nil))
		So(err, ShouldBeNil)
		So(v, ShouldEqual, 0)

		fb.BreakFeatures(nil, "GetMulti")

		v, err = GetEntityGroupVersion(c, aKey)
		So(err.Error(), ShouldContainSubstring, "INTERNAL_ERROR")
	})
}