diff --git a/deploy/seed/generators/people.py b/deploy/seed/generators/people.py index f4101c7f4..25ac58fe9 100644 --- a/deploy/seed/generators/people.py +++ b/deploy/seed/generators/people.py @@ -158,6 +158,11 @@ def seed_identity_persons( `value_effective` is what the macro reads (lowercased and trimmed on both sides of the join); `id` is the resolution tiebreak, so it must be distinct and stable per person or which row wins becomes arbitrary. + + Each person also gets a `value_type='id'` binding row carrying a synthetic + source account, because that is what `person_account_assignments_current` + projects — an email-only log leaves that relation empty and every + account-keyed lookup silently resolves nothing. """ truncate(client, "identity", "identity_persons") @@ -167,6 +172,7 @@ def seed_identity_persons( "insight_source_type", "insight_source_id", "insight_tenant_id", + "value_id", "value_effective", "person_id", "author_person_id", @@ -177,24 +183,46 @@ def seed_identity_persons( author = deterministic_uuid("identity_persons", "author") stamped = _dt.datetime(2026, 1, 1, tzinfo=_dt.UTC) - rows: list[tuple[object, ...]] = [ + # The whole roster, not `_measured_persons`: the admin operator holds no + # activity but still has to RESOLVE, or any request naming them reads as + # an unknown person rather than a person with nothing. + emails: list[tuple[object, ...]] = [ ( index + 1, "email", "seed", source_id, tenant_uuid, + None, p.email.lower(), p.uuid, author, stamped, stamped, ) - # The whole roster, not `_measured_persons`: the admin operator holds no - # activity but still has to RESOLVE, or any request naming them reads as - # an unknown person rather than a person with nothing. for index, p in enumerate(roster) ] + + # `id` rows carry the account in `value_id`; ids continue past the email + # block so the resolution tiebreak stays distinct across both kinds. + bindings: list[tuple[object, ...]] = [ + ( + len(roster) + index + 1, + "id", + "seed", + source_id, + tenant_uuid, + f"seed-account-{index + 1}", + f"seed-account-{index + 1}", + p.uuid, + author, + stamped, + stamped, + ) + for index, p in enumerate(roster) + ] + + rows = emails + bindings return bulk_insert(client, "identity", "identity_persons", cols, rows) diff --git a/docs/components/backend/identity-resolution/openapi.json b/docs/components/backend/identity-resolution/openapi.json index 308bbe37d..319cd124c 100644 --- a/docs/components/backend/identity-resolution/openapi.json +++ b/docs/components/backend/identity-resolution/openapi.json @@ -2,7 +2,6 @@ "components": { "schemas": { "AttributeReconcileListResponse": { - "description": "List response wrapper (typed for OpenAPI); `next_cursor` is declared but\nalways `null`, same non-paginating contract as the other journals.", "properties": { "items": { "items": { @@ -23,7 +22,6 @@ "type": "object" }, "AttributeReconcileOperationResponse": { - "description": "One reconcile operation's status.", "properties": { "author_person_id": { "format": "uuid", @@ -65,7 +63,6 @@ "type": "string" }, "summary": { - "description": "On completion: the [`ReconcileSummary`] — discovered / created /\nrefreshed / `skipped_invalid` counts.\n\n[`ReconcileSummary`]: crate::domain::attribute_reconcile::ReconcileSummary", "type": [ "object", "null" @@ -161,7 +158,6 @@ "type": "object" }, "PersonAttributeListResponse": { - "description": "List response wrapper (typed for OpenAPI).", "properties": { "items": { "items": { @@ -176,7 +172,6 @@ "type": "object" }, "PersonAttributeResponse": { - "description": "One attribute definition with its current policy.", "properties": { "first_observed_at": { "type": "string" @@ -532,8 +527,85 @@ ], "type": "object" }, + "PolicyPublishListResponse": { + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/PolicyPublishOperationResponse" + }, + "type": "array" + }, + "next_cursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "items" + ], + "type": "object" + }, + "PolicyPublishOperationResponse": { + "properties": { + "author_person_id": { + "format": "uuid", + "type": "string" + }, + "completed_at": { + "type": [ + "string", + "null" + ] + }, + "error_message": { + "type": [ + "string", + "null" + ] + }, + "insight_tenant_id": { + "format": "uuid", + "type": "string" + }, + "operation_id": { + "format": "uuid", + "type": "string" + }, + "operation_type": { + "type": "string" + }, + "request": { + "type": [ + "object", + "null" + ] + }, + "started_at": { + "type": "string" + }, + "status": { + "type": "string" + }, + "summary": { + "type": [ + "object", + "null" + ] + } + }, + "required": [ + "operation_id", + "operation_type", + "status", + "insight_tenant_id", + "author_person_id", + "started_at" + ], + "type": "object" + }, "PolicyResponse": { - "description": "The current policy revision of one definition.", "properties": { "actor_person_id": { "format": "uuid", @@ -583,7 +655,6 @@ "type": "object" }, "PolicyUpdateRequest": { - "description": "Full policy body for the next revision. `expected_revision` is the\nrevision the caller read; a stale value yields 409.", "properties": { "comparison_enabled": { "type": "boolean" @@ -950,7 +1021,6 @@ "type": "object" }, "ValueModeDto": { - "description": "Declared value mode of an attribute, as an OpenAPI enum.", "enum": [ "single", "multi" @@ -1174,6 +1244,283 @@ "summary": "List discovered person attributes with their current policy (admin)" } }, + "/v1/person-attributes-policy-publish": { + "get": { + "operationId": "identity_resolution.person_attributes_policy_publish.list", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PolicyPublishListResponse" + } + } + }, + "description": "Operations" + }, + "400": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Bad Request" + }, + "401": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Unauthorized" + }, + "403": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Forbidden" + }, + "404": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Not Found" + }, + "409": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Conflict" + }, + "429": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Too Many Requests" + }, + "500": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Internal Server Error" + } + }, + "security": [ + { + "bearerAuth": [] + } + ], + "summary": "List policy-publish operations" + }, + "post": { + "operationId": "identity_resolution.person_attributes_policy_publish.create", + "responses": { + "202": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PolicyPublishOperationResponse" + } + } + }, + "description": "Publish accepted; poll the operation" + }, + "400": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Bad Request" + }, + "401": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Unauthorized" + }, + "403": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Forbidden" + }, + "404": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Not Found" + }, + "409": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Conflict" + }, + "429": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Too Many Requests" + }, + "500": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Internal Server Error" + } + }, + "security": [ + { + "bearerAuth": [] + } + ], + "summary": "Trigger a policy-snapshot publish (admin)" + } + }, + "/v1/person-attributes-policy-publish/{id}": { + "get": { + "operationId": "identity_resolution.person_attributes_policy_publish.get", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PolicyPublishOperationResponse" + } + } + }, + "description": "Operation status" + }, + "400": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Bad Request" + }, + "401": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Unauthorized" + }, + "403": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Forbidden" + }, + "404": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Not Found" + }, + "409": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Conflict" + }, + "429": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Too Many Requests" + }, + "500": { + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + }, + "description": "Internal Server Error" + } + }, + "security": [ + { + "bearerAuth": [] + } + ], + "summary": "Poll one policy-publish operation" + } + }, "/v1/person-attributes-reconcile": { "get": { "operationId": "identity_resolution.person_attributes_reconcile.list", diff --git a/src/backend/Cargo.lock b/src/backend/Cargo.lock index 2d6a42f4b..523439cca 100644 --- a/src/backend/Cargo.lock +++ b/src/backend/Cargo.lock @@ -3118,6 +3118,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-util", "tracing", "tracing-subscriber", "utoipa", diff --git a/src/backend/services/identity-resolution/Cargo.toml b/src/backend/services/identity-resolution/Cargo.toml index 80bd41b3c..638d10363 100644 --- a/src/backend/services/identity-resolution/Cargo.toml +++ b/src/backend/services/identity-resolution/Cargo.toml @@ -34,6 +34,7 @@ types_registry = { workspace = true } anyhow = { workspace = true } tokio = { workspace = true } +tokio-util = "0.7" clap = { workspace = true } # Domain gear: config deserialization, async trait impls, HTTP router type, diff --git a/src/backend/services/identity-resolution/helm/templates/publish-policy-cronjob.yaml b/src/backend/services/identity-resolution/helm/templates/publish-policy-cronjob.yaml new file mode 100644 index 000000000..bdf621d3f --- /dev/null +++ b/src/backend/services/identity-resolution/helm/templates/publish-policy-cronjob.yaml @@ -0,0 +1,102 @@ +{{- if .Values.publishPolicy.enabled }} +# Scheduled policy publish: copies the registry's current per-attribute +# policy into ClickHouse `identity.person_attribute_policy_snapshot`, the +# relation the query path enforces policy from. Hourly and offset from the +# reconcile run, so a field registered in the morning is publishable the same +# hour and an admin's policy edit reaches analytics within the hour. +# +# Same image/config/secret wiring as the seed/sync CronJobs; the +# `publish-policy` subcommand runs once and exits (exit codes: 0 ok / +# 1 failed / 2 lock busy). An unchanged policy set short-circuits after +# verifying the published snapshot still matches, so most runs write nothing. +# Runs serialize on a 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" . }}-publish-policy publish-manual-$USER +apiVersion: batch/v1 +kind: CronJob +metadata: + name: {{ include "insight-identity-resolution.fullname" . }}-publish-policy + labels: + {{- include "insight-identity-resolution.labels" . | nindent 4 }} + app.kubernetes.io/component: policy-publish +spec: + schedule: {{ .Values.publishPolicy.schedule | quote }} + concurrencyPolicy: {{ .Values.publishPolicy.concurrencyPolicy }} + successfulJobsHistoryLimit: {{ .Values.publishPolicy.successfulJobsHistoryLimit }} + failedJobsHistoryLimit: {{ .Values.publishPolicy.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 publish timeout. + backoffLimit: {{ .Values.publishPolicy.backoffLimit }} + activeDeadlineSeconds: {{ .Values.publishPolicy.activeDeadlineSeconds }} + template: + metadata: + # NOT the shared selectorLabels: the Service selects on + # name+instance alone, and a job pod carrying them would enter the + # Service's endpoints (it listens on nothing). + labels: + app.kubernetes.io/name: identity-resolution-policy-publish + app.kubernetes.io/instance: {{ .Release.Name }} + app.kubernetes.io/component: policy-publish + 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: policy-publish + 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", "publish-policy"] + volumeMounts: + - name: identity-resolution-config + mountPath: /app/config/insight.yaml + subPath: insight.yaml + readOnly: true + # The gears host creates `server.home_dir` at boot; the + # subcommand never starts grpc-hub, so /app/data is the only + # writable path it needs under a read-only root. + - name: data + mountPath: /app/data + envFrom: + # Same Secret as the deployment: database_url, clickhouse_*, + # tenant_default_id — everything the publish needs. The + # ClickHouse user needs CREATE/INSERT/EXCHANGE in the + # `identity` database (the snapshot-swap write path), which + # persons-sync already requires. + - secretRef: + name: {{ required "existingSecret is required (umbrella provides `insight-identity-resolution-config`; standalone installs supply their own)" .Values.existingSecret | quote }} + {{- with .Values.publishPolicy.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 + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + resources: + {{- toYaml .Values.publishPolicy.resources | nindent 16 }} + volumes: + - name: identity-resolution-config + configMap: + name: {{ include "insight-identity-resolution.fullname" . }}-gears-config + - name: data + emptyDir: {} +{{- 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 2446dcd1b..153c4693c 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 @@ -50,6 +50,7 @@ "seed": ("30 6 * * *", "seed", "seed"), "sync": ("45 6 * * *", "sync", "sync"), "reconcile-attributes": ("0 7 * * *", "reconcile-attributes", "reconcileAttributes"), + "publish-policy": ("15 * * * *", "publish-policy", "publishPolicy"), } # Minimum viable subchart install (mirrors the umbrella's wiring). diff --git a/src/backend/services/identity-resolution/helm/values.yaml b/src/backend/services/identity-resolution/helm/values.yaml index 1653e4f33..7f75e08ac 100644 --- a/src/backend/services/identity-resolution/helm/values.yaml +++ b/src/backend/services/identity-resolution/helm/values.yaml @@ -150,6 +150,33 @@ reconcileAttributes: cpu: 500m memory: 512Mi +publishPolicy: + enabled: true + # Journal-row tenant (UUID); same override semantics as seed.tenantDefaultId + # (the snapshot spans every tenant in the registry — this only scopes the + # operations journal the admin GET routes read). + tenantDefaultId: "" + # Hourly, offset from the reconcile run at 07:00 so newly registered fields + # publish the same hour; an unchanged policy set short-circuits. + schedule: "15 * * * *" + # 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 publish 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/attribute_reconcile.rs b/src/backend/services/identity-resolution/src/api/attribute_reconcile.rs index 19dedf6a7..f93b61623 100644 --- a/src/backend/services/identity-resolution/src/api/attribute_reconcile.rs +++ b/src/backend/services/identity-resolution/src/api/attribute_reconcile.rs @@ -1,12 +1,3 @@ -//! Attribute-reconcile operations journal — read-only HTTP surface. -//! -//! The reconcile itself is CLI-only (`identity-resolution -//! reconcile-attributes`, run by the Helm `CronJob` or a manual Job — see -//! `crate::attribute_reconcile_runner`); these GETs are the observability -//! window over its `operations` rows: status, summary (discovered / created / -//! refreshed / skipped counts), error per run. Same wire conventions and -//! admin gate as the seed and sync journals. - use std::sync::Arc; use axum::Json; @@ -27,7 +18,6 @@ use crate::infra::db::ops_repo::{self, Operation, PERSON_ATTRIBUTES_RECONCILE_OP const LIST_DEFAULT_LIMIT: u64 = 50; const LIST_MAX_LIMIT: u64 = 500; -/// One reconcile operation's status. #[derive(Debug, Serialize, ToSchema)] pub struct AttributeReconcileOperationResponse { pub operation_id: Uuid, @@ -37,10 +27,6 @@ pub struct AttributeReconcileOperationResponse { pub author_person_id: Uuid, #[schema(value_type = Option)] pub request: Option, - /// On completion: the [`ReconcileSummary`] — discovered / created / - /// refreshed / `skipped_invalid` counts. - /// - /// [`ReconcileSummary`]: crate::domain::attribute_reconcile::ReconcileSummary #[schema(value_type = Option)] pub summary: Option, pub error_message: Option, @@ -66,8 +52,6 @@ impl From for AttributeReconcileOperationResponse { } } -/// List response wrapper (typed for OpenAPI); `next_cursor` is declared but -/// always `null`, same non-paginating contract as the other journals. #[derive(Debug, Serialize, ToSchema)] pub struct AttributeReconcileListResponse { pub items: Vec, @@ -75,7 +59,6 @@ pub struct AttributeReconcileListResponse { } impl toolkit::api::api_dto::ResponseApiDto for AttributeReconcileListResponse {} -/// `GET /v1/person-attributes-reconcile/{id}` — poll one operation. pub async fn get_attribute_reconcile( Extension(state): Extension>, Extension(ctx): Extension, @@ -98,9 +81,6 @@ pub async fn get_attribute_reconcile( Ok(Json(AttributeReconcileOperationResponse::from(op))) } -/// `GET /v1/person-attributes-reconcile` — list reconcile operations. -/// Optional `?status=` (unknown values ignored) and `?limit=` (default 50, -/// capped 500), same semantics as the seed and sync journals. pub async fn list_attribute_reconcile( Extension(state): Extension>, Extension(ctx): Extension, diff --git a/src/backend/services/identity-resolution/src/api/mod.rs b/src/backend/services/identity-resolution/src/api/mod.rs index 5f84acf35..b9dc2b0b9 100644 --- a/src/backend/services/identity-resolution/src/api/mod.rs +++ b/src/backend/services/identity-resolution/src/api/mod.rs @@ -8,6 +8,7 @@ mod gate; mod handlers; pub mod person_attributes; pub mod person_roles; +pub mod policy_publish; pub mod roles; pub mod seed; pub mod subchart; @@ -33,6 +34,9 @@ pub struct AppState { pub db: DatabaseConnection, /// Gear config (`org_chart_source_type`, `clickhouse_*`, …). pub config: GearConfig, + /// Host shutdown signal, so request-spawned runs can end their journal row + /// instead of vanishing mid-write. + pub cancel: tokio_util::sync::CancellationToken, } /// Mount the identity-resolution routes onto the host's router. @@ -225,6 +229,48 @@ fn build_operations(router: Router, openapi: &dyn OpenApiRegistry) -> Router { .handler(attribute_reconcile::list_attribute_reconcile) .register(router, openapi); + let router = OperationBuilder::post("/v1/person-attributes-policy-publish") + .operation_id("identity_resolution.person_attributes_policy_publish.create") + .summary("Trigger a policy-snapshot publish (admin)") + .authenticated() + .no_license_required() + .json_response_with_schema::( + openapi, + StatusCode::ACCEPTED, + "Publish accepted; poll the operation", + ) + .standard_errors(openapi) + .handler(policy_publish::create_policy_publish) + .register(router, openapi); + + let router = OperationBuilder::get("/v1/person-attributes-policy-publish/{id}") + .operation_id("identity_resolution.person_attributes_policy_publish.get") + .summary("Poll one policy-publish operation") + .authenticated() + .no_license_required() + .json_response_with_schema::( + openapi, + StatusCode::OK, + "Operation status", + ) + .standard_errors(openapi) + .handler(policy_publish::get_policy_publish) + .register(router, openapi); + + let router = OperationBuilder::get("/v1/person-attributes-policy-publish") + .operation_id("identity_resolution.person_attributes_policy_publish.list") + .summary("List policy-publish operations") + .authenticated() + .no_license_required() + .json_response_with_schema::( + openapi, + StatusCode::OK, + "Operations", + ) + .standard_errors(openapi) + .handler(policy_publish::list_policy_publish) + .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/person_attributes.rs b/src/backend/services/identity-resolution/src/api/person_attributes.rs index f8bd2e059..fec8b58e9 100644 --- a/src/backend/services/identity-resolution/src/api/person_attributes.rs +++ b/src/backend/services/identity-resolution/src/api/person_attributes.rs @@ -1,17 +1,3 @@ -//! Person-attribute registry — admin HTTP surface. -//! -//! Definitions are discovered by the `reconcile-attributes` CLI run (see -//! `crate::attribute_reconcile_runner`); this surface reads them and revises -//! their policy. A policy revision is append-only: `PUT …/policy` writes the -//! next revision, never mutates one, and carries the caller as its actor. -//! Concurrency is optimistic — the request names the revision it saw, and a -//! stale value is a 409 `aborted` (the canonical model has no 422; see the -//! divergence note in `super::error`). -//! -//! Tenant scope: definitions store the RAW warehouse tenant string; the -//! caller's gateway-JWT tenant UUID is matched against it in canonical -//! string form. - use std::sync::Arc; use axum::Json; @@ -30,7 +16,6 @@ use crate::infra::db::person_attributes_repo::{ self, DefinitionWithPolicy, PolicyInput, ValueMode, }; -/// One attribute definition with its current policy. #[derive(Debug, Serialize, ToSchema)] pub struct PersonAttributeResponse { pub id: Uuid, @@ -43,7 +28,6 @@ pub struct PersonAttributeResponse { } impl toolkit::api::api_dto::ResponseApiDto for PersonAttributeResponse {} -/// The current policy revision of one definition. #[derive(Debug, Serialize, ToSchema)] pub struct PolicyResponse { pub revision: i32, @@ -57,14 +41,12 @@ pub struct PolicyResponse { pub reason: String, } -/// List response wrapper (typed for OpenAPI). #[derive(Debug, Serialize, ToSchema)] pub struct PersonAttributeListResponse { pub items: Vec, } impl toolkit::api::api_dto::ResponseApiDto for PersonAttributeListResponse {} -/// Declared value mode of an attribute, as an OpenAPI enum. #[derive(Debug, Clone, Copy, Deserialize, ToSchema)] #[serde(rename_all = "snake_case")] pub enum ValueModeDto { @@ -81,8 +63,6 @@ impl From for ValueMode { } } -/// Full policy body for the next revision. `expected_revision` is the -/// revision the caller read; a stale value yields 409. #[derive(Debug, Deserialize, ToSchema)] pub struct PolicyUpdateRequest { pub expected_revision: i32, @@ -130,8 +110,6 @@ fn to_response(d: DefinitionWithPolicy) -> PersonAttributeResponse { } } -/// `GET /v1/person-attributes` — list the tenant's discovered attribute -/// definitions with their current policy. pub async fn list_person_attributes( Extension(state): Extension>, Extension(ctx): Extension, @@ -151,7 +129,6 @@ pub async fn list_person_attributes( Ok(Json(PersonAttributeListResponse { items })) } -/// `GET /v1/person-attributes/{id}` — one definition with its current policy. pub async fn get_person_attribute( Extension(state): Extension>, Extension(ctx): Extension, @@ -174,10 +151,6 @@ pub async fn get_person_attribute( Ok(Json(to_response(definition))) } -/// `PUT /v1/person-attributes/{id}/policy` — append the next policy revision. -/// The 200 body re-reads the current state, which a concurrent append may -/// already have advanced past the revision this call wrote — acceptable -/// under optimistic concurrency (the response always shows current truth). pub async fn put_person_attribute_policy( Extension(state): Extension>, Extension(ctx): Extension, diff --git a/src/backend/services/identity-resolution/src/api/policy_publish.rs b/src/backend/services/identity-resolution/src/api/policy_publish.rs new file mode 100644 index 000000000..f531aedfa --- /dev/null +++ b/src/backend/services/identity-resolution/src/api/policy_publish.rs @@ -0,0 +1,253 @@ +use std::sync::Arc; + +use axum::Json; +use axum::extract::{Extension, Path, Query}; +use axum::http::{StatusCode, header::LOCATION}; +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::{AccessError, PersonAttributeError}; +use super::gate::require_admin; +use super::seed::ListParams; +use crate::infra::db::ops_repo::{self, Operation, PERSON_ATTRIBUTES_POLICY_PUBLISH_OP}; +use crate::infra::db::{self}; +use crate::publish_policy_runner::{self, PublishTrigger}; + +const LIST_DEFAULT_LIMIT: u64 = 50; +const LIST_MAX_LIMIT: u64 = 500; + +#[derive(Debug, Serialize, ToSchema)] +pub struct PolicyPublishOperationResponse { + 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, + #[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 PolicyPublishOperationResponse {} + +impl From for PolicyPublishOperationResponse { + 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), + } + } +} + +impl PolicyPublishOperationResponse { + fn queued( + operation_id: Uuid, + tenant: Uuid, + author: Uuid, + trigger: PublishTrigger, + started_at: sea_orm::prelude::DateTime, + ) -> Self { + Self { + operation_id, + operation_type: PERSON_ATTRIBUTES_POLICY_PUBLISH_OP.to_owned(), + status: "queued".to_owned(), + insight_tenant_id: tenant, + author_person_id: author, + request: super::seed::parse_or_null(Some(trigger.request_json())), + summary: None, + error_message: None, + started_at: super::seed::fmt_ts(started_at), + completed_at: None, + } + } +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct PolicyPublishListResponse { + pub items: Vec, + pub next_cursor: Option, +} +impl toolkit::api::api_dto::ResponseApiDto for PolicyPublishListResponse {} + +fn publish_in_progress() -> CanonicalError { + PersonAttributeError::aborted("a policy publish is already running") + .with_reason("PUBLISH_IN_PROGRESS") + .create() +} + +fn publish_unavailable() -> CanonicalError { + CanonicalError::internal("failed to start policy publish").create() +} + +fn tenant_mismatch() -> CanonicalError { + AccessError::failed_precondition() + .with_precondition_violation( + "tenant", + "the caller's tenant is not the tenant this deployment journals policy publishes under", + "tenant_mismatch", + ) + .create() +} + +pub async fn create_policy_publish( + Extension(state): Extension>, + Extension(ctx): Extension, +) -> Result { + let author = require_admin(&state.db, &ctx).await?; + + let tenant = publish_policy_runner::resolve_journal_tenant(&state.db, &state.config) + .await + .map_err(|e| { + tracing::error!(error = %e, "policy-publish: journal tenant unresolved"); + publish_unavailable() + })?; + if tenant != ctx.subject_tenant_id() { + return Err(tenant_mismatch()); + } + + let Some(lock) = db::PolicyPublishLockGuard::try_acquire(&state.config.database_url) + .await + .map_err(|e| { + tracing::error!(error = %e, "policy-publish: lock acquire failed"); + publish_unavailable() + })? + else { + return Err(publish_in_progress()); + }; + + let started_at = chrono::Utc::now().naive_utc(); + let operation_id = + publish_policy_runner::enqueue_run(&state.db, tenant, author, PublishTrigger::Http) + .await + .map_err(|e| { + tracing::error!(error = %e, "policy-publish: enqueue failed"); + publish_unavailable() + })?; + + tokio::spawn(publish_policy_runner::run_detached( + state.db.clone(), + state.config.clone(), + state.cancel.clone(), + tenant, + operation_id, + lock, + )); + + let body = PolicyPublishOperationResponse::queued( + operation_id, + tenant, + author, + PublishTrigger::Http, + started_at, + ); + let location = format!("/v1/person-attributes-policy-publish/{operation_id}"); + Ok((StatusCode::ACCEPTED, [(LOCATION, location)], Json(body))) +} + +pub async fn get_policy_publish( + 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 == PERSON_ATTRIBUTES_POLICY_PUBLISH_OP) + .ok_or_else(|| { + PersonAttributeError::not_found("operation not found") + .with_resource(id.to_string()) + .create() + })?; + Ok(Json(PolicyPublishOperationResponse::from(op))) +} + +pub async fn list_policy_publish( + 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(PERSON_ATTRIBUTES_POLICY_PUBLISH_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(PolicyPublishOperationResponse::from) + .collect(); + Ok(Json(PolicyPublishListResponse { + items, + next_cursor: None, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_concurrent_publish_is_refused_as_409() { + assert_eq!(publish_in_progress().status_code(), StatusCode::CONFLICT); + } + + #[test] + fn a_caller_outside_the_journal_tenant_is_refused_as_400() { + assert_eq!(tenant_mismatch().status_code(), StatusCode::BAD_REQUEST); + } + + #[test] + fn the_accepted_body_reports_queued_with_no_outcome_yet() -> anyhow::Result<()> { + let started = + sea_orm::prelude::DateTime::parse_from_str("2026-08-06 12:00:00", "%Y-%m-%d %H:%M:%S")?; + let id = Uuid::from_u128(9); + + let body = PolicyPublishOperationResponse::queued( + id, + Uuid::from_u128(1), + Uuid::from_u128(2), + PublishTrigger::Http, + started, + ); + + assert_eq!(body.operation_id, id); + assert_eq!(body.status, "queued"); + assert_eq!(body.summary, None); + assert_eq!(body.completed_at, None); + assert_eq!(body.error_message, None); + Ok(()) + } +} diff --git a/src/backend/services/identity-resolution/src/attribute_reconcile_runner.rs b/src/backend/services/identity-resolution/src/attribute_reconcile_runner.rs index 52d26d436..daceb6608 100644 --- a/src/backend/services/identity-resolution/src/attribute_reconcile_runner.rs +++ b/src/backend/services/identity-resolution/src/attribute_reconcile_runner.rs @@ -1,25 +1,3 @@ -//! CLI attribute-reconcile runner — the engine behind the -//! `reconcile-attributes` subcommand. -//! -//! Discovers person-attribute fields in the warehouse claim relation and -//! registers them in the MariaDB registry (definition + default revision-1 -//! policy: grouping allowed, comparison denied). Same execution model as the -//! seed/sync runners: a Helm `CronJob` / manual Job runs -//! `identity-resolution reconcile-attributes`; only GET journal routes exist -//! on the API. -//! -//! One run: advisory lock → zombie sweep → `operations` journal row → -//! discover → guards → register → journal completed/failed. -//! -//! Guards (exit code 3, journalled as `failed` so the journal explains why -//! nothing was registered): -//! - the claims relation does not exist — this service can deploy ahead of -//! the ingestion release that creates it; refusing beats a red `CronJob` -//! that alerts until the other repo ships. -//! - fields were discovered but every one had empty key components — a run -//! that green-completes registering nothing would hide a broken claims -//! contract indefinitely (see `domain::attribute_reconcile`). - use std::time::Duration; use async_trait::async_trait; @@ -39,8 +17,6 @@ const RECONCILE_TIMEOUT: Duration = Duration::from_mins(5); const RUN_TIMEOUT: Duration = Duration::from_mins(7); const ZOMBIE_CUTOFF_HOURS: i64 = 1; -/// Why a reconcile run did not complete — mapped to the shared exit-code -/// scheme (0 ok / 1 failed / 2 lock busy / 3 guard). #[derive(Debug)] pub enum ReconcileRunError { LockBusy, @@ -54,11 +30,6 @@ impl From for ReconcileRunError { } } -/// Run one CLI attribute reconciliation end to end. -/// -/// # Errors -/// -/// [`ReconcileRunError`] — lock busy, guard refusal, or a failed run. pub async fn run(config: &GearConfig) -> Result { let db = db::connect(&config.database_url).await?; @@ -206,9 +177,6 @@ async fn guarded_reconcile( }) } -/// Adapter feeding already-read fields through the domain reader port, so the -/// missing-relation classification stays in the infra layer while the domain -/// loop keeps a single entry point. struct PreRead(Vec); #[async_trait] diff --git a/src/backend/services/identity-resolution/src/domain/attribute_reconcile.rs b/src/backend/services/identity-resolution/src/domain/attribute_reconcile.rs index e2bdb97df..3795473ed 100644 --- a/src/backend/services/identity-resolution/src/domain/attribute_reconcile.rs +++ b/src/backend/services/identity-resolution/src/domain/attribute_reconcile.rs @@ -1,16 +1,9 @@ -//! Attribute reconciliation core: turns fields discovered in the warehouse -//! claim relations into registry registrations. Pure over its two ports — -//! a reader of discovered fields and a registrar — so the loop, its guards -//! and its accounting are testable without a database. - use async_trait::async_trait; use serde::Serialize; use uuid::Uuid; use crate::infra::db::person_attributes_repo::{DefinitionKey, RegisterOutcome}; -/// One distinct (tenant, source type, source instance, field) seen in claims, -/// with its latest observation instant (warehouse-formatted timestamp). #[derive(Debug, Clone)] pub struct DiscoveredField { pub insight_tenant_id: String, @@ -20,13 +13,11 @@ pub struct DiscoveredField { pub last_observed_at: String, } -/// Port: reads the distinct discovered fields from the warehouse. #[async_trait] pub trait DiscoveredFieldsReader { async fn discover(&self) -> anyhow::Result>; } -/// Port: registers one discovered field in the registry. #[async_trait] pub trait FieldRegistrar { async fn register( @@ -36,24 +27,17 @@ pub trait FieldRegistrar { ) -> anyhow::Result; } -/// Accounting of one reconciliation run, journalled as `summary_json`. #[derive(Debug, Default, Serialize, PartialEq, Eq)] pub struct ReconcileSummary { pub discovered: usize, pub created: usize, pub refreshed: usize, pub skipped_invalid: usize, - /// Fields registered under a tenant string that is not a canonical UUID. - /// They persist and reconcile fine but no gateway JWT can ever address - /// them through the admin API, which matches tenants in canonical form — - /// a nonzero count is the only signal of that mismatch. pub non_canonical_tenants: usize, } -/// Why a reconciliation run refused to proceed. #[derive(Debug)] pub enum ReconcileError { - /// Guard refusal with an operator-facing message (journalled verbatim). Guard(String), Failed(anyhow::Error), } @@ -68,9 +52,6 @@ fn is_canonical_uuid(tenant: &str) -> bool { Uuid::parse_str(tenant).is_ok_and(|u| u.to_string() == tenant) } -/// A field is registrable only when every key component is non-empty; claims -/// normalize absent values to `''`, and an empty component would mint a -/// definition no request could ever address. fn to_key(field: &DiscoveredField) -> Option { let all_present = !field.insight_tenant_id.is_empty() && !field.insight_source_type.is_empty() @@ -84,15 +65,6 @@ fn to_key(field: &DiscoveredField) -> Option { }) } -/// Register every valid discovered field. Guards rather than completes when -/// fields were read but none was registrable — a green run that registered -/// nothing would hide a broken contract indefinitely. An EMPTY read is fine -/// (no claims yet), distinct from an all-invalid read. -/// -/// # Errors -/// -/// [`ReconcileError::Guard`] on the all-invalid case; `Failed` when a port -/// fails. pub async fn run_reconcile( reader: &dyn DiscoveredFieldsReader, registrar: &dyn FieldRegistrar, diff --git a/src/backend/services/identity-resolution/src/gear.rs b/src/backend/services/identity-resolution/src/gear.rs index a9cbedab5..59b4ca966 100644 --- a/src/backend/services/identity-resolution/src/gear.rs +++ b/src/backend/services/identity-resolution/src/gear.rs @@ -33,12 +33,13 @@ impl Gear for IdentityResolutionGear { tracing::info!("starting identity-resolution gear"); // Self-managed MariaDB pool (same approach as the analytics gear). - // No background workers: the persons-seed runs as the `seed` CLI - // subcommand (CronJob / manual Job — see `crate::seed_runner`), so the - // server process only serves reads + the operations journal. let db = crate::infra::db::connect(&config.database_url).await?; - let state = AppState { db, config }; + let state = AppState { + db, + config, + cancel: ctx.cancellation_token().clone(), + }; self.state .set(Arc::new(state)) .map_err(|_| anyhow::anyhow!("{} gear already initialized", Self::MODULE_NAME))?; @@ -155,6 +156,30 @@ pub async fn run_reconcile_attributes( Ok(()) } +/// `publish-policy` subcommand: publish the current person-attribute policy +/// snapshot to ClickHouse once and exit. Same shape as [`run_seed`]. +/// +/// # Errors +/// +/// [`crate::publish_policy_runner::PublishRunError`] — the caller maps each +/// variant to a distinct process exit code. +pub async fn run_publish_policy( + app: &toolkit::bootstrap::AppConfig, +) -> Result<(), crate::publish_policy_runner::PublishRunError> { + let cfg = + extract_gear_config(app).map_err(crate::publish_policy_runner::PublishRunError::Failed)?; + if cfg.database_url.is_empty() { + return Err(crate::publish_policy_runner::PublishRunError::Failed( + anyhow::anyhow!( + "`gears.identity-resolution.config.database_url` is required for publish-policy" + ), + )); + } + let summary = crate::publish_policy_runner::run(&cfg).await?; + tracing::info!(?summary, "policy-publish run finished"); + Ok(()) +} + impl RestApiCapability for IdentityResolutionGear { fn register_rest( &self, diff --git a/src/backend/services/identity-resolution/src/infra/attribute_claims.rs b/src/backend/services/identity-resolution/src/infra/attribute_claims.rs index 0e5f5e12d..918c2970a 100644 --- a/src/backend/services/identity-resolution/src/infra/attribute_claims.rs +++ b/src/backend/services/identity-resolution/src/infra/attribute_claims.rs @@ -1,19 +1,3 @@ -//! ClickHouse reader for `silver.class_person_attribute_claims` — discovers -//! which source fields exist so the attribute reconciliation can register -//! them. Aggregate-only read: one row per distinct -//! (tenant, source type, source instance, field) with its latest observation. -//! -//! Multi-tenant raw scan, same posture as `identity_inputs.rs`: claims carry -//! RAW connector-config tenant strings, deployments are single-tenant, and -//! the registry stores the raw string (see `sql/015_person_attributes.sql`). -//! -//! `silver.` is fully qualified because the client's configured database is -//! `identity`; the deployed ClickHouse user therefore needs a `silver` grant. -//! Aliases differ from source column names (ClickHouse "Cyclic aliases" -//! gotcha, same as `identity_inputs.rs`). FINAL is required: the relation is -//! `ReplacingMergeTree` and pre-merge duplicates would be harmless for -//! `max()` but not for future aggregate changes — keep the read canonical. - use std::time::Duration; use async_trait::async_trait; @@ -37,10 +21,6 @@ const DISCOVER_SQL: &str = r" ORDER BY tenant, source_type, source_instance, field "; -/// ClickHouse exception token for a missing table — the pre-ingestion-release -/// deploy case the runner turns into a guard refusal rather than a failure. -/// Matched as the named token, not the numeric code: a bare "Code: 60" -/// substring would also match codes 600-609. const UNKNOWN_TABLE_TOKEN: &str = "UNKNOWN_TABLE"; #[derive(Debug, Row, Deserialize)] @@ -52,7 +32,6 @@ struct FieldRow { last_observed: String, } -/// Reads discovered attribute fields from ClickHouse via the shared client. pub struct ClickHouseDiscoveredFieldsReader { client: Client, } @@ -63,7 +42,6 @@ impl ClickHouseDiscoveredFieldsReader { Self { client } } - /// Build a reader from connection settings (empty user → no auth). #[must_use] pub fn connect(url: &str, database: &str, user: &str, password: &str) -> Self { let mut config = Config::new(url, database).with_query_timeout(READ_TIMEOUT); @@ -74,7 +52,6 @@ impl ClickHouseDiscoveredFieldsReader { } } -/// Read outcome distinguishing "relation absent" from real failures. pub enum DiscoverOutcome { Fields(Vec), ClaimsRelationMissing, @@ -89,15 +66,6 @@ impl DiscoveredFieldsReader for ClickHouseDiscoveredFieldsReader { } impl ClickHouseDiscoveredFieldsReader { - /// Like [`DiscoveredFieldsReader::discover`], but classifies a missing - /// claims relation as its own outcome so the runner can refuse (guard) - /// instead of failing: this service can deploy ahead of the ingestion - /// release that creates the relation. - /// - /// # Errors - /// - /// Returns an error for any ClickHouse failure other than the missing - /// relation. pub async fn discover_or_missing(&self) -> anyhow::Result { match self .client @@ -114,6 +82,8 @@ impl ClickHouseDiscoveredFieldsReader { } } +// WORKAROUND: matched on the named token rather than the numeric code — a bare +// "Code: 60" substring also matches codes 600-609. fn is_unknown_table(err: &clickhouse::error::Error) -> bool { matches!(err, clickhouse::error::Error::BadResponse(msg) if msg.contains(UNKNOWN_TABLE_TOKEN)) } 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 b93687e72..6fb122566 100644 --- a/src/backend/services/identity-resolution/src/infra/db/mod.rs +++ b/src/backend/services/identity-resolution/src/infra/db/mod.rs @@ -257,6 +257,55 @@ impl ReconcileLockGuard { } } +/// Name of the GLOBAL advisory lock serializing policy-publish runs. +const POLICY_PUBLISH_LOCK: &str = "person-attributes-policy-publish"; + +/// RAII holder of the policy-publish advisory lock. +pub struct PolicyPublishLockGuard { + conn: DatabaseConnection, +} + +impl PolicyPublishLockGuard { + /// Try to take the policy-publish lock without waiting; `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)", + [POLICY_PUBLISH_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; dropping the session releases 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(?)", + [POLICY_PUBLISH_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 96fc36de7..01cdb687b 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 @@ -1,27 +1,29 @@ //! Operations audit/job-tracking store (MariaDB `operations` table). //! -//! An async operation (persons-seed) moves `queued → running → completed/failed`. -//! The POST handler enqueues a row; the worker flips it to `running` -//! (`try_start`, so two workers can't double-run), then `complete`s or `fail`s -//! it. GETs poll status. SQL ported from the .NET `Sql.Operations.cs`. +//! An operation moves `queued → running → completed/failed`. Whoever starts a run +//! enqueues the row, flips it with `try_start` (atomic, so two runs can't double-start), +//! then `complete`s or `fail`s it. GETs poll status. SQL ported from the .NET +//! `Sql.Operations.cs`. //! //! Raw SQL on the self-managed pool (like the rest of `infra::db`): the atomic -//! `queued→running` transition (`try_start`) and the cross-tenant startup -//! `sweep_zombies` are conditional DML that `toolkit-db`'s scoped builder can't -//! express, so the whole repo stays on raw SQL for consistency. Values are -//! bound params; see `infra::db` module docs + constructorfabric/gears-rust#4239. +//! `queued→running` transition (`try_start`) and `sweep_zombies` are conditional DML +//! that `toolkit-db`'s scoped builder can't express, so the whole repo stays on raw SQL +//! for consistency. Values are bound params; see `infra::db` module docs + +//! constructorfabric/gears-rust#4239. use sea_orm::prelude::DateTime; use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; 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). +/// (writes) and the 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"; /// Operation type of the attribute-reconcile runner. pub const PERSON_ATTRIBUTES_RECONCILE_OP: &str = "person-attributes-reconcile"; +/// Operation type of the policy-snapshot publisher. +pub const PERSON_ATTRIBUTES_POLICY_PUBLISH_OP: &str = "person-attributes-policy-publish"; /// Lifecycle phase of an operation. DB column is a `VARCHAR(16)`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -246,15 +248,14 @@ pub async fn list( } /// Fail every `queued`/`running` operation whose `started_at` is older than -/// `older_than`. Run once at worker startup so a pod restart cannot leave a row -/// stuck in `running` forever (its in-memory job is gone). Intentionally NOT -/// tenant-scoped — the single-process worker owns all in-flight operations -/// across tenants. Mirrors `Sql.Operations.cs::SweepZombies`. Returns the number -/// of rows reclaimed. +/// `older_than`, so a killed run cannot leave a row stuck in `running` forever. +/// Called at the head of every run. Mirrors `Sql.Operations.cs::SweepZombies`. +/// Returns the number of rows reclaimed. /// /// # Errors /// /// Returns an error if the update fails. +// INVARIANT: deliberately not tenant-scoped — the sweep reclaims rows of every tenant. pub async fn sweep_zombies(db: &DatabaseConnection, older_than: DateTime) -> anyhow::Result { const SQL: &str = r" UPDATE operations diff --git a/src/backend/services/identity-resolution/src/infra/db/person_attributes_repo.rs b/src/backend/services/identity-resolution/src/infra/db/person_attributes_repo.rs index abad67215..a17386d28 100644 --- a/src/backend/services/identity-resolution/src/infra/db/person_attributes_repo.rs +++ b/src/backend/services/identity-resolution/src/infra/db/person_attributes_repo.rs @@ -1,15 +1,6 @@ -//! Person-attribute registry: connector-discovered attribute definitions and -//! their append-only policy revisions. -//! -//! Definitions are keyed by the RAW string identifiers the warehouse claim -//! relations carry (see the deviation note in `sql/015_person_attributes.sql`). -//! Policy revisions never mutate; the current policy is the row with the -//! highest revision per definition, and every revision carries its actor. - use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, SqlErr, Statement}; use uuid::Uuid; -/// Stable identity of one discovered source field. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct DefinitionKey { pub insight_tenant_id: String, @@ -18,7 +9,6 @@ pub struct DefinitionKey { pub source_field_id: String, } -/// One registry row joined with its current (highest-revision) policy. #[derive(Debug, Clone)] pub struct DefinitionWithPolicy { pub id: Uuid, @@ -28,7 +18,6 @@ pub struct DefinitionWithPolicy { pub policy: PolicyRevision, } -/// One immutable policy revision (read model). #[derive(Debug, Clone)] pub struct PolicyRevision { pub revision: i32, @@ -42,8 +31,6 @@ pub struct PolicyRevision { pub reason: String, } -/// Policy fields a caller may set; revision and actor are assigned by the -/// append itself. #[derive(Debug, Clone)] pub struct PolicyInput { pub label_override: Option, @@ -70,9 +57,6 @@ impl ValueMode { } } - /// # Errors - /// - /// Returns an error for a value outside the `value_mode` enum. pub fn from_db(s: &str) -> anyhow::Result { match s { "single" => Ok(Self::Single), @@ -82,30 +66,28 @@ impl ValueMode { } } -/// Outcome of registering one discovered field. +#[derive(Debug, Clone)] +pub struct CurrentPolicyRow { + pub definition_id: Uuid, + pub insight_tenant_id: String, + pub insight_source_type: String, + pub insight_source_id: String, + pub source_field_id: String, + pub revision: i32, + pub label_override: Option, + pub sensitivity_class: Option, + pub grouping_enabled: bool, + pub comparison_enabled: bool, + pub value_mode: ValueMode, + pub retired: bool, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RegisterOutcome { Created, Refreshed, } -/// Register a discovered field: insert the definition plus its revision-1 -/// default policy (grouping allowed, comparison denied) when the key is new; -/// otherwise only advance `last_observed_at`. Idempotent per key. -/// -/// The insert and its revision-1 row are two statements without a wrapping -/// transaction: a crash between them leaves a definition whose revision 1 is -/// re-inserted by the next run (the `INSERT IGNORE` below), never a definition -/// that reconciliation refuses to touch again. -/// -/// The outcome is classified from the `INSERT IGNORE` (1 row = the field had -/// no revision 1, so it registers as created): the definition insert's -/// affected-rows count is unreliable under `CLIENT_FOUND_ROWS`, which the -/// driver enables — a duplicate-key no-op reports 1 there, not 0. -/// -/// # Errors -/// -/// Returns an error if a statement fails. pub async fn register_discovered( db: &DatabaseConnection, key: &DefinitionKey, @@ -147,6 +129,8 @@ pub async fn register_discovered( ); db.execute(insert).await?; + // WORKAROUND: the driver enables CLIENT_FOUND_ROWS, under which a duplicate-key no-op + // reports 1 affected row — so the outcome is classified from the INSERT IGNORE below. let policy = Statement::from_sql_and_values( DbBackend::MySql, INSERT_INITIAL_POLICY, @@ -168,17 +152,6 @@ pub async fn register_discovered( }) } -/// Append the next policy revision iff the caller saw the current one. -/// Returns `false` when `expected_revision` is stale (or the definition does -/// not exist / belongs to another tenant) — the API maps that to 409. The -/// guarded `INSERT ... SELECT` makes check-and-insert one statement; if two -/// writers still race past it, `uq_definition_revision` rejects the loser, -/// which is reported as the same stale-revision outcome rather than an error. -/// -/// # Errors -/// -/// Returns an error if the statement fails for any reason other than the -/// revision uniqueness key. pub async fn append_policy_revision( db: &DatabaseConnection, tenant_id: &str, @@ -235,12 +208,6 @@ fn is_duplicate_key(err: &sea_orm::DbErr) -> bool { matches!(err.sql_err(), Some(SqlErr::UniqueConstraintViolation(_))) } -/// List a tenant's definitions with their current policy, ordered by source -/// then field for stable presentation. -/// -/// # Errors -/// -/// Returns an error if the query fails. pub async fn list_definitions( db: &DatabaseConnection, tenant_id: &str, @@ -256,11 +223,6 @@ pub async fn list_definitions( rows.iter().map(row_to_definition).collect() } -/// One definition with its current policy, tenant-scoped. -/// -/// # Errors -/// -/// Returns an error if the query fails. pub async fn get_definition( db: &DatabaseConnection, tenant_id: &str, @@ -329,6 +291,37 @@ fn row_to_definition(row: &sea_orm::QueryResult) -> anyhow::Result anyhow::Result> { + let stmt = Statement::from_string( + DbBackend::MySql, + format!( + "{SELECT_WITH_POLICY} ORDER BY d.insight_tenant_id, d.insight_source_type, \ + d.insight_source_id, d.source_field_id" + ), + ); + let rows = db.query_all(stmt).await?; + rows.iter().map(row_to_current_policy).collect() +} + +fn row_to_current_policy(row: &sea_orm::QueryResult) -> anyhow::Result { + let id: Vec = row.try_get("", "definition_id")?; + let value_mode: String = row.try_get("", "value_mode")?; + Ok(CurrentPolicyRow { + definition_id: Uuid::from_slice(&id)?, + insight_tenant_id: row.try_get("", "tenant_id")?, + insight_source_type: row.try_get("", "source_type")?, + insight_source_id: row.try_get("", "source_instance")?, + source_field_id: row.try_get("", "field_id")?, + revision: row.try_get("", "revision")?, + label_override: row.try_get("", "label_override")?, + sensitivity_class: row.try_get("", "sensitivity_class")?, + grouping_enabled: row.try_get("", "grouping_enabled")?, + comparison_enabled: row.try_get("", "comparison_enabled")?, + value_mode: ValueMode::from_db(&value_mode)?, + retired: row.try_get("", "retired")?, + }) +} + #[cfg(test)] mod tests { use super::ValueMode; diff --git a/src/backend/services/identity-resolution/src/infra/identity_persons.rs b/src/backend/services/identity-resolution/src/infra/identity_persons.rs index ddb9dd57e..791b9e5e8 100644 --- a/src/backend/services/identity-resolution/src/infra/identity_persons.rs +++ b/src/backend/services/identity-resolution/src/infra/identity_persons.rs @@ -1,62 +1,17 @@ -//! 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}; +use crate::infra::snapshot_writer::{SnapshotSpec, SnapshotWriter}; -/// 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, @@ -74,10 +29,16 @@ const COLUMNS_DDL: &str = r" _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). +pub(crate) const SPEC: SnapshotSpec = SnapshotSpec { + database: "identity", + target: "identity_persons", + staging_prefix: "identity_persons_staging_", + columns_ddl: COLUMNS_DDL, + order_by: "id", + watermark_column: "_synced_at", + log_label: "persons-sync", +}; + #[derive(Debug, Row, Serialize)] struct WireRow { id: u64, @@ -105,173 +66,25 @@ struct WireRow { synced_at: chrono::DateTime, } -/// [`IdentityPersonsWriter`] over the shared `insight-clickhouse` client. pub struct ClickHouseIdentityPersonsWriter { - client: Client, + writer: SnapshotWriter, } 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"); - } - } + Self { + writer: SnapshotWriter::connect(url, user, password, SPEC, WRITE_TIMEOUT), } } - - /// 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 + let wire: Vec = rows.iter().map(|r| to_wire_row(r, synced_at)).collect(); + self.writer.replace(&wire, synced_at).await } } diff --git a/src/backend/services/identity-resolution/src/infra/mod.rs b/src/backend/services/identity-resolution/src/infra/mod.rs index b2e23e169..b1514fbb5 100644 --- a/src/backend/services/identity-resolution/src/infra/mod.rs +++ b/src/backend/services/identity-resolution/src/infra/mod.rs @@ -4,3 +4,5 @@ pub mod attribute_claims; pub mod db; pub mod identity_inputs; pub mod identity_persons; +pub mod policy_snapshot; +pub mod snapshot_writer; diff --git a/src/backend/services/identity-resolution/src/infra/policy_snapshot.rs b/src/backend/services/identity-resolution/src/infra/policy_snapshot.rs new file mode 100644 index 000000000..d5eede606 --- /dev/null +++ b/src/backend/services/identity-resolution/src/infra/policy_snapshot.rs @@ -0,0 +1,141 @@ +use std::time::Duration; + +use chrono::Utc; +use clickhouse::Row; +use serde::Serialize; +use uuid::Uuid; + +use crate::infra::db::person_attributes_repo::CurrentPolicyRow; +use crate::infra::snapshot_writer::{SnapshotSpec, SnapshotWriter}; + +const WRITE_TIMEOUT: Duration = Duration::from_mins(5); + +const COLUMNS_DDL: &str = r" + definition_id UUID, + insight_tenant_id String, + insight_source_type String, + insight_source_id String, + source_field_id String, + revision Int32, + label_override Nullable(String), + sensitivity_class Nullable(String), + grouping_enabled Bool, + comparison_enabled Bool, + value_mode LowCardinality(String), + retired Bool, + _published_at DateTime64(3, 'UTC') +"; + +pub(crate) const SPEC: SnapshotSpec = SnapshotSpec { + database: "identity", + target: "person_attribute_policy_snapshot", + staging_prefix: "person_attribute_policy_snapshot_staging_", + columns_ddl: COLUMNS_DDL, + order_by: "(insight_tenant_id, insight_source_type, insight_source_id, source_field_id)", + watermark_column: "_published_at", + log_label: "policy-publish", +}; + +#[derive(Debug, Row, Serialize)] +struct WireRow { + #[serde(with = "clickhouse::serde::uuid")] + definition_id: Uuid, + insight_tenant_id: String, + insight_source_type: String, + insight_source_id: String, + source_field_id: String, + revision: i32, + label_override: Option, + sensitivity_class: Option, + grouping_enabled: bool, + comparison_enabled: bool, + value_mode: String, + retired: bool, + #[serde( + rename = "_published_at", + with = "clickhouse::serde::chrono::datetime64::millis" + )] + published_at: chrono::DateTime, +} + +pub struct ClickHousePolicySnapshotWriter { + writer: SnapshotWriter, +} + +impl ClickHousePolicySnapshotWriter { + #[must_use] + pub fn connect(url: &str, user: &str, password: &str) -> Self { + Self { + writer: SnapshotWriter::connect(url, user, password, SPEC, WRITE_TIMEOUT), + } + } + + pub async fn replace( + &self, + rows: &[CurrentPolicyRow], + published_at: chrono::DateTime, + ) -> anyhow::Result<()> { + let wire: Vec = rows.iter().map(|r| to_wire_row(r, published_at)).collect(); + self.writer.replace(&wire, published_at).await + } + + pub async fn published_row_count(&self) -> anyhow::Result { + self.writer.published_row_count().await + } +} + +fn to_wire_row(r: &CurrentPolicyRow, published_at: chrono::DateTime) -> WireRow { + WireRow { + definition_id: r.definition_id, + insight_tenant_id: r.insight_tenant_id.clone(), + insight_source_type: r.insight_source_type.clone(), + insight_source_id: r.insight_source_id.clone(), + source_field_id: r.source_field_id.clone(), + revision: r.revision, + label_override: r.label_override.clone(), + sensitivity_class: r.sensitivity_class.clone(), + grouping_enabled: r.grouping_enabled, + comparison_enabled: r.comparison_enabled, + value_mode: r.value_mode.as_db().to_owned(), + retired: r.retired, + published_at, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::infra::db::person_attributes_repo::ValueMode; + + fn row() -> CurrentPolicyRow { + CurrentPolicyRow { + definition_id: Uuid::from_u128(7), + insight_tenant_id: "Tenant-Raw".to_owned(), + insight_source_type: "bamboohr".to_owned(), + insight_source_id: "hr-main".to_owned(), + source_field_id: "jobTitle".to_owned(), + revision: 3, + label_override: None, + sensitivity_class: Some("restricted".to_owned()), + grouping_enabled: true, + comparison_enabled: false, + value_mode: ValueMode::Single, + retired: false, + } + } + + #[test] + fn publishes_registry_keys_verbatim() -> anyhow::Result<()> { + let at = chrono::DateTime::parse_from_rfc3339("2026-08-06T10:00:00Z")?.with_timezone(&Utc); + + let wire = to_wire_row(&row(), at); + + // Byte-equal keys are the contract: the query path joins these against + // the claim relations without normalizing either side. + assert_eq!(wire.insight_tenant_id, "Tenant-Raw"); + assert_eq!(wire.insight_source_id, "hr-main"); + assert_eq!(wire.value_mode, "single"); + assert_eq!(wire.published_at, at); + Ok(()) + } +} diff --git a/src/backend/services/identity-resolution/src/infra/snapshot_writer.rs b/src/backend/services/identity-resolution/src/infra/snapshot_writer.rs new file mode 100644 index 000000000..1c09ef6b9 --- /dev/null +++ b/src/backend/services/identity-resolution/src/infra/snapshot_writer.rs @@ -0,0 +1,277 @@ +use std::time::Duration; + +use chrono::Utc; +use clickhouse::Row; +use insight_clickhouse::{Client, Config}; +use serde::Serialize; +use uuid::Uuid; + +const STAGING_GC_AGE_SECONDS: u32 = 3600; +const STAGING_SUFFIX_LEN: usize = 32; + +// INVARIANT: no `staging_prefix` may be a prefix of another spec's — the GC lists by +// `LIKE '%'`, so an overlap lets one relation's sweep drop another's live staging. +#[derive(Debug, Clone, Copy)] +pub struct SnapshotSpec { + pub database: &'static str, + pub target: &'static str, + pub staging_prefix: &'static str, + pub columns_ddl: &'static str, + pub order_by: &'static str, + pub watermark_column: &'static str, + pub log_label: &'static str, +} + +pub struct SnapshotWriter { + client: Client, + spec: SnapshotSpec, +} + +impl SnapshotWriter { + #[must_use] + pub fn new(client: Client, spec: SnapshotSpec) -> Self { + Self { client, spec } + } + + #[must_use] + pub fn connect( + url: &str, + user: &str, + password: &str, + spec: SnapshotSpec, + write_timeout: Duration, + ) -> Self { + let mut config = Config::new(url, spec.database).with_query_timeout(write_timeout); + if !user.is_empty() { + config = config.with_auth(user, password); + } + Self::new(Client::new(config), spec) + } + + pub async fn published_row_count(&self) -> anyhow::Result { + let SnapshotSpec { + database, target, .. + } = self.spec; + let count: u64 = self + .client + .query(&format!("SELECT count() FROM {database}.{target}")) + .fetch_one() + .await?; + Ok(count) + } + + async fn execute(&self, sql: &str) -> anyhow::Result<()> { + self.client.query(sql).execute().await?; + Ok(()) + } + + async fn ensure_database(&self) -> anyhow::Result<()> { + let database = self.spec.database; + // WORKAROUND: ClickHouse resolves a request's database before executing it, so + // `CREATE DATABASE` on a client pinned to that database is rejected outright. + self.client + .inner() + .clone() + .with_database("default") + .query(&format!("CREATE DATABASE IF NOT EXISTS {database}")) + .execute() + .await?; + Ok(()) + } + + async fn drop_stale_stagings(&self) { + let SnapshotSpec { + database, + staging_prefix, + log_label, + .. + } = self.spec; + 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, "{log_label}: staging GC listing failed (skipped)"); + return; + } + }; + + for name in stale { + let Some(suffix) = name.strip_prefix(staging_prefix) else { + continue; + }; + if suffix.len() != STAGING_SUFFIX_LEN || !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, "{log_label}: dropped orphaned staging"), + Err(e) => { + tracing::warn!(error = %e, table = %name, "{log_label}: staging GC drop failed"); + } + } + } + } + + async fn fill_and_swap( + &self, + staging: &str, + rows: &[R], + watermark: chrono::DateTime, + ) -> anyhow::Result<()> + where + R: Row + Serialize + Send + Sync, + for<'a> R: Row = R>, + { + let SnapshotSpec { + database, + target, + watermark_column, + .. + } = self.spec; + + let mut insert = self.client.inner().insert::(staging).await?; + for row in rows { + insert.write(row).await?; + } + insert.end().await?; + + 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)" + ); + + // INVARIANT: this guard is a backstop, not the serialization — concurrent runs are + // serialized by the caller's advisory lock, which is what makes check-then-swap safe. + let published_ms: i64 = self + .client + .query(&format!( + "SELECT toUnixTimestamp64Milli(max({watermark_column})) FROM {database}.{target}" + )) + .fetch_one() + .await?; + anyhow::ensure!( + published_ms <= watermark.timestamp_millis(), + "a newer snapshot ({watermark_column}={published_ms}ms) is already published; \ + discarding this run's older snapshot ({}ms)", + watermark.timestamp_millis() + ); + + self.execute(&format!( + "EXCHANGE TABLES {database}.`{staging}` AND {database}.{target}" + )) + .await?; + Ok(()) + } + + pub async fn replace( + &self, + rows: &[R], + watermark: chrono::DateTime, + ) -> anyhow::Result<()> + where + R: Row + Serialize + Send + Sync, + for<'a> R: Row = R>, + { + let SnapshotSpec { + database, + target, + staging_prefix, + columns_ddl, + order_by, + log_label, + .. + } = self.spec; + + let staging = format!("{staging_prefix}{}", Uuid::now_v7().simple()); + + self.ensure_database().await?; + // EXCHANGE needs both sides to exist, so the target is created before the staging. + self.execute(&format!( + "CREATE TABLE IF NOT EXISTS {database}.{target} ({columns_ddl}) \ + ENGINE = MergeTree ORDER BY {order_by}" + )) + .await?; + self.drop_stale_stagings().await; + + self.execute(&format!( + "CREATE TABLE {database}.`{staging}` ({columns_ddl}) \ + ENGINE = MergeTree ORDER BY {order_by}" + )) + .await?; + + let result = self.fill_and_swap(&staging, rows, watermark).await; + + if let Err(e) = self + .execute(&format!("DROP TABLE IF EXISTS {database}.`{staging}`")) + .await + { + tracing::warn!(error = %e, table = %staging, "{log_label}: dropping own staging failed"); + } + result + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn all_specs() -> Vec { + vec![ + crate::infra::identity_persons::SPEC, + crate::infra::policy_snapshot::SPEC, + ] + } + + #[test] + fn no_staging_prefix_is_a_prefix_of_another() { + let specs = all_specs(); + for outer in &specs { + for inner in &specs { + if outer.target == inner.target { + continue; + } + assert!( + !inner.staging_prefix.starts_with(outer.staging_prefix), + "staging prefix {:?} would be GC'd by {:?}'s sweep", + inner.staging_prefix, + outer.staging_prefix + ); + } + } + } + + #[test] + fn every_spec_targets_a_distinct_relation() { + let specs = all_specs(); + for (i, outer) in specs.iter().enumerate() { + for inner in &specs[i + 1..] { + assert_ne!( + (outer.database, outer.target), + (inner.database, inner.target), + "two specs publish the same relation" + ); + } + } + } +} diff --git a/src/backend/services/identity-resolution/src/main.rs b/src/backend/services/identity-resolution/src/main.rs index efb7904ec..67e5d2cfc 100644 --- a/src/backend/services/identity-resolution/src/main.rs +++ b/src/backend/services/identity-resolution/src/main.rs @@ -16,6 +16,7 @@ mod domain; mod gear; mod infra; mod migration; +mod publish_policy_runner; mod seed_runner; mod sync_runner; @@ -95,6 +96,10 @@ enum Commands { /// 1 failed / 2 another run holds the lock / 3 refused by a guard /// (claims relation absent, or a broken claims contract). ReconcileAttributes, + /// Publish the registry's current person-attribute policy into ClickHouse + /// and exit. Same execution model as `sync` — Helm `CronJob` / manual Job. + /// Exit codes: 0 ok / 1 failed / 2 another run holds the lock. + PublishPolicy, } /// Exit codes of the `seed` / `sync` subcommands (one shared scheme), @@ -152,6 +157,20 @@ async fn main() -> Result<()> { } } } + Commands::PublishPolicy => { + init_subcommand_logging(); + match gear::run_publish_policy(&config).await { + Ok(()) => Ok(()), + Err(publish_policy_runner::PublishRunError::LockBusy) => { + tracing::warn!("another policy-publish run holds the lock; exiting"); + std::process::exit(EXIT_SEED_LOCK_BUSY); + } + Err(publish_policy_runner::PublishRunError::Failed(e)) => { + tracing::error!(error = %format!("{e:#}"), "policy publish failed"); + std::process::exit(EXIT_SEED_FAILED); + } + } + } Commands::Openapi => print_openapi(), Commands::ReconcileAttributes => { init_subcommand_logging(); diff --git a/src/backend/services/identity-resolution/src/publish_policy_runner.rs b/src/backend/services/identity-resolution/src/publish_policy_runner.rs new file mode 100644 index 000000000..ebb5ab1b0 --- /dev/null +++ b/src/backend/services/identity-resolution/src/publish_policy_runner.rs @@ -0,0 +1,348 @@ +use std::time::Duration; + +use sea_orm::DatabaseConnection; +use serde::{Deserialize, Serialize}; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use crate::config::GearConfig; +use crate::infra::db::person_attributes_repo::{self, CurrentPolicyRow}; +use crate::infra::db::{self, ops_repo, seed_repo}; +use crate::infra::policy_snapshot::ClickHousePolicySnapshotWriter; +use crate::seed_runner::{SYSTEM_AUTHOR, resolve_tenant}; + +const PUBLISH_TIMEOUT: Duration = Duration::from_mins(5); +const RUN_TIMEOUT: Duration = Duration::from_mins(7); +const ZOMBIE_CUTOFF_HOURS: i64 = 1; +const ABORTED_BY_SHUTDOWN: &str = "policy publish aborted by server shutdown"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PublishTrigger { + Cli, + Http, +} + +impl PublishTrigger { + pub(crate) fn request_json(self) -> &'static str { + match self { + Self::Cli => r#"{"trigger":"cli"}"#, + Self::Http => r#"{"trigger":"http"}"#, + } + } +} + +#[derive(Debug)] +pub enum PublishRunError { + LockBusy, + Failed(anyhow::Error), +} + +impl From for PublishRunError { + fn from(e: anyhow::Error) -> Self { + Self::Failed(e) + } +} + +#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct PublishSummary { + pub rows: u64, + pub checksum: String, + pub skipped: bool, +} + +pub(crate) async fn resolve_journal_tenant( + db: &DatabaseConnection, + config: &GearConfig, +) -> anyhow::Result { + let distinct = seed_repo::distinct_tenants(db, 2).await?; + resolve_tenant(&config.tenant_default_id, &distinct).map_err(|msg| anyhow::anyhow!(msg)) +} + +pub(crate) async fn enqueue_run( + db: &DatabaseConnection, + tenant: Uuid, + author: Uuid, + trigger: PublishTrigger, +) -> anyhow::Result { + let operation_id = Uuid::now_v7(); + ops_repo::enqueue( + db, + operation_id, + ops_repo::PERSON_ATTRIBUTES_POLICY_PUBLISH_OP, + tenant, + author, + Some(trigger.request_json()), + ) + .await?; + Ok(operation_id) +} + +pub async fn run(config: &GearConfig) -> Result { + let db = db::connect(&config.database_url).await?; + let tenant = resolve_journal_tenant(&db, config).await?; + + let Some(lock) = db::PolicyPublishLockGuard::try_acquire(&config.database_url).await? else { + return Err(PublishRunError::LockBusy); + }; + let operation_id = enqueue_run(&db, tenant, SYSTEM_AUTHOR, PublishTrigger::Cli).await?; + + let result = run_bounded(&db, config, &lock, tenant, operation_id).await; + + lock.release().await; + result +} + +pub(crate) async fn run_detached( + db: DatabaseConnection, + config: GearConfig, + cancel: CancellationToken, + tenant: Uuid, + operation_id: Uuid, + lock: db::PolicyPublishLockGuard, +) { + let outcome = tokio::select! { + biased; + () = cancel.cancelled() => None, + result = run_bounded(&db, &config, &lock, tenant, operation_id) => Some(result), + }; + + match outcome { + Some(Ok(summary)) => { + tracing::info!(%operation_id, ?summary, "policy-publish: triggered run finished"); + } + Some(Err(e)) => { + tracing::error!(error = ?e, %operation_id, "policy-publish: triggered run failed"); + } + None => { + tracing::warn!(%operation_id, "policy-publish: run cut short by shutdown"); + if let Err(e) = ops_repo::fail(&db, operation_id, ABORTED_BY_SHUTDOWN).await { + tracing::error!(error = %e, %operation_id, "shutdown fail-update failed"); + } + return; + } + } + + lock.release().await; +} + +async fn run_bounded( + db: &DatabaseConnection, + config: &GearConfig, + _lock: &db::PolicyPublishLockGuard, + tenant: Uuid, + operation_id: Uuid, +) -> Result { + tokio::time::timeout(RUN_TIMEOUT, run_locked(db, config, tenant, operation_id)) + .await + .unwrap_or_else(|_| { + Err(PublishRunError::Failed(anyhow::anyhow!( + "policy publish timed out after {}s inside the lock-held critical section", + RUN_TIMEOUT.as_secs() + ))) + }) +} + +async fn run_locked( + db: &DatabaseConnection, + config: &GearConfig, + tenant: Uuid, + operation_id: Uuid, +) -> Result { + 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, "policy-publish: reclaimed zombie operations"), + Ok(_) => {} + Err(e) => tracing::error!(error = %e, "policy-publish: zombie sweep failed"), + } + + let previous = last_activation(db, tenant).await; + ops_repo::try_start(db, operation_id).await?; + tracing::info!(%operation_id, %tenant, "policy-publish: run started"); + + match guarded_publish(db, config, previous.as_ref()).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, "policy-publish: completed"); + Ok(summary) + } + Err(e) => { + if let Err(e2) = + ops_repo::fail(db, operation_id, "policy publish failed; see job logs").await + { + tracing::error!(error = %e2, %operation_id, "fail update failed"); + } + Err(e) + } + } +} + +async fn last_activation(db: &DatabaseConnection, tenant: Uuid) -> Option { + // INVARIANT: filtering on `completed` is what keeps this run from reading its own row — + // it is `queued` until this same run completes it. + let ops = ops_repo::list( + db, + tenant, + Some(ops_repo::PERSON_ATTRIBUTES_POLICY_PUBLISH_OP), + Some(ops_repo::OperationStatus::Completed), + 1, + ) + .await + .inspect_err(|e| tracing::warn!(error = %e, "policy-publish: activation lookup failed")) + .ok()?; + let summary_json = ops.into_iter().next()?.summary_json?; + serde_json::from_str(&summary_json).ok() +} + +async fn guarded_publish( + db: &DatabaseConnection, + config: &GearConfig, + previous: Option<&PublishSummary>, +) -> Result { + let writer = ClickHousePolicySnapshotWriter::connect( + &config.clickhouse_url, + &config.clickhouse_user, + &config.clickhouse_password, + ); + + let run = async { + let policies = person_attributes_repo::current_policies(db).await?; + let checksum = checksum(&policies); + let rows = policies.len() as u64; + + if is_already_published(previous, &checksum, rows, &writer).await { + tracing::info!(rows, %checksum, "policy-publish: snapshot already current, skipping"); + return Ok(PublishSummary { + rows, + checksum, + skipped: true, + }); + } + + writer.replace(&policies, chrono::Utc::now()).await?; + Ok(PublishSummary { + rows, + checksum, + skipped: false, + }) + }; + + tokio::time::timeout(PUBLISH_TIMEOUT, run) + .await + .unwrap_or_else(|_| { + Err(PublishRunError::Failed(anyhow::anyhow!( + "policy publish timed out after {}s", + PUBLISH_TIMEOUT.as_secs() + ))) + }) +} + +async fn is_already_published( + previous: Option<&PublishSummary>, + checksum: &str, + rows: u64, + writer: &ClickHousePolicySnapshotWriter, +) -> bool { + let Some(previous) = previous else { + return false; + }; + if previous.checksum != checksum || previous.rows != rows { + return false; + } + match writer.published_row_count().await { + Ok(published) => published == rows, + Err(e) => { + tracing::warn!(error = %e, "policy-publish: published-count check failed; publishing"); + false + } + } +} + +fn checksum(policies: &[CurrentPolicyRow]) -> String { + use std::hash::{DefaultHasher, Hash, Hasher}; + + let mut hasher = DefaultHasher::new(); + for p in policies { + p.definition_id.hash(&mut hasher); + p.insight_tenant_id.hash(&mut hasher); + p.insight_source_type.hash(&mut hasher); + p.insight_source_id.hash(&mut hasher); + p.source_field_id.hash(&mut hasher); + p.revision.hash(&mut hasher); + p.label_override.hash(&mut hasher); + p.sensitivity_class.hash(&mut hasher); + p.grouping_enabled.hash(&mut hasher); + p.comparison_enabled.hash(&mut hasher); + p.value_mode.as_db().hash(&mut hasher); + p.retired.hash(&mut hasher); + } + format!("{:016x}", hasher.finish()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::infra::db::person_attributes_repo::ValueMode; + + fn policy(field: &str, comparison_enabled: bool) -> CurrentPolicyRow { + CurrentPolicyRow { + definition_id: Uuid::from_u128(1), + insight_tenant_id: "t".to_owned(), + insight_source_type: "bamboohr".to_owned(), + insight_source_id: "hr-main".to_owned(), + source_field_id: field.to_owned(), + revision: 1, + label_override: None, + sensitivity_class: None, + grouping_enabled: true, + comparison_enabled, + value_mode: ValueMode::Single, + retired: false, + } + } + + #[test] + fn checksum_is_stable_for_an_unchanged_policy_set() { + let set = vec![policy("jobTitle", false), policy("department", false)]; + assert_eq!(checksum(&set), checksum(&set.clone())); + } + + #[test] + fn checksum_changes_when_any_published_field_changes() { + let before = vec![policy("jobTitle", false)]; + let after = vec![policy("jobTitle", true)]; + assert_ne!( + checksum(&before), + checksum(&after), + "enabling comparison must re-publish" + ); + } + + #[test] + fn checksum_of_an_empty_registry_is_stable_and_distinct() { + let empty = checksum(&[]); + assert_eq!(empty, checksum(&[])); + assert_ne!(empty, checksum(&[policy("jobTitle", false)])); + } + + #[test] + fn the_journalled_request_names_the_trigger() { + for (trigger, expected) in [ + (PublishTrigger::Cli, r#"{"trigger":"cli"}"#), + (PublishTrigger::Http, r#"{"trigger":"http"}"#), + ] { + assert_eq!( + trigger.request_json(), + expected, + "journal request payload changed for {trigger:?}" + ); + } + } + + #[tokio::test] + async fn default_config_fails_cleanly() { + let cfg = crate::config::GearConfig::default(); + let err = run(&cfg).await; + assert!(matches!(err, Err(PublishRunError::Failed(_)))); + } +} diff --git a/src/ingestion/dbt/identity/person_account_assignments_current.sql b/src/ingestion/dbt/identity/person_account_assignments_current.sql new file mode 100644 index 000000000..d6f8a19de --- /dev/null +++ b/src/ingestion/dbt/identity/person_account_assignments_current.sql @@ -0,0 +1,49 @@ +{{ config( + materialized='view', + schema='identity', + tags=['identity'] +) }} + +-- Current account→person assignment: for every stable source account, the +-- person it is bound to right now. +-- +-- A VIEW, not a published table: `identity.identity_persons` is replaced +-- wholesale by an atomic EXCHANGE, so a view over it is always consistent +-- with the latest published journal and carries no staleness window of its +-- own. Materializing would add a second publication step that could only +-- ever be older than its source. +-- +-- Latest-wins is keyed by ACCOUNT, not by person: partitioning by person +-- answers "which account does this person use", and the two disagree +-- precisely when an account is rebound from one person to another — the +-- case this relation exists to resolve. Mirrors the account-keyed window in +-- the service's own resolver. +-- +-- No FINAL: identity_persons is a plain MergeTree holding a full snapshot, +-- so `LIMIT 1 BY` here is choosing the winning observation, not collapsing +-- duplicate parts. + +SELECT + insight_tenant_id, + insight_source_type, + insight_source_id, + value_id AS source_account_id, + person_id, + created_at AS assigned_at, + _synced_at +FROM identity.identity_persons +WHERE value_type = 'id' + AND value_id IS NOT NULL + AND value_id != '' +ORDER BY + insight_tenant_id, + insight_source_type, + insight_source_id, + value_id, + created_at DESC, + id DESC +LIMIT 1 BY + insight_tenant_id, + insight_source_type, + insight_source_id, + value_id diff --git a/src/ingestion/dbt/identity/schema.yml b/src/ingestion/dbt/identity/schema.yml index 63fd04da8..1e96219db 100644 --- a/src/ingestion/dbt/identity/schema.yml +++ b/src/ingestion/dbt/identity/schema.yml @@ -229,3 +229,47 @@ models: - not_null - accepted_values: values: ['UPSERT'] + + - name: person_account_assignments_current + description: > + Current account→person assignment: one row per stable source account, + carrying the person it is bound to now. A view over + identity.identity_persons, which is replaced by an atomic swap, so the + relation is always consistent with the latest published journal. + Latest-wins is keyed by account, so rebinding an account to another + person changes this row rather than leaving both bindings visible. + JOIN RECIPE for raw-keyed relations (the person-attribute claim + relations and the attribute policy snapshot): hash their raw tenant and + source-instance identifiers with the insight_uuid_from_raw macro, and + join source_account_id raw-to-raw — value_id is not hashed. + columns: + - name: insight_tenant_id + description: "Tenant UUID (raw tenant string hashed via insight_uuid_from_raw)" + tests: + - not_null + - name: insight_source_type + description: "Source system discriminator (bamboohr, ms-entra, …)" + tests: + - not_null + - name: insight_source_id + description: "Source instance UUID (raw instance string hashed via insight_uuid_from_raw)" + tests: + - not_null + - name: source_account_id + description: "Native account identifier, exactly as the source issued it" + tests: + - not_null + - name: person_id + description: "Canonical person the account currently resolves to" + tests: + - not_null + - name: assigned_at + description: "When the winning binding observation was recorded" + tests: + - not_null + - name: _synced_at + description: > + Publication stamp of the underlying journal snapshot, identical for + every row; consumers pin it as the assignment revision. + tests: + - not_null diff --git a/src/ingestion/dbt/macros/identity_inputs_from_history.sql b/src/ingestion/dbt/macros/identity_inputs_from_history.sql index f69c60521..c7b87ede7 100644 --- a/src/ingestion/dbt/macros/identity_inputs_from_history.sql +++ b/src/ingestion/dbt/macros/identity_inputs_from_history.sql @@ -41,7 +41,8 @@ Types: `insight_tenant_id` and `insight_source_id` are emitted as UUID, derived - from the source's raw `tenant_id` / `source_id` strings via sipHash128. + from the source's raw `tenant_id` / `source_id` strings by the shared + `insight_uuid_from_raw` macro (the join recipe for raw-keyed relations). `_synced_at` is emitted as DateTime64(3). All three match what the seed- style identity inputs (seed_identity_inputs_from_cursor / _from_claude_admin) emit so the `silver/_shared/identity_inputs.sql` @@ -76,8 +77,8 @@ upserts AS ( 'UPSERT-', toString(toUnixTimestamp64Milli(toDateTime64(updated_at, 3))) ) AS String) AS unique_key, - toUUID(UUIDNumToString(sipHash128(coalesce(tenant_id, '')))) AS insight_tenant_id, - toUUID(UUIDNumToString(sipHash128(coalesce(source_id, '')))) AS insight_source_id, + {{ insight_uuid_from_raw('tenant_id') }} AS insight_tenant_id, + {{ insight_uuid_from_raw('source_id') }} AS insight_source_id, '{{ source_type }}' AS insight_source_type, entity_id AS source_account_id, '{{ f.value_type }}' AS value_type, @@ -115,8 +116,8 @@ deletes AS ( 'DELETE-', toString(toUnixTimestamp64Milli(toDateTime64(d.updated_at, 3))) ) AS String) AS unique_key, - toUUID(UUIDNumToString(sipHash128(coalesce(d.tenant_id, '')))) AS insight_tenant_id, - toUUID(UUIDNumToString(sipHash128(coalesce(d.source_id, '')))) AS insight_source_id, + {{ insight_uuid_from_raw('d.tenant_id') }} AS insight_tenant_id, + {{ insight_uuid_from_raw('d.source_id') }} AS insight_source_id, '{{ source_type }}' AS insight_source_type, d.entity_id AS source_account_id, '{{ f.value_type }}' AS value_type, @@ -144,8 +145,8 @@ id_upserts AS ( 'UPSERT-', toString(toUnixTimestamp64Milli(toDateTime64(updated_at, 3))) ) AS String) AS unique_key, - toUUID(UUIDNumToString(sipHash128(coalesce(tenant_id, '')))) AS insight_tenant_id, - toUUID(UUIDNumToString(sipHash128(coalesce(source_id, '')))) AS insight_source_id, + {{ insight_uuid_from_raw('tenant_id') }} AS insight_tenant_id, + {{ insight_uuid_from_raw('source_id') }} AS insight_source_id, '{{ source_type }}' AS insight_source_type, entity_id AS source_account_id, 'id' AS value_type, @@ -169,8 +170,8 @@ id_deletes AS ( 'DELETE-', toString(toUnixTimestamp64Milli(toDateTime64(d.updated_at, 3))) ) AS String) AS unique_key, - toUUID(UUIDNumToString(sipHash128(coalesce(d.tenant_id, '')))) AS insight_tenant_id, - toUUID(UUIDNumToString(sipHash128(coalesce(d.source_id, '')))) AS insight_source_id, + {{ insight_uuid_from_raw('d.tenant_id') }} AS insight_tenant_id, + {{ insight_uuid_from_raw('d.source_id') }} AS insight_source_id, '{{ source_type }}' AS insight_source_type, d.entity_id AS source_account_id, 'id' AS value_type, diff --git a/src/ingestion/dbt/macros/insight_uuid_from_raw.sql b/src/ingestion/dbt/macros/insight_uuid_from_raw.sql new file mode 100644 index 000000000..17e097574 --- /dev/null +++ b/src/ingestion/dbt/macros/insight_uuid_from_raw.sql @@ -0,0 +1,23 @@ +{% macro insight_uuid_from_raw(column) %} +{#- + The project-wide mapping from a raw connector-config identifier (tenant id, + source instance id) to the UUID the identity relations carry. + + Identity relations type these columns as UUID so the silver UNION ALL + type-checks (ClickHouse rejects UNION across UUID and String, NO_COMMON_TYPE), + while connector configuration supplies free-form strings — so the raw value + is hashed into a UUID. The same raw string always maps to the same UUID, which + is what keeps cross-source joins consistent. + + This is the JOIN RECIPE for anything keyed by raw strings — the person + attribute claim relations and the attribute policy snapshot both carry raw + identifiers, so joining them to an identity relation means hashing the raw + side with this macro. Writing the expression by hand is how a join silently + returns nothing: a different hash width or a missing coalesce still compiles + and still produces a UUID, just not the same one. + + TEMPORARY, pending a real tenants registry that issues actual UUIDs; when + that lands this macro is the single place the mapping is retired from. +-#} +toUUID(UUIDNumToString(sipHash128(coalesce({{ column }}, '')))) +{% endmacro %} diff --git a/tests/generate_schemas.py b/tests/generate_schemas.py index 999d73155..08c3f07f4 100644 --- a/tests/generate_schemas.py +++ b/tests/generate_schemas.py @@ -132,6 +132,29 @@ ''' +IDENTITY_RESOLUTION_HEADER = '''"""Identity-resolution response shapes — GENERATED, do not edit. + +Regenerate with: + + uv run --project tests --frozen python tests/generate_schemas.py + +Source: `docs/components/backend/identity-resolution/openapi.json`, emitted by +`cargo run -p identity-resolution -- openapi` from the same route table the +service serves and drift-gated in CI. These models therefore describe the +structs that serialize the wire, which the hand-written ones they replaced could +not: the committed document used to be the retired .NET contract. + +`extra="forbid"` throughout: an undeclared field is drift. + +The four journal responses (persons-seed, persons-sync, attribute-reconcile, +policy-publish) are separate types with identical fields, because they are +separate operations whose summaries are free to diverge. `schemas/__init__.py` +re-exports one of them as `Operation` for suites that assert the shared shape. +""" + +''' + + @dataclass(frozen=True) class Generated: """A service whose document describes bodies: models are generated and committed.""" @@ -229,19 +252,11 @@ def declares_only_200(spec_path: Path) -> str | None: spec=_SPECS / "gateway" / "openapi.json", reason="publishes no OpenAPI document (NGINX + Lua; `GET /healthz` is its only own route)", ), - # The committed document is still the retired .NET one — it declares routes - # the service does not serve, omits ones it does, and lists only `200` - # everywhere (enumerated in `stand/api/schemas/__init__.py`). `identity.py` is - # hand-written from the Rust DTOs until identity grows an `openapi` - # subcommand of its own, as analytics and authenticator have. - # - # NOT `Bodyless`: the document does describe one body (`POST - # /v1/visible-persons`), and body count says nothing about provenance. - Untrusted( + Generated( name="identity-resolution", spec=_SPECS / "identity-resolution" / "openapi.json", - reason="the committed document is the retired .NET contract", - still_the_wrong_document=declares_only_200, + output=_SCHEMAS / "identity.py", + header=IDENTITY_RESOLUTION_HEADER, ), ) diff --git a/tests/stand/api/identity/test_admin.py b/tests/stand/api/identity/test_admin.py index 177ef49d5..ae154c28a 100644 --- a/tests/stand/api/identity/test_admin.py +++ b/tests/stand/api/identity/test_admin.py @@ -1,6 +1,6 @@ """The admin-gated half of identity-resolution. -Thirteen operations sit behind `require_admin`, which resolves the caller from +Sixteen operations sit behind `require_admin`, which resolves the caller from the gateway JWT and requires an active `admin` row in `identity.person_roles`. It never reads the `insight-admin` REALM role — so the CEO, who holds that role, is refused exactly like everybody else. The seed grants the row to one account: diff --git a/tests/stand/api/operations.py b/tests/stand/api/operations.py index 17c97d071..ca14abe9b 100644 --- a/tests/stand/api/operations.py +++ b/tests/stand/api/operations.py @@ -99,7 +99,7 @@ def _i(method: str, suffix: str) -> Operation: _a("POST", "/v1/metric-drilldown/export"), ) -#: identity-resolution — 17 operations. `/health` and `/healthz` are the host +#: identity-resolution — 20 operations. `/health` and `/healthz` are the host #: router's, not the product API, and are deliberately absent: the real probes #: address the pod directly rather than passing the gateway. IDENTITY_OPERATIONS: Final[tuple[Operation, ...]] = ( @@ -110,6 +110,9 @@ def _i(method: str, suffix: str) -> Operation: _i("GET", f"/v1/persons-seed/{SOME_ID}"), _i("GET", "/v1/persons-sync"), _i("GET", f"/v1/persons-sync/{SOME_ID}"), + _i("GET", "/v1/person-attributes-policy-publish"), + _i("GET", f"/v1/person-attributes-policy-publish/{SOME_ID}"), + _i("POST", "/v1/person-attributes-policy-publish"), _i("GET", "/v1/roles"), _i("POST", "/v1/roles"), _i("DELETE", f"/v1/roles/{SOME_ID}"), @@ -126,7 +129,7 @@ def _i(method: str, suffix: str) -> Operation: ALL_OPERATIONS: Final[tuple[Operation, ...]] = ANALYTICS_OPERATIONS + IDENTITY_OPERATIONS -#: The 13 identity operations behind `require_admin`, which resolves the caller +#: The 16 identity operations behind `require_admin`, which resolves the caller #: from the gateway JWT and requires an active `admin` row in `person_roles` — #: it never reads the `insight-admin` REALM role. The seed grants nobody that #: row, so every persona is refused; see out/endpoint-coverage-preconditions.md. @@ -135,6 +138,13 @@ def _i(method: str, suffix: str) -> Operation: for op in IDENTITY_OPERATIONS if any( seg in op.path - for seg in ("/persons-seed", "/persons-sync", "/roles", "/person-roles", "/visibility") + for seg in ( + "/persons-seed", + "/persons-sync", + "/person-attributes", + "/roles", + "/person-roles", + "/visibility", + ) ) ) diff --git a/tests/stand/api/schemas/__init__.py b/tests/stand/api/schemas/__init__.py index c58c49ee5..4b20ff158 100644 --- a/tests/stand/api/schemas/__init__.py +++ b/tests/stand/api/schemas/__init__.py @@ -4,15 +4,10 @@ * `common.py` — the error envelope and the listing wrapper, hand-written from the bodies the stand returns. -* `identity.py` — hand-written from the Rust DTOs in - `src/backend/services/identity-resolution/src/api/`. **Not** generated, - because the committed contract for that service is still the .NET document: - it declares `/v1/persons/{email}` (which identity answers 404 for), declares - `POST /v1/persons-seed` (405), spells the - subchart parameter `{personId}` where the service serves `{person_id}`, omits - both persons-sync operations, and lists only `200` for all 18 operations. - Generating from it would record every one of those errors as fact. -* `analytics.py`, `authenticator.py` — GENERATED from documents the services +* `identity_internal.py` — hand-written, and only for the `/internal/*` routes + the published document deliberately omits (service-to-service only, mounted + outside the OpenAPI registry). +* `identity.py`, `analytics.py`, `authenticator.py` — GENERATED from documents the services emit themselves (`cargo run -p -- openapi`) and CI drift-gates in `.github/workflows/openapi-specs.yml`, so the models describe the very structs that serialize the wire. `authenticator.py` is currently just the error @@ -26,9 +21,14 @@ That asymmetry is a real difference in what the models mean. The generated ones are a **contract test** — a mismatch says the service and its published contract -disagree. The hand-written ones are a **description of observed behaviour**, and -they should be deleted in favour of generated ones once the identity contract is -regenerated from the service. +disagree. The hand-written ones are a **description of observed behaviour**, kept +only where no contract describes the route. + +The identity models are re-exported below under the names the suites already +use: the generated ones carry the DTO names the service serializes +(`RoleResponse`, `ProfileResponse`, …), and the four journal responses are +field-identical, so one of them is re-exported as `Operation` for assertions +about the shared journal shape. The strictness follows from that. Generated models set `extra="forbid"`: they are regenerated in the same change that adds a field, so strictness costs nothing and @@ -58,20 +58,46 @@ ProblemDocument, ) from .identity import ( - IdentityValue, - Operation, - OperationList, - PersonRole, - PersonRoleList, - Profile, - Role, - RoleList, - Subchart, - SubchartForest, + PersonRoleListResponse as PersonRoleList, +) +from .identity import ( + PersonRoleResponse as PersonRole, +) +from .identity import ( + PersonsSeedListResponse as OperationList, +) +from .identity import ( + PersonsSeedOperationResponse as Operation, +) +from .identity import ( + ProfileResponse as Profile, +) +from .identity import ( + RoleListResponse as RoleList, +) +from .identity import ( + RoleResponse as Role, +) +from .identity import ( + SubchartForestResponse as SubchartForest, +) +from .identity import ( SubchartNode, - Visibility, - VisibilityList, - VisiblePersons, +) +from .identity import ( + SubchartResponse as Subchart, +) +from .identity import ( + VisibilityListResponse as VisibilityList, +) +from .identity import ( + VisibilityResponse as Visibility, +) +from .identity import ( + VisiblePersonsResponse as VisiblePersons, +) +from .identity_internal import ( + IdentityValue, ) __all__: Sequence[str] = ( diff --git a/tests/stand/api/schemas/identity.py b/tests/stand/api/schemas/identity.py index 6b3e3d559..35a01b334 100644 --- a/tests/stand/api/schemas/identity.py +++ b/tests/stand/api/schemas/identity.py @@ -1,251 +1,477 @@ -"""identity-resolution response shapes, hand-written from the Rust DTOs. - -Sources, field for field: - - domain/subchart.rs SubchartNode · SubchartResponse · SubchartForestResponse - domain/profile.rs ProfileResponse - api/roles.rs RoleResponse - api/person_roles.rs PersonRoleResponse - api/visibility.rs VisibilityResponse - api/seed.rs PersonsSeedOperationResponse - api/sync.rs PersonsSyncOperationResponse - -**Not generated**, and not from the committed contract — see `schemas/__init__.py` -for why `docs/components/backend/identity-resolution/openapi.json` cannot be -trusted. These models therefore describe OBSERVED behaviour, and `extra` is left -at its default so an added field does not fail the suite. When that contract is -regenerated from the service, this module should be deleted in favour of -generated models with `extra="forbid"`. - -Timestamps stay `str`. The service serialises them itself (`api/datetime.rs` -normalises to naive-UTC on the way in), and coercing to `datetime` here would -make the tests assert a parse this suite does not perform — the wire format is -the contract, not Python's reading of it. +"""Identity-resolution response shapes — GENERATED, do not edit. + +Regenerate with: + + uv run --project tests --frozen python tests/generate_schemas.py + +Source: `docs/components/backend/identity-resolution/openapi.json`, emitted by +`cargo run -p identity-resolution -- openapi` from the same route table the +service serves and drift-gated in CI. These models therefore describe the +structs that serialize the wire, which the hand-written ones they replaced could +not: the committed document used to be the retired .NET contract. + +`extra="forbid"` throughout: an undeclared field is drift. + +The four journal responses (persons-seed, persons-sync, attribute-reconcile, +policy-publish) are separate types with identical fields, because they are +separate operations whose summaries are free to diverge. `schemas/__init__.py` +re-exports one of them as `Operation` for suites that assert the shared shape. """ from __future__ import annotations - -from collections.abc import Sequence from uuid import UUID +from typing import Any +from pydantic import AwareDatetime, BaseModel, ConfigDict, Field +from enum import StrEnum -from insight_stand import JsonValue -from pydantic import BaseModel, Field -from .common import ListResponse +class AttributeReconcileOperationResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + completed_at: str | None = None + error_message: str | None = None + insight_tenant_id: UUID + operation_id: UUID + operation_type: str + request: dict[str, Any] | None = None + started_at: str + status: str + summary: dict[str, Any] | None = None -# --------------------------------------------------------------------------- -# Org subchart -# --------------------------------------------------------------------------- +class CreatePersonRoleRequest(BaseModel): + """ + Body of `POST /v1/person-roles` — grant a role to a person. + """ + model_config = ConfigDict( + extra='forbid', + ) + person_id: UUID + reason: str | None = None + role_id: UUID + valid_from: AwareDatetime | None = Field(None, description='Optional assignment start; defaults to now when omitted. Accepts RFC-3339\n(`Z`/offset), zone-less, or date-only, normalised to naive-UTC.') -class SubchartNode(BaseModel): - """One person in an org tree. Self-referential through `subordinates`. - Everything but `person_id` is nullable: a node exists because an `org_chart` - edge points at it, and the person's attributes are a separate observation - that may be absent. +class CreateRoleRequest(BaseModel): + """ + Body of `POST /v1/roles`. """ + model_config = ConfigDict( + extra='forbid', + ) + name: str - person_id: UUID - email: str | None = None - display_name: str | None = None - job_title: str | None = None - status: str | None = None - subordinates: list[SubchartNode] = Field(default_factory=list) - def walk(self) -> list[SubchartNode]: - """This node and every descendant, at any depth.""" - found = [self] - for child in self.subordinates: - found += child.walk() - return found +class CreateVisibilityRequest(BaseModel): + """ + Body of `POST /v1/visibility` — grant a viewer visibility over a target + (or the whole tree when `viewed_person_id` is omitted). + """ + model_config = ConfigDict( + extra='forbid', + ) + reason: str | None = None + valid_from: AwareDatetime | None = Field(None, description='Optional grant start; defaults to now when omitted. Accepts RFC-3339\n(`Z`/offset), zone-less, or date-only, normalised to naive-UTC.') + viewed_person_id: UUID | None = None + viewer_person_id: UUID - def emails(self) -> set[str]: - """Every email in this subtree — the shape scope assertions compare.""" - return {node.email for node in self.walk() if node.email} +class PersonResponse(BaseModel): + """ + A person node in the org tree (subordinate of a profile), matching the .NET + `PersonResponse`. Unlike `ProfileResponse`, the attribute fields are plain + strings (empty when absent, not omitted) and the `supervisor_*`/`parent_*` + fields serialize as `null` rather than being dropped. + """ + model_config = ConfigDict( + extra='forbid', + ) + department: str + display_name: str + division: str + email: str + first_name: str + job_title: str + last_name: str + parent_email: str | None = None + parent_id: str | None = None + parent_person_id: UUID | None = None + person_id: UUID + status: str + subordinates: list[PersonResponse] + supervisor_email: str | None = None + supervisor_name: str | None = None -class SubchartForest(BaseModel): - """`GET /v1/subchart` — the forest the CALLER can see. - Empty when the caller has no visible membership, which is the normal state - for an account outside the org chart rather than an error. +class PersonRoleResponse(BaseModel): """ + One role assignment. + """ + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + created_at: str + insight_tenant_id: UUID + person_id: UUID + person_role_id: UUID + reason: str | None = None + role_id: UUID + valid_from: str + valid_to: str | None = None - roots: list[SubchartNode] = Field(default_factory=list) - - def emails(self) -> set[str]: - return {email for root in self.roots for email in root.emails()} +class PersonsSeedOperationResponse(BaseModel): + """ + One operation's status. Wire shape mirrors the .NET + `PersonsSeedOperationResponse`: `request` and `summary` are surfaced as + parsed JSON (not double-encoded strings), the tenant/author ids are + included, timestamps are ISO-8601, and null fields are emitted (the .NET + serializer does not drop nulls). + """ + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + completed_at: str | None = None + error_message: str | None = None + insight_tenant_id: UUID + operation_id: UUID + operation_type: str + request: dict[str, Any] | None = None + started_at: str + status: str + summary: dict[str, Any] | None = None -class Subchart(BaseModel): - """`GET /v1/subchart/{person_id}` — one named person's subtree. - A `root` object rather than a bare node so the response can gain sibling - fields without breaking clients. +class PersonsSyncOperationResponse(BaseModel): """ + One operation's status. Wire shape matches the seed journal's: + `request` and `summary` surfaced as parsed JSON, ISO-8601 timestamps, + null fields emitted. + """ + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + completed_at: str | None = None + error_message: str | None = None + insight_tenant_id: UUID + operation_id: UUID + operation_type: str + request: dict[str, Any] | None = None + started_at: str + status: str + summary: dict[str, Any] | None = Field(None, description='On completion: the [`SyncSummary`] — rows copied, `max_id` /\n`max_created_at` watermarks, `synced_at`.\n\n[`SyncSummary`]: crate::domain::sync_service::SyncSummary') - root: SubchartNode - - def emails(self) -> set[str]: - return self.root.emails() +class PolicyPublishOperationResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + completed_at: str | None = None + error_message: str | None = None + insight_tenant_id: UUID + operation_id: UUID + operation_type: str + request: dict[str, Any] | None = None + started_at: str + status: str + summary: dict[str, Any] | None = None -# --------------------------------------------------------------------------- -# Profiles -# --------------------------------------------------------------------------- +class PolicyResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + actor_person_id: UUID + comparison_enabled: bool + grouping_enabled: bool + label_override: str | None = None + reason: str + retired: bool + revision: int + sensitivity_class: str | None = None + value_mode: str -class Profile(BaseModel): - """`POST /v1/profiles` — a person resolved by email or source-native id. - Only the fields the tests assert are declared. The DTO carries many more - attribute fields, all omitted from JSON when null, and modelling them would - be describing the seed rather than the contract. +class Problem(BaseModel): + """ + RFC 9457 problem+json. `context` varies by error category. + """ + model_config = ConfigDict( + extra='forbid', + ) + context: dict[str, Any] + detail: str + instance: str | None = None + status: int + title: str + trace_id: str | None = None + type: str + + +class ProfileIdEntry(BaseModel): + """ + One source-native account id bound to the person — the latest + `value_type='id'` observation per source instance. Ported from the .NET + `ProfileIdEntry`. """ + model_config = ConfigDict( + extra='forbid', + ) + insight_source_id: UUID + insight_source_type: str + value: str - person_id: UUID - insight_tenant_id: UUID - email: str | None = None - display_name: str | None = None +class ProfileResponse(BaseModel): + """ + Response body of `POST /v1/profiles` — the resolved person's profile: + current attributes, the org tree (`supervisor_*` / `parent_*` / + `subordinates[]`), and every current source-native id (`ids[]`). Null + attribute fields are omitted from JSON; `subordinates`/`ids` are always + present (empty when none), matching the .NET contract. + """ + model_config = ConfigDict( + extra='forbid', + ) + department: str | None = None + display_name: str | None = None + division: str | None = None + email: str | None = None + employee_id: str | None = None + first_name: str | None = None + ids: list[ProfileIdEntry] = Field(..., description='Every current source-native id for the person (one per source instance).\nAlways serialized — an empty array when the person has no ids — matching\nthe .NET contract (unlike the attributes above, which are omitted).') + insight_tenant_id: UUID + job_title: str | None = None + last_name: str | None = None + parent_email: str | None = None + parent_id: str | None = None + parent_person_id: UUID | None = None + person_id: UUID + status: str | None = None + subordinates: list[PersonResponse] = Field(..., description='Recursive subordinates subtree (direct reports and their reports), on the\nconfigured `org_chart` source. Always serialized (empty when none).') + supervisor_email: str | None = None + supervisor_name: str | None = None + username: str | None = None -class IdentityValue(BaseModel): - """`GET /internal/persons/by-email/{email}` — the login-bootstrap lookup. - NOT a `Profile`, though both are "a person looked up by email". This route - answers the identity VALUE that matched — the alias row, pointing at what it - resolved to — because at login the caller has an email and needs to learn - which person it belongs to, not to read that person's attributes. Hence - `insight_source_id` rather than `person_id`, and no tenant at all: the - tenant is exactly what is still unknown at that point. +class ResolveProfileRequest(BaseModel): + """ + Body of `POST /v1/profiles`. `value_type = "email"` matches across all + sources for the tenant; `value_type = "id"` matches a source-native account + id within one source instance (needs `insight_source_type` + `insight_source_id`); + `value_type = "person_id"` takes the canonical person UUID itself — the key + the metrics runtime and its routes use since the identity cutover. + """ + model_config = ConfigDict( + extra='forbid', + ) + insight_source_id: UUID | None = Field(None, description='Required when `value_type = "id"`.') + insight_source_type: str | None = Field(None, description='Required when `value_type = "id"` — the source instance to scope to.') + value: str + value_type: str +class RoleResponse(BaseModel): """ - - value_type: str - value: str - insight_source_type: str - insight_source_id: UUID + One role in the catalogue. + """ + model_config = ConfigDict( + extra='forbid', + ) + name: str + role_id: UUID -# --------------------------------------------------------------------------- -# Admin: roles, assignments, visibility -# --------------------------------------------------------------------------- +class SubchartNode(BaseModel): + """ + One node in the org subchart tree. + """ + model_config = ConfigDict( + extra='forbid', + ) + display_name: str | None = None + email: str | None = None + job_title: str | None = None + person_id: UUID + status: str | None = None + subordinates: list[SubchartNode] -class Role(BaseModel): - """An entry in the global role catalogue. Deleted, not revoked.""" +class SubchartResponse(BaseModel): + """ + `{ "root": { … } }` — single-root wrapper (locked by the #348 acceptance + criteria so the response can gain sibling fields without breaking clients). + """ + model_config = ConfigDict( + extra='forbid', + ) + root: SubchartNode - role_id: UUID - name: str +class ValueModeDto(StrEnum): + single = 'single' + multi = 'multi' -class PersonRole(BaseModel): - """A role assignment. Temporal: `DELETE` sets `valid_to` rather than removing. - `valid_to is None` is therefore the only meaning of "in force", and it is - what the leak sweep in `scratch.py` checks. +class VisibilityResponse(BaseModel): """ - - person_role_id: UUID + One visibility grant. + """ + model_config = ConfigDict( + extra='forbid', + ) + author_person_id: UUID + created_at: str insight_tenant_id: UUID - person_id: UUID - role_id: UUID + reason: str | None = None valid_from: str valid_to: str | None = None - author_person_id: UUID - reason: str | None = None - created_at: str + viewed_person_id: UUID | None = None + viewer_person_id: UUID + visibility_id: UUID - @property - def in_force(self) -> bool: - return self.valid_to is None +class VisiblePersonsRequest(BaseModel): + """ + Canonical person UUIDs to check (the metric runtime's key since the + identity cutover — the earlier email-based draft of this endpoint never + shipped). + """ + model_config = ConfigDict( + extra='forbid', + ) + person_ids: list[UUID] -class Visibility(BaseModel): - """A visibility grant. Temporal, exactly like `PersonRole`. - `viewed_person_id is None` is a grant over everything the viewer's source - membership covers rather than one named person. - """ +class VisiblePersonsResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + visible: list[UUID] - visibility_id: UUID - insight_tenant_id: UUID - viewer_person_id: UUID - viewed_person_id: UUID | None = None - valid_from: str - valid_to: str | None = None - author_person_id: UUID - reason: str | None = None - created_at: str - @property - def in_force(self) -> bool: - return self.valid_to is None +class AttributeReconcileListResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + items: list[AttributeReconcileOperationResponse] + next_cursor: str | None = None + +class PersonAttributeResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + first_observed_at: str + id: UUID + last_observed_at: str + policy: PolicyResponse + source_field_id: str + source_instance: str + source_type: str -class VisiblePersons(BaseModel): - """`POST /v1/visible-persons` — the subset of the asked-about person ids. - A list of what survived, not a per-id verdict: a person the caller may not - see is absent rather than present-and-false, which is the same - non-disclosure choice `/v1/subchart/{id}` makes by answering 404. +class PersonRoleListResponse(BaseModel): + """ + List wrapper. + """ + model_config = ConfigDict( + extra='forbid', + ) + items: list[PersonRoleResponse] + next_cursor: str | None = Field(None, description='Wire parity with the .NET `ListResponse`: the cursor is declared\nbut pagination is not implemented — always `null` (both\nimplementations return every row; consumers already tolerate it).') + - Person UUIDs since the identity cutover (#2098), like every other - person-keyed route. +class PersonsSeedListResponse(BaseModel): """ + List response wrapper (typed for OpenAPI). + """ + model_config = ConfigDict( + extra='forbid', + ) + items: list[PersonsSeedOperationResponse] + next_cursor: str | None = Field(None, description='Wire parity with the .NET `ListResponse`: the cursor is declared\nbut pagination is not implemented — always `null` (both\nimplementations return every row; consumers already tolerate it).') - visible: list[UUID] +class PersonsSyncListResponse(BaseModel): + """ + List response wrapper (typed for OpenAPI). `next_cursor` is declared but + always `null` — same non-paginating contract as the seed journal. + """ + model_config = ConfigDict( + extra='forbid', + ) + items: list[PersonsSyncOperationResponse] + next_cursor: str | None = None + + +class PolicyPublishListResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + items: list[PolicyPublishOperationResponse] + next_cursor: str | None = None + + +class PolicyUpdateRequest(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + comparison_enabled: bool + expected_revision: int + grouping_enabled: bool + label_override: str | None = None + reason: str + retired: bool + sensitivity_class: str | None = None + value_mode: ValueModeDto + + +class RoleListResponse(BaseModel): + """ + List wrapper. + """ + model_config = ConfigDict( + extra='forbid', + ) + items: list[RoleResponse] + next_cursor: str | None = Field(None, description='Wire parity with the .NET `ListResponse`: the cursor is declared\nbut pagination is not implemented — always `null` (both\nimplementations return every row; consumers already tolerate it).') -# --------------------------------------------------------------------------- -# Seed / sync journals -# --------------------------------------------------------------------------- +class SubchartForestResponse(BaseModel): + """ + `{ "roots": [ … ] }` — forest wrapper (#344). Empty when the caller has no + visible-in-source membership. + """ + model_config = ConfigDict( + extra='forbid', + ) + roots: list[SubchartNode] -class Operation(BaseModel): - """One persons-seed or persons-sync run. - The two DTOs are field-identical, so one model serves both journals. `request` - and `summary` are free-form objects the service echoes back, kept as - `JsonValue` rather than modelled — their shape belongs to whichever seed - version wrote them. +class VisibilityListResponse(BaseModel): """ + List wrapper. + """ + model_config = ConfigDict( + extra='forbid', + ) + items: list[VisibilityResponse] + next_cursor: str | None = Field(None, description='Wire parity with the .NET `ListResponse`: the cursor is declared\nbut pagination is not implemented — always `null` (both\nimplementations return every row; consumers already tolerate it).') - operation_id: UUID - operation_type: str - status: str - insight_tenant_id: UUID - author_person_id: UUID - request: JsonValue = None - summary: JsonValue = None - error_message: str | None = None - started_at: str - completed_at: str | None = None + +class PersonAttributeListResponse(BaseModel): + model_config = ConfigDict( + extra='forbid', + ) + items: list[PersonAttributeResponse] -# --------------------------------------------------------------------------- -# Listings -# --------------------------------------------------------------------------- - -RoleList = ListResponse[Role] -PersonRoleList = ListResponse[PersonRole] -VisibilityList = ListResponse[Visibility] -OperationList = ListResponse[Operation] - - -__all__: Sequence[str] = ( - "Operation", - "OperationList", - "PersonRole", - "PersonRoleList", - "Profile", - "Role", - "RoleList", - "Subchart", - "SubchartForest", - "SubchartNode", - "Visibility", - "VisibilityList", -) +PersonResponse.model_rebuild() +SubchartNode.model_rebuild() diff --git a/tests/stand/api/schemas/identity_internal.py b/tests/stand/api/schemas/identity_internal.py new file mode 100644 index 000000000..54ed34d8b --- /dev/null +++ b/tests/stand/api/schemas/identity_internal.py @@ -0,0 +1,35 @@ +"""Identity shapes the published contract deliberately does not describe. + +`identity.py` is generated from `docs/components/backend/identity-resolution/ +openapi.json`. The `/internal/*` routes are mounted outside the OpenAPI +registry on purpose — they are service-to-service only and are not part of the +public surface — so nothing in that document describes them and no model for +them can be generated. + +Hand-written models therefore describe **observed behaviour**, not a contract: +`extra` stays at its default, because a benign upstream addition should not +fail a suite that never had a contract to check against in the first place. +""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel + + +class IdentityValue(BaseModel): + """`GET /internal/persons/by-email/{email}` — the login-bootstrap lookup. + + NOT a `ProfileResponse`, though both are "a person looked up by email". + This route answers the identity VALUE that matched — the alias row, + pointing at what it resolved to — because at login the caller has an email + and needs to learn which person it belongs to, not to read that person's + attributes. Hence `insight_source_id` rather than `person_id`, and no + tenant at all: the tenant is exactly what is still unknown at that point. + """ + + value_type: str + value: str + insight_source_type: str + insight_source_id: UUID