feat(queue/sql): add subscriber with partition leasing (#23)

## Summary
## Why?
Need Subscriber implementation with partition leasing to enable
distributed message processing from SQL queue.

## What?
- Subscriber discovers partitions and acquires leases for distributed
work
- Partition leasing ensures each partition processed by exactly one
worker
- Automatic failover via stale lease stealing
- Visibility timeout with automatic retry on worker crash
- DLQ integration for messages exceeding retry limit
- Thread-safe delivery acknowledgment
- Moved Store in separate package to avoid circular dependency with
mocks

## Test Plan
bazel test

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -4,15 +4,18 @@ go_library(
     name = "sql",
     srcs = [
         "config.go",
+        "errors.go",
         "mock_stores.go",
         "publisher.go",
         "stores.go",
+        "subscriber.go",
         "validation.go",
     ],
     importpath = "github.com/uber/submitqueue/extensions/queue/sql",
     visibility = ["//visibility:public"],
     deps = [
         "//entities/queue",
+        "//extensions/queue",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_mock//gomock",
         "@org_uber_go_zap//:zap",
@@ -24,6 +27,7 @@ go_test(
     srcs = [
         "config_test.go",
         "publisher_test.go",
+        "subscriber_test.go",
     ],
     embed = [":sql"],
     deps = [

extensions/queue/sql/errors.go

@@ -0,0 +1,12 @@
+package sql
+
+import "fmt"
+
+// ErrAlreadyAcknowledged is returned when attempting to ack/nack a delivery that was already processed
+type ErrAlreadyAcknowledged struct {
+	DeliveryID string
+}
+
+func (e *ErrAlreadyAcknowledged) Error() string {
+	return fmt.Sprintf("delivery %s already acknowledged or nacked", e.DeliveryID)
+}

extensions/queue/sql/mock_stores.go

@@ -15,13 +15,13 @@ type publisher struct {
 	config       Config
 	logger       *zap.SugaredLogger
 	metrics      tally.Scope
-	messageStore MessageStore
+	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 {
+func NewPublisher(config Config, logger *zap.SugaredLogger, metrics tally.Scope, messageStore messageStore) *publisher {
 	return &publisher{
 		config:       config,
 		logger:       logger,