feat(identity): sync CLI — copy the persons log into ClickHouse (metrics person_id, iteration 1)

src/backend/Cargo.toml

@@ -33,7 +33,7 @@ dbg_macro = "deny"
 
 [workspace.dependencies]
 # ClickHouse client
-clickhouse = { version = "0.14", features = ["uuid"] }
+clickhouse = { version = "0.14", features = ["uuid", "chrono"] }
 
 sqlparser = "0.62"
 

src/backend/services/identity-resolution/helm/templates/sync-cronjob.yaml

@@ -0,0 +1,91 @@
+{{- if .Values.sync.enabled }}
+# Scheduled persons-sync: copies the MariaDB `persons` observation log into
+# ClickHouse `identity.identity_persons` — the table the metrics dbt builds
+# resolve email → person_id against. Scheduled AFTER the seed CronJob by
+# default: the seed rewrites the log, the sync publishes it (each run is a
+# full snapshot + atomic swap, so ordering is a freshness concern only,
+# never a correctness one).
+#
+# Same image/config/secret wiring as the seed CronJob; the `sync`
+# subcommand runs one copy and exits (exit codes: 0 ok / 1 failed /
+# 2 lock busy / 3 empty-log guard — override the guard with a manual
+# `--force` Job). Runs serialize on a global MariaDB advisory lock, so
+# `concurrencyPolicy: Forbid` is belt-and-braces for cron-vs-cron only.
+#
+# Manual run:
+#   kubectl create job --from=cronjob/{{ include "insight-identity-resolution.fullname" . }}-sync sync-manual-$USER
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: {{ include "insight-identity-resolution.fullname" . }}-sync
+  labels:
+    {{- include "insight-identity-resolution.labels" . | nindent 4 }}
+    app.kubernetes.io/component: persons-sync
+spec:
+  schedule: {{ .Values.sync.schedule | quote }}
+  concurrencyPolicy: {{ .Values.sync.concurrencyPolicy }}
+  successfulJobsHistoryLimit: {{ .Values.sync.successfulJobsHistoryLimit }}
+  failedJobsHistoryLimit: {{ .Values.sync.failedJobsHistoryLimit }}
+  jobTemplate:
+    spec:
+      # Retries for transient connect blips; the advisory lock + the
+      # snapshot-swap idempotency make a repeated run safe. The deadline caps
+      # a wedged pod well past the in-binary 5-minute sync timeout.
+      backoffLimit: {{ .Values.sync.backoffLimit }}
+      activeDeadlineSeconds: {{ .Values.sync.activeDeadlineSeconds }}
+      template:
+        metadata:
+          # NOT the shared selectorLabels: the Service selects on
+          # name+instance alone, and a sync pod carrying them would enter the
+          # Service's endpoints (it listens on nothing).
+          labels:
+            app.kubernetes.io/name: identity-resolution-sync
+            app.kubernetes.io/instance: {{ .Release.Name }}
+            app.kubernetes.io/component: persons-sync
+        spec:
+          {{- with .Values.global }}
+          {{- with .imagePullSecrets }}
+          imagePullSecrets:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- end }}
+          enableServiceLinks: false
+          restartPolicy: Never
+          securityContext:
+            runAsNonRoot: true
+            runAsUser: 1000
+            fsGroup: 1000
+          containers:
+            - name: persons-sync
+              image: "{{ required "image.repository is required" .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+              imagePullPolicy: {{ .Values.image.pullPolicy }}
+              command: ["/app/identity-resolution"]
+              args: ["-c", "/app/config/insight.yaml", "sync"]
+              volumeMounts:
+                - name: identity-resolution-config
+                  mountPath: /app/config/insight.yaml
+                  subPath: insight.yaml
+                  readOnly: true
+              envFrom:
+                # Same Secret as the deployment: database_url, clickhouse_*,
+                # tenant_default_id — everything the sync needs. The
+                # ClickHouse user must be able to CREATE/INSERT/EXCHANGE in
+                # the `identity` database (the snapshot-swap write path).
+                - secretRef:
+                    name: {{ required "existingSecret is required (umbrella provides `insight-identity-resolution-config`; standalone installs supply their own)" .Values.existingSecret | quote }}
+              {{- with .Values.sync.tenantDefaultId }}
+              env:
+                # Explicit tenant for the JOURNAL row (the copy itself is
+                # tenant-agnostic). Same override semantics as the seed's.
+                - name: APP__gears__identity-resolution__config__tenant_default_id
+                  value: {{ . | quote }}
+              {{- end }}
+              securityContext:
+                allowPrivilegeEscalation: false
+              resources:
+                {{- toYaml .Values.sync.resources | nindent 16 }}
+          volumes:
+            - name: identity-resolution-config
+              configMap:
+                name: {{ include "insight-identity-resolution.fullname" . }}-gears-config
+{{- end }}

