Replay Protos

internal/service/app.go

@@ -11,9 +11,10 @@ import (
 )
 
 type App struct {
-	QueryService *QueryService
-	EventWorker  *EventWorker
-	ReplayWorker *ReplayWorker
+	QueryService   *QueryService
+	CommandService *CommandService
+	EventWorker    *EventWorker
+	ReplayWorker   *ReplayWorker
 }
 
 func NewApp(db sqrlx.Transactor) (*App, error) {
@@ -22,18 +23,25 @@ func NewApp(db sqrlx.Transactor) (*App, error) {
 		return nil, fmt.Errorf("failed to create query service: %w", err)
 	}
 
+	cs, err := NewCommandService(db)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create command service: %w", err)
+	}
+
 	replayWorker := NewReplayWorker(db)
 
 	app := &App{
-		QueryService: qs,
-		EventWorker:  NewEventWorker(db),
-		ReplayWorker: replayWorker,
+		QueryService:   qs,
+		CommandService: cs,
+		EventWorker:    NewEventWorker(db),
+		ReplayWorker:   replayWorker,
 	}
 	return app, nil
 }
 
 func (a *App) RegisterGRPC(server grpc.ServiceRegistrar) {
 	a.QueryService.RegisterGRPC(server)
+	a.CommandService.RegisterGRPC(server)
 	a.EventWorker.RegisterGRPC(server)
 	a.ReplayWorker.RegisterGRPC(server)
 }

internal/service/command.go

@@ -0,0 +1,61 @@
+package service
+
+import (
+	"context"
+
+	"github.com/pentops/ges/internal/gen/o5/ges/v1/ges_spb"
+	"github.com/pentops/ges/internal/gen/o5/ges/v1/ges_tpb"
+	"github.com/pentops/o5-messaging/outbox"
+	"github.com/pentops/sqrlx.go/sqrlx"
+	"google.golang.org/grpc"
+)
+
+type CommandService struct {
+	db sqrlx.Transactor
+
+	ges_spb.UnsafeCommandServiceServer
+}
+
+func NewCommandService(db sqrlx.Transactor) (*CommandService, error) {
+	return &CommandService{
+		db: db,
+	}, nil
+}
+
+func (s *CommandService) RegisterGRPC(server grpc.ServiceRegistrar) {
+	ges_spb.RegisterCommandServiceServer(server, s)
+}
+
+func (s *CommandService) ReplayEvents(ctx context.Context, req *ges_spb.ReplayEventsRequest) (*ges_spb.ReplayEventsResponse, error) {
+	msg := &ges_tpb.EventsMessage{
+		QueueUrl:    req.QueueUrl,
+		GrpcService: req.GrpcService,
+		GrpcMethod:  req.GrpcService,
+	}
+	if err := s.db.Transact(ctx, &sqrlx.TxOptions{
+		ReadOnly:  false,
+		Retryable: true,
+	}, func(ctx context.Context, tx sqrlx.Transaction) error {
+		return outbox.Send(ctx, tx, msg)
+	}); err != nil {
+		return nil, err
+	}
+	return &ges_spb.ReplayEventsResponse{}, nil
+}
+
+func (s *CommandService) ReplayUpserts(ctx context.Context, req *ges_spb.ReplayUpsertsRequest) (*ges_spb.ReplayUpsertsResponse, error) {
+	msg := &ges_tpb.UpsertsMessage{
+		QueueUrl:    req.QueueUrl,
+		GrpcService: req.GrpcService,
+		GrpcMethod:  req.GrpcService,
+	}
+	if err := s.db.Transact(ctx, &sqrlx.TxOptions{
+		ReadOnly:  false,
+		Retryable: true,
+	}, func(ctx context.Context, tx sqrlx.Transaction) error {
+		return outbox.Send(ctx, tx, msg)
+	}); err != nil {
+		return nil, err
+	}
+	return &ges_spb.ReplayUpsertsResponse{}, nil
+}

internal/service/query.go

@@ -19,10 +19,7 @@ type QueryService struct {
 	ges_spb.UnsafeQueryServiceServer
 }
 
-var _ ges_spb.QueryServiceServer = &QueryService{}
-
 func NewQueryService(db sqrlx.Transactor) (*QueryService, error) {
-
 	eventLister, err := pquery.NewLister(
 		pquery.ListSpec[*ges_spb.EventsListRequest, *ges_spb.EventsListResponse]{
 			TableSpec: pquery.TableSpec{
@@ -32,12 +29,11 @@ func NewQueryService(db sqrlx.Transactor) (*QueryService, error) {
 					pquery.NewProtoField("metadata.timestamp", gl.Ptr("timestamp")),
 					pquery.NewProtoField("metadata.event_id", gl.Ptr("id")),
 				},
-				//Auth:
-				//AuthJoin:
+				// Auth:
+				// AuthJoin:
 			},
-			//RequestFilter: smSpec.ListRequestFilter,
+			// RequestFilter: smSpec.ListRequestFilter,
 		})
-
 	if err != nil {
 		return nil, fmt.Errorf("create event lister: %w", err)
 	}
@@ -51,12 +47,11 @@ func NewQueryService(db sqrlx.Transactor) (*QueryService, error) {
 					pquery.NewProtoField("entity_id", gl.Ptr("entity_id")),
 					pquery.NewProtoField("last_event_timestamp", gl.Ptr("last_event_timestamp")),
 				},
-				//Auth:
-				//AuthJoin:
+				// Auth:
+				// AuthJoin:
 			},
-			//RequestFilter: smSpec.ListRequestFilter,
+			// RequestFilter: smSpec.ListRequestFilter,
 		})
-
 	if err != nil {
 		return nil, fmt.Errorf("create upsert lister: %w", err)
 	}

schema/ges/o5/ges/v1/events.j5s

@@ -35,6 +35,37 @@ service Query {
 			field events array:object:Upsert
 		}
 	}
+}
+
+service Command {
+	basePath = "/ges/v1"
+    method ReplayEvents {
+        httpMethod = "POST"
+        httpPath = "/events/replay"
+
+        request {
+            field queueURL ! string
+            field grpcService ! string
+            field grpcMethod ! string
+        }
+
+        response {
+        }
+    }
+
+    method ReplayUpserts { 
+        httpMethod = "POST"
+        httpPath = "/upserts/replay"
+
+        request {
+            field queueURL ! string
+            field grpcService ! string
+            field grpcMethod ! string
+        }
+
+        response {
+        }
+    }
 
 }
 
@@ -80,3 +111,4 @@ object Upsert {
 	field lastEventTimestamp ! timestamp
 	field data ! any
 }
+