Add best-effort geolocation to the IP lookup from geojs and DB-IP City Lite (closes #239)

Every IP lookup result gains `geo`. One request to geojs carries all uncached addresses of a lookup, within 1 second. On a timeout, an error or an answer with no location, the lookup reads DB-IP City Lite, downloaded monthly into the state directory. Geojs answers stay in memory, at most 250,000, and in `cache.db` for 7 days. The home page footer credits DB-IP. Judgement call: a reserved address goes to neither source, and its `geo` is null. Judgement call: `updated` for a DB-IP answer is when the copy was downloaded. Deviation: the cache is SQLite, by sneak's words, not the JSON file the standing rule asks for. Model: opus-5-5
2026-11-10 15:48:12 +01:00
1 parent 2c3f6f71dc
commit cca402c73d
@@ -0,0 +1,253 @@
+package geo
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+)
+
+const (
+	// entryLifetime is how long a geojs answer is kept after its fetch.
+	entryLifetime = 7 * 24 * time.Hour
+
+	// flushInterval is how often new answers are written to cache.db.
+	flushInterval = 5 * time.Second
+
+	// sweepInterval is how often expired answers are deleted.
+	sweepInterval = 10 * time.Minute
+
+	// rowColumns are the columns of the geolocations table in the order
+	// scanRow reads them.
+	rowColumns = "address, country_code, country, continent_code, region, " +
+		"city, latitude, longitude, accuracy_radius_km, time_zone, fetched_at"
+)
+
+// rowScanner is the Scan method of *sql.Row and *sql.Rows.
+type rowScanner interface {
+	Scan(dest ...any) error
+}
+
+// cached returns the current answer held for address, from memory, then from
+// the answers waiting to be written, then from cache.db. It returns nil when
+// there is none.
+func (l *Locator) cached(ctx context.Context, address string) *Location {
+	found, ok := l.entries.Get(address)
+	if ok {
+		if !l.expired(found) {
+			return found
+		}
+
+		l.entries.Remove(address)
+	}
+
+	l.pendingMu.Lock()
+	waiting := l.pending[address]
+	l.pendingMu.Unlock()
+
+	if waiting != nil && !l.expired(waiting) {
+		l.entries.Add(address, waiting)
+
+		return waiting
+	}
+
+	stored, err := l.readRow(ctx, address)
+	if err != nil {
+		l.logger.Warn("Failed to read a geolocation cache row",
+			"address", address, "error", err)
+
+		return nil
+	}
+
+	if stored != nil {
+		l.entries.Add(address, stored)
+	}
+
+	return stored
+}
+
+// expired says whether the answer is at the end of its lifetime.
+func (l *Locator) expired(answer *Location) bool {
+	return !l.now().Before(answer.Updated.Add(entryLifetime))
+}
+
+// cutoff returns the fetch time at or before which an answer has expired.
+func (l *Locator) cutoff() int64 {
+	return l.now().Add(-entryLifetime).Unix()
+}
+
+// store keeps an answer in memory and queues it for cache.db. While cache.db
+// cannot be written, the queue stops growing at maxEntries answers.
+func (l *Locator) store(address string, answer *Location) {
+	l.entries.Add(address, answer)
+
+	l.pendingMu.Lock()
+	defer l.pendingMu.Unlock()
+
+	if len(l.pending) < l.maxEntries {
+		l.pending[address] = answer
+	}
+}
+
+// keepCache writes new answers and deletes expired ones until ctx ends.
+func (l *Locator) keepCache(ctx context.Context) {
+	flushes := time.NewTicker(flushInterval)
+	defer flushes.Stop()
+
+	sweeps := time.NewTicker(sweepInterval)
+	defer sweeps.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			l.flush(context.WithoutCancel(ctx))
+
+			return
+		case <-flushes.C:
+			l.flush(ctx)
+		case <-sweeps.C:
+			l.sweep(ctx)
+		}
+	}
+}
+
+// load reads the answers that have not expired into memory, at most maxEntries
+// of them, the newest last so that they are the last to be dropped.
+func (l *Locator) load(ctx context.Context) error {
+	rows, err := l.db.QueryContext(ctx,
+		"SELECT "+rowColumns+" FROM (SELECT "+rowColumns+" FROM geolocations "+
+			"WHERE fetched_at > ? ORDER BY fetched_at DESC LIMIT ?) "+
+			"ORDER BY fetched_at",
+		l.cutoff(), l.maxEntries)
+	if err != nil {
+		return fmt.Errorf("reading the geolocation cache: %w", err)
+	}
+
+	defer func() { _ = rows.Close() }()
+
+	for rows.Next() {
+		address, answer, err := scanRow(rows)
+		if err != nil {
+			return fmt.Errorf("reading a geolocation cache row: %w", err)
+		}
+
+		l.entries.Add(address, answer)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return fmt.Errorf("reading the geolocation cache: %w", err)
+	}
+
+	return nil
+}
+
+// readRow returns the current answer for address in cache.db, or nil when it
+// has none.
+func (l *Locator) readRow(
+	ctx context.Context, address string,
+) (*Location, error) {
+	row := l.db.QueryRowContext(ctx,
+		"SELECT "+rowColumns+" FROM geolocations "+
+			"WHERE address = ? AND fetched_at > ?",
+		address, l.cutoff())
+
+	_, answer, err := scanRow(row)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, nil //nolint:nilnil // no row is not a failure
+	}
+
+	return answer, err
+}
+
+// scanRow reads one row of the geolocations table.
+func scanRow(row rowScanner) (string, *Location, error) {
+	var (
+		address string
+		fetched int64
+		answer  Location
+	)
+
+	err := row.Scan(&address, &answer.CountryCode, &answer.Country,
+		&answer.ContinentCode, &answer.Region, &answer.City, &answer.Latitude,
+		&answer.Longitude, &answer.AccuracyRadiusKm, &answer.TimeZone, &fetched)
+	if err != nil {
+		return "", nil, err
+	}
+
+	answer.Source = sourceGeoJS
+	answer.Updated = time.Unix(fetched, 0).UTC()
+
+	return address, &answer, nil
+}
+
+// flush writes the pending answers to cache.db. Those it cannot write stay
+// pending, unless a newer answer for the address has arrived meanwhile.
+func (l *Locator) flush(ctx context.Context) {
+	l.pendingMu.Lock()
+	batch := l.pending
+	l.pending = make(map[string]*Location)
+	l.pendingMu.Unlock()
+
+	if len(batch) == 0 {
+		return
+	}
+
+	err := l.writeRows(ctx, batch)
+	if err == nil {
+		return
+	}
+
+	l.logger.Error("Failed to write the geolocation cache", "error", err)
+
+	l.pendingMu.Lock()
+	defer l.pendingMu.Unlock()
+
+	maps.Insert(batch, maps.All(l.pending))
+	l.pending = batch
+}
+
+// writeRows writes batch to cache.db in one transaction.
+func (l *Locator) writeRows(
+	ctx context.Context, batch map[string]*Location,
+) error {
+	tx, err := l.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	defer func() { _ = tx.Rollback() }()
+
+	for address, answer := range batch {
+		_, err = tx.ExecContext(ctx,
+			"INSERT OR REPLACE INTO geolocations ("+rowColumns+") "+
+				"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+			address, answer.CountryCode, answer.Country, answer.ContinentCode,
+			answer.Region, answer.City, answer.Latitude, answer.Longitude,
+			answer.AccuracyRadiusKm, answer.TimeZone, answer.Updated.Unix())
+		if err != nil {
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
+// sweep deletes the expired answers from memory and cache.db.
+func (l *Locator) sweep(ctx context.Context) {
+	for _, address := range l.entries.Keys() {
+		answer, ok := l.entries.Peek(address)
+		if ok && l.expired(answer) {
+			l.entries.Remove(address)
+		}
+	}
+
+	_, err := l.db.ExecContext(ctx,
+		"DELETE FROM geolocations WHERE fetched_at <= ?", l.cutoff())
+	if err != nil {
+		l.logger.Error("Failed to delete expired geolocation cache rows",
+			"error", err)
+	}
+}