Connectors: getting data in

Intake: the scoped way in

The document store behind everything (repo-service, covered on its own page) checks every caller itself. Its gRPC listener requires the operator API token to open, and each call presents that token or a credential the host resolves through an access policy, OIDC token introspection, or a JDBC caller table; its HTTP routes also accept a console session signed in through the organisation's OpenID provider. A scoped caller reaches only the accounts its bindings name, and writes only as the intake owners those bindings list (see callers and sign-in). Intake is the surface built for ingest: the network door where a program proves itself with an API key, gets bound to a narrow allowance of account, datasources, drives, and MIME types, and hands over bytes.

The proof of identity never travels inside a message field. It rides in gRPC metadata, the standard x-api-key header or the usual authorization header, and the server resolves it to an IntakeScope: an account id, the datasource ids this key may use, the drives it may write, the MIME types it may send, and a payload size cap. The contract says it in a header comment worth reading whole:

dev-tools/protomolt intake module, intake_service.proto, the auth rule stated in the contract

// AUTH IS NOT A PROTO FIELD. The credential rides gRPC metadata
// (x-api-key, or the standard authorization header) and is resolved
// server-side to a scope {account_id, allowed datasource_ids,
// allowed drives, permissions}; the request messages below only
// NARROW within that scope, never widen it.

The narrowing rule has two visible consequences. A request aimed at a datasource or drive outside its scope fails with PERMISSION_DENIED: the server knows who you are, and this is not yours. A key that resolves to nothing fails earlier with UNAUTHENTICATED. The error split is deliberate: "who are you?" and "not for you" are different answers, and callers should be able to tell them apart.

One more rule falls out of the scope model: ownership fields the scope dictates win over whatever the caller supplied. If the scope says this key writes to the intake drive, a request claiming the pipeline drive is rewritten, not executed.

Three lanes, one gate

The same gate serves three shapes of upload. IngestDocument is a single unary call for a program that already has a complete document or a raw payload (bytes, a filename, and a declared MIME type that is treated as a hint; the parsing coordinator sniffs the real content type from magic bytes later). IngestStream is a client-streaming call for large payloads, with strict frame discipline: the first frame must carry metadata and every later frame must carry data, and breaking that order is an INVALID_ARGUMENT. And POST /v1/intake:upload is a plain HTTP route for the simplest possible client, a curl one-liner that sends the key as a header:

The HTTP lane; identical scope rules, no gRPC needed

curl -X POST http://intake-host/v1/intake:upload \
  -H 'x-api-key: <your key>' \
  --data-binary @quarterly-plan.pdf

All three lanes return the same receipt vocabulary: a doc_id, a node_id and address inside the store, the drive, the size, a server-computed SHA-256, and a deduplicated flag. And all three save through repo-service's wire contract; intake itself never touches object storage.

The document id is the content hash

This is the detail that makes re-uploads safe. The doc id is derived from the payload's SHA-256, so identical bytes always map to the same document. Re-running a crawler that already ran yesterday does not create duplicates; the second upload converges on the same id and the receipt says deduplicated: true. Deterministic ids turn "import everything again" from a cleanup chore into a no-op.

Where keys live

Three interchangeable stores sit behind the key-resolution seam. An in-memory store covers tests, demos, and single-tenant runs. A JDBC store covers the real case: keys live in PostgreSQL, and the table is keyed by the lowercase hex SHA-256 of the credential, never the credential itself, so a dump of the key table authenticates nobody. Revocation is a revoked_at timestamp kept as an audit row rather than a delete, which makes key rotation with a grace period just two live rows resolving to the same account. An OIDC introspection store (RFC 7662) treats an API key as an identity-provider-issued opaque token, with the scope riding the token's claims and revocation taking effect on the next call. One error discipline binds all three: a broken key store throws an internal error rather than answering like a bad key, because an outage must never masquerade as the caller's fault.

The reach-in side: acquire, pulling from sources that keep their own documents

Push is not always the right shape. A bucket full of PDFs, a table with a LONGVARBINARY column, a wiki that already organizes a company's knowledge: these sources already hold the documents, so ProtoMolt's acquire family reaches in instead. It splits into three groups with very different levels of maturity. A web crawler for public sites and a Google Drive connector are planned but do not exist yet.

Verbs for buckets and databases

Two contributed verbs cover the simplest sources. pull-s3 pulls objects from an S3 bucket and pull-jdbc pulls rows over JDBC. Both are one bounded pass per call: there is no crawl engine and no scheduler anywhere in the tree, so each invocation copies a defined slice and finishes. The verbs ship and work, but nothing schedules them: they run when a caller invokes them, not on their own.

The gRParse fleet proxies: Confluence and Microsoft Graph as typed gRPC

Hosted knowledge bases are harder than buckets: they paginate, they update, they carry attachments, and their native APIs are REST with vendor quirks. The gRParse fleet provides two proxies that speak a vendor REST API upstream and typed gRPC downstream. Its Confluence service exposes spaces, pages, blog posts, comments, and attachments, including optional attachment bytes up to a 25 MiB cap. A cursor-based Sync stream sends a bounded incremental crawl, and completed runs can write an Open Knowledge Format bundle, a sibling WARC 1.1 archive, a protobuf Kafka stream, and sync-ledger records. Its Microsoft Graph service exposes sites, drives, and files through GetItem, DownloadItem, and Sync, with SharePoint columns mapped to typed fields. The fleet also has Streamable HTTP MCP tools for connection setup, output settings, and syncs. These are independently deployable gRParse fleet services; the Java acquire/msgraph library is a separate framework building block.

