Checkpoint.

lib/debezium/converters/date.go

@@ -0,0 +1,26 @@
+package converters
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/artie-labs/transfer/lib/typing"
+	"github.com/artie-labs/transfer/lib/typing/ext"
+)
+
+type Date struct{}
+
+func (Date) ToKindDetails() typing.KindDetails {
+	return typing.NewKindDetailsFromTemplate(typing.ETime, ext.DateKindType)
+}
+
+func (Date) Convert(value any) (any, error) {
+	valueInt64, isOk := value.(int64)
+	if !isOk {
+		return nil, fmt.Errorf("expected int64 got '%v' with type %T", value, value)
+	}
+
+	unix := time.UnixMilli(0).In(time.UTC) // 1970-01-01
+	// Represents the number of days since the epoch.
+	return ext.NewExtendedTime(unix.AddDate(0, 0, int(valueInt64)), ext.DateKindType, ""), nil
+}

lib/debezium/converters/date_test.go

@@ -0,0 +1,18 @@
+package converters
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDate_Convert(t *testing.T) {
+	{
+		// Invalid data type
+		_, err := Date{}.Convert("invalid")
+		assert.ErrorContains(t, err, "expected int64 got 'invalid' with type string")
+	}
+	{
+
+	}
+}

lib/debezium/schema.go

@@ -92,6 +92,8 @@ func (f Field) ToValueConverter() converters.ValueConverter {
 		return converters.Geometry{}
 	case JSON:
 		return converters.JSON{}
+	case Date, DateKafkaConnect:
+		return converters.Date{}
 	}
 
 	return nil
@@ -110,8 +112,6 @@ func (f Field) ToKindDetails() typing.KindDetails {
 	switch f.DebeziumType {
 	case Timestamp, MicroTimestamp, NanoTimestamp, DateTimeKafkaConnect:
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.DateTimeKindType)
-	case Date, DateKafkaConnect:
-		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.DateKindType)
 	case Time, MicroTime, NanoTime, TimeKafkaConnect:
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimeKindType)
 	case KafkaDecimalType:

lib/debezium/types.go

@@ -170,10 +170,6 @@ func FromDebeziumTypeToTime(supportedType SupportedDebeziumType, val int64) (*ex
 	case NanoTimestamp:
 		// Represents the number of nanoseconds past the epoch, and does not include timezone information.
 		extTime = ext.NewExtendedTime(time.UnixMicro(val/1_000).In(time.UTC), ext.DateTimeKindType, time.RFC3339Nano)
-	case Date, DateKafkaConnect:
-		unix := time.UnixMilli(0).In(time.UTC) // 1970-01-01
-		// Represents the number of days since the epoch.
-		extTime = ext.NewExtendedTime(unix.AddDate(0, 0, int(val)), ext.DateKindType, "")
 	case Time, TimeKafkaConnect:
 		// Represents the number of milliseconds past midnight, and does not include timezone information.
 		extTime = ext.NewExtendedTime(time.UnixMilli(val).In(time.UTC), ext.TimeKindType, "")