New Messaging

buf.gen.yaml

@@ -4,7 +4,7 @@ plugins:
     out: .
     opt: paths=import,module=github.com/pentops/registry
 
-  - name: go-messaging # github.com/pentops/protoc-gen-go-messaging
+  - name: go-o5-messaging # github.com/pentops/o5-messaging.go/cmd/protoc-gen-go-o5-messaging
     out: .
     opt: paths=import,module=github.com/pentops/registry
 

buildwrap/worker.go

@@ -14,7 +14,7 @@ import (
 	"github.com/pentops/j5/gen/j5/source/v1/source_j5pb"
 	"github.com/pentops/log.go/log"
 	"github.com/pentops/o5-go/messaging/v1/messaging_pb"
-	"github.com/pentops/outbox.pg.go/outbox"
+	"github.com/pentops/o5-messaging.go/o5msg"
 	"github.com/pentops/registry/gen/o5/registry/builder/v1/builder_tpb"
 	"github.com/pentops/registry/github"
 	"google.golang.org/protobuf/types/known/emptypb"
@@ -26,7 +26,7 @@ type Storage interface {
 }
 
 type Publisher interface {
-	Publish(ctx context.Context, msg ...outbox.OutboxMessage) error
+	Publish(ctx context.Context, msg o5msg.Message) error
 }
 
 type BuildWorker struct {
@@ -48,7 +48,7 @@ type IGithub interface {
 	GetCommit(ctx context.Context, ref github.RepoRef) (*source_j5pb.CommitInfo, error)
 }
 
-func NewBuildWorker(builder J5Builder, github IGithub, store Storage, publisher Publisher) *BuildWorker {
+func NewBuildWorker(builder J5Builder, github IGithub, store Storage, publisher o5msg.Publisher) *BuildWorker {
 	return &BuildWorker{
 		builder:   builder,
 		github:    github,

cmd/registry/main.go

@@ -31,7 +31,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
-	"github.com/pentops/outbox.pg.go/outbox"
+	"github.com/pentops/o5-messaging.go/outbox"
 )
 
 var Version = "0.0.0"
@@ -124,7 +124,7 @@ func runCombinedServer(ctx context.Context, cfg struct {
 		return err
 	}
 
-	dbPublisher, err := outbox.NewDBPublisher(db)
+	dbPublisher, err := outbox.NewDirectPublisher(db, outbox.DefaultSender)
 	if err != nil {
 		return err
 	}
@@ -137,12 +137,7 @@ func runCombinedServer(ctx context.Context, cfg struct {
 		return err
 	}
 
-	publisher, err := outbox.NewDBPublisher(db)
-	if err != nil {
-		return err
-	}
-
-	githubWorker, err := service.NewWebhookWorker(refStore, githubClient, publisher)
+	githubWorker, err := service.NewWebhookWorker(refStore, githubClient, dbPublisher)
 	if err != nil {
 		return err
 	}

ext/db/00001_outbox.sql

@@ -1,8 +1,7 @@
 -- +goose Up
 CREATE TABLE outbox (
 	id uuid PRIMARY KEY,
-	destination text NOT NULL,
-	message bytea NOT NULL,
+	data jsonb NOT NULL,
 	headers text NOT NULL
 );