# NexusCRM Outbound Job Framework

Status: Target design; the first Month End ARR MVP implements a bounded subset.

The implementation in this change keeps the target architecture's core safety
properties—typed mapping, full source preflight, deterministic external-ID
upsert, target-contract validation, a single-writer lease, and per-record
attempt evidence—but does not yet materialize a run-scoped candidate table or
maintain separate current-state and readiness tables. The operational contract
for what is implemented today is the
[Month End ARR outbound runbook](month-end-arr-outbound-runbook.md); the
additional lifecycle and retry-state machinery below remains the intended
framework evolution.

## 1. Decision

Implement the NexusCRM writeback in this repository as a dedicated Cloud Run
Job. It shares the repository, container image, CI pipeline, and a small amount
of task infrastructure with the existing extract/load jobs, but it has its own:

- command and configuration;
- Cloud Run Job and schedule;
- runtime service account and secrets;
- run lease, state ledger, and record-attempt history;
- failure policy, metrics, and alerts.

The first implementation targets one known custom NexusCRM object. The
framework must not become a generic reverse-ETL engine or a dynamic field
mapping DSL. Source-to-object mapping remains explicit, typed Python code.

Although this repository calls the direction "outbound", NexusCRM sees the
traffic as inbound record writes. Use `nexuscrm-writeback` in resource and code
names where that avoids confusion with NexusCRM's own outbound webhook/callout
subsystem.

## 2. Goals and non-goals

### Goals

1. Read a deterministic set of candidate rows from BigQuery.
2. Map them to one fixed custom object and a declared set of owned fields.
3. Upsert records through the NexusCRM v63 REST API by a unique external ID.
4. Make retries and whole-job replay safe.
5. Preserve per-record success and failure evidence.
6. Detect schema, permission, authentication, and source-data drift before or
   during a run.
7. Reuse the existing build/deploy path without coupling the writeback failure
   domain to the current extract/load jobs.

### Non-goals for v1

- real-time or event-driven synchronization;
- a general-purpose workflow/orchestration platform;
- multiple target CRMs or arbitrary target objects;
- hard delete propagation;
- two-way conflict resolution;
- direct writes to NexusCRM's TiDB tables;
- Bulk API v2 as the initial write path.

## 3. Architecture

```text
BigQuery source view/query
        |
        v
run-scoped BigQuery candidate table
        |
        v
source contract + full mapping validation
        |
        v
object-specific typed mapper
        |
        v
canonical payload + payload hash
        |
        +------ unchanged ------> state: skipped
        |
        v
NexusCRM Composite Collections Upsert (<= 200 records)
        |
        +------ success --------> current state + attempt history
        |
        +------ retryable ------> backoff/retry + current state
        |
        +------ permanent ------> rejected state + attempt history
```

The job performs a full logical source selection in v1 and uses the canonical
payload hash to skip unchanged records. This favors correctness for late
corrections and failed-row replay over early watermark optimization. If volume
later makes a full selection expensive, add an incremental candidate query that
unions:

1. rows newer than the last successful cutoff; and
2. external IDs in retryable state.

The state ledger remains the correctness boundary after that optimization.

Materialize the source SQL once into a run-scoped candidate table in the
control dataset. Give it a 24-hour expiration and delete it in normal cleanup.
The candidate table provides a stable source snapshot for validation and
writing, including runs where the upstream view changes while the job is in
progress. It may contain customer data, so it is temporary working state and
must not be retained as an audit payload.

The runner makes two deterministic passes over the candidate table:

1. validate null/duplicate keys, required types, and every mapped payload;
2. map again, compare payload hashes with current state, and write batches.

This ensures a mapper/schema error is discovered before the first CRM write
without retaining mapped customer payloads in the control ledger.

## 4. Repository layout

