feat(queue/sql): add publisher with gomock testing

To implement the SQL-based queue publisher with proper test coverage using gomock for mocking dependencies, avoiding manual mock implementations.

- Add MessageStore interface for message table operations
- Implement Publisher using MessageStore for inserting messages
- Add gomock-based tests for Publisher (no manual mocks)
- Generate gomock mocks for MessageStore interface
- Add README.md with instructions for generating mocks

Test Plan:
- Unit tests pass: `bazel test //extensions/queue/sql:sql_test`
- All tests use gomock exclusively, no manual mocks
- `make build` and `make test` pass successfully

Author: behinddwalls

MODULE.bazel

@@ -36,6 +36,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,36 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "sql",
-    srcs = ["config.go"],
+    srcs = [
+        "config.go",
+        "publisher.go",
+        "stores.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_zap//:zap",
+    ],
 )
 
 go_test(
     name = "sql_test",
-    srcs = ["config_test.go"],
+    srcs = [
+        "config_test.go",
+        "publisher_test.go",
+    ],
+    data = ["//extensions/queue/sql/schema"],
     embed = [":sql"],
     deps = [
+        "//entities/queue",
+        "//extensions/queue",
+        "//extensions/queue/sql/mocks",
         "@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/mocks/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "mocks",
+    srcs = ["mock_stores.go"],
+    importpath = "github.com/uber/submitqueue/extensions/queue/sql/mocks",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//entities/queue",
+        "@org_uber_go_mock//gomock",
+    ],
+)

extensions/queue/sql/mocks/mock_stores.go

@@ -0,0 +1,77 @@
+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
+}
+
+// Publish sends a message to the specified topic
+func (p *publisher) Publish(ctx context.Context, topic string, message queue.Message) error {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	if p.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 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 err
+	}
+
+	p.metrics.Tagged(map[string]string{"topic": topic}).Counter("messages_published").Inc(1)
+	p.logger.Infow("published message", "topic", topic, "message_id", message.ID)
+
+	return nil
+}
+
+// Close gracefully shuts down the publisher
+func (p *publisher) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.closed = true
+	p.logger.Info("publisher closed")
+	return nil
+}
+
+// validateTopicName ensures topic name is safe for use as SQL table name
+func validateTopicName(topic string) error {
+	if topic == "" {
+		return fmt.Errorf("topic name cannot be empty")
+	}
+	if len(topic) > 255 {
+		return fmt.Errorf("topic name too long (max 255 characters)")
+	}
+	// Only allow lowercase letters, numbers, and underscores
+	for _, c := range topic {
+		if !((c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_') {
+			return fmt.Errorf("topic name must contain only lowercase letters, numbers, and underscores")
+		}
+	}
+	return nil
+}