goclaw/internal/store/sqlitestore/cron_crud.go

//go:build sqlite || sqliteonly

package sqlitestore

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/google/uuid"

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

func (s *SQLiteCronStore) AddJob(ctx context.Context, name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*store.CronJob, error) {
	if schedule.TZ == "" && schedule.Kind == "cron" && s.defaultTZ != "" {
		schedule.TZ = s.defaultTZ
	}
	if schedule.TZ != "" {
		if _, err := time.LoadLocation(schedule.TZ); err != nil {
			return nil, fmt.Errorf("invalid timezone: %s", schedule.TZ)
		}
	}

	payload := store.CronPayload{
		Kind:             "agent_turn",
		Message:          message,
		CredentialUserID: store.ExplicitCredentialUserIDFromContext(ctx),
	}
	payloadJSON, _ := json.Marshal(payload)

	id := uuid.Must(uuid.NewV7())
	now := time.Now()
	scheduleKind := schedule.Kind
	deleteAfterRun := schedule.Kind == "at"

	var cronExpr, tz *string
	var runAt *time.Time
	if schedule.Expr != "" {
		cronExpr = &schedule.Expr
	}
	if schedule.AtMS != nil {
		t := time.UnixMilli(*schedule.AtMS)
		runAt = &t
	}
	if schedule.TZ != "" {
		tz = &schedule.TZ
	}

	var agentUUID *uuid.UUID
	if agentID != "" {
		if aid, err := uuid.Parse(agentID); err == nil {
			agentUUID = &aid
		}
	}

	var userIDPtr *string
	if userID != "" {
		userIDPtr = &userID
	}

	var intervalMS *int64
	if schedule.EveryMS != nil {
		intervalMS = schedule.EveryMS
	}

	nextRun := computeNextRun(&schedule, now, s.defaultTZ)

	_, err := s.db.ExecContext(ctx,
		`INSERT INTO cron_jobs (id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
		 interval_ms, payload, delete_after_run, deliver, deliver_channel, deliver_to, wake_heartbeat, next_run_at, created_at, updated_at)
		 VALUES (?,?,?,?,?,1,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
		id, tenantIDForInsert(ctx), agentUUID, userIDPtr, name, scheduleKind, cronExpr, runAt, tz,
		intervalMS, payloadJSON, deleteAfterRun, deliver, channel, to, false, nextRun, now, now,
	)
	if err != nil {
		return nil, fmt.Errorf("create cron job: %w", err)
	}

	s.InvalidateCache()

	// Read back with a tenant-consistent context. The INSERT above uses
	// tenantIDForInsert(ctx), which falls back to MasterTenantID when the
	// caller has no tenant in context. scanJob rejects a nil-tenant context
	// with "tenant_id required", so reading back under the same tenant the row
	// was inserted with keeps insert and readback symmetric (matches pg store).
	readCtx := store.WithTenantID(ctx, tenantIDForInsert(ctx))
	job, ok := s.GetJob(readCtx, id.String())
	if !ok || job == nil {
		return nil, fmt.Errorf("cron job %s created but readback failed", id)
	}
	return job, nil
}

func (s *SQLiteCronStore) GetJob(ctx context.Context, jobID string) (*store.CronJob, bool) {
	id, err := uuid.Parse(jobID)
	if err != nil {
		return nil, false
	}
	job, err := s.scanJob(ctx, id)
	if err != nil {
		return nil, false
	}
	return job, true
}

func (s *SQLiteCronStore) ListJobs(ctx context.Context, includeDisabled bool, agentID, userID string) []store.CronJob {
	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 1=1`

	var args []any

	if !includeDisabled {
		q += " AND enabled = ?"
		args = append(args, true)
	}
	if agentID != "" {
		if aid, err := uuid.Parse(agentID); err == nil {
			q += " AND agent_id = ?"
			args = append(args, aid)
		}
	}
	if userID != "" {
		q += " AND user_id = ?"
		args = append(args, userID)
	}

	clause, targs, tErr := scopeClause(ctx)
	if tErr != nil {
		slog.Warn("cron.ListJobs: tenant context missing, returning empty (fail-closed)", "error", tErr)
		return nil
	}
	if clause != "" {
		q += clause
		args = append(args, targs...)
	}

	q += " ORDER BY created_at DESC"

	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil
	}
	defer rows.Close()

	var result []store.CronJob
	for rows.Next() {
		job, err := scanCronRow(rows)
		if err != nil {
			continue
		}
		result = append(result, *job)
	}
	if err := rows.Err(); err != nil {
		slog.Warn("cron: list jobs iteration error", "error", err)
	}
	return result
}

