Build connectors that fit your stack.

# Add the SDK
go get zipline.run/sdk-go

# The reference connectors live in the same tree -
# copy connectors/postgres or connectors/mssql to start.
zipline connector list

package zip

// Every connector implements Init + Stop.
type Source interface {
    Init(ctx context.Context, cfg Config) error
    Stop(ctx context.Context) error
}

type Sink interface {
    Init(ctx context.Context, cfg Config) error
    Write(ctx context.Context, batch []Event) error // all-or-nothing, durable on return
    Stop(ctx context.Context) error
}

// A Source implements exactly one read style - PullReader is the common one.
type PullReader interface {
    Start(ctx context.Context, resume Position, p StartParams) error
    Read(ctx context.Context) (Batch, error)
}

// Optional capabilities - implement only what you need; the host detects them.
type Acker           interface { Ack(ctx context.Context, durable Position) error }
type Pauser          interface { Pause(context.Context) error; Resume(context.Context) error }
type SchemaDescriber interface { DescribeSchemas(context.Context) ([]Schema, error) }
type MetricsReporter interface { Metrics(context.Context) MetricSnapshot }

type Event struct {
    Stream     string    // relation name, e.g. "public.orders"
    Key        []byte    // entity identity (partition / compaction)
    Op         Op        // Insert | Update | Delete | Snapshot | SchemaChange | Truncate
    Before     *Record   // nil for inserts
    After      *Record   // nil for deletes
    Position   Position  // opaque, per-stream monotonic, bytewise-comparable
    OccurredAt time.Time
    // … SchemaRef, CapturedAt, Transaction
}

type Batch struct {
    Events       []Event
    SafePosition Position // safe-resume low-water at batch end
}

package redis

import (
    "context"
    "time"

    "github.com/redis/go-redis/v9"
    zip "zipline.run/sdk-go"
)

type redisConfig struct {
    Addr     string `json:"addr"`
    Stream   string `json:"stream"`
    Group    string `json:"group"`
    Password string `json:"password" zip:"secret"`
}

type Source struct {
    cfg redisConfig
    rdb *redis.Client
}

// Compile-time proof we satisfy the contract + the capabilities the host wires.
var (
    _ zip.Source     = (*Source)(nil)
    _ zip.PullReader = (*Source)(nil)
    _ zip.Acker      = (*Source)(nil)
)

func (s *Source) Init(ctx context.Context, cfg zip.Config) error {
    if err := cfg.Map(&s.cfg); err != nil {
        return err
    }
    s.rdb = redis.NewClient(&redis.Options{Addr: s.cfg.Addr, Password: s.cfg.Password})
    return s.rdb.XGroupCreateMkStream(ctx, s.cfg.Stream, s.cfg.Group, "0").Err()
}

// Start is a no-op here: XReadGroup with ">" resumes from the group cursor.
func (s *Source) Start(_ context.Context, _ zip.Position, _ zip.StartParams) error { return nil }

func (s *Source) Read(ctx context.Context) (zip.Batch, error) {
    res, err := s.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group:    s.cfg.Group,
        Consumer: "zipline",
        Streams:  []string{s.cfg.Stream, ">"},
        Count:    256,
        Block:    2 * time.Second,
    }).Result()
    if err == redis.Nil { return zip.Batch{}, nil } // idle poll: nothing new
    if err != nil { return zip.Batch{}, err }

    var b zip.Batch
    for _, msg := range res[0].Messages {
        rec := zip.NewRecord(msg.Values)              // map[string]any → *Record
        pos := zip.BytesPosition([]byte(msg.ID)) // stream ids sort bytewise
        b.Add(s.cfg.Stream, zip.OpInsert, rec, pos)
    }
    return b, nil
}

func (s *Source) Ack(ctx context.Context, durable zip.Position) error {
    return s.rdb.XAck(ctx, s.cfg.Stream, s.cfg.Group, string(durable)).Err()
}

func (s *Source) Stop(_ context.Context) error { return s.rdb.Close() }

