txutil

type BulkSyncTxDelete ¶

type BulkSyncTxDelete[ID any, Topic comparable] struct {
	// contains filtered or unexported fields
}

BulkSyncTxDelete wraps a SQL transaction and tracks delete events to publish grouped by topic after successful commit.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewBulkDeleteTx[IDType, TopicType](tx, extractDeleteTopicFn)
defer devtoolsdb.TxnRollback(tx)

for _, id := range ids {
    service.Delete(ctx, id)
    syncTx.Track(id, workspaceID, isDelta)
}

err := syncTx.CommitAndPublish(ctx, publishBulkDeleteFn)

func NewBulkDeleteTx[ID any, Topic comparable](
	tx *sql.Tx,
	topicExtractor func(DeleteEvent[ID]) Topic,
) *BulkSyncTxDelete[ID, Topic]

func (s *BulkSyncTxDelete[ID, Topic]) CommitAndPublish(
	ctx context.Context,
	publishFn func(Topic, []DeleteEvent[ID]),
) error

func (s *BulkSyncTxDelete[ID, Topic]) Track(id ID, workspaceID ID, isDelta bool)

type BulkSyncTxInsert ¶

type BulkSyncTxInsert[T any, Topic comparable] struct {
	// contains filtered or unexported fields
}

BulkSyncTxInsert wraps a SQL transaction and tracks items to publish bulk sync events grouped by topic after successful commit. This eliminates the need for manual loops and ensures events are never forgotten.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewBulkInsertTx[ItemType, TopicType](tx, extractTopicFn)
defer devtoolsdb.TxnRollback(tx)

for _, item := range items {
    service.Create(ctx, item)
    syncTx.Track(item)
}

err := syncTx.CommitAndPublish(ctx, publishBulkInsertFn)

func NewBulkInsertTx[T any, Topic comparable](
	tx *sql.Tx,
	topicExtractor TopicExtractor[T, Topic],
) *BulkSyncTxInsert[T, Topic]

func (s *BulkSyncTxInsert[T, Topic]) CommitAndPublish(
	ctx context.Context,
	publishFn func(Topic, []T),
) error

func (s *BulkSyncTxInsert[T, Topic]) Track(item T)

type BulkSyncTxUpdate ¶

type BulkSyncTxUpdate[T any, P any, Topic comparable] struct {
	// contains filtered or unexported fields
}

BulkSyncTxUpdate wraps a SQL transaction and tracks update events to publish grouped by topic after successful commit.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewBulkUpdateTx[ItemType, PatchType, TopicType](tx, extractTopicFn)
defer devtoolsdb.TxnRollback(tx)

for _, update := range updates {
    service.Update(ctx, update.Item)
    syncTx.Track(update.Item, update.Patch)
}

err := syncTx.CommitAndPublish(ctx, publishBulkUpdateFn)

func NewBulkUpdateTx[T any, P any, Topic comparable](
	tx *sql.Tx,
	topicExtractor TopicExtractor[T, Topic],
) *BulkSyncTxUpdate[T, P, Topic]

func (s *BulkSyncTxUpdate[T, P, Topic]) CommitAndPublish(
	ctx context.Context,
	publishFn func(Topic, []UpdateEvent[T, P]),
) error

func (s *BulkSyncTxUpdate[T, P, Topic]) Track(item T, patch P)

type DeleteEvent ¶

type DeleteEvent[ID any] struct {
	ID          ID
	WorkspaceID ID
	IsDelta     bool
}

DeleteEvent represents a delete operation with the ID and workspace context

type SyncTxDelete ¶

type SyncTxDelete[ID any] struct {
	// contains filtered or unexported fields
}

SyncTxDelete wraps a SQL transaction and tracks delete events to publish after successful commit.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewDeleteTx[idwrap.IDWrap](tx)
defer devtoolsdb.TxnRollback(tx)

for _, id := range ids {
    service.Delete(ctx, id)
    syncTx.Track(id, workspaceID, isDelta)
}

err := syncTx.CommitAndPublish(ctx, publishDeleteEvent)

func NewDeleteTx[ID any](tx *sql.Tx) *SyncTxDelete[ID]

func (s *SyncTxDelete[ID]) CommitAndPublish(
	ctx context.Context,
	publishFn func(ID, ID, bool),
) error

func (s *SyncTxDelete[ID]) Track(id ID, workspaceID ID, isDelta bool)

type SyncTxInsert ¶

type SyncTxInsert[T any] struct {
	// contains filtered or unexported fields
}

SyncTxInsert wraps a SQL transaction and tracks items to publish sync events after successful commit. This ensures sync events are never forgotten.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewInsertTx[mhttp.HTTP](tx)
defer devtoolsdb.TxnRollback(tx)

for _, item := range items {
    service.Create(ctx, item)
    syncTx.Track(item)
}

err := syncTx.CommitAndPublish(ctx, publishInsertEvent)

func NewInsertTx[T any](tx *sql.Tx) *SyncTxInsert[T]

func (s *SyncTxInsert[T]) CommitAndPublish(
	ctx context.Context,
	publishFn func(T),
) error

func (s *SyncTxInsert[T]) Track(item T)

type SyncTxUpdate ¶

type SyncTxUpdate[T any, P any] struct {
	// contains filtered or unexported fields
}

SyncTxUpdate wraps a SQL transaction and tracks update events to publish after successful commit.

Usage:

tx, _ := db.BeginTx(ctx, nil)
syncTx := txutil.NewUpdateTx[mhttp.HTTP, patch.HTTPDeltaPatch](tx)
defer devtoolsdb.TxnRollback(tx)

for _, update := range updates {
    service.Update(ctx, update.Item)
    syncTx.Track(update.Item, update.Patch)
}

err := syncTx.CommitAndPublish(ctx, publishUpdateEvent)

func NewUpdateTx[T any, P any](tx *sql.Tx) *SyncTxUpdate[T, P]

func (s *SyncTxUpdate[T, P]) CommitAndPublish(
	ctx context.Context,
	publishFn func(T, P),
) error

func (s *SyncTxUpdate[T, P]) Track(item T, patch P)

type TopicExtractor ¶

type TopicExtractor[T any, Topic any] func(item T) Topic

TopicExtractor is a function that extracts a topic from an item. Used for auto-grouping items by topic before bulk publishing.

type UpdateEvent ¶

type UpdateEvent[T any, P any] struct {
	Item  T
	Patch P
}

UpdateEvent represents an update operation with the updated item and its delta patch