package invocation

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	bindinglang "github.com/tidbcloud/AgentTeam/internal/workflow/binding"
	domain "github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
)

const (
	scheduleCompletedReason   = "schedule_completed"
	scheduleBlockedRetryDelay = 30 * time.Second
)

var _ domain.ScheduledMaterializer = (*Store)(nil)

// ListDueScheduleCandidates returns at most one due occurrence per active
// schedule source. next_occurrence_at is the source's durable earliest
// unaccepted occurrence, so a single row per source also provides bounded
// catch-up: a recurring source can be selected again only by a later pass.
func (store *Store) ListDueScheduleCandidates(ctx context.Context, input domain.ListDueScheduleInput) ([]domain.ScheduleCandidate, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	if input.Now.IsZero() {
		return nil, storeerr.BadRequest("schedule list now is required")
	}
	now := input.Now.UTC().Truncate(time.Microsecond)
	limit := input.Limit
	if limit <= 0 {
		limit = defaultListLimit
	}
	if limit > maxListLimit {
		limit = maxListLimit
	}

	rows, err := store.db.Query(ctx, `
SELECT source.id, source.workspace_id, source.binding_id,
       source.next_occurrence_at, source.activation_number,
       placement.workspace_group_id, placement.state, placement.placement_epoch
FROM workflow_binding_sources source
JOIN workflow_bindings binding
  ON binding.workspace_id = source.workspace_id
 AND binding.id = source.binding_id
JOIN workspace_placements placement
  ON placement.workspace_id = source.workspace_id
WHERE source.source_kind = $1
  AND source.status = 'active'
  AND source.next_occurrence_at IS NOT NULL
  AND source.next_occurrence_at <= $2
  AND (source.materialization_retry_at IS NULL OR source.materialization_retry_at <= $2)
  AND binding.state = 'enabled'
  AND binding.active_source_id = source.id
  AND binding.active_revision_id = source.binding_revision_id
  AND placement.workspace_group_id = COALESCE(NULLIF($3, ''), placement.workspace_group_id)
  AND placement.state IN ($4, $5, $6)
-- NULL retry_at means this due occurrence has never been deferred. Always
-- drain that unseen set before retrying blocked rows, regardless of how old a
-- blocked occurrence is. Otherwise a prefix larger than batch*backoff/poll
-- can become eligible again before the scanner ever reaches healthy rows.
ORDER BY source.materialization_retry_at ASC,
         source.next_occurrence_at ASC, source.id ASC
LIMIT $7
`, SourceKindSchedule, now, store.schedulerWorkspaceGroupID(),
		schedulePlacementStateActive, schedulePlacementStateDraining, schedulePlacementStateMoving, limit)
	if err != nil {
		return nil, fmt.Errorf("list due Workflow schedule candidates: %w", err)
	}
	defer rows.Close()

	candidates := make([]domain.ScheduleCandidate, 0, limit)
	for rows.Next() {
		var candidate scheduleCandidate
		if err := rows.Scan(
			&candidate.sourceID, &candidate.workspaceID, &candidate.bindingID,
			&candidate.occurrenceAt, &candidate.activation,
			&candidate.workspaceGroup, &candidate.placementState, &candidate.placementEpoch,
		); err != nil {
			return nil, fmt.Errorf("scan due Workflow schedule candidate: %w", err)
		}
		candidate.occurrenceAt = candidate.occurrenceAt.UTC().Truncate(time.Microsecond)
		if _, err := requireStoreScheduleCandidate(candidate); err != nil {
			return nil, fmt.Errorf("validate due Workflow schedule candidate: %w", err)
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate due Workflow schedule candidates: %w", err)
	}
	return candidates, nil
}

// MaterializeScheduled implements invocation.ScheduledMaterializer.
func (store *Store) MaterializeScheduled(ctx context.Context, candidate domain.ScheduleCandidate, now time.Time) (domain.ScheduleMaterializeResult, error) {
	return store.AcceptScheduled(ctx, candidate, now)
}

// AcceptScheduled revalidates and accepts one store-issued schedule candidate
// in a single transaction. Lock order is deliberately explicit and stable:
// placement, Binding, Source, immutable BindingRevision, Plan aggregate,
// immutable PlanVersion, owner member, Invocation+Run, then source cursor
// mutation.
func (store *Store) AcceptScheduled(ctx context.Context, opaque domain.ScheduleCandidate, now time.Time) (domain.ScheduleMaterializeResult, error) {
	if err := store.Validate(); err != nil {
		return domain.ScheduleMaterializeResult{}, err
	}
	candidate, err := requireStoreScheduleCandidate(opaque)
	if err != nil {
		return domain.ScheduleMaterializeResult{}, err
	}
	if now.IsZero() {
		return domain.ScheduleMaterializeResult{}, storeerr.BadRequest("schedule materialization now is required")
	}
	now = now.UTC().Truncate(time.Microsecond)

	var materialized domain.ScheduleMaterializeResult
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		placementCurrent, err := store.lockScheduleCandidatePlacement(ctx, tx, candidate)
		if err != nil {
			return err
		}
		if !placementCurrent {
			materialized.Disposition = domain.ScheduleMaterializeFenced
			return nil
		}

		binding, found, err := lockScheduleBinding(ctx, tx, candidate.workspaceID, candidate.bindingID)
		if err != nil {
			return err
		}
		if !found {
			materialized.Disposition = domain.ScheduleMaterializeStale
			return nil
		}

		source, found, err := lockScheduleSource(ctx, tx, candidate.workspaceID, candidate.sourceID)
		if err != nil {
			return err
		}
		if !found || source.BindingID != binding.ID || source.Activation != candidate.activation {
			materialized.Disposition = domain.ScheduleMaterializeStale
			return nil
		}

		revision, found, err := lockScheduleRevision(ctx, tx, candidate.workspaceID, binding.ID, source.BindingRevisionID)
		if err != nil {
			return err
		}
		if !found {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}
		plan, found, err := lockSchedulePlan(ctx, tx, candidate.workspaceID, revision.PlanID)
		if err != nil {
			return err
		}
		if !found {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}
		version, found, err := lockSchedulePlanVersion(ctx, tx, candidate.workspaceID, revision.PlanID, revision.PlanVersionID)
		if err != nil {
			return err
		}
		if !found {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}

		ownerAvailable, err := lockScheduleOwner(ctx, tx, binding)
		if err != nil {
			return err
		}

		occurrenceRef := domain.ScheduleOccurrenceRef(candidate.sourceID, candidate.occurrenceAt)
		existing, replay, err := lockInvocationByOccurrence(
			ctx, tx, candidate.workspaceID, candidate.sourceID, occurrenceRef,
		)
		if err != nil {
			return err
		}

		expected := ScheduleDeploymentFence{
			BindingSourceID: candidate.sourceID, BindingRevisionID: revision.ID,
			PlanVersionID: version.ID, RevisionDigest: revision.Digest,
		}
		if replay {
			// Replays deliberately use the immutable accepted payload and receipt
			// key. They therefore remain available after source lifecycle,
			// aggregate pointer, or owner membership changes.
			result, err := store.AcceptInTx(ctx, tx, AcceptParams{
				WorkspaceID: candidate.workspaceID, BindingID: candidate.bindingID,
				Expected: expected, SourceKind: SourceKindSchedule, OccurrenceAt: timePointer(candidate.occurrenceAt),
				RequestInputJSON: existing.InputJSON, InputJSON: existing.InputJSON,
				IdempotencyKey: existing.IdempotencyKey, Now: now,
			})
			if err != nil {
				return err
			}
			if !result.Replayed {
				return fmt.Errorf("replay scheduled Workflow Invocation %s returned a fresh acceptance", existing.ID)
			}
			materialized = domain.ScheduleMaterializeResult{
				Disposition: domain.ScheduleMaterializeReplayed,
				Acceptance:  result,
			}
			return nil
		}

		if !scheduleCandidateStillDue(candidate, now, binding, source, revision, plan, version) {
			materialized.Disposition = domain.ScheduleMaterializeStale
			return nil
		}
		if !ownerAvailable {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}

		inputJSON, schedule, err := normalizeScheduledInput(revision, version)
		if err != nil {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}
		nextOccurrence, err := bindinglang.NextOccurrenceAfter(schedule, candidate.occurrenceAt)
		if err != nil {
			if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
				return err
			}
			materialized.Disposition = domain.ScheduleMaterializeBlocked
			return nil
		}

		result, err := store.AcceptInTx(ctx, tx, AcceptParams{
			WorkspaceID: candidate.workspaceID, BindingID: candidate.bindingID,
			Expected: expected, SourceKind: SourceKindSchedule, OccurrenceAt: timePointer(candidate.occurrenceAt),
			RequestInputJSON: inputJSON, InputJSON: inputJSON,
			IdempotencyKey: occurrenceRef, Now: now,
		})
		if err != nil {
			switch {
			case errors.Is(err, ErrInactiveDeployment), errors.Is(err, ErrDeploymentChanged):
				materialized.Disposition = domain.ScheduleMaterializeStale
				return nil
			case errors.Is(err, ErrOwnerUnavailable), errors.Is(err, ErrPlanVersionUnavailable):
				if err := deferBlockedScheduleSource(ctx, tx, source, candidate.occurrenceAt, now); err != nil {
					return err
				}
				materialized.Disposition = domain.ScheduleMaterializeBlocked
				return nil
			default:
				return err
			}
		}

		// A replay discovered only inside AcceptInTx can happen if an older
		// schedule acceptor did not use this adapter. Because the locked source
		// was fully validated as still owning the occurrence, advancing it here
		// repairs the atomic cursor invariant without skipping another due item.
		if err := advanceScheduleSource(ctx, tx, source, candidate.occurrenceAt, nextOccurrence, now); err != nil {
			return err
		}
		disposition := domain.ScheduleMaterializeCreated
		if result.Replayed {
			disposition = domain.ScheduleMaterializeReplayed
		}
		materialized = domain.ScheduleMaterializeResult{Disposition: disposition, Acceptance: result}
		return nil
	})
	if err != nil {
		return domain.ScheduleMaterializeResult{}, err
	}
	return materialized, nil
}

