Fix timestamp/time type handling in diff and repair pipelines

* Fix timestamp/time type handling in diff and repair pipelines

- Split combined timestamp/timestamptz case in ConvertToPgxType; return
  pgxv5type.Timestamp for "timestamp without time zone" so pgx sends
  the correct wire type
- Add time/timetz support: serialize pgxv5type.Time in diff output,
  parse time strings back to pgxv5type.Time in ConvertToPgxType, and
  pass timetz as string (unregistered in pgx v5 type map)
- Exclude "time with time zone" from IsKnownScalarType so the diff
  layer casts it to ::TEXT (avoids scan error on unregistered type)
- Wrap time.Time in pgxv5type.Timestamp in repair's convertValueForType
  for timestamp-without-tz columns
- Add unit tests for all new type handling paths
- Add integration test exercising full diff→repair round-trip for all
  four temporal types

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Add bounds checks and fractional truncation to parseTimeToMicroseconds

Reject out-of-range hours (>23), minutes (>59), and seconds (>59).
Truncate fractional second digits beyond 6 to microsecond precision.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: mason-sharp

internal/consistency/diff/table_diff.go

@@ -33,6 +33,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgtype"
 	"github.com/jackc/pgx/v5"
+	pgxv5type "github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/pgedge/ace/db/queries"
 	auth "github.com/pgedge/ace/internal/infra/db"
