fix: check errors in cron store Scan and Unmarshal to prevent data corruption (#422)

* fix: check errors in cron store Scan and Unmarshal to prevent data corruption

Three unchecked errors in the cron store could silently corrupt data:

1. cron_update.go:44 — QueryRowContext().Scan() error ignored when
   fetching current schedule during partial updates. If DB fails,
   all schedule fields default to zero values, corrupting schedule_kind
   to empty string and bypassing all validation.

2. cron_update.go:167 — json.Unmarshal error ignored when parsing
   existing payload for merge. Corrupted JSON leaves payload zero-valued,
   wiping all existing fields (Message, Deliver, Channel, To) when
   only one field was being updated.

3. cron_scan.go:63 — json.Unmarshal error ignored when scanning cron
   job rows. Malformed payload JSON silently returns zero-valued payload,
   masking data corruption.

All three now return descriptive errors instead of proceeding with
corrupted data. NULL payload (valid case) is handled separately.

* fix: add NULL payload guard and check Marshal error in cron update

- Add len(payloadJSON) > 0 guard before Unmarshal in UpdateJob payload
  merge path, consistent with the same guard in scanCronRow. Without
  this, NULL payload (valid for new jobs) would return a false error
  instead of proceeding with the merge.

- Check json.Marshal error instead of discarding it, consistent with
  the error-checking theme of the parent commit.

---------

Co-authored-by: viettranx <viettranx@gmail.com>

internal/store/pg/cron_scan.go

@@ -60,7 +60,11 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 	}
 
 	var payload store.CronPayload