type lockedScheduleBinding struct {
	ID               string
	WorkspaceID      string
	PlanID           string
	State            string
	ActiveRevisionID string
	ActiveSourceID   string
	OwnerUserID      string
	OwnerMemberID    string
}

type lockedScheduleSource struct {
	ID                string
	WorkspaceID       string
	BindingID         string
	BindingRevisionID string
	Activation        int64
	SourceKind        string
	Status            string
	Revision          int64
	NextOccurrenceAt  *time.Time
	RetryAt           *time.Time
}

type lockedScheduleRevision struct {
	ID            string
	WorkspaceID   string
	BindingID     string
	PlanID        string
	Revision      int64
	PlanVersionID string
	SourceKind    string
	ConfigJSON    []byte
	Digest        string
}

type lockedSchedulePlan struct {
	ID          string
	WorkspaceID string
}

type lockedSchedulePlanVersion struct {
	ID              string
	WorkspaceID     string
	PlanID          string
	ArtifactJSON    []byte
	CompilerVersion string
	Digest          string
}

func lockScheduleBinding(ctx context.Context, tx coredb.Tx, workspaceID string, bindingID string) (lockedScheduleBinding, bool, error) {
	var value lockedScheduleBinding
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id, plan_id, state,
       COALESCE(active_revision_id, ''), COALESCE(active_source_id, ''),
       owner_user_id, owner_member_id
FROM workflow_bindings
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, bindingID).Scan(
		&value.ID, &value.WorkspaceID, &value.PlanID, &value.State,
		&value.ActiveRevisionID, &value.ActiveSourceID,
		&value.OwnerUserID, &value.OwnerMemberID,
	)
	if coredb.IsNotFound(err) {
		return lockedScheduleBinding{}, false, nil
	}
	if err != nil {
		return lockedScheduleBinding{}, false, fmt.Errorf("lock scheduled Workflow Binding: %w", err)
	}
	return value, true, nil
}

