Change Data Capture

type StreamCapable interface {
    SupportsStreaming() bool
    SupportsCDC() bool
}

type ChangeOp int

const (
    ChangeInsert  ChangeOp = iota // INSERT
    ChangeUpdate                   // UPDATE
    ChangeDelete                   // DELETE
    ChangeReplace                  // REPLACE (MongoDB-specific)
)

type ChangeEvent[T any] struct {
    // Operation is the type of change (INSERT, UPDATE, DELETE, REPLACE).
    Operation ChangeOp

    // Before is the previous state of the row.
    // nil for inserts; populated when the driver supports it.
    Before *T

    // After is the new state of the row.
    // nil for deletes.
    After *T

    // Timestamp is the server timestamp of the change.
    Timestamp time.Time

    // ResumeToken is an opaque token for resuming the stream after disconnect.
    ResumeToken any
}

type ChangeSource[T any] interface {
    Next(ctx context.Context) bool
    Event() ChangeEvent[T]
    Err() error
    Close() error
    ResumeToken() any
}

type ChangeStream[T any] struct { /* unexported fields */ }

func NewChangeStream[T any](source ChangeSource[T]) *ChangeStream[T]

cs := stream.NewChangeStream(source)
defer cs.Close()

for cs.Next(ctx) {
    event := cs.Event()

    switch event.Operation {
    case stream.ChangeInsert:
        fmt.Printf("New row: %+v\n", *event.After)
    case stream.ChangeUpdate:
        fmt.Printf("Updated row: %+v\n", *event.After)
    case stream.ChangeDelete:
        if event.Before != nil {
            fmt.Printf("Deleted row: %+v\n", *event.Before)
        }
    }
}
if err := cs.Err(); err != nil {
    log.Error("change stream error", "err", err)
}

for event, err := range cs.All {
    if err != nil {
        log.Error("change stream error", "err", err)
        break
    }

    if event.Operation == stream.ChangeInsert {
        processNewRecord(event.After)
    }
}

// Save the resume token periodically
token := cs.ResumeToken()
saveCheckpoint(token)

// Later, resume from the saved token
source := driver.WatchCollection("users", WithResumeToken(savedToken))
cs := stream.NewChangeStream(source)

// Create a listener from the pgdriver
pgdb := pgdriver.Unwrap(db)
listener := pgdb.NewListener()

// Start listening (acquires a dedicated connection)
err := listener.Start(ctx)

// Subscribe to a channel
err = listener.Listen(ctx, "user_changes")

// Register a handler
listener.OnNotification("user_changes", func(n *pgdriver.Notification) {
    fmt.Printf("Channel: %s, Payload: %s\n", n.Channel, n.Payload)
})

pgdb := pgdriver.Unwrap(db)

listener, err := pgdb.Listen(ctx, "order_events", func(n *pgdriver.Notification) {
    fmt.Printf("Order event: %s\n", n.Payload)
})
if err != nil {
    return err
}
defer listener.Close()

err := listener.Notify(ctx, "user_changes", `{"op":"INSERT","id":42}`)

type Notification struct {
    Channel string // notification channel name
    Payload string // message payload (typically JSON)
    PID     uint32 // backend PID of the sender
}

-- Create a notification function
CREATE OR REPLACE FUNCTION notify_changes()
RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(
        TG_TABLE_NAME || '_changes',
        json_build_object(
            'op', TG_OP,
            'before', CASE WHEN TG_OP = 'DELETE' THEN row_to_json(OLD) ELSE NULL END,
            'after', CASE WHEN TG_OP = 'DELETE' THEN NULL ELSE row_to_json(NEW) END,
            'timestamp', now()
        )::text
    );
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

-- Attach to a table
CREATE TRIGGER users_notify
    AFTER INSERT OR UPDATE OR DELETE ON users
    FOR EACH ROW EXECUTE FUNCTION notify_changes();

pgdb := pgdriver.Unwrap(db)

listener, err := pgdb.Listen(ctx, "users_changes", func(n *pgdriver.Notification) {
    var event struct {
        Op        string          `json:"op"`
        Before    json.RawMessage `json:"before"`
        After     json.RawMessage `json:"after"`
        Timestamp time.Time       `json:"timestamp"`
    }
    if err := json.Unmarshal([]byte(n.Payload), &event); err != nil {
        log.Error("failed to parse notification", "err", err)
        return
    }

    switch event.Op {
    case "INSERT":
        var user User
        json.Unmarshal(event.After, &user)
        fmt.Printf("New user: %s\n", user.Email)
    case "UPDATE":
        var user User
        json.Unmarshal(event.After, &user)
        fmt.Printf("Updated user: %s\n", user.Email)
    case "DELETE":
        var user User
        json.Unmarshal(event.Before, &user)
        fmt.Printf("Deleted user: %s\n", user.Email)
    }
})
if err != nil {
    return err
}
defer listener.Close()

// Block until context is cancelled
<-ctx.Done()