docs: godoc and package-doc pass

doc.go

@@ -0,0 +1,133 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package sdk is the Go SDK for building Conduit processors. A processor
+// receives records flowing through a pipeline, transforms them, and returns the
+// result. Authors implement the [Processor] interface (or adapt a function with
+// [NewProcessorFunc]) and the SDK handles the rest: configuration parsing,
+// schema encode/decode middleware, and the plumbing that connects the processor
+// to the Conduit engine.
+//
+// # Standalone vs. built-in processors
+//
+// The same [Processor] implementation can run in two ways:
+//
+//   - Standalone: the processor is compiled to a WebAssembly module
+//     (GOOS=wasip1 GOARCH=wasm) and executed by Conduit in a wazero runtime.
+//     The module's main function calls [Run], which becomes the entry point.
+//     This is the default and recommended mode — it isolates the processor from
+//     the engine and lets it be distributed as a single portable binary.
+//   - Built-in: the processor is compiled natively into a custom Conduit
+//     build. This avoids the WebAssembly boundary (and its per-call
+//     serialization cost) at the price of coupling the processor to the engine
+//     binary. Built-in processors do not call [Run]; the engine invokes the
+//     [Processor] methods directly.
+//
+// Author code is identical in both modes. Only the entry point and build
+// constraints differ, so write to the [Processor] contract and let the build
+// target decide how the processor is hosted.
+//
+// # Implementing a Processor
+//
+// Embed [UnimplementedProcessor] in your type. It provides no-op implementations
+// of the optional methods and satisfies the unexported marker method that keeps
+// the interface closed, so adding a method to [Processor] in a later release is
+// not a breaking change for existing processors:
+//
+//	type myProcessor struct {
+//		sdk.UnimplementedProcessor
+//		cfg myConfig
+//	}
+//
+//	func (p *myProcessor) Specification() (sdk.Specification, error) { ... }
+//	func (p *myProcessor) Configure(ctx context.Context, cfg config.Config) error { ... }
+//	func (p *myProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord { ... }
+//
+// # Lifecycle
+//
+// The runtime calls a processor's methods in a fixed order, and (for a single
+// processor instance) never concurrently — the standalone command loop in [Run]
+// processes one command at a time. A processor therefore does not need to guard
+// its own fields against concurrent access by the SDK, but it must not assume
+// any parallelism either.
+//
+//  1. Specification — called to discover the processor's name, version, and
+//     configuration parameters. Must be side-effect free; it may be called
+//     before Configure and without any configuration.
+//  2. Configure — called once with the user's configuration. Validate and store
+//     it here. Do not open connections or start background work; that is Open's
+//     job. See [ParseConfig] for turning the raw config map into a typed struct.
+//  3. Open — called once after Configure. Acquire resources and start any
+//     background work here.
+//  4. Process — called repeatedly, once per incoming batch, until shutdown. See
+//     the record-handling contract below.
+//  5. Teardown — called once when the pipeline is shutting down. No other method
+//     is called after Teardown returns; the processor is then discarded. Release
+//     everything Open acquired.
+//
+// Process may be called more than once with the same records (for example after
+// a restart when records were not flushed downstream), so processing must be
+// idempotent.
+//
+// # Record handling and error propagation
+//
+// Process receives a batch of [opencdc.Record] values and returns a
+// [ProcessedRecord] for each. The returned slice is positional: the result at
+// index i is the outcome of the input record at index i. Each input record may
+// carry raw or structured data in its key and payload; a processor that reads
+// structured fields should enable the schema-decode middleware (see below) or
+// handle both shapes.
+//
+// The concrete [ProcessedRecord] type an author returns decides how the record
+// continues through the pipeline:
+//
+//   - [SingleRecord] — the transformed record continues downstream. This is the
+//     common case.
+//   - [MultiRecord] — the record is split into zero or more records. Returning
+//     an empty MultiRecord is equivalent to [FilterRecord]; returning one record
+//     is equivalent to [SingleRecord].
+//   - [FilterRecord] — the record is acknowledged and dropped from the pipeline.
+//     Use this to intentionally discard records; it is not an error.
+//   - [ErrorRecord] — processing failed. The record is nacked and handled
+//     according to the pipeline's error policy (for example routed to a dead-
+//     letter queue or halting the pipeline). Returning an ErrorRecord is the
+//     only way to signal a per-record failure — a processor must not drop a
+//     record it could not process, or at-least-once delivery is violated.
+//
+// Because filtering, splitting, and failing are all expressed through the
+// return value rather than through the process's exit or a returned error,
+// Process itself does not return an error: a batch always produces a result for
+// every record it accounts for.
+//
+// # Middleware
+//
+// [DefaultProcessorMiddleware] wraps every processor run through [Run] with
+// schema decode and encode middleware. Decode middleware fetches the schema
+// referenced in a record's metadata and turns raw key/payload bytes into
+// [opencdc.StructuredData] before Process sees them; encode middleware reverses
+// that afterwards, so a processor can operate on structured data without dealing
+// with schema resolution. A processor tunes this via [Processor.MiddlewareOptions]
+// (see [ProcessorWithSchemaDecodeConfig] and [ProcessorWithSchemaEncodeConfig]).
+//
+// # Neighboring packages
+//
+//   - github.com/conduitio/conduit-commons/opencdc defines the Record type that
+//     flows through Process.
+//   - github.com/conduitio/conduit-commons/config defines the configuration and
+//     parameter types used in Specification and Configure.
+//   - The schema subpackage is the author-facing API for creating and fetching
+//     schemas from within a processor.
+//   - The wasm and pprocutils subpackages are engine plumbing and are not meant
+//     to be imported by processor authors.
+package sdk

errors.go