-	json.Unmarshal(payloadJSON, &payload)
+	if len(payloadJSON) > 0 {
+		if err := json.Unmarshal(payloadJSON, &payload); err != nil {
+			return nil, fmt.Errorf("failed to parse cron job payload: %w", err)
+		}
+	}
 
 	job := &store.CronJob{
 		ID:       id.String(),

internal/store/pg/cron_scan_test.go

@@ -0,0 +1,169 @@
+package pg
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+)
+
+// mockRowScanner satisfies cronRowScanner for unit-testing scanCronRow
+// without hitting a real database.
+type mockRowScanner struct {
+	values []any
+	err    error
+}
+
+func (m *mockRowScanner) Scan(dest ...any) error {
+	if m.err != nil {
+		return m.err
+	}
+	// Copy pre-set values into dest pointers.
+	for i, d := range dest {
+		if i >= len(m.values) {
+			break
+		}
+		switch dp := d.(type) {
+		case *uuid.UUID:
+			if v, ok := m.values[i].(uuid.UUID); ok {
+				*dp = v
+			}
+		case **uuid.UUID:
+			if v, ok := m.values[i].(*uuid.UUID); ok {
+				*dp = v
+			}
+		case *string:
+			if v, ok := m.values[i].(string); ok {
+				*dp = v
+			}
+		case **string:
+			if v, ok := m.values[i].(*string); ok {
+				*dp = v
+			}
+		case *bool:
+			if v, ok := m.values[i].(bool); ok {
+				*dp = v
+			}
+		case **time.Time:
+			if v, ok := m.values[i].(*time.Time); ok {
+				*dp = v
+			}
+		case **int64:
+			if v, ok := m.values[i].(*int64); ok {
+				*dp = v
+			}
+		case *[]byte:
+			if v, ok := m.values[i].([]byte); ok {
+				*dp = v
+			}
+		case *time.Time:
+			if v, ok := m.values[i].(time.Time); ok {
+				*dp = v
+			}
+		}
+	}
+	return nil
+}
+
+// validCronRow returns mock scanner values for a valid cron job row.
+func validCronRow(payloadJSON []byte) []any {
+	id := uuid.New()
+	tenantID := uuid.New()
+	now := time.Now()
+	expr := "*/5 * * * *"
+	tz := "UTC"
+	return []any{
+		id,                  // id
+		tenantID,            // tenant_id
+		(*uuid.UUID)(nil),   // agent_id
+		(*string)(nil),      // user_id
+		"test-job",          // name
+		true,                // enabled
+		"cron",              // schedule_kind
+		&expr,               // cron_expression
+		(*time.Time)(nil),   // run_at
+		&tz,                 // timezone
+		(*int64)(nil),       // interval_ms
+		payloadJSON,         // payload
+		false,               // delete_after_run
+		(*time.Time)(nil),   // next_run_at
+		(*time.Time)(nil),   // last_run_at
+		(*string)(nil),      // last_status
+		(*string)(nil),      // last_error
+		now,                 // created_at
+		now,                 // updated_at
+	}
+}
+
+// TestScanCronRow_MalformedPayloadJSON verifies that malformed payload JSON
+// returns an error instead of silently returning a zero-valued payload.
+func TestScanCronRow_MalformedPayloadJSON(t *testing.T) {
+	malformedJSON := []byte(`{invalid json!!!}`)
+	row := &mockRowScanner{values: validCronRow(malformedJSON)}
+
+	_, err := scanCronRow(row)
+	if err == nil {
+		t.Fatal("expected error for malformed payload JSON, got nil")
+	}
+	if !strings.Contains(err.Error(), "payload") {
+		t.Errorf("error should mention payload, got: %v", err)
+	}
+}
+
+// TestScanCronRow_ValidPayload verifies correct behavior with valid JSON.
+func TestScanCronRow_ValidPayload(t *testing.T) {
+	payload := store.CronPayload{
+		Kind:    "message",
+		Message: "hello world",
+		Deliver: true,
+		Channel: "telegram",
+		To:      "user123",
+	}
+	payloadJSON, _ := json.Marshal(payload)
+	row := &mockRowScanner{values: validCronRow(payloadJSON)}
+
+	job, err := scanCronRow(row)
+	if err != nil {
+		t.Fatalf("scanCronRow returned unexpected error: %v", err)
+	}
+
+	if job.Payload.Message != "hello world" {
+		t.Errorf("expected Message 'hello world', got %q", job.Payload.Message)
+	}
+	if job.Payload.Channel != "telegram" {
+		t.Errorf("expected Channel 'telegram', got %q", job.Payload.Channel)
+	}
+	if job.Payload.To != "user123" {
+		t.Errorf("expected To 'user123', got %q", job.Payload.To)
+	}
+}
+
+// TestScanCronRow_NullPayload verifies that NULL payload (empty []byte) is handled.
+func TestScanCronRow_NullPayload(t *testing.T) {
+	row := &mockRowScanner{values: validCronRow(nil)}
+
+	job, err := scanCronRow(row)
+	if err != nil {
+		t.Fatalf("scanCronRow returned unexpected error: %v", err)
+	}
+
+	// NULL payload should result in zero-valued payload (not an error).
+	if job.Payload.Message != "" {
+		t.Errorf("expected empty Message for NULL payload, got %q", job.Payload.Message)
+	}
+}
+
+// TestScanCronRow_ScanError verifies that a DB scan error is propagated.
+func TestScanCronRow_ScanError(t *testing.T) {
+	row := &mockRowScanner{err: errors.New("connection reset")}
+
+	_, err := scanCronRow(row)
+	if err == nil {
+		t.Fatal("expected error from scanCronRow when scan fails")
+	}
+}

internal/store/pg/cron_update.go

@@ -41,10 +41,12 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
 		var curExpr, curTZ *string
 		var curIntervalMS *int64
 		var curRunAt *time.Time
-		s.db.QueryRowContext(ctx,
+		if err := s.db.QueryRowContext(ctx,
 			"SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = $1"+tenantSuffix,
 			append([]any{id}, tenantArg...)...,
-		).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt)
+		).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt); err != nil {
+			return nil, fmt.Errorf("failed to fetch current schedule for job %s: %w", jobID, err)
+		}
 
 		// Resolve the effective schedule kind
 		newKind := patch.Schedule.Kind
@@ -164,7 +166,11 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
 		var payloadJSON []byte
 		if scanErr := s.db.QueryRowContext(ctx, "SELECT payload FROM cron_jobs WHERE id = $1"+tenantSuffix, append([]any{id}, tenantArg...)...).Scan(&payloadJSON); scanErr == nil {
 			var payload store.CronPayload
-			json.Unmarshal(payloadJSON, &payload)
+			if len(payloadJSON) > 0 {
+				if err := json.Unmarshal(payloadJSON, &payload); err != nil {
+					return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", jobID, err)
+				}
+			}
 
 			if patch.Message != "" {
 				payload.Message = patch.Message
@@ -182,7 +188,10 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
 				payload.WakeHeartbeat = *patch.WakeHeartbeat
 			}
 
-			merged, _ := json.Marshal(payload)
+			merged, err := json.Marshal(payload)
+			if err != nil {
+				return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+			}
 			updates["payload"] = merged
 		}
 	}

