feat(queue/sql): add data access layer stores

MODULE.bazel

@@ -30,6 +30,7 @@ go_deps.from_file(go_mod = "//:go.mod")
 # All *direct* Go dependencies of the module have to be listed explicitly
 use_repo(
     go_deps,
+    "com_github_data_dog_go_sqlmock",
     "com_github_go_sql_driver_mysql",
     "com_github_gogo_protobuf",
     "com_github_stretchr_testify",

extensions/queue/sql/BUILD.bazel

@@ -4,8 +4,12 @@ go_library(
     name = "sql",
     srcs = [
         "config.go",
+        "constants.go",
         "errors.go",
+        "message_store.go",
         "mock_stores.go",
+        "offset_store.go",
+        "partition_lease_store.go",
         "publisher.go",
         "stores.go",
         "subscriber.go",
@@ -26,13 +30,17 @@ go_test(
     name = "sql_test",
     srcs = [
         "config_test.go",
+        "message_store_test.go",
+        "offset_store_test.go",
+        "partition_lease_store_test.go",
         "publisher_test.go",
         "subscriber_test.go",
     ],
     embed = [":sql"],
     deps = [
         "//entities/queue",
         "//extensions/queue",
+        "@com_github_data_dog_go_sqlmock//:go-sqlmock",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_uber_go_tally_v4//:tally",

extensions/queue/sql/config_test.go

@@ -30,7 +30,6 @@ func TestConfigValidation(t *testing.T) {
 		name        string
 		config      Config
 		expectError bool
-		errorMsg    string
 	}{
 		{
 			name:        "valid config",
@@ -50,7 +49,6 @@ func TestConfigValidation(t *testing.T) {
 				Retry:                DefaultConfig("dummy", "dummy").Retry,
 			},
 			expectError: true,
-			errorMsg:    "ConsumerGroup is required",
 		},
 		{
 			name: "empty worker ID",
@@ -65,7 +63,6 @@ func TestConfigValidation(t *testing.T) {
 				Retry:                DefaultConfig("dummy", "dummy").Retry,
 			},
 			expectError: true,
-			errorMsg:    "WorkerID is required",
 		},
 		{
 			name: "invalid poll interval",
@@ -80,7 +77,6 @@ func TestConfigValidation(t *testing.T) {
 				Retry:                DefaultConfig("dummy", "dummy").Retry,
 			},
 			expectError: true,
-			errorMsg:    "PollInterval must be positive",
 		},
 		{
 			name: "invalid batch size",
@@ -95,7 +91,6 @@ func TestConfigValidation(t *testing.T) {
 				Retry:                DefaultConfig("dummy", "dummy").Retry,
 			},
 			expectError: true,
-			errorMsg:    "BatchSize must be positive",
 		},
 	}
 
@@ -104,7 +99,6 @@ func TestConfigValidation(t *testing.T) {
 			err := tt.config.Validate()
 			if tt.expectError {
 				require.Error(t, err)
-				assert.Contains(t, err.Error(), tt.errorMsg)
 			} else {
 				require.NoError(t, err)
 			}

extensions/queue/sql/constants.go

@@ -0,0 +1,18 @@
+package sql
+
+// Common constants for frequently repeated strings across stores
+
+const (
+	// Tag key (used in every Tagged() call)
+	tagErrorType = "error_type"
+
+	// Common log field names (used extensively across all stores)
+	logTopic        = "topic"
+	logPartitionKey = "partition_key"
+	logMessageID    = "message_id"
+	logError        = "error"
+
+	// Error types used across multiple methods/stores
+	errorBeginTx = "begin_transaction"
+	errorCommit  = "commit"
+)