meshcore-analyzer/internal/dbschema/dedup_index.go

package dbschema

import (
	"database/sql"
	"errors"
	"fmt"
	"strconv"

	sqlite3 "github.com/mattn/go-sqlite3"
)

// upsertMergedColumns are the columns the ingestor's observation UPSERT writes
// on conflict, in cmd/ingestor/db.go:
//
//	ON CONFLICT(...) DO UPDATE SET
//	  snr           = COALESCE(excluded.snr,           snr),
//	  rssi          = COALESCE(excluded.rssi,          rssi),
//	  score         = COALESCE(excluded.score,         score),
//	  raw_hex       = COALESCE(excluded.raw_hex,       raw_hex),
//	  resolved_path = COALESCE(excluded.resolved_path, resolved_path)
//
// Two things follow, and collapseDuplicatesAndIndex has to honour both to be a
// faithful replay rather than an approximation of it.
//
// First, `excluded` is the incoming row, so a non-NULL later value REPLACES an
// earlier one. Applied down a group in insertion order the result is the LAST
// non-NULL value, not the first.
//
// Second, every other column is absent from the SET list, so the UPSERT never
// touches it: the surviving row keeps its own direction, timestamp, path_json
// and so on. Merging those too would invent history the ingestor would not have
// written.
var upsertMergedColumns = []string{"snr", "rssi", "score", "raw_hex", "resolved_path"}

// dupGroupsDDL materialises one row per duplicate group: the group key plus the
// earliest id, which is the row that is kept and merged into.
//
// Materialised rather than left as a repeated subquery on purpose. In the
// subquery form every merged column's UPDATE re-ran this GROUP BY over the whole
// observations table — measured at 9.7s on 2.4M rows holding only 5 duplicates,
// nearly all of it the same scan performed repeatedly. This runs inside a write
// transaction and holds the write lock for its duration, so the repetition is
// worth removing.
// The WHERE clause is the whole correctness of this query, and leaving it out
// destroys data.
//
// GROUP BY folds all NULLs into one group. A UNIQUE index does the opposite:
// SQLite treats NULLs as distinct, so a row with NULL in any indexed column can
// never violate it. Group without excluding them and the repair calls rows
// duplicates that the index would have accepted, then deletes them to build an
// index that did not need them gone.
//
// Not hypothetical: on an 11.2M-row instance, 198 of the 222 groups this
// reported had observer_idx IS NULL — 238 of the 262 rows it planned to delete.
// They were direction='tx' rows, and the index built without complaint once
// they were left alone.
//
// transmission_id is NOT NULL in the schema and COALESCE(path_json, ”) can
// never be NULL, so observer_idx is the only one that needs the guard; it is
// written out in full anyway, because the next person to add a column to this
// index should see the rule rather than infer it.
const dupGroupsDDL = `CREATE TEMP TABLE dedup_groups AS
	SELECT transmission_id, observer_idx, COALESCE(path_json, '') AS p,
	       MIN(id) AS keep, COUNT(*) AS n
	  FROM observations
	 WHERE transmission_id IS NOT NULL
	   AND observer_idx IS NOT NULL
	 GROUP BY transmission_id, observer_idx, COALESCE(path_json, '')
	HAVING COUNT(*) > 1`

const dedupIndexDDL = `CREATE UNIQUE INDEX IF NOT EXISTS idx_observations_dedup ` +
	`ON observations(transmission_id, observer_idx, COALESCE(path_json, ''))`

