postgres: reset result writer save interval after flush (#281)

Update the Postgres result writer so a successful timed batch save refreshes lastSave. Previously lastSave was initialized once and never updated, causing the time-based flush condition to remain true after the first minute and flush every subsequent result immediately. Add a regression test with a controlled clock to verify that a timed flush does not force the next row to flush until the interval elapses again.
2026-09-19 07:27:12 +08:00 · 2026-05-30 10:32:30 +03:00
parent f21caaa179
commit ec730616ac
4 changed files with 222 additions and 40 deletions
@@ -73,13 +73,13 @@ type Review struct {
 	TextOriginal        string  `json:"text_original"`
 	TextTranslated      string  `json:"text_translated"`

-	ReplyText                string `json:"reply_text,omitempty"`
-	ReplyTextOriginal        string `json:"reply_text_original,omitempty"`
-	ReplyLanguage            string `json:"reply_language,omitempty"`
-	ReplyTranslatedLang      string `json:"reply_translated_lang,omitempty"`
-	ReplyPostedAtUnixMicros  int64  `json:"reply_posted_at_unix_micros,omitempty"`
-	ReplyUpdatedAtUnixMicros int64  `json:"reply_updated_at_unix_micros,omitempty"`
-	PublishedAt    *time.Time `json:"published_at,omitempty"`
+	ReplyText                string     `json:"reply_text,omitempty"`
+	ReplyTextOriginal        string     `json:"reply_text_original,omitempty"`
+	ReplyLanguage            string     `json:"reply_language,omitempty"`
+	ReplyTranslatedLang      string     `json:"reply_translated_lang,omitempty"`
+	ReplyPostedAtUnixMicros  int64      `json:"reply_posted_at_unix_micros,omitempty"`
+	ReplyUpdatedAtUnixMicros int64      `json:"reply_updated_at_unix_micros,omitempty"`
+	PublishedAt              *time.Time `json:"published_at,omitempty"`
 }

 const reviewPublishedAtFutureSkew = 24 * time.Hour
@@ -97,38 +97,38 @@ type Entry struct {
 	OpenHours  map[string][]string `json:"open_hours"`
 	// PopularTImes is a map with keys the days of the week
 	// and value is a map with key the hour and value the traffic in that time
-	PopularTimes        map[string]map[int]int `json:"popular_times"`
-	WebSite             string                 `json:"web_site"`
-	Phone               string                 `json:"phone"`
-	PlusCode            string                 `json:"plus_code"`
-	ReviewCount         int                    `json:"review_count"`
-	ReviewRating        float64                `json:"review_rating"`
-	ReviewsPerRating    map[int]int            `json:"reviews_per_rating"`
-	Latitude            float64                `json:"latitude"`
+	PopularTimes     map[string]map[int]int `json:"popular_times"`
+	WebSite          string                 `json:"web_site"`
+	Phone            string                 `json:"phone"`
+	PlusCode         string                 `json:"plus_code"`
+	ReviewCount      int                    `json:"review_count"`
+	ReviewRating     float64                `json:"review_rating"`
+	ReviewsPerRating map[int]int            `json:"reviews_per_rating"`
+	Latitude         float64                `json:"latitude"`
 	// Longtitude holds the longitude. The struct field and the legacy JSON
 	// key are misspelled ("longtitude"); MarshalJSON also emits the correctly
 	// spelled "longitude" key, and UnmarshalJSON accepts either. The field
 	// name is kept for backwards compatibility with existing imports.
-	Longtitude          float64                `json:"longtitude"`
-	Status              string                 `json:"status"`
-	Description         string                 `json:"description"`
-	ReviewsLink         string                 `json:"reviews_link"`
-	Thumbnail           string                 `json:"thumbnail"`
-	Timezone            string                 `json:"timezone"`
-	PriceRange          string                 `json:"price_range"`
-	DataID              string                 `json:"data_id"`
-	StreetViewURL       string                 `json:"street_view_url"`
-	PlaceID             string                 `json:"place_id"`
-	Images              []Image                `json:"images"`
-	Reservations        []LinkSource           `json:"reservations"`
-	OrderOnline         []LinkSource           `json:"order_online"`
-	Menu                LinkSource             `json:"menu"`
-	Owner               Owner                  `json:"owner"`
-	CompleteAddress     Address                `json:"complete_address"`
-	About               []About                `json:"about"`
-	UserReviews         []Review               `json:"user_reviews"`
-	UserReviewsExtended []Review               `json:"user_reviews_extended"`
-	Emails              []string               `json:"emails"`
+	Longtitude          float64      `json:"longtitude"`
+	Status              string       `json:"status"`
+	Description         string       `json:"description"`
+	ReviewsLink         string       `json:"reviews_link"`
+	Thumbnail           string       `json:"thumbnail"`
+	Timezone            string       `json:"timezone"`
+	PriceRange          string       `json:"price_range"`
+	DataID              string       `json:"data_id"`
+	StreetViewURL       string       `json:"street_view_url"`
+	PlaceID             string       `json:"place_id"`
+	Images              []Image      `json:"images"`
+	Reservations        []LinkSource `json:"reservations"`
+	OrderOnline         []LinkSource `json:"order_online"`
+	Menu                LinkSource   `json:"menu"`
+	Owner               Owner        `json:"owner"`
+	CompleteAddress     Address      `json:"complete_address"`
+	About               []About      `json:"about"`
+	UserReviews         []Review     `json:"user_reviews"`
+	UserReviewsExtended []Review     `json:"user_reviews_extended"`
+	Emails              []string     `json:"emails"`
 }

 // entryAlias is used inside Marshal/UnmarshalJSON to avoid infinite recursion
@@ -138,6 +138,8 @@ type entryAlias Entry
 // MarshalJSON emits both the legacy "longtitude" key (preserved for backwards
 // compatibility) and the correctly spelled "longitude" key so downstream
 // consumers can migrate without a flag day.
+//
+//nolint:gocritic // value receiver preserves json.Marshaler behavior for Entry values.
 func (e Entry) MarshalJSON() ([]byte, error) {
 	return json.Marshal(struct {
 		Longitude float64 `json:"longitude"`
@@ -161,9 +163,11 @@ func (e *Entry) UnmarshalJSON(data []byte) error {
 	if err := json.Unmarshal(data, &aux); err != nil {
 		return err
 	}
+
 	if e.Longtitude == 0 && aux.Longitude != nil {
 		e.Longtitude = *aux.Longitude
 	}
+
 	return nil
 }

@@ -581,7 +585,7 @@ func parseReviews(reviewsI []any) []Review {
 		if isAggregator {
 			review.RatingFloat = getNthElementAndCast[float64](el, 2, 8, 1)
 		} else {
-			review.RatingFloat = float64(rating)
+			review.RatingFloat = float64(review.Rating)
 		}

 		r3 := getNthElementAndCast[[]any](el, 3)
@@ -211,6 +211,7 @@ func Test_EntryMarshalEmitsBothLongitudeKeys(t *testing.T) {
 	require.NoError(t, err)

 	var got map[string]json.RawMessage
+
 	require.NoError(t, json.Unmarshal(raw, &got))

 	require.JSONEq(t, "2.5", string(got["longtitude"]), "legacy key preserved")
@@ -219,16 +220,19 @@ func Test_EntryMarshalEmitsBothLongitudeKeys(t *testing.T) {

 func Test_EntryUnmarshalAcceptsEitherLongitudeKey(t *testing.T) {
 	var legacy gmaps.Entry
+
 	require.NoError(t, json.Unmarshal([]byte(`{"longtitude":42.5}`), &legacy))
 	require.Equal(t, 42.5, legacy.Longtitude)

 	var modern gmaps.Entry
+
 	require.NoError(t, json.Unmarshal([]byte(`{"longitude":42.5}`), &modern))
 	require.Equal(t, 42.5, modern.Longtitude)

 	// When both are present, the legacy spelling wins so existing files
 	// round-trip byte-identical.
 	var both gmaps.Entry
+
 	require.NoError(t, json.Unmarshal([]byte(`{"longtitude":1.0,"longitude":2.0}`), &both))
 	require.Equal(t, 1.0, both.Longtitude)
 }
@@ -18,18 +18,24 @@ import (
 )

 func NewResultWriter(db *sql.DB) scrapemate.ResultWriter {
-	return &resultWriter{db: db}
+	return &resultWriter{
+		db:           db,
+		now:          time.Now,
+		saveInterval: time.Minute,
+	}
 }

 type resultWriter struct {
-	db *sql.DB
+	db           *sql.DB
+	now          func() time.Time
+	saveInterval time.Duration
 }

 func (r *resultWriter) Run(ctx context.Context, in <-chan scrapemate.Result) error {
 	const maxBatchSize = 50

 	buff := make([]*gmaps.Entry, 0, 50)
-	lastSave := time.Now().UTC()
+	lastSave := r.currentTime()

 	for result := range in {
 		entry, ok := result.Data.(*gmaps.Entry)
@@ -40,13 +46,14 @@ func (r *resultWriter) Run(ctx context.Context, in <-chan scrapemate.Result) err

 		buff = append(buff, entry)

-		if len(buff) >= maxBatchSize || time.Since(lastSave) >= time.Minute {
+		if len(buff) >= maxBatchSize || r.currentTime().Sub(lastSave) >= r.saveEvery() {
 			err := r.batchSave(ctx, buff)
 			if err != nil {
 				return err
 			}

 			buff = buff[:0]
+			lastSave = r.currentTime()
 		}
 	}

@@ -60,6 +67,22 @@ func (r *resultWriter) Run(ctx context.Context, in <-chan scrapemate.Result) err
 	return nil
 }

+func (r *resultWriter) currentTime() time.Time {
+	if r.now == nil {
+		return time.Now()
+	}
+
+	return r.now()
+}
+
+func (r *resultWriter) saveEvery() time.Duration {
+	if r.saveInterval == 0 {
+		return time.Minute
+	}
+
+	return r.saveInterval
+}
+
 func (r *resultWriter) batchSave(ctx context.Context, entries []*gmaps.Entry) error {
 	if len(entries) == 0 {
 		return nil
@@ -0,0 +1,151 @@
+package postgres //nolint:testpackage // tests need unexported clock hooks on resultWriter.
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"fmt"
+	"io"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gosom/scrapemate"
+	"github.com/stretchr/testify/require"
+
+	"github.com/gosom/google-maps-scraper/gmaps"
+)
+
+func TestResultWriterResetsSaveIntervalAfterTimedFlush(t *testing.T) {
+	db, execs := newCountingDB(t)
+	defer db.Close()
+
+	base := time.Date(2026, time.May, 30, 12, 0, 0, 0, time.UTC)
+	clock := &testClock{now: base}
+	writer := &resultWriter{
+		db:           db,
+		now:          clock.Now,
+		saveInterval: time.Minute,
+	}
+
+	in := make(chan scrapemate.Result)
+	done := make(chan error, 1)
+
+	go func() {
+		done <- writer.Run(context.Background(), in)
+	}()
+
+	in <- resultWithEntry("first")
+
+	clock.Set(base.Add(time.Minute + time.Second))
+	in <- resultWithEntry("second")
+
+	require.Eventually(t, func() bool {
+		return execs.Load() == 1
+	}, time.Second, 10*time.Millisecond)
+
+	clock.Set(base.Add(time.Minute + 2*time.Second))
+	in <- resultWithEntry("third")
+
+	require.Never(t, func() bool {
+		return execs.Load() > 1
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	close(in)
+	require.NoError(t, <-done)
+	require.Equal(t, int64(2), execs.Load())
+}
+
+func resultWithEntry(id string) scrapemate.Result {
+	return scrapemate.Result{
+		Data: &gmaps.Entry{
+			ID:         id,
+			Title:      id,
+			Latitude:   1,
+			Longtitude: 2,
+		},
+	}
+}
+
+type testClock struct {
+	mu  sync.Mutex
+	now time.Time
+}
+
+func (c *testClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.now
+}
+
+func (c *testClock) Set(now time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.now = now
+}
+
+var countingDriverSeq atomic.Int64
+
+func newCountingDB(t *testing.T) (*sql.DB, *atomic.Int64) {
+	t.Helper()
+
+	execs := &atomic.Int64{}
+	driverName := fmt.Sprintf("counting-resultwriter-%d", countingDriverSeq.Add(1))
+	sql.Register(driverName, &countingDriver{execs: execs})
+
+	db, err := sql.Open(driverName, "")
+	require.NoError(t, err)
+
+	return db, execs
+}
+
+type countingDriver struct {
+	execs *atomic.Int64
+}
+
+func (d *countingDriver) Open(string) (driver.Conn, error) {
+	return &countingConn{execs: d.execs}, nil
+}
+
+type countingConn struct {
+	execs *atomic.Int64
+}
+
+func (c *countingConn) Prepare(string) (driver.Stmt, error) {
+	return nil, fmt.Errorf("prepare is not implemented")
+}
+
+func (c *countingConn) Close() error {
+	return nil
+}
+
+func (c *countingConn) Begin() (driver.Tx, error) {
+	return countingTx{}, nil
+}
+
+func (c *countingConn) BeginTx(context.Context, driver.TxOptions) (driver.Tx, error) {
+	return countingTx{}, nil
+}
+
+func (c *countingConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) {
+	c.execs.Add(1)
+
+	return driver.RowsAffected(1), nil
+}
+
+func (c *countingConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) {
+	return nil, io.EOF
+}
+
+type countingTx struct{}
+
+func (countingTx) Commit() error {
+	return nil
+}
+
+func (countingTx) Rollback() error {
+	return nil
+}