Connectors & Intake

All data enters ProtoMolt through one hardened front door: connector-intake-service, which authenticates every push, derives a deterministic document identity, applies a content policy, and stages accepted work onto a Redis stream that the engine drains on its own schedule. Connectors — database crawlers, S3, filesystems, Confluence, SharePoint — live outside the processing mesh and only push. Intake is admission, not processing: it never parses, chunks, embeds, or routes.

The intake front door: three push paths

Intake terminates on a single port, 18108, with HTTP and gRPC sharing one listener (unified server mode is the platform default). Three paths in, all authenticated the same way:

• gRPC unary — UploadPipeDoc for a fully-hydrated PipeDoc, and UploadBlob for raw bytes plus minimal metadata.
• gRPC bidirectional streaming — UploadPipeDocStream: the first message is a StreamContext, the server answers with a StreamHandshake carrying the resolved identity, and subsequent messages are PipeDocItem documents or delete_ref requests. One long-lived stream per crawl.
• Chunked HTTP POST — POST /uploads/raw accepts a raw body up to 10 GB (quarkus.http.limits.max-body-size=10G) with metadata in headers (x-api-key, x-doc-id/x-source-doc-id, x-filename, x-checksum-sha256). The body is streamed straight through to repository-service with HTTP/2-native backpressure — never buffered to memory or disk in intake.

pipestream-protos/intake/proto/ai/pipestream/connector/intake/v1/connector_intake_service.proto

service ConnectorIntakeService {
  // 1. The "Power User" path: Client sends a full PipeDoc.
  rpc UploadPipeDoc(UploadPipeDocRequest) returns (UploadPipeDocResponse);

  // 2. The "Simple" path: Client sends raw bytes (Blob) + minimal metadata.
  rpc UploadBlob(UploadBlobRequest) returns (UploadBlobResponse);

  // 3. High-Performance Streaming path: Bidirectional stream for documents and deletes.
  rpc UploadPipeDocStream(stream UploadPipeDocStreamRequest) returns (stream UploadPipeDocStreamResponse);

Intake as the cop

There is no anonymous path and no degraded mode. Every call carries an API key in the x-api-key gRPC metadata header (or the same header over HTTP). Intake resolves it against connector-admin (port 18107, the API-key authority) and then checks the account is active against account-service (18105). A missing or unresolvable key returns UNAUTHENTICATED; an inactive or missing account returns PERMISSION_DENIED. Whatever ownership the caller asserted on the document is overwritten with the key-resolved identity — callers never assert a datasource id; identity comes back from ResolveApiKey.

Keys have the wire format psk_<keyId>_<secret>. The 12–16 character keyId is stored plaintext (unique, indexed) and resolves the credential row; the secret exists only as an Argon2id hash, returned in plaintext exactly once at issuance. Unknown keyId and secret mismatch return the identical "Invalid API key" message, so the endpoint is not an enumeration oracle. A hot crawl does not re-run Argon2id per document: the resolution cache is keyed by a SHA-256 of the entire key string, so a cache hit proves the caller presented byte-for-byte the same secret that already passed verification — the cache cannot bypass auth.

connector-admin/…/credentials/ApiKeyFormat.java

/** Wire prefix identifying a Pipestream API key. */
public static final String PREFIX = "psk_";
/** Minimum keyId length accepted by the parser. */
public static final int KEY_ID_MIN_LENGTH = 12;
/** Maximum keyId length accepted by the parser (and the generated length). */
public static final int KEY_ID_MAX_LENGTH = 16;

Credentials are structurally banned from message bodies: the upload request protos reserve the field number and name api_key, so a credential can never be serialized into a logged or persisted message.

connector_intake_service.proto — UploadPipeDocRequest

message UploadPipeDocRequest {
  // Optional datasource cross-check. Authentication resolves the authoritative
  // datasource from the x-api-key metadata; a non-blank mismatch is rejected.
  string datasource_id = 1;
  // Authentication travels in the x-api-key gRPC metadata header, never in
  // the message body.
  reserved 2;
  reserved "api_key";

Key rotation retires the outgoing primary key into the delegated-key table as holder rotation-grace for a bounded grace window (pipestream.connector-admin.api-key.rotation-grace, default PT15M) so running crawls finish; invalidate_old_immediately=true is the incident-response kill switch. Connectors running async crawls hold delegated keys (IssueDelegatedApiKey / RevokeDelegatedApiKey, optional ttl_seconds) instead of borrowing the customer's primary; issuance under a presented key requires proof of possession — the presented key must resolve to the same datasource.

Accounts are top-level tenants, and provisioning is fail-closed: CreateAccount synchronously provisions both storage drives (<account>:intake and <account>:pipeline) via the repository's FilesystemService.CreateDrive. On failure the account is marked inactive and the call fails with FAILED_PRECONDITION — an account never exists without its drives. Each DriveSpec chooses its backing: the shared default bucket, a dedicated BUCKET_TYPE_MANAGED bucket (created), or BUCKET_TYPE_CUSTOMER (an existing bucket, never created, with no silent fallback).

Deterministic document identity

Every document gets an id derived from its canonical source identity: {datasource_id}:{name-UUID(canonical source identity UTF-8)}, with the canonical form picked by priority — client doc_id, then source_doc_id, then a canonicalized source_uri, then a normalized source_path. Retries, duplicates, and even rejected items converge on the same id, so a rejected upload still gets a pollable tracking token and a retry of the same source document updates the same receipt row. Derived ids are uniformly hashed rather than pasting the raw source URL into the id — an earlier scheme leaked spaces and unicode into identity and broke SigV4-signed S3 headers.

connector-intake-service/…/pipedoc/PipeDocIdDeriver.java

private static String sourceUuid(String canonical) {
    String seed = canonical == null ? "" : canonical;
    return UUID.nameUUIDFromBytes(seed.getBytes(StandardCharsets.UTF_8)).toString();
}

Staging: Redis, claim-check, and signed status tokens

The intake→engine transport is Redis and nothing else: one XADD per accepted document onto the stream pipestream:intake:ingress. The old direct gRPC handoff to the engine was removed; the engine drains the stream via consumer group engine-intake, with XAUTOCLAIM for stalled entries and a poison stream pipestream:intake:ingress:poison for irrecoverable ones. Acceptance returns to the caller as soon as the envelope is in Redis — intake never blocks on the engine.

connector-intake-service/…/ingress/RedisStreamIntakeIngressProducer.java (the entire enqueue path)

@ConfigProperty(name = "pipestream.intake.ingress.stream-key", defaultValue = "pipestream:intake:ingress")
String streamKey;

@Override
public EnqueueResult enqueue(IntakeHandoffRequest request, String sourceDocId) {
    String messageId = streams.xadd(streamKey, IntakeIngressEnvelope.fromRequest(request, sourceDocId).toRedisFields());
    return new EnqueueResult(messageId);
}

Stream entries come in two payload modes. Small documents ride inline in the envelope. Large bodies are claim-check: HTTP and gRPC blob bodies go to repository-service (S3-backed) first; the repository emits an IntakeRepoEvent(CREATED) on Kafka, and intake's consumer then XADDs a reference-only handoff (document_ref with keep=true) — payload mode reference. One stream, two payload modes.

Every admission decision emits DocumentAccepted / DocumentRejected / DocumentError to Kafka, projected back into an upload_receipts Postgres table. Clients poll GetUploadStatus over gRPC or GET /uploads/status/{token} with an HMAC-SHA256-signed status token scoped to one document; an unknown receipt returns PENDING, never a 404 enumeration oracle. The token signing key has no production default — an unset INTAKE_UPLOAD_TOKEN_SIGNING_KEY fails startup.

Crawl sessions, replay, and content policy

A crawl is a session, not just a burst of uploads: StartCrawlSession, Heartbeat, and EndCrawlSession record runs in the crawl_sessions table, and EndCrawlSession — the only close primitive — stamps COMPLETED or FAILED from the caller's summary. GetCrawlHistory serves past sessions, and every intake-bound call carries an x-connector-instance header (<logical id>#<per-boot run id>) recorded into sessions and handoffs for audit — explicitly never an authorization input.

Replay without re-crawl: ReplayDocuments (server-streaming) queries the repository and re-stages reference handoffs onto the same ingress stream, so a new pipeline graph can be run over already-crawled content without touching the source system again. Husked documents (receipt-only) are skipped and counted; each run writes a replay_requests row, and the replay id doubles as the crawl id in traces.

At the boundary, a content policy rejects blocked extensions — executables, installers, archives — as UNSUPPORTED_MEDIA_TYPE before acceptance. The check is case-insensitive, compound-extension aware (report.pdf.exe is caught), and strips query/fragment parts; gz and tar.gz are deliberately allowed because gzip is a first-class crawl corpus format. Failure status is honest about retryability: only RESOURCE_EXHAUSTED, UNAVAILABLE, DEADLINE_EXCEEDED, ABORTED, and half-closed INTERNAL are marked retryable=true; validation, auth, and policy failures are permanent and never retried.

Connectors

Connectors run outside the processing mesh as ordinary services and push into intake; the engine never dials them on the data path. The connector-admin seeds a catalog with s3, file-crawler, and jdbc entries (all UNMANAGED), plus JSON Schemas for frontend configuration wizards. Both shipped crawlers seal their credentials — the intake API key, the JDBC target password, S3 static credentials — as AES-GCM tokens via the shared connector extension (pskms1 for KMS-wrapped per-record data keys, pslocal1 for an env key). Sealing is fail-closed: no usable key material means registration fails, never a plaintext fallback, and past migrations destroyed previously-plaintext columns rather than converting them, forcing rotation.

JDBC connector

The JDBC connector (port 18121) maps SQL rows to PipeDocs and pushes them over one UploadPipeDocStream per crawl. Column conventions do the mapping: aliases _ps_doc_id, _ps_title, _ps_body, _ps_source_uri, _ps_language, _ps_author, _ps_category, _ps_account_id, _ps_acl, _ps_created_at, _ps_updated_at map to PipeDoc fields; _ps_tags_<group> and _ps_meta_<key> go to metadata; every other column is packed as typed values (short strings auto-promote to searchable metadata). Crawl definitions can additionally carry a projection — ordered {source, target, transform} rows with CEL expressions (compiled once per crawl, value and the whole row in scope) targeting a bounded vocabulary such as search_metadata.title or ownership.acl; targets outside the vocabulary are rejected with a 400 at save time. For live freshness, an embedded Debezium Postgres logical-replication slot streams changes with dirty-key coalescing (refresh_flush_millis, default 500 ms). The plain query path works on anything with a JDBC driver (Postgres, MySQL, and H2 drivers ship); CDC requires wal_level=logical and a REPLICATION role on the target. The uploader is backpressure-honest: since the async gRPC StreamObserver buffers rather than parking on HTTP/2 flow control, the crawl stream enforces a 512-permit in-flight semaphore, keeping a large crawl's memory O(512), not O(rows).

S3 connector

The S3 connector (port 18120) crawls buckets, publishes one Kafka event per object to s3-crawl-events, and its consumer streams each object's bytes to intake as a raw POST with SHA-256 checksum headers. Objects up to 32 MB (s3.connector.checksum-max-buffer-bytes) are buffered to compute the checksum before upload; x-checksum-sha256 then arms repository-side dedupe so re-crawling unchanged bytes skips the S3 PUT. Larger objects stream without a checksum — a streamed body cannot know its digest before headers go out. Live updates are opt-in: standard bucket notifications posted to POST /api/s3/events turn s3:ObjectCreated:* into single-object crawls with source LIVE, and a two-queue snapshot crawl (a durable event buffer plus a checkpointed CSV manifest) closes the initial-crawl race window with key-level last-write-wins. An object deleted between listing and fetch settles as a delete on the spot — not an error, never retried — so a previously indexed revision does not linger.

Filesystem crawler

A filesystem crawler (file-crawler) is seeded in the connector catalog alongside S3 and JDBC, with JSON Schemas for the frontend wizards in the connector-admin resources.

Confluence

The grpc-confluence project is a full Confluence Cloud proxy: REST v2 upstream, typed gRPC downstream, with the interface taken from the Confluence API spec and designed to carry all of the same features — if the API sends it, gRPC captures it. Clients read spaces, pages, blog posts, comments, and attachments over one generated contract (ListSpaces, GetPage, GetBlogPost, ListPages, ListBlogPosts, ListAttachments; GetAttachment can inline bytes with a 25 MiB cap, following the media-CDN redirect server-side). Cursor-based incremental Sync streams deliver automatic updates, and a queryable sync ledger (grpc-sync-service, memory or SQLite) records crawl, update, and delete rows per asset, including attachments. Every completed crawl can land as an Open Knowledge Format v0.2 bundle with a sibling WARC 1.1 archive (one resource record per live web_url, a conversion record for the OKF markdown), plus a raw protobuf Kafka change stream via bundled Kafka Connect source plugins. An MCP endpoint (Streamable HTTP, official MCP Java SDK, port 8090) exposes the same surface as tools so an LLM can set up connections, configure output, and run syncs at runtime.

grpc-confluence/README.md — Confluence proxy configuration

CONFLUENCE_BASE_URL=https://example.atlassian.net/wiki
[email protected]
CONFLUENCE_API_TOKEN=...
# aliases: CONFLUENCE_USER / CONFLUENCE_TOKEN
CONFLUENCE_SPACES=ENG,DOCS          # optional allowlist
CONFLUENCE_GRPC_PORT=9095
# optional raw-bytes Kafka sink
CONFLUENCE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092

SharePoint Online (Microsoft Graph proxy)

The same repository ships grpc-microsoft-service (port 9096), a live Microsoft Graph proxy for SharePoint Online: GetMe, ListSites, ListDrives, ListChildren (streaming), GetItem, DownloadItem, and a streaming Sync with the same sync-ledger and OKF/WARC output options as Confluence. GetItem, DownloadItem, and Sync flatten SharePoint list-item columns into typed values. A third process, grpc-microsoft-connector (port 30303), is a Microsoft Copilot connector adapter: it implements Microsoft's four connector services (ConnectorInfoService, ConnectionManagementService, ConnectorCrawlerService, ConnectorOAuthService) and forwards crawls to the Graph proxy, so the same crawl can feed Copilot while Graph tokens stay on the proxy.

Related

• Engine — drains the ingress stream and owns the document's journey from there.
• Repository — where claim-check bodies and the document ledger live.
• Mapping & validation — the CEL machinery the JDBC connector's projections share with pipeline hops.
• AI-friendly output formats — the OKF and WARC output the Confluence and Graph proxies write.