@@ -624,6 +625,19 @@ func (t *TableDiffTask) fetchRows(nodeName string, r Range) ([]types.OrderedMap,
 					// nil caught above
 					processedVal = fmt.Sprintf("%x-%x-%x-%x-%x",
 						v[0:4], v[4:6], v[6:8], v[8:10], v[10:16])
+				case pgxv5type.Time: // pgx/v5 returns "time without time zone" as pgtype.Time
+					if v.Valid {
+						usec := v.Microseconds
+						hours := usec / 3_600_000_000
+						usec -= hours * 3_600_000_000
+						minutes := usec / 60_000_000
+						usec -= minutes * 60_000_000
+						seconds := usec / 1_000_000
+						usec -= seconds * 1_000_000
+						processedVal = fmt.Sprintf("%02d:%02d:%02d.%06d", hours, minutes, seconds, usec)
+					} else {
+						processedVal = nil
+					}
 				case time.Time:
 					processedVal = v
 				case string:

internal/consistency/repair/table_repair.go

@@ -27,6 +27,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pglogrepl"
 	"github.com/jackc/pgx/v5"
+	pgxv5type "github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/pgedge/ace/db/queries"
 	planner "github.com/pgedge/ace/internal/consistency/repair/plan"
@@ -1227,6 +1228,13 @@ func convertValueForType(val any, colType string) (any, error) {
 	}
 
 	if tVal, ok := val.(time.Time); ok {
+		// For "timestamp without time zone" columns, wrap in pgxv5type.Timestamp
+		// so pgx sends the value as "timestamp" instead of "timestamptz". This prevents
+		// PostgreSQL from applying a session-timezone conversion.
+		lower := strings.ToLower(colType)
+		if lower == "timestamp without time zone" || (strings.HasPrefix(lower, "timestamp") && !strings.Contains(lower, "with time zone")) {
+			return pgxv5type.Timestamp{Time: tVal, Valid: true}, nil
+		}
 		return tVal, nil
 	}
 

internal/consistency/repair/table_repair_test.go

@@ -0,0 +1,256 @@
+// ///////////////////////////////////////////////////////////////////////////
+//
+// # ACE - Active Consistency Engine
+//
+// Copyright (C) 2023 - 2026, pgEdge (https://www.pgedge.com/)
+//
+// This software is released under the PostgreSQL License:
+// https://opensource.org/license/postgresql
+//
+// ///////////////////////////////////////////////////////////////////////////
+
+package repair
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	pgxv5type "github.com/jackc/pgx/v5/pgtype"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConvertValueForType_TimestampWrapsInPgxTimestamp(t *testing.T) {
+	// When a time.Time value is passed for a "timestamp without time zone"
+	// column, convertValueForType must wrap it in pgxv5type.Timestamp so
+	// pgx sends the value as "timestamp" instead of "timestamptz". This prevents PostgreSQL from
+	// applying a session-timezone conversion that shifts the value.
+	ts := time.Date(2024, 6, 15, 10, 30, 0, 0, time.UTC)
+
+	tests := []struct {
+		name    string
+		colType string
+		wantPgx bool // true → expect pgxv5type.Timestamp; false → expect time.Time
+	}{
+		{"timestamp without time zone", "timestamp without time zone", true},
+		{"TIMESTAMP WITHOUT TIME ZONE (upper)", "TIMESTAMP WITHOUT TIME ZONE", true},
+		{"timestamp with time zone", "timestamp with time zone", false},
+		{"timestamptz via format_type", "timestamp with time zone", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := convertValueForType(ts, tt.colType)
+			require.NoError(t, err)
+
+			if tt.wantPgx {
+				pgTS, ok := got.(pgxv5type.Timestamp)
+				require.True(t, ok, "expected pgxv5type.Timestamp, got %T", got)
+				require.True(t, pgTS.Valid)
+				require.True(t, pgTS.Time.Equal(ts))
+			} else {
+				tv, ok := got.(time.Time)
+				require.True(t, ok, "expected time.Time, got %T", got)
+				require.True(t, tv.Equal(ts))
+			}
+		})
+	}
+}
+
+func TestConvertValueForType_StringTimestamp(t *testing.T) {
+	// Values coming from diff JSON are strings. They should be parsed and
+	// wrapped in pgxv5type.Timestamp for "timestamp without time zone".
+	val, err := convertValueForType("2024-06-15T10:30:00Z", "timestamp without time zone")
+	require.NoError(t, err)
+
+	pgTS, ok := val.(pgxv5type.Timestamp)
+	require.True(t, ok, "expected pgxv5type.Timestamp, got %T", val)
+	require.True(t, pgTS.Valid)
+
+	expected := time.Date(2024, 6, 15, 10, 30, 0, 0, time.UTC)
+	require.True(t, pgTS.Time.Equal(expected))
+}
+
+func TestConvertValueForType_StringTimestamptz(t *testing.T) {
+	// Values for "timestamp with time zone" columns should remain time.Time.
+	val, err := convertValueForType("2024-06-15T10:30:00Z", "timestamp with time zone")
+	require.NoError(t, err)
+
+	tv, ok := val.(time.Time)
+	require.True(t, ok, "expected time.Time, got %T", val)
+
+	expected := time.Date(2024, 6, 15, 10, 30, 0, 0, time.UTC)
+	require.True(t, tv.Equal(expected))
+}
+
+func TestConvertValueForType_Nil(t *testing.T) {
+	val, err := convertValueForType(nil, "timestamp without time zone")
+	require.NoError(t, err)
+	require.Nil(t, val)
+}
+
+func TestBuildFixNullsBatchSQL_TimestampType(t *testing.T) {
+	// Verify that buildFixNullsBatchSQL generates valid SQL for timestamp
+	// columns and that values are properly converted.
+	task := &TableRepairTask{}
+	task.Schema = "public"
+	task.Table = "test_table"
+	task.Key = []string{"id"}
+
+	colTypes := map[string]string{
+		"id":     "integer",
+		"ts_col": "timestamp without time zone",
+	}
+
+	batch := []*nullUpdate{
+		{
+			pkValues: []any{float64(1)},
+			columns:  map[string]any{"ts_col": "2024-06-15T10:30:00Z"},
+		},
+	}
+
+	sql, args, err := task.buildFixNullsBatchSQL("ts_col", "timestamp without time zone", batch, colTypes)
+	require.NoError(t, err)
+
+	// The SQL should contain the correct type cast
+	require.True(t, strings.Contains(sql, "::timestamp without time zone"),
+		"expected SQL to contain '::timestamp without time zone', got: %s", sql)
+
+	// The args should contain the converted values
+	require.Len(t, args, 2) // 1 PK + 1 value
+
+	// The timestamp value should be pgxv5type.Timestamp, not time.Time
+	pgTS, ok := args[1].(pgxv5type.Timestamp)
+	require.True(t, ok, "expected timestamp arg to be pgxv5type.Timestamp, got %T", args[1])
+	require.True(t, pgTS.Valid)
+
+	expected := time.Date(2024, 6, 15, 10, 30, 0, 0, time.UTC)
+	require.True(t, pgTS.Time.Equal(expected))
+}
+
+func TestBuildFixNullsBatchSQL_TimestamptzType(t *testing.T) {
+	// Verify that timestamptz columns use time.Time (not pgxv5type.Timestamp).
+	task := &TableRepairTask{}
+	task.Schema = "public"
+	task.Table = "test_table"
+	task.Key = []string{"id"}
+
+	colTypes := map[string]string{
+		"id":      "integer",
+		"tstz_col": "timestamp with time zone",
+	}
+
+	batch := []*nullUpdate{
+		{
+			pkValues: []any{float64(1)},
+			columns:  map[string]any{"tstz_col": "2024-06-15T10:30:00Z"},
+		},
+	}
+
+	sql, args, err := task.buildFixNullsBatchSQL("tstz_col", "timestamp with time zone", batch, colTypes)
+	require.NoError(t, err)
+
+	require.True(t, strings.Contains(sql, "::timestamp with time zone"),
+		"expected SQL to contain '::timestamp with time zone', got: %s", sql)
+
+	require.Len(t, args, 2)
+
+	// The timestamptz value should be a plain time.Time
+	tv, ok := args[1].(time.Time)
+	require.True(t, ok, "expected timestamptz arg to be time.Time, got %T", args[1])
+
+	expected := time.Date(2024, 6, 15, 10, 30, 0, 0, time.UTC)
+	require.True(t, tv.Equal(expected))
+}
+
+func TestConvertValueForType_StringTime(t *testing.T) {
+	// Values from diff JSON for "time without time zone" columns should be
+	// parsed into pgxv5type.Time via ConvertToPgxType.
+	val, err := convertValueForType("10:30:00.000000", "time without time zone")
+	require.NoError(t, err)
+
+	tv, ok := val.(pgxv5type.Time)
+	require.True(t, ok, "expected pgxv5type.Time, got %T", val)
+	require.True(t, tv.Valid)
+
+	expectedUsec := int64(10*3_600_000_000 + 30*60_000_000)
+	require.Equal(t, expectedUsec, tv.Microseconds)
+}
+
+func TestConvertValueForType_StringTimetz(t *testing.T) {
+	// Values for "time with time zone" columns should remain as strings.
+	val, err := convertValueForType("10:30:00-05:00", "time with time zone")
+	require.NoError(t, err)
+
+	s, ok := val.(string)
+	require.True(t, ok, "expected string, got %T", val)
+	require.Equal(t, "10:30:00-05:00", s)
+}
+
+func TestBuildFixNullsBatchSQL_TimeType(t *testing.T) {
+	// Verify that buildFixNullsBatchSQL generates valid SQL for time columns
+	// and that values are properly converted to pgxv5type.Time.
+	task := &TableRepairTask{}
+	task.Schema = "public"
+	task.Table = "test_table"
+	task.Key = []string{"id"}
+
+	colTypes := map[string]string{
+		"id":       "integer",
+		"time_col": "time without time zone",
+	}
+
+	batch := []*nullUpdate{
+		{
+			pkValues: []any{float64(1)},
+			columns:  map[string]any{"time_col": "14:30:00.000000"},
+		},
+	}
+
+	sql, args, err := task.buildFixNullsBatchSQL("time_col", "time without time zone", batch, colTypes)
+	require.NoError(t, err)
+
+	require.True(t, strings.Contains(sql, "::time without time zone"),
+		"expected SQL to contain '::time without time zone', got: %s", sql)
+
+	require.Len(t, args, 2) // 1 PK + 1 value
+
+	tv, ok := args[1].(pgxv5type.Time)
+	require.True(t, ok, "expected time arg to be pgxv5type.Time, got %T", args[1])
+	require.True(t, tv.Valid)
+
+	expectedUsec := int64(14*3_600_000_000 + 30*60_000_000)
+	require.Equal(t, expectedUsec, tv.Microseconds)
+}
+
+func TestBuildFixNullsBatchSQL_TimetzType(t *testing.T) {
+	// Verify that timetz columns pass through as strings.
+	task := &TableRepairTask{}
+	task.Schema = "public"
+	task.Table = "test_table"
+	task.Key = []string{"id"}
+
+	colTypes := map[string]string{
+		"id":         "integer",
+		"timetz_col": "time with time zone",
+	}
+
+	batch := []*nullUpdate{
+		{
+			pkValues: []any{float64(1)},
+			columns:  map[string]any{"timetz_col": "14:30:00-05:00"},
+		},
+	}
+
+	sql, args, err := task.buildFixNullsBatchSQL("timetz_col", "time with time zone", batch, colTypes)
+	require.NoError(t, err)
+
+	require.True(t, strings.Contains(sql, "::time with time zone"),
+		"expected SQL to contain '::time with time zone', got: %s", sql)
+
+	require.Len(t, args, 2)
+
+	s, ok := args[1].(string)
+	require.True(t, ok, "expected timetz arg to be string, got %T", args[1])
+	require.Equal(t, "14:30:00-05:00", s)
+}