adds snapshot iterator

[dylanlott]

Description

Adds the snapshot iterator to the logrepl package.

This is the first step towards adding a snapshot that works seamlessly with CDC mode.

We define a SnapshotIterator that takes a config which allows for a SnapshotName to be passed and iterates over
the rows from the configured table at that snapshot's point in time.

We necessarily secure a transaction with READ REPEATABLE from the database for our iterator to read.

Related to #15

Quick checks:

• I have followed the Code Guidelines.
• There is no other pull request for the same update/change.
• I have written unit tests.
• I have made sure that the PR is of reasonable size and can be easily reviewed.

[lovromazgon]

A couple of nitpicks, but the main things are the open transaction and the unchecked context in Next, after this we should be good 👍 The error related issues will be addressed in #27 so I skipped those.

[lovromazgon]

Nitpick: we can simplify using QueryRow, we know there will be only one row. Also name doesn't have to be a pointer, and we can use ctx as the context.

Suggested change

	rows, err := tx.Query(context.Background(), query)
	is.NoErr(err)
	var name *string
	is.True(rows.Next())
	err = rows.Scan(&name)
	is.NoErr(err)
	row := tx.QueryRow(ctx, query)
	var name string
	err = row.Scan(&name)
	is.NoErr(err)

[dylanlott]

We actually can't use QueryRow here, because it messes with our transaction when we call Scan, which handles its own connection cleanup. Will update to match the rest, though 👍

[lovromazgon]

I tried it out and it works for me 🤔

[lovromazgon]

If this indeed doesn't work we should open an issue in the driver repo because this doesn't do anything out of the ordinary, just a query within a transaction.

EDIT: I realized you are talking about the closing of the rows - that is implicit in here yes, but that doesn't mean that the connection or transaction is cleaned up.

[dylanlott]

So, I tested this before I wrote my first comment to verify that behavior still persisted, and it broke exactly as I expected it to the first time... but after you said you tested it and it worked for you, I went back and tested it again, and... it worked? 🤔 I guess I'll accept a passing test 😅

[lovromazgon]

LGTM 👍 Left two more comments, but approving already!

[lovromazgon]

Suggested change

	if rollErr := s.tx.Rollback(ctx); err != nil {
	if rollErr := s.tx.Rollback(ctx); rollErr != nil {

[lovromazgon]

Nitpick: I'd rather turn the logic around and just log the rollback error (same as we do in line 107)

Suggested change

	sdk.Logger(ctx).Err(err).Msg("load rows failed")
	return nil, fmt.Errorf("rollback failed: %w", rollErr)
	sdk.Logger(ctx).Err(rollErr).Msg("rollback failed")