cmd/main.go

@@ -71,7 +71,7 @@ func main() {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		// syncer.Run()
+		syncer.Run()
 	}()
 
 	<-ctx.Done()
@@ -80,7 +80,7 @@ func main() {
 
 	wg.Add(1)
 	go func() {
-		// syncer.Stop()
+		syncer.Stop()
 		defer wg.Done()
 	}()
 

internal/syncer/syncer.go

@@ -11,11 +11,10 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
-const syncInterval = time.Hour * 1
+const syncInterval = time.Second * 5
 
 type (
 	KVRow struct {
-		ID      int       `db:"id"`
 		Key     []byte    `db:"key"`
 		Value   []byte    `db:"value"`
 		Updated time.Time `db:"updated"`
@@ -74,7 +73,7 @@ func (s *Syncer) Process(ctx context.Context) error {
 
 		kvRows, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (KVRow, error) {
 			var kvRow KVRow
-			err := row.Scan(&kvRow.ID, &kvRow.Key, &kvRow.Value, &kvRow.Updated)
+			err := row.Scan(&kvRow.Key, &kvRow.Value, &kvRow.Updated)
 			return kvRow, err
 		})
 		if err != nil {
@@ -92,8 +91,7 @@ func (s *Syncer) Process(ctx context.Context) error {
 				}
 				decodedValue := data.DecodeValue(row.Value)
 
-				s.logg.Info("processing row", "id", row.ID)
-				s.logg.Debug("processing row", "id", row.ID, "key_type", decodedKeyDataType, "session_id", sessionID, "value", decodedValue, "timestamp", row.Updated, "column", column)
+				s.logg.Debug("processing row", "key_type", decodedKeyDataType, "session_id", sessionID, "value", decodedValue, "timestamp", row.Updated, "column", column)
 				query := fmt.Sprintf(`
 				INSERT INTO ussd_data (phone_number, %s) 
 				VALUES ($1, $2)

queries.sql

@@ -1,11 +1,8 @@
 --name: extract-entries
-SELECT id, key, value, updated FROM kv_vise
+SELECT key, value, updated FROM kv_vise
 WHERE updated > (SELECT last_sync FROM ussd_sync LIMIT 1)
 ORDER BY updated DESC
 
 --name: update-cursor
 -- $1: timestamp
-UPDATE ussd_sync SET last_sync = ($1)
-
--- 202b3235343732343934323039370013
--- 202b3235343731313030303132330013
+UPDATE ussd_sync SET last_sync = ($1)