Location history retention & background pruning

README.md

@@ -218,6 +218,34 @@ curl -i -X POST http://localhost:8080/api/v1/locations \
   -d '{"vehicle_id":"bus-1","latitude":-1.29,"longitude":36.82,"timestamp":1752566400}{"extra":1}'
 ```
 
+**Data Retention & Privacy**
+
+`location_points` is the highest-volume table in the system: the server stores one row per vehicle per reporting interval. Fifty vehicles reporting every 10 seconds is 5 rows a second — about 157.7 million rows a year, or roughly 105 million if vehicles only report across a 16-hour service day. That data is also a per-driver GPS trace, which most agencies should not keep indefinitely.
+
+The server can delete location points once they pass a configured age:
+
+|Variable                   |Default|Purpose                                               |
+|---------------------------|-------|------------------------------------------------------|
+|`LOCATION_RETENTION_PERIOD`|`0`    |How long to keep location points. `0` disables pruning|
+|`LOCATION_PRUNE_INTERVAL`  |`1h`   |How often the pruner runs                             |
+|`LOCATION_PRUNE_BATCH_SIZE`|`10000`|Maximum rows deleted per statement                    |
+
+Notes for operators:
+
+- **Retention is off by default.** With `LOCATION_RETENTION_PERIOD` unset (or `0`), history is kept forever, which is the behavior of every release before this feature. Pruning starts only when an agency opts in.
+- **Deletion is permanent.** There is no archival or export step; pruned points are gone. Export anything worth keeping before enabling retention.
+- **Retention is measured from server receipt time** (`received_at`), not the device-reported `timestamp` in the payload. A driver phone with a wrong clock can therefore neither keep its history past the retention period nor have it deleted early.
+- Deletes run in batches of `LOCATION_PRUNE_BATCH_SIZE`, each in its own transaction, so clearing a large backlog does not hold locks long enough to stall location ingest.
+- The first pass runs one full interval after startup, not at boot.
+- Agencies should pick a retention period consistent with local law and their own driver-privacy policy.
+
+Example — keep 90 days of history, sweeping hourly:
+
+```bash
+export LOCATION_RETENTION_PERIOD=2160h
+export LOCATION_PRUNE_INTERVAL=1h
+```
+
 **Technology Stack:**
 
 - **Language:** Go (aligns with Maglev and OTSF’s server-side direction)

db/query.sql

@@ -181,3 +181,14 @@ FROM trips t
 JOIN users u ON u.id = t.user_id
 WHERE t.status = 'active'
 ORDER BY t.vehicle_id, t.start_time DESC;
+
+-- name: DeleteLocationPointsBefore :execrows
+-- Batched retention delete. The ctid subquery bounds each statement so a large
+-- backlog is removed over many small transactions instead of one long lock.
+DELETE FROM location_points
+WHERE ctid IN (
+    SELECT expired.ctid FROM location_points AS expired
+    WHERE expired.received_at < sqlc.arg('cutoff')
+    ORDER BY expired.received_at
+    LIMIT sqlc.arg('batch_size')
+);

docs/development.md

@@ -62,6 +62,14 @@ You can run Postgres in Docker and run the Go server directly:
    export STALENESS_THRESHOLD=5m
    ```
 
+   Location retention is optional and off unless you set it:
+
+   ```bash
+   export LOCATION_RETENTION_PERIOD=720h   # keep 30 days; 0 or unset keeps forever
+   export LOCATION_PRUNE_INTERVAL=1h       # how often the pruner sweeps
+   export LOCATION_PRUNE_BATCH_SIZE=10000  # rows deleted per statement
+   ```
+
 3. Run server:
 
    ```bash
@@ -142,6 +150,38 @@ Custom example:
 go run ./cmd/simulator -url http://localhost:8080 -vehicles 20 -interval 2s -duration 2m
 ```
 
+## Watching Retention Prune Locally
+
+Retention deletes location points older than `LOCATION_RETENTION_PERIOD`, measured from
+`received_at` (when the server stored the point), not the device-reported `timestamp`.
+The first sweep runs one full `LOCATION_PRUNE_INTERVAL` after startup.
+
+To watch it work without waiting hours, run the server with a very short retention:
+
+```bash
+export LOCATION_RETENTION_PERIOD=2m
+export LOCATION_PRUNE_INTERVAL=30s
+make run
+```
+
+Then generate some points and wait for the sweep:
+
+```bash
+make simulate
+```
+
+The server logs `location retention enabled` at startup, and each sweep that removes
+anything logs `pruned expired location points` with the row count and cutoff. Confirm
+against the database:
+
+```bash
+docker compose exec db psql -U postgres -d vehicle_positions \
+  -c "SELECT count(*), min(received_at) FROM location_points;"
+```
+
+Deletion is permanent, so use a scratch database for this rather than one holding data
+you care about.
+
 ## API Sanity Checks
 
 ### Submit one location

