ConfluentWire

A stripped Confluent frame: the schema id and the Avro binary body.

body is a COPY of the payload bytes, not a zero-copy offset view — Array[Byte] cannot carry an offset, and every downstream consumer takes whole arrays. The copy is one arraycopy of length - 5 bytes: noise next to any decode that follows.

Resolve a Confluent schema id to the writer Schema it names. Synchronous by design: the caller owns the registry client, the cache, and any effect wrapping, and hands this surface an already-synchronous lookup. May throw (a registry miss, a network error) — callers here catch that into AvroFailure.SchemaResolutionFailed.

Frame an Avro binary body under schemaId — the inverse of strip: strip(attach(id, body)) == Right(Framed(id, body)) for every id / body.

resolve as a composable eo dev.constructive.eo.optics.Prism over bytes: Array[Byte] (a full Confluent frame) ↔ Array[Byte] (the byte-exact gated body). Drop it BEFORE any byte optic and compose:

 val cf = ConfluentWire.confluent(schemaById, readerSchema, frameId)
 cf.andThen(codecPrism[A].field(_.x)).getOption(framedBytes)   // Option[X]
 cf.getOption(framedBytes)                                     // Option[Array[Byte]]

• getOption(framed) runs strip + resolve + fingerprint-gate; a bad frame, an unresolvable id, or a fingerprint mismatch all yield None. Use resolve when you need the specific AvroFailure instead of None.
• reverseGet(body) re-frames via attach under frameId — the schema id to publish under (typically the reader schema's registry id). Only exercised on write-back (modify / replace). A monomorphic byte Prism can't thread the original per-message id from read to write, so re-framing uses this fixed id; to preserve the incoming id, use resolve + attach by hand.

The reader fingerprint is computed once at construction, not per payload.

Gate + graft a Confluent-framed fragment into parent at into's focus, byte-exact: refuse any writer/focus drift rather than splice bytes that would misdecode. Per call:

1. strip the fragment's 5-byte header → (writerId, body);
2. resolve the writer schema for writerId via schemaById (cached per id);
3. gate by parsing-canonical-form fingerprint against into's focus schema (into.codec.schema); on equality graft the bare body into parent via AvroPrism.graftBytes (branch index re-synthesised), else refuse with AvroFailure.SchemaMismatch.

Distinct Left causes for metering: AvroFailure.SchemaMismatch (permanent structural drift) vs AvroFailure.SchemaResolutionFailed (transient — the hook threw) vs AvroFailure.NotConfluentFramed (bad fragment frame) vs the graft's own locate failure. The caller owns the fallback on a Left (decode-and-re-encode, alert, drop). To ABSORB compatible drift automatically instead of refusing it, use graftResolving.

Graft a Confluent-framed fragment into parent at into's focus, ABSORBING writer/focus drift instead of refusing it: on fingerprint match the body is spliced verbatim (zero re-encode); on drift the body is resolve-decoded writer → into's focus schema and re-encoded into that shape, so a compatible-but-not-identical fragment grafts as correct bytes rather than the silent garbage a raw byte splice would produce. Always yields valid parent bytes for a resolvable, decodable fragment — the fingerprint gate never surfaces.

The auto-translate twin of resolvingBytes. Per writer id the cache holds a bridge — identity (verbatim) on match, resolve + re-encode on drift — built once. Left only on the irreducibly-unrecoverable: AvroFailure.NotConfluentFramed (bad frame), AvroFailure.SchemaResolutionFailed (the hook threw), AvroFailure.ResolveFailed / AvroFailure.EncodeFailed (the translate itself failed), or the graft's own locate failure. Use graftGated when drift is a bug to surface and meter, not a condition to absorb.

Typed per-message Confluent reader: Array[Byte] => F[A], ready for an fs2 Stream.evalMap. Per payload: strip the header, look the writer schema up by id (schemaById, effectful), resolve-decode into A (reader shape = the codec's own schema).

Strict on the frame: a payload that does not parse as a Confluent frame raises AvroFailure.NotConfluentFramed — no silent fallback to a direct decode, which could accidentally succeed on corrupt bytes and yield garbage. A topic with mixed framed / unframed producers opts into its own fallback by catching that failure and decoding directly (AvroCodec.decodeValue[A]).

All failures are raised in F: NotConfluentFramed / ResolveFailed / DecodeFailed / BinaryParseFailed via AvroFailureException; a schemaById failure as the effect's own error.

Generic counterpart of reader — yields Array[Byte] => F[IndexedRecord] resolved into the caller-supplied writeSchema, for when no reader case class exists. Same strict frame contract as reader (opt-in fallback decode: AvroCodec.decodeRecord).

Strip + resolve + fingerprint-gate a Confluent-framed payload down to its BODY BYTES, without decoding. Per payload:

1. strip the 5-byte header → (schemaId, body);
2. resolve the writer schema for schemaId via schemaById;
3. gate by parsing-canonical-form fingerprint (org.apache.avro.SchemaNormalization.parsingFingerprint64): when the writer fingerprint equals the reader's, the body is byte-identical under both schemas — return it as-is; else refuse with AvroFailure.SchemaMismatch rather than hand back bytes that would misdecode.

The returned bytes are the caller's to decode (or hash, or forward) with whatever they own. Failures: AvroFailure.NotConfluentFramed (bad frame), AvroFailure.SchemaResolutionFailed (the hook threw), AvroFailure.SchemaMismatch (writer ≠ reader fingerprint). To ABSORB compatible drift instead of refusing it, use the translating surface below.

Read+write Confluent optic for a KNOWN writer schema (single-schema topic, or a producer's own output). to strips the header and resolve-decodes the body from readSchema into A (the write schema = AvroCodec[A].schema); from re-encodes A and re-frames under frameId. Affine-carried with the same T = Either[AvroFailure, Array[Byte]] fallible-build shape as AvroBridge. For a mixed-schema stream use reader (typed) or resolvingBytes (bytes), which look the writer schema up per message.

Framed bytes → reader-layout framed bytes for a MIXED-schema stream: per-message writer lookup (like recordReader) fused with drift translation (Avro schema resolution), handing back Array[Byte] rather than a typed A or an F[A]. Per payload: strip, resolve-decode the body writer → reader, re-encode under readerSchema, re-frame under frameId. Because the output is reader-layout, it is stable across writer-schema evolution within a reader generation — so digests, slices, and forwards computed downstream don't churn when producers upgrade. That is the property the gate (resolve) cannot give.

A factory, like confluent: the returned function closes over a per-writer-id cache (java.util.concurrent.ConcurrentHashMap), so schemaById is consulted once per DISTINCT writer id — one entry per schema version seen on the stream. Keep the returned function; re-calling resolvingBytes per message discards the cache.

Failures as Left: AvroFailure.NotConfluentFramed (bad or null frame, checked before schemaById), AvroFailure.SchemaResolutionFailed (the hook threw), AvroFailure.ResolveFailed / AvroFailure.EncodeFailed (resolve-decode / re-encode failed).

Validate and strip the 5-byte Confluent header. Fails structurally (AvroFailure.NotConfluentFramed) on a null payload (a Kafka tombstone / mis-produced record — a defined failure rather than an NPE), on inputs shorter than the header, or whose magic byte isn't 0x00.

Header length: 1 magic byte + 4 big-endian schema-id bytes.

The Confluent magic byte — always 0x00 in the current wire format.