goclaw/internal/store/pg/cron_scan.go

package pg

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

// scanJob fetches a single cron job by ID with tenant filtering.
func (s *PGCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob, error) {
	q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
		 interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
		 next_run_at, last_run_at, last_status, last_error,
		 created_at, updated_at, provider_id, model FROM cron_jobs WHERE id = $1`
	args := []any{id}

	if !store.IsCrossTenant(ctx) {
		tid := store.TenantIDFromContext(ctx)
		if tid == uuid.Nil {
			return nil, fmt.Errorf("tenant_id required")
		}
		q += fmt.Sprintf(" AND tenant_id = $%d", len(args)+1)
		args = append(args, tid)
	}

	row := s.db.QueryRowContext(ctx, q, args...)
	return scanCronSingleRow(row)
}

// --- Scan helpers ---

type cronRowScanner interface {
	Scan(dest ...any) error
}

func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
	var id uuid.UUID
	var tenantID uuid.UUID
	var agentID *uuid.UUID
	var userID *string
	var name, scheduleKind string
	var enabled, deleteAfterRun bool
	var stateless, deliver, wakeHeartbeat bool
	var deliverChannel, deliverTo string
	var cronExpr, tz, lastStatus, lastError *string
	var runAt, nextRunAt, lastRunAt *time.Time
	var intervalMS *int64
	var payloadJSON []byte
	var createdAt, updatedAt time.Time
	var providerID *uuid.UUID
	var model *string

	err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
		&intervalMS, &payloadJSON, &deleteAfterRun, &stateless, &deliver, &deliverChannel, &deliverTo, &wakeHeartbeat,
		&nextRunAt, &lastRunAt, &lastStatus, &lastError,
		&createdAt, &updatedAt, &providerID, &model)
	if err != nil {
		return nil, err
	}

	var payload store.CronPayload
	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(),
		TenantID: tenantID,
		Name:     name,
		Enabled:  enabled,
		Schedule: store.CronSchedule{
			Kind: scheduleKind,
		},
		Payload:        payload,
		CreatedAtMS:    createdAt.UnixMilli(),
		UpdatedAtMS:    updatedAt.UnixMilli(),
		DeleteAfterRun: deleteAfterRun,
		Stateless:      stateless,
		Deliver:        deliver,
		DeliverChannel: deliverChannel,
		DeliverTo:      deliverTo,
		WakeHeartbeat:  wakeHeartbeat,
	}

	job.ProviderID = providerID
	job.Model = model

	if agentID != nil {
		job.AgentID = agentID.String()
	}
	if userID != nil {
		job.UserID = *userID
	}
	if cronExpr != nil {
		job.Schedule.Expr = *cronExpr
	}
	if runAt != nil {
		ms := runAt.UnixMilli()
		job.Schedule.AtMS = &ms
	}
	if intervalMS != nil {
		job.Schedule.EveryMS = intervalMS
	}
	if tz != nil {
		job.Schedule.TZ = *tz
	}
	if nextRunAt != nil {
		ms := nextRunAt.UnixMilli()
		job.State.NextRunAtMS = &ms
	}
	if lastRunAt != nil {
		ms := lastRunAt.UnixMilli()
		job.State.LastRunAtMS = &ms
	}
	if lastStatus != nil {
		job.State.LastStatus = *lastStatus
	}
	if lastError != nil {
		job.State.LastError = *lastError
	}

	return job, nil
}

func scanCronSingleRow(row *sql.Row) (*store.CronJob, error) {
	return scanCronRow(row)
}

// --- Helpers ---

// computeNextRun calculates the next run time for a schedule.
// defaultTZ is the gateway-level fallback IANA timezone used when the
// schedule itself does not specify a timezone (existing jobs with TZ = NULL).
func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ string) *time.Time {
	return store.ComputeNextRun(schedule, now, defaultTZ)
}