feat(outbox): implements outboxsql adapter

Author: notjustmoney

outbox/adapter/sql/config.go

@@ -0,0 +1,5 @@
+package outboxsql
+
+type Config struct {
+	TableName string
+}

outbox/adapter/sql/db_ctx.go

@@ -0,0 +1,31 @@
+package outboxsql
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/Masterminds/squirrel"
+)
+
+type dbCtx struct {
+	db *sql.DB
+}
+
+func (d *dbCtx) executor(ctx context.Context) squirrel.BaseRunner {
+	tx, ok := txFromContext(ctx)
+	if !ok {
+		return d.db
+	}
+	return tx
+}
+
+type txContextKey struct{}
+
+func withTx(ctx context.Context, tx *sql.Tx) context.Context {
+	return context.WithValue(ctx, txContextKey{}, tx)
+}
+
+func txFromContext(ctx context.Context) (*sql.Tx, bool) {
+	tx, ok := ctx.Value(txContextKey{}).(*sql.Tx)
+	return tx, ok
+}

outbox/adapter/sql/go.mod

@@ -0,0 +1,17 @@
+module github.com/notjustmoney/protobox/outbox/adapter/sql
+
+go 1.23.4
+
+require (
+	github.com/Masterminds/squirrel v1.5.4
+	github.com/notjustmoney/protobox v0.0.0-20250128131955-5f749f4fabb3
+	github.com/samber/lo v1.49.1
+)
+
+require (
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
+	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+)

outbox/adapter/sql/go.sum

@@ -0,0 +1,25 @@
+github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8afzqM=
+github.com/Masterminds/squirrel v1.5.4/go.mod h1:NNaOrjSoIDfDA40n7sr2tPNZRfjzjA400rg+riTZj10=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 h1:SOEGU9fKiNWd/HOJuq6+3iTQz8KNCLtVX6idSoTLdUw=
+github.com/lann/builder v0.0.0-20180802200727-47ae307949d0/go.mod h1:dXGbAdH5GtBTC4WfIxhKZfyBF/HBFgRZSWwZ9g/He9o=
+github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 h1:P6pPBnrTSX3DEVR4fDembhRWSsG5rVo6hYhAB/ADZrk=
+github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0/go.mod h1:vmVJ0l/dxyfGW6FmdpVm2joNMFikkuWg0EoCKLGUMNw=
+github.com/notjustmoney/protobox v0.0.0-20250128131955-5f749f4fabb3 h1:TZRn8YQrtF5Xj2DTFVuDkoYwjwOOBdNyrJUdDwK3nvk=
+github.com/notjustmoney/protobox v0.0.0-20250128131955-5f749f4fabb3/go.mod h1:c47ESQ/AG/c9d7YZIFTk5BsgTnTmNbIUpSsXojs2dC8=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/samber/lo v1.49.1 h1:4BIFyVfuQSEpluc7Fua+j1NolZHiEHEpaSEKdsH0tew=
+github.com/samber/lo v1.49.1/go.mod h1:dO6KHFzUKXgP8LDhU0oI8d2hekjXnGOu0DB8Jecxd6o=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

outbox/adapter/sql/marker.go

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2025. protobox
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package outboxsql
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/samber/lo"
+
+	"github.com/notjustmoney/protobox/outbox"
+)
+
+type MarkerFunc func(ctx context.Context, messages []outbox.PublishedMessage) (int, error)
+
+func (f MarkerFunc) Mark(ctx context.Context, messages []outbox.PublishedMessage) (int, error) {
+	return f(ctx, messages)
+}
+
+func Marker(
+	dbCtx *dbCtx,
+	cfg Config,
+) MarkerFunc {
+	return func(ctx context.Context, messages []outbox.PublishedMessage) (int, error) {
+		args := make([]interface{}, 0, len(messages)*3)
+		for i := 0; i < len(messages); i++ {
+			args = append(args, messages[i].ID)
+			args = append(args, messages[i].ProcessedAt.UTC())
+			args = append(args, messages[i].Error)
+		}
+
+		if _, err := sq.
+			Update(cfg.TableName).
+			Set("processed_at", sq.Expr("v.processed_at")).
+			Set("error", sq.Expr("v.error")).
+			Suffix("FROM").
+			SuffixExpr(
+				sq.Alias(
+					sq.Expr(fmt.Sprintf("VALUES %s", strings.Join(lo.Map(messages, func(_ outbox.PublishedMessage, index int) string {
+						return fmt.Sprintf("($%d, $%d::timestamp with time zone, $%d)", index*3+1, index*3+2, index*3+3)
+					}), ",")), args...),
+					"v(id, processed_at, error)",
+				),
+			).
+			Suffix("WHERE " + cfg.TableName + ".id = v.id::uuid").
+			PlaceholderFormat(sq.Dollar).
+			RunWith(dbCtx.executor(ctx)).
+			ExecContext(ctx); err != nil {
+			return 0, err
+		}
+
+		return len(messages), nil
+	}
+}

outbox/adapter/sql/poller.go

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2025. protobox
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package outboxsql
+
+import (
+	"context"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/notjustmoney/protobox/outbox"
+)
+
+type PollerFunc func(ctx context.Context, cfg outbox.PollConfig) ([]outbox.Message, error)
+
+func (f PollerFunc) Poll(ctx context.Context, cfg outbox.PollConfig) ([]outbox.Message, error) {
+	return f(ctx, cfg)
+}
+
+func Poller(dbCtx *dbCtx, cfg Config) PollerFunc {
+	return func(ctx context.Context, pollCfg outbox.PollConfig) ([]outbox.Message, error) {
+		rows, err := sq.
+			Select("id", "topic", "payload").
+			From(cfg.TableName).
+			Where(sq.Eq{"processed_at": nil}).
+			OrderBy("created_at").
+			Limit(uint64(pollCfg.BatchSize)).
+			Suffix("FOR UPDATE SKIP LOCKED").
+			PlaceholderFormat(sq.Dollar).
+			RunWith(dbCtx.executor(ctx)).
+			QueryContext(ctx)
+		if err != nil {
+			return nil, err
+		}
+
+		var messages []outbox.Message
+		for rows.Next() {
+			var message outbox.Message
+			if err := rows.Scan(&message.ID, &message.Topic, &message.Payload); err != nil {
+				return nil, err
+			}
+			messages = append(messages, message)
+		}
+
+		return messages, nil
+	}
+}

outbox/adapter/sql/relay_handler.go

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2025. protobox
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package outboxsql
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/notjustmoney/protobox/outbox"
+)
+
+type RelayHandler struct {
+	db           *sql.DB
+	relayHandler *outbox.RelayHandler
+}
+
+func NewRelayHandler(
+	db *sql.DB,
+	publisher outbox.Publisher,
+	cfg Config,
+) *RelayHandler {
+	dbCtx := &dbCtx{db: db}
+	return &RelayHandler{
+		db: db,
+		relayHandler: outbox.NewRelayHandler(
+			Poller(dbCtx, cfg),
+			publisher,
+			Marker(dbCtx, cfg),
+		),
+	}
+}
+
+func (h *RelayHandler) Handle(ctx context.Context) (int, error) {
+	tx, err := h.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, fmt.Errorf("outboxsql: %w", err)
+	}
+	defer func() {
+		if p := recover(); p != nil {
+			_ = tx.Rollback()
+			panic(p)
+		}
+		if err != nil {
+			_ = tx.Rollback()
+			return
+		}
+		_ = tx.Commit()
+	}()
+
+	processed, err := h.Handle(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("outboxsql: %w", err)
+	}
+	return processed, nil
+}