# Confluent — API: The Complete HTTP Surface for Confluent Schema Registry Governance

This is the reference we wish every integration shipped with: every route, every role, every payload - nothing implied. CoreModels exposes eight core verbs for a Schema Registry estate across two HTTP surfaces, and this article walks all of them with real request and response bodies. The vendor key is `confluent`; the connector declares **Import, Audit, Generate**, so every verb below is genuinely available - including artifact generation, which not all of our connectors support.

# The Complete HTTP Surface for Confluent Schema Registry Governance

This is the reference we wish every integration shipped with: every route, every role, every payload - nothing implied. CoreModels exposes eight core verbs for a Schema Registry estate across two HTTP surfaces, and this article walks all of them with real request and response bodies. The vendor key is `confluent`; the connector declares **Import, Audit, Generate**, so every verb below is genuinely available - including artifact generation, which not all of our connectors support.

Two surfaces, two token types:

- The **interactive surface** (`graph/integrations/...`) authenticates with your normal CoreModels login token. It carries the full verb set.
- The **machine-to-machine surface** (`v1/...`) accepts user API keys and is what CI should call. It carries `audit` and `badge`; `reaudit` and `history` live on the interactive surface only.

Throughout, `https://coremodels.example.com` stands in for your deployment's API host, `$TOKEN` for the bearer token, and `$PROJECT_ID` for the 32-character hex id of the governing project.

## Discovery - `GET graph/integrations/vendors`

Any authenticated user can list the registered connectors and what each expects:

```bash
curl -sS "https://coremodels.example.com/graph/integrations/vendors" \
  -H "Authorization: Bearer $TOKEN"
```

The Confluent entry in the response tells you the whole artifact contract in one line:

```json
{ "key": "confluent", "displayName": "Confluent Schema Registry",
  "capabilities": "Import, Audit, Generate",
  "artifacts": { "subjects": "required - JSON array of GET /subjects/{subject}/versions/latest responses ({subject, version, id, schemaType, schema})" } }
```

## Import - `POST graph/integrations/confluent/import/{projectId}` (Admin)

Import writes the estate into the graph, additively - existing governed nodes are never mutated. The body is an artifacts map; for this connector the only artifact name is `subjects`, and its value is the raw registry export as a string:

```http
POST https://coremodels.example.com/graph/integrations/confluent/import/$PROJECT_ID
Authorization: Bearer $TOKEN
Content-Type: application/json

{ "artifacts": { "subjects": "<contents of subjects.json>" },
  "spaces": [] }
```

`spaces` is optional; empty means the project's main space. Omitting `artifacts` entirely earns you the exact error `Body must include 'artifacts': { "<name>": "<content>" } (e.g. manifest for dbt).`, and sending an artifacts map without `subjects` fails with `The 'subjects' artifact (aggregated Schema Registry export) is required.` The success response is a counts object - `datasetsAdded`, `fieldsAdded`, `nodesEnriched`, `snapshotStored`, and the `lossiness`/`errors` channels.

## Audit - `POST graph/integrations/confluent/audit/{projectId}` (Viewer)

The audit is strictly read-only. It parses a fresh export and compares it to the governed model across coverage, drift, and conformance:

```http
POST https://coremodels.example.com/graph/integrations/confluent/audit/$PROJECT_ID
Authorization: Bearer $TOKEN
Content-Type: application/json

{ "artifacts": { "subjects": "<subjects.json>" }, "recordHistory": true }
```

`recordHistory` defaults to false - the audit verb stays strictly read-only unless asked. The response carries counts, per-code totals, drifted object identities, an artifact `fingerprint`, structured findings, and a `markdown` report ready to paste into a pull request:

```json
{ "success": true, "vendor": "confluent", "projectName": "schema-registry",
  "errorCount": 1, "warningCount": 0, "infoCount": 1,
  "codes": { "field-type-drift": 1, "fields-no-doc": 1 },
  "driftedObjects": ["payments-value.amount"],
  "fingerprint": "9f2c4b1a8e7d3c05",
  "findings": [
    { "section": "Drift", "severity": "Error", "code": "field-type-drift",
      "subject": "payments-value.amount", "message": "Field type changed since the last import.",
      "detail": "governed: double, estate: string" },
    { "section": "Conformance", "severity": "Info", "code": "fields-no-doc",
      "subject": "payments-value",
      "message": "2 field(s) carry no doc - consumers and agents are guessing their meaning.",
      "detail": "amount, currency" }
  ],
  "markdown": "…", "historyRecorded": true, "lossiness": [] }
```

## Re-audit - `POST graph/integrations/confluent/reaudit/{projectId}` (Viewer)

The audit asks whether fresh artifacts still conform to the governed model. Re-audit asks the opposite question - the governed model changed; does the *last-known estate* still conform? It runs the same audit engine over the snapshot stored at import time, so it needs no artifacts at all:

```http
POST https://coremodels.example.com/graph/integrations/confluent/reaudit/$PROJECT_ID
Authorization: Bearer $TOKEN
Content-Type: application/json

{}
```

An optional body of `{ "projectName": "schema-registry" }` selects a specific stored snapshot; null means the latest. Unlike the audit verb, a re-audit run is always recorded in the history. Note the dependency: re-audit requires that import stored a snapshot. Very large registries whose encoded snapshot exceeds the storage cap (roughly 1.5 MB) report `snapshotStored: false` at import with a lossiness record - fresh-artifact audits still work, but there is nothing for `reaudit` to run against.

## History - `GET graph/integrations/confluent/history/{projectId}` (Viewer)

The rolling audit trail, newest first:

```bash
curl -sS "https://coremodels.example.com/graph/integrations/confluent/history/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN"
```

```json
{ "success": true, "vendor": "confluent",
  "projects": [ { "projectName": "schema-registry",
    "runs": [ { "at": "2026-07-27T09:14:02.4610215+00:00", "trigger": "reaudit",
                "errorCount": 0, "warningCount": 0, "infoCount": 2,
                "codes": { "fields-no-doc": 2 }, "fingerprint": "9f2c4b1a8e7d3c05" } ] } ] }
```

`trigger` is `audit`, `reaudit`, or `ci` depending on which surface and verb recorded the run (`scheduled` marks runs recorded by the server-side heartbeat, where a deployment enables it). Matching fingerprints across runs mean the registry export did not change between them.

## Badge - `GET graph/integrations/confluent/badge/{projectId}` (Viewer)

Returns a self-contained shields-style SVG (`image/svg+xml`) reflecting the latest recorded run: green for clean, yellow for warnings, red for errors, gray when no runs are recorded. The label reads `confluent audit`.

## Generate - `POST graph/integrations/confluent/generate/{projectId}` (Viewer)

This connector closes the loop: it emits one registry-ready Avro record schema per governed Type, named `schemas/{Record}.avsc`, with kind `json`:

```http
POST https://coremodels.example.com/graph/integrations/confluent/generate/$PROJECT_ID
Authorization: Bearer $TOKEN
Content-Type: application/json

{ "typeNames": [] }
```

`typeNames` restricts generation to the named Types; empty means everything eligible. The request shape also accepts `targetVersion` and `extra` for connectors with dialect versions - the Confluent generator does not consume them, and we say so rather than let you wonder. The response:

```json
{ "success": true,
  "artifacts": [ { "name": "schemas/Order.avsc", "kind": "json",
    "content": "{\n  \"type\": \"record\",\n  \"name\": \"Order\", …" } ],
  "lossiness": [], "errors": [] }
```

Inside each `.avsc`: required fields as bare Avro types, optional fields as `["null", T]` with `default: null`, governed taxonomies as sanitized Avro enums, DateTime as `timestamp-millis` logical types, and references emitted as named record types when the target record is generated in the same call. Governed Types with no elements are skipped with a declared lossiness record (an Avro record needs at least one field), and if nothing at all is eligible the call fails honestly with `No eligible governed types found to emit as Avro records.`

## Status - `GET graph/integrations/confluent/status/{projectId}` (Viewer)

The last-import state: whether an import has happened, when, the artifact fingerprint, and counts - including the facts the parser recorded, such as how many subjects were Protobuf or structurally unparseable. The response also reports `governedDatasets`, the number of imported subjects currently resolvable to governed Types.

## The machine-to-machine surface

Two routes accept user API keys, both at Viewer role. The audit response is wrapped in the standard `ApiResponse` envelope (counts live under `data.*`); the badge returns the raw SVG:

```
POST https://coremodels.example.com/v1/$PROJECT_ID/integrations/confluent/audit
GET  https://coremodels.example.com/v1/$PROJECT_ID/integrations/confluent/badge
```

The v1 audit takes the same body as the interactive audit; when `recordHistory` is true the run lands in the trail with trigger `ci`. The contract your pipeline gates on is `data.errorCount > 0` ⇒ fail the build. The v1 badge is the same SVG, embeddable in a README with an API key.

## Roles, errors, and the routes beside the verbs

Import requires **Admin** on the project. Audit, re-audit, history, badge, generate, and status all run at **Viewer** - they never write governed meaning (recording an audit run is opt-in bookkeeping; re-audit always records). An unknown vendor key returns `success: false` with `Unknown vendor '<v>'. Registered: <the connector keys>.` - the list is live, so it also doubles as discovery when you're debugging a typo.

One further route sits beside the per-vendor verbs rather than inside them: `POST graph/integrations/reconcile/{projectId}` (Admin) takes `{ "vendorA": "...", "vendorB": "..." }` and links datasets that two governed estates describe as the same physical relation, writing reciprocal `sameAs` mappings on both sides. It matches on normalized physical names, and for a registry estate the physical name is the topic - so it pays off when your topic names line up with another estate's relation names, and reports the rest as unmatched rather than guessing.

The interactive API also carries the sync-plan surface, newer than the verbs above and all at Viewer role: `POST graph/integrations/{vendor}/sync/propose/{projectId}` classifies fresh artifacts into a stored, reviewable sync plan (its only writes are bookkeeping - the plan blob and a ledger entry, never governed meaning), while `GET graph/integrations/sync/plan/{projectId}/{planId}` and `GET graph/integrations/sync/ledger/{projectId}` read a stored plan and the rolling plan ledger back.

That is the surface: the eight core routes plus reconcile and the sync-plan trio on the interactive API, two on the machine surface, one artifact name, and no credentials held by us at any point. For the extraction recipe and a guided first run, see the Confluent Schema Registry quickstart in the CoreModels docs.