ipe/api/handlers.go

// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package api

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"sort"
	"strings"

	log "github.com/golang/glog"
	"github.com/gorilla/mux"

	"ipe/events"
	"ipe/storage"
	"ipe/utils"
)

// // Maximum event size permitted 10 kB
// See: http://blogs.gnome.org/cneumair/2008/09/30/1-kb-1024-bytes-no-1-kb-1000-bytes/
const maxDataEventSize = 10 * 1000

// Prepare QueryString
func prepareQueryString(params url.Values) string {
	var keys []string

	for key := range params {
		keys = append(keys, strings.ToLower(key))
	}

	sort.Strings(keys)

	var pieces []string

	for _, key := range keys {
		pieces = append(pieces, key+"="+params.Get(key))
	}

	return strings.Join(pieces, "&")
}

// Authenticate pusher
// see: https://gist.github.com/mloughran/376898
//
// The signature is a HMAC SHA256 hex digest.
// This is generated by signing a string made up of the following components concatenated with newline characters \n.
//
//  * The uppercase request method (e.g. POST)
//  * The request path (e.g. /some/resource)
//  * The query parameters sorted by key, with keys converted to lowercase, then joined as in the query string.
//    Note that the string must not be url escaped (e.g. given the keys auth_key: foo, Name: Something else, you get auth_key=foo&name=Something else)
func Authentication(storage storage.Storage) func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		fn := func(w http.ResponseWriter, r *http.Request) {
			var (
				pathVars = mux.Vars(r)
				appID    = pathVars["app_id"]
			)

			app, err := storage.GetAppByAppID(appID)

			if err != nil {
				log.Error(err)
				http.Error(w, "Not authorized", http.StatusUnauthorized)
				return
			}

			query := r.URL.Query()

			signature := query.Get("auth_signature")
			query.Del("auth_signature")

			queryString := prepareQueryString(query)

			toSign := strings.ToUpper(r.Method) + "\n" + r.URL.Path + "\n" + queryString

			if utils.HashMAC([]byte(toSign), []byte(app.Secret)) == signature {
				next.ServeHTTP(w, r)
			} else {
				log.Error("Not authorized")
				http.Error(w, "Not authorized", http.StatusUnauthorized)
			}
		}

		return http.HandlerFunc(fn)
	}
}

// Check if the application is disabled
func CheckAppDisabled(storage storage.Storage) func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		fn := func(w http.ResponseWriter, r *http.Request) {
			var (
				pathVars = mux.Vars(r)
				appID    = pathVars["app_id"]
			)

			currentApp, err := storage.GetAppByAppID(appID)

			if err != nil {
				http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusForbidden)
				return
			}

			if !currentApp.Enabled {
				http.Error(w, "Application disabled", http.StatusForbidden)
				return
			}

			next.ServeHTTP(w, r)
		}
		return http.HandlerFunc(fn)
	}
}

type PostEvents struct{ storage storage.Storage }

func NewPostEvents(storage storage.Storage) *PostEvents {
	return &PostEvents{storage: storage}
}

// ServeHTTPC An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels.
// This is conventionally known as triggering an event.
//
// The body should contain a Hash of parameters encoded as JSON where data parameter itself is JSON encoded.
//
// Not Implemented:
// Note that these parameters may be provided in the query string, although this is discouraged.
//
// Example:
//
// {"name":"foo","channels":["project-3"],"data":"{\"some\":\"data\"}"}
//
// Response is an empty JSON hash.
//
// POST /apps/{app_id}/events
func (h *PostEvents) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var (
		pathVars = mux.Vars(r)
		appID    = pathVars["app_id"]
	)

	app, err := h.storage.GetAppByAppID(appID)

	if err != nil {
		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
	}

	var input struct {
		Name     string          `json:"name"`
		Data     json.RawMessage `json:"data"`
		Channels []string        `json:"channels,omitempty"`
		Channel  string          `json:"channel,omitempty"`
		SocketID string          `json:"socket_id,omitempty"`
	}

	err = json.NewDecoder(r.Body).Decode(&input)

	if err != nil {
		http.Error(w, "Bad request", http.StatusBadRequest)
		return
	}

	// The event data should not be larger than 10KB.
	if len(input.Data) > maxDataEventSize {
		http.Error(w, "Request too large.", http.StatusRequestEntityTooLarge)
		return
	}

	log.Info(input.Channels)
	if len(input.Channel) > 0 && len(input.Channels) == 0 {
		input.Channels = append(input.Channels, input.Channel)
	}

	for _, c := range input.Channels {
		channel := app.FindOrCreateChannelByChannelID(c)

		if err := app.Publish(channel, events.Raw{Event: input.Name, Channel: c, Data: input.Data}, input.SocketID); err != nil {
			log.Errorf("error publishing event %+v", err)
			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
		}
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	if _, err := w.Write([]byte("{}")); err != nil {
		log.Errorf("unexpected error while writing into response %+v", err)
	}
}

type GetChannels struct{ storage storage.Storage }

func NewGetChannels(storage storage.Storage) *GetChannels {
	return &GetChannels{storage: storage}
}

