resgate notification handler

access_handler.go

package main

import (
	"encoding/json"

	"github.com/jirenius/go-res"
	log "github.com/sirupsen/logrus"
)

//Introspection is the response from the introspection endpoint
type Introspection struct {
	Active bool `json:"active"`
}

//Token back to the client
type Token struct {
	Jwt string `json:"jwt,omitempty"`

	Token string `json:"token,omitempty"`
}

// AccessHandler struct for UserHandler
type AccessHandler struct {
	jwt *JwtClient
}

// SetOption sets the res.Handler options.
func (ah *AccessHandler) SetOption(rh *res.Handler) {
	rh.Option(
		res.Access(ah.verifyJwt),
	)
}

func (ah *AccessHandler) verifyJwt(r res.AccessRequest) {

	log.Infof("<AccessHandler>.verifyJwt: verifying access %s", r.ResourceName())

	var token Token
	json.Unmarshal([]byte(r.RawToken()), &token)

	//login to jwt service using the jwt
	resp, err := ah.jwt.JwtIntrospection(token.Jwt, "loginrole", &Introspection{})
	if err != nil {
		log.Error(err)
		r.AccessDenied()
		return
	}

	if resp.Active != true {
		r.AccessDenied()
		return
	}

	addresses, err := ah.jwt.getAddressArray(token)
	if err != nil {
		log.Error(err)
		r.AccessDenied()
		return
	}

	//only the account which owns the address can receive messages
	for _, item := range addresses {
		if item == r.PathParam("address") {
			r.Access(true, "set")
			return
		}
	}

	//anyone with a valid active account can set
	r.Access(false, "set")
}

main.go

package main

import (
	"net/http"
	"time"

	"github.com/heptiolabs/healthcheck"
	res "github.com/jirenius/go-res"
	log "github.com/sirupsen/logrus"
	bolt "go.etcd.io/bbolt"
)

const (
	natsURL     = "nats://nats-client.default.svc.cluster.local:4222"
)

func main() {
	log.Info("account notification service starting...")
	//log.SetLevel(log.DebugLevel)

	health := healthcheck.NewHandler()
	jwt := NewJWTService()
	db, err := bolt.Open("notification.db", 0600, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	s := res.NewService("system")
	s.SetOwnedResources(
		[]string{"system.account.notification", "system.account.notification.>"},
		[]string{"system.account.notification", "system.account.notification.>"},
	)

	s.Handle("account.notification.$address",
		&AccessHandler{jwt},
		&NotificationHandler{db},
	)

	health.AddLivenessCheck("goroutine-threshold", healthcheck.GoroutineCountCheck(100))
	health.AddReadinessCheck("jwt-upstream-dep-dns", healthcheck.TCPDialCheck("jwt.default.svc.cluster.local:8000", 50*time.Millisecond))
	health.AddReadinessCheck("nats-upstream-dep-dns", healthcheck.TCPDialCheck("nats-client.default.svc.cluster.local:4222", 50*time.Millisecond))
	go http.ListenAndServe("0.0.0.0:8080", health)

	s.ListenAndServe(natsURL)
}

notification_handler.go

package main

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"strconv"

	"github.com/jirenius/go-res"
	bolt "go.etcd.io/bbolt"
)

// NotificationHandler for handling system notifications
type NotificationHandler struct {
	db *bolt.DB
}

// SetOption sets the res.Handler options.
func (nh *NotificationHandler) SetOption(rh *res.Handler) {
	rh.Option(
		res.GetModel(nh.get),
		res.Call("set", nh.set),
	)
}

func (nh *NotificationHandler) get(r res.ModelRequest) {

	address := r.PathParam("address")

	nh.createDB(address)

	err := nh.db.View(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte(address))
		c := b.Cursor()

		model := make(map[string]string)

		for k, v := c.First(); k != nil; k, v = c.Next() {
			model[string(k)] = string(v)
		}

		r.Model(model)

		return nil
	})

	if err != nil {
		r.InvalidQuery(err.Error())
		return
	}

}

func (nh *NotificationHandler) set(r res.CallRequest) {

	address := r.PathParam("address")
	nh.createDB(address)

	var param map[string]interface{}
	r.ParseParams(&param)

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

	err := nh.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte(address))
		id, _ := b.NextSequence()
		data, err := json.Marshal(param)
		if err != nil {
			return err
		}
		err = b.Put([]byte(strconv.Itoa(int(id))), []byte(r.RawParams()))
		event[strconv.Itoa(int(id))] = string(data)
		return err
	})

	if err != nil {
		r.InvalidQuery(err.Error())
		return
	}

	r.ChangeEvent(event)

	r.OK("done")
}

func (nh *NotificationHandler) createDB(bucket string) {
	nh.db.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists([]byte(bucket))
		if err != nil {
			return fmt.Errorf("create bucket: %s", err)
		}
		return nil
	})
}

func itob(v int) []byte {
	b := make([]byte, 8)
	binary.BigEndian.PutUint64(b, uint64(v))
	return b
}

test.js

function getNotifications() {
	client
		.get('system.account.notification.' + account_id)
		.then((resp) => {
			if (!resp) {
				console.error('there was a problem');
				return;
			}

			console.log(resp);

			resp.on('change', () => {
				console.log('Updated!', resp);
			});
		})
		.catch(showError);
}

function sendNotification() {
	let date = Date.now();
	client
		.setModel('system.account.notification.' + account_id, {
			from: 'GBIB5ILR5VGHUG2LZVHXRSBGJ7GSTK4UAEYAGENOK6FK43UN6ZP2UHML',
			to: 'GBIKPXBXOJJXG3M247IB3KFG7G2SVXGAXQY35NIF4ZYFTJT2LVLJKRJ2',
			message: 'Someone would like to contact you',
			action: 'contactrequest',
			read: false,
			responsed: false,
			completed: false,
			created_at: date,
		})
		.then((resp) => {
			
			if (!resp) {
				console.error('there was a problem');
				return;
			}

			console.log(resp);
		})
		.catch(showError);
}