borealis.go/raw_event.go

package borealis

import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"reflect"
	"time"

	"github.com/fxamacker/cbor/v2"
	"github.com/johncgriffin/overflow"
	"github.com/segmentio/ksuid"
)

const btcEpoch = 1231006505 // Bitcoin genesis, 2009-01-03T18:15:05Z

type RawEvent struct {
	_            struct{} `cbor:",toarray" json:"-"`
	Type         uint16
	SequentialID uint64
	TimestampS   uint32
	TimestampMS  uint16
	UniqueID     []byte
	Payload      interface{}
}

func NewRawEvent(type_ uint16, payload interface{}) (*RawEvent, error) {
	now := time.Now()
	ksuid, err := ksuid.NewRandomWithTime(now)
	if err != nil {
		return nil, err
	}
	return &RawEvent{
		Type:         type_,
		SequentialID: 0,
		TimestampS:   uint32(now.Unix() - btcEpoch),
		TimestampMS:  uint16(now.UnixMilli() % 1000),
		UniqueID:     ksuid[4:],
		Payload:      payload,
	}, nil
}

func (rawEvent RawEvent) Check() (*Event, error) {
	timestampS, ok := overflow.Add64(btcEpoch, int64(rawEvent.TimestampS))
	if !ok || rawEvent.TimestampMS > 999 {
		return nil, fmt.Errorf("timestamp overflow")
	}
	timestampNS := int64(rawEvent.TimestampMS) * 1000000
	timestamp := time.Unix(timestampS, timestampNS)

	ksuid, err := ksuid.FromParts(timestamp, rawEvent.UniqueID)
	if err != nil {
		return nil, err
	}

	event := Event{
		Type:         rawEvent.Type,
		SequentialID: rawEvent.SequentialID,
		Timestamp:    timestamp,
		UniqueID:     ksuid,
		Payload:      rawEvent.Payload,
	}
	return &event, nil
}

func (event RawEvent) Equal(other RawEvent) bool {
	if event.Type != other.Type ||
		event.SequentialID != other.SequentialID ||
		event.TimestampS != other.TimestampS ||
		event.TimestampMS != other.TimestampMS ||
		!bytes.Equal(event.UniqueID, other.UniqueID) {
		return false
	}
	return reflect.DeepEqual(event.Payload, other.Payload)
}

func (event RawEvent) MarshalJSON() ([]byte, error) {
	return json.Marshal([]interface{}{
		event.Type,
		event.SequentialID,
		event.TimestampS,
		event.TimestampMS,
		event.UniqueID, // automatically Base64-encoded
		event.Payload,
	})
}

func (event RawEvent) MarshalCBOR() ([]byte, error) {
	return cbor.Marshal([]interface{}{
		event.Type,
		event.SequentialID,
		event.TimestampS,
		event.TimestampMS,
		event.UniqueID,
		event.Payload,
	})
}

func (event *RawEvent) UnmarshalJSON(input []byte) error {
	var err error
	array := []interface{}{}
	if err = json.Unmarshal(input, &array); err != nil {
		return err
	}
	if len(array) != 6 {
		return fmt.Errorf("event must be an array of length %d, but got %d", 6, len(array))
	}

	event.Type = uint16(array[0].(float64))
	event.SequentialID = uint64(array[1].(float64))
	event.TimestampS = uint32(array[2].(float64))
	event.TimestampMS = uint16(array[3].(float64))
	event.UniqueID, err = base64.StdEncoding.DecodeString(array[4].(string))
	if err != nil {
		return err
	}

	payload, ok := array[5].(map[string]interface{})
	if !ok {
		return fmt.Errorf("event payload must be a map")
	}
	event.Payload = payload

	return nil
}

func (event *RawEvent) UnmarshalCBOR(input []byte) error {
	var err error
	array := []interface{}{}
	if err = cbor.Unmarshal(input, &array); err != nil {
		return err
	}
	if len(array) != 6 {
		return fmt.Errorf("event must be an array of length %d, but got %d", 6, len(array))
	}

	event.Type = uint16(array[0].(uint64))
	event.SequentialID = uint64(array[1].(uint64))
	event.TimestampS = uint32(array[2].(uint64))
	event.TimestampMS = uint16(array[3].(uint64))
	event.UniqueID = array[4].([]byte)

	payloadIn, ok := array[5].(map[interface{}]interface{})
	if !ok {
		return fmt.Errorf("event payload must be a map")
	}
	payloadOut := make(map[string]interface{})
	for k, v := range payloadIn {
		var key string
		if key, ok = k.(string); !ok {
			return fmt.Errorf("event payload must be a map with string keys, but got the key %v", k)
		}
		payloadOut[key] = v
	}
	event.Payload = payloadOut

	return nil
}