Merge pull request #2 from udacity/bernielomax/feat/change-event-incl…

…udes-id

feat: Change events include the changeset id

changeset.go

@@ -34,6 +34,7 @@ func ParseChangesetKind(kind string) ChangesetKind {
 
 // Changeset represents a changeset for a record on a Postgres table.
 type Changeset struct {
+	ID        int64              `json:"id"`
 	Kind      ChangesetKind      `json:"kind"`
 	Schema    string             `json:"schema"`
 	Table     string             `json:"table"`

db/wal2json.go

@@ -8,6 +8,7 @@ type Wal2JSONMessage struct {
 
 // Wal2JSONChange represents a changeset within a Wal2JSONMessage.
 type Wal2JSONChange struct {
+	ID           int64            `json:"id"`
 	Kind         string           `json:"kind"`
 	Schema       string           `json:"schema"`
 	Table        string           `json:"table"`

lr_listener.go

@@ -222,6 +222,7 @@ func (l *LogicalReplicationListener) processMessage(msg *pgx.ReplicationMessage)
 
 	for _, change := range w2jmsg.Changes {
 		cs := &Changeset{
+			ID:     change.ID,
 			Kind:   ParseChangesetKind(change.Kind),
 			Schema: change.Schema,
 			Table:  change.Table,

notify_listener.go

@@ -8,8 +8,9 @@ import (
 	"time"
 
 	"github.com/jackc/pgx"
-	"github.com/perangel/warp-pipe/internal/store"
 	log "github.com/sirupsen/logrus"
+
+	"github.com/perangel/warp-pipe/internal/store"
 )
 
 // NotifyOption is a NotifyListener option function
@@ -167,6 +168,7 @@ func (l *NotifyListener) processMessage(msg *pgx.Notification) {
 
 func (l *NotifyListener) processChangeset(event *store.Event) {
 	cs := &Changeset{
+		ID:        event.ID,
 		Kind:      ParseChangesetKind(event.Action),
 		Schema:    event.SchemaName,
 		Table:     event.TableName,