func lockScheduleSource(ctx context.Context, tx coredb.Tx, workspaceID string, sourceID string) (lockedScheduleSource, bool, error) {
	var value lockedScheduleSource
	var next sql.NullTime
	var retry sql.NullTime
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id, binding_id, binding_revision_id,
       activation_number, source_kind, status, revision, next_occurrence_at,
       materialization_retry_at
FROM workflow_binding_sources
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, sourceID).Scan(
		&value.ID, &value.WorkspaceID, &value.BindingID, &value.BindingRevisionID,
		&value.Activation, &value.SourceKind, &value.Status, &value.Revision, &next, &retry,
	)
	if coredb.IsNotFound(err) {
		return lockedScheduleSource{}, false, nil
	}
	if err != nil {
		return lockedScheduleSource{}, false, fmt.Errorf("lock scheduled Workflow Binding source: %w", err)
	}
	if next.Valid {
		occurrence := next.Time.UTC().Truncate(time.Microsecond)
		value.NextOccurrenceAt = &occurrence
	}
	if retry.Valid {
		retryAt := retry.Time.UTC().Truncate(time.Microsecond)
		value.RetryAt = &retryAt
	}
	return value, true, nil
}

func lockScheduleRevision(ctx context.Context, tx coredb.Tx, workspaceID string, bindingID string, revisionID string) (lockedScheduleRevision, bool, error) {
	var value lockedScheduleRevision
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id, binding_id, plan_id, revision_number,
       plan_version_id, source_kind, config_json, digest
FROM workflow_binding_revisions
WHERE workspace_id = $1 AND binding_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, bindingID, revisionID).Scan(
		&value.ID, &value.WorkspaceID, &value.BindingID, &value.PlanID, &value.Revision,
		&value.PlanVersionID, &value.SourceKind, &value.ConfigJSON, &value.Digest,
	)
	if coredb.IsNotFound(err) {
		return lockedScheduleRevision{}, false, nil
	}
	if err != nil {
		return lockedScheduleRevision{}, false, fmt.Errorf("lock scheduled Workflow Binding revision: %w", err)
	}
	return value, true, nil
}

