-
-
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
-}
-
-
-