Example CDC usage in Go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/neo4j/neo4j-go-driver/v5/neo4j"
	"github.com/spf13/cobra"
	"github.com/tidwall/pretty"
)

func applyChange(record *neo4j.Record) error { (1)
	jsonOutput, err := json.Marshal(asMap(record))
	if err != nil {
		return fmt.Errorf("unable to jsonify record: %w", err)
	}

	fmt.Println(string(pretty.Color(pretty.Pretty(jsonOutput), pretty.TerminalStyle)))

	return nil
}

func queryChangeID(ctx context.Context, driver neo4j.DriverWithContext, database string, query string) (string, error) {
	result, err := neo4j.ExecuteQuery(ctx, driver, query, nil, neo4j.EagerResultTransformer, neo4j.ExecuteQueryWithDatabase(database), neo4j.ExecuteQueryWithReadersRouting())
	if err != nil {
		return "", fmt.Errorf("unable to query change identifier: %w", err)
	}

	if len(result.Records) != 1 {
		return "", fmt.Errorf("expected one record, but got %d", len(result.Records))
	}

	id, _, err := neo4j.GetRecordValue[string](result.Records[0], "id")
	if err != nil {
		return "", fmt.Errorf("unable to extract id: %w", err)
	}

	return id, nil
}

func asMap(record *neo4j.Record) map[string]any {
	result := make(map[string]any, len(record.Keys))

	for i := 0; i < len(record.Keys); i++ {
		result[record.Keys[i]] = record.Values[i]
	}

	return result
}

type CDCService struct {
	driver    neo4j.DriverWithContext
	database  string
	waitGroup sync.WaitGroup
	cursor    atomic.Pointer[string]
	selectors []any
}

func (s *CDCService) queryChanges(ctx context.Context) error { (2)
	session := s.driver.NewSession(ctx, neo4j.SessionConfig{DatabaseName: s.database})

	current, err := s.currentChangeID(ctx)
	if err != nil {
		return err
	}
	_, err = session.ExecuteRead(ctx, func(tx neo4j.ManagedTransaction) (any, error) {
		result, err := tx.Run(ctx, "CALL cdc.query($from, $selectors)", map[string]any{
			"from":      s.from(),
			"selectors": s.selectors,
		})
		if err != nil {
			return "", err
		}

		var record *neo4j.Record
		if !result.Peek(ctx) {
			s.setFrom(current) (3)
		} else {
			for result.NextRecord(ctx, &record) {
				err := applyChange(record) (4)
				if err != nil {
					return "", fmt.Errorf("error processing record: %w", err)
				}

				id, isNil, err := neo4j.GetRecordValue[string](record, "id")
				if err != nil || isNil {
					return "", fmt.Errorf("missing or invalid id value returned")
				}
				s.setFrom(id) (5)
			}
		}

		return s.from(), nil
	})
	if err != nil {
		return fmt.Errorf("unable to query/process changes: %w", err)
	}

	return nil
}

func (s *CDCService) earliestChangeID(ctx context.Context) (string, error) { (6)
	return queryChangeID(ctx, s.driver, s.database, "CALL cdc.earliest()")
}

func (s *CDCService) currentChangeID(ctx context.Context) (string, error) { (7)
	return queryChangeID(ctx, s.driver, s.database, "CALL cdc.current()")
}

func (s *CDCService) from() string {
	return *s.cursor.Load()
}

func (s *CDCService) setFrom(from string) {
	s.cursor.Store(&from)
}

func (s *CDCService) Start(ctx context.Context) error {
	if s.from() == "" {
		current, err := s.currentChangeID(ctx)
		if err != nil {
			return err
		}
		s.setFrom(current)
	}

	s.waitGroup.Add(1)
	go func(ctx context.Context) {
		defer func() {
			s.waitGroup.Done()
		}()

		timer := time.NewTimer(0 * time.Millisecond)
		for {
			select {
			case <-ctx.Done():
				return
			case <-timer.C:
				{
					err := s.queryChanges(ctx)
					if err != nil {
						log.Printf("error querying/processing changes: %v", err)
						return
					}

					timer.Reset(500 * time.Millisecond) (8)
				}
			}
		}
	}(ctx)

	return nil
}

func (s *CDCService) WaitForExit() {
	s.waitGroup.Wait()
}

func NewCDCService(uri string, username string, password string, database string, from string, selectors []any) (*CDCService, error) {
	driver, err := neo4j.NewDriverWithContext(uri, neo4j.BasicAuth(username, password, ""))
	if err != nil {
		return nil, fmt.Errorf("unable to create driver: %w", err)
	}

	cdc := &CDCService{
		driver:    driver,
		database:  database,
		waitGroup: sync.WaitGroup{},
		cursor:    atomic.Pointer[string]{},
		selectors: selectors,
	}
	cdc.setFrom(from)

	return cdc, nil
}

var (
	address  string
	database string
	username string
	password string
	from     string
)

func main() {
	rootCmd := &cobra.Command{
		Run: func(cmd *cobra.Command, args []string) {
			ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt)

			selectors := []any{
				//map[string]any{"select": "n", "labels": []string{"Person", "Employee"}}, (9)
			}

			cdc, err := NewCDCService(address, username, password, database, from, selectors)
			if err != nil {
				log.Fatal(err)
			}

			if err := cdc.Start(ctx); err != nil {
				log.Fatal(err)
			}

			fmt.Printf("starting...\n")
			cdc.WaitForExit()
			fmt.Printf("quitting...\n")
		},
	}

	rootCmd.Flags().StringVarP(&address, "address", "a", "bolt://localhost:7687", "Bolt URI")
	rootCmd.Flags().StringVarP(&database, "database", "d", "", "Database")
	rootCmd.Flags().StringVarP(&username, "username", "u", "neo4j", "Username")
	rootCmd.Flags().StringVarP(&password, "password", "p", "passw0rd", "Password")
	rootCmd.Flags().StringVarP(&from, "from", "f", "", "Change identifier to query changes from")

	cobra.CheckErr(rootCmd.Execute())
}