feat: Listener outbox + Lots of fixes

Author: Unbreathable

listener_creation.go

@@ -5,20 +5,20 @@ import (
 	"sync"
 
 	"github.com/Liphium/neogate"
+	"github.com/bytedance/sonic"
 	"github.com/dgraph-io/ristretto/v2"
 )
 
-type DatabaseListenerCreate[C Change[C]] struct {
+type DatabaseListenerCreate[DB any, C Change[C]] struct {
 	Identifier string                                           // Identifier for the listener (REQUIRED)
-	Get        func([]string) (map[string]C, error)             // Get the base data from results of listeners or just with key (required)
+	Get        func(DB, []string) (map[string]C, error)         // Get the base data from results of listeners or just with key (required)
 	ToEvent    func(key string, change Change[C]) neogate.Event // Should convert string and change info into an event that can be sent with Neo (required)
-	FromEvent  func(key string, encodedEvent []byte) Change[C]  // Should convert an event back to a change and its key
 
 	PoolConfig PoolConfig // Config for the pooling of subscription workers
 }
 
 // Helper function for initializing a new listener dictionary properly
-func NewListenerDictionary[T any, PS IPubSubBackend, DB any, C Change[C]](instance *Instance[T, PS], outbox *PubSubOutbox[DB, PS], create DatabaseListenerCreate[C]) *DatabaseListenerDictionary[T, PS, DB, C] {
+func NewListenerDictionary[T any, PS IPubSubBackend, DB any, C Change[C]](instance *Instance[T, PS], outbox *PubSubOutbox[DB, PS], create DatabaseListenerCreate[DB, C]) *DatabaseListenerDictionary[T, PS, DB, C] {
 	subDict, err := ristretto.NewCache(&ristretto.Config[string, *ListenerSubscriptions[T, PS, C]]{
 		MaxCost:     10_000,      // Maximum 10.000 stored items
 		NumCounters: 10_000 * 10, // 10x what we want to store
@@ -42,9 +42,12 @@ func NewListenerDictionary[T any, PS IPubSubBackend, DB any, C Change[C]](instan
 
 	// Create the pool to forward messages to the dictionary
 	dictionary.pool.OnMessage(func(channel, message string) {
-		key := dictionary.channelToKey(channel)
-		event := create.FromEvent(key, []byte(message))
-		dictionary.onChange(dictionary.channelToKey(channel), event)
+		var change C
+		if err := sonic.UnmarshalString(message, &change); err != nil {
+			Log.Println("ERROR: Couln't process event received through pubsub ("+create.Identifier+"):", err)
+			return
+		}
+		dictionary.onChange(dictionary.channelToKey(channel), change)
 	})
 
 	// Just print warning when an error happens in a channel for now

listener_dictionary.go

@@ -23,9 +23,8 @@ type DatabaseListenerDictionary[T any, PS IPubSubBackend, DB any, C Change[C]] s
 	// Dictionary for managing the subscriptions by key
 	subDict *ristretto.Cache[string, *ListenerSubscriptions[T, PS, C]]
 
-	get         func([]string) (map[string]C, error)
+	get         func(DB, []string) (map[string]C, error)
 	toEvent     func(string, Change[C]) neogate.Event
-	toChange    func(neogate.Event) Change[C]
 	createMutex *sync.Mutex
 	pool        *SubPool[PS]
 }
@@ -48,7 +47,7 @@ func (ld *DatabaseListenerDictionary[T, PS, DB, C]) channelToKey(channel string)
 	return values[2]
 }
 
-func DictionarySubscribe[T any, PS IPubSubBackend, DB any, C Change[C], S Subscription[C]](ld *DatabaseListenerDictionary[T, PS, DB, C], keys []string, identifier string, subscription S) error {
+func DictionarySubscribe[T any, PS IPubSubBackend, DB any, C Change[C], S Subscription[C]](ld *DatabaseListenerDictionary[T, PS, DB, C], db DB, keys []string, identifier string, subscription S) error {
 
 	// Find all listeners that are not already available
 	nonCached := []string{}
@@ -66,12 +65,12 @@ func DictionarySubscribe[T any, PS IPubSubBackend, DB any, C Change[C], S Subscr
 	}
 
 	// Create subscriptions so that they can start receiving stuff already
-	return createSubscriptions(ld, keys, identifier, subscription)
+	return createSubscriptions(ld, db, keys, identifier, subscription)
 }
 
 // Create subscriptions in the ListenerDictionary
 // TODO(unbreathable): How do we fix broken subscriptions in case an error is returned below subscription creation?
-func createSubscriptions[T any, PS IPubSubBackend, DB any, C Change[C], S Subscription[C]](ld *DatabaseListenerDictionary[T, PS, DB, C], keys []string, identifier string, subscription S) error {
+func createSubscriptions[T any, PS IPubSubBackend, DB any, C Change[C], S Subscription[C]](ld *DatabaseListenerDictionary[T, PS, DB, C], db DB, keys []string, identifier string, subscription S) error {
 	ctx := context.Background()
 
 	toGet := []string{}
@@ -100,7 +99,7 @@ func createSubscriptions[T any, PS IPubSubBackend, DB any, C Change[C], S Subscr
 	}
 
 	// Get the base data for all listeners that were created
-	results, err := ld.Get(toGet)
+	results, err := ld.Get(db, toGet)
 	if err != nil {
 		return fmt.Errorf("couldn't get from base data: %v", err)
 	}
@@ -119,9 +118,41 @@ func createSubscriptions[T any, PS IPubSubBackend, DB any, C Change[C], S Subscr
 	return nil
 }
 
+// Reset all values for the keys back to the original value by re-getting them and pushing that update
+func (ld *DatabaseListenerDictionary[T, PS, DB, C]) Reset(db DB, keys []string) error {
+	results, err := ld.get(db, keys)
+	if err != nil {
+		return err
+	}
+
+	// Build all of the messages for the outbox
+	messages := make([]OutboxMessage, len(keys))
+	for i, key := range keys {
+		change, ok := results[key]
+		if !ok {
+			return fmt.Errorf("couldn't find result for key %s", key)
+		}
+
+		// Marshal the event for the reset
+		event := ld.toEvent(key, change)
+		bytes, err := sonic.Marshal(event)
+		if err != nil {
+			return err
+		}
+
+		messages[i] = OutboxMessage{
+			Identifier: ld.keyToChannel(key),
+			Data:       bytes,
+		}
+	}
+
+	// Publish all the messages to the outbox
+	return ld.outbox.save(db, messages)
+}
+
 // Get the value for keys from the listener dictionary (makes sure we can add batching in the future)
-func (ld *DatabaseListenerDictionary[T, PS, DB, C]) Get(keys []string) (map[string]C, error) {
-	return ld.get(keys)
+func (ld *DatabaseListenerDictionary[T, PS, DB, C]) Get(db DB, keys []string) (map[string]C, error) {
+	return ld.get(db, keys)
 }
 
 // Handle a change for a specific key
@@ -131,16 +162,28 @@ func (ld *DatabaseListenerDictionary[T, PS, DB, C]) onChange(key string, change
 	}
 }
 
+// Package a key and change for the outbox
+func (ld *DatabaseListenerDictionary[T, PS, DB, C]) packageForOutbox(key string, change Change[C]) (OutboxMessage, error) {
+	var message OutboxMessage
+	bytes, err := sonic.Marshal(change)
+	if err != nil {
+		return message, err
+	}
+	message = OutboxMessage{
+		Identifier: ld.keyToChannel(key),
+		Data:       bytes,
+	}
+	return message, nil
+}
+
 // Save a change to the outbox, makes sure all of this stays transactional
 func (ld *DatabaseListenerDictionary[T, PS, DB, C]) Save(db DB, key string, change Change[C]) error {
-	event := ld.toEvent(key, change)
-	bytes, err := sonic.Marshal(event)
+	message, err := ld.packageForOutbox(key, change)
 	if err != nil {
 		return err
 	}
 
-	return ld.outbox.save(db, OutboxMessage{
-		Identifier: ld.keyToChannel(key),
-		Data:       bytes,
+	return ld.outbox.save(db, []OutboxMessage{
+		message,
 	})
 }

listener_dictionary_outbox.go

@@ -0,0 +1,29 @@
+package hydro
+
+// Ignore, needed for type check below
+type s struct{}
+
+func (d s) Stack(s Change[s]) Change[s] {
+	return d
+}
+
+// For making sure the thing actually implements the interface
+var _ PackedMessage = dbldOutboxMessage[any, IPubSubBackend, any, s]{}
+
+type dbldOutboxMessage[T any, PS IPubSubBackend, DB any, C Change[C]] struct {
+	dict   *DatabaseListenerDictionary[T, PS, DB, C]
+	key    string
+	change C
+}
+
+func (m dbldOutboxMessage[T, PS, DB, C]) convertToOutbox() (OutboxMessage, error) {
+	return m.dict.packageForOutbox(m.key, m.change)
+}
+
+func (ld *DatabaseListenerDictionary[T, PS, DB, C]) ForOutbox(key string, change C) dbldOutboxMessage[T, PS, DB, C] {
+	return dbldOutboxMessage[T, PS, DB, C]{
+		dict:   ld,
+		key:    key,
+		change: change,
+	}
+}

pubsub_outbox.go

@@ -2,6 +2,7 @@ package hydro
 
 import (
 	"context"
+	"errors"
 	"time"
 )
 
@@ -20,7 +21,7 @@ type PubSubOutbox[DB any, PS IPubSubBackend] struct {
 
 	closeChan chan struct{} // For closing the outbox
 
-	save func(database DB, message OutboxMessage) error
+	save func(database DB, messages []OutboxMessage) error
 }
 
 type OutboxMessage struct {
@@ -34,7 +35,7 @@ type OutboxCreate[DB any] struct {
 	WaitDuration time.Duration
 
 	// This function should save an event with an identifier to the database.
-	Save func(database DB, message OutboxMessage) error
+	Save func(database DB, messages []OutboxMessage) error
 
 	// This is the main function handling the pulling of messages for the outbox.
 	//
@@ -80,7 +81,7 @@ func NewOutbox[T any, DB any, PS IPubSubBackend](instance *Instance[T, PS], conn
 
 						// Send the encoded event to pub/sub
 						err := outbox.backend.Publish(context.Background(), message.Identifier, string(message.Data))
-						if err != nil {
+						if err != nil && !errors.Is(err, ErrChannelNotRegistered) {
 							// On publish error, use exponential backoff
 							backoff *= 2
 							if backoff > maxBackoff {
@@ -101,9 +102,11 @@ func NewOutbox[T any, DB any, PS IPubSubBackend](instance *Instance[T, PS], conn
 
 // Save an event to the outbox. Use this for transactional pub/sub using the database.
 func (o *PubSubOutbox[DB, PS]) Save(db DB, identifier string, data []byte) error {
-	return o.save(db, OutboxMessage{
-		Identifier: identifier,
-		Data:       data,
+	return o.save(db, []OutboxMessage{
+		{
+			Identifier: identifier,
+			Data:       data,
+		},
 	})
 }
 
@@ -121,3 +124,21 @@ func (o *PubSubOutbox[DB, PS]) CreateWorker() ISubWorker {
 func (o *PubSubOutbox[DB, PS]) Close() {
 	o.closeChan <- struct{}{}
 }
+
+type PackedMessage interface {
+	convertToOutbox() (OutboxMessage, error)
+}
+
+// Method for inserting multiple things into the outbox with one query
+func (o *PubSubOutbox[DB, PS]) SaveMultiple(db DB, sends []PackedMessage) error {
+	messages := make([]OutboxMessage, len(sends))
+	for i, send := range sends {
+		result, err := send.convertToOutbox()
+		if err != nil {
+			return err
+		}
+		messages[i] = result
+	}
+
+	return o.save(db, messages)
+}