fix(staging): revert to concrete return types, add interface conformance checks

Author: whites11

flow/connectors/utils/staging.go

@@ -5,6 +5,11 @@ import (
 	"io"
 )
 
+var (
+	_ StagingStore = (*S3StagingStore)(nil)
+	_ StagingStore = (*GCSStagingStore)(nil)
+)
+
 // StagingStore abstracts cloud storage used by PeerDB's ClickHouse connector
 // for staging Avro files. Files are written by PeerDB and read by ClickHouse
 // via table functions (s3(), url(), etc.).

flow/connectors/utils/staging_gcs.go

@@ -26,7 +26,7 @@ type GCSStagingStore struct {
 	fullPath string
 }
 
-func NewGCSStagingStore(ctx context.Context, bucketPath string) (StagingStore, error) {
+func NewGCSStagingStore(ctx context.Context, bucketPath string) (*GCSStagingStore, error) {
 	// bucketPath is "gcs://bucket/prefix" or just "bucket/prefix"
 	path := strings.TrimPrefix(bucketPath, "gcs://")
 	path = strings.TrimPrefix(path, "gs://")

flow/connectors/utils/staging_s3.go

@@ -24,7 +24,7 @@ type S3StagingStore struct {
 	fullPath string // original "s3://bucket/prefix" for logging
 }
 
-func NewS3StagingStore(bucketPath string, creds AWSCredentialsProvider) (StagingStore, error) {
+func NewS3StagingStore(bucketPath string, creds AWSCredentialsProvider) (*S3StagingStore, error) {
 	s3o, err := NewS3BucketAndPrefix(bucketPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse S3 bucket path: %w", err)

flow/connectors/utils/staging_test.go

@@ -19,6 +19,8 @@ func TestNewS3StagingStore(t *testing.T) {
 
 	store, err := NewS3StagingStore("s3://my-bucket/some/prefix", creds)
 	require.NoError(t, err)
+	require.Equal(t, "my-bucket", store.Bucket())
+	require.Equal(t, "some/prefix", store.Prefix())
 	require.Equal(t, "some/prefix", store.KeyPrefix())
 	require.Equal(t, "s3://my-bucket/some/prefix", store.BucketPath())
 }
@@ -92,6 +94,6 @@ func TestNewS3StagingStoreNoBucket(t *testing.T) {
 
 	store, err := NewS3StagingStore("s3://bucket-only", creds)
 	require.NoError(t, err)
-	require.Equal(t, "s3://bucket-only", store.BucketPath())
-	require.Empty(t, store.KeyPrefix())
+	require.Equal(t, "bucket-only", store.Bucket())
+	require.Empty(t, store.Prefix())
 }