src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py

@@ -1,23 +1,31 @@
-"""Helm render-contract for the persons-seed CronJob (#1690).
-
-The original bug was the ABSENCE of scheduling — the seed existed but nothing
-ever ran it — so the schedule wiring itself is contract, not plumbing: these
-tests render the chart(s) with `helm template` and assert the manifests the
-cluster would actually get. No cluster involved; runs anywhere helm + PyYAML
-exist (CI: .github/workflows/identity-resolution-helm.yml).
-
-Covered:
-  * the CronJob exists by default with the documented schedule and the exact
-    `seed` command/args against the mounted gears config;
+"""Helm render-contract for the persons-seed AND persons-sync CronJobs.
+
+The original bug was the ABSENCE of scheduling (#1690) — the seed existed but
+nothing ever ran it — so the schedule wiring itself is contract, not
+plumbing: these tests render the chart(s) with `helm template` and assert the
+manifests the cluster would actually get. No cluster involved; runs anywhere
+helm + PyYAML exist (CI: .github/workflows/identity-resolution-helm.yml).
+
+The chart now ships TWO CronJobs — seed (rebuilds the persons log from
+identity_inputs) and sync (publishes the log into ClickHouse
+`identity.identity_persons` for the metrics resolve path, scheduled 15
+minutes after the seed). CronJobs are selected BY NAME, never as "the sole
+CronJob in the render" — the suite must not break again when a third job
+appears.
+
+Covered, per job:
+  * exists by default with its documented schedule and the exact
+    subcommand/args against the mounted gears config (never `--force`);
   * config comes from the SAME Secret/ConfigMap pair the deployment uses;
-  * `seed.tenantDefaultId` env-overrides the Secret (k8s `env` beats
-    `envFrom`) — the standalone-install tenant source;
-  * `seed.enabled=false` removes the CronJob and nothing else;
-  * the seed pod labels do NOT match the Service selector (a pod that
-    listens on nothing must never enter the Service's endpoints);
-  * the umbrella refuses to render when the seed is enabled but no tenant is
-    configured — only on the path where the umbrella composes the config
-    Secret itself (`credentials.autoGenerate`).
+  * `<job>.tenantDefaultId` env-overrides the Secret (k8s `env` beats
+    `envFrom`);
+  * `<job>.enabled=false` removes THAT CronJob and nothing else;
+  * the job pod labels do NOT match the Service selector (a pod that
+    listens on nothing must never enter the Service's endpoints).
+
+Umbrella: the seed tenant render guard (unchanged — the sync only journals
+under the tenant, so it carries no equivalent guard) and both CronJobs
+rendering when a tenant is configured.
 """
 
 from __future__ import annotations
@@ -35,6 +43,12 @@
 
 TENANT = "3e1d5a65-434c-95b4-8c1b-eb8f53a39bab"
 
+# name suffix -> (schedule, subcommand) — the per-job contract facts.
+JOBS = {
+    "seed": ("30 6 * * *", "seed"),
+    "sync": ("45 6 * * *", "sync"),
+}
+
 # Minimum viable subchart install (mirrors the umbrella's wiring).
 SUBCHART_BASE = [
     "--set",
@@ -109,6 +123,31 @@ def _the(docs: list[dict], kind: str) -> dict:
     return matches[0]
 
 
+def _cronjobs(docs: list[dict]) -> dict[str, dict]:
+    """All CronJobs in the render, keyed by metadata name."""
+    return {d["metadata"]["name"]: d for d in docs if d.get("kind") == "CronJob"}
+
+
+def _cronjob(docs: list[dict], job: str) -> dict:
+    """The seed/sync CronJob selected BY NAME — never 'the sole CronJob'.
+
+    Matched on the identity-resolution fullname + the job suffix rather than
+    an exact literal, so the same helper works for subchart renders
+    (`contract-test-identity-resolution-<job>`) and umbrella renders (whose
+    release/alias prefix differs).
+    """
+    matches = {
+        name: doc
+        for name, doc in _cronjobs(docs).items()
+        if "identity-resolution" in name and name.endswith(f"-{job}")
+    }
+    assert len(matches) == 1, (
+        f"expected exactly one identity-resolution {job} CronJob; "
+        f"present: {sorted(_cronjobs(docs))}"
+    )
+    return next(iter(matches.values()))
+
+
 def _subchart_docs(*extra: str) -> list[dict]:
     rc, out, err = _render(SUBCHART, *SUBCHART_BASE, *extra)
     assert rc == 0, err
@@ -120,32 +159,42 @@ def default_docs() -> list[dict]:
     return _subchart_docs()
 
 
-def _seed_container(cronjob: dict) -> dict:
+def _job_container(cronjob: dict) -> dict:
     pod = cronjob["spec"]["jobTemplate"]["spec"]["template"]["spec"]
     assert pod["restartPolicy"] == "Never", pod
     (container,) = pod["containers"]
     return container
 
 
-def test_cronjob_exists_by_default_with_documented_schedule(default_docs) -> None:
-    cj = _the(default_docs, "CronJob")
-    assert cj["metadata"]["name"] == "contract-test-identity-resolution-seed"
-    assert cj["spec"]["schedule"] == "30 6 * * *"
+def test_default_render_ships_exactly_the_two_documented_cronjobs(default_docs) -> None:
+    names = sorted(_cronjobs(default_docs))
+    assert len(names) == len(JOBS), names
+    for job in JOBS:
+        _cronjob(default_docs, job)
+
+
+@pytest.mark.parametrize("job", JOBS)
+def test_cronjob_exists_by_default_with_documented_schedule(default_docs, job: str) -> None:
+    cj = _cronjob(default_docs, job)
+    assert cj["spec"]["schedule"] == JOBS[job][0]
     assert cj["spec"]["concurrencyPolicy"] == "Forbid"
 
 
-def test_cronjob_runs_the_seed_subcommand_against_the_mounted_config(default_docs) -> None:
-    container = _seed_container(_the(default_docs, "CronJob"))
+@pytest.mark.parametrize("job", JOBS)
+def test_cronjob_runs_its_subcommand_against_the_mounted_config(default_docs, job: str) -> None:
+    container = _job_container(_cronjob(default_docs, job))
     assert container["command"] == ["/app/identity-resolution"]
-    assert container["args"] == ["-c", "/app/config/insight.yaml", "seed"]
-    # The CronJob must never run forced — --force is a deliberate manual act.
+    assert container["args"] == ["-c", "/app/config/insight.yaml", JOBS[job][1]]
+    # A CronJob must never run forced — --force is a deliberate manual act
+    # (seed: input guards; sync: the empty-log guard).
     assert "--force" not in container["args"]
 
 
-def test_cronjob_uses_the_deployments_secret_and_configmap(default_docs) -> None:
-    cj = _the(default_docs, "CronJob")
+@pytest.mark.parametrize("job", JOBS)
+def test_cronjob_uses_the_deployments_secret_and_configmap(default_docs, job: str) -> None:
+    cj = _cronjob(default_docs, job)
     deploy = _the(default_docs, "Deployment")
-    container = _seed_container(cj)
+    container = _job_container(cj)
 
     secret_refs = [e["secretRef"]["name"] for e in container["envFrom"]]
     deploy_secret_refs = [
@@ -160,34 +209,41 @@ def test_cronjob_uses_the_deployments_secret_and_configmap(default_docs) -> None
     assert cj_cm == deploy_cm
 
 
-def test_tenant_value_overrides_the_secret_via_env(default_docs) -> None:
+@pytest.mark.parametrize("job", JOBS)
+def test_tenant_value_overrides_the_secret_via_env(default_docs, job: str) -> None:
     # Default: no explicit env — the Secret is the tenant source.
-    container = _seed_container(_the(default_docs, "CronJob"))
+    container = _job_container(_cronjob(default_docs, job))
     assert "env" not in container, container.get("env")
 
-    docs = _subchart_docs("--set", f"seed.tenantDefaultId={TENANT}")
-    container = _seed_container(_the(docs, "CronJob"))
+    docs = _subchart_docs("--set", f"{job}.tenantDefaultId={TENANT}")
+    container = _job_container(_cronjob(docs, job))
     env = {e["name"]: e["value"] for e in container["env"]}
     assert env == {"APP__gears__identity-resolution__config__tenant_default_id": TENANT}
 
 
-def test_seed_disabled_removes_only_the_cronjob() -> None:
-    docs = _subchart_docs("--set", "seed.enabled=false")
-    assert not [d for d in docs if d.get("kind") == "CronJob"]
-    # The rest of the chart is untouched.
+@pytest.mark.parametrize("job", JOBS)
+def test_disabling_one_job_removes_only_that_cronjob(job: str) -> None:
+    docs = _subchart_docs("--set", f"{job}.enabled=false")
+    jobs = _cronjobs(docs)
+    assert f"contract-test-identity-resolution-{job}" not in jobs, sorted(jobs)
+    # The sibling CronJob and the rest of the chart are untouched.
+    (other,) = [j for j in JOBS if j != job]
+    _cronjob(docs, other)
     _the(docs, "Deployment")
     _the(docs, "Service")
 
 
-def test_seed_pod_labels_never_match_the_service_selector(default_docs) -> None:
-    """A seed pod listens on nothing: if the Service selector matched it, it
-    would enter the endpoints and blackhole live traffic during every run."""
+@pytest.mark.parametrize("job", JOBS)
+def test_job_pod_labels_never_match_the_service_selector(default_docs, job: str) -> None:
+    """A seed/sync pod listens on nothing: if the Service selector matched
+    it, it would enter the endpoints and blackhole live traffic during every
+    run."""
     selector = _the(default_docs, "Service")["spec"]["selector"]
-    pod_labels = _the(default_docs, "CronJob")["spec"]["jobTemplate"]["spec"]["template"][
+    pod_labels = _cronjob(default_docs, job)["spec"]["jobTemplate"]["spec"]["template"][
         "metadata"
     ]["labels"]
     assert any(pod_labels.get(k) != v for k, v in selector.items()), (
-        f"seed pod labels {pod_labels} satisfy the Service selector {selector}"
+        f"{job} pod labels {pod_labels} satisfy the Service selector {selector}"
     )
 
 
@@ -212,20 +268,22 @@ def test_umbrella_refuses_enabled_seed_without_a_tenant(umbrella_deps) -> None:
     assert "requires a tenant" in err, err
 
 
-def test_umbrella_renders_the_cronjob_with_a_tenant(umbrella_deps) -> None:
+def test_umbrella_renders_both_cronjobs_with_a_tenant(umbrella_deps) -> None:
     rc, out, err = _render(
         umbrella_deps, *UMBRELLA_BASE, "--set", f"global.tenantDefaultId={TENANT}"
     )
     assert rc == 0, err
-    _the(_docs(out), "CronJob")
+    docs = _docs(out)
+    for job in JOBS:
+        _cronjob(docs, job)
 
 
 def test_umbrella_accepts_the_explicit_seed_tenant_alone(umbrella_deps) -> None:
     rc, out, err = _render(
         umbrella_deps, *UMBRELLA_BASE, "--set", f"identityResolution.seed.tenantDefaultId={TENANT}"
     )
     assert rc == 0, err
-    container = _seed_container(_the(_docs(out), "CronJob"))
+    container = _job_container(_cronjob(_docs(out), "seed"))
     env = {e["name"]: e["value"] for e in container.get("env", [])}
     assert env.get("APP__gears__identity-resolution__config__tenant_default_id") == TENANT
 
@@ -235,4 +293,9 @@ def test_umbrella_disabled_seed_needs_no_tenant(umbrella_deps) -> None:
         umbrella_deps, *UMBRELLA_BASE, "--set", "identityResolution.seed.enabled=false"
     )
     assert rc == 0, err
-    assert not [d for d in _docs(out) if d.get("kind") == "CronJob"]
+    jobs = _cronjobs(_docs(out))
+    # The seed CronJob is gone; the sync one legitimately remains (it has no
+    # tenant render guard — the tenant only scopes its journal row).
+    assert not any(
+        "identity-resolution" in n and n.endswith("-seed") for n in jobs
+    ), sorted(jobs)

src/backend/services/identity-resolution/helm/values.yaml

@@ -93,6 +93,36 @@ seed:
       cpu: 500m
       memory: 512Mi
 
+# Persons-sync CronJob: publishes the `persons` log into ClickHouse
+# `identity.identity_persons` (the metrics email→person_id resolve source).
+# Scheduled after the seed by default — the seed rewrites the log, the sync
+# publishes it.
+sync:
+  enabled: true
+  # Journal-row tenant (UUID); same override semantics as seed.tenantDefaultId
+  # (the copy itself is tenant-agnostic — this only scopes the operations
+  # journal the admin GET routes read).
+  tenantDefaultId: ""
+  # Daily, 15 minutes after the seed's 06:30 run.
+  schedule: "45 6 * * *"
+  # Belt-and-braces for cron-vs-cron only — manual Jobs and other instances
+  # serialize on the MariaDB advisory lock regardless.
+  concurrencyPolicy: Forbid
+  # Retries for transient connect blips; the lock + snapshot-swap idempotency
+  # make a repeated run safe.
+  backoffLimit: 2
+  # Caps a wedged pod well past the in-binary 5-minute sync timeout.
+  activeDeadlineSeconds: 600
+  successfulJobsHistoryLimit: 3
+  failedJobsHistoryLimit: 3
+  resources:
+    requests:
+      cpu: 100m
+      memory: 128Mi
+    limits:
+      cpu: 500m
+      memory: 512Mi
+
 # TCP gate before start: the service connects to MariaDB at boot. Same
 # busybox wait as the .NET identity chart.
 waitForMariadb:

src/backend/services/identity-resolution/src/api/error.rs

@@ -12,6 +12,9 @@ pub struct ProfileError;
 #[resource_error("gts.cf.insight.identity_resolution.persons_seed.v1~")]
 pub struct PersonsSeedError;
 
+#[resource_error("gts.cf.insight.identity_resolution.persons_sync.v1~")]
+pub struct PersonsSyncError;
+
 /// Shared admin-gate errors (401 no caller / 403 not admin), used by every
 /// admin-gated endpoint via [`crate::api::gate`].
 #[resource_error("gts.cf.insight.identity_resolution.access.v1~")]