// pingDevicesAsync schedules len(endpoints) tasks of /ping-device.
// d specifies the duration the tasker must wait before executing the task.
// If scheduling fails for some endpoints, those will be in the returned values
// along with a non-nil error.
func pingDevicesAsync(c context.Context, uid string, endpoints []string, d time.Duration) ([]string, error) {
	if len(endpoints) == 0 {
		return nil, nil
	}
	p := path.Join(config.Prefix, "/task/ping-device")
	jobs := make([]*taskqueue.Task, 0, len(endpoints))
	for _, endpoint := range endpoints {
		t := taskqueue.NewPOSTTask(p, url.Values{
			"uid":      {uid},
			"endpoint": {endpoint},
		})
		t.Delay = d
		jobs = append(jobs, t)
	}

	_, err := taskqueue.AddMulti(c, jobs, "")
	merr, mok := err.(appengine.MultiError)
	if !mok {
		return nil, err
	}

	errEndpoints := make([]string, 0)
	for i, e := range merr {
		if e == nil {
			continue
		}
		errEndpoints = append(errEndpoints, endpoints[i])
	}
	if len(errEndpoints) == 0 {
		return nil, nil
	}
	return errEndpoints, fmt.Errorf("pingDevicesAsync: %v", err)
}

// Grab all users, and enqueue them for batch processing
func upgradeHandler(w http.ResponseWriter, r *http.Request) {
	ctx := req2ctx(r)
	cdb := complaintdb.NewDB(ctx)

	var cps = []types.ComplainerProfile{}
	cps, err := cdb.GetAllProfiles()
	if err != nil {
		cdb.Errorf("upgradeHandler: getallprofiles: %v", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	for _, cp := range cps {
		t := taskqueue.NewPOSTTask("/backend/cdb-batch-user", map[string][]string{
			"email": {cp.EmailAddress},
		})
		if _, err := taskqueue.Add(cdb.Ctx(), t, "batch"); err != nil {
			cdb.Errorf("upgradeHandler: enqueue: %v", err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
	}
	cdb.Infof("enqueued %d batch", len(cps))
	w.Write([]byte(fmt.Sprintf("OK, enqueued %d", len(cps))))
}

func TestTaskQueue(t *testing.T) {
	// Only run the test if APPENGINE_DEV_APPSERVER is explicitly set.
	if os.Getenv("APPENGINE_DEV_APPSERVER") == "" {
		t.Skip("APPENGINE_DEV_APPSERVER not set")
	}
	queueNames := []string{
		"taskQueueName",
	}
	ctx, done, err := NewContextOptions(&Options{
		TaskQueues: queueNames,
	})
	queueNames = append(queueNames, "default")
	if err != nil {
		t.Fatalf("NewContext: %v", err)
	}
	defer done()

	for _, queueName := range queueNames {
		_, err = taskqueue.Add(ctx, taskqueue.NewPOSTTask("/worker", url.Values{
			"key": {"value"},
		}), queueName)
		if err != nil {
			t.Errorf("Unable to add task to queue - %v", err)
		}
		if stats, err := taskqueue.QueueStats(ctx, []string{queueName}); err != nil {
			t.Errorf("Unable to fetch queue statistics - %v", err)
		} else if len(stats) == 0 {
			t.Errorf("No stats found for the default taskqueue!")
		} else if stats[0].Tasks != 1 {
			t.Errorf("Wrong number of tasks found in queue, wanted 1, got %d", stats[0].Tasks)
		}
	}
}

// Writes them all into a batch queue
func publishAllComplaintsHandler(w http.ResponseWriter, r *http.Request) {
	ctx := appengine.NewContext(r)
	str := ""

	s, e, _ := widget.FormValueDateRange(r)
	days := date.IntermediateMidnights(s.Add(-1*time.Second), e) // decrement start, to include it
	url := "/backend/publish-complaints"

	for i, day := range days {
		dayStr := day.Format("2006.01.02")

		thisUrl := fmt.Sprintf("%s?datestring=%s", url, dayStr)
		if r.FormValue("skipload") != "" {
			thisUrl += "&skipload=" + r.FormValue("skipload")
		}

		t := taskqueue.NewPOSTTask(thisUrl, map[string][]string{})
		// Give ourselves time to get all these tasks posted, and stagger them out a bit
		t.Delay = time.Minute + time.Duration(i)*15*time.Second

		if _, err := taskqueue.Add(ctx, t, "batch"); err != nil {
			log.Errorf(ctx, "publishAllComplaintsHandler: enqueue: %v", err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		str += " * posting for " + thisUrl + "\n"
	}

	w.Header().Set("Content-Type", "text/plain")
	w.Write([]byte(fmt.Sprintf("OK, enqueued %d\n--\n%s", len(days), str)))
}

func notifyShardAsync(c context.Context, shard, changes string, all bool) error {
	p := path.Join(config.Prefix, "/task/notify-shard")
	t := taskqueue.NewPOSTTask(p, url.Values{
		"shard":   {shard},
		"changes": {changes},
		"all":     {fmt.Sprintf("%v", all)},
	})
	_, err := taskqueue.Add(c, t, "")
	return err
}

// pingUserAsync creates an async job to send a push notification to user devices.
// sessions are session IDs used to compare against user bookmarks.
// TODO: add ioext support
func pingUserAsync(c context.Context, uid string, sessions []string, all bool) error {
	p := path.Join(config.Prefix, "/task/ping-user")
	t := taskqueue.NewPOSTTask(p, url.Values{
		"uid":      {uid},
		"sessions": {strings.Join(sessions, " ")},
		"all":      {fmt.Sprintf("%v", all)},
	})
	_, err := taskqueue.Add(c, t, "")
	return err
}

// pingExtPartyAsync notifies extra parties at config.ExtPingURL about data updates.
func pingExtPartyAsync(c context.Context, key string) error {
	if key == "" || config.ExtPingURL == "" {
		return nil
	}
	p := path.Join(config.Prefix, "/task/ping-ext")
	t := taskqueue.NewPOSTTask(p, url.Values{
		"key": {key},
	})
	_, err := taskqueue.Add(c, t, "")
	return err
}

// This enqueues tasks for each individual day, or flight
func batchFlightScanHandler(w http.ResponseWriter, r *http.Request) {
	c := appengine.NewContext(r)

	tags := []string{} //"ADSB"} // Maybe make this configurable ...

	n := 0
	str := ""
	s, e, _ := widget.FormValueDateRange(r)
	job := r.FormValue("job")
	if job == "" {
		http.Error(w, "Missing argument: &job=foo", http.StatusInternalServerError)
	}

	days := date.IntermediateMidnights(s.Add(-1*time.Second), e) // decrement start, to include it
	for _, day := range days {
		// Get the keys for all the flights on this day.
		fdb := oldfgae.FlightDB{C: oldappengine.NewContext(r)}

		dStart, dEnd := date.WindowForTime(day)
		dEnd = dEnd.Add(-1 * time.Second)
		keys, err := fdb.KeysInTimeRangeByTags(tags, dStart, dEnd)
		if err != nil {
			log.Errorf(c, "upgradeHandler: enqueue: %v", err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		singleFlightUrl := "/backend/fdb-batch/flight"
		for _, key := range keys {
			str += fmt.Sprintf("Enqueing day=%s: %s?job=%s&key=%s\n",
				day.Format("2006.01.02"), singleFlightUrl, job, key.Encode())

			t := taskqueue.NewPOSTTask(singleFlightUrl, map[string][]string{
				"date": {day.Format("2006.01.02")},
				"key":  {key.Encode()},
				"job":  {job},
			})

			if _, err := taskqueue.Add(c, t, "batch"); err != nil {
				log.Errorf(c, "upgradeHandler: enqueue: %v", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}

			n++
		}
	}

	log.Infof(c, "enqueued %d batch items for '%s'", n, job)

	w.Header().Set("Content-Type", "text/plain")
	w.Write([]byte(fmt.Sprintf("OK, batch, enqueued %d tasks for %s\n%s", n, job, str)))
}

// Dequeue a single day, and enqueue a job for each flight on that day
func batchFlightDayHandler(w http.ResponseWriter, r *http.Request) {
	ctx := req2ctx(r)

	tags := []string{} //"ADSB"} // Maybe make this configurable ...

	n := 0
	str := ""
	job := r.FormValue("job")
	if job == "" {
		http.Error(w, "Missing argument: &job=foo", http.StatusInternalServerError)
	}

	day := date.ArbitraryDatestring2MidnightPdt(r.FormValue("day"), "2006/01/02")

	fdb := oldfgae.NewDB(r)

	dStart, dEnd := date.WindowForTime(day)
	dEnd = dEnd.Add(-1 * time.Second)
	keys, err := fdb.KeysInTimeRangeByTags(tags, dStart, dEnd)
	if err != nil {
		log.Errorf(ctx, "upgradeHandler: enqueue: %v", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	singleFlightUrl := "/backend/fdb-batch/flight"
	for _, key := range keys {
		str += fmt.Sprintf("Enqueing day=%s: %s?job=%s&key=%s\n",
			day.Format("2006.01.02"), singleFlightUrl, job, key.Encode())

		if r.FormValue("dryrun") == "" {
			t := taskqueue.NewPOSTTask(singleFlightUrl, map[string][]string{
				// "date": {day.Format("2006.01.02")},
				"key": {key.Encode()},
				"job": {job},
			})

			if _, err := taskqueue.Add(ctx, t, "batch"); err != nil {
				log.Errorf(ctx, "upgradeHandler: enqueue: %v", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
		}

		n++
	}

	log.Infof(ctx, "enqueued %d batch items for '%s'", n, job)

	w.Header().Set("Content-Type", "text/plain")
	w.Write([]byte(fmt.Sprintf("OK, batch, enqueued %d tasks for %s\n%s", n, job, str)))
}

// PushTask to push a task into the queue.
func (queue *PushQueue) PushTask(req *wcg.Request, urlPath string, form url.Values) error {
	var queueName string
	if req.IsTest() || lib.IsOnLocalGAE() {
		queueName = "default"
	} else {
		queueName = queue.Name
	}
	if _, err := taskqueue.Add(gae.NewContext(req), taskqueue.NewPOSTTask(urlPath, form), queueName); err != nil {
		req.Logger.Errorf("[Queue] Error adding a task (%s) into the queue (%q): %v", urlPath, queueName, err)
		return err
	}
	return nil
}

// notifySubscriberAsync creates an async job to begin notify subscribers.
func notifySubscribersAsync(c context.Context, d *dataChanges, all bool) error {
	changes, err := json.Marshal(d)
	if err != nil {
		return err
	}
	p := path.Join(config.Prefix, "/task/notify-subscribers")
	t := taskqueue.NewPOSTTask(p, url.Values{
		"changes": {string(changes)},
		"all":     {fmt.Sprintf("%v", all)},
	})
	_, err = taskqueue.Add(c, t, "")
	return err
}

func notifyUserAsync(c context.Context, uid, shard string, m *pushMessage) error {
	p := path.Join(config.Prefix, "/task/notify-user")
	msg, err := json.Marshal(m)
	if err != nil {
		return err
	}
	t := taskqueue.NewPOSTTask(p, url.Values{
		"uid":     {uid},
		"shard":   {shard},
		"message": {string(msg)},
	})
	_, err = taskqueue.Add(c, t, "")
	return err
}

// notifySubscriberAsync creates an async job to begin notify subscribers.
func notifySubscribersAsync(c context.Context, d *dataChanges, all bool) error {
	skeys := make([]string, 0, len(d.Sessions))
	for id, _ := range d.Sessions {
		skeys = append(skeys, id)
	}
	p := path.Join(config.Prefix, "/task/notify-subscribers")
	// TODO: add ioext to the payload
	t := taskqueue.NewPOSTTask(p, url.Values{
		"sessions": {strings.Join(skeys, " ")},
		"all":      {fmt.Sprintf("%v", all)},
	})
	_, err := taskqueue.Add(c, t, "")
	return err
}

func CallMinecraftTQ(c context.Context, minecraftKey *datastore.Key, operationID string) (*taskqueue.Task, error) {
	log.Infof(c, "Call Minecraft TQ, key = %v, operationID = %s", minecraftKey, operationID)
	if minecraftKey == nil {
		return nil, errors.New("key is required")
	}
	if len(operationID) < 1 {
		return nil, errors.New("operationID is required")
	}

	t := taskqueue.NewPOSTTask("/tq/1/minecraft", url.Values{
		"keyStr":      {minecraftKey.Encode()},
		"operationID": {operationID},
	})
	t.Delay = time.Second * 30
	return taskqueue.Add(c, t, "minecraft")
}

func (a *ServerTQApi) CallDeleteInstance(c context.Context, minecraftKey *datastore.Key, operationID string, latestSnapshot string) (*taskqueue.Task, error) {
	log.Infof(c, "Call Minecraft TQ, key = %v, operationID = %s", minecraftKey, operationID)
	if minecraftKey == nil {
		return nil, errors.New("key is required")
	}
	if len(operationID) < 1 {
		return nil, errors.New("operationID is required")
	}

	t := taskqueue.NewPOSTTask("/tq/1/server/instance/delete", url.Values{
		"keyStr":         {minecraftKey.Encode()},
		"operationID":    {operationID},
		"latestSnapshot": {latestSnapshot},
	})
	t.Delay = time.Second * 30
	return taskqueue.Add(c, t, "minecraft")
}

func callbackHandler(w http.ResponseWriter, r *http.Request) {

	ctx := appengine.NewContext(r)

	chi, err := strconv.ParseInt(os.Getenv("CHANNEL_ID"), 10, 64)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	chs := os.Getenv("CHANNEL_SECRET")
	mid := os.Getenv("MID")

	bot, err := linebot.NewClient(chi, chs, mid)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	received, err := bot.ParseRequest(r)
	if err != nil {
		if err == linebot.ErrInvalidSignature {
			w.WriteHeader(http.StatusBadRequest)
		} else {
			w.WriteHeader(http.StatusInternalServerError)
		}
		return
	}
	for _, result := range received.Results {
		content := result.Content()
		if content != nil && content.IsMessage && content.ContentType == linebot.ContentTypeText {
			text, _ := content.TextContent()
			log.Debugf(ctx, "id: %s, text: %s, from: %s, to: %v", content.ID, text.Text, text.From, text.To)

			values := url.Values{}
			values.Set("to", text.From)
			values.Set("text", text.Text)
			t := taskqueue.NewPOSTTask(TEXT_URI, values)
			_, err = taskqueue.Add(ctx, t, TEXT_MESSAGE_QUEUE)
			if err != nil {
				w.WriteHeader(http.StatusInternalServerError)
				return
			}
		}
	}
	w.WriteHeader(http.StatusOK)
}

func f(ctx context.Context) {
	err := datastore.RunInTransaction(ctx, func(ctx context.Context) error {
		t := taskqueue.NewPOSTTask("/worker", url.Values{
		// ...
		})
		// Use the transaction's context when invoking taskqueue.Add.
		_, err := taskqueue.Add(ctx, t, "")
		if err != nil {
			// Handle error
		}
		// ...
		return nil
	}, nil)
	if err != nil {
		// Handle error
	}
	// ...
}

// AddToSearchIndex adds an Item to the search index.
// To current implementation uses task queues so this operation will
// be executed in the background
func AddToSearchIndex(con *data.Context, i data.Item) {
	// We'll update the search index next
	// FIRST: Store the HTML of the item in the memcache.
	//        We do that because it is often larger than the maximum
	//        task size allowed at the GAE.
	memI := &memcache.Item{
		Key:   i.DSKey,
		Value: []byte(i.HTMLforSearch),
	}
	if err := memcache.Set(con.C, memI); err != nil {
		con.Log.Infof("Error while storing the search HTML in the memcache for URL %v", i.URL)
	}

	// SECOND: Put the search index update task in the queue
	task := taskqueue.NewPOSTTask("/t/search/add_to_index", itemToSearchIndexTask(i))
	if _, err := taskqueue.Add(con.C, task, "search-index"); err != nil {
		con.Log.Errorf("Error while triggering the add to index: %v", err)
	}
}

// Enqueues one 'day' task per day in the range
func batchFlightDateRangeHandler(w http.ResponseWriter, r *http.Request) {
	ctx := req2ctx(r)

	n := 0
	str := ""
	s, e, _ := widget.FormValueDateRange(r)
	job := r.FormValue("job")
	if job == "" {
		http.Error(w, "Missing argument: &job=foo", http.StatusInternalServerError)
		return
	}

	str += fmt.Sprintf("** s: %s\n** e: %s\n", s, e)

	days := date.IntermediateMidnights(s.Add(-1*time.Second), e) // decrement start, to include it
	for _, day := range days {

		dayUrl := "/backend/fdb-batch/day"
		dayStr := day.Format("2006/01/02")

		str += fmt.Sprintf(" * adding %s, %s via %s\n", job, dayStr, dayUrl)

		if r.FormValue("dryrun") == "" {
			t := taskqueue.NewPOSTTask(dayUrl, map[string][]string{
				"day": {dayStr},
				"job": {job},
			})

			if _, err := taskqueue.Add(ctx, t, "batch"); err != nil {
				log.Errorf(ctx, "upgradeHandler: enqueue: %v", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
		}

		n++
	}

	log.Infof(ctx, "enqueued %d batch items for '%s'", n, job)

	w.Header().Set("Content-Type", "text/plain")
	w.Write([]byte(fmt.Sprintf("OK, batch, enqueued %d tasks for %s\n%s", n, job, str)))
}

func handler(w http.ResponseWriter, r *http.Request) {
	ctx := appengine.NewContext(r)
	if name := r.FormValue("name"); name != "" {
		t := taskqueue.NewPOSTTask("/worker", map[string][]string{"name": {name}})
		if _, err := taskqueue.Add(ctx, t, ""); err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
	}
	q := datastore.NewQuery("Counter")
	var counters []Counter
	if _, err := q.GetAll(ctx, &counters); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if err := handlerTemplate.Execute(w, counters); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	// OK
}