fix(engine): update ReplicationLag format to be passed as integer

Author: denis256

engine/internal/retrieval/status/retrieval_status.go

@@ -97,30 +97,26 @@ func FetchSyncMetrics(ctx context.Context, config *global.Config, socketPath str
 		return &sync, fmt.Errorf("failed to read Postgres version %w", err)
 	}
 
-	var replicationLag string
-
 	var query = lag9xQuery
 
 	if pgVersion >= pgVersion10 {
 		query = lagQuery
 	}
 
-	replicationLag, err = lag(ctx, conn, query)
+	replicationLag, err := lag(ctx, conn, query)
 	if err != nil {
 		log.Warn("Failed to fetch replication lag", err)
 	} else {
 		sync.ReplicationLag = replicationLag
 	}
 
-	var replayedLsn, lastReplayedLsnAt string
-
 	query = lastReplayedLsn9xQuery
 
 	if pgVersion >= pgVersion10 {
 		query = lastReplayedLsnQuery
 	}
 
-	lastReplayedLsnAt, replayedLsn, err = lastReplayedLsn(ctx, conn, query)
+	lastReplayedLsnAt, replayedLsn, err := lastReplayedLsn(ctx, conn, query)
 	if err != nil {
 		log.Warn("Failed to fetch last replayed lsn", err)
 	} else {
@@ -166,16 +162,16 @@ func version(ctx context.Context, conn *pgx.Conn) (int, error) {
 	return pgVersion, nil
 }
 
-func lag(ctx context.Context, conn *pgx.Conn, query string) (string, error) {
-	var lagSec sql.NullString
+func lag(ctx context.Context, conn *pgx.Conn, query string) (int, error) {
+	var lagSec int
 
 	row := conn.QueryRow(ctx, query)
 
 	if err := row.Scan(&lagSec); err != nil {
-		return "", fmt.Errorf("failed to read replication lag: %w", err)
+		return 0, fmt.Errorf("failed to read replication lag: %w", err)
 	}
 
-	return lagSec.String, nil
+	return lagSec, nil
 }
 
 func lastReplayedLsn(ctx context.Context, conn *pgx.Conn, query string) (string, string, error) {

engine/pkg/models/sync.go

@@ -6,6 +6,6 @@ type Sync struct {
 	StartedAt         string `json:"startedAt,omitempty"`
 	LastReplayedLsn   string `json:"lastReplayedLsn"`
 	LastReplayedLsnAt string `json:"lastReplayedLsnAt"`
-	ReplicationLag    string `json:"replicationLag"`
+	ReplicationLag    int    `json:"replicationLag"`
 	ReplicationUptime int    `json:"replicationUptime"`
 }