perf: batch insert attemptDetail

Author: vm-001

app/app.go

@@ -132,7 +132,7 @@ func (app *Application) initialize() error {
 			PoolConcurrency: int(cfg.Worker.Pool.Concurrency),
 		}
 		deliverer := deliverer.NewHTTPDeliverer(&cfg.Worker.Deliverer)
-		app.worker = worker.NewWorker(opts, db, deliverer, queue, app.metrics, tracer, app.bus)
+		app.worker = worker.NewWorker(opts, db, deliverer, queue, app.metrics, tracer, app.bus, app.dispatcher)
 	}
 
 	// admin

db/dao/attempt_detail_dao.go

@@ -30,22 +30,25 @@ func NewAttemptDetailDao(db *sqlx.DB, bus *eventbus.EventBus, workspace bool) At
 	}
 }
 
-func (dao *attemptDetailDao) Insert(ctx context.Context, attemptDetail *entities.AttemptDetail) error {
-	ctx, span := tracing.Start(ctx, fmt.Sprintf("dao.%s.insert", dao.opts.Table), trace.WithSpanKind(trace.SpanKindServer))
+func (dao *attemptDetailDao) BatchInsert(ctx context.Context, entities []*entities.AttemptDetail) error {
+	ctx, span := tracing.Start(ctx, fmt.Sprintf("dao.%s.batch_insert", dao.opts.Table), trace.WithSpanKind(trace.SpanKindServer))
 	defer span.End()
 
 	now := time.Now()
-	values := []interface{}{attemptDetail.ID, attemptDetail.RequestHeaders, attemptDetail.RequestBody, attemptDetail.ResponseHeaders, attemptDetail.ResponseBody, now, now, attemptDetail.WorkspaceId}
 
-	sql := `INSERT INTO attempt_details (id, request_headers, request_body, response_headers, response_body, created_at, updated_at, ws_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) 
-		ON CONFLICT (id) DO UPDATE SET 
-		request_headers = EXCLUDED.request_headers, 
-		request_body = EXCLUDED.request_body, 
-		response_headers = EXCLUDED.response_headers, 
-		response_body = EXCLUDED.response_body, 
-		updated_at = EXCLUDED.updated_at`
-
-	result, err := dao.DB(ctx).ExecContext(ctx, sql, values...)
+	builder := psql.Insert(dao.opts.Table).Columns("id", "request_headers", "request_body", "response_headers", "response_body", "created_at", "updated_at", "ws_id")
+	for _, entity := range entities {
+		builder = builder.Values(entity.ID, entity.RequestHeaders, entity.RequestBody, entity.ResponseHeaders, entity.ResponseBody, now, now, entity.WorkspaceId)
+	}
+	sql, args := builder.Suffix(`
+		ON CONFLICT (id) DO UPDATE SET
+		request_headers = EXCLUDED.request_headers,
+		request_body = EXCLUDED.request_body,
+		response_headers = EXCLUDED.response_headers,
+		response_body = EXCLUDED.response_body,
+		updated_at = EXCLUDED.updated_at`).MustSql()
+	dao.debugSQL(sql, args)
+	result, err := dao.DB(ctx).ExecContext(ctx, sql, args...)
 	if err != nil {
 		return err
 	}

db/dao/daos.go

@@ -50,7 +50,7 @@ type SourceDAO interface {
 
 type AttemptDetailDAO interface {
 	BaseDAO[entities.AttemptDetail]
-	Insert(ctx context.Context, attemptDetail *entities.AttemptDetail) error
+	BatchInsert(ctx context.Context, entities []*entities.AttemptDetail) error
 }
 
 type PluginDAO interface {

db/entities/types.go

@@ -3,6 +3,7 @@ package entities
 import (
 	"database/sql/driver"
 	"encoding/json"
+	"errors"
 	"github.com/lib/pq"
 	"github.com/webhookx-io/webhookx/pkg/types"
 )
@@ -26,7 +27,14 @@ type BaseModel struct {
 type Headers map[string]string
 
 func (m *Headers) Scan(src interface{}) error {
-	return json.Unmarshal(src.([]byte), m)
+	switch v := src.(type) {
+	case string:
+		return json.Unmarshal([]byte(v), m)
+	case []byte:
+		return json.Unmarshal(v, m)
+	default:
+		return errors.New("unknown type")
+	}
 }
 
 func (m Headers) Value() (driver.Value, error) {

db/migrations/7_update_attempt_details.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE IF EXISTS ONLY "attempt_details" ALTER COLUMN request_headers TYPE JSONB USING request_headers::JSONB;
+ALTER TABLE IF EXISTS ONLY "attempt_details" ALTER COLUMN response_headers TYPE JSONB USING response_headers::JSONB;
+CREATE INDEX idx_attempt_details_ws_id ON attempt_details (ws_id);

db/migrations/7_update_attempt_details.up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE IF EXISTS ONLY "attempt_details" ALTER COLUMN request_headers TYPE TEXT;
+ALTER TABLE IF EXISTS ONLY "attempt_details" ALTER COLUMN response_headers TYPE TEXT;
+DROP INDEX idx_attempt_details_ws_id;

dispatcher/dispatcher.go

@@ -13,7 +13,6 @@ import (
 	"github.com/webhookx-io/webhookx/pkg/tracing"
 	"github.com/webhookx-io/webhookx/pkg/types"
 	"github.com/webhookx-io/webhookx/utils"
-	"github.com/webhookx-io/webhookx/worker"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"time"
@@ -96,7 +95,7 @@ func (d *Dispatcher) dispatchBatch(ctx context.Context, events []*entities.Event
 		return d.db.Attempts.BatchInsert(ctx, attempts)
 	})
 	if err == nil {
-		go d.sendToQueue(context.WithoutCancel(ctx), attempts)
+		go d.SendToQueue(context.WithoutCancel(ctx), attempts)
 	}
 	return n, err
 }
@@ -130,19 +129,19 @@ func (d *Dispatcher) DispatchEndpoint(ctx context.Context, event *entities.Event
 		return err
 	}
 
-	d.sendToQueue(ctx, attempts)
+	d.SendToQueue(ctx, attempts)
 
 	return nil
 }
 
-func (d *Dispatcher) sendToQueue(ctx context.Context, attempts []*entities.Attempt) {
+func (d *Dispatcher) SendToQueue(ctx context.Context, attempts []*entities.Attempt) {
 	tasks := make([]*taskqueue.TaskMessage, 0, len(attempts))
 	ids := make([]string, 0, len(attempts))
 	for _, attempt := range attempts {
 		tasks = append(tasks, &taskqueue.TaskMessage{
 			ID:          attempt.ID,
 			ScheduledAt: attempt.ScheduledAt.Time,
-			Data: &worker.MessageData{
+			Data: &taskqueue.MessageData{
 				EventID:    attempt.EventId,
 				EndpointId: attempt.EndpointId,
 				Attempt:    attempt.AttemptNumber,

pkg/taskqueue/queue.go

@@ -13,6 +13,13 @@ type TaskMessage struct {
 	data        []byte
 }
 
+type MessageData struct {
+	EventID    string `json:"event_id"`
+	EndpointId string `json:"endpoint_id"`
+	Attempt    int    `json:"attempt"`
+	Event      string `json:"event"`
+}
+
 func (t *TaskMessage) String() string {
 	return t.ID + ":" + string(t.data)
 }

test/tracing/worker_test.go

@@ -53,13 +53,12 @@ var _ = Describe("tracing worker", Ordered, func() {
 					"github.com/webhookx-io/webhookx",
 				}
 				expectedScopeSpans := map[string]map[string]string{
-					"worker.submit":              {},
-					"worker.handle_task":         {},
-					"dao.endpoints.get":          {},
-					"dao.plugins.list":           {},
-					"worker.deliver":             {},
-					"dao.attempt_details.insert": {},
-					"taskqueue.redis.delete":     {},
+					"worker.submit":          {},
+					"worker.handle_task":     {},
+					"dao.endpoints.get":      {},
+					"dao.plugins.list":       {},
+					"worker.deliver":         {},
+					"taskqueue.redis.delete": {},
 				}
 
 				n, err := helper.FileCountLine(helper.OtelCollectorTracesFile)

test/worker/requeue_test.go

@@ -45,7 +45,7 @@ var _ = Describe("processRequeue", Ordered, func() {
 		assert.NoError(GinkgoT(), err)
 		w = worker.NewWorker(worker.WorkerOptions{
 			RequeueJobInterval: time.Second,
-		}, db, deliverer.NewHTTPDeliverer(&config.WorkerDeliverer{}), queue, metrics, tracer, mocks.MockBus{})
+		}, db, deliverer.NewHTTPDeliverer(&config.WorkerDeliverer{}), queue, metrics, tracer, mocks.MockBus{}, nil)
 
 		// data
 		ws := utils.Must(db.Workspaces.GetDefault(context.TODO()))