@@ -23,5 +23,9 @@ var (
 		"this action, please check the source code of the processor and make sure " +
 		"all required processor methods are implemented")
 
+	// ErrFilterRecord is a sentinel error a function passed to
+	// [NewProcessorFunc] can return to filter a record out of the pipeline
+	// instead of failing it. The record is acked and dropped ([FilterRecord]),
+	// not nacked. Returning any other error yields an [ErrorRecord].
 	ErrFilterRecord = errors.New("filter out this record")
 )

processor_func.go

@@ -47,6 +47,12 @@ func NewProcessorFunc(specs Specification, f func(context.Context, opencdc.Recor
 
 func (f ProcessorFunc) Specification() (Specification, error) { return f.specs, nil }
 
+// Process applies the wrapped function to each record in order. A record for
+// which the function returns [ErrFilterRecord] becomes a [FilterRecord]; any
+// other error becomes an [ErrorRecord] and processing stops there, so the
+// returned slice is truncated at the first failing record and is shorter than
+// the input. Records before the error are returned as [SingleRecord] values;
+// records after it are left for Conduit to reprocess.
 func (f ProcessorFunc) Process(ctx context.Context, records []opencdc.Record) []ProcessedRecord {
 	outRecs := make([]ProcessedRecord, len(records))
 	for i, inRec := range records {

schema/doc.go

@@ -0,0 +1,35 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package schema is the processor-facing API for creating and fetching schemas.
+// Processors use it to register the schema of records they emit and to look up
+// the schema of records they receive, keeping schema identity consistent across
+// a pipeline.
+//
+// The package exposes two entry points, [Get] and [Create], both backed by the
+// package-level [SchemaService]. The service a processor talks to depends on how
+// it is hosted:
+//
+//   - Standalone (WebAssembly): the engine replaces [SchemaService] at startup
+//     with an implementation that forwards calls to Conduit's schema registry
+//     over the host boundary, so schemas are shared with the rest of the
+//     pipeline.
+//   - Built-in / tests: the default [SchemaService] is an in-process,
+//     [NewInMemoryService]-backed store wrapped in a cache. It has no
+//     persistence and is not shared with a real registry — useful for unit
+//     tests, not for cross-processor schema sharing.
+//
+// Get and Create results are cached, so repeated lookups of the same schema do
+// not cross the host boundary again.
+package schema

schema/in_memory.go

@@ -23,6 +23,11 @@ import (
 	"github.com/conduitio/conduit-processor-sdk/pprocutils"
 )
 
+// InMemoryService is a non-persistent [pprocutils.SchemaService] that keeps all
+// schemas in memory. It is the default backing store for tests and built-in
+// processors; it is safe for concurrent use but its contents are lost when the
+// process exits and are not shared with a real schema registry. Versions per
+// subject start at 1 and increment on each [InMemoryService.CreateSchema] call.
 type InMemoryService struct {
 	// schemas is a map of schema subjects to all the versions of that schema
 	// versioning starts at 1, newer versions are appended to the end of the versions slice.
@@ -33,6 +38,7 @@ type InMemoryService struct {
 	idSequence int
 }
 
+// NewInMemoryService returns an empty [InMemoryService] ready for use.
 func NewInMemoryService() *InMemoryService {
 	return &InMemoryService{
 		schemas: make(map[string][]schema.Schema),

schema/schema.go

@@ -22,16 +22,31 @@ import (
 	"github.com/conduitio/conduit-processor-sdk/pprocutils"
 )
 
+// TypeAvro is the Avro schema type. It is currently the only type accepted by
+// [Create].
 const TypeAvro = schema.TypeAvro
 
+// SchemaService is the service backing [Get] and [Create]. In a standalone
+// (WebAssembly) processor the engine overwrites this at startup with a client
+// for Conduit's schema registry; otherwise it defaults to an in-process,
+// cache-wrapped in-memory store. Replace it in tests to stub schema resolution.
 var SchemaService pprocutils.SchemaService = newCachedSchemaService(NewInMemoryService())
 
 var (
+	// ErrSubjectNotFound is returned by [Get] when no schema exists for the
+	// requested subject.
 	ErrSubjectNotFound = pprocutils.ErrSubjectNotFound
+	// ErrVersionNotFound is returned by [Get] when the subject exists but the
+	// requested version does not.
 	ErrVersionNotFound = pprocutils.ErrVersionNotFound
-	ErrInvalidSchema   = pprocutils.ErrInvalidSchema
+	// ErrInvalidSchema is returned by [Create] when the supplied bytes are not a
+	// valid schema of the requested type.
+	ErrInvalidSchema = pprocutils.ErrInvalidSchema
 )
 
+// Get fetches the schema registered under the given subject and version.
+// Versions start at 1. It returns [ErrSubjectNotFound] or [ErrVersionNotFound]
+// (wrapped) if the schema is not registered.
 func Get(ctx context.Context, subject string, version int) (schema.Schema, error) {
 	resp, err := SchemaService.GetSchema(ctx, pprocutils.GetSchemaRequest{
 		Subject: subject,
@@ -43,6 +58,10 @@ func Get(ctx context.Context, subject string, version int) (schema.Schema, error
 	return resp.Schema, nil
 }
 
+// Create registers a new version of the schema for the given subject and
+// returns it with its assigned ID and version. Each call to Create appends a new
+// version; there is no deduplication of identical bytes. Only [TypeAvro] is
+// currently accepted — other types return [ErrInvalidSchema].
 func Create(ctx context.Context, typ schema.Type, subject string, bytes []byte) (schema.Schema, error) {
 	resp, err := SchemaService.CreateSchema(ctx, pprocutils.CreateSchemaRequest{
 		Subject: subject,