From b1ee354cc3d7f17ba13af25adb7b571c113d5cbd Mon Sep 17 00:00:00 2001 From: Sergei Mozhaev Date: Thu, 30 Jul 2026 14:05:39 +0300 Subject: [PATCH] =?UTF-8?q?feat(identity):=20`sync`=20CLI=20=E2=80=94=20co?= =?UTF-8?q?py=20the=20persons=20log=20into=20ClickHouse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Iteration 1 of the metrics person_id rework (#1873 consumer side), reworked onto the CLI trigger model the persons-seed established in #1690 — the earlier POST /v1/persons-sync + queue + worker draft never merged; this lands the surface CLI-only from birth. `identity-resolution sync` runs one copy of the MariaDB `persons` observation log into ClickHouse `identity.identity_persons` (next to identity_inputs) and exits — the table dbt gold builds resolve email → person_id against. Execution mirrors the seed subcommand: global MariaDB GET_LOCK advisory lock held for the whole run (RAII guard, dies with the session), zombie sweep, `operations` journal row under the resolved tenant with the SYSTEM_AUTHOR nil UUID, exit codes 0 ok / 1 failed / 2 lock busy / 3 guard. The natural pairing is "sync after seed": the seed rewrites the log, the sync publishes it — the Helm CronJob is scheduled 15 minutes after the seed's (sync-cronjob.yaml, same wiring). Copy semantics: full snapshot per run — per-run staging table, count-verify, atomic EXCHANGE TABLES swap, `_synced_at` watermark guard as a backstop, >1h stale-staging GC. Readers never see an empty or partial table; a failed run leaves the live snapshot untouched. The log is copied verbatim (source provenance columns, NULLs preserved) so future per-source resolution lives entirely in the dbt-side resolve macro. An empty-log run is refused by a guard (publishing it would erase a populated snapshot — the destructive-zero-rows lesson); `--force` overrides deliberately. HTTP keeps only the read-only journal (GET /v1/persons-sync + /{id}), admin-gated, same wire conventions as the seed journal. e2e: rust-only CLI contract module — exit 0 + terminal journal row, snapshot verification in ClickHouse against the fixture dataset (fixed person UUIDs, dup pair stays unresolved, watermark stamped), replace-not-append, journal type-scoping vs the seed surface, status filter, 401/403/404. Full identity suite green (124 passed, 5 skipped). Co-Authored-By: Claude Fable 5 Signed-off-by: Sergei Mozhaev --- src/backend/Cargo.lock | 1 + src/backend/Cargo.toml | 2 +- .../helm/templates/sync-cronjob.yaml | 91 +++++ .../helm/tests/test_seed_cronjob_contract.py | 159 ++++++--- .../identity-resolution/helm/values.yaml | 30 ++ .../identity-resolution/src/api/error.rs | 3 + .../identity-resolution/src/api/mod.rs | 32 ++ .../identity-resolution/src/api/seed.rs | 7 +- .../identity-resolution/src/api/sync.rs | 139 ++++++++ .../identity-resolution/src/domain/mod.rs | 1 + .../src/domain/sync_service.rs | 203 +++++++++++ .../services/identity-resolution/src/gear.rs | 23 ++ .../identity-resolution/src/infra/db/mod.rs | 58 +++ .../src/infra/db/ops_repo.rs | 2 + .../src/infra/db/persons_log_repo.rs | 86 +++++ .../src/infra/identity_persons.rs | 335 ++++++++++++++++++ .../identity-resolution/src/infra/mod.rs | 1 + .../services/identity-resolution/src/main.rs | 32 +- .../identity-resolution/src/seed_runner.rs | 9 +- .../identity-resolution/src/sync_runner.rs | 250 +++++++++++++ .../tests/e2e/identity/test_persons_sync.py | 194 ++++++++++ src/ingestion/tests/e2e/lib/identity.py | 51 +++ 22 files changed, 1654 insertions(+), 55 deletions(-) create mode 100644 src/backend/services/identity-resolution/helm/templates/sync-cronjob.yaml create mode 100644 src/backend/services/identity-resolution/src/api/sync.rs create mode 100644 src/backend/services/identity-resolution/src/domain/sync_service.rs create mode 100644 src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs create mode 100644 src/backend/services/identity-resolution/src/infra/identity_persons.rs create mode 100644 src/backend/services/identity-resolution/src/sync_runner.rs create mode 100644 src/ingestion/tests/e2e/identity/test_persons_sync.py diff --git a/src/backend/Cargo.lock b/src/backend/Cargo.lock index 31790c216..4c334835b 100644 --- a/src/backend/Cargo.lock +++ b/src/backend/Cargo.lock @@ -1433,6 +1433,7 @@ dependencies = [ "bnum", "bstr", "bytes", + "chrono", "cityhash-rs", "clickhouse-macros", "clickhouse-types", diff --git a/src/backend/Cargo.toml b/src/backend/Cargo.toml index 016015dd0..527d7b3ff 100644 --- a/src/backend/Cargo.toml +++ b/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" diff --git a/src/backend/services/identity-resolution/helm/templates/sync-cronjob.yaml b/src/backend/services/identity-resolution/helm/templates/sync-cronjob.yaml new file mode 100644 index 000000000..d333a5fc8 --- /dev/null +++ b/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 }} diff --git a/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py b/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py index 3011be398..e66781cc4 100644 --- a/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py +++ b/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`). + * `.tenantDefaultId` env-overrides the Secret (k8s `env` beats + `envFrom`); + * `.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-`) 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,12 +268,14 @@ 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: @@ -225,7 +283,7 @@ def test_umbrella_accepts_the_explicit_seed_tenant_alone(umbrella_deps) -> None: 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) diff --git a/src/backend/services/identity-resolution/helm/values.yaml b/src/backend/services/identity-resolution/helm/values.yaml index 5f80fd93d..c8836cc15 100644 --- a/src/backend/services/identity-resolution/helm/values.yaml +++ b/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: diff --git a/src/backend/services/identity-resolution/src/api/error.rs b/src/backend/services/identity-resolution/src/api/error.rs index f9afac4a7..b9844f53f 100644 --- a/src/backend/services/identity-resolution/src/api/error.rs +++ b/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~")] diff --git a/src/backend/services/identity-resolution/src/api/mod.rs b/src/backend/services/identity-resolution/src/api/mod.rs index 2b25c955d..762c2e624 100644 --- a/src/backend/services/identity-resolution/src/api/mod.rs +++ b/src/backend/services/identity-resolution/src/api/mod.rs @@ -9,6 +9,7 @@ pub mod person_roles; pub mod roles; pub mod seed; pub mod subchart; +pub mod sync; pub mod visibility; use std::sync::Arc; @@ -110,6 +111,37 @@ fn build_operations(router: Router, openapi: &dyn OpenApiRegistry) -> Router { .handler(seed::list_persons_seed) .register(router, openapi); + // Persons-sync journal (read-only, admin-gated): the sync itself is + // CLI-only (`identity-resolution sync` — see crate::sync_runner), same + // trigger model as the seed after #1690. + let router = OperationBuilder::get("/v1/persons-sync/{id}") + .operation_id("identity_resolution.persons_sync.get") + .summary("Get a persons-sync operation") + .authenticated() + .no_license_required() + .json_response_with_schema::( + openapi, + StatusCode::OK, + "Operation status", + ) + .standard_errors(openapi) + .handler(sync::get_persons_sync) + .register(router, openapi); + + let router = OperationBuilder::get("/v1/persons-sync") + .operation_id("identity_resolution.persons_sync.list") + .summary("List persons-sync operations") + .authenticated() + .no_license_required() + .json_response_with_schema::( + openapi, + StatusCode::OK, + "Operations", + ) + .standard_errors(openapi) + .handler(sync::list_persons_sync) + .register(router, openapi); + // Roles catalogue (admin-gated CRUD over the global `roles` table). let router = OperationBuilder::post("/v1/roles") .operation_id("identity_resolution.roles.create") diff --git a/src/backend/services/identity-resolution/src/api/seed.rs b/src/backend/services/identity-resolution/src/api/seed.rs index 7d5b0327e..ef7e4f9e6 100644 --- a/src/backend/services/identity-resolution/src/api/seed.rs +++ b/src/backend/services/identity-resolution/src/api/seed.rs @@ -72,7 +72,8 @@ impl From for PersonsSeedOperationResponse { /// Surface a stored JSON column as a parsed value (not a double-encoded string); /// `None` for absent/empty/unparseable. Mirrors the .NET `ParseOrNull`. -fn parse_or_null(json: Option<&str>) -> Option { +/// `pub(crate)`: shared with the persons-sync journal (same wire conventions). +pub(crate) fn parse_or_null(json: Option<&str>) -> Option { let s = json?; if s.is_empty() { return None; @@ -83,7 +84,7 @@ fn parse_or_null(json: Option<&str>) -> Option { /// Format a DB `DateTime` (naive) as ISO-8601 with a `T` separator, matching the /// .NET `System.Text.Json` `DateTime` output (`NaiveDateTime::to_string` uses a /// space, which breaks ISO-8601 parsers). -fn fmt_ts(dt: sea_orm::prelude::DateTime) -> String { +pub(crate) fn fmt_ts(dt: sea_orm::prelude::DateTime) -> String { dt.format("%Y-%m-%dT%H:%M:%S%.6f").to_string() } @@ -162,7 +163,7 @@ pub async fn list_persons_seed( /// Map the `?status=` query to a filter. An unknown/blank value is ignored /// (returns all statuses), matching the .NET `_ => null` — not a 400. -fn status_filter(raw: Option<&str>) -> Option { +pub(crate) fn status_filter(raw: Option<&str>) -> Option { match raw { Some("queued") => Some(OperationStatus::Queued), Some("running") => Some(OperationStatus::Running), diff --git a/src/backend/services/identity-resolution/src/api/sync.rs b/src/backend/services/identity-resolution/src/api/sync.rs new file mode 100644 index 000000000..ee8f53b7d --- /dev/null +++ b/src/backend/services/identity-resolution/src/api/sync.rs @@ -0,0 +1,139 @@ +//! Persons-sync operations journal — read-only HTTP surface. +//! +//! The sync itself is CLI-only (`identity-resolution sync`, run by the Helm +//! `CronJob` or a manual Job — see `crate::sync_runner`); like the +//! persons-seed after #1690, there is no HTTP trigger. The GETs are the +//! observability window over the `operations` rows the CLI runs write: +//! status, summary (`rows` / `max_id` / `max_created_at` / `synced_at` — the +//! resolution watermark), error per run. +//! +//! Own DTO types rather than reusing the seed's: same wire conventions +//! (parsed JSON `request`/`summary`, ISO-8601 timestamps, nulls emitted) but +//! an independent OpenAPI schema, free to grow sync-specific fields. +//! +//! Admin-gated like the seed journal: the caller is the gateway-JWT subject +//! and must hold an active `admin` role in the tenant. + +use std::sync::Arc; + +use axum::Json; +use axum::extract::{Extension, Path, Query}; +use axum::response::IntoResponse; +use serde::Serialize; +use toolkit_canonical_errors::CanonicalError; +use toolkit_security::SecurityContext; +use utoipa::ToSchema; +use uuid::Uuid; + +use super::AppState; +use super::error::PersonsSyncError; +use super::gate::require_admin; +use super::seed::ListParams; +use crate::infra::db::ops_repo::{self, Operation, PERSONS_SYNC_OP}; + +/// Default page size / cap for the list endpoint (same as the seed journal). +const LIST_DEFAULT_LIMIT: u64 = 50; +const LIST_MAX_LIMIT: u64 = 500; + +/// One operation's status. Wire shape matches the seed journal's: +/// `request` and `summary` surfaced as parsed JSON, ISO-8601 timestamps, +/// null fields emitted. +#[derive(Debug, Serialize, ToSchema)] +pub struct PersonsSyncOperationResponse { + pub operation_id: Uuid, + pub operation_type: String, + pub status: String, + pub insight_tenant_id: Uuid, + pub author_person_id: Uuid, + #[schema(value_type = Option)] + pub request: Option, + /// On completion: the [`SyncSummary`] — rows copied, `max_id` / + /// `max_created_at` watermarks, `synced_at`. + /// + /// [`SyncSummary`]: crate::domain::sync_service::SyncSummary + #[schema(value_type = Option)] + pub summary: Option, + pub error_message: Option, + pub started_at: String, + pub completed_at: Option, +} +impl toolkit::api::api_dto::ResponseApiDto for PersonsSyncOperationResponse {} + +impl From for PersonsSyncOperationResponse { + fn from(op: Operation) -> Self { + Self { + operation_id: op.operation_id, + operation_type: op.operation_type, + status: op.status.as_db().to_owned(), + insight_tenant_id: op.insight_tenant_id, + author_person_id: op.author_person_id, + request: super::seed::parse_or_null(op.request_json.as_deref()), + summary: super::seed::parse_or_null(op.summary_json.as_deref()), + error_message: op.error_message, + started_at: super::seed::fmt_ts(op.started_at), + completed_at: op.completed_at.map(super::seed::fmt_ts), + } + } +} + +/// List response wrapper (typed for OpenAPI). `next_cursor` is declared but +/// always `null` — same non-paginating contract as the seed journal. +#[derive(Debug, Serialize, ToSchema)] +pub struct PersonsSyncListResponse { + pub items: Vec, + pub next_cursor: Option, +} +impl toolkit::api::api_dto::ResponseApiDto for PersonsSyncListResponse {} + +/// `GET /v1/persons-sync/{id}` — poll one operation. +pub async fn get_persons_sync( + Extension(state): Extension>, + Extension(ctx): Extension, + Path(id): Path, +) -> Result { + let tenant = ctx.subject_tenant_id(); + require_admin(&state.db, &ctx).await?; + let op = ops_repo::get_by_id(&state.db, tenant, id) + .await + .map_err(|e| { + tracing::error!(error = %e, "get operation failed"); + CanonicalError::internal("failed to read operation").create() + })? + .filter(|o| o.operation_type == PERSONS_SYNC_OP) + .ok_or_else(|| { + PersonsSyncError::not_found("operation not found") + .with_resource(id.to_string()) + .create() + })?; + Ok(Json(PersonsSyncOperationResponse::from(op))) +} + +/// `GET /v1/persons-sync` — list persons-sync operations. Optional `?status=` +/// (unknown values ignored) and `?limit=` (default 50, capped 500), same +/// semantics as the seed journal list. +pub async fn list_persons_sync( + Extension(state): Extension>, + Extension(ctx): Extension, + Query(params): Query, +) -> Result { + let tenant = ctx.subject_tenant_id(); + require_admin(&state.db, &ctx).await?; + let status = super::seed::status_filter(params.status.as_deref()); + let limit = params.limit.map_or(LIST_DEFAULT_LIMIT, |l| { + u64::try_from(l).unwrap_or(1).clamp(1, LIST_MAX_LIMIT) + }); + let ops = ops_repo::list(&state.db, tenant, Some(PERSONS_SYNC_OP), status, limit) + .await + .map_err(|e| { + tracing::error!(error = %e, "list operations failed"); + CanonicalError::internal("failed to list operations").create() + })?; + let items = ops + .into_iter() + .map(PersonsSyncOperationResponse::from) + .collect(); + Ok(Json(PersonsSyncListResponse { + items, + next_cursor: None, + })) +} diff --git a/src/backend/services/identity-resolution/src/domain/mod.rs b/src/backend/services/identity-resolution/src/domain/mod.rs index 69acbb196..3b6a3b9c4 100644 --- a/src/backend/services/identity-resolution/src/domain/mod.rs +++ b/src/backend/services/identity-resolution/src/domain/mod.rs @@ -4,3 +4,4 @@ pub mod profile; pub mod seed; pub mod seed_service; pub mod subchart; +pub mod sync_service; diff --git a/src/backend/services/identity-resolution/src/domain/sync_service.rs b/src/backend/services/identity-resolution/src/domain/sync_service.rs new file mode 100644 index 000000000..5f3448c00 --- /dev/null +++ b/src/backend/services/identity-resolution/src/domain/sync_service.rs @@ -0,0 +1,203 @@ +//! Persons sync — copy the MariaDB `persons` observation log into ClickHouse +//! (`identity.identity_persons`) so the metrics pipeline (dbt gold builds) can +//! resolve `email -> person_id` next to the observation tables it already +//! reads. Iteration 1 of the metrics `person_id` rework: the ClickHouse copy is +//! a disposable full snapshot — MariaDB stays the only source of truth, and +//! every run replaces the whole table (self-healing, no incremental state). +//! +//! Shaped like [`run_seed`]: pure orchestration over two ports so the flow is +//! unit-testable without either database. +//! +//! [`run_seed`]: crate::domain::seed_service::run_seed + +use async_trait::async_trait; +use sea_orm::prelude::DateTime; +use serde::Serialize; +use uuid::Uuid; + +/// One row of the `persons` observation log, copied VERBATIM — including +/// nullability (`reason` is nullable per migration 009; NULL and `''` stay +/// distinct in `identity_persons`). Matches the MariaDB schema column-for-column +/// except `value_hash` (a generated convenience column, cheap to recompute in +/// ClickHouse if ever needed). +#[derive(Debug, Clone)] +pub struct PersonsLogRow { + pub id: u64, + pub value_type: String, + pub insight_source_type: String, + pub insight_source_id: Uuid, + pub insight_tenant_id: Uuid, + pub value_id: Option, + pub value_full_text: Option, + pub value: Option, + pub value_effective: Option, + pub person_id: Uuid, + pub author_person_id: Uuid, + pub reason: Option, + pub created_at: DateTime, +} + +/// Reads the full `persons` log from the identity store. +#[async_trait] +pub trait PersonsLogReader: Send + Sync { + /// All rows ordered by `id`. Materialized (not streamed) — same trade-off + /// as the seed's `IdentityInputsReader`: fine at current sizes. + async fn read_all(&self) -> anyhow::Result>; +} + +/// Replaces ClickHouse `identity.identity_persons` with a new snapshot. +#[async_trait] +pub trait IdentityPersonsWriter: Send + Sync { + /// Load `rows` (stamped with `synced_at`) into a staging table and swap it + /// in atomically. Must leave the previous snapshot untouched on any failure. + async fn replace(&self, rows: &[PersonsLogRow], synced_at: DateTime) -> anyhow::Result<()>; +} + +/// What a completed sync reports (stored as the operation's `summary_json`). +#[derive(Debug, Serialize, PartialEq, Eq)] +pub struct SyncSummary { + /// Rows copied into `identity_persons`. + pub rows: u64, + /// Highest log `id` in the snapshot — the resolution watermark; `null` + /// for an empty log. + pub max_id: Option, + /// Latest observation timestamp in the snapshot (ISO-8601), `null` for an + /// empty log. + pub max_created_at: Option, + /// When this snapshot was taken (ISO-8601). Also stamped on every copied + /// row as `_synced_at`. + pub synced_at: String, +} + +/// Copy the whole log through the two ports. An empty log is a valid snapshot +/// (`identity_persons` is emptied) — deleting every person in MariaDB should not leave +/// stale resolutions behind. +/// +/// # Errors +/// +/// Propagates reader/writer failures; the writer contract guarantees the +/// previous snapshot survives them. +pub async fn run_sync( + reader: &dyn PersonsLogReader, + writer: &dyn IdentityPersonsWriter, + now: DateTime, +) -> anyhow::Result { + let rows = reader.read_all().await?; + writer.replace(&rows, now).await?; + + Ok(SyncSummary { + rows: rows.len() as u64, + max_id: rows.iter().map(|r| r.id).max(), + max_created_at: rows.iter().map(|r| r.created_at).max().map(fmt_iso), + synced_at: fmt_iso(now), + }) +} + +/// ISO-8601 with a `T` separator (same rationale as the seed API's `fmt_ts`). +fn fmt_iso(dt: DateTime) -> String { + dt.format("%Y-%m-%dT%H:%M:%S%.6f").to_string() +} + +#[cfg(test)] +mod tests { + use tokio::sync::Mutex; + + use super::*; + + fn row(id: u64, created_at: &str) -> anyhow::Result { + Ok(PersonsLogRow { + id, + value_type: "email".to_owned(), + insight_source_type: "bamboohr".to_owned(), + insight_source_id: Uuid::from_u128(1), + insight_tenant_id: Uuid::from_u128(2), + value_id: Some("a@x.com".to_owned()), + value_full_text: None, + value: None, + value_effective: Some("a@x.com".to_owned()), + person_id: Uuid::from_u128(3), + author_person_id: Uuid::from_u128(4), + reason: None, + created_at: parse(created_at)?, + }) + } + + fn parse(s: &str) -> anyhow::Result { + Ok(DateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S")?) + } + + struct FakeReader(Vec); + #[async_trait] + impl PersonsLogReader for FakeReader { + async fn read_all(&self) -> anyhow::Result> { + Ok(self.0.clone()) + } + } + + /// Records what was written; async `Mutex` because the trait takes `&self`. + #[derive(Default)] + struct FakeWriter { + written: Mutex>, + } + #[async_trait] + impl IdentityPersonsWriter for FakeWriter { + async fn replace(&self, rows: &[PersonsLogRow], synced_at: DateTime) -> anyhow::Result<()> { + *self.written.lock().await = Some((rows.len(), synced_at)); + Ok(()) + } + } + + #[tokio::test] + async fn summarizes_rows_and_watermarks() -> anyhow::Result<()> { + let reader = FakeReader(vec![ + row(7, "2026-07-01 10:00:00")?, + row(3, "2026-07-20 09:30:00")?, + ]); + let writer = FakeWriter::default(); + let now = parse("2026-07-29 12:00:00")?; + + let summary = run_sync(&reader, &writer, now).await?; + + assert_eq!(summary.rows, 2); + assert_eq!(summary.max_id, Some(7)); + assert_eq!( + summary.max_created_at.as_deref(), + Some("2026-07-20T09:30:00.000000") + ); + assert_eq!(summary.synced_at, "2026-07-29T12:00:00.000000"); + let written = writer.written.lock().await.take(); + assert_eq!(written, Some((2, now))); + Ok(()) + } + + #[tokio::test] + async fn empty_log_is_a_valid_snapshot() -> anyhow::Result<()> { + let reader = FakeReader(vec![]); + let writer = FakeWriter::default(); + let now = parse("2026-07-29 12:00:00")?; + + let summary = run_sync(&reader, &writer, now).await?; + + assert_eq!(summary.rows, 0); + assert_eq!(summary.max_id, None); + assert_eq!(summary.max_created_at, None); + // The writer still ran — an empty snapshot must clear the table. + assert!(writer.written.lock().await.is_some()); + Ok(()) + } + + #[tokio::test] + async fn writer_failure_propagates() -> anyhow::Result<()> { + struct FailingWriter; + #[async_trait] + impl IdentityPersonsWriter for FailingWriter { + async fn replace(&self, _: &[PersonsLogRow], _: DateTime) -> anyhow::Result<()> { + anyhow::bail!("clickhouse is down") + } + } + let reader = FakeReader(vec![row(1, "2026-07-01 10:00:00")?]); + let result = run_sync(&reader, &FailingWriter, parse("2026-07-29 12:00:00")?).await; + assert!(result.is_err()); + Ok(()) + } +} diff --git a/src/backend/services/identity-resolution/src/gear.rs b/src/backend/services/identity-resolution/src/gear.rs index a6e5aff7c..06f31fc2d 100644 --- a/src/backend/services/identity-resolution/src/gear.rs +++ b/src/backend/services/identity-resolution/src/gear.rs @@ -107,6 +107,29 @@ pub async fn run_seed( Ok(()) } +/// `sync` subcommand: copy the `persons` log into ClickHouse +/// `identity.identity_persons` once and exit (the metrics resolve source). +/// Same shape as [`run_seed`]. +/// +/// # Errors +/// +/// [`crate::sync_runner::SyncRunError`] — the caller maps each variant to a +/// distinct process exit code. +pub async fn run_sync( + app: &toolkit::bootstrap::AppConfig, + force: bool, +) -> Result<(), crate::sync_runner::SyncRunError> { + let cfg = extract_gear_config(app).map_err(crate::sync_runner::SyncRunError::Failed)?; + if cfg.database_url.is_empty() { + return Err(crate::sync_runner::SyncRunError::Failed(anyhow::anyhow!( + "`gears.identity-resolution.config.database_url` is required for sync" + ))); + } + let summary = crate::sync_runner::run(&cfg, force).await?; + tracing::info!(?summary, "persons-sync run finished"); + Ok(()) +} + impl RestApiCapability for IdentityResolutionGear { fn register_rest( &self, diff --git a/src/backend/services/identity-resolution/src/infra/db/mod.rs b/src/backend/services/identity-resolution/src/infra/db/mod.rs index 1c67487af..ff0fb9aa0 100644 --- a/src/backend/services/identity-resolution/src/infra/db/mod.rs +++ b/src/backend/services/identity-resolution/src/infra/db/mod.rs @@ -23,6 +23,7 @@ pub mod bootstrap; pub mod entities; pub mod ops_repo; pub mod person_roles_repo; +pub mod persons_log_repo; pub mod persons_repo; pub mod roles_repo; pub mod seed_repo; @@ -144,6 +145,63 @@ impl SeedLockGuard { } } +/// Name of the GLOBAL advisory lock serializing persons-sync runs (the sync +/// copies the whole log regardless of tenant, so unlike the seed there is no +/// per-tenant suffix). Same cross-process/cross-instance properties as the +/// seed lock; distinct from it so a sync never contends with a seed. +const SYNC_LOCK: &str = "persons-sync"; + +/// RAII holder of the persons-sync advisory lock — the global sibling of +/// [`SeedLockGuard`], with identical lifetime semantics (owns its dedicated +/// single-connection session; every exit path up to and including process +/// death releases the lock server-side). +pub struct SyncLockGuard { + conn: DatabaseConnection, +} + +impl SyncLockGuard { + /// Try to take the global sync lock without waiting (`GET_LOCK` timeout 0 + /// — a concurrent run fails fast instead of publishing a stale snapshot + /// after the active one). Returns `None` when another run holds it. + /// + /// # Errors + /// + /// Returns an error if the connection or the query fails. + pub async fn try_acquire(database_url: &str) -> anyhow::Result> { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + let conn = connect_single(database_url).await?; + let acquired: Option = conn + .query_one(Statement::from_sql_and_values( + DbBackend::MySql, + "SELECT GET_LOCK(?, 0)", + [SYNC_LOCK.into()], + )) + .await? + .map(|r| r.try_get_by_index::>(0)) + .transpose()? + .flatten(); + if acquired == Some(1) { + Ok(Some(Self { conn })) + } else { + Ok(None) + } + } + + /// Explicit best-effort release (the happy path); dropping the session + /// releases the lock anyway. + pub async fn release(self) { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + let _ = self + .conn + .execute(Statement::from_sql_and_values( + DbBackend::MySql, + "SELECT RELEASE_LOCK(?)", + [SYNC_LOCK.into()], + )) + .await; + } +} + /// Name of the cross-process advisory lock serializing schema migration runs. const MIGRATION_LOCK: &str = "identity_resolution_migrations"; /// How long a second migrator waits for the lock before giving up (seconds). diff --git a/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs b/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs index f5950052e..a1ba5064c 100644 --- a/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs +++ b/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs @@ -18,6 +18,8 @@ use uuid::Uuid; /// `operation_type` value of persons-seed runs — shared by the CLI runner /// (writes) and the `GET /v1/persons-seed*` journal endpoints (filter). pub const PERSONS_SEED_OP: &str = "persons-seed"; +/// Operation type of the persons→ClickHouse sync (`sync` subcommand). +pub const PERSONS_SYNC_OP: &str = "persons-sync"; /// Lifecycle phase of an operation. DB column is a `VARCHAR(16)`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs b/src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs new file mode 100644 index 000000000..6ed06506a --- /dev/null +++ b/src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs @@ -0,0 +1,86 @@ +//! Full-log read of `persons` for the ClickHouse `identity_persons` sync. +//! +//! Unlike the resolver queries in [`persons_repo`], this needs no windowing — +//! the ClickHouse copy carries the raw log verbatim and all "which observation wins" +//! logic lives on the ClickHouse side (the dbt resolve macro). Plain +//! entity scan, ordered by `id` so the snapshot is deterministic. +//! +//! [`persons_repo`]: super::persons_repo + +use async_trait::async_trait; +use sea_orm::{DatabaseConnection, EntityTrait, PaginatorTrait, QueryOrder}; +use uuid::Uuid; + +use super::entities::persons; +use crate::domain::sync_service::{PersonsLogReader, PersonsLogRow}; + +/// [`PersonsLogReader`] over the service's MariaDB pool. +pub struct MariaDbPersonsLogReader<'a> { + db: &'a DatabaseConnection, +} + +impl<'a> MariaDbPersonsLogReader<'a> { + #[must_use] + pub fn new(db: &'a DatabaseConnection) -> Self { + Self { db } + } + + /// Cheap row count for the sync runner's empty-log guard — avoids + /// materializing the whole log just to learn it is empty. + /// + /// # Errors + /// + /// Returns an error if the query fails. + pub async fn count(&self) -> anyhow::Result { + Ok(persons::Entity::find().count(self.db).await?) + } +} + +#[async_trait] +impl PersonsLogReader for MariaDbPersonsLogReader<'_> { + async fn read_all(&self) -> anyhow::Result> { + let models = persons::Entity::find() + .order_by_asc(persons::Column::Id) + .all(self.db) + .await?; + models.into_iter().map(map_row).collect() + } +} + +fn map_row(m: persons::Model) -> anyhow::Result { + Ok(PersonsLogRow { + id: m.id, + value_type: m.value_type, + insight_source_type: m.insight_source_type, + insight_source_id: uuid16(&m.insight_source_id, "insight_source_id", m.id)?, + insight_tenant_id: uuid16(&m.insight_tenant_id, "insight_tenant_id", m.id)?, + value_id: m.value_id, + value_full_text: m.value_full_text, + value: m.value, + value_effective: m.value_effective, + person_id: uuid16(&m.person_id, "person_id", m.id)?, + author_person_id: uuid16(&m.author_person_id, "author_person_id", m.id)?, + // Nullable since migration 009 — copied verbatim, NULL stays NULL. + reason: m.reason, + created_at: m.created_at, + }) +} + +/// BINARY(16) → `Uuid` (canonical big-endian, as written by `Uuid::as_bytes`). +/// A wrong-length value means a corrupt row — fail the sync loudly rather than +/// copy garbage. +fn uuid16(bytes: &[u8], column: &str, row_id: u64) -> anyhow::Result { + Uuid::from_slice(bytes) + .map_err(|e| anyhow::anyhow!("persons.{column} of row id={row_id} is not a UUID: {e}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn uuid16_rejects_wrong_length() { + assert!(uuid16(&[0u8; 15], "person_id", 42).is_err()); + assert!(uuid16(Uuid::from_u128(7).as_bytes(), "person_id", 42).is_ok()); + } +} diff --git a/src/backend/services/identity-resolution/src/infra/identity_persons.rs b/src/backend/services/identity-resolution/src/infra/identity_persons.rs new file mode 100644 index 000000000..ddb9dd57e --- /dev/null +++ b/src/backend/services/identity-resolution/src/infra/identity_persons.rs @@ -0,0 +1,335 @@ +//! ClickHouse writer for `identity.identity_persons` — the persons-log copy +//! the metrics dbt builds resolve against. +//! +//! Full-snapshot replace with an atomic swap: +//! +//! 1. `CREATE TABLE IF NOT EXISTS` the target (first run / dbt-hook parity); +//! 2. create a staging table UNIQUE to this run (suffix = UUIDv7) with the +//! CURRENT schema — concurrent syncs (another replica's worker) can never +//! write into or drop each other's staging, and the swap below upgrades +//! the live table's schema for free on the run after a schema change; +//! 3. stream every row into staging (readers keep seeing the old snapshot); +//! 4. count-verify staging against what we sent — a short write MUST NOT be +//! swapped in; +//! 5. watermark guard: if the live table already carries a `_synced_at` +//! NEWER than this snapshot's, abort — a swap would regress the table. +//! This is a BACKSTOP, not the serialization: concurrent runs are +//! serialized cluster-wide by the persons-sync advisory lock the worker +//! holds around the whole run (`infra::db::persons_sync_lock`), which is +//! what makes check→swap safe. The guard still catches anything that +//! bypasses the worker (a by-hand EXCHANGE, a future lock-free caller); +//! 6. `EXCHANGE TABLES` — atomic, readers never observe an empty/partial +//! table (requires an Atomic database, ClickHouse's default); +//! 7. drop this run's staging (post-swap it holds the previous snapshot); +//! stagings orphaned by crashed runs are garbage-collected at the start +//! of every run once they are an hour old. +//! +//! Any failure before the swap leaves the live table untouched. + +use std::time::Duration; + +use async_trait::async_trait; +use chrono::Utc; +use clickhouse::Row; +use insight_clickhouse::{Client, Config}; +use sea_orm::prelude::DateTime; +use serde::Serialize; +use uuid::Uuid; + +use crate::domain::sync_service::{IdentityPersonsWriter, PersonsLogRow}; + +/// The whole snapshot (DDL + insert + verify) rides one client; generous bound, +/// the sync operation as a whole is separately bounded by the worker. +const WRITE_TIMEOUT: Duration = Duration::from_mins(5); + +const DATABASE: &str = "identity"; +const TARGET: &str = "identity_persons"; +/// Per-run staging tables are `identity_persons_staging_`. +const STAGING_PREFIX: &str = "identity_persons_staging_"; +/// How old an orphaned staging table must be before the GC drops it — old +/// enough that no live run (bounded well under this by `SYNC_TIMEOUT`) can +/// still be writing to it. +const STAGING_GC_AGE_SECONDS: u32 = 3600; + +/// Column block shared by the target and staging DDL. Mirrors the MariaDB +/// `persons` log (`001_persons.sql`, nullability per +/// `009_align_existing_tables_to_conventions.sql`) minus the generated +/// `value_hash`, plus the `_synced_at` watermark (same convention as +/// `identity_inputs`). Keep in sync with the dbt on-run-start hook that +/// creates the empty table for builds that run before the first sync. +const COLUMNS_DDL: &str = r" + id UInt64, + value_type String, + insight_source_type String, + insight_source_id UUID, + insight_tenant_id UUID, + value_id Nullable(String), + value_full_text Nullable(String), + value Nullable(String), + value_effective Nullable(String), + person_id UUID, + author_person_id UUID, + reason Nullable(String), + created_at DateTime64(6, 'UTC'), + _synced_at DateTime64(3, 'UTC') +"; + +/// Wire row for the `RowBinary` insert. Field order and names must match the +/// DDL above — the clickhouse client sends `INSERT INTO … (field names)`. +/// The watermark field is serde-renamed to `_synced_at` (a Rust field can't +/// comfortably live with the underscore prefix under clippy). +#[derive(Debug, Row, Serialize)] +struct WireRow { + id: u64, + value_type: String, + insight_source_type: String, + #[serde(with = "clickhouse::serde::uuid")] + insight_source_id: Uuid, + #[serde(with = "clickhouse::serde::uuid")] + insight_tenant_id: Uuid, + value_id: Option, + value_full_text: Option, + value: Option, + value_effective: Option, + #[serde(with = "clickhouse::serde::uuid")] + person_id: Uuid, + #[serde(with = "clickhouse::serde::uuid")] + author_person_id: Uuid, + reason: Option, + #[serde(with = "clickhouse::serde::chrono::datetime64::micros")] + created_at: chrono::DateTime, + #[serde( + rename = "_synced_at", + with = "clickhouse::serde::chrono::datetime64::millis" + )] + synced_at: chrono::DateTime, +} + +/// [`IdentityPersonsWriter`] over the shared `insight-clickhouse` client. +pub struct ClickHouseIdentityPersonsWriter { + client: Client, +} + +impl ClickHouseIdentityPersonsWriter { + #[must_use] + pub fn new(client: Client) -> Self { + Self { client } + } + + /// Build a writer from connection settings (empty user → no auth). The + /// client database is pinned to `identity` regardless of the configured + /// read database — the table's home is fixed by contract. + #[must_use] + pub fn connect(url: &str, user: &str, password: &str) -> Self { + let mut config = Config::new(url, DATABASE).with_query_timeout(WRITE_TIMEOUT); + if !user.is_empty() { + config = config.with_auth(user, password); + } + Self::new(Client::new(config)) + } + + async fn execute(&self, sql: &str) -> anyhow::Result<()> { + self.client.query(sql).execute().await?; + Ok(()) + } + + /// Drop staging tables orphaned by crashed runs. Only tables older than + /// [`STAGING_GC_AGE_SECONDS`] — a younger one may belong to a live + /// concurrent run on another replica. Best-effort: GC failures must not + /// fail the sync. + async fn drop_stale_stagings(&self) { + let stale: Result, _> = self + .client + .query( + "SELECT name FROM system.tables \ + WHERE database = ? AND name LIKE ? \ + AND metadata_modification_time < now() - INTERVAL ? SECOND", + ) + .bind(DATABASE) + .bind(format!("{STAGING_PREFIX}%")) + .bind(STAGING_GC_AGE_SECONDS) + .fetch_all() + .await; + let stale = match stale { + Ok(names) => names, + Err(e) => { + tracing::warn!(error = %e, "persons-sync: staging GC listing failed (skipped)"); + return; + } + }; + for name in stale { + // Defense in depth: only names matching exactly what this code + // mints (prefix + 32 lowercase hex chars of a simple-format UUID) + // ever reach the identifier-interpolated DROP, even if the LIKE + // above were somehow loosened. + let Some(suffix) = name.strip_prefix(STAGING_PREFIX) else { + continue; + }; + if suffix.len() != 32 || !suffix.bytes().all(|b| b.is_ascii_hexdigit()) { + continue; + } + match self + .execute(&format!("DROP TABLE IF EXISTS {DATABASE}.`{name}`")) + .await + { + Ok(()) => tracing::info!(table = %name, "persons-sync: dropped orphaned staging"), + Err(e) => { + tracing::warn!(error = %e, table = %name, "persons-sync: staging GC drop failed"); + } + } + } + } + + /// Insert + verify + guard + swap against `staging`. Split out so + /// [`replace`](IdentityPersonsWriter::replace) can unconditionally drop this + /// run's staging afterwards, on success and failure alike. + async fn fill_and_swap( + &self, + staging: &str, + rows: &[PersonsLogRow], + synced_at: chrono::DateTime, + ) -> anyhow::Result<()> { + let mut insert = self.client.inner().insert::(staging).await?; + for row in rows { + insert.write(&to_wire_row(row, synced_at)).await?; + } + insert.end().await?; + + // A lost batch must never be swapped in as "the new truth". + let count: u64 = self + .client + .query(&format!("SELECT count() FROM {DATABASE}.`{staging}`")) + .fetch_one() + .await?; + let expected = rows.len() as u64; + anyhow::ensure!( + count == expected, + "staging count mismatch: inserted {expected}, staging holds {count}; \ + aborting swap (live table left untouched)" + ); + + // Watermark guard (empty table → epoch 0, always passes). Equal + // stamps pass: re-publishing an identical-instant snapshot is + // harmless, and the replica clocks feeding `_synced_at` are the + // service's own. + let published_ms: i64 = self + .client + .query(&format!( + "SELECT toUnixTimestamp64Milli(max(_synced_at)) FROM {DATABASE}.{TARGET}" + )) + .fetch_one() + .await?; + anyhow::ensure!( + published_ms <= synced_at.timestamp_millis(), + "a newer snapshot (_synced_at={published_ms}ms) is already published; \ + discarding this run's older snapshot ({}ms)", + synced_at.timestamp_millis() + ); + + self.execute(&format!( + "EXCHANGE TABLES {DATABASE}.`{staging}` AND {DATABASE}.{TARGET}" + )) + .await?; + Ok(()) + } +} + +#[async_trait] +impl IdentityPersonsWriter for ClickHouseIdentityPersonsWriter { + async fn replace(&self, rows: &[PersonsLogRow], synced_at: DateTime) -> anyhow::Result<()> { + let synced_at = synced_at.and_utc(); + // Unique per run: concurrent syncs never touch each other's staging. + let staging = format!("{STAGING_PREFIX}{}", Uuid::now_v7().simple()); + + // The database normally pre-exists (init-identity migration), but a + // fresh environment may not have run it yet — idempotent and cheap. + self.execute(&format!("CREATE DATABASE IF NOT EXISTS {DATABASE}")) + .await?; + // Target first: EXCHANGE requires both sides to exist, and the very + // first sync runs against a cluster that may only have the database. + self.execute(&format!( + "CREATE TABLE IF NOT EXISTS {DATABASE}.{TARGET} ({COLUMNS_DDL}) \ + ENGINE = MergeTree ORDER BY id" + )) + .await?; + self.drop_stale_stagings().await; + + self.execute(&format!( + "CREATE TABLE {DATABASE}.`{staging}` ({COLUMNS_DDL}) \ + ENGINE = MergeTree ORDER BY id" + )) + .await?; + + let result = self.fill_and_swap(&staging, rows, synced_at).await; + + // Unconditional cleanup of THIS run's staging: after a successful swap + // it holds the previous snapshot; after a failure, the partial write. + // Best-effort — an orphan is reclaimed by the next run's GC. + if let Err(e) = self + .execute(&format!("DROP TABLE IF EXISTS {DATABASE}.`{staging}`")) + .await + { + tracing::warn!(error = %e, table = %staging, "persons-sync: dropping own staging failed"); + } + result + } +} + +fn to_wire_row(r: &PersonsLogRow, synced_at: chrono::DateTime) -> WireRow { + WireRow { + id: r.id, + value_type: r.value_type.clone(), + insight_source_type: r.insight_source_type.clone(), + insight_source_id: r.insight_source_id, + insight_tenant_id: r.insight_tenant_id, + value_id: r.value_id.clone(), + value_full_text: r.value_full_text.clone(), + value: r.value.clone(), + value_effective: r.value_effective.clone(), + person_id: r.person_id, + author_person_id: r.author_person_id, + reason: r.reason.clone(), + // MariaDB `TIMESTAMP(6)` comes back naive; the pool session runs in + // UTC, so re-attaching Utc is a re-labeling, not a conversion. + created_at: r.created_at.and_utc(), + synced_at, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::sync_service::PersonsLogRow; + + #[test] + fn maps_log_row_preserving_micros_and_nulls() -> anyhow::Result<()> { + let created = + DateTime::parse_from_str("2026-07-29 12:34:56.123456", "%Y-%m-%d %H:%M:%S%.f")?; + let row = PersonsLogRow { + id: 9, + value_type: "email".to_owned(), + insight_source_type: "bamboohr".to_owned(), + insight_source_id: Uuid::from_u128(1), + insight_tenant_id: Uuid::from_u128(2), + value_id: Some("a@x.com".to_owned()), + value_full_text: None, + value: None, + value_effective: Some("a@x.com".to_owned()), + person_id: Uuid::from_u128(3), + author_person_id: Uuid::from_u128(4), + reason: None, + created_at: created, + }; + let synced = + DateTime::parse_from_str("2026-07-29 13:00:00", "%Y-%m-%d %H:%M:%S")?.and_utc(); + + let wire = to_wire_row(&row, synced); + + assert_eq!(wire.id, 9); + assert_eq!(wire.created_at.timestamp_subsec_micros(), 123_456); + assert_eq!(wire.synced_at, synced); + // A NULL reason stays NULL — the copy is verbatim, not normalized. + assert_eq!(wire.reason, None); + Ok(()) + } +} diff --git a/src/backend/services/identity-resolution/src/infra/mod.rs b/src/backend/services/identity-resolution/src/infra/mod.rs index 005c27481..54e8d57af 100644 --- a/src/backend/services/identity-resolution/src/infra/mod.rs +++ b/src/backend/services/identity-resolution/src/infra/mod.rs @@ -2,3 +2,4 @@ pub mod db; pub mod identity_inputs; +pub mod identity_persons; diff --git a/src/backend/services/identity-resolution/src/main.rs b/src/backend/services/identity-resolution/src/main.rs index 10710037f..a35175d6f 100644 --- a/src/backend/services/identity-resolution/src/main.rs +++ b/src/backend/services/identity-resolution/src/main.rs @@ -15,6 +15,7 @@ mod gear; mod infra; mod migration; mod seed_runner; +mod sync_runner; // System gears — linked via inventory for the REST host and the gateway-JWT auth // pipeline. `use … as _;` is load-bearing: the gears register through `inventory` @@ -71,9 +72,20 @@ enum Commands { #[arg(long)] force: bool, }, + /// Copy the `persons` log into ClickHouse `identity.identity_persons` + /// (the metrics email→`person_id` resolve source) and exit. Same execution + /// model as `seed` — Helm `CronJob` / manual Job; pairs naturally as + /// "sync after seed". Exit codes: 0 ok / 1 failed / 2 another run holds + /// the lock / 3 refused by the empty-log guard. + Sync { + /// Override the empty-log guard (publish an empty snapshot). + #[arg(long)] + force: bool, + }, } -/// Exit codes of the `seed` subcommand, mirrored in the Job monitoring docs. +/// Exit codes of the `seed` / `sync` subcommands (one shared scheme), +/// mirrored in the Job monitoring docs. const EXIT_SEED_FAILED: i32 = 1; const EXIT_SEED_LOCK_BUSY: i32 = 2; const EXIT_SEED_GUARD: i32 = 3; @@ -109,6 +121,24 @@ async fn main() -> Result<()> { } } } + Commands::Sync { force } => { + init_subcommand_logging(); + match gear::run_sync(&config, force).await { + Ok(()) => Ok(()), + Err(sync_runner::SyncRunError::LockBusy) => { + tracing::warn!("another persons-sync run holds the lock; exiting"); + std::process::exit(EXIT_SEED_LOCK_BUSY); + } + Err(sync_runner::SyncRunError::Guard(msg)) => { + tracing::error!(%msg, "persons-sync refused by the empty-log guard"); + std::process::exit(EXIT_SEED_GUARD); + } + Err(sync_runner::SyncRunError::Failed(e)) => { + tracing::error!(error = %format!("{e:#}"), "persons-sync failed"); + std::process::exit(EXIT_SEED_FAILED); + } + } + } } } diff --git a/src/backend/services/identity-resolution/src/seed_runner.rs b/src/backend/services/identity-resolution/src/seed_runner.rs index 707e2ec6b..307910f28 100644 --- a/src/backend/services/identity-resolution/src/seed_runner.rs +++ b/src/backend/services/identity-resolution/src/seed_runner.rs @@ -255,8 +255,13 @@ async fn guarded_seed( /// The pure tenant-resolution decision (split out for unit tests): an /// explicitly configured tenant always wins; an empty config falls back to /// the SOLE tenant present in the persons log; anything ambiguous refuses -/// with an operator-facing message. -fn resolve_tenant(configured: &str, distinct_in_persons: &[Uuid]) -> Result { +/// with an operator-facing message. `pub(crate)`: the sync runner journals +/// its runs under the same resolved tenant (its GET journal routes are +/// tenant-scoped, so a made-up tenant would hide the rows from admins). +pub(crate) fn resolve_tenant( + configured: &str, + distinct_in_persons: &[Uuid], +) -> Result { let configured = configured.trim(); if !configured.is_empty() { return Uuid::parse_str(configured).map_err(|e| format!("invalid tenant_default_id: {e}")); diff --git a/src/backend/services/identity-resolution/src/sync_runner.rs b/src/backend/services/identity-resolution/src/sync_runner.rs new file mode 100644 index 000000000..b334c1dc9 --- /dev/null +++ b/src/backend/services/identity-resolution/src/sync_runner.rs @@ -0,0 +1,250 @@ +//! CLI persons-sync runner — the engine behind the `sync` subcommand. +//! +//! Copies the MariaDB `persons` observation log into ClickHouse +//! `identity.identity_persons` (full snapshot, atomic swap — see +//! `infra::identity_persons`) so the metrics dbt builds can resolve +//! `email -> person_id`. CLI-only from the start, mirroring the persons-seed +//! shape (#1690): a Helm `CronJob` / manual Job runs +//! `identity-resolution sync` — no HTTP trigger, no auth; only the GET +//! journal routes remain on the API. The natural pairing is "sync right +//! after seed" — the seed rewrites the log, the sync publishes it. +//! +//! One run: advisory lock → zombie sweep → `operations` journal row → log +//! read → guard → snapshot replace → journal completed/failed. +//! +//! Concurrency: runs serialize on a GLOBAL MariaDB `GET_LOCK` +//! (`infra::db::SyncLockGuard` — global, not per-tenant: the sync copies the +//! whole log). This is the actual serialization of the publish step; the +//! writer's `_synced_at` watermark guard is a backstop for anything that +//! bypasses the runner. A concurrent run fails fast +//! ([`SyncRunError::LockBusy`], exit code 2). +//! +//! Guard (overridable with `--force`, recorded as a `failed` operation so +//! the journal explains why nothing was published): an EMPTY `persons` log. +//! An empty read usually means a misconfigured database or a wiped stand, +//! not "nobody exists" — and publishing it would atomically erase a +//! populated mirror (the destructive-zero-rows lesson of the seed, #1550). +//! A deliberate wipe is what `--force` is for; the domain layer itself +//! treats an empty snapshot as valid. + +use std::time::Duration; + +use uuid::Uuid; + +use crate::config::GearConfig; +use crate::domain::sync_service::{SyncSummary, run_sync}; +use crate::infra::db::{self, ops_repo, persons_log_repo::MariaDbPersonsLogReader, seed_repo}; +use crate::infra::identity_persons::ClickHouseIdentityPersonsWriter; +use crate::seed_runner::{SYSTEM_AUTHOR, resolve_tenant}; + +/// Upper bound on the read + replace — a healthy run is seconds; this only +/// trips on a real MariaDB/ClickHouse stall, failing the Job instead of +/// wedging it into the `CronJob`'s next tick. +const SYNC_TIMEOUT: Duration = Duration::from_mins(5); + +/// Backstop over the WHOLE lock-held critical section (sweep + journal +/// writes are outside [`SYNC_TIMEOUT`]'s scope). A run cut off here may +/// leave its `operations` row `running`; the next run's zombie sweep +/// reclaims it. +const RUN_TIMEOUT: Duration = Duration::from_mins(7); + +/// How stale a `queued`/`running` operation must be before the pre-run +/// sweep reclaims it (same convention as the seed runner). +const ZOMBIE_CUTOFF_HOURS: i64 = 1; + +/// Why a sync run did not complete — the `sync` subcommand maps each variant +/// to a distinct exit code (0 ok / 1 failed / 2 lock busy / 3 guard), same +/// scheme as the seed. +#[derive(Debug)] +pub enum SyncRunError { + /// Another run holds the global sync advisory lock. + LockBusy, + /// The guard refused the run (operator-facing message, persisted + /// verbatim as the operation's `error_message`). + Guard(String), + /// The run itself failed (connect, read, replace, or journal write). + Failed(anyhow::Error), +} + +impl From for SyncRunError { + fn from(e: anyhow::Error) -> Self { + Self::Failed(e) + } +} + +/// Run one CLI persons-sync end to end. See the module docs for the shape. +/// +/// # Errors +/// +/// [`SyncRunError`] — lock busy, guard refusal, or a failed run. +pub async fn run(config: &GearConfig, force: bool) -> Result { + let db = db::connect(&config.database_url).await?; + + // The sync itself is tenant-agnostic (whole-log copy), but its journal + // row needs a real tenant: the GET journal routes are tenant-scoped, so + // rows under a made-up tenant would be invisible to the admins who need + // them. Same resolution rule as the seed (configured || sole-in-log). + let distinct = seed_repo::distinct_tenants(&db, 2).await?; + let tenant = match resolve_tenant(&config.tenant_default_id, &distinct) { + Ok(t) => t, + Err(msg) => return Err(SyncRunError::Failed(anyhow::anyhow!(msg))), + }; + + let Some(lock) = db::SyncLockGuard::try_acquire(&config.database_url).await? else { + return Err(SyncRunError::LockBusy); + }; + + let result = tokio::time::timeout(RUN_TIMEOUT, run_locked(&db, config, tenant, force)) + .await + .unwrap_or_else(|_| { + Err(SyncRunError::Failed(anyhow::anyhow!( + "persons-sync run timed out after {}s inside the lock-held critical section", + RUN_TIMEOUT.as_secs() + ))) + }); + + lock.release().await; + result +} + +/// Everything that happens while the advisory lock is held: zombie sweep, +/// journal row, guarded sync, journal resolution. +async fn run_locked( + db: &sea_orm::DatabaseConnection, + config: &GearConfig, + tenant: Uuid, + force: bool, +) -> Result { + // Reclaim rows a killed run left behind. Log-only failure: a broken + // sweep must not block the sync itself. + let cutoff = chrono::Utc::now().naive_utc() - chrono::Duration::hours(ZOMBIE_CUTOFF_HOURS); + match ops_repo::sweep_zombies(db, cutoff).await { + Ok(n) if n > 0 => tracing::warn!(swept = n, "persons-sync: reclaimed zombie operations"), + Ok(_) => {} + Err(e) => tracing::error!(error = %e, "persons-sync: zombie sweep failed"), + } + + // Journal row first, so every later failure (guard included) is recorded + // and visible over the GET /v1/persons-sync endpoints. + let operation_id = Uuid::now_v7(); + let request_json = serde_json::json!({ "trigger": "cli", "force": force }).to_string(); + ops_repo::enqueue( + db, + operation_id, + ops_repo::PERSONS_SYNC_OP, + tenant, + SYSTEM_AUTHOR, + Some(&request_json), + ) + .await?; + ops_repo::try_start(db, operation_id).await?; + tracing::info!(%operation_id, %tenant, force, "persons-sync: cli run started"); + + match guarded_sync(db, config, force).await { + Ok(summary) => { + let summary_json = serde_json::to_string(&summary).unwrap_or_else(|_| "{}".to_owned()); + ops_repo::complete(db, operation_id, &summary_json).await?; + tracing::info!(%operation_id, ?summary, "persons-sync: completed"); + Ok(summary) + } + Err(SyncRunError::Guard(msg)) => { + // Deliberate operator-facing text — safe to persist verbatim. + tracing::warn!(%operation_id, %msg, "persons-sync: refused by guard"); + if let Err(e) = ops_repo::fail(db, operation_id, &msg).await { + tracing::error!(error = %e, %operation_id, "fail update failed"); + } + Err(SyncRunError::Guard(msg)) + } + Err(e) => { + // Persist only a generic message: `error_message` is returned + // verbatim by the GET endpoints, so raw driver/anyhow text must + // not leak to callers. + if let Err(e2) = + ops_repo::fail(db, operation_id, "persons-sync failed; see job logs").await + { + tracing::error!(error = %e2, %operation_id, "fail update failed"); + } + Err(e) + } + } +} + +/// Log read → guard → snapshot replace, bounded by [`SYNC_TIMEOUT`]. +async fn guarded_sync( + db: &sea_orm::DatabaseConnection, + config: &GearConfig, + force: bool, +) -> Result { + let reader = MariaDbPersonsLogReader::new(db); + let writer = ClickHouseIdentityPersonsWriter::connect( + &config.clickhouse_url, + &config.clickhouse_user, + &config.clickhouse_password, + ); + + let run = async { + let rows = reader.count().await?; + if let Err(msg) = empty_log_guard(rows, force) { + return Err(SyncRunError::Guard(msg)); + } + run_sync(&reader, &writer, chrono::Utc::now().naive_utc()) + .await + .map_err(SyncRunError::Failed) + }; + + tokio::time::timeout(SYNC_TIMEOUT, run) + .await + .unwrap_or_else(|_| { + Err(SyncRunError::Failed(anyhow::anyhow!( + "persons-sync timed out after {}s", + SYNC_TIMEOUT.as_secs() + ))) + }) +} + +/// The pure guard decision: refuse publishing an EMPTY log unless `--force`. +/// An empty read is far more often a misconfigured database / wiped stand +/// than a real "no people", and publishing it atomically erases a populated +/// mirror. The message is operator-facing. +fn empty_log_guard(log_rows: u64, force: bool) -> Result<(), String> { + if force || log_rows > 0 { + return Ok(()); + } + Err( + "empty-log guard: the persons log has 0 rows — publishing would erase the \ + ClickHouse snapshot (misconfigured database_url? wiped stand?); re-run with \ + --force to publish the empty snapshot deliberately" + .to_owned(), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_log_refused_and_names_the_fix() -> anyhow::Result<()> { + let Err(msg) = empty_log_guard(0, false) else { + anyhow::bail!("empty log must be refused"); + }; + assert!(msg.contains("--force"), "{msg}"); + Ok(()) + } + + #[test] + fn force_publishes_the_empty_snapshot() { + assert!(empty_log_guard(0, true).is_ok()); + } + + #[test] + fn non_empty_log_passes() { + assert!(empty_log_guard(1, false).is_ok()); + } + + #[tokio::test] + async fn default_config_fails_cleanly() { + let cfg = crate::config::GearConfig::default(); + let err = run(&cfg, false).await; + assert!(matches!(err, Err(SyncRunError::Failed(_)))); + } +} diff --git a/src/ingestion/tests/e2e/identity/test_persons_sync.py b/src/ingestion/tests/e2e/identity/test_persons_sync.py new file mode 100644 index 000000000..0e2cc932a --- /dev/null +++ b/src/ingestion/tests/e2e/identity/test_persons_sync.py @@ -0,0 +1,194 @@ +"""Contract: persons-sync — the `sync` CLI trigger + the GET journal routes. + +The sync copies the ENTIRE MariaDB `persons` observation log into ClickHouse +`identity.identity_persons` (full snapshot, atomic swap) so the metrics +pipeline can resolve `email -> person_id` at dbt-build time. CLI-only from +birth, mirroring the persons-seed trigger model (#1690): `identity-resolution +sync` run synchronously (Helm CronJob / manual Job), exit codes +0 ok / 1 failed / 2 lock busy / 3 empty-log guard; only the GET journal +routes exist over HTTP. Rust-only module (`supports_persons_sync`) — on the +`dotnet` run every case skips. + +The fixture dataset (lib/identity_seed.py) gives the log a deterministic +floor — the fixture people's observations — which the end-to-end case +asserts against in ClickHouse, keyed by fixed person UUIDs. + +The empty-log guard (exit 3) is deliberately NOT covered here: triggering it +requires an empty `persons` table, and this suite shares one seeded database +across modules — wiping it would destroy the fixture tree every read test +depends on. The guard decision is a pure function unit-tested in the binary +(sync_runner::tests). + +Unlike persons-seed there is no tenant scoping to isolate: the sync copies +the whole log verbatim (single-tenant reality, #1550) and writes nothing back +to MariaDB, so it cannot disturb the fixture tree. The journal row lands +under TEST_TENANT_ID (passed as the run's tenant) so the fixture admin +(alice) reads it over the GETs. +""" + +from __future__ import annotations + +import uuid + +import pytest + +from identity.contract import items_of +from lib import clickhouse +from lib import identity_seed as seed +from lib.config import TEST_TENANT_ID, SessionConfig + +pytestmark = [pytest.mark.identity, pytest.mark.mutating] + +IDENTITY_PERSONS = "identity.identity_persons" + +# Author the CLI stamps on its journal rows (no JWT on that path) — the +# SYSTEM_AUTHOR nil-UUID convention shared with the seed CLI. +SYSTEM_AUTHOR = "00000000-0000-0000-0000-000000000000" + + +@pytest.fixture(autouse=True) +def _rust_only(identity_svc): + if not identity_svc.supports_persons_sync: + pytest.skip("the sync CLI exists only in the Rust implementation") + + +def _run_sync(identity_svc, **kwargs): + """One CLI run under the fixture tenant, asserted successful.""" + proc = identity_svc.run_sync_cli(tenant=str(TEST_TENANT_ID), **kwargs) + assert proc.returncode == 0, f"rc={proc.returncode}\nstdout={proc.stdout}\nstderr={proc.stderr}" + return proc + + +def _latest_op(api) -> dict: + """Newest persons-sync journal row (the CLI is synchronous, so the run + that just returned is terminal and first in the DESC-ordered list).""" + r = api.get("/v1/persons-sync?limit=1") + assert r.status_code == 200, f"status={r.status_code} body={r.text}" + ops = items_of(r.json()) + assert len(ops) == 1, ops + return ops[0] + + +def _snapshot_count(cfg: SessionConfig) -> int: + return clickhouse.query(cfg, f"SELECT count() FROM {IDENTITY_PERSONS}")[0][0] + + +def test_persons_sync_end_to_end(identity_svc, api, compose_stack: SessionConfig) -> None: + """CLI exit 0 → journal row completed (system author, cli trigger) → + ClickHouse holds the snapshot: row count matches the summary, the fixture + people's email observations arrived with their fixed person UUIDs, and + the watermark is stamped.""" + _run_sync(identity_svc) + + op = _latest_op(api) + assert op["status"] == "completed", op + assert op["operation_type"] == "persons-sync", op + assert op["author_person_id"] == SYSTEM_AUTHOR, op + assert (op.get("request") or {}).get("trigger") == "cli", op + summary = op.get("summary") or {} + # The fixture dataset alone puts dozens of observations in the log. + assert summary.get("rows", 0) > 0, op + assert summary.get("max_id"), op + assert summary.get("synced_at"), op + + # identity_persons carries EXACTLY the snapshot the summary reports — the + # binary's own count-verify ran before the swap; this re-checks it + # end-to-end through an independent client. + assert _snapshot_count(compose_stack) == summary["rows"] + + # Fixture people arrived intact: raw email observation, fixed person UUID. + rows = clickhouse.query( + compose_stack, + f"SELECT DISTINCT person_id FROM {IDENTITY_PERSONS} " # noqa: S608 — fixed table, fixed test literal + f"WHERE value_type = 'email' AND value_id = '{seed.ALICE_EMAIL}'", + ) + assert [str(row[0]) for row in rows] == [str(seed.ALICE)], rows + + # The shared-email pair kept DISTINCT person ids — identity_persons is the + # raw log, not a resolution: collapsing dup1/dup2 is the resolve macro's + # call, downstream in dbt. + dup_rows = clickhouse.query( + compose_stack, + f"SELECT DISTINCT person_id FROM {IDENTITY_PERSONS} " # noqa: S608 — fixed table, fixed test literal + f"WHERE value_type = 'email' AND value_id = '{seed.DUP_EMAIL}'", + ) + assert {str(row[0]) for row in dup_rows} == {str(seed.DUP1), str(seed.DUP2)}, dup_rows + + # Every copied row is stamped with the run's watermark. + stamped = clickhouse.query( + compose_stack, + f"SELECT count() FROM {IDENTITY_PERSONS} WHERE _synced_at > toDateTime64(0, 3)", + ) + assert stamped[0][0] == summary["rows"], stamped + + +def test_persons_sync_replaces_not_appends( + identity_svc, api, compose_stack: SessionConfig +) -> None: + """Two consecutive runs leave the table equal to ONE snapshot — the + replace-swap semantics: a re-run must never double it.""" + _run_sync(identity_svc) + first = _latest_op(api) + assert first["status"] == "completed", first + _run_sync(identity_svc) + second = _latest_op(api) + assert second["status"] == "completed", second + assert second["operation_id"] != first["operation_id"], (first, second) + + # The log only grows (append-only), and the table equals the LAST + # snapshot — not the sum of both. + assert second["summary"]["rows"] >= first["summary"]["rows"], (first, second) + assert _snapshot_count(compose_stack) == second["summary"]["rows"] + + +def test_persons_sync_journal_get_by_id(identity_svc, api) -> None: + """The single-operation GET returns the run the list shows.""" + _run_sync(identity_svc) + listed = _latest_op(api) + r = api.get(f"/v1/persons-sync/{listed['operation_id']}") + assert r.status_code == 200, f"status={r.status_code} body={r.text}" + assert r.json()["operation_id"] == listed["operation_id"], r.json() + assert r.json()["status"] == "completed", r.json() + + +def test_persons_sync_journal_lists_only_sync_ops(identity_svc, api) -> None: + """Operation-type scoping both ways: the sync journal carries ONLY + persons-sync rows, and a sync operation is 404 on the SEED journal — one + `operations` table, two disjoint API surfaces.""" + _run_sync(identity_svc) + ops = items_of(api.get("/v1/persons-sync").json()) + assert ops, "at least the run above must be listed" + assert all(op["operation_type"] == "persons-sync" for op in ops), ops + + sync_op = ops[0]["operation_id"] + r = api.get(f"/v1/persons-seed/{sync_op}") + assert r.status_code == 404, f"status={r.status_code} body={r.text}" + seed_ops = items_of(api.get("/v1/persons-seed").json()) + assert sync_op not in {op["operation_id"] for op in seed_ops}, seed_ops + + +def test_persons_sync_journal_status_filter(identity_svc, api) -> None: + """`?status=` filters: a finished CLI run is `completed` (the CLI is + synchronous — no queued/running to race against) and absent from the + `failed` view.""" + _run_sync(identity_svc) + op = _latest_op(api) + completed = items_of(api.get("/v1/persons-sync?status=completed").json()) + assert op["operation_id"] in {o["operation_id"] for o in completed}, completed + failed = items_of(api.get("/v1/persons-sync?status=failed").json()) + assert op["operation_id"] not in {o["operation_id"] for o in failed}, failed + + +def test_persons_sync_get_unknown_id_404(api) -> None: + r = api.get(f"/v1/persons-sync/{uuid.uuid4()}") + assert r.status_code == 404, f"status={r.status_code} body={r.text}" + + +def test_persons_sync_journal_403_non_admin(bob_api) -> None: + """bob is not an admin anywhere — the journal is refused.""" + assert bob_api.get("/v1/persons-sync").status_code == 403 + assert bob_api.get(f"/v1/persons-sync/{uuid.uuid4()}").status_code == 403 + + +def test_persons_sync_journal_401_unauthenticated(anon_api) -> None: + assert anon_api.get("/v1/persons-sync").status_code == 401 diff --git a/src/ingestion/tests/e2e/lib/identity.py b/src/ingestion/tests/e2e/lib/identity.py index 5bc4e3fb4..b3630eea8 100644 --- a/src/ingestion/tests/e2e/lib/identity.py +++ b/src/ingestion/tests/e2e/lib/identity.py @@ -97,6 +97,14 @@ def supports_seed_cli(implementation: str) -> bool: return implementation == "rust" +def supports_persons_sync(implementation: str) -> bool: + """Whether the binary has the `sync` subcommand (copy the `persons` log + into ClickHouse `identity.identity_persons`, the metrics email→person_id + resolve source) + its GET journal routes. A NEW Rust-only surface, + deliberately never backported to the frozen, outgoing .NET service.""" + return implementation == "rust" + + def supports_strict_input_validation(implementation: str) -> bool: """Strict input validation added by the Rust service (reviewed on epic #1602): too-long revoke `reason` → 400, present-but-nil @@ -215,6 +223,49 @@ def supports_seed_http_trigger(self) -> bool: def supports_seed_cli(self) -> bool: return supports_seed_cli(self.implementation) + @property + def supports_persons_sync(self) -> bool: + return supports_persons_sync(self.implementation) + + def run_sync_cli( + self, + *, + tenant: str | None, + force: bool = False, + timeout_s: float = 300.0, + extra_env: dict[str, str] | None = None, + ) -> subprocess.CompletedProcess[str]: + """Run `identity-resolution sync` — copy the `persons` log into + ClickHouse `identity.identity_persons`. Synchronous: when it returns, + the run's `operations` row is terminal. Exit codes: 0 ok / 1 failed / + 2 lock busy / 3 empty-log guard. + + `tenant` scopes the run's JOURNAL row (the copy itself is + tenant-agnostic); pass the tenant whose admin will read the journal. + """ + if not self.supports_persons_sync: + raise ApiSpawnError( + f"the sync CLI exists only on the rust implementation " + f"(selected: {self.implementation})" + ) + cmd = locate_rust_app(self.cfg) + env = self._rust_env() + if tenant is not None: + env["APP__gears__identity-resolution__config__tenant_default_id"] = tenant + if extra_env: + env.update(extra_env) + args = [*cmd, "-c", str(self._rig_config_path), "sync"] + if force: + args.append("--force") + return subprocess.run( # noqa: S603 — harness-controlled argv + args, + env=env, + capture_output=True, + text=True, + timeout=timeout_s, + check=False, + ) + def run_seed_cli( self, *,