feat(db): add batched presence touch queries

single-statement flush targets for coalesced last_seen/last_heard_at bumps

db/presence.go

@@ -0,0 +1,44 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"context"
+	"time"
+
+	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+func (s *Store) TouchObservers(ctx context.Context, ids []uuid.UUID, seen []time.Time, counts []int32) error {
+	return s.q.TouchObservers(ctx, sqlc.TouchObserversParams{
+		Column1: ids,
+		Column2: toTimestamptzs(seen),
+		Column3: counts,
+	})
+}
+
+func (s *Store) TouchObserverBrokers(ctx context.Context, ids []uuid.UUID, brokers []string, seen []time.Time) error {
+	return s.q.TouchObserverBrokers(ctx, sqlc.TouchObserverBrokersParams{
+		Column1: ids,
+		Column2: brokers,
+		Column3: toTimestamptzs(seen),
+	})
+}
+
+func (s *Store) TouchPackets(ctx context.Context, hashes [][]byte, heard []time.Time) error {
+	return s.q.TouchPackets(ctx, sqlc.TouchPacketsParams{
+		Column1: hashes,
+		Column2: toTimestamptzs(heard),
+	})
+}
+
+func toTimestamptzs(ts []time.Time) []pgtype.Timestamptz {
+	out := make([]pgtype.Timestamptz, len(ts))
+	for i, t := range ts {
+		out[i] = pgtype.Timestamptz{Time: t, Valid: true}
+	}
+	return out
+}

db/queries/queries.sql

@@ -109,6 +109,19 @@ UPDATE observers SET
 WHERE public_key = $1
 RETURNING id;
 
+-- name: TouchObservers :exec
+-- Batched flush of coalesced presence bumps. GREATEST keeps a late flush
+-- from regressing a newer write-through (e.g. a status update).
+UPDATE observers o SET
+  last_seen         = GREATEST(o.last_seen, v.seen),
+  observation_count = COALESCE(o.observation_count, 0) + v.delta
+FROM (
+  SELECT unnest($1::uuid[]) AS id,
+         unnest($2::timestamptz[]) AS seen,
+         unnest($3::int[]) AS delta
+) v
+WHERE o.id = v.id;
+
 -- name: UpsertObserverScope :exec
 INSERT INTO observer_scopes (observer_id, scope_id, last_seen)
 VALUES ($1, $2, NOW())
@@ -270,6 +283,17 @@ ON CONFLICT (observer_id, broker_name) DO UPDATE SET
   last_seen      = NOW(),
   last_packet_at = NOW();
 
+-- name: TouchObserverBrokers :exec
+UPDATE observer_brokers ob SET
+  last_seen      = GREATEST(ob.last_seen, v.seen),
+  last_packet_at = GREATEST(ob.last_packet_at, v.seen)
+FROM (
+  SELECT unnest($1::uuid[]) AS observer_id,
+         unnest($2::text[]) AS broker_name,
+         unnest($3::timestamptz[]) AS seen
+) v
+WHERE ob.observer_id = v.observer_id AND ob.broker_name = v.broker_name;
+
 -- ============================================================
 -- PACKETS
 -- ============================================================
@@ -303,6 +327,15 @@ AS inserted;
 -- name: SetPacketDecrypted :exec
 UPDATE packets SET decrypted = true WHERE packet_hash = $1;
 
+-- name: TouchPackets :exec
+UPDATE packets p SET
+  last_heard_at = GREATEST(p.last_heard_at, v.heard)
+FROM (
+  SELECT unnest($1::bytea[]) AS packet_hash,
+         unnest($2::timestamptz[]) AS heard
+) v
+WHERE p.packet_hash = v.packet_hash;
+
 -- name: GetPacketByHash :one
 SELECT p.*, ts.name AS scope_name,
     cm.sender_name AS cm_sender_name,

db/sqlc/mock/querier.go

@@ -1036,6 +1036,48 @@ func (mr *MockQuerierMockRecorder) SetPacketDecrypted(ctx, packetHash any) *gomo
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetPacketDecrypted", reflect.TypeOf((*MockQuerier)(nil).SetPacketDecrypted), ctx, packetHash)
 }
 