// ensureObservationsDedupIndex creates the unique expression index that the
// ingestor's observation UPSERT resolves its ON CONFLICT target against:
//
//	ON CONFLICT(transmission_id, observer_idx, COALESCE(path_json, ''))
//
// cmd/ingestor/db.go creates this index, but only inside the branch that
// creates the observations table for the first time. Any database whose
// observations table predates that branch therefore never got one, so the
// UPSERT had no conflict target to resolve against. Under modernc.org/sqlite
// that surfaced late — statements compiled lazily, so it failed on the first
// insert. github.com/mattn/go-sqlite3 compiles inside Prepare, so it now
// surfaces at OpenStore. Same bug either way; creating the index here repairs
// the database rather than just moving the error.
//
// Because the index is what was supposed to prevent duplicates, a database that
// never had it can already hold rows violating it — the repo's own
// test-fixtures/e2e-fixture.db did. Those rows are collapsed first. Refusing
// would not be safer: without the index the ingestor cannot prepare its UPSERT,
// so it cannot run at all.
//
// Skipped on v2 schemas (observer_id rather than observer_idx): the UPSERT does
// not apply there, and indexing a column that does not exist would fail.
func ensureObservationsDedupIndex(rw *sql.DB, logf Logger) error {
	hasIdx, err := TableHasColumn(rw, "observations", "observer_idx")
	if err != nil {
		return err
	}
	if !hasIdx {
		return nil
	}
	// Fast path: no duplicates, so the index just builds. Atomic on its own.
	_, err = rw.Exec(dedupIndexDDL)
	if err == nil {
		return nil
	}
	if !isConstraintViolation(err) {
		return err
	}

	// Deleting rows is not something to do quietly, and on a large table this
	// holds the write lock long enough that an operator watching startup
	// deserves to know why before it happens rather than after.
	logf("[dbschema] idx_observations_dedup cannot be created: duplicate observations exist. " +
		"Repairing now — on a large observations table this takes tens of seconds, and " +
		"writers are blocked for part of it.")

	removed, err := collapseDuplicatesAndIndex(rw, logf)
	if err != nil {
		return fmt.Errorf("collapse duplicate observations: %w", err)
	}
	logf("[dbschema] collapsed %d duplicate observation row(s); idx_observations_dedup created", removed)
	return nil
}

// isConstraintViolation reports whether err is SQLite refusing a constraint.
//
// Deliberately a typed check. This used to match on
// strings.Contains(err, "UNIQUE constraint failed"), which made the entire
// repair path — including the row deletion — hinge on the driver's prose, in
// the same change that swapped the driver. A reworded or wrapped message would
// silently skip the repair and surface later as an OpenStore failure with no
// hint as to why. CREATE UNIQUE INDEX can only hit a constraint error because
// the data violates the uniqueness it asks for, so the primary code is the
// right granularity.
func isConstraintViolation(err error) bool {
	var se sqlite3.Error
	if errors.As(err, &se) {
		return se.Code == sqlite3.ErrConstraint
	}
	return false
}

// collapseDuplicatesAndIndex merges duplicate observation rows into the lowest
// id of each group, deletes the rest, and creates the unique index — all in one
// transaction. It returns the number of rows deleted.
//
// The index creation belongs inside the same transaction as the deletion. With
// them separated, a writer inserting a duplicate in the gap makes the index
// creation fail while leaving the deletions committed: rows destroyed and no
// index to show for it. One transaction makes the repair all-or-nothing.
func collapseDuplicatesAndIndex(rw *sql.DB, logf Logger) (int64, error) {
	tx, err := rw.Begin()
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()

	// Every reference to the TEMP table goes through tx, so its whole lifecycle
	// sits on one connection: created here, dropped before COMMIT below, and
	// removed by ROLLBACK on any error path (a TEMP table created inside a
	// transaction does not survive its rollback — verified, not assumed).
	//
	// It has to be tx, not rw. A TEMP table belongs to a single connection and
	// database/sql hands out pooled ones, so an `rw.Exec` drop can land on a
	// different connection and leave the table behind on the one that has it.
	// The leading DROP is still here because this package cannot prove nothing
	// else ever left one on this connection.
	if _, err := tx.Exec(`DROP TABLE IF EXISTS temp.dedup_groups`); err != nil {
		return 0, fmt.Errorf("drop stale dedup_groups: %w", err)
	}
	if _, err := tx.Exec(dupGroupsDDL); err != nil {
		return 0, fmt.Errorf("build dedup_groups: %w", err)
	}
	if _, err := tx.Exec(`CREATE INDEX temp.dedup_groups_key
		ON dedup_groups(transmission_id, observer_idx, p)`); err != nil {
		return 0, fmt.Errorf("index dedup_groups: %w", err)
	}

	// The group keys exist right now and cease to after the delete. Record them
	// before destroying anything: a row count is not enough to reconstruct from
	// if the merge direction is ever wrong again.
	if err := logDuplicateGroups(tx, logf); err != nil {
		return 0, err
	}

	// Only merge columns this database actually has. Apply runs this step
	// before ensureResolvedPathColumn and ensureObservationsRawHexColumn, so on
	// a database old enough to be missing the dedup index, resolved_path and
	// raw_hex may not exist yet either.
	cols, err := existingColumns(tx, "observations", upsertMergedColumns)
	if err != nil {
		return 0, err
	}

	// Replay the UPSERT: for each merged column take the LAST non-NULL value in
	// the group by id. The subquery spans the whole group, the survivor
	// included, so when every row is NULL the result is NULL — which is what
	// the UPSERT would also have left behind.
	for _, c := range cols {
		q := `UPDATE observations SET ` + c + ` = (
			SELECT o2.` + c + ` FROM observations o2
			WHERE o2.transmission_id = observations.transmission_id
			  AND o2.observer_idx = observations.observer_idx
			  AND COALESCE(o2.path_json, '') = COALESCE(observations.path_json, '')
			  AND o2.` + c + ` IS NOT NULL
			ORDER BY o2.id DESC LIMIT 1)
			WHERE id IN (SELECT keep FROM dedup_groups)`
		if _, err := tx.Exec(q); err != nil {
			return 0, fmt.Errorf("merge %s: %w", c, err)
		}
	}

	res, err := tx.Exec(`DELETE FROM observations WHERE id IN (
		SELECT o.id FROM observations o
		JOIN dedup_groups d
		  ON d.transmission_id = o.transmission_id
		 AND d.observer_idx = o.observer_idx
		 AND d.p = COALESCE(o.path_json, '')
		WHERE o.id <> d.keep)`)
	if err != nil {
		return 0, err
	}
	removed, err := res.RowsAffected()
	if err != nil {
		return 0, err
	}

	// Same transaction: if this fails, the deletions above roll back with it.
	if _, err := tx.Exec(`DROP TABLE temp.dedup_groups`); err != nil {
		return 0, fmt.Errorf("drop dedup_groups: %w", err)
	}

	if _, err := tx.Exec(dedupIndexDDL); err != nil {
		return 0, fmt.Errorf("create idx_observations_dedup after collapsing %d row(s): %w", removed, err)
	}
	if err := tx.Commit(); err != nil {
		return 0, err
	}
	return removed, nil
}