internal/store/pg/cron_update_test.go

@@ -0,0 +1,103 @@
+package pg
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+)
+
+// TestScheduleKindCorruption_LogicPath verifies that empty schedule_kind
+// (which would happen if the pre-fetch Scan failed and error was ignored)
+// falls through validation with no error — proving why the Scan error
+// MUST be checked (as it now is in the fix).
+//
+// The fix: UpdateJob now returns an error when the schedule-fetch Scan fails,
+// so this corrupted logic path is no longer reachable.
+func TestScheduleKindCorruption_LogicPath(t *testing.T) {
+	// When Scan fails, curKind is "" (zero value).
+	// With partial patch (no kind specified), newKind inherits "".
+	curKind := "" // simulates failed Scan
+	patchKind := ""
+
+	newKind := patchKind
+	if newKind == "" {
+		newKind = curKind
+	}
+
+	// Verify the corruption: empty kind bypasses all validation cases
+	merged := store.CronSchedule{Kind: newKind}
+	validationRan := false
+	switch merged.Kind {
+	case "cron", "every", "at":
+		validationRan = true
+	}
+
+	if validationRan {
+		t.Error("expected validation to be skipped for empty kind")
+	}
+	if newKind != "" {
+		t.Error("expected empty newKind from failed Scan")
+	}
+
+	// This proves the corruption path exists.
+	// The fix (returning error from Scan) prevents reaching this path.
+	t.Log("Confirmed: empty schedule_kind bypasses validation — fix prevents this by checking Scan error")
+}
+
+// TestPayloadUnmarshal_CorruptedJSON verifies that corrupted JSON in the
+// payload column causes json.Unmarshal to fail, which would wipe existing
+// fields if the error is ignored — proving why the error MUST be checked.
+//
+// The fix: UpdateJob now returns an error when payload Unmarshal fails,
+// so existing fields are never overwritten with zero values.
+func TestPayloadUnmarshal_CorruptedJSON(t *testing.T) {
+	corruptedJSON := []byte(`{"kind":"message","message":"daily report","deliver":true,CORRUPT`)
+
+	var payload store.CronPayload
+	err := json.Unmarshal(corruptedJSON, &payload)
+
+	// Prove that Unmarshal fails
+	if err == nil {
+		t.Fatal("expected json.Unmarshal to fail on corrupted JSON")
+	}
+
+	// Prove that payload is zero-valued after failed Unmarshal
+	if payload.Message != "" {
+		t.Error("expected empty Message after failed Unmarshal")
+	}
+	if payload.Deliver {
+		t.Error("expected false Deliver after failed Unmarshal")
+	}
+
+	// If we ignored this error and applied patch fields, we'd wipe existing data.
+	// The fix returns an error instead.
+	t.Log("Confirmed: failed Unmarshal leaves zero payload — fix prevents wipe by returning error")
+}
+
+// TestPayloadUnmarshal_ValidJSON verifies normal payload round-trip works.
+func TestPayloadUnmarshal_ValidJSON(t *testing.T) {
+	original := store.CronPayload{
+		Kind:    "message",
+		Message: "daily report",
+		Deliver: true,
+		Channel: "telegram",
+		To:      "admin-group",
+	}
+	data, _ := json.Marshal(original)
+
+	var decoded store.CronPayload
+	if err := json.Unmarshal(data, &decoded); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if decoded.Message != original.Message {
+		t.Errorf("Message mismatch: got %q, want %q", decoded.Message, original.Message)
+	}
+	if decoded.Channel != original.Channel {
+		t.Errorf("Channel mismatch: got %q, want %q", decoded.Channel, original.Channel)
+	}
+	if decoded.To != original.To {
+		t.Errorf("To mismatch: got %q, want %q", decoded.To, original.To)
+	}
+}