func lockSchedulePlan(ctx context.Context, tx coredb.Tx, workspaceID string, planID string) (lockedSchedulePlan, bool, error) {
	var value lockedSchedulePlan
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id
FROM workflow_plans
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, planID).Scan(&value.ID, &value.WorkspaceID)
	if coredb.IsNotFound(err) {
		return lockedSchedulePlan{}, false, nil
	}
	if err != nil {
		return lockedSchedulePlan{}, false, fmt.Errorf("lock scheduled Workflow Plan: %w", err)
	}
	return value, true, nil
}

func lockSchedulePlanVersion(ctx context.Context, tx coredb.Tx, workspaceID string, planID string, versionID string) (lockedSchedulePlanVersion, bool, error) {
	var value lockedSchedulePlanVersion
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id, plan_id, artifact, compiler_version, digest
FROM workflow_plan_versions
WHERE workspace_id = $1 AND plan_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, planID, versionID).Scan(
		&value.ID, &value.WorkspaceID, &value.PlanID,
		&value.ArtifactJSON, &value.CompilerVersion, &value.Digest,
	)
	if coredb.IsNotFound(err) {
		return lockedSchedulePlanVersion{}, false, nil
	}
	if err != nil {
		return lockedSchedulePlanVersion{}, false, fmt.Errorf("lock scheduled Workflow Plan version: %w", err)
	}
	return value, true, nil
}