// existingColumns filters want down to the columns table actually has, keeping
// the given order.
//
// Takes a Querier, not *sql.DB, and callers inside a transaction must pass the
// transaction. Probing the pool from inside one deadlocks the ingestor, whose
// pool is capped at a single connection: the probe waits for the connection the
// transaction is holding, and nothing ever releases it.
func existingColumns(rw Querier, table string, want []string) ([]string, error) {
	out := make([]string, 0, len(want))
	for _, c := range want {
		ok, err := TableHasColumn(rw, table, c)
		if err != nil {
			return nil, err
		}
		if ok {
			out = append(out, c)
		}
	}
	return out, nil
}

// maxLoggedDupGroups bounds the audit log. A database missing the index for
// long enough can have a great many duplicate groups, and an unbounded dump at
// startup is its own operational problem.
const maxLoggedDupGroups = 20

// logDuplicateGroups records what is about to be collapsed, while dedup_groups
// still holds it. Reads through tx because the TEMP table lives on that
// transaction's connection.
func logDuplicateGroups(tx *sql.Tx, logf Logger) error {
	var groups, extra int64
	if err := tx.QueryRow(`SELECT COUNT(*), COALESCE(SUM(n - 1), 0) FROM dedup_groups`).
		Scan(&groups, &extra); err != nil {
		return fmt.Errorf("count dedup_groups: %w", err)
	}
	logf("[dbschema] %d duplicate observation group(s), %d row(s) to remove", groups, extra)

	rows, err := tx.Query(`SELECT transmission_id, observer_idx, p, keep, n
		FROM dedup_groups ORDER BY keep LIMIT ?`, maxLoggedDupGroups)
	if err != nil {
		return fmt.Errorf("list dedup_groups: %w", err)
	}
	defer rows.Close()
	var listed int64
	for rows.Next() {
		var txID, keep, n int64
		var observerIdx sql.NullInt64
		var pathJSON string
		if err := rows.Scan(&txID, &observerIdx, &pathJSON, &keep, &n); err != nil {
			return fmt.Errorf("scan dedup_groups: %w", err)
		}
		// Print NULL as NULL. Rendering observerIdx.Int64 unconditionally shows
		// a NULL as 0, which is the one value that would hide the grouping bug
		// above from the only person able to notice it.
		idx := "NULL"
		if observerIdx.Valid {
			idx = strconv.FormatInt(observerIdx.Int64, 10)
		}
		logf("[dbschema]   transmission_id=%d observer_idx=%s path_json=%q rows=%d keeping id=%d",
			txID, idx, pathJSON, n, keep)
		listed++
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate dedup_groups: %w", err)
	}
	if groups > listed {
		logf("[dbschema]   ... and %d more group(s) not listed", groups-listed)
	}
	return nil
}