golangci-lint 1.62

.github/workflows/golang-lint.yml

@@ -29,6 +29,6 @@ jobs:
       - name: golangci-lint
         uses: golangci/golangci-lint-action@971e284b6050e8a5849b72094c50ab08da042db8 # v6
         with:
-          version: v1.61
+          version: v1.62
           working-directory: ./flow
           args: --timeout=10m

flow/.golangci.yml

@@ -15,6 +15,7 @@ linters:
     - gofumpt
     - gosec
     - gosimple
+    - iface
     - ineffassign
     - intrange
     - lll
@@ -25,6 +26,7 @@ linters:
     - nonamedreturns
     - perfsprint
     - prealloc
+    - recvcheck
     - rowserrcheck
     - staticcheck
     - stylecheck
@@ -66,6 +68,11 @@ linters-settings:
     enable-all: true
     disable:
       - shadow
+  iface:
+    enable:
+      - identical
+      - unused
+      - opaque
   stylecheck:
     checks:
       - all

flow/connectors/core.go

@@ -463,6 +463,8 @@ var (
 	_ CDCSyncConnector = &connclickhouse.ClickHouseConnector{}
 	_ CDCSyncConnector = &connelasticsearch.ElasticsearchConnector{}
 
+	_ CDCSyncPgConnector = &connpostgres.PostgresConnector{}
+
 	_ CDCNormalizeConnector = &connpostgres.PostgresConnector{}
 	_ CDCNormalizeConnector = &connbigquery.BigQueryConnector{}
 	_ CDCNormalizeConnector = &connsnowflake.SnowflakeConnector{}

flow/connectors/postgres/qrep_query_executor_test.go

@@ -186,8 +186,7 @@ func TestAllDataTypes(t *testing.T) {
 	expectedBytea := []byte("bytea")
 	require.Equal(t, expectedBytea, record[6].Value(), "expected 'bytea'")
 
-	expectedJSON := `{"key":"value"}`
-	require.Equal(t, expectedJSON, record[7].Value(), "expected '{\"key\":\"value\"}'")
+	require.JSONEq(t, `{"key":"value"}`, record[7].Value().(string), "expected '{\"key\":\"value\"}'")
 
 	actualUUID := record[8].Value().(uuid.UUID)
 	require.Equal(t, savedUUID[:], actualUUID[:], "expected savedUUID: %v", savedUUID)

flow/connectors/sql/query_executor.go

@@ -19,6 +19,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/model/qvalue"
 )
 
+//nolint:iface
 type SQLQueryExecutor interface {
 	ConnectionActive(context.Context) error
 	Close() error

flow/connectors/utils/aws.go

@@ -103,7 +103,7 @@ func (r *ConfigBasedAWSCredentialsProvider) Retrieve(ctx context.Context) (AWSCr
 	}, nil
 }
 
-func NewConfigBasedAWSCredentialsProvider(config aws.Config) AWSCredentialsProvider {
+func NewConfigBasedAWSCredentialsProvider(config aws.Config) *ConfigBasedAWSCredentialsProvider {
 	return &ConfigBasedAWSCredentialsProvider{config: config}
 }
 
@@ -132,7 +132,7 @@ func (s *StaticAWSCredentialsProvider) GetEndpointURL() string {
 	return ""
 }
 
-func NewStaticAWSCredentialsProvider(credentials AWSCredentials, region string) AWSCredentialsProvider {
+func NewStaticAWSCredentialsProvider(credentials AWSCredentials, region string) *StaticAWSCredentialsProvider {
 	return &StaticAWSCredentialsProvider{
 		credentials: credentials,
 		region:      region,

flow/e2e/bigquery/peer_flow_bq_test.go

@@ -55,7 +55,7 @@ func (s PeerFlowE2ETestSuiteBQ) attachSuffix(input string) string {
 	return fmt.Sprintf("%s_%s", input, s.bqSuffix)
 }
 
-func (s *PeerFlowE2ETestSuiteBQ) checkPeerdbColumns(dstQualified string, softDelete bool) error {
+func (s PeerFlowE2ETestSuiteBQ) checkPeerdbColumns(dstQualified string, softDelete bool) error {
 	qualifiedTableName := fmt.Sprintf("`%s.%s`", s.bqHelper.Config.DatasetId, dstQualified)
 	selector := "`_PEERDB_SYNCED_AT`"
 	if softDelete {

flow/e2e/sqlserver/sqlserver_helper.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"strconv"
 
-	peersql "github.com/PeerDB-io/peer-flow/connectors/sql"
 	connsqlserver "github.com/PeerDB-io/peer-flow/connectors/sqlserver"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/model/qvalue"
@@ -16,7 +15,7 @@ import (
 type SQLServerHelper struct {
 	config *protos.SqlServerConfig
 
-	E          peersql.SQLQueryExecutor
+	E          *connsqlserver.SQLServerConnector
 	SchemaName string
 	tables     []string
 }

flow/shared/telemetry/incidentio_message_sender.go

@@ -30,10 +30,6 @@ type IncidentIoResponse struct {
 }
 
 type IncidentIoMessageSender struct {
-	Sender
-}
-
-type IncidentIoMessageSenderImpl struct {
 	http   *http.Client
 	config IncidentIoMessageSenderConfig
 }
@@ -43,7 +39,7 @@ type IncidentIoMessageSenderConfig struct {
 	Token string
 }
 
-func (i *IncidentIoMessageSenderImpl) SendMessage(
+func (i *IncidentIoMessageSender) SendMessage(
 	ctx context.Context,
 	subject string,
 	body string,
@@ -117,12 +113,12 @@ func (i *IncidentIoMessageSenderImpl) SendMessage(
 	return incidentResponse.Status, nil
 }
 
-func NewIncidentIoMessageSender(_ context.Context, config IncidentIoMessageSenderConfig) (Sender, error) {
+func NewIncidentIoMessageSender(_ context.Context, config IncidentIoMessageSenderConfig) (*IncidentIoMessageSender, error) {
 	client := &http.Client{
 		Timeout: time.Second * 5,
 	}
 
-	return &IncidentIoMessageSenderImpl{
+	return &IncidentIoMessageSender{
 		config: config,
 		http:   client,
 	}, nil

flow/shared/telemetry/incidentio_message_sender_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestIncidentIoMessageSenderImpl_SendMessage(t *testing.T) {
+func TestIncidentIoMessageSender_SendMessage(t *testing.T) {
 	tests := []struct {
 		serverResponse     IncidentIoResponse
 		name               string
@@ -168,7 +168,7 @@ func TestIncidentIoMessageSenderImpl_SendMessage(t *testing.T) {
 				URL:   fakeIncidentIoServer.URL,
 				Token: "test-token",
 			}
-			sender := &IncidentIoMessageSenderImpl{
+			sender := &IncidentIoMessageSender{
 				http:   &http.Client{Timeout: time.Second * 5},
 				config: config,
 			}

flow/shared/telemetry/interface.go → flow/shared/telemetry/sender.go

@@ -8,6 +8,11 @@ type Sender interface {
 	SendMessage(ctx context.Context, subject string, body string, attributes Attributes) (string, error)
 }
 
+var (
+	_ Sender = &IncidentIoMessageSender{}
+	_ Sender = &SNSMessageSender{}
+)
+
 type Attributes struct {
 	DeploymentUID string
 	Type          string

flow/shared/telemetry/sns_message_sender.go

@@ -15,11 +15,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/shared/aws_common"
 )
 
-type SNSMessageSender interface {
-	Sender
-}
-
-type SNSMessageSenderImpl struct {
+type SNSMessageSender struct {
 	client *sns.Client
 	topic  string
 }
@@ -28,7 +24,7 @@ type SNSMessageSenderConfig struct {
 	Topic string `json:"topic"`
 }
 
-func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string, body string, attributes Attributes) (string, error) {
+func (s *SNSMessageSender) SendMessage(ctx context.Context, subject string, body string, attributes Attributes) (string, error) {
 	activityInfo := activity.Info{}
 	if activity.IsActivity(ctx) {
 		activityInfo = activity.GetInfo(ctx)
@@ -92,14 +88,14 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 	return *publish.MessageId, nil
 }
 
-func NewSNSMessageSenderWithNewClient(ctx context.Context, config *SNSMessageSenderConfig) (SNSMessageSender, error) {
+func NewSNSMessageSenderWithNewClient(ctx context.Context, config *SNSMessageSenderConfig) (*SNSMessageSender, error) {
 	// Topic Region must match client region
 	region := strings.Split(strings.TrimPrefix(config.Topic, "arn:aws:sns:"), ":")[0]
 	client, err := newSnsClient(ctx, &region)
 	if err != nil {
 		return nil, err
 	}
-	return &SNSMessageSenderImpl{
+	return &SNSMessageSender{
 		client: client,
 		topic:  config.Topic,
 	}, nil