diff --git a/db/presence.go b/db/presence.go new file mode 100644 index 0000000..ea2d6fe --- /dev/null +++ b/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 +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index a9dabe2..ef18ee5 100644 --- a/db/queries/queries.sql +++ b/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, diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index de3bda4..58cb8d3 100644 --- a/db/sqlc/mock/querier.go +++ b/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() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 42a8426..9e3b903 100644 --- a/db/sqlc/querier.go +++ b/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 diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 11c5099..c31ce17 100644 --- a/db/sqlc/queries.sql.go +++ b/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),