func (s *SQLiteCronStore) RemoveJob(ctx context.Context, jobID string) error {
	id, err := uuid.Parse(jobID)
	if err != nil {
		return fmt.Errorf("invalid job ID: %s", jobID)
	}

	q := "DELETE FROM cron_jobs WHERE id = ?"
	args := []any{id}

	if !store.IsCrossTenant(ctx) {
		tid := store.TenantIDFromContext(ctx)
		if tid == uuid.Nil {
			return fmt.Errorf("tenant_id required")
		}
		q += " AND tenant_id = ?"
		args = append(args, tid)
	}

	res, err := s.db.ExecContext(ctx, q, args...)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("job not found")
	}
	s.InvalidateCache()
	return nil
}

func (s *SQLiteCronStore) EnableJob(ctx context.Context, jobID string, enabled bool) error {
	id, err := uuid.Parse(jobID)
	if err != nil {
		return fmt.Errorf("invalid job ID: %s", jobID)
	}

	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	current, err := s.lockCronJobForMutation(ctx, tx, id, false)
	if err != nil {
		return err
	}

	now := time.Now()
	nextRun, err := store.NextRunForToggle(&current.Schedule, enabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
	if err != nil {
		return err
	}

	if err := execCronJobUpdateTx(ctx, tx, id, map[string]any{
		"enabled":     enabled,
		"next_run_at": nextRun,
		"updated_at":  now,
	}); err != nil {
		return err
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	s.InvalidateCache()
	return nil
}

func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch store.CronJobPatch) (*store.CronJob, error) {
	id, err := uuid.Parse(jobID)
	if err != nil {
		return nil, fmt.Errorf("invalid job ID: %s", jobID)
	}

	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	current, err := s.lockCronJobForMutation(ctx, tx, id, true)
	if err != nil {
		return nil, err
	}

	now := time.Now()
	updates := make(map[string]any)
	effectiveEnabled := current.Enabled
	if patch.Enabled != nil {
		effectiveEnabled = *patch.Enabled
		updates["enabled"] = effectiveEnabled
	}

	if patch.Name != "" {
		updates["name"] = patch.Name
	}
	if patch.DeleteAfterRun != nil {
		updates["delete_after_run"] = *patch.DeleteAfterRun
	}
	if patch.AgentID != nil {
		if *patch.AgentID == "" {
			updates["agent_id"] = nil
		} else if aid, parseErr := uuid.Parse(*patch.AgentID); parseErr == nil {
			updates["agent_id"] = aid
		}
	}

	if patch.Schedule != nil {
		merged := store.MergeCronSchedule(current.Schedule, patch.Schedule)
		if err := store.ValidateCronSchedule(&merged); err != nil {
			return nil, err
		}

		store.ApplyCronScheduleUpdates(updates, merged)

		nextRun, err := store.NextRunForSchedule(&merged, effectiveEnabled, now, s.defaultTZ)
		if err != nil {
			return nil, err
		}
		updates["next_run_at"] = nextRun
	} else if patch.Enabled != nil {
		nextRun, err := store.NextRunForToggle(&current.Schedule, effectiveEnabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
		if err != nil {
			return nil, err
		}
		updates["next_run_at"] = nextRun
	}

	if patch.Stateless != nil {
		updates["stateless"] = *patch.Stateless
	}
	if patch.Deliver != nil {
		updates["deliver"] = *patch.Deliver
	}
	if patch.DeliverChannel != nil {
		updates["deliver_channel"] = *patch.DeliverChannel
	}
	if patch.DeliverTo != nil {
		updates["deliver_to"] = *patch.DeliverTo
	}
	if patch.WakeHeartbeat != nil {
		updates["wake_heartbeat"] = *patch.WakeHeartbeat
	}
	if patch.ProviderID != nil {
		updates["provider_id"] = *patch.ProviderID
	}
	if patch.Model != nil {
		updates["model"] = *patch.Model
	}

	if patch.Message != "" || patch.Command != nil {
		payload := current.Payload
		if patch.Message != "" {
			payload.Message = patch.Message
		}
		if patch.Command != nil {
			payload.Kind = store.CronPayloadKindCommand
			payload.Command = patch.Command
		}
		mergedPayload, err := json.Marshal(payload)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
		}
		updates["payload"] = mergedPayload
	}

	updates["updated_at"] = now
	if err := execCronJobUpdateTx(ctx, tx, id, updates); err != nil {
		return nil, err
	}

	if err := tx.Commit(); err != nil {
		return nil, err
	}

	s.InvalidateCache()
	job, _ := s.scanJob(ctx, id)
	return job, nil
}

func (s *SQLiteCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id uuid.UUID, loadPayload bool) (*store.CronJobMutableState, error) {
	q := `SELECT enabled, schedule_kind, cron_expression, run_at, timezone, interval_ms, next_run_at, payload
		FROM cron_jobs WHERE id = ?`
	args := []any{id}

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

	var (
		state        store.CronJobMutableState
		scheduleKind string
		cronExpr     *string
		runAt        nullSqliteTime
		tz           *string
		intervalMS   *int64
		nextRunAt    nullSqliteTime
		payloadJSON  []byte
	)

	if err := tx.QueryRowContext(ctx, q, args...).Scan(
		&state.Enabled,
		&scheduleKind,
		&cronExpr,
		&runAt,
		&tz,
		&intervalMS,
		&nextRunAt,
		&payloadJSON,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, store.ErrCronJobNotFound
		}
		return nil, err
	}

	state.Schedule = store.CronSchedule{Kind: scheduleKind}
	if cronExpr != nil {
		state.Schedule.Expr = *cronExpr
	}
	if runAt.Valid {
		ms := runAt.Time.UnixMilli()
		state.Schedule.AtMS = &ms
	}
	if tz != nil {
		state.Schedule.TZ = *tz
	}
	if intervalMS != nil {
		state.Schedule.EveryMS = intervalMS
	}
	if nextRunAt.Valid {
		next := nextRunAt.Time
		state.NextRunAt = &next
	}

	if loadPayload && len(payloadJSON) > 0 {
		if err := json.Unmarshal(payloadJSON, &state.Payload); err != nil {
			return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", id, err)
		}
	}

	return &state, nil
}

func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error {
	if len(updates) == 0 {
		return nil
	}

	var (
		setClauses []string
		args       []any
	)
	for _, col := range store.SortedUpdateColumns(updates) {
		if !validColumnName.MatchString(col) {
			return fmt.Errorf("invalid column name: %q", col)
		}
		setClauses = append(setClauses, col+" = ?")
		args = append(args, sqliteVal(updates[col]))
	}

	args = append(args, id)
	q := "UPDATE cron_jobs SET " + strings.Join(setClauses, ", ") + " WHERE id = ?"
	if !store.IsCrossTenant(ctx) {
		tid := store.TenantIDFromContext(ctx)
		if tid == uuid.Nil {
			return fmt.Errorf("tenant_id required")
		}
		args = append(args, tid)
		q += " AND tenant_id = ?"
	}

	res, err := tx.ExecContext(ctx, q, args...)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return store.ErrCronJobNotFound
	}
	return nil
}