func lockScheduleOwner(ctx context.Context, tx coredb.Tx, binding lockedScheduleBinding) (bool, error) {
	var memberType string
	var userID string
	var status string
	err := tx.QueryRow(ctx, `
SELECT type, COALESCE(user_id, ''), status
FROM members
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, binding.WorkspaceID, binding.OwnerMemberID).Scan(&memberType, &userID, &status)
	if coredb.IsNotFound(err) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("lock scheduled Workflow execution owner: %w", err)
	}
	return memberType == ActorTypeHuman && status == "active" && userID != "" && userID == binding.OwnerUserID, nil
}

func scheduleCandidateStillDue(
	candidate scheduleCandidate,
	now time.Time,
	binding lockedScheduleBinding,
	source lockedScheduleSource,
	revision lockedScheduleRevision,
	plan lockedSchedulePlan,
	version lockedSchedulePlanVersion,
) bool {
	return !candidate.occurrenceAt.After(now) &&
		binding.WorkspaceID == candidate.workspaceID && binding.ID == candidate.bindingID &&
		binding.State == "enabled" && binding.ActiveSourceID == source.ID && binding.ActiveRevisionID == revision.ID &&
		source.WorkspaceID == candidate.workspaceID && source.BindingID == binding.ID &&
		source.BindingRevisionID == revision.ID && source.Activation == candidate.activation &&
		source.SourceKind == SourceKindSchedule && source.Status == "active" && source.NextOccurrenceAt != nil &&
		source.NextOccurrenceAt.Equal(candidate.occurrenceAt) && (source.RetryAt == nil || !source.RetryAt.After(now)) &&
		revision.WorkspaceID == candidate.workspaceID && revision.BindingID == binding.ID && revision.PlanID == binding.PlanID &&
		revision.PlanVersionID == version.ID && revision.SourceKind == SourceKindSchedule && revision.Revision > 0 &&
		plan.WorkspaceID == candidate.workspaceID && plan.ID == binding.PlanID &&
		version.WorkspaceID == candidate.workspaceID && version.PlanID == plan.ID && version.ID != ""
}

func normalizeScheduledInput(revision lockedScheduleRevision, version lockedSchedulePlanVersion) ([]byte, bindinglang.ScheduleSource, error) {
	artifact, err := plan.DecodeArtifact(version.ArtifactJSON)
	if err != nil || artifact == nil {
		return nil, bindinglang.ScheduleSource{}, ErrPlanVersionUnavailable
	}
	if artifact.CompilerVersion != strings.TrimSpace(version.CompilerVersion) || artifact.Digest != strings.TrimSpace(version.Digest) {
		return nil, bindinglang.ScheduleSource{}, ErrPlanVersionUnavailable
	}
	spec, err := bindinglang.DecodeConfig(revision.PlanID, revision.PlanVersionID, revision.ConfigJSON)
	if err != nil {
		return nil, bindinglang.ScheduleSource{}, fmt.Errorf("decode scheduled Workflow Binding revision config: %w", err)
	}
	normalized, err := bindinglang.NormalizeRevision(spec, artifact.Contracts.Input)
	if err != nil {
		return nil, bindinglang.ScheduleSource{}, fmt.Errorf("verify scheduled Workflow Binding revision config: %w", err)
	}
	if normalized.Digest != strings.TrimSpace(revision.Digest) || !bindinglang.VerifyDigest(spec, revision.Digest) {
		return nil, bindinglang.ScheduleSource{}, fmt.Errorf("verify scheduled Workflow Binding revision config: persisted digest mismatch")
	}
	if normalized.Spec.Source.Type != bindinglang.SourceTypeSchedule || normalized.Spec.Source.Schedule == nil {
		return nil, bindinglang.ScheduleSource{}, ErrDeploymentChanged
	}
	input, err := bindinglang.NormalizeScheduleInvocationInput(artifact.Contracts.Input, normalized.Spec)
	if err != nil {
		return nil, bindinglang.ScheduleSource{}, fmt.Errorf("normalize scheduled Workflow Invocation input: %w", err)
	}
	payload, err := input.DeterministicJSON()
	if err != nil {
		return nil, bindinglang.ScheduleSource{}, fmt.Errorf("encode scheduled Workflow Invocation input: %w", err)
	}
	return payload, *normalized.Spec.Source.Schedule, nil
}

// deferBlockedScheduleSource durably removes one blocked source from the next
// few bounded scans. Without this operational backoff, an earliest prefix of
// permanently blocked sources can consume every batch forever and starve
// healthy due sources ordered behind it. The authored revision/cursor are not
// changed; a later pass retries the exact same occurrence.
func deferBlockedScheduleSource(
	ctx context.Context,
	tx coredb.Tx,
	source lockedScheduleSource,
	occurrence time.Time,
	now time.Time,
) error {
	occurrence = occurrence.UTC().Truncate(time.Microsecond)
	retryAt := now.UTC().Truncate(time.Microsecond).Add(scheduleBlockedRetryDelay)
	result, err := tx.Exec(ctx, `
UPDATE workflow_binding_sources
SET materialization_retry_at = $1
WHERE workspace_id = $2 AND binding_id = $3 AND id = $4
  AND revision = $5 AND status = 'active' AND source_kind = $6
  AND next_occurrence_at = $7
`, retryAt, source.WorkspaceID, source.BindingID, source.ID,
		source.Revision, SourceKindSchedule, occurrence)
	if err != nil {
		return fmt.Errorf("defer blocked scheduled Workflow source: %w", err)
	}
	if result.RowsAffected() != 1 {
		return fmt.Errorf("defer blocked scheduled Workflow source: source changed while locked")
	}
	return nil
}

func advanceScheduleSource(
	ctx context.Context,
	tx coredb.Tx,
	source lockedScheduleSource,
	acceptedOccurrence time.Time,
	nextOccurrence *time.Time,
	now time.Time,
) error {
	acceptedOccurrence = acceptedOccurrence.UTC().Truncate(time.Microsecond)
	if nextOccurrence != nil {
		next := nextOccurrence.UTC().Truncate(time.Microsecond)
		if !next.After(acceptedOccurrence) {
			return fmt.Errorf("next scheduled Workflow occurrence must be strictly after the accepted occurrence")
		}
		result, err := tx.Exec(ctx, `
UPDATE workflow_binding_sources
SET next_occurrence_at = $1,
    materialization_retry_at = NULL,
    revision = revision + 1,
    updated_at = $2
WHERE workspace_id = $3 AND binding_id = $4 AND id = $5
  AND revision = $6 AND status = 'active' AND source_kind = $7
  AND next_occurrence_at = $8
`, next, now, source.WorkspaceID, source.BindingID, source.ID,
			source.Revision, SourceKindSchedule, acceptedOccurrence)
		if err != nil {
			return fmt.Errorf("advance scheduled Workflow source cursor: %w", err)
		}
		if result.RowsAffected() != 1 {
			return fmt.Errorf("advance scheduled Workflow source cursor: source changed while locked")
		}
		return nil
	}

	result, err := tx.Exec(ctx, `
UPDATE workflow_binding_sources
SET status = 'completed',
    next_occurrence_at = NULL,
    materialization_retry_at = NULL,
    closed_at = $1,
    closed_reason = $2,
    revision = revision + 1,
    updated_at = $1
WHERE workspace_id = $3 AND binding_id = $4 AND id = $5
  AND revision = $6 AND status = 'active' AND source_kind = $7
  AND next_occurrence_at = $8
`, now, scheduleCompletedReason, source.WorkspaceID, source.BindingID, source.ID,
		source.Revision, SourceKindSchedule, acceptedOccurrence)
	if err != nil {
		return fmt.Errorf("complete one-shot Workflow schedule source: %w", err)
	}
	if result.RowsAffected() != 1 {
		return fmt.Errorf("complete one-shot Workflow schedule source: source changed while locked")
	}
	return nil
}

func timePointer(value time.Time) *time.Time { return &value }