func init() {
    zip.Register[redisConfig](zip.ConnectorDescriptor{
        Type:      "redis",
        Direction: zip.DirectionSource,
        Version:   "0.1.0",
    }, zip.Factory{NewSource: func() zip.Source { return &Source{} }})
}

package clickhouse

import (
    "context"
    "encoding/json"

    "github.com/ClickHouse/clickhouse-go/v2"
    "github.com/ClickHouse/clickhouse-go/v2/lib/driver"
    zip "zipline.run/sdk-go"
)

type chConfig struct {
    Addr     string `json:"addr"`
    Database string `json:"database"`
    Table    string `json:"table"`
    Username string `json:"username"`
    Password string `json:"password" zip:"secret"`
}

type Sink struct {
    cfg  chConfig
    conn driver.Conn
}

var (
    _ zip.Sink           = (*Sink)(nil)
    _ zip.IdempotentSink = (*Sink)(nil)
)

func (s *Sink) Init(ctx context.Context, cfg zip.Config) error {
    if err := cfg.Map(&s.cfg); err != nil {
        return err
    }
    conn, err := clickhouse.Open(&clickhouse.Options{
        Addr: []string{s.cfg.Addr},
        Auth: clickhouse.Auth{Database: s.cfg.Database, Username: s.cfg.Username, Password: s.cfg.Password},
    })
    if err != nil { return err }
    s.conn = conn
    return conn.Ping(ctx)
}

// Write applies the whole batch in one INSERT - all-or-nothing, durable on return.
func (s *Sink) Write(ctx context.Context, batch []zip.Event) error {
    bw, err := s.conn.PrepareBatch(ctx,
        "INSERT INTO "+s.cfg.Table+" (stream, op, key, after, ts)")
    if err != nil { return err }

    for _, ev := range batch {
        after := []byte("{}")
        if ev.After != nil {
            after, _ = json.Marshal(ev.After.Map()) // Record → map[string]any
        }
        if err := bw.Append(ev.Stream, ev.Op.String(), string(ev.Key),
            string(after), ev.OccurredAt); err != nil {
            return err // abort the whole batch → full replay
        }
    }
    return bw.Send() // ReplacingMergeTree collapses replays
}

func (s *Sink) Idempotent() bool { return true }

func (s *Sink) Stop(_ context.Context) error { return s.conn.Close() }

func init() {
    zip.Register[chConfig](zip.ConnectorDescriptor{
        Type:      "clickhouse",
        Direction: zip.DirectionSink,
        Version:   "0.1.0",
    }, zip.Factory{NewSink: func() zip.Sink { return &Sink{} }})
}

import (
    "context"
    "errors"
    "fmt"

    zip "zipline.run/sdk-go"
)

func (s *Sink) Write(ctx context.Context, batch []zip.Event) error {
    if err := s.apply(ctx, batch); err != nil {
        switch {
        case errors.Is(ctx.Err(), context.Canceled):
            return ctx.Err()                          // clean stop - checkpoint preserved
        case errors.Is(err, errBadSchema):
            return fmt.Errorf("clickhouse: %v: %w", err, zip.ErrPermanent) // dead-letter
        default:
            return err                                // transient - host retries with backoff
        }
    }
    return nil
}

func TestSource_ReadsEntries(t *testing.T) {
    src := &Source{}
    cfg := zip.ConfigFromJSON([]byte(`{"addr":"localhost:6379","stream":"orders","group":"zip"}`))

    if err := src.Init(t.Context(), cfg); err != nil {
        t.Fatal(err)
    }
    defer src.Stop(t.Context())

    if err := src.Start(t.Context(), nil, zip.StartParams{Mode: zip.ModeStream}); err != nil {
        t.Fatal(err)
    }
    batch, err := src.Read(t.Context())
    if err != nil { t.Fatal(err) }

    for _, ev := range batch.Events {
        amount, _ := ev.After.Int("amount")
        t.Logf("%s %s amount=%d", ev.Op, ev.Stream, amount)
    }
}