(temp): fix extract entries query

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

@@ -15,6 +15,7 @@ const syncInterval = time.Hour * 1
 
 type (
 	KVRow struct {
+		ID      int       `db:"id"`
 		Key     []byte    `db:"key"`
 		Value   []byte    `db:"value"`
 		Updated time.Time `db:"updated"`
@@ -73,7 +74,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.Key, &kvRow.Value, &kvRow.Updated)
+			err := row.Scan(&kvRow.ID, &kvRow.Key, &kvRow.Value, &kvRow.Updated)
 			return kvRow, err
 		})
 		if err != nil {
@@ -91,7 +92,8 @@ func (s *Syncer) Process(ctx context.Context) error {
 				}
 				decodedValue := data.DecodeValue(row.Value)
 
-				s.logg.Debug("processing row", "key_type", decodedKeyDataType, "session_id", sessionID, "value", decodedValue, "timestamp", row.Updated, "column", column)
+				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)
 				query := fmt.Sprintf(`
 				INSERT INTO ussd_data (phone_number, %s) 
 				VALUES ($1, $2)

queries.sql

@@ -1,8 +1,11 @@
 --name: extract-entries
-SELECT key, value, updated FROM kv_vise
+SELECT id, 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)
+UPDATE ussd_sync SET last_sync = ($1)
+
+-- 202b3235343732343934323039370013
+-- 202b3235343731313030303132330013