feat(queue/sql): add publisher implementation

MODULE.bazel

@@ -39,6 +39,7 @@ use_repo(
     "org_golang_google_grpc",
     "org_golang_google_protobuf",
     "org_uber_go_fx",
+    "org_uber_go_mock",
     "org_uber_go_yarpc",
     "org_uber_go_zap",
 )

extensions/queue/sql/BUILD.bazel

@@ -2,17 +2,37 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "sql",
-    srcs = ["config.go"],
+    srcs = [
+        "config.go",
+        "mock_stores.go",
+        "publisher.go",
+        "stores.go",
+        "validation.go",
+    ],
     importpath = "github.com/uber/submitqueue/extensions/queue/sql",
     visibility = ["//visibility:public"],
+    deps = [
+        "//entities/queue",
+        "@com_github_uber_go_tally_v4//:tally",
+        "@org_uber_go_mock//gomock",
+        "@org_uber_go_zap//:zap",
+    ],
 )
 
 go_test(
     name = "sql_test",
-    srcs = ["config_test.go"],
+    srcs = [
+        "config_test.go",
+        "publisher_test.go",
+    ],
     embed = [":sql"],
     deps = [
+        "//entities/queue",
+        "//extensions/queue",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally_v4//:tally",
+        "@org_uber_go_mock//gomock",
+        "@org_uber_go_zap//zaptest",
     ],
 )

extensions/queue/sql/config.go

@@ -5,14 +5,6 @@ import (
 	"time"
 )
 
-const (
-	// Fixed table names for single-table design
-	MessagesTableName        = "queue_messages"
-	PartitionLeasesTableName = "queue_partition_leases"
-	OffsetsTableName         = "queue_offsets"
-	DLQTableName             = "queue_dlq"
-)
-
 // Config holds configuration for SQL-based queue.
 // DB connection, logger, and metrics are passed separately to NewFactory.
 type Config struct {

extensions/queue/sql/publisher.go

@@ -0,0 +1,72 @@
+package sql
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/uber-go/tally/v4"
+	"go.uber.org/zap"
+
+	"github.com/uber/submitqueue/entities/queue"
+)
+
+type publisher struct {
+	config       Config
+	logger       *zap.SugaredLogger
+	metrics      tally.Scope
+	messageStore MessageStore
+	mu           sync.RWMutex
+	closed       bool
+}
+
+// NewPublisher creates a publisher with the given configuration and dependencies
+func NewPublisher(config Config, logger *zap.SugaredLogger, metrics tally.Scope, messageStore MessageStore) *publisher {
+	return &publisher{
+		config:       config,
+		logger:       logger,
+		metrics:      metrics,
+		messageStore: messageStore,
+	}
+}
+
+// Publish sends a message to the specified topic
+func (p *publisher) Publish(ctx context.Context, topic string, message queue.Message) error {
+	// Check if closed (under lock)
+	p.mu.RLock()
+	closed := p.closed
+	p.mu.RUnlock()
+
+	if closed {
+		p.logger.Errorw("publish failure: publisher is closed", "topic", topic)
+		return fmt.Errorf("publisher is closed")
+	}
+
+	// Validate topic name (SQL-safe)
+	if err := validateTopicName(topic); err != nil {
+		p.logger.Errorw("publish failure: invalid topic name", "topic", topic, "error", err)
+		p.metrics.Tagged(map[string]string{"topic": topic}).Counter("publish_errors").Inc(1)
+		return fmt.Errorf("publish invalid topic name: %w", err)
+	}
+
+	if err := p.messageStore.Insert(ctx, topic, []queue.Message{message}); err != nil {
+		p.metrics.Tagged(map[string]string{"topic": topic}).Counter("publish_errors").Inc(1)
+		p.logger.Errorw("publish failure: message store insert error", "topic", topic, "error", err)
+		return fmt.Errorf("publish message store insert error: %w", err)
+	}
+
+	p.metrics.Tagged(map[string]string{"topic": topic}).Counter("messages_published").Inc(1)
+	p.logger.Debugw("published message", "topic", topic, "message_id", message.ID)
+
+	return nil
+}
+
+// Close gracefully shuts down the publisher
+func (p *publisher) Close() error {
+	p.mu.Lock()
+	p.closed = true
+	p.mu.Unlock()
+
+	p.logger.Info("publisher closed")
+	return nil
+}