// Allows fetching a hash of occupied channels (optionally filtered by prefix),
// and optionally one or more attributes for each channel.
//
// Notes:
// 'user_count' is the only attribute documented on the Pusher API
//
// Example:
// {
//   "channels": {
//     "presence-foobar": {
//       user_count: 42
//     },
//     "presence-another": {
//       user_count: 123
//     }
//   }
// }
//
// GET /apps/{app_id}/channels
func (h *GetChannels) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var (
		pathVars   = mux.Vars(r)
		queryVars  = r.URL.Query()
		appID      = pathVars["app_id"]
		filter     = queryVars.Get("filter_by_prefix")
		info       = queryVars.Get("info")
		attributes = strings.Split(info, ",")
	)

	requestedUserCount := false

	for _, a := range attributes {
		if a == "user_count" {
			requestedUserCount = true
		}
	}

	// If an attribute such as user_count is requested, and the request is not limited
	// to presence channels, the API will return an error (400 code)
	if requestedUserCount && filter != "presence-" {
		http.Error(w, "Attribute user_count is restricted to presence channels", http.StatusBadRequest)
		return
	}

	app, err := h.storage.GetAppByAppID(appID)

	if err != nil {
		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
	}

	channels := make(map[string]interface{})

	switch filter {
	case "presence-":
		for _, c := range app.PresenceChannels() {
			if requestedUserCount {
				channels[c.ID] = struct {
					UserCount int `json:"user_count"`
				}{
					c.TotalUsers(),
				}
			} else {
				channels[c.ID] = struct{}{}
			}
		}
	case "public-":
		for _, c := range app.PublicChannels() {
			channels[c.ID] = struct{}{}
		}
	case "private-":
		for _, c := range app.PrivateChannels() {
			channels[c.ID] = struct{}{}
		}
	default:
		for _, c := range app.Channels() {
			channels[c.ID] = struct{}{}
		}
	}

	w.Header().Set("Content-Type", "application/json")

	js := make(map[string]interface{}, 1)
	js["channels"] = channels

	if err := json.NewEncoder(w).Encode(js); err != nil {
		log.Error(err)
		http.Error(w, "Internal server error", http.StatusInternalServerError)
	}
}

type GetChannel struct{ storage storage.Storage }

func NewGetChannel(storage storage.Storage) *GetChannel {
	return &GetChannel{storage: storage}
}

// Fetch info for one channel
//
// Example:
// {
//   occupied: true,
//   user_count: 42,
//   subscription_count: 42
// }
//
// GET /apps/{app_id}/channels/{channel_name}
func (h *GetChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var (
		pathVars    = mux.Vars(r)
		queryVars   = r.URL.Query()
		appID       = pathVars["app_id"]
		channelName = pathVars["channel_name"]
		info        = queryVars.Get("info")
		attributes  = strings.Split(info, ",")
	)

	app, err := h.storage.GetAppByAppID(appID)

	if err != nil {
		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
	}

	// Channel name could not be empty
	if strings.TrimSpace(channelName) == "" {
		http.Error(w, "Empty channel name", http.StatusBadRequest)
		return
	}

	// Attributes requested
	requestedUserCount := false
	requestedSubscriptionCount := false

	for _, a := range attributes {
		switch a {
		case "subscription_count":
			requestedSubscriptionCount = true
		case "user_count":
			requestedUserCount = true
		}
	}

	channel, err := app.FindChannelByChannelID(channelName)

	// Channel exists?
	if err != nil {
		http.Error(w, fmt.Sprintf("Could not find a channel with id %s", channelName), http.StatusBadRequest)
		return
	}

	// If an attribute such as user_count is requested, and the request is not limited
	// to presence channels, the API will return an error (400 code)
	if requestedUserCount && !channel.IsPresence() {
		http.Error(w, "Attribute user_count is restricted to presence channels", http.StatusBadRequest)
		return
	}

	// Output
	dtoChannel := struct {
		Occupied          bool `json:"occupied"`
		UserCount         int  `json:"user_count,omitempty"`
		SubscriptionCount int  `json:"subscription_count,omitempty"`
	}{Occupied: channel.IsOccupied()}

	switch {
	case requestedSubscriptionCount && requestedUserCount:
		dtoChannel.UserCount = channel.TotalUsers()
		dtoChannel.SubscriptionCount = channel.TotalSubscriptions()

	case requestedUserCount:
		dtoChannel.UserCount = channel.TotalUsers()

	case requestedSubscriptionCount:
		dtoChannel.SubscriptionCount = channel.TotalSubscriptions()
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(dtoChannel); err != nil {
		log.Error(err)
		http.Error(w, "Internal server error", http.StatusInternalServerError)
	}
}

type GetChannelUsers struct{ storage storage.Storage }

func NewGetChannelUsers(storage storage.Storage) *GetChannelUsers {
	return &GetChannelUsers{storage: storage}
}

// Allowed only for presence-channels
//
// Example:
// {
//  "users": [
//    { "id": "1" },
//    { "id": "2" }
//  ]
// }
//
// GET /apps/{app_id}/channels/{channel_name}/users
func (h *GetChannelUsers) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var (
		pathVars    = mux.Vars(r)
		appID       = pathVars["app_id"]
		channelName = pathVars["channel_name"]
	)

	isPresence := utils.IsPresenceChannel(channelName)

	if !isPresence {
		http.Error(w, "This api endpoint is restricted to presence channels.", http.StatusBadRequest)
		return
	}

	app, err := h.storage.GetAppByAppID(appID)

	if err != nil {
		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
	}

	// Get the channel
	channel, err := app.FindChannelByChannelID(channelName)

	// Channel exists?
	if err != nil {
		http.Error(w, fmt.Sprintf("Could not find a channel with id %s", channelName), http.StatusBadRequest)
		return
	}

	result := make(map[string][]interface{})

	var users []interface{}

	for _, s := range channel.Subscriptions() {
		users = append(users, struct {
			ID string `json:"id"`
		}{s.ID})
	}

	result["users"] = users

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(result); err != nil {
		http.Error(w, "Internal server error", http.StatusInternalServerError)
		log.Error(err)
	}
}