func (cv *Conversation) handleUnknownUserMessage(msg sarif.Message) {
	pl := &MsgErrNatural{
		Original: msg.Text,
	}

	cv.SendToClient(msg.Reply(sarif.CreateMessage("err/natural", pl)))
}

func (s *Service) handleEventRecord(msg sarif.Message) {
	var p recordPayload
	if err := msg.DecodePayload(&p); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	if p.Action == "" {
		s.ReplyBadRequest(msg, errors.New("No action specified"))
		return
	}

	var cfg Config
	s.cfg.Get(&cfg)
	if cfg.RecordedActions == nil {
		cfg.RecordedActions = make(map[string]bool)
	}
	if enabled := cfg.RecordedActions[p.Action]; !enabled {
		cfg.RecordedActions[p.Action] = true
		s.cfg.Set(cfg)
		s.Subscribe(p.Action, "", s.handleEventNew)
	}

	s.Log("debug", "recording action:", p.Action)
	s.Reply(msg, sarif.CreateMessage("event/recording", p))
}

func (s *Service) handleEventList(msg sarif.Message) {
	var filter map[string]interface{}
	if err := msg.DecodePayload(&filter); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	if filter == nil {
		filter = make(map[string]interface{})
	}
	reverse := false
	if len(filter) == 0 {
		filter["time >="] = time.Now().Add(-24 * time.Hour)
		reverse = true
	}

	s.Log("debug", "list by filter:", filter)
	var events []Event
	err := s.Store.Scan("events", store.Scan{
		Only:    "values",
		Filter:  filter,
		Reverse: reverse,
	}, &events)
	if err != nil {
		s.ReplyInternalError(msg, err)
	}
	s.Log("debug", "list - found", len(events))

	s.Reply(msg, sarif.CreateMessage("events/listed", &aggPayload{
		Type:   "list",
		Filter: filter,
		Events: events,
		Value:  float64(len(events)),
	}))
}

func (s *Service) HandleQueryExternal(msg sarif.Message) {
	var f Fact
	if err := msg.DecodePayload(&f); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	facts := []*Fact{&f}
	FillVariables(facts)

	var r sparql.ResourceResponse
	q := sparql.DBPedia.Query()
	q = BuildQuery(q, facts)
	if err := q.Exec(&r); err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	result := ApplyBindings(facts, r.Results.Bindings, sparql.CommonPrefixes)
	s.Reply(msg, sarif.CreateMessage("concepts/result", &resultPayload{
		ToJsonLd(result),
		result,
	}))

	for _, f := range result {
		if err := s.DB.FirstOrCreate(&f, &f).Error; err != nil {
			s.Log("err", "[reasoner] error updating external fact: "+err.Error())
		}
	}
}

func (cv *Conversation) answer(a *schema.Action, text string) (sarif.Message, bool) {
	reply := sarif.Message{
		Action: a.Reply,
		Text:   text,
	}
	if text == ".cancel" || text == "cancel" || strings.HasPrefix(text, "cancel ") {
		return reply, false
	}

	t := a.SchemaType
	if t == "ConfirmAction" || t == "DeleteAction" || t == "CancelAction" {
		ctx := &natural.Context{Text: text, ExpectedReply: "affirmative"}
		r, err := cv.service.Parse(ctx)
		if err != nil || len(r.Intents) == 0 {
			return reply, false
		}
		if r.Intents[0].Type == "neg" {
			reply.Action = a.ReplyNegative
			return reply, reply.Action != ""
		}
	}

	if a.Payload != nil {
		reply.EncodePayload(a.Payload)
	}
	return reply, true
}

func (s *Service) HandleQuery(msg sarif.Message) {
	var f Fact
	if err := msg.DecodePayload(&f); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}

	f, err := s.InterpretLiterals(f)
	if err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	var facts []*Fact
	if err := s.DB.Where(f).Limit(100).Find(&facts).Error; err != nil {
		s.ReplyInternalError(msg, err)
		return
	}
	if facts, err = s.AddLabelFacts(facts); err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	s.Reply(msg, sarif.CreateMessage("concepts/result", &resultPayload{
		ToJsonLd(facts),
		facts,
	}))
}