main.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 )
@@ -181,6 +182,33 @@ func main() {
 	loginLimiter := NewLoginRateLimiter()
 	defer loginLimiter.Stop()
 
+	// Retention is opt-in: a zero period means keep location history forever,
+	// which is the behavior every existing deployment has today. Any other value
+	// is a deliberate request to delete data, so a bad one is reported rather
+	// than quietly ignored.
+	if retentionPeriod := envDurationOrDefault("LOCATION_RETENTION_PERIOD", 0); retentionPeriod != 0 {
+		pruneInterval := envDurationOrDefault("LOCATION_PRUNE_INTERVAL", time.Hour)
+		batchSize := envInt32OrDefault("LOCATION_PRUNE_BATCH_SIZE", 10_000)
+
+		pruner, err := NewLocationPruner(store, retentionPeriod, pruneInterval, batchSize)
+		if err != nil {
+			// Refusing to start the whole server would take live vehicle
+			// tracking down over an optional feature, so carry on with
+			// retention off — the safe direction, since nothing is deleted.
+			slog.Error("location retention disabled: invalid configuration", "error", err)
+		} else {
+			defer pruner.Stop()
+
+			if pruneInterval > retentionPeriod {
+				slog.Warn("prune interval is longer than the retention period, so points outlive it by up to one interval",
+					"interval", pruneInterval.String(), "retention", retentionPeriod.String())
+			}
+
+			slog.Info("location retention enabled",
+				"retention", retentionPeriod.String(), "interval", pruneInterval.String(), "batch_size", batchSize)
+		}
+	}
+
 	cutoff := time.Now().Add(-maxAge)
 	recentLocations, err := store.GetRecentLocations(ctx, cutoff)
 	if err != nil {
@@ -246,6 +274,21 @@ func envDurationOrDefault(key string, fallback time.Duration) time.Duration {
 	return fallback
 }
 
+// envInt32OrDefault reads a positive 32-bit integer from the environment. The
+// 32-bit bound is deliberate: these values reach the database as int32 query
+// parameters, and parsing wider would let an oversized value wrap to a negative.
+func envInt32OrDefault(key string, fallback int32) int32 {
+	if v := os.Getenv(key); v != "" {
+		n, err := strconv.ParseInt(v, 10, 32)
+		if err != nil || n < 1 {
+			slog.Warn("invalid positive integer, using default", "key", key, "value", v, "default", fallback)
+			return fallback
+		}
+		return int32(n)
+	}
+	return fallback
+}
+
 type statusRecorder struct {
 	http.ResponseWriter
 	status int

main_test.go

@@ -72,3 +72,34 @@ func TestStatusRecorder_CapturesStatus(t *testing.T) {
 	assert.Equal(t, http.StatusNotFound, rec.status)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
+
+func TestEnvInt32OrDefault(t *testing.T) {
+	// Not safe for t.Parallel(); uses t.Setenv and the global logger
+	const key = "TEST_PRUNE_BATCH_SIZE"
+
+	tests := []struct {
+		name     string
+		value    string
+		set      bool
+		expected int32
+	}{
+		{name: "valid", value: "5000", set: true, expected: 5000},
+		{name: "unset", set: false, expected: 10_000},
+		{name: "empty", value: "", set: true, expected: 10_000},
+		{name: "non-numeric", value: "many", set: true, expected: 10_000},
+		{name: "negative", value: "-1", set: true, expected: 10_000},
+		{name: "zero", value: "0", set: true, expected: 10_000},
+		{name: "exceeds int32", value: "2147483648", set: true, expected: 10_000},
+		{name: "max int32", value: "2147483647", set: true, expected: 2147483647},
+		{name: "float", value: "1.5", set: true, expected: 10_000},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.set {
+				t.Setenv(key, tt.value)
+			}
+			assert.Equal(t, tt.expected, envInt32OrDefault(key, 10_000))
+		})
+	}
+}

migrations/000011_add_location_retention_index.down.sql

@@ -0,0 +1 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_location_points_received_at;

migrations/000011_add_location_retention_index.up.sql

@@ -0,0 +1,7 @@
+-- Supports the retention pruner's global scan over expired rows.
+-- The existing (vehicle_id, received_at DESC) index cannot serve this predicate
+-- because received_at is not its leading column.
+-- CONCURRENTLY keeps location ingest writable while the index builds. It cannot
+-- run inside a transaction, which is fine here: golang-migrate's postgres driver
+-- executes migration files directly on the connection, not in a transaction.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_location_points_received_at ON location_points (received_at);

migrations_test.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"io/fs"
+	"regexp"
+	"strconv"
+	"testing"
+
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var migrationFilePattern = regexp.MustCompile(`^(\d+)_([a-z0-9_]+)\.(up|down)\.sql$`)
+
+// TestMigrations_SourceLoads guards the failure this repo keeps hitting: two
+// branches each add a migration with the same version, git merges them without
+// a conflict because the filenames differ, and the server then exits at startup
+// because iofs.New rejects the duplicate version. Needs no database, so CI
+// catches it on every PR.
+func TestMigrations_SourceLoads(t *testing.T) {
+	src, err := iofs.New(migrationsFS, "migrations")
+	require.NoError(t, err, "migrations must load; a duplicate version stops the server from starting")
+	require.NoError(t, src.Close())
+}
+
+func TestMigrations_VersionsAreUniqueAndPaired(t *testing.T) {
+	entries, err := fs.ReadDir(migrationsFS, "migrations")
+	require.NoError(t, err)
+	require.NotEmpty(t, entries, "no migrations found; the embed pattern may be wrong")
+
+	// version -> direction -> filename
+	byVersion := make(map[uint64]map[string]string)
+	for _, entry := range entries {
+		match := migrationFilePattern.FindStringSubmatch(entry.Name())
+		require.NotNil(t, match, "migration filename does not match <version>_<name>.<up|down>.sql: %s", entry.Name())
+
+		version, err := strconv.ParseUint(match[1], 10, 64)
+		require.NoError(t, err)
+		direction := match[3]
+
+		if byVersion[version] == nil {
+			byVersion[version] = make(map[string]string)
+		}
+		if existing, duplicate := byVersion[version][direction]; duplicate {
+			assert.Fail(t, "duplicate migration version",
+				"version %d has two %s migrations (%s and %s) — renumber the newer one to the next free version",
+				version, direction, existing, entry.Name())
+		}
+		byVersion[version][direction] = entry.Name()
+	}
+
+	for version, files := range byVersion {
+		assert.Contains(t, files, "up", "version %d has no up migration", version)
+		assert.Contains(t, files, "down", "version %d has no down migration", version)
+	}
+}