expose sync status

errors.go

@@ -0,0 +1,15 @@
+package sqlitestream
+
+import "errors"
+
+// ErrNotReplicating is returned when an operation requires a configured
+// replica but replication is off.
+var ErrNotReplicating = errors.New("sqlitestream: replication not configured")
+
+// ErrExists is returned by Restore when the output path already exists
+// and force is false.
+var ErrExists = errors.New("sqlitestream: file exists")
+
+// ErrReplicaEmpty is returned by Restore when the replica holds no
+// snapshots. Callers can treat this as a fresh install.
+var ErrReplicaEmpty = errors.New("sqlitestream: replica is empty — nothing to restore")

restore.go

@@ -17,21 +17,21 @@ import (
 func Restore(ctx context.Context, outputPath string, force bool, opts ...Option) error {
 	cfg := apply(opts)
 	if cfg.replicaURL == "" {
-		return errors.New("WithReplicaURL is required for restore")
+		return fmt.Errorf("%w: WithReplicaURL is required for restore", ErrNotReplicating)
 	}
 
 	switch _, err := os.Stat(outputPath); {
 	case err == nil:
 		if !force {
-			return fmt.Errorf("%s exists (pass force=true to overwrite)", outputPath)
+			return fmt.Errorf("%w: %s (pass force=true to overwrite)", ErrExists, outputPath)
 		}
 		if err := removeDBFiles(outputPath); err != nil {
 			return fmt.Errorf("clear destination: %w", err)
 		}
 	case errors.Is(err, os.ErrNotExist):
 		// ok — nothing to clear
 	default:
-		return fmt.Errorf("unexpected error openning path: %s (%w)", outputPath, err)
+		return fmt.Errorf("unexpected error opening path: %s (%w)", outputPath, err)
 	}
 
 	db, err := newLitestreamDB(outputPath, cfg)
@@ -41,8 +41,10 @@ func Restore(ctx context.Context, outputPath string, force bool, opts ...Option)
 	opt := litestream.NewRestoreOptions()
 	opt.OutputPath = outputPath
 	if err := db.Replica.Restore(ctx, opt); err != nil {
-		if errors.Is(err, litestream.ErrNoSnapshots) {
-			return errors.New("replica is empty — nothing to restore")
+		// litestream reports an empty replica as either error, depending
+		// on the client.
+		if errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable) {
+			return ErrReplicaEmpty
 		}
 		return fmt.Errorf("restore: %w", err)
 	}

sqlitestream.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net/url"
+	"time"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/s3" // also registers the s3:// scheme
@@ -64,6 +65,37 @@ func (d *DB) DB() *sql.DB {
 	return d.sql
 }
 
+// Replicating reports whether background replication is configured.
+func (d *DB) Replicating() bool {
+	return d != nil && d.ldb != nil
+}
+
+// Retention reports whether litestream is handling retention.
+func (d *DB) Retention() bool {
+	return d != nil && d.ldb != nil && d.ldb.RetentionEnabled
+}
+
+// LastSyncedAt returns the time of the last successful sync to the
+// replica. The zero time means replication is off or nothing has been
+// synced yet.
+func (d *DB) LastSyncedAt() time.Time {
+	if d == nil || d.ldb == nil {
+		return time.Time{}
+	}
+	return d.ldb.LastSuccessfulSyncAt()
+}
+
+// SyncStatus reports the status between the local database and the
+// configured (cloud) storage by comparing transaction records. This
+// may entail I/O. Returns ErrNotReplicating when no replica is
+// configured.
+func (d *DB) SyncStatus(ctx context.Context) (litestream.SyncStatus, error) {
+	if d == nil || d.ldb == nil {
+		return litestream.SyncStatus{}, ErrNotReplicating
+	}
+	return d.ldb.SyncStatus(ctx)
+}
+
 // Close flushes any pending WAL to S3 (when replication is on), stops the
 // replicator, and closes the SQL handle.
 //

sqlitestream_test.go

@@ -4,7 +4,9 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
+	_ "github.com/benbjohnson/litestream/file" // registers the file:// scheme for tests
 	"github.com/luzilla/sqlitestream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -41,10 +43,53 @@ func TestSuite(t *testing.T) {
 		assert.NoError(t, db.Close(t.Context()))
 	})
 
+	// Sync status accessors are safe on a nil receiver and without a replica.
+	t.Run("SyncStatusNoReplica", func(t *testing.T) {
+		var nilDB *sqlitestream.DB
+		assert.False(t, nilDB.Replicating())
+		assert.False(t, nilDB.Retention())
+		assert.True(t, nilDB.LastSyncedAt().IsZero())
+		_, err := nilDB.SyncStatus(t.Context())
+		assert.ErrorIs(t, err, sqlitestream.ErrNotReplicating)
+
+		db, err := sqlitestream.Open(t.Context(), filepath.Join(t.TempDir(), "test.db"))
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = db.Close(t.Context()) })
+
+		assert.False(t, db.Replicating())
+		assert.False(t, db.Retention())
+		assert.True(t, db.LastSyncedAt().IsZero())
+		_, err = db.SyncStatus(t.Context())
+		assert.ErrorIs(t, err, sqlitestream.ErrNotReplicating)
+	})
+
+	// With a replica configured, LastSyncedAt reflects the background sync.
+	t.Run("SyncStatusWithReplica", func(t *testing.T) {
+		replicaURL := "file://" + filepath.Join(t.TempDir(), "replica")
+		db, err := sqlitestream.Open(t.Context(), filepath.Join(t.TempDir(), "test.db"),
+			sqlitestream.WithReplicaURL(replicaURL))
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = db.Close(t.Context()) })
+
+		assert.True(t, db.Replicating())
+		assert.True(t, db.Retention()) // litestream default
+
+		_, err = db.DB().Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)`)
+		require.NoError(t, err)
+
+		require.Eventually(t, func() bool {
+			return !db.LastSyncedAt().IsZero()
+		}, 10*time.Second, 100*time.Millisecond, "no sync recorded")
+		assert.WithinDuration(t, time.Now(), db.LastSyncedAt(), time.Minute)
+
+		_, err = db.SyncStatus(t.Context())
+		assert.NoError(t, err)
+	})
+
 	t.Run("restore", func(t *testing.T) {
 		t.Run("RequiresReplicaURL", func(t *testing.T) {
 			err := sqlitestream.Restore(t.Context(), filepath.Join(t.TempDir(), "out.db"), false)
-			require.Error(t, err)
+			require.ErrorIs(t, err, sqlitestream.ErrNotReplicating)
 			assert.Contains(t, err.Error(), "WithReplicaURL")
 		})
 
@@ -56,9 +101,13 @@ func TestSuite(t *testing.T) {
 
 			err := sqlitestream.Restore(t.Context(), path, false,
 				sqlitestream.WithReplicaURL("file:///nonexistent"))
-			require.Error(t, err)
-			assert.Contains(t, err.Error(), "exists")
+			require.ErrorIs(t, err, sqlitestream.ErrExists)
 		})
 
+		t.Run("EmptyReplica", func(t *testing.T) {
+			err := sqlitestream.Restore(t.Context(), filepath.Join(t.TempDir(), "out.db"), false,
+				sqlitestream.WithReplicaURL("file://"+t.TempDir()))
+			require.ErrorIs(t, err, sqlitestream.ErrReplicaEmpty)
+		})
 	})
 }