websubhub/hub/main.go

package main // import "git.elis.nu/etu/websubhub/hub"

import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"strings"
	"sync"

	"github.com/gorilla/mux"
)

type Subscriber struct {
	URL          string
	Topic        string
	Secret       string
	LeaseSeconds string
}

type MovieResponse struct {
	Name       string  `json:"name"`
	ImdbLink   string  `json:"imdb_link"`
	ImdbRating float64 `json:"imdb_rating"`
}

var subscriptions struct {
	sync.RWMutex
	subscriptions map[string]Subscriber
}

func main() {
	router := mux.NewRouter()

	subscriptions = struct {
		sync.RWMutex
		subscriptions map[string]Subscriber
	}{subscriptions: make(map[string]Subscriber)}

	router.HandleFunc("/", subscribeHandler).Methods("POST")
	router.HandleFunc("/spam/{topic}", publishToAll)

	log.Fatal(http.ListenAndServe(":8080", router))
}

func subscribeHandler(w http.ResponseWriter, r *http.Request) {
	hubCallback, callbackErr := url.ParseRequestURI(r.FormValue("hub.callback"))
	hubMode := r.FormValue("hub.mode")
	hubTopic := r.FormValue("hub.topic")
	hubLeaseSeconds := r.FormValue("hub.lease_seconds")
	hubSecret := r.FormValue("hub.secret")

	if callbackErr != nil {
		w.WriteHeader(http.StatusBadRequest)
		w.Write([]byte("Invalid callback provided"))
		return
	}

	// Determine separator to append get parameters
	containsQ := strings.Contains(hubCallback.String(), "?")
	getSeparator := "?"
	if containsQ {
		getSeparator = "&"
	}

	// Do verification request back to the subscriber
	resp, err := http.Get(
		fmt.Sprintf(
			"%s%shub.mode=%s&hub.topic=%s&hub.challange=%s",
			hubCallback.String(),
			getSeparator,
			hubMode,
			hubTopic,
			"abc123", // TODO: This should be a randomized string
		),
	)

	// Read response body
	respText, respErr := ioutil.ReadAll(resp.Body)
	defer resp.Body.Close()

	// No error in the verification request, let's store the state of
	// this subscription change.
	if err == nil && respErr == nil && len(respText) == 0 {
		log.Printf("Created subscription for %s", hubTopic)

		// Make a subscription and store it
		subscriptions.Lock()
		subscriptions.subscriptions[hubCallback.String()] = Subscriber{
			URL:          hubCallback.String(),
			Topic:        hubTopic,
			Secret:       hubSecret,
			LeaseSeconds: hubLeaseSeconds,
		}
		subscriptions.Unlock()

		return
	}

	// Log eventual error
	log.Printf("Something is weird with this subscription: Message: '%s', Request Error: '%s', Body Parsing: '%s'", respText, err, respErr)
}

func publishToAll(w http.ResponseWriter, r *http.Request) {
	// Get variables
	vars := mux.Vars(r)

	// Get topic
	topic := vars["topic"]

	response := MovieResponse{
		Name:       "Contact",
		ImdbLink:   "https://www.imdb.com/title/tt0118884/",
		ImdbRating: 7.4,
	}

	// Claim lock, defer unlock and copy current subscriptions
	subscriptions.RLock()
	defer subscriptions.RUnlock()
	subs := subscriptions.subscriptions

	if len(subs) > 0 {
		for _, subscriber := range subs {
			// Skip this subscripber if it's the wrong topic
			if topic != subscriber.Topic {
				continue
			}

			// Encode JSON
			data, jsonErr := json.Marshal(response)

			// Log failure and return server error
			if jsonErr != nil {
				log.Fatalf("Failed to encode json: '%s'", jsonErr)
				w.WriteHeader(http.StatusInternalServerError)
				return
			}

			// Create a new hmac and write the json to it
			h := hmac.New(sha256.New, []byte(subscriber.Secret))
			h.Write(data)

			// Build request
			req, requestErr := http.NewRequest("POST", subscriber.URL, bytes.NewBuffer(data))

			// Log failure and return server error
			if requestErr != nil {
				log.Fatalf("Failed to set up http request for sending payload: '%s'", requestErr)
				w.WriteHeader(http.StatusInternalServerError)
				return
			}

			req.Header.Set("Content-Type", "application/json")
			req.Header.Set("X-Hub-Signature", "sha256="+hex.EncodeToString(h.Sum(nil)))

			// Do request
			httpResp, httpErr := http.DefaultClient.Do(req)

			if httpErr != nil {
				log.Printf("Got error when pushing data to subscriber: '%s'", httpErr)
				continue
			}

			if httpResp.StatusCode == 410 {
				// We could have triggered an unsubscribe for this subscriber if we got this
				// code, but I never got it to produce a 410
			}

			respText, respErr := ioutil.ReadAll(httpResp.Body)

			if respErr != nil {
				log.Printf("Failed to parse response text: '%s'", respErr)
			}

			log.Printf(
				"Sent message to subscriber with url '%s', got '%d' as status back with the following body: '%s'",
				subscriber.URL,
				httpResp.StatusCode,
				string(respText),
			)
		}
	}

	w.Write([]byte("I've now posted a message to all subscribers that cares about " + topic))
}