func (s *Service) handlePut(msg sarif.Message) {
	collection, key := parseAction("store/put/", msg.Action)
	if collection == "" {
		s.ReplyBadRequest(msg, errors.New("No collection specified."))
		return
	}

	if len(msg.Payload.Raw) == 0 && msg.Text != "" {
		v, _ := json.Marshal(msg.Text)
		msg.Payload.Raw = json.RawMessage(v)
	}
	var p interface{}
	if err := msg.DecodePayload(&p); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	// TODO: maybe a JSON payload consistency check
	doc, err := s.Store.Put(&Document{
		Collection: collection,
		Key:        key,
		Value:      msg.Payload.Raw,
	})
	if err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	doc.Value = nil
	reply := sarif.CreateMessage("store/updated/"+doc.Collection+"/"+doc.Key, doc)
	s.Reply(msg, reply)

	pub := sarif.CreateMessage("store/updated/"+doc.Collection+"/"+doc.Key, doc)
	s.Publish(pub)
}

func (s *Service) handleScan(msg sarif.Message) {
	collection, prefix := parseAction("store/scan/", msg.Action)
	if collection == "" {
		s.ReplyBadRequest(msg, errors.New("No collection specified."))
		return
	}

	var p scanMessage
	if err := msg.DecodePayload(&p); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	if p.Start == "" && p.End == "" {
		if p.Prefix == "" {
			p.Prefix = prefix
		}
	}

	got, err := s.doScan(collection, p)
	if err != nil {
		s.ReplyInternalError(msg, err)
		return
	}
	s.Reply(msg, sarif.CreateMessage("store/scanned/"+collection, got))
}

func (s *Service) handleCounter(msg sarif.Message) {
	if msg.Text == "" {
		s.ReplyBadRequest(msg, errors.New("Please specify a counter name!"))
		return
	}
	name := msg.Text
	cnt, err := s.counterGet(name)
	if err != nil {
		s.ReplyInternalError(msg, err)
		return
	}
	newCnt := cnt
	if msg.IsAction("cmd/decrement") {
		newCnt--
	} else if msg.IsAction("cmd/increment") {
		newCnt++
	}
	if newCnt != cnt {
		if err := s.counterSet(name, newCnt); err != nil {
			s.ReplyInternalError(msg, err)
			return
		}
	}
	s.Reply(msg, sarif.CreateMessage("counter/changed/"+name, &counterMessage{name, newCnt}))
}

func (app *App) Up() {
	if flag.NArg() <= 1 {
		app.Log.Fatal("Please specify an action to send to.")
	}
	action := flag.Arg(1)

	in, err := ioutil.ReadAll(os.Stdin)
	if err != nil {
		app.Log.Fatal(err)
	}

	msg := sarif.Message{
		Action: action,
	}

	pl := ContentPayload{
		Content: content.PutData(in),
	}
	if strings.HasPrefix(pl.Content.Type, "text/") {
		msg.Text = string(in)
	}
	if err := msg.EncodePayload(pl); err != nil {
		app.Log.Fatal(err)
	}

	msg, ok := <-app.Client.Request(msg)
	if !ok {
		app.Log.Fatal("No reply received.")
	}
	fmt.Println(strings.TrimSpace(msg.Text))
}

func (s *Service) handleLuaLoad(msg sarif.Message) {
	gen := false
	name := strings.TrimPrefix(strings.TrimPrefix(msg.Action, "lua/load"), "/")
	if name == "" {
		name, gen = sarif.GenerateId(), true
	}
	if _, ok := s.Machines[name]; ok {
		s.destroyMachine(name)
	}

	m, err := s.createMachine(name)
	if err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	var ctp ContentPayload
	if err := msg.DecodePayload(&ctp); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	text := msg.Text
	if ctp.Content.Url != "" {
		ct, err := content.Get(ctp.Content)
		if err != nil {
			s.ReplyBadRequest(msg, err)
		}
		text = string(ct.Data)
	}

	var gp interface{}
	msg.DecodePayload(&gp)
	out, err, _ := m.Do(text, gp)
	if err != nil {
		s.ReplyBadRequest(msg, err)
		s.destroyMachine(name)
		return
	}

	if !gen {
		f, err := os.Create(s.cfg.ScriptDir + "/" + name + ".lua")
		if err == nil {
			_, err = f.Write([]byte(text))
			defer f.Close()
		}
		if err != nil {
			s.ReplyInternalError(msg, err)
			s.destroyMachine(name)
			return
		}
	}

	s.Reply(msg, sarif.CreateMessage("lua/loaded", &MsgMachineStatus{
		name,
		"up",
		out,
	}))
}