+// TouchObserverBrokers mocks base method.
+func (m *MockQuerier) TouchObserverBrokers(ctx context.Context, arg db.TouchObserverBrokersParams) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TouchObserverBrokers", ctx, arg)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TouchObserverBrokers indicates an expected call of TouchObserverBrokers.
+func (mr *MockQuerierMockRecorder) TouchObserverBrokers(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TouchObserverBrokers", reflect.TypeOf((*MockQuerier)(nil).TouchObserverBrokers), ctx, arg)
+}
+
+// TouchObservers mocks base method.
+func (m *MockQuerier) TouchObservers(ctx context.Context, arg db.TouchObserversParams) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TouchObservers", ctx, arg)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TouchObservers indicates an expected call of TouchObservers.
+func (mr *MockQuerierMockRecorder) TouchObservers(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TouchObservers", reflect.TypeOf((*MockQuerier)(nil).TouchObservers), ctx, arg)
+}
+
+// TouchPackets mocks base method.
+func (m *MockQuerier) TouchPackets(ctx context.Context, arg db.TouchPacketsParams) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TouchPackets", ctx, arg)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TouchPackets indicates an expected call of TouchPackets.
+func (mr *MockQuerierMockRecorder) TouchPackets(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TouchPackets", reflect.TypeOf((*MockQuerier)(nil).TouchPackets), ctx, arg)
+}
+
 // UpdateObserverStatus mocks base method.
 func (m *MockQuerier) UpdateObserverStatus(ctx context.Context, arg db.UpdateObserverStatusParams) (uuid.UUID, error) {
 	m.ctrl.T.Helper()

db/sqlc/querier.go

@@ -140,6 +140,11 @@ type Querier interface {
 	SetNodeMultibytePaths(ctx context.Context, id uuid.UUID) error
 	SetNodeMultibyteTraces(ctx context.Context, id uuid.UUID) error
 	SetPacketDecrypted(ctx context.Context, packetHash []byte) error
+	TouchObserverBrokers(ctx context.Context, arg TouchObserverBrokersParams) error
+	// Batched flush of coalesced presence bumps. GREATEST keeps a late flush
+	// from regressing a newer write-through (e.g. a status update).
+	TouchObservers(ctx context.Context, arg TouchObserversParams) error
+	TouchPackets(ctx context.Context, arg TouchPacketsParams) error
 	UpdateObserverStatus(ctx context.Context, arg UpdateObserverStatusParams) (uuid.UUID, error)
 	// ============================================================
 	// CHANNELS

db/sqlc/queries.sql.go

@@ -2959,6 +2959,74 @@ func (q *Queries) SetPacketDecrypted(ctx context.Context, packetHash []byte) err
 	return err
 }
 
+const touchObserverBrokers = `-- name: TouchObserverBrokers :exec
+UPDATE observer_brokers ob SET
+  last_seen      = GREATEST(ob.last_seen, v.seen),
+  last_packet_at = GREATEST(ob.last_packet_at, v.seen)
+FROM (
+  SELECT unnest($1::uuid[]) AS observer_id,
+         unnest($2::text[]) AS broker_name,
+         unnest($3::timestamptz[]) AS seen
+) v
+WHERE ob.observer_id = v.observer_id AND ob.broker_name = v.broker_name
+`
+
+type TouchObserverBrokersParams struct {
+	Column1 []uuid.UUID          `json:"column_1"`
+	Column2 []string             `json:"column_2"`
+	Column3 []pgtype.Timestamptz `json:"column_3"`
+}
+
+func (q *Queries) TouchObserverBrokers(ctx context.Context, arg TouchObserverBrokersParams) error {
+	_, err := q.db.Exec(ctx, touchObserverBrokers, arg.Column1, arg.Column2, arg.Column3)
+	return err
+}
+
+const touchObservers = `-- name: TouchObservers :exec
+UPDATE observers o SET
+  last_seen         = GREATEST(o.last_seen, v.seen),
+  observation_count = COALESCE(o.observation_count, 0) + v.delta
+FROM (
+  SELECT unnest($1::uuid[]) AS id,
+         unnest($2::timestamptz[]) AS seen,
+         unnest($3::int[]) AS delta
+) v
+WHERE o.id = v.id
+`
+
+type TouchObserversParams struct {
+	Column1 []uuid.UUID          `json:"column_1"`
+	Column2 []pgtype.Timestamptz `json:"column_2"`
+	Column3 []int32              `json:"column_3"`
+}
+
+// Batched flush of coalesced presence bumps. GREATEST keeps a late flush
+// from regressing a newer write-through (e.g. a status update).
+func (q *Queries) TouchObservers(ctx context.Context, arg TouchObserversParams) error {
+	_, err := q.db.Exec(ctx, touchObservers, arg.Column1, arg.Column2, arg.Column3)
+	return err
+}
+
+const touchPackets = `-- name: TouchPackets :exec
+UPDATE packets p SET
+  last_heard_at = GREATEST(p.last_heard_at, v.heard)
+FROM (
+  SELECT unnest($1::bytea[]) AS packet_hash,
+         unnest($2::timestamptz[]) AS heard
+) v
+WHERE p.packet_hash = v.packet_hash
+`
+
+type TouchPacketsParams struct {
+	Column1 [][]byte             `json:"column_1"`
+	Column2 []pgtype.Timestamptz `json:"column_2"`
+}
+
+func (q *Queries) TouchPackets(ctx context.Context, arg TouchPacketsParams) error {
+	_, err := q.db.Exec(ctx, touchPackets, arg.Column1, arg.Column2)
+	return err
+}
+
 const updateObserverStatus = `-- name: UpdateObserverStatus :one
 UPDATE observers SET
   display_name     = COALESCE(NULLIF($2, ''), display_name),