Incremental Re-ingest (Operator): Full-History Namespace Clone + Per-Repository Cron Schedule -- Implementation Plan¶

package ingest

import "testing"

func TestNamespacePath(t *testing.T) {
    tests := []struct {
        name string
        url  string
        want string
    }{
        {"https github with .git", "https://github.com/szymonrychu/tatara-cli.git", "szymonrychu/tatara-cli"},
        {"https github no .git", "https://github.com/szymonrychu/tatara-cli", "szymonrychu/tatara-cli"},
        {"https gitlab subgroups", "https://gitlab.com/szymonrychu/infra/helmfile", "szymonrychu/infra/helmfile"},
        {"scp-like git@", "git@github.com:szymonrychu/tatara-cli.git", "szymonrychu/tatara-cli"},
        {"ssh url with port", "ssh://git@host:22/group/sub/repo.git", "group/sub/repo"},
        {"trailing slash", "https://github.com/acme/widgets/", "acme/widgets"},
        {"https with userinfo", "https://x-access-token@github.com/acme/widgets.git", "acme/widgets"},
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            if got := namespacePath(tt.url); got != tt.want {
                t.Errorf("namespacePath(%q) = %q, want %q", tt.url, got, tt.want)
            }
        })
    }
}

package ingest

import "strings"

// namespacePath maps a git clone URL to the on-disk subpath:
// owner[/subgroups]/repo, dropping scheme, host, userinfo, and a trailing
// ".git". Keeps the owner.
//
//  https://github.com/szymonrychu/tatara-cli.git   -> szymonrychu/tatara-cli
//  https://gitlab.com/szymonrychu/infra/helmfile    -> szymonrychu/infra/helmfile
//  git@github.com:szymonrychu/tatara-cli.git        -> szymonrychu/tatara-cli
//  ssh://git@host:22/group/sub/repo.git             -> group/sub/repo
func namespacePath(cloneURL string) string {
    s := strings.TrimSpace(cloneURL)

    // Drop scheme (https://, ssh://, git://, ...).
    if i := strings.Index(s, "://"); i >= 0 {
        s = s[i+3:]
    }

    // scp-like "git@host:owner/repo" has no scheme; split host from path on
    // the first ":" when there is no "/" before it.
    if !strings.Contains(s[:firstSlash(s)], "/") {
        if i := strings.Index(s, ":"); i >= 0 {
            s = s[:i] + "/" + s[i+1:]
        }
    }

    // Drop userinfo before the host ("git@host", "x-access-token@host").
    if i := strings.Index(s, "@"); i >= 0 && i < firstSlash(s) {
        s = s[i+1:]
    }

    // Drop the host segment (everything up to and including the first slash).
    if i := strings.Index(s, "/"); i >= 0 {
        s = s[i+1:]
    }

    s = strings.Trim(s, "/")
    s = strings.TrimSuffix(s, ".git")
    return s
}

// firstSlash returns the index of the first "/" in s, or len(s) when absent.
func firstSlash(s string) int {
    if i := strings.Index(s, "/"); i >= 0 {
        return i
    }
    return len(s)
}

func TestBuildJob_FullHistoryClone(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "", testBaseURL, testConfig())
    clone := job.Spec.Template.Spec.InitContainers[0]
    cloneCmd := strings.Join(clone.Command, " ") + " " + strings.Join(clone.Args, " ")
    if strings.Contains(cloneCmd, "--depth") {
        t.Errorf("clone must be full history (no --depth): %q", cloneCmd)
    }
    if !strings.Contains(cloneCmd, "--branch main") {
        t.Errorf("clone cmd missing branch: %q", cloneCmd)
    }
}

