postgres

func (*PostgresSource) Ack ¶

func (p *PostgresSource) Ack(ctx context.Context, msg hermod.Message) error

func (*PostgresSource) Close ¶

func (p *PostgresSource) Close() error

func (*PostgresSource) DiscoverColumns ¶

func (p *PostgresSource) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)

func (*PostgresSource) DiscoverDatabases ¶

func (p *PostgresSource) DiscoverDatabases(ctx context.Context) ([]string, error)

func (*PostgresSource) DiscoverPublications ¶

func (p *PostgresSource) DiscoverPublications(ctx context.Context) ([]hermod.PublicationInfo, error)

DiscoverPublications returns all publications and the tables each one covers so the user can pick an existing publication that already includes their table or decide to create a new one.

func (*PostgresSource) DiscoverReplicationSlots ¶

func (p *PostgresSource) DiscoverReplicationSlots(ctx context.Context) ([]hermod.ReplicationSlotInfo, error)

DiscoverReplicationSlots returns all logical replication slots present in the connected PostgreSQL instance. The UI uses this so users can reuse an existing slot instead of always creating a new one.

func (*PostgresSource) DiscoverTables ¶

func (p *PostgresSource) DiscoverTables(ctx context.Context) ([]string, error)

func (*PostgresSource) ExecuteSQL ¶

func (p *PostgresSource) ExecuteSQL(ctx context.Context, query string) ([]map[string]any, error)

func (*PostgresSource) GetLag ¶

func (p *PostgresSource) GetLag(ctx context.Context) (uint64, error)

func (*PostgresSource) IsReady ¶

func (p *PostgresSource) IsReady(ctx context.Context) error

func (*PostgresSource) LastStreamActivity ¶

func (p *PostgresSource) LastStreamActivity() time.Time

LastStreamActivity implements hermod.StreamLivenessReporter.

func (*PostgresSource) PendingWork ¶

func (p *PostgresSource) PendingWork() (pending bool, known bool)

PendingWork implements hermod.PendingWorkReporter. It reports whether this source has handed over changes that were never acknowledged.

Only positions actually delivered count. WAL that arrived and was filtered out — a table this workflow does not follow, or traffic in another database on the same server — is not work this pipeline owes, even though it shows up as replication lag. Conflating the two is what would make an idle workflow on a busy server look permanently wedged.

func (*PostgresSource) Ping ¶

func (p *PostgresSource) Ping(ctx context.Context) error

func (*PostgresSource) Read ¶

func (p *PostgresSource) Read(ctx context.Context) (hermod.Message, error)

func (*PostgresSource) Sample ¶

func (p *PostgresSource) Sample(ctx context.Context, table string) (hermod.Message, error)

func (*PostgresSource) SetInitialLoad ¶

func (p *PostgresSource) SetInitialLoad(enabled bool)

SetInitialLoad asks for a one-time backfill of the watched tables before streaming starts.

It happens only when the replication slot is created, which is the source's own record of having run before: if the slot exists, changes have been streamed from it and the rows are already downstream. That makes the backfill once-only without any extra bookkeeping, and means enabling this on a running workflow does nothing until the slot is dropped.

func (*PostgresSource) SetLogger ¶

func (p *PostgresSource) SetLogger(logger hermod.Logger)

func (*PostgresSource) SetPersistentSlot ¶

func (p *PostgresSource) SetPersistentSlot(persistent bool)

func (*PostgresSource) Snapshot ¶

func (p *PostgresSource) Snapshot(ctx context.Context, tables ...string) error

func (*PostgresSource) StreamSilenceThreshold ¶

func (p *PostgresSource) StreamSilenceThreshold() time.Duration

StreamSilenceThreshold implements hermod.StreamLivenessReporter. It is derived from the server's own wal_sender_timeout rather than assumed, because a server configured with a long timeout sends keepalives correspondingly rarely and a hardcoded deadline would declare a healthy stream dead.