storebus/storebus.go

package storebus

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"sync"

	"github.com/gopatchy/bus"
	"github.com/gopatchy/jsrest"
	"github.com/gopatchy/metadata"
	"github.com/gopatchy/store"
)

type StoreBus struct {
	store *store.Store
	bus   *bus.Bus

	// This lock ensures that no writes interleave with read/subscribe pairs
	orderMu sync.RWMutex

	chanMap   map[<-chan []any]<-chan any
	chanMapMu sync.Mutex
}

func NewStoreBus(dbname string) (*StoreBus, error) {
	st, err := store.NewStore(dbname)
	if err != nil {
		return nil, err
	}

	return &StoreBus{
		store:   st,
		bus:     bus.NewBus(),
		chanMap: map[<-chan []any]<-chan any{},
	}, nil
}

func (sb *StoreBus) Close() {
	sb.store.Close()
}

func (sb *StoreBus) Write(ctx context.Context, t string, obj any) error {
	sb.orderMu.Lock()
	defer sb.orderMu.Unlock()

	if err := updateHash(obj); err != nil {
		return jsrest.Errorf(jsrest.ErrInternalServerError, "hash update failed (%w)", err)
	}

	if err := sb.store.Write(ctx, t, obj); err != nil {
		return jsrest.Errorf(jsrest.ErrInternalServerError, "write failed (%w)", err)
	}

	sb.bus.Announce(t, obj)

	return nil
}

func (sb *StoreBus) Delete(ctx context.Context, t, id string) error {
	sb.orderMu.Lock()
	defer sb.orderMu.Unlock()

	if err := sb.store.Delete(ctx, t, id); err != nil {
		return jsrest.Errorf(jsrest.ErrInternalServerError, "delete failed (%w)", err)
	}

	sb.bus.Delete(t, id)

	return nil
}

func (sb *StoreBus) Read(ctx context.Context, t, id string, factory func() any) (any, error) {
	return sb.store.Read(ctx, t, id, factory)
}

func (sb *StoreBus) ReadStream(ctx context.Context, t, id string, factory func() any) (<-chan any, error) {
	sb.orderMu.RLock()
	defer sb.orderMu.RUnlock()

	initial, err := sb.store.Read(ctx, t, id, factory)
	if err != nil {
		return nil, jsrest.Errorf(jsrest.ErrInternalServerError, "read failed (%w)", err)
	}

	c := sb.bus.SubscribeKey(t, id, initial)

	return c, nil
}

func (sb *StoreBus) CloseReadStream(t, id string, c <-chan any) {
	sb.bus.UnsubscribeKey(t, id, c)
}

func (sb *StoreBus) List(ctx context.Context, t string, factory func() any) ([]any, error) {
	return sb.store.List(ctx, t, factory)
}

func (sb *StoreBus) ListStream(ctx context.Context, t string, factory func() any) (<-chan []any, error) {
	sb.orderMu.RLock()
	defer sb.orderMu.RUnlock()

	initial, err := sb.store.List(ctx, t, factory)
	if err != nil {
		return nil, jsrest.Errorf(jsrest.ErrInternalServerError, "list failed (%w)", err)
	}

	c := sb.bus.SubscribeType(t, initial)

	ret := make(chan []any, 100)

	sb.registerChan(c, ret)

	go func() {
		defer close(ret)

		for range c {
			// List() results are always at least (but not exactly) as new as the write that triggered it
			l, err := sb.store.List(ctx, t, factory)
			if err != nil {
				break
			}

			select {
			case ret <- l:
			default:
				break
			}
		}
	}()

	return ret, nil
}

func (sb *StoreBus) CloseListStream(t string, c <-chan []any) {
	sb.chanMapMu.Lock()
	defer sb.chanMapMu.Unlock()

	sb.bus.UnsubscribeType(t, sb.chanMap[c])

	delete(sb.chanMap, c)
}

func (sb *StoreBus) registerChan(in <-chan any, out <-chan []any) {
	sb.chanMapMu.Lock()
	defer sb.chanMapMu.Unlock()

	sb.chanMap[out] = in
}

func updateHash(obj any) error {
	m := *metadata.GetMetadata(obj)
	metadata.ClearMetadata(obj)

	defer metadata.SetMetadata(obj, &m)

	hash := sha256.New()
	enc := json.NewEncoder(hash)

	if err := enc.Encode(obj); err != nil {
		return jsrest.Errorf(jsrest.ErrInternalServerError, "JSON encode failed (%w)", err)
	}

	m.ETag = fmt.Sprintf("etag:%x", hash.Sum(nil))

	return nil
}