```text
configs/
  jobs/
    <job-name>.yaml
docs/
  nexuscrm-outbound-framework.md
sql/
  jobs/
    <job-name>.sql
    <job-name>-readiness.sql
src/pingcap_datapipelines/
  jobs/
    __init__.py
    base.py
    registry.py
    nexuscrm_custom_object/
      __init__.py
      config.py
      mapping.py
      runner.py
  integrations/
    __init__.py
    nexuscrm.py
  outbound_state.py
tests/
  test_job_config.py
  test_job_mapping.py
  test_job_runner.py
  test_nexuscrm_client.py
  test_outbound_state.py
```

Responsibilities:

- `jobs/base.py`: stable job lifecycle types only.
- `jobs/registry.py`: explicit name-to-job factory mapping; no dynamic imports.
- `jobs/nexuscrm_custom_object/config.py`: validates this job's configuration.
- `jobs/nexuscrm_custom_object/mapping.py`: owns source schema, field
  ownership, null behavior, normalization, and target payload construction.
- `jobs/nexuscrm_custom_object/runner.py`: coordinates query, mapping,
  deduplication, batching, state, and result handling.
- `integrations/nexuscrm.py`: transport-only OAuth and REST client. It must not
  know the target custom object's fields or business rules.
- `outbound_state.py`: BigQuery DDL and batched state transitions shared by
  outbound jobs.

## 5. Minimal job contract

Do not introduce a full workflow engine. The initial reusable contract is:

```python
@dataclass(frozen=True)
class JobContext:
    run_id: str
    started_at: datetime
    dry_run: bool


@dataclass(frozen=True)
class JobResult:
    status: Literal["succeeded", "completed_with_errors", "failed"]
    selected: int
    unchanged: int
    succeeded: int
    retryable_failed: int
    permanently_failed: int


class Job(Protocol):
    name: str

    def validate(self, context: JobContext) -> None: ...
    def run(self, context: JobContext) -> JobResult: ...
```

Execution order is fixed by the runner:

1. load and validate configuration;
2. acquire the job lease and create the `running` run row;
3. evaluate source readiness/lineage;
4. authenticate and perform NexusCRM object/field preflight;
5. materialize the run-scoped candidate table;
6. validate source keys and every mapped payload with no target writes;
7. read current outbound record states;
8. upsert changed records in batches and persist each batch's evidence;
9. finalize the run counts/status;
10. delete the candidate table and release the lease in cleanup.

A failure before step 8 produces no CRM mutations. A failure during or after
step 8 is recovered by replaying external-ID upserts and reconciling against the
record ledger.

CLI shape:

```bash
pingcap-pipeline run-job \
  --job <job-name> \
  --config /app/configs/jobs/<job-name>.yaml

pingcap-pipeline run-job \
  --job <job-name> \
  --config /app/configs/jobs/<job-name>.yaml \
  --dry-run
```

Operational replay flags should be added with the implementation:

- `--run-id`: caller-supplied deterministic run ID for recovery/testing;
- `--retry-failed`: select rows whose current state is retryable failure;
- `--external-id`: run a one-record canary or repair;
- `--force`: ignore an unchanged payload hash, but preserve upsert semantics.

## 6. Configuration contract

Example:

```yaml
version: 1

source:
  project: gcp-digital-bi
  location: us-west2
  sql_file: ../../sql/jobs/<job-name>.sql
  readiness_sql_file: ../../sql/jobs/<job-name>-readiness.sql
  freshness_max_age_minutes: 180

target:
  base_url_env: NEXUSCRM_BASE_URL
  api_version: v63.0
  client_id_env: NEXUSCRM_CLIENT_ID
  client_secret_env: NEXUSCRM_CLIENT_SECRET

state:
  project: gcp-digital-bi
  dataset: nexuscrm_control

write:
  batch_size: 200
  all_or_none: false
  skip_unchanged: true
  delete_policy: ignore
  max_attempts: 5
  base_backoff_seconds: 1
  max_backoff_seconds: 30

schedule:
  cron: "0 3 * * *"
  timezone: UTC
  paused: true
```

Configuration owns environment, transport, state, and execution-policy facts.
The registered job implementation owns its job name. The object mapper is the
single source of truth for the target object, external ID field, field mapping,
and transformation logic; these facts must not be repeated in YAML. The
implementation must reject:

- unknown top-level keys;
- batch sizes outside `1..200` for the Composite Collections path;
- non-HTTPS target URLs outside local tests;
- `delete_policy` values other than `ignore` in v1;
- missing environment-variable names;
- source SQL paths that escape the repository root.

## 7. Source query contract

The job first runs a readiness query. It returns exactly one row:

| Column | Type | Meaning |
|---|---|---|
| `_outbound_source_run_id` | STRING | Authoritative upstream load/materialization run |
| `_outbound_source_as_of_at` | TIMESTAMP | When that source run became available |
| `_outbound_source_ready` | BOOL | Whether the source is complete and safe to publish |
| `_outbound_source_reason` | STRING | Optional not-ready diagnostic |

Freshness is calculated from `_outbound_source_as_of_at`. The readiness query
must derive it from authoritative upstream load metadata, such as
`pipeline_runs`, rather than `CURRENT_TIMESTAMP()`. This keeps readiness valid
when the data query legitimately returns zero rows.

After readiness succeeds, the data query returns exactly one row per external
ID and includes these framework columns:

| Column | Type | Meaning |
|---|---|---|
| `_outbound_external_id` | STRING | Stable target upsert key; required and non-empty |
| `_outbound_source_version` | STRING | Source revision/timestamp used for evidence |
| `_outbound_source_updated_at` | TIMESTAMP | Freshness and lineage signal |

All other columns belong to the object-specific mapper.

Before any write, the runner must fail closed if:

- an external ID is null or empty;
- the query returns duplicate external IDs;
- the source is older than `freshness_max_age_minutes`;
- a required mapper input is missing or has an invalid type;
- the mapper tries to emit a field outside its declared ownership set.

Rows are processed in stable `_outbound_external_id` order so batching and test
evidence are deterministic.

Materialize the query result to a run-scoped candidate table before validation.
Validate uniqueness and non-null keys with set-based BigQuery queries, not one
query per record. The mapper must be pure and deterministic because the runner
uses it once for the no-write validation pass and once for the write pass.

If this outbound depends on one of this repository's ingestion sources, record
the readiness query's `pipeline_runs.run_id` in `outbound_runs` as
`source_run_id`. The writeback remains a separate scheduled execution; source
readiness and lineage, rather than a chained process, form the dependency
contract.

## 8. Mapping and field ownership

The custom-object mapper is the single source of truth for the target write
contract and declares:

```python
TARGET_OBJECT = "<Custom_Object__c>"
EXTERNAL_ID_FIELD = "<Source_Record_Id__c>"
OWNED_FIELDS = frozenset({
    "<Source_Record_Id__c>",
    "<Field_A__c>",
    "<Field_B__c>",
})
```

Rules:

1. Emit only owned, createable/updateable fields.
2. Never emit server-managed fields such as `Id`, `CreatedDate`, or
   `LastModifiedDate`.
3. Omitted and `null` are different:
   - omit a field to leave the CRM value unchanged;
   - emit `null` only when the source contract explicitly owns clearing it.
4. Normalize timestamps to UTC ISO-8601 and decimals without binary-float
   conversion.
5. Reject unexpected truncation; do not silently clip strings or decimals.
6. Resolve parent references before child writes if the target object has
   dependencies. Parent/child phases are explicit in the object job, not a
   generic graph engine.

Compute `payload_hash` from canonical JSON containing the target object,
external ID field/value, and owned payload:

```text
sha256(UTF8(JSON(sort_keys=true, normalized values)))
```

Do not include run IDs, timestamps generated by the job, or OAuth metadata in
the hash.

## 9. NexusCRM API contract

### Authentication

Use OAuth `client_credentials` with a dedicated Connected App and a dedicated
run-as integration user. The token needs the `api` scope for writes. The user
must have only the target object's required object CRUD and field-level
permissions.

Secrets are provided through Secret Manager environment variables. Never use
password authentication, never put a secret in YAML, and never log tokens or
client credentials.

Implement the transport with one pooled `httpx.Client` per run and explicit
connect/read/write/pool timeouts. Add `httpx` as a direct project dependency;
do not rely on an indirect dependency or shell out to `curl`.

### Preflight

At the start of each execution:

1. acquire a token;
2. call object describe;
3. verify the object and external ID field exist;
4. verify every owned field is present and createable/updateable as required;
5. fail before selecting or writing batches if the contract drifted.

The unique constraint on the external ID is a CRM metadata/deployment
precondition and must also be verified by the pack's tests. API upsert uses a
read-then-write path, so the database unique constraint is required to close
the concurrent-create race.

### Write path

Use Composite Collections Upsert:

```http
PATCH /services/data/v63.0/composite/sobjects/{Object}/{ExternalIdField}
Authorization: Bearer <token>
Content-Type: application/json

{
  "allOrNone": false,
  "records": [ ... up to 200 records ... ]
}
```

The client returns a normalized result for every external ID:

```python
@dataclass(frozen=True)
class RecordWriteResult:
    external_id: str
    success: bool
    crm_record_id: str | None
    error_code: str | None
    error_message: str | None
    retryable: bool
```

Do not use Bulk API v2 in v1. The currently inspected runtime exposes the job
lifecycle but not a sufficient per-record failed-result retrieval contract for
this ledger. Bulk can replace the transport later without changing the job,
mapper, or state contracts once staging proves durable upload storage, complete
result retrieval, external-ID propagation, and restart behavior.

## 10. State and idempotency

Use four BigQuery tables under `nexuscrm_control`.

### `outbound_job_leases`

One current lease per job:

| Column | Type |
|---|---|
| `job_name` | STRING NOT NULL |
| `owner_run_id` | STRING NOT NULL |
| `acquired_at` | TIMESTAMP NOT NULL |
| `heartbeat_at` | TIMESTAMP NOT NULL |
| `expires_at` | TIMESTAMP NOT NULL |

Acquire/renew with an atomic conditional `MERGE`. Only the lease owner may
write a heartbeat or release it. An unexpired lease causes a new execution to
exit successfully as `skipped_already_running`, not start a competing writer.

### `outbound_runs`

One row per execution:

| Column | Type |
|---|---|
| `job_name` | STRING NOT NULL |
| `run_id` | STRING NOT NULL |
| `source_run_id` | STRING |
| `source_cutoff_at` | TIMESTAMP |
| `status` | STRING NOT NULL |
| `selected_rows` | INT64 |
| `unchanged_rows` | INT64 |
| `succeeded_rows` | INT64 |
| `retryable_failed_rows` | INT64 |
| `permanently_failed_rows` | INT64 |
| `started_at` | TIMESTAMP NOT NULL |
| `ended_at` | TIMESTAMP |
| `error_message` | STRING |

Allowed terminal states are `succeeded`, `completed_with_errors`, `failed`,
`dry_run`, and `skipped_already_running`.

### `outbound_records`

Current state per `(job_name, target_object, external_id)`:

| Column | Type |
|---|---|
| `job_name` | STRING NOT NULL |
| `target_object` | STRING NOT NULL |
| `external_id` | STRING NOT NULL |
| `payload_hash` | STRING NOT NULL |
| `source_version` | STRING |
| `crm_record_id` | STRING |
| `status` | STRING NOT NULL |
| `attempt_count` | INT64 NOT NULL |
| `last_attempt_run_id` | STRING |
| `last_success_run_id` | STRING |
| `next_retry_at` | TIMESTAMP |
| `error_code` | STRING |
| `error_message` | STRING |
| `updated_at` | TIMESTAMP NOT NULL |

Current record states are `succeeded`, `retryable_failed`, and
`permanent_failed`. A succeeded row with the same payload hash is skipped. An
unchanged permanent failure is also skipped until its source payload changes or
an operator explicitly forces replay; this prevents the same known data error
from failing and alerting every scheduled run.

### `outbound_record_attempts`

Append-only, time-partitioned evidence with 90-day retention:

| Column | Type |
|---|---|
| `attempted_at` | TIMESTAMP NOT NULL |
| `job_name` | STRING NOT NULL |
| `run_id` | STRING NOT NULL |
| `batch_id` | STRING NOT NULL |
| `external_id` | STRING NOT NULL |
| `payload_hash` | STRING NOT NULL |
| `attempt_number` | INT64 NOT NULL |
| `status` | STRING NOT NULL |
| `crm_record_id` | STRING |
| `http_status` | INT64 |
| `error_code` | STRING |
| `error_message` | STRING |
| `duration_ms` | INT64 |

Do not store the full payload in control tables because it can contain customer
data. Store the source version and payload hash; use the source query to
reconstruct a payload during controlled troubleshooting.

State persistence is batch-oriented. Append attempt rows and `MERGE` current
record states once per API batch; never issue one BigQuery DML statement per
record. The update for a batch must be complete before the runner submits the
next batch.

## 11. Retry and failure policy

| Condition | Behavior |
|---|---|
| OAuth 401 | Refresh token once, then retry the batch once |
| HTTP 408, 429, 5xx, connection reset/timeout | Exponential backoff with full jitter; honor `Retry-After` |
| Validation, invalid field/value, FLS/CRUD denial | Permanent per-record failure |
| Object/field missing during preflight | Fail the whole run before writes |
| Duplicate external ID / unique conflict | Permanent failure and blocking alert |
| Malformed or incomplete response | Treat every unaccounted record as retryable; never assume success |
| Lease loss | Stop before the next batch and fail the run |

Retries happen in application code. Start the Cloud Run Job with
`max_retries = 0`; the external-ID upsert makes a later operator or scheduler
replay safe without an infrastructure layer restarting the whole process at an
unknown point.

`write.max_attempts` limits transport attempts for one batch in one execution;
it does not turn a transient remote failure into a permanent data failure.
`outbound_records.attempt_count` counts attempts for the current payload hash
and resets when that hash changes. A retryable row is eligible again after
`next_retry_at` in a later execution.

After each batch, persist attempts and current state before moving to the next
batch. If state persistence fails after the CRM request, stop the run. A replay
may send the batch again, but external-ID upsert makes that safer than losing
the success evidence and advancing.

Run exit policy:

- exit `0` for `succeeded`, `dry_run`, or `skipped_already_running`;
- exit non-zero for infrastructure/auth/schema failures or remaining retryable
  failures;
- exit non-zero when a new or payload-changed permanent failure is observed;
- do not re-alert unchanged permanent failures on later runs.

## 12. Deletion and conflict policy

`delete_policy: ignore` is mandatory in v1. A source row disappearing must not
delete a CRM record. Future deletion behavior needs a separately reviewed
business contract, preferably an explicit source tombstone mapped to a target
inactive/status field rather than hard delete.

Each emitted field has one declared system of record. The job must never read a
field from `nexuscrm_final` and write that same fact back into NexusCRM. Any
NexusCRM-to-BigQuery-to-NexusCRM flow requires an explicit field-level ownership
review to prevent feedback loops.

## 13. Security and IAM

Create a dedicated GCP service account for this job with only:

- `roles/bigquery.jobUser` on the project;
- `roles/bigquery.dataViewer` on the source dataset/view;
- data-editor permission limited to `nexuscrm_control`;
- Secret Manager accessor permission on the exact NexusCRM client ID/secret
  secrets, not all project secrets.

Logging rules:

- structured JSON with `job_name`, `run_id`, `batch_id`, counts, durations, and
  error codes;
- never log OAuth responses, authorization headers, full record payloads, or
  unredacted customer fields;
- external IDs may be hashed in logs when they are customer-identifying;
- truncate and sanitize remote error messages before persistence/logging.

## 14. Terraform and deployment

Add a dedicated `google_cloud_run_v2_job`, scheduler, service account, secret
bindings, and monitoring policy wiring. Suggested resource name:

```text
pingcap-<job-name>-nexuscrm-writeback
```

Initial runtime settings:

- task count: `1`;
- parallelism: `1`;
- timeout: `3600s`, adjusted after measured runs;
- max retries: `0`;
- scheduler paused on first deployment;
- same container image as the existing pipeline jobs;
- command: `run-job --job <job-name> --config /app/configs/jobs/<job-name>.yaml`.

The explicit job allowlist in `.github/workflows/deploy.yml` and its contract
test must be updated in the same change so the new job receives image updates.
The implementation change must also:

- add `httpx` to `pyproject.toml` and refresh `uv.lock`;
- make `Dockerfile` copy the new `sql/` directory into the image;
- update the README's pipeline shape, runtime variables, and validation
  commands;
- add the new managed job to the deploy workflow allowlist and its test.

The existing `pingcap-*` Cloud Run failure alert will cover whole-job failures;
add log-based metrics or a custom metric for:

- selected/succeeded/unchanged/failed row counts;
- request and batch latency;
- HTTP 429/5xx counts;
- auth/schema preflight failures;
- lease contention and lease loss.

## 15. Testing strategy

### Unit tests

- strict YAML validation and safe relative SQL path resolution;
- readiness must return exactly one authoritative source run/as-of row;
- source row parsing and required-column enforcement;
- duplicate/null external ID rejection;
- mapper field ownership, type conversion, and null-vs-omit behavior;
- canonical payload hashing stability;
- response normalization and retry classification;
- batch size and stable ordering;
- state transitions and unchanged-row skipping;
- token refresh, `Retry-After`, timeout, and malformed-response handling;
- secret/payload redaction in logs.

### Contract tests with a fake HTTP server

- OAuth token exchange;
- describe preflight;
- a mixed-success Composite Upsert response;
- 401 refresh and retry;
- 429 and 5xx backoff;
- response missing one requested record;
- connection timeout after the server accepted a request.

### Staging acceptance

1. Upsert one new external ID.
2. Upsert the same ID with the same payload; confirm one CRM row and a skipped
   subsequent pipeline write.
3. Change one owned field; confirm update, not create.
4. Send one valid and one invalid record in the same `allOrNone=false` batch;
   confirm per-record evidence.
5. Confirm target validation rules, flows, FLS, audit, and ownership behavior.
6. Kill the job after a successful API request but before state commit; replay
   and confirm no duplicate row.
7. Attempt two simultaneous executions; confirm one obtains the lease.

## 16. Rollout

1. Land framework, job code, Terraform, and tests with the scheduler paused.
2. Verify target object metadata, unique external ID, Connected App, and
   integration-user permissions in staging.
3. Run `--dry-run` and reconcile selected counts with the source owner.
4. Run a small allowlisted canary using `--external-id`.
5. Run the full job manually in staging and then production.
6. Reconcile CRM IDs and sampled owned fields against BigQuery.
7. Enable the schedule and monitor at least three complete cycles.

## 17. Inputs required before implementation

The framework design is complete, but the concrete job needs these reviewed
facts:

1. job/resource name;
2. custom object API name;
3. unique external ID field API name;
4. source BigQuery query or view;
5. exact owned target fields and null/clearing semantics;
6. source freshness expectation and schedule;
7. expected row count and change volume;
8. parent-reference requirements, if any.