mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-11-04 09:47:21 +00:00
MrAlders0n
16fb706d17
* feat(db): packet_sites table and indexes * feat(ingest): keep each packet's site row in the observation insert * feat(db): packet_sites backfill, reconcile and audit * feat(db): region packet pages from packet_sites when trusted, with an exact page cursor * feat(db): packet_sites front sweeper with plan and liveness metrics * feat: boot repair, reconcile, audit, sweep and first-time verification for packet_sites * fix(db): read no site range when no packet was first heard in the window * fix(db): read the window guard's maxima through indexes * test(planguard): packet_sites statements active, data-derived cases, diff and storediff * docs(api): pageCursor applies to region packet pages * fix(db): drop packet_sites trust again after the unsited fallback write * fix(db): the packet_sites audit refills a pair at its newest hearing * fix(db): bounded region reads fill the page within the request, else today's statement answers * fix(planguard): tie_cut needs both pages at the limit and members of other identities * fix: packet_sites sweep without JIT, one verification VACUUM, read-only checks to the newest id * docs(api): cursor and pageCursor on region pages served without nextPageCursor * test: drop local baselines of entries the copy skips; plain race schedule names * test(planguard): storediff walks of the bounded region shapes; local baselines after the overlay rebuild * fix(db): drop message-site trust again after the unseeded insert * fix(db): a region page cursor's fallback to today's statement continues from its exact position
1257 lines
52 KiB
Go
1257 lines
52 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: packet_sites.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const checkPacketSitesSlice = `-- name: CheckPacketSitesSlice :one
|
|
WITH pk AS MATERIALIZED (
|
|
SELECT DISTINCT po.packet_hash FROM packet_observations po
|
|
WHERE po.id > $1::bigint AND po.id <= $2::bigint
|
|
), def AS (
|
|
SELECT po.packet_hash, po.iata, max(po.heard_at) AS site_heard_at,
|
|
p.first_heard_at, p.payload_type, p.route_type, p.scope_id
|
|
FROM pk
|
|
JOIN packet_observations po ON po.packet_hash = pk.packet_hash
|
|
JOIN packets p ON p.packet_hash = pk.packet_hash
|
|
GROUP BY po.packet_hash, po.iata, p.first_heard_at, p.payload_type, p.route_type, p.scope_id
|
|
), have AS (
|
|
SELECT ps.packet_hash, ps.iata, ps.site_heard_at, ps.first_heard_at, ps.payload_type, ps.route_type, ps.scope_id
|
|
FROM pk JOIN packet_sites ps ON ps.packet_hash = pk.packet_hash
|
|
)
|
|
SELECT
|
|
(SELECT count(*) FROM def d WHERE NOT EXISTS
|
|
(SELECT 1 FROM have h WHERE h.packet_hash = d.packet_hash AND h.iata = d.iata))::bigint AS missing,
|
|
(SELECT count(*) FROM def d JOIN have h ON h.packet_hash = d.packet_hash AND h.iata = d.iata
|
|
WHERE (h.site_heard_at, h.first_heard_at, h.payload_type, h.route_type, h.scope_id)
|
|
IS DISTINCT FROM (d.site_heard_at, d.first_heard_at, d.payload_type, d.route_type, d.scope_id))::bigint AS wrong,
|
|
(SELECT count(*) FROM have h WHERE NOT EXISTS
|
|
(SELECT 1 FROM def d WHERE d.packet_hash = h.packet_hash AND d.iata = h.iata))::bigint AS extra
|
|
`
|
|
|
|
type CheckPacketSitesSliceParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
type CheckPacketSitesSliceRow struct {
|
|
Missing int64 `json:"missing"`
|
|
Wrong int64 `json:"wrong"`
|
|
Extra int64 `json:"extra"`
|
|
}
|
|
|
|
// Exact check of every packet with an observation in (after_id, through_id]: site rows the observations imply
|
|
// and the table lacks, rows whose hearing time or packet columns differ, and rows no observation backs.
|
|
func (q *Queries) CheckPacketSitesSlice(ctx context.Context, arg CheckPacketSitesSliceParams) (CheckPacketSitesSliceRow, error) {
|
|
row := q.db.QueryRow(ctx, checkPacketSitesSlice, arg.AfterID, arg.ThroughID)
|
|
var i CheckPacketSitesSliceRow
|
|
err := row.Scan(&i.Missing, &i.Wrong, &i.Extra)
|
|
return i, err
|
|
}
|
|
|
|
const fillPacketSitesAudit = `-- name: FillPacketSitesAudit :execrows
|
|
INSERT INTO packet_sites (packet_hash, iata, site_heard_at, first_heard_at, payload_type, route_type, scope_id)
|
|
SELECT g.packet_hash, g.iata, g.mx, p.first_heard_at, p.payload_type, p.route_type, p.scope_id
|
|
FROM (
|
|
SELECT po.packet_hash, po.iata, max(po.heard_at) AS mx
|
|
FROM (SELECT DISTINCT s.packet_hash, s.iata FROM packet_observations s
|
|
WHERE s.id > $1::bigint AND s.id <= $2::bigint) t
|
|
JOIN packet_observations po ON po.packet_hash = t.packet_hash AND po.iata = t.iata
|
|
GROUP BY po.packet_hash, po.iata
|
|
) g
|
|
JOIN packets p ON p.packet_hash = g.packet_hash
|
|
WHERE NOT EXISTS (SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = g.packet_hash AND x.iata = g.iata AND x.site_heard_at >= g.mx)
|
|
ON CONFLICT (packet_hash, iata) DO UPDATE SET site_heard_at = EXCLUDED.site_heard_at
|
|
WHERE packet_sites.site_heard_at < EXCLUDED.site_heard_at
|
|
`
|
|
|
|
type FillPacketSitesAuditParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
// The audit's fill: each (packet, site) of the observations in (after_id, through_id], at its latest hearing there
|
|
// among all of the packet's observations, so a pair heard again in a later slice comes back at that hearing.
|
|
func (q *Queries) FillPacketSitesAudit(ctx context.Context, arg FillPacketSitesAuditParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, fillPacketSitesAudit, arg.AfterID, arg.ThroughID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const fillPacketSitesSlice = `-- name: FillPacketSitesSlice :execrows
|
|
|
|
INSERT INTO packet_sites (packet_hash, iata, site_heard_at, first_heard_at, payload_type, route_type, scope_id)
|
|
SELECT g.packet_hash, g.iata, g.mx, g.first_heard_at, g.payload_type, g.route_type, g.scope_id
|
|
FROM (
|
|
SELECT po.packet_hash, po.iata, max(po.heard_at) AS mx, p.first_heard_at, p.payload_type, p.route_type, p.scope_id
|
|
FROM packet_observations po
|
|
JOIN packets p ON p.packet_hash = po.packet_hash
|
|
WHERE po.id > $1::bigint AND po.id <= $2::bigint
|
|
GROUP BY po.packet_hash, po.iata, p.first_heard_at, p.payload_type, p.route_type, p.scope_id
|
|
) g
|
|
WHERE NOT EXISTS (SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = g.packet_hash AND x.iata = g.iata AND x.site_heard_at >= g.mx)
|
|
ON CONFLICT (packet_hash, iata) DO UPDATE SET site_heard_at = EXCLUDED.site_heard_at
|
|
WHERE packet_sites.site_heard_at < EXCLUDED.site_heard_at
|
|
`
|
|
|
|
type FillPacketSitesSliceParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
// Copyright 2026 Beacon Contributors
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
// Each (packet, site) of the observations in (after_id, through_id], at its latest hearing there.
|
|
// Pairs already at or past that hearing are skipped before the insert, so a clean slice writes nothing.
|
|
func (q *Queries) FillPacketSitesSlice(ctx context.Context, arg FillPacketSitesSliceParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, fillPacketSitesSlice, arg.AfterID, arg.ThroughID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const filterKnownSites = `-- name: FilterKnownSites :one
|
|
SELECT COALESCE(array_agg(DISTINCT u ORDER BY u), '{}')::bpchar[] AS iatas
|
|
FROM unnest($1::bpchar[]) AS u
|
|
WHERE EXISTS (SELECT 1 FROM packet_sites ps WHERE ps.iata = u)
|
|
`
|
|
|
|
// The requested sites that have at least one row: a site with none adds no row to any site page.
|
|
func (q *Queries) FilterKnownSites(ctx context.Context, iatas []string) ([]string, error) {
|
|
row := q.db.QueryRow(ctx, filterKnownSites, iatas)
|
|
var iatas_2 []string
|
|
err := row.Scan(&iatas_2)
|
|
return iatas_2, err
|
|
}
|
|
|
|
const listPacketSiteSamples = `-- name: ListPacketSiteSamples :many
|
|
SELECT po.iata, count(DISTINCT po.packet_hash)::bigint AS n
|
|
FROM packet_observations po
|
|
WHERE po.id > $1::bigint
|
|
GROUP BY po.iata
|
|
ORDER BY n DESC, po.iata
|
|
LIMIT $2
|
|
`
|
|
|
|
type ListPacketSiteSamplesParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
RowLimit int32 `json:"row_limit"`
|
|
}
|
|
|
|
type ListPacketSiteSamplesRow struct {
|
|
Iata string `json:"iata"`
|
|
N int64 `json:"n"`
|
|
}
|
|
|
|
// The sites with the most packets among observations above after_id; the first-time verification pages them.
|
|
func (q *Queries) ListPacketSiteSamples(ctx context.Context, arg ListPacketSiteSamplesParams) ([]ListPacketSiteSamplesRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketSiteSamples, arg.AfterID, arg.RowLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketSiteSamplesRow{}
|
|
for rows.Next() {
|
|
var i ListPacketSiteSamplesRow
|
|
if err := rows.Scan(&i.Iata, &i.N); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySite = `-- name: ListPacketsBySite :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT ps.packet_hash, ps.site_heard_at AS k
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = $1::bpchar
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($2::timestamptz, $3::bytea)
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $4
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type ListPacketsBySiteParams struct {
|
|
Iata string `json:"iata"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySiteRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// One site: one range of its (site_heard_at, packet_hash) index.
|
|
func (q *Queries) ListPacketsBySite(ctx context.Context, arg ListPacketsBySiteParams) ([]ListPacketsBySiteRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySite,
|
|
arg.Iata,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySiteRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySiteRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySiteKeys = `-- name: ListPacketsBySiteKeys :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT h.packet_hash, t.k, h.ord
|
|
FROM unnest($1::bytea[]) WITH ORDINALITY AS h(packet_hash, ord)
|
|
JOIN unnest($2::timestamptz[]) WITH ORDINALITY AS t(k, ord) ON t.ord = h.ord
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.ord
|
|
`
|
|
|
|
type ListPacketsBySiteKeysParams struct {
|
|
PacketHashes [][]byte `json:"packet_hashes"`
|
|
SiteHeardAts []pgtype.Timestamptz `json:"site_heard_ats"`
|
|
}
|
|
|
|
type ListPacketsBySiteKeysRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// The page rows for keys from ListPacketsBySitesSince or ListPacketsBySitesScoped, in the given order.
|
|
func (q *Queries) ListPacketsBySiteKeys(ctx context.Context, arg ListPacketsBySiteKeysParams) ([]ListPacketsBySiteKeysRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySiteKeys, arg.PacketHashes, arg.SiteHeardAts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySiteKeysRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySiteKeysRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySites = `-- name: ListPacketsBySites :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.packet_hash, ps.site_heard_at
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($2::timestamptz, $3::bytea)
|
|
AND NOT EXISTS (SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = ps.packet_hash
|
|
AND x.iata = ANY($1::bpchar[])
|
|
AND (x.site_heard_at, x.packet_hash) >= ($2::timestamptz, $3::bytea)
|
|
OFFSET 0) -- a per-row SubPlan in both plan modes, never an anti join
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $4) s
|
|
GROUP BY s.packet_hash
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $4
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type ListPacketsBySitesParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// Two or more sites with a cursor: one range per site; a packet whose newest requested-site row is at or above
|
|
// the cursor was already returned, so the in-range re-check drops it.
|
|
func (q *Queries) ListPacketsBySites(ctx context.Context, arg ListPacketsBySitesParams) ([]ListPacketsBySitesRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySites,
|
|
arg.Iatas,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySitesFirst = `-- name: ListPacketsBySitesFirst :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.packet_hash, ps.site_heard_at
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($2::timestamptz, $3::bytea)
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $4) s
|
|
GROUP BY s.packet_hash
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $4
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type ListPacketsBySitesFirstParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesFirstRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// ListPacketsBySites for the first page or one site, where the re-check cannot match.
|
|
func (q *Queries) ListPacketsBySitesFirst(ctx context.Context, arg ListPacketsBySitesFirstParams) ([]ListPacketsBySitesFirstRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySitesFirst,
|
|
arg.Iatas,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesFirstRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesFirstRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySitesScoped = `-- name: ListPacketsBySitesScoped :many
|
|
WITH scanned AS MATERIALIZED (
|
|
SELECT w.iata, w.scope_id, w.packet_hash, w.site_heard_at, w.first_heard_at, w.payload_type, w.route_type
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN (SELECT id FROM transport_scopes WHERE name = ANY($2::text[])) sc
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.iata, ps.scope_id, ps.packet_hash, ps.site_heard_at, ps.first_heard_at, ps.payload_type, ps.route_type
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata AND ps.scope_id = sc.id
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($3::timestamptz, $4::bytea)
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $5) w
|
|
-- No packet was first heard at or after since: nothing can match, so no range is read. A packet's last_heard_at
|
|
-- is at or after its first_heard_at, except the rows idx_packets_heard_inverted holds.
|
|
-- Both maxima are one index probe each.
|
|
WHERE (SELECT max(g.last_heard_at) FROM packets g) >= $6::timestamptz
|
|
OR (SELECT max(g.first_heard_at) FROM packets g WHERE (g.first_heard_at > g.last_heard_at) IS TRUE) >= $6::timestamptz
|
|
), floor AS (
|
|
-- Every range that filled scan_depth has unread entries below its last one; the highest such entry bounds
|
|
-- what this page knows exactly.
|
|
SELECT f.site_heard_at, f.packet_hash
|
|
FROM (SELECT DISTINCT ON (w.iata, w.scope_id) w.site_heard_at, w.packet_hash
|
|
FROM scanned w
|
|
WHERE (w.iata, w.scope_id) IN (SELECT w.iata, w.scope_id FROM scanned w GROUP BY w.iata, w.scope_id HAVING count(*) >= $5)
|
|
ORDER BY w.iata, w.scope_id, w.site_heard_at, w.packet_hash) f
|
|
ORDER BY f.site_heard_at DESC, f.packet_hash DESC
|
|
LIMIT 1
|
|
), page AS (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM scanned s
|
|
WHERE s.first_heard_at >= $6::timestamptz AND s.first_heard_at <= $7::timestamptz
|
|
AND s.payload_type = ANY($8::smallint[]) AND s.route_type = ANY($9::smallint[])
|
|
GROUP BY s.packet_hash
|
|
HAVING NOT EXISTS (SELECT 1 FROM floor f WHERE (max(s.site_heard_at), s.packet_hash) < (f.site_heard_at, f.packet_hash))
|
|
AND (NOT $10::boolean OR NOT EXISTS (
|
|
SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = s.packet_hash
|
|
AND x.iata = ANY($1::bpchar[])
|
|
AND (x.site_heard_at, x.packet_hash) >= ($3::timestamptz, $4::bytea)))
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $11
|
|
)
|
|
SELECT false AS is_floor, pg.packet_hash, pg.k::timestamptz AS site_heard_at FROM page pg
|
|
UNION ALL
|
|
SELECT true, f.packet_hash, f.site_heard_at FROM floor f
|
|
ORDER BY 1, 3 DESC, 2 DESC
|
|
`
|
|
|
|
type ListPacketsBySitesScopedParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
ScopeNames []string `json:"scope_names"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
ScanDepth int32 `json:"scan_depth"`
|
|
SinceTs pgtype.Timestamptz `json:"since_ts"`
|
|
UntilTs pgtype.Timestamptz `json:"until_ts"`
|
|
PayloadTypes []int16 `json:"payload_types"`
|
|
RouteTypes []int16 `json:"route_types"`
|
|
Recheck bool `json:"recheck"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesScopedRow struct {
|
|
IsFloor bool `json:"is_floor"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
}
|
|
|
|
// ListPacketsBySitesSince with one range per (site, named scope).
|
|
func (q *Queries) ListPacketsBySitesScoped(ctx context.Context, arg ListPacketsBySitesScopedParams) ([]ListPacketsBySitesScopedRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySitesScoped,
|
|
arg.Iatas,
|
|
arg.ScopeNames,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.ScanDepth,
|
|
arg.SinceTs,
|
|
arg.UntilTs,
|
|
arg.PayloadTypes,
|
|
arg.RouteTypes,
|
|
arg.Recheck,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesScopedRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesScopedRow
|
|
if err := rows.Scan(&i.IsFloor, &i.PacketHash, &i.SiteHeardAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySitesSince = `-- name: ListPacketsBySitesSince :many
|
|
WITH scanned AS MATERIALIZED (
|
|
SELECT w.iata, w.scope_id, w.packet_hash, w.site_heard_at, w.first_heard_at, w.payload_type, w.route_type
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.iata, ps.scope_id, ps.packet_hash, ps.site_heard_at, ps.first_heard_at, ps.payload_type, ps.route_type
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($2::timestamptz, $3::bytea)
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $4) w
|
|
-- No packet was first heard at or after since: nothing can match, so no range is read. A packet's last_heard_at
|
|
-- is at or after its first_heard_at, except the rows idx_packets_heard_inverted holds.
|
|
-- Both maxima are one index probe each.
|
|
WHERE (SELECT max(g.last_heard_at) FROM packets g) >= $5::timestamptz
|
|
OR (SELECT max(g.first_heard_at) FROM packets g WHERE (g.first_heard_at > g.last_heard_at) IS TRUE) >= $5::timestamptz
|
|
), floor AS (
|
|
-- Every range that filled scan_depth has unread entries below its last one; the highest such entry bounds
|
|
-- what this page knows exactly.
|
|
SELECT f.site_heard_at, f.packet_hash
|
|
FROM (SELECT DISTINCT ON (w.iata) w.site_heard_at, w.packet_hash
|
|
FROM scanned w
|
|
WHERE (w.iata) IN (SELECT w.iata FROM scanned w GROUP BY w.iata HAVING count(*) >= $4)
|
|
ORDER BY w.iata, w.site_heard_at, w.packet_hash) f
|
|
ORDER BY f.site_heard_at DESC, f.packet_hash DESC
|
|
LIMIT 1
|
|
), page AS (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM scanned s
|
|
WHERE s.first_heard_at >= $5::timestamptz AND s.first_heard_at <= $6::timestamptz
|
|
AND s.payload_type = ANY($7::smallint[]) AND s.route_type = ANY($8::smallint[])
|
|
GROUP BY s.packet_hash
|
|
HAVING NOT EXISTS (SELECT 1 FROM floor f WHERE (max(s.site_heard_at), s.packet_hash) < (f.site_heard_at, f.packet_hash))
|
|
AND (NOT $9::boolean OR NOT EXISTS (
|
|
SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = s.packet_hash
|
|
AND x.iata = ANY($1::bpchar[])
|
|
AND (x.site_heard_at, x.packet_hash) >= ($2::timestamptz, $3::bytea)))
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $10
|
|
)
|
|
SELECT false AS is_floor, pg.packet_hash, pg.k::timestamptz AS site_heard_at FROM page pg
|
|
UNION ALL
|
|
SELECT true, f.packet_hash, f.site_heard_at FROM floor f
|
|
ORDER BY 1, 3 DESC, 2 DESC
|
|
`
|
|
|
|
type ListPacketsBySitesSinceParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
ScanDepth int32 `json:"scan_depth"`
|
|
SinceTs pgtype.Timestamptz `json:"since_ts"`
|
|
UntilTs pgtype.Timestamptz `json:"until_ts"`
|
|
PayloadTypes []int16 `json:"payload_types"`
|
|
RouteTypes []int16 `json:"route_types"`
|
|
Recheck bool `json:"recheck"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesSinceRow struct {
|
|
IsFloor bool `json:"is_floor"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
}
|
|
|
|
// Page keys for a first_heard_at window or many type pairs (absent window sides are -infinity and infinity).
|
|
// Each site range is read to at most scan_depth entries, as deep as the observation statement reads, and the
|
|
// window and type tests filter what was read; the last row (is_floor) is the floor below which this page is
|
|
// incomplete, when a range filled scan_depth. Rows below the floor are left to the next page.
|
|
func (q *Queries) ListPacketsBySitesSince(ctx context.Context, arg ListPacketsBySitesSinceParams) ([]ListPacketsBySitesSinceRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySitesSince,
|
|
arg.Iatas,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.ScanDepth,
|
|
arg.SinceTs,
|
|
arg.UntilTs,
|
|
arg.PayloadTypes,
|
|
arg.RouteTypes,
|
|
arg.Recheck,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesSinceRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesSinceRow
|
|
if err := rows.Scan(&i.IsFloor, &i.PacketHash, &i.SiteHeardAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySitesTyped = `-- name: ListPacketsBySitesTyped :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN unnest($2::smallint[]) AS pt(t)
|
|
CROSS JOIN unnest($3::smallint[]) AS rt(t)
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.packet_hash, ps.site_heard_at
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata AND ps.payload_type = pt.t AND ps.route_type = rt.t
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($4::timestamptz, $5::bytea)
|
|
AND NOT EXISTS (SELECT 1 FROM packet_sites x
|
|
WHERE x.packet_hash = ps.packet_hash
|
|
AND x.iata = ANY($1::bpchar[])
|
|
AND (x.site_heard_at, x.packet_hash) >= ($4::timestamptz, $5::bytea)
|
|
OFFSET 0) -- a per-row SubPlan in both plan modes, never an anti join
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $6) s
|
|
GROUP BY s.packet_hash
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $6
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type ListPacketsBySitesTypedParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
PayloadTypes []int16 `json:"payload_types"`
|
|
RouteTypes []int16 `json:"route_types"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesTypedRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// One range per (site, payload type, route type); Go passes each type list deduplicated and sends only a few
|
|
// type pairs here.
|
|
func (q *Queries) ListPacketsBySitesTyped(ctx context.Context, arg ListPacketsBySitesTypedParams) ([]ListPacketsBySitesTypedRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySitesTyped,
|
|
arg.Iatas,
|
|
arg.PayloadTypes,
|
|
arg.RouteTypes,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesTypedRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesTypedRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPacketsBySitesTypedFirst = `-- name: ListPacketsBySitesTypedFirst :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT s.packet_hash, max(s.site_heard_at) AS k
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN unnest($2::smallint[]) AS pt(t)
|
|
CROSS JOIN unnest($3::smallint[]) AS rt(t)
|
|
CROSS JOIN LATERAL (
|
|
SELECT ps.packet_hash, ps.site_heard_at
|
|
FROM packet_sites ps
|
|
WHERE ps.iata = req.iata AND ps.payload_type = pt.t AND ps.route_type = rt.t
|
|
AND (ps.site_heard_at, ps.packet_hash) < ($4::timestamptz, $5::bytea)
|
|
ORDER BY ps.site_heard_at DESC, ps.packet_hash DESC
|
|
LIMIT $6) s
|
|
GROUP BY s.packet_hash
|
|
ORDER BY max(s.site_heard_at) DESC, s.packet_hash DESC
|
|
LIMIT $6
|
|
)
|
|
SELECT p.packet_hash, p.payload_type,
|
|
COALESCE(CASE
|
|
WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
|
|
THEN p.parsed_payload #>> '{appData,name}'
|
|
WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK'
|
|
AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string'
|
|
AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum')
|
|
WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING')
|
|
AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string'
|
|
AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$'
|
|
THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag')
|
|
END, '')::text AS summary,
|
|
p.route_type, p.first_heard_at, p.last_heard_at, p.scope_id, ts.name AS scope_name,
|
|
pg.k::timestamptz AS site_heard_at,
|
|
p.observation_count AS observation_count,
|
|
-- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL.
|
|
po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name,
|
|
COALESCE(po.iata, ''::bpchar) AS latest_observer_iata,
|
|
COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte,
|
|
COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size,
|
|
COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count,
|
|
po.path_bytes AS latest_observer_path_bytes, p.raw_payload, p.origin_pubkey
|
|
FROM page pg
|
|
JOIN packets p ON p.packet_hash = pg.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
|
|
FROM packet_observations WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC, id DESC LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type ListPacketsBySitesTypedFirstParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
PayloadTypes []int16 `json:"payload_types"`
|
|
RouteTypes []int16 `json:"route_types"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListPacketsBySitesTypedFirstRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
PayloadType int16 `json:"payload_type"`
|
|
Summary string `json:"summary"`
|
|
RouteType int16 `json:"route_type"`
|
|
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
|
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
ScopeName *string `json:"scope_name"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
|
LatestObserverName *string `json:"latest_observer_name"`
|
|
LatestObserverIata string `json:"latest_observer_iata"`
|
|
LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"`
|
|
LatestObserverHashSize int16 `json:"latest_observer_hash_size"`
|
|
LatestObserverHopCount int16 `json:"latest_observer_hop_count"`
|
|
LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"`
|
|
RawPayload []byte `json:"raw_payload"`
|
|
OriginPubkey []byte `json:"origin_pubkey"`
|
|
}
|
|
|
|
// ListPacketsBySitesTyped for the first page or one site.
|
|
func (q *Queries) ListPacketsBySitesTypedFirst(ctx context.Context, arg ListPacketsBySitesTypedFirstParams) ([]ListPacketsBySitesTypedFirstRow, error) {
|
|
rows, err := q.db.Query(ctx, listPacketsBySitesTypedFirst,
|
|
arg.Iatas,
|
|
arg.PayloadTypes,
|
|
arg.RouteTypes,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListPacketsBySitesTypedFirstRow{}
|
|
for rows.Next() {
|
|
var i ListPacketsBySitesTypedFirstRow
|
|
if err := rows.Scan(
|
|
&i.PacketHash,
|
|
&i.PayloadType,
|
|
&i.Summary,
|
|
&i.RouteType,
|
|
&i.FirstHeardAt,
|
|
&i.LastHeardAt,
|
|
&i.ScopeID,
|
|
&i.ScopeName,
|
|
&i.SiteHeardAt,
|
|
&i.ObservationCount,
|
|
&i.LatestObserverID,
|
|
&i.LatestObserverName,
|
|
&i.LatestObserverIata,
|
|
&i.LatestObserverPathLengthByte,
|
|
&i.LatestObserverHashSize,
|
|
&i.LatestObserverHopCount,
|
|
&i.LatestObserverPathBytes,
|
|
&i.RawPayload,
|
|
&i.OriginPubkey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const minObservationID = `-- name: MinObservationID :one
|
|
SELECT COALESCE(min(id), 0)::bigint AS id FROM packet_observations
|
|
`
|
|
|
|
func (q *Queries) MinObservationID(ctx context.Context) (int64, error) {
|
|
row := q.db.QueryRow(ctx, minObservationID)
|
|
var id int64
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
const packetSitesTotalBytes = `-- name: PacketSitesTotalBytes :one
|
|
SELECT pg_total_relation_size('packet_sites'::regclass)::bigint AS bytes
|
|
`
|
|
|
|
func (q *Queries) PacketSitesTotalBytes(ctx context.Context) (int64, error) {
|
|
row := q.db.QueryRow(ctx, packetSitesTotalBytes)
|
|
var bytes int64
|
|
err := row.Scan(&bytes)
|
|
return bytes, err
|
|
}
|
|
|
|
const sweepPacketSitesFronts = `-- name: SweepPacketSitesFronts :one
|
|
WITH site AS MATERIALIZED (
|
|
SELECT ps.iata, ps.payload_type, ps.route_type, ps.scope_id
|
|
FROM iata_codes c
|
|
CROSS JOIN LATERAL (
|
|
SELECT p.iata, p.payload_type, p.route_type, p.scope_id
|
|
FROM packet_sites p
|
|
WHERE p.iata = c.iata AND p.site_heard_at > now() - $1::interval
|
|
ORDER BY p.site_heard_at DESC LIMIT 100000) ps -- idx_packet_sites_site, Index Scan
|
|
), typ AS (
|
|
SELECT count(*) AS n
|
|
FROM (SELECT DISTINCT iata, payload_type, route_type FROM site) f
|
|
CROSS JOIN LATERAL (
|
|
SELECT 1 FROM packet_sites p
|
|
WHERE p.iata = f.iata AND p.payload_type = f.payload_type AND p.route_type = f.route_type
|
|
AND p.site_heard_at > now() - $1::interval
|
|
ORDER BY p.site_heard_at DESC LIMIT 100000) x -- idx_packet_sites_site_type, Index Only Scan
|
|
), scp AS (
|
|
SELECT count(*) AS n
|
|
FROM (SELECT DISTINCT iata, scope_id FROM site WHERE scope_id IS NOT NULL) f
|
|
CROSS JOIN LATERAL (
|
|
SELECT 1 FROM packet_sites p
|
|
WHERE p.iata = f.iata AND p.scope_id = f.scope_id AND p.scope_id IS NOT NULL
|
|
AND p.site_heard_at > now() - $1::interval
|
|
ORDER BY p.site_heard_at DESC LIMIT 100000) x -- idx_packet_sites_site_scope, Index Only Scan
|
|
)
|
|
SELECT (SELECT count(*) FROM site) AS site_live, (SELECT n FROM typ) AS type_live, (SELECT n FROM scp) AS scope_live
|
|
`
|
|
|
|
type SweepPacketSitesFrontsRow struct {
|
|
SiteLive int64 `json:"site_live"`
|
|
TypeLive int64 `json:"type_live"`
|
|
ScopeLive int64 `json:"scope_live"`
|
|
}
|
|
|
|
// Read-only. Visits, through each time-ordered packet_sites index, every entry whose key was written in the last
|
|
// lookback (2 minutes). An index scan that reaches an entry whose heap tuple is dead to every snapshot sets the
|
|
// entry's LP_DEAD hint; when that leaf later fills, the inserting backend deletes hinted entries instead of splitting
|
|
// (nbtree simple deletion). Keeps the three indexes at about 2x a fresh build instead of 4-9x. Changes no row.
|
|
// The ORDER BY ... LIMIT keeps every lateral a per-front index range scan (never flattened into a join).
|
|
// The cap is far above any front's 2-minute volume (the busiest site: about 150 entries) and only bounds the gauge.
|
|
func (q *Queries) SweepPacketSitesFronts(ctx context.Context, lookback pgtype.Interval) (SweepPacketSitesFrontsRow, error) {
|
|
row := q.db.QueryRow(ctx, sweepPacketSitesFronts, lookback)
|
|
var i SweepPacketSitesFrontsRow
|
|
err := row.Scan(&i.SiteLive, &i.TypeLive, &i.ScopeLive)
|
|
return i, err
|
|
}
|
|
|
|
const truthPacketsBySitesPage = `-- name: TruthPacketsBySitesPage :many
|
|
WITH page AS MATERIALIZED (
|
|
SELECT po.packet_hash, max(po.heard_at) AS k
|
|
FROM packet_observations po
|
|
WHERE po.iata = ANY($1::bpchar[]) AND po.heard_at >= $2::timestamptz
|
|
GROUP BY po.packet_hash
|
|
HAVING (max(po.heard_at), po.packet_hash) < ($3::timestamptz, $4::bytea)
|
|
ORDER BY max(po.heard_at) DESC, po.packet_hash DESC
|
|
LIMIT $5
|
|
)
|
|
SELECT pg.packet_hash, pg.k::timestamptz AS site_heard_at FROM page pg ORDER BY pg.k DESC, pg.packet_hash DESC
|
|
`
|
|
|
|
type TruthPacketsBySitesPageParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
FloorTs pgtype.Timestamptz `json:"floor_ts"`
|
|
CursorTs pgtype.Timestamptz `json:"cursor_ts"`
|
|
CursorHash []byte `json:"cursor_hash"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type TruthPacketsBySitesPageRow struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"`
|
|
}
|
|
|
|
// Page keys straight from the observations, independent of packet_sites. floor_ts bounds the read: keys at or
|
|
// above it are exact, because a packet's newest requested-site hearing is unchanged by dropping older ones.
|
|
func (q *Queries) TruthPacketsBySitesPage(ctx context.Context, arg TruthPacketsBySitesPageParams) ([]TruthPacketsBySitesPageRow, error) {
|
|
rows, err := q.db.Query(ctx, truthPacketsBySitesPage,
|
|
arg.Iatas,
|
|
arg.FloorTs,
|
|
arg.CursorTs,
|
|
arg.CursorHash,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []TruthPacketsBySitesPageRow{}
|
|
for rows.Next() {
|
|
var i TruthPacketsBySitesPageRow
|
|
if err := rows.Scan(&i.PacketHash, &i.SiteHeardAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|