feat(record): added WithRecordChanges

record/record_updates.go

@@ -0,0 +1,9 @@
+package record
+
+import "github.com/dal-go/dalgo/dal"
+
+// Updates defines updates for a record
+type Updates struct {
+	Record  dal.Record
+	Updates []dal.Update
+}

record/with_record_changes.go

@@ -0,0 +1,82 @@
+package record
+
+import (
+	"context"
+	"fmt"
+	"github.com/dal-go/dalgo/dal"
+)
+
+type WithRecordChanges struct {
+	recordsToInsert []dal.Record
+	RecordsToUpdate []*Updates
+	RecordsToDelete []*dal.Key
+}
+
+func (v *WithRecordChanges) RecordsToInsert() (records []dal.Record) {
+	if len(v.recordsToInsert) == 0 {
+		return v.recordsToInsert
+	}
+	records = make([]dal.Record, len(v.recordsToInsert))
+	copy(records, v.recordsToInsert)
+	return
+}
+
+func (v *WithRecordChanges) QueueForInsert(records ...dal.Record) {
+	for i, record := range records {
+		if record == nil {
+			panic(fmt.Sprintf("record #%d is required", i))
+		}
+		if record.Key() == nil {
+			panic(fmt.Sprintf("record #%d.Key() is required", i))
+		}
+		if record.Data() == nil {
+			panic(fmt.Sprintf("record #%d.Data() is required", i))
+		}
+		v.recordsToInsert = append(v.recordsToInsert, record)
+	}
+}
+
+func excludeRecords(records []dal.Record, excludeKeys []*dal.Key) (result []dal.Record) {
+	if len(excludeKeys) == 0 {
+		return records
+	}
+	result = make([]dal.Record, 0, len(records))
+external:
+	for _, record := range records {
+		for _, excludeKey := range excludeKeys {
+			if record.Key() == excludeKey {
+				continue external
+			}
+		}
+		result = append(result, record)
+	}
+	return
+}
+
+func (v *WithRecordChanges) ApplyChanges(ctx context.Context, tx dal.ReadwriteTransaction, excludeKeys ...*dal.Key) (err error) {
+
+	if records := excludeRecords(v.recordsToInsert, excludeKeys); len(records) > 0 {
+		if err = tx.InsertMulti(ctx, records); err != nil {
+			err = fmt.Errorf("failed to insert records: %w", err)
+			return
+		}
+	}
+	if len(v.RecordsToUpdate) > 0 {
+		for _, record2update := range v.RecordsToUpdate {
+			key := record2update.Record.Key()
+			if err = tx.Update(ctx, key, record2update.Updates); err != nil {
+				return fmt.Errorf("failed to update record %s: %w", key, err)
+			}
+		}
+	}
+	if len(v.RecordsToDelete) > 0 {
+		if err = tx.DeleteMulti(ctx, v.RecordsToDelete); err != nil {
+			err = fmt.Errorf("failed to delete records: %w", err)
+			return
+		}
+	}
+	v.recordsToInsert = nil
+	v.RecordsToUpdate = nil
+	v.RecordsToDelete = nil
+	return
+}

record/with_record_changes_test.go

@@ -0,0 +1,109 @@
+package record
+
+import (
+	"context"
+	"github.com/dal-go/dalgo/dal"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestWithRecordChanges_ApplyChanges(t *testing.T) {
+	type fields struct {
+		recordsToInsert []dal.Record
+		RecordsToUpdate []*Updates
+		RecordsToDelete []*dal.Key
+	}
+	type args struct {
+		ctx         context.Context
+		tx          dal.ReadwriteTransaction
+		excludeKeys []*dal.Key
+	}
+	tests := []struct {
+		name      string
+		fields    fields
+		args      args
+		assertErr assert.ErrorAssertionFunc
+	}{
+		{
+			name: "nil",
+			fields: fields{
+				recordsToInsert: nil,
+				RecordsToUpdate: nil,
+				RecordsToDelete: nil,
+			},
+			assertErr: assert.NoError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := &WithRecordChanges{
+				recordsToInsert: tt.fields.recordsToInsert,
+				RecordsToUpdate: tt.fields.RecordsToUpdate,
+				RecordsToDelete: tt.fields.RecordsToDelete,
+			}
+			err := v.ApplyChanges(tt.args.ctx, tt.args.tx, tt.args.excludeKeys...)
+			tt.assertErr(t, err)
+		})
+	}
+}
+
+func TestWithRecordChanges_QueueForInsert(t *testing.T) {
+	type fields struct {
+		recordsToInsert []dal.Record
+	}
+	type args struct {
+		records []dal.Record
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+	}{
+		{
+			name: "nil",
+			fields: fields{
+				recordsToInsert: nil,
+			},
+			args: args{
+				records: nil,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := &WithRecordChanges{
+				recordsToInsert: tt.fields.recordsToInsert,
+			}
+			v.QueueForInsert(tt.args.records...)
+		})
+	}
+}
+
+func TestWithRecordChanges_RecordsToInsert(t *testing.T) {
+	type fields struct {
+		recordsToInsert []dal.Record
+		RecordsToUpdate []*Updates
+		RecordsToDelete []*dal.Key
+	}
+	tests := []struct {
+		name   string
+		fields fields
+	}{
+		{
+			name: "nil",
+			fields: fields{
+				recordsToInsert: nil,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := &WithRecordChanges{
+				recordsToInsert: tt.fields.recordsToInsert,
+				RecordsToUpdate: tt.fields.RecordsToUpdate,
+				RecordsToDelete: tt.fields.RecordsToDelete,
+			}
+			assert.Equalf(t, tt.fields.recordsToInsert, v.RecordsToInsert(), "RecordsToInsert()")
+		})
+	}
+}