func TestBuildJob_NamespaceCloneDir(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "", testBaseURL, testConfig())

    // widgets repo URL is https://github.com/acme/widgets.git -> acme/widgets
    const wantDir = "/workspace/acme/widgets"

    clone := job.Spec.Template.Spec.InitContainers[0]
    cloneCmd := strings.Join(clone.Command, " ") + " " + strings.Join(clone.Args, " ")
    if !strings.Contains(cloneCmd, wantDir) {
        t.Errorf("clone must target namespace dir %q: %q", wantDir, cloneCmd)
    }

    main := job.Spec.Template.Spec.Containers[0]
    cmd := strings.Join(main.Command, " ") + " " + strings.Join(main.Args, " ")
    if !strings.Contains(cmd, "--repo-root "+wantDir) {
        t.Errorf("ingest cmd must use namespace repo-root %q: %q", wantDir, cmd)
    }
    if !strings.Contains(cmd, "git -C "+wantDir+" rev-parse HEAD") {
        t.Errorf("HEAD resolution must run in namespace dir %q: %q", wantDir, cmd)
    }
}

    if !strings.Contains(cmd, "tatara-ingest --repo-root /workspace/repo --repo-name widgets --base-url http://mem-acme.tatara.svc:8080") {

    if !strings.Contains(cmd, "tatara-ingest --repo-root /workspace/acme/widgets --repo-name widgets --base-url http://mem-acme.tatara.svc:8080") {

const (
    workspaceVolume = "workspace"
    workspaceMount  = "/workspace"
    repoDir         = "/workspace/repo"
)

const (
    workspaceVolume = "workspace"
    workspaceMount  = "/workspace"
)

    // Use git credential helper to inject SCM_TOKEN without embedding it in
    // the URL string. The full URL appears literally in the command so tests
    // can assert on it; the token is supplied via SecretKeyRef env var.
    cloneCmd := fmt.Sprintf(
        `set -e; git -c "credential.helper=!f() { echo username=x-access-token; echo password=${SCM_TOKEN}; }; f" `+
            `clone --depth 1 --branch %s %s %s`,
        repo.Spec.DefaultBranch, repo.Spec.URL, repoDir)

    ingestArgs := fmt.Sprintf(
        "tatara-ingest --repo-root %s --repo-name %s --base-url %s",
        repoDir, repo.Name, baseURL)
    if since != "" {
        ingestArgs += " --since " + since
    }
    // After a successful ingest, resolve HEAD and patch the result ConfigMap
    // via the in-cluster API (the Job ServiceAccount has patch on it).
    resultCM := ResultConfigMapName(repo)
    mainScript := fmt.Sprintf(
        "set -e; %s; "+
            "SHA=$(git -C %s rev-parse HEAD); "+
            "kubectl -n %s patch configmap %s --type merge "+
            "-p \"{\\\"data\\\":{\\\"sha\\\":\\\"${SHA}\\\"}}\"",
        ingestArgs, repoDir, cfg.Namespace, resultCM)

    // Clone into a directory that mirrors the repo namespace (owner/.../repo),
    // not a flat "/workspace/repo", so concurrent clones never collide.
    repoDir := workspaceMount + "/" + namespacePath(repo.Spec.URL)

    // Use git credential helper to inject SCM_TOKEN without embedding it in
    // the URL string. The full URL appears literally in the command so tests
    // can assert on it; the token is supplied via SecretKeyRef env var.
    // Full-history clone (no --depth): the incremental diff needs <since> in
    // history, and a shallow clone exits 128 when <since> is absent.
    cloneCmd := fmt.Sprintf(
        `set -e; git -c "credential.helper=!f() { echo username=x-access-token; echo password=${SCM_TOKEN}; }; f" `+
            `clone --branch %s %s %s`,
        repo.Spec.DefaultBranch, repo.Spec.URL, repoDir)

    ingestArgs := fmt.Sprintf(
        "tatara-ingest --repo-root %s --repo-name %s --base-url %s",
        repoDir, repo.Name, baseURL)
    if since != "" {
        ingestArgs += " --since " + since
    }
    // After a successful ingest, resolve HEAD and patch the result ConfigMap
    // via the in-cluster API (the Job ServiceAccount has patch on it).
    resultCM := ResultConfigMapName(repo)
    mainScript := fmt.Sprintf(
        "set -e; %s; "+
            "SHA=$(git -C %s rev-parse HEAD); "+
            "kubectl -n %s patch configmap %s --type merge "+
            "-p \"{\\\"data\\\":{\\\"sha\\\":\\\"${SHA}\\\"}}\"",
        ingestArgs, repoDir, cfg.Namespace, resultCM)

package v1alpha1

import (
    "testing"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func TestRepositorySpec_ReingestScheduleField(t *testing.T) {
    r := &Repository{}
    r.Spec.ReingestSchedule = "0 6 * * *"
    if r.Spec.ReingestSchedule != "0 6 * * *" {
        t.Fatalf("ReingestSchedule = %q, want %q", r.Spec.ReingestSchedule, "0 6 * * *")
    }
}

func TestRepositoryStatus_LastScheduledReingestField(t *testing.T) {
    r := &Repository{}
    now := metav1.NewTime(time.Now())
    r.Status.LastScheduledReingest = &now
    if r.Status.LastScheduledReingest == nil {
        t.Fatal("LastScheduledReingest should round-trip a *metav1.Time")
    }
}

func TestRepository_DeepCopyCopiesLastScheduledReingest(t *testing.T) {
    now := metav1.NewTime(time.Now())
    r := &Repository{}
    r.Spec.ReingestSchedule = "0 6 * * *"
    r.Status.LastScheduledReingest = &now

    cp := r.DeepCopy()
    if cp.Spec.ReingestSchedule != "0 6 * * *" {
        t.Errorf("deepcopy lost ReingestSchedule: %q", cp.Spec.ReingestSchedule)
    }
    if cp.Status.LastScheduledReingest == nil {
        t.Fatal("deepcopy lost LastScheduledReingest")
    }
    // Must be a distinct pointer (deep, not shallow).
    if cp.Status.LastScheduledReingest == r.Status.LastScheduledReingest {
        t.Error("deepcopy must allocate a new LastScheduledReingest pointer")
    }
}

// RepositorySpec defines the desired state of a Repository.
type RepositorySpec struct {
    ProjectRef string `json:"projectRef"`
    URL        string `json:"url"`
    // +kubebuilder:default="main"
    // +optional
    DefaultBranch string `json:"defaultBranch,omitempty"`
    // +kubebuilder:default=true
    // +optional
    IngestEnabled bool `json:"ingestEnabled,omitempty"`
}

// RepositorySpec defines the desired state of a Repository.
type RepositorySpec struct {
    ProjectRef string `json:"projectRef"`
    URL        string `json:"url"`
    // +kubebuilder:default="main"
    // +optional
    DefaultBranch string `json:"defaultBranch,omitempty"`
    // +kubebuilder:default=true
    // +optional
    IngestEnabled bool `json:"ingestEnabled,omitempty"`
    // ReingestSchedule is a standard 5-field cron expression (e.g. "0 6 * * *")
    // that triggers a periodic catch-up re-ingest in addition to push webhooks.
    // +kubebuilder:validation:Required
    // +kubebuilder:validation:MinLength=9
    // +kubebuilder:validation:Pattern=`^(\S+\s+){4}\S+$`
    ReingestSchedule string `json:"reingestSchedule"`
}

// RepositoryStatus defines the observed state of a Repository.
type RepositoryStatus struct {
    // +kubebuilder:validation:Enum=Pending;Ingesting;Ingested;Failed
    // +optional
    Phase string `json:"phase,omitempty"`
    // +optional
    LastIngestedCommit string `json:"lastIngestedCommit,omitempty"`
    // +optional
    LastIngestTime *metav1.Time `json:"lastIngestTime,omitempty"`
    // LastScheduledReingest is the last time the cron schedule stamped a
    // reingest-requested annotation; used as the base for the next fire.
    // +optional
    LastScheduledReingest *metav1.Time `json:"lastScheduledReingest,omitempty"`
    // +optional
    JobName string `json:"jobName,omitempty"`
    // +optional
    // +listType=map
    // +listMapKey=type
    Conditions []metav1.Condition `json:"conditions,omitempty"`
}

func mkRepo(t *testing.T, name, projectRef string) *tataradevv1alpha1.Repository {
    t.Helper()
    r := &tataradevv1alpha1.Repository{}
    r.Name = name
    r.Namespace = testNS
    r.Spec.ProjectRef = projectRef
    r.Spec.URL = "https://github.com/acme/" + name + ".git"
    r.Spec.DefaultBranch = "main"
    r.Spec.IngestEnabled = true
    if err := k8sClient.Create(context.Background(), r); err != nil {
        t.Fatalf("create repo %s: %v", name, err)
    }
    return r
}

func mkRepo(t *testing.T, name, projectRef string) *tataradevv1alpha1.Repository {
    t.Helper()
    r := &tataradevv1alpha1.Repository{}
    r.Name = name
    r.Namespace = testNS
    r.Spec.ProjectRef = projectRef
    r.Spec.URL = "https://github.com/acme/" + name + ".git"
    r.Spec.DefaultBranch = "main"
    r.Spec.IngestEnabled = true
    r.Spec.ReingestSchedule = "0 6 * * *"
    if err := k8sClient.Create(context.Background(), r); err != nil {
        t.Fatalf("create repo %s: %v", name, err)
    }
    return r
}

func mkTaskRepository(t *testing.T, name, projectRef string) {
    t.Helper()
    r := &tatarav1alpha1.Repository{}
    r.Name = name
    r.Namespace = testNS
    r.Spec.ProjectRef = projectRef
    r.Spec.URL = "https://git/acme/" + name
    r.Spec.DefaultBranch = "main"
    if err := k8sClient.Create(context.Background(), r); err != nil {
        t.Fatalf("create repository: %v", err)
    }
}

func mkTaskRepository(t *testing.T, name, projectRef string) {
    t.Helper()
    r := &tatarav1alpha1.Repository{}
    r.Name = name
    r.Namespace = testNS
    r.Spec.ProjectRef = projectRef
    r.Spec.URL = "https://git/acme/" + name
    r.Spec.DefaultBranch = "main"
    r.Spec.ReingestSchedule = "0 6 * * *"
    if err := k8sClient.Create(context.Background(), r); err != nil {
        t.Fatalf("create repository: %v", err)
    }
}

func setRepoIngested(t *testing.T, name, sha string, lastIngest time.Time) {
    t.Helper()
    r := getRepo(t, name)
    r.Status.LastIngestedCommit = sha
    lt := metav1.NewTime(lastIngest)
    r.Status.LastIngestTime = &lt
    r.Status.Phase = "Ingested"
    if err := k8sClient.Status().Update(context.Background(), r); err != nil {
        t.Fatalf("seed ingested status for %s: %v", name, err)
    }
}

func TestRepoReconcile_ScheduleStampsAnnotationWhenDue(t *testing.T) {
    mkProject(t, "rp-sch1", "rp-sch1-scm")
    mkSecret(t, "rp-sch1-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "sch1", "rp-sch1")
    setProjectMemoryReady(t, "rp-sch1", "http://mem-rp-sch1.tatara.svc:8080")

    // Schedule fires every minute; last ingest was an hour ago, so it is due now.
    r := getRepo(t, "sch1")
    r.Spec.ReingestSchedule = "* * * * *"
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set schedule: %v", err)
    }
    setRepoIngested(t, "sch1", "shaSch1", time.Now().Add(-1*time.Hour))

    if _, err := reconcileRepo(t, "sch1"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }

    got := getRepo(t, "sch1")
    if got.Annotations[ReingestAnnotation] == "" {
        t.Fatal("due schedule must stamp the reingest-requested annotation")
    }
    if got.Status.LastScheduledReingest == nil {
        t.Fatal("due schedule must set status.lastScheduledReingest")
    }
    // No Job yet: the annotation re-triggers reconcile via the watch; this
    // reconcile pass only stamps.
    if jobs := listIngestJobs(t, "sch1"); len(jobs) != 0 {
        t.Fatalf("schedule stamp pass must not itself launch a job, got %d", len(jobs))
    }
}

func TestRepoReconcile_ScheduleRequeuesWhenNotDue(t *testing.T) {
    mkProject(t, "rp-sch2", "rp-sch2-scm")
    mkSecret(t, "rp-sch2-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "sch2", "rp-sch2")
    setProjectMemoryReady(t, "rp-sch2", "http://mem-rp-sch2.tatara.svc:8080")

    // Far-future daily schedule + a fresh ingest => not due; expect a requeue.
    r := getRepo(t, "sch2")
    r.Spec.ReingestSchedule = "0 6 * * *"
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set schedule: %v", err)
    }
    setRepoIngested(t, "sch2", "shaSch2", time.Now())

    res, err := reconcileRepo(t, "sch2")
    if err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    if res.RequeueAfter <= 0 {
        t.Errorf("not-due schedule must set RequeueAfter, got %v", res.RequeueAfter)
    }
    if res.RequeueAfter > 6*time.Hour {
        t.Errorf("RequeueAfter must be clamped to 6h, got %v", res.RequeueAfter)
    }
    if getRepo(t, "sch2").Annotations[ReingestAnnotation] != "" {
        t.Error("not-due schedule must not stamp the annotation")
    }
    if getRepo(t, "sch2").Status.LastScheduledReingest != nil {
        t.Error("not-due schedule must not set lastScheduledReingest")
    }
}

func TestRepoReconcile_ScheduleBadCronSkips(t *testing.T) {
    mkProject(t, "rp-sch3", "rp-sch3-scm")
    mkSecret(t, "rp-sch3-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "sch3", "rp-sch3")
    setProjectMemoryReady(t, "rp-sch3", "http://mem-rp-sch3.tatara.svc:8080")

    // A syntactically-shaped but semantically invalid cron (bad minute field).
    r := getRepo(t, "sch3")
    r.Spec.ReingestSchedule = "99 6 * * *"
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set schedule: %v", err)
    }
    setRepoIngested(t, "sch3", "shaSch3", time.Now().Add(-1*time.Hour))

    res, err := reconcileRepo(t, "sch3")
    if err != nil {
        t.Fatalf("bad cron must not error the reconcile: %v", err)
    }
    if res.RequeueAfter != 0 {
        t.Errorf("bad cron must skip scheduling (no requeue), got %v", res.RequeueAfter)
    }
    if getRepo(t, "sch3").Annotations[ReingestAnnotation] != "" {
        t.Error("bad cron must not stamp the annotation")
    }
}

func TestRepoReconcile_ScheduleNoDoubleFireWithinInterval(t *testing.T) {
    mkProject(t, "rp-sch4", "rp-sch4-scm")
    mkSecret(t, "rp-sch4-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "sch4", "rp-sch4")
    setProjectMemoryReady(t, "rp-sch4", "http://mem-rp-sch4.tatara.svc:8080")

    r := getRepo(t, "sch4")
    r.Spec.ReingestSchedule = "0 6 * * *"
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set schedule: %v", err)
    }
    setRepoIngested(t, "sch4", "shaSch4", time.Now().Add(-25*time.Hour))

    // lastScheduledReingest is recent => next fire is in the future => not due,
    // even though lastIngestTime is old. Guards against double-fire.
    r = getRepo(t, "sch4")
    just := metav1.NewTime(time.Now().Add(-1 * time.Minute))
    r.Status.LastScheduledReingest = &just
    if err := k8sClient.Status().Update(context.Background(), r); err != nil {
        t.Fatalf("seed lastScheduledReingest: %v", err)
    }

    res, err := reconcileRepo(t, "sch4")
    if err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    if getRepo(t, "sch4").Annotations[ReingestAnnotation] != "" {
        t.Error("recent lastScheduledReingest must prevent a second stamp this interval")
    }
    if res.RequeueAfter <= 0 {
        t.Errorf("expected a requeue to the next fire, got %v", res.RequeueAfter)
    }
}

import (
    "context"
    "fmt"
    "time"

    "github.com/robfig/cron/v3"
    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/ingest"
    "github.com/szymonrychu/tatara-operator/internal/obs"
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    apierrors "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/api/meta"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/types"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
    "sigs.k8s.io/controller-runtime/pkg/log"
)

// maxScheduleRequeue bounds the cron requeue so clock skew or long sleeps still
// re-evaluate the schedule reasonably soon.
const maxScheduleRequeue = 6 * time.Hour

    since, want := r.ingestDecision(&repo)
    if !want {
        r.Metrics.ReconcileResult("Repository", "success")
        return ctrl.Result{}, nil
    }

    since, want := r.ingestDecision(&repo)
    if !want {
        res, err := r.scheduleNextReingest(ctx, &repo)
        if err != nil {
            r.Metrics.ReconcileResult("Repository", "error")
            return ctrl.Result{}, err
        }
        r.Metrics.ReconcileResult("Repository", "success")
        return res, nil
    }

// scheduleNextReingest applies the per-Repository cron schedule for an
// already-ingested repo. It parses spec.reingestSchedule and computes the next
// fire from base = lastScheduledReingest | lastIngestTime | creationTimestamp.
// When the fire is due (and strictly after lastIngestTime, so an in-flight
// ingest from another trigger is not double-stamped), it stamps the existing
// reingest-requested annotation and records lastScheduledReingest; the
// annotation change re-triggers reconcile, which launches the Job via the
// existing path. Otherwise it requeues at the next fire (clamped). A bad cron
// expression is logged at ERROR and skipped (no requeue, no error).
func (r *RepositoryReconciler) scheduleNextReingest(ctx context.Context, repo *tataradevv1alpha1.Repository) (ctrl.Result, error) {
    l := log.FromContext(ctx)

    // Only schedule once a repo has been ingested at least once; a first full
    // ingest is driven by ingestDecision, not the cron.
    if repo.Status.LastIngestedCommit == "" || repo.Spec.ReingestSchedule == "" {
        return ctrl.Result{}, nil
    }

    schedule, err := cron.ParseStandard(repo.Spec.ReingestSchedule)
    if err != nil {
        l.Error(err, "invalid reingestSchedule, skipping cron",
            "action", "ingest_schedule_invalid", "resource_id", repo.Name,
            "schedule", repo.Spec.ReingestSchedule)
        return ctrl.Result{}, nil
    }

    var lastIngestTime time.Time
    if repo.Status.LastIngestTime != nil {
        lastIngestTime = repo.Status.LastIngestTime.Time
    }

    base := repo.CreationTimestamp.Time
    if repo.Status.LastIngestTime != nil {
        base = repo.Status.LastIngestTime.Time
    }
    if repo.Status.LastScheduledReingest != nil {
        base = repo.Status.LastScheduledReingest.Time
    }

    now := time.Now()
    next := schedule.Next(base)

    if now.Before(next) {
        requeue := next.Sub(now)
        if requeue > maxScheduleRequeue {
            requeue = maxScheduleRequeue
        }
        return ctrl.Result{RequeueAfter: requeue}, nil
    }

    // Due. Guard against firing while an ingest from another trigger is still
    // in flight or just finished: only stamp when now is strictly after the
    // last successful ingest.
    if !now.After(lastIngestTime) {
        return ctrl.Result{RequeueAfter: maxScheduleRequeue}, nil
    }

    if repo.Annotations == nil {
        repo.Annotations = map[string]string{}
    }
    repo.Annotations[ReingestAnnotation] = now.UTC().Format(time.RFC3339)
    if err := r.Update(ctx, repo); err != nil {
        return ctrl.Result{}, fmt.Errorf("stamp scheduled reingest annotation: %w", err)
    }

    scheduled := metav1.NewTime(now)
    repo.Status.LastScheduledReingest = &scheduled
    if err := r.Status().Update(ctx, repo); err != nil {
        return ctrl.Result{}, fmt.Errorf("update lastScheduledReingest: %w", err)
    }

    l.Info("scheduled re-ingest requested",
        "action", "ingest_schedule_fire", "resource_id", repo.Name,
        "schedule", repo.Spec.ReingestSchedule)
    return ctrl.Result{}, nil
}