func (cv *Conversation) HandleClientMessage(msg sarif.Message) {
	if msg.Text == ".full" || msg.Text == "/full" {
		text, err := json.MarshalIndent(cv.LastMessage, "", "    ")
		if err != nil {
			panic(err)
		}
		cv.service.Reply(msg, sarif.Message{
			Action: "natural/full",
			Text:   string(text),
		})
		return
	}

	// Check if client answers a conversation.
	if time.Now().Sub(cv.LastTime) < 5*time.Minute {
		if cv.LastMessageAction.IsAction() {
			parsed, ok := cv.answer(cv.LastMessageAction.Action, msg.Text)
			cv.LastTime = time.Time{}
			parsed.Destination = cv.LastMessage.Source
			if ok {
				cv.PublishForClient(parsed)
			}
			return
		}
	}

	// Otherwise parse message as normal request.
	ctx := &natural.Context{
		Text:      msg.Text,
		Sender:    "user",
		Recipient: "sarif",
	}
	res, err := cv.service.Parse(ctx)
	if err != nil || len(res.Intents) == 0 {
		cv.handleUnknownUserMessage(msg)
		return
	}
	pred := res.Intents[0]
	if pred.Type == "exclamatory" {
		cv.SendToClient(msg.Reply(sarif.Message{
			Action: "natural/phrase",
			Text:   cv.service.phrases.Answer(msg.Text),
		}))
		return
	}

	if pred.Message.Text == "" && pred.Type != "simple" {
		pred.Message.Text = msg.Text
	}
	cv.LastUserTime = time.Now()
	cv.LastUserText = msg.Text
	cv.LastUserMessage = pred.Message
	pred.Message.CorrId = msg.Id
	cv.PublishForClient(pred.Message)
}

func (c *WebSocketConn) Write(msg sarif.Message) error {
	if err := msg.IsValid(); err != nil {
		return err
	}
	w, err := c.conn.NextWriter(websocket.TextMessage)
	if err != nil {
		return err
	}
	defer w.Close()
	return json.NewEncoder(w).Encode(msg)
}

func (c *WebSocketConn) Read() (sarif.Message, error) {
	var msg sarif.Message
	_, r, err := c.conn.NextReader()
	if err != nil {
		return msg, err
	}
	if err := json.NewDecoder(r).Decode(&msg); err != nil {
		return msg, err
	}
	return msg, msg.IsValid()
}

func (s *Scheduler) handle(msg sarif.Message) {
	var t ScheduleMessage
	if err := msg.DecodePayload(&t); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}

	now := time.Now()
	t.Task.Time = now
	if t.Time != "" {
		t.Task.Time = futureTime(util.ParseTime(t.Time, now))
	}
	if t.RandomAfter != "" && t.RandomBefore != "" {
		after := futureTime(util.ParseTime(t.RandomAfter, t.Task.Time))
		before := futureTime(util.ParseTime(t.RandomBefore, t.Task.Time))
		if before.Before(after) {
			after, before = before, after
		}
		maxDur := int64(before.Sub(after))
		ranDur := time.Duration(rand.Int63n(maxDur))
		t.Task.Time = after.Add(ranDur)
	}
	if t.Duration != "" {
		dur, err := util.ParseDuration(t.Duration)
		if err != nil {
			s.ReplyBadRequest(msg, err)
			return
		}
		t.Task.Time = t.Task.Time.Add(dur)
	}
	if t.Task.Reply.Action == "" {
		text := msg.Text
		if text == "" {
			text = "Reminder from " + time.Now().Format(time.RFC3339) + " finished."
		}
		t.Task.Reply = sarif.Message{
			Action:      "schedule/finished",
			Destination: msg.Source,
			Text:        text,
		}
	}
	if t.Task.Reply.CorrId == "" {
		t.Reply.CorrId = msg.Id
	}
	s.Log("info", "new task:", t)

	if _, err := s.Store.Put(t.Task.Key(), &t.Task); err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	go s.recalculateTimer()
	s.Reply(msg, sarif.CreateMessage("schedule/created", t.Task))
}

func (s *Service) handleServingRecord(msg sarif.Message) {
	var sv Serving
	if err := msg.DecodePayload(&sv); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	size, name := splitSizeName(msg.Text)
	if sv.Size == 0 {
		sv.Size = size
	}
	if sv.Name == "" {
		sv.Name = name
	}
	if sv.Product == nil {
		ps, err := s.findProduct(sv.Name)
		if err != nil {
			s.ReplyBadRequest(msg, err)
			return
		}

		if len(ps) > 1 {
			pList := ""
			for _, p := range ps {
				pList += "\n- " + p.Name
			}
			s.ReplyBadRequest(msg, fmt.Errorf("%d products named %s found.%s", len(ps), sv.Name, pList))
			return
		}
		if len(ps) == 1 {
			sv.Product = &ps[0]
		}
	}
	if sv.AmountWeight <= 0 {
		sv.AmountWeight = Weight(sv.Size) * sv.Product.ServingWeight
	}
	if sv.AmountVolume <= 0 {
		sv.AmountVolume = Volume(sv.Size) * sv.Product.ServingVolume
	}
	if sv.AmountWeight <= 0 && sv.AmountVolume <= 0 {
		s.ReplyBadRequest(msg, errors.New("No serving amount specified."))
		return
	}
	if sv.Time.IsZero() {
		sv.Time = time.Now()
	}

	if err := s.DB.Save(&sv).Error; err != nil {
		s.ReplyInternalError(msg, err)
		return
	}

	s.Reply(msg, sarif.CreateMessage("meal/serving/recorded", &sv))
}

func checkErr(reply sarif.Message, ok bool) error {
	if !ok {
		return ErrNoReply
	}
	if reply.IsAction("err/notfound") {
		return ErrNotFound
	}
	if reply.IsAction("err") {
		return errors.New(reply.Text)
	}
	return nil
}

func (s *Server) clientIsAllowed(client string, msg sarif.Message) bool {
	allowed, ok := s.cfg.AllowedActions[client]
	if !ok {
		return true
	}
	for _, action := range allowed {
		if msg.IsAction(action) {
			return true
		}
	}
	return false
}

func (s *MessageSchemaStore) AddMessage(msg *sarif.Message) {
	schema := &MessageSchema{
		Action: msg.Action,
		Fields: make(map[string]string),
	}
	p := make(map[string]interface{})
	msg.DecodePayload(&p)
	for v, _ := range p {
		schema.Fields[v] = "text"
	}

	s.Add(schema)
}

func (s *Service) handleLocationUpdate(msg sarif.Message) {
	loc := Location{}
	if err := msg.DecodePayload(&loc); err != nil {
		s.ReplyBadRequest(msg, err)
		return
	}
	if loc.Time.IsZero() {
		loc.Time = time.Now()
	}
	loc.Geohash = EncodeGeohash(loc.Latitude, loc.Longitude, 12)
	s.Log("debug", "store update", loc)

	var last []Location
	err := s.Store.Scan("locations", store.Scan{
		Reverse: true,
		Only:    "values",
		Limit:   1,
	}, &last)
	if err != nil {
		s.Log("err/internal", "retrieve last err: "+err.Error())
	}
	if len(last) > 0 {
		loc.Distance = HaversineDistance(last[0], loc)
		loc.Speed = loc.Distance / loc.Time.Sub(last[0].Time).Seconds()
	}

	if _, err := s.Store.Put(loc.Key(), &loc); err != nil {
		s.ReplyInternalError(msg, err)
	}

	if changed := s.Clusters.Advance(loc); changed {
		c := s.Clusters.Current()
		status := "enter"
		if c.Status != ConfirmedCluster {
			status = "leave"
			c = s.Clusters.LastCompleted()
			s.Clusters.ClearCompleted()
		}

		// TODO: make optional
		if place, err := ReverseGeocode(c.Location); err == nil {
			c.Address = place.Pretty()
		}

		s.Publish(sarif.CreateMessage("location/cluster/"+status, c))
	}

	if len(last) > 0 {
		s.checkGeofences(last[0], loc)
	}
}