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.

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 (like resolvingRecord), 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).

Generic counterpart of resolving — resolves readSchema → the caller-supplied writeSchema into an IndexedRecord, for when no reader codec / case class exists.

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.