flowchart TD
          a["Request arrives, key in metadata"] --> k["Does the key resolve?"]
          k -- "no: UNAUTHENTICATED" --> r1["Refused: unknown key"]
          k -- "yes" --> s["Targets inside the scope?"]
          s -- "no: PERMISSION_DENIED" --> r2["Refused: outside this key's scope"]
          s -- "yes" --> p["Within the payload cap?"]
          p -- "no: RESOURCE_EXHAUSTED" --> r3["Refused: over the cap"]
          p -- "yes" --> d["doc_id = SHA-256 of the payload"]
          d --> v["Bytes already known?"]
          v -- "yes" --> dup["Same id, receipt says deduplicated"]
          v -- "no" --> save["Saved through repo-service, receipt returned"]

The standing bridge: Kafka Connect, in both directions

The pull verbs above are one bounded pass per call. For data that flows continuously, ProtoMolt ships a Kafka family under sink/kafka: a serde that puts protobuf messages on topics in the standard wire format, a Kafka Connect sink and source that bridge topics to gRPC and back, and transforms that apply the schema's own rules inside the Connect worker. Kafka Connect is Kafka's own framework for running connectors as managed, long-lived workers, so ProtoMolt's pieces plug into infrastructure many teams already operate.

The serde: standard frames, loud refusal

A serde (serializer and deserializer) is the pair of adapters a Kafka client uses to turn a typed message into topic bytes and back. ProtoMolt's speaks the Confluent wire format, the de facto standard framing where each record carries the id of the exact schema that wrote it, with subject naming strategies identical to Confluent's. That means the registry behind it can be ProtoMolt's own git-backed registry or any Confluent-compatible one, and consumers that know the wire format can read the topics without ProtoMolt on the classpath. The write path enforces the schema's declared validate.v1 rules before a frame leaves the producer, and the read path checks them again, so a topic guarded by this serde carries only messages that pass their own declarations. Schema ids are cached with independent backoff windows per subject, and when the registry cannot answer, the serde refuses loudly rather than falling back to a guessed id, because a wrong id silently poisons every downstream consumer.

The sink: a topic delivered into any gRPC method

The Connect sink takes records from a topic and delivers each one as a call to a gRPC method you name, any method on any service, resolved through the same descriptor machinery as the rest of the platform. Pointed at intake's IngestDocument, it turns a Kafka topic into a document feed with the full admission gate applied to every record. Pointed at anything else, it is a generic topic-to-RPC bridge with no adapter code to write. The services workbench in the browser console generates a ready-made Connect sink configuration for any registered service, the same way it emits the claude mcp add line for agents.

The source: a server stream becomes a topic, resumably

The Connect source runs the other way: it subscribes to a gRPC server-streaming method and publishes each streamed message to a topic. Its distinguishing feature is how it survives restarts. Connect asks every source connector to report an offset, a bookmark it can resume from; this source computes that bookmark by evaluating a CEL expression you declare against the last message it saw, so any field that orders the stream (a sequence number, a modified-at timestamp, an opaque cursor) can serve as the resume token. The README is direct about its standing: this is the only acquisition path in the tree with durable resumption. pull-s3 and pull-jdbc finish their pass and forget; the source picks up where it stopped.

Transforms: the schema's rules, applied in flight

Between a topic and its destination, Connect can run single-message transforms, and ProtoMolt ships four: validate a record against its schema's declared rules, map fields with the same rule language the mapper family uses, redact sensitivity-classified fields, and filter records with a CEL predicate. None of them introduces a new configuration vocabulary. They read the same descriptor options as every other gate, and the CEL engine is the same single module that serves validation, projection, and routing everywhere else, so a rule tested with map-message behaves identically inside a Connect worker. One declaration, every boundary.

Two further Connect sinks carry documents onward rather than inward: an Iceberg sink writes topics into lake tables (the Parquet files the metric service's DuckDB backend aggregates over), and an OpenSearch sink indexes them. On the Iceberg side, equality deletes and v3 variant columns are later phases. A planned S3 bundle sink under sink/emit does not exist yet.

flowchart LR
          topic["Kafka topic, frames in the Confluent wire format"] --> tf["Connect transforms: validate, map, redact, CEL filter"]
          tf --> rpc["gRPC sink: any method, e.g. intake IngestDocument"]
          tf --> ice["Iceberg sink: lake tables as Parquet"]
          tf --> os["OpenSearch sink: index documents"]
          ice --> duck["metric service, DuckDB backend: SELECT ... GROUP BY over the lake"]
          os --> q["search queries"]

The platform also eats its own cooking here: the document store's lifecycle events (saved, deleted, purge requested, purged) leave through a transactional outbox onto a document-events topic via this serde, keyed by document id so each document's history stays ordered, and the optional Kafka purge queue publishes purge commands validated on write by their own declared rules.