mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-11-12 11:17:55 +00:00
MrAlders0n
5ded8d925f
* feat(db): channel_message_iatas table and indexes * feat(ingest): record channel message sites at insert and on every later hearing * feat(db): channel message site backfill and two-range reconcile * feat(db): serve region and scope message pages from the site read model when trusted * feat: boot repair, periodic reconcile and first-time verification for the message site read model * test(planguard): channel message site statements, diff and storediff cases * feat(background): log when a read-only server trusts the message site read model * test(planguard): storediff vars in query values, and message cases read from the data * fix(ingest): record a stored hearing's channel message site right after the observation insert * fix: bounded message dispatch for filtered all-channel pages, fresher read-only trust, earlier invalidation * fix: record a hearing's site after its path is marked and skip the backstop when the message was found; read-only trust ends at the writer's deadline
955 lines
33 KiB
Go
955 lines
33 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: channel_message_iatas.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const addChannelMessageIATA = `-- name: AddChannelMessageIATA :one
|
|
|
|
WITH m AS (
|
|
SELECT cm.id, cm.channel_id, p.scope_id
|
|
FROM channel_messages cm
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
WHERE cm.packet_hash = $1::bytea
|
|
), added AS (
|
|
INSERT INTO channel_message_iatas (message_id, iata, channel_id, scope_id)
|
|
SELECT m.id, $2::bpchar, m.channel_id, m.scope_id FROM m
|
|
ON CONFLICT (message_id, iata) DO NOTHING
|
|
)
|
|
SELECT EXISTS (SELECT 1 FROM m) AS found
|
|
`
|
|
|
|
type AddChannelMessageIATAParams struct {
|
|
PacketHash []byte `json:"packet_hash"`
|
|
Iata string `json:"iata"`
|
|
}
|
|
|
|
// Copyright 2026 Beacon Contributors
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
// A stored observation's site for a message another statement created. A no-op when the packet has no message;
|
|
// found says whether the packet has a committed message (then the site is recorded, now or before).
|
|
func (q *Queries) AddChannelMessageIATA(ctx context.Context, arg AddChannelMessageIATAParams) (bool, error) {
|
|
row := q.db.QueryRow(ctx, addChannelMessageIATA, arg.PacketHash, arg.Iata)
|
|
var found bool
|
|
err := row.Scan(&found)
|
|
return found, err
|
|
}
|
|
|
|
const checkChannelMessageIATAs = `-- name: CheckChannelMessageIATAs :one
|
|
WITH def AS (
|
|
SELECT DISTINCT cm.id AS message_id, po.iata, cm.channel_id, p.scope_id, po.id <= $1::bigint AS settled
|
|
FROM channel_messages cm
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
WHERE cm.id > $2::bigint AND cm.id <= $3::bigint
|
|
), have AS (
|
|
SELECT i.message_id, i.iata, i.channel_id, i.scope_id
|
|
FROM channel_message_iatas i
|
|
WHERE i.message_id > $2::bigint AND i.message_id <= $3::bigint
|
|
)
|
|
SELECT
|
|
(SELECT count(*) FROM (SELECT message_id, iata, channel_id, scope_id FROM def WHERE settled
|
|
EXCEPT SELECT message_id, iata, channel_id, scope_id FROM have) m)::bigint AS missing,
|
|
(SELECT count(*) FROM (SELECT message_id, iata, channel_id, scope_id FROM have
|
|
EXCEPT SELECT message_id, iata, channel_id, scope_id FROM def) x)::bigint AS extra
|
|
`
|
|
|
|
type CheckChannelMessageIATAsParams struct {
|
|
ObsThrough int64 `json:"obs_through"`
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
type CheckChannelMessageIATAsRow struct {
|
|
Missing int64 `json:"missing"`
|
|
Extra int64 `json:"extra"`
|
|
}
|
|
|
|
// Exact check of messages in (after_id, through_id]: pairs the definition has and the table lacks (counting
|
|
// observations up to obs_through only), and table pairs no stored observation backs.
|
|
func (q *Queries) CheckChannelMessageIATAs(ctx context.Context, arg CheckChannelMessageIATAsParams) (CheckChannelMessageIATAsRow, error) {
|
|
row := q.db.QueryRow(ctx, checkChannelMessageIATAs, arg.ObsThrough, arg.AfterID, arg.ThroughID)
|
|
var i CheckChannelMessageIATAsRow
|
|
err := row.Scan(&i.Missing, &i.Extra)
|
|
return i, err
|
|
}
|
|
|
|
const fillChannelMessageIATAsByMessages = `-- name: FillChannelMessageIATAsByMessages :execrows
|
|
INSERT INTO channel_message_iatas (message_id, iata, channel_id, scope_id)
|
|
SELECT DISTINCT cm.id, po.iata, cm.channel_id, p.scope_id
|
|
FROM channel_messages cm
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
WHERE cm.id > $1::bigint AND cm.id <= $2::bigint
|
|
ON CONFLICT (message_id, iata) DO NOTHING
|
|
`
|
|
|
|
type FillChannelMessageIATAsByMessagesParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
// Every site of every message in (after_id, through_id]: the backfill and the reconcile's message part.
|
|
func (q *Queries) FillChannelMessageIATAsByMessages(ctx context.Context, arg FillChannelMessageIATAsByMessagesParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, fillChannelMessageIATAsByMessages, arg.AfterID, arg.ThroughID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const fillChannelMessageIATAsByObservations = `-- name: FillChannelMessageIATAsByObservations :execrows
|
|
INSERT INTO channel_message_iatas (message_id, iata, channel_id, scope_id)
|
|
SELECT DISTINCT cm.id, po.iata, cm.channel_id, p.scope_id
|
|
FROM packet_observations po
|
|
JOIN channel_messages cm ON cm.packet_hash = po.packet_hash
|
|
JOIN packets p ON p.packet_hash = po.packet_hash
|
|
WHERE po.id > $1::bigint AND po.id <= $2::bigint AND COALESCE(po.payload_type, 5) = 5
|
|
ON CONFLICT (message_id, iata) DO NOTHING
|
|
`
|
|
|
|
type FillChannelMessageIATAsByObservationsParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
ThroughID int64 `json:"through_id"`
|
|
}
|
|
|
|
// The site of every group-text observation in (after_id, through_id] whose packet has a message.
|
|
// An untyped observation is checked too: older rows may have no payload type.
|
|
func (q *Queries) FillChannelMessageIATAsByObservations(ctx context.Context, arg FillChannelMessageIATAsByObservationsParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, fillChannelMessageIATAsByObservations, arg.AfterID, arg.ThroughID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const listAllChannelMessagesByIATAs = `-- name: ListAllChannelMessagesByIATAs :many
|
|
WITH hits AS MATERIALIZED (
|
|
SELECT m.message_id
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT i.message_id FROM channel_message_iatas i
|
|
WHERE i.iata = req.iata AND i.message_id < $2::bigint
|
|
ORDER BY i.message_id DESC
|
|
LIMIT $3) m
|
|
GROUP BY m.message_id
|
|
ORDER BY m.message_id DESC
|
|
LIMIT $3)
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM hits h
|
|
JOIN channel_messages cm ON cm.id = h.message_id
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY cm.id DESC
|
|
`
|
|
|
|
type ListAllChannelMessagesByIATAsParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListAllChannelMessagesByIATAsRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// Every channel filtered by sites: each site's newest message ids from the read model, merged.
|
|
func (q *Queries) ListAllChannelMessagesByIATAs(ctx context.Context, arg ListAllChannelMessagesByIATAsParams) ([]ListAllChannelMessagesByIATAsRow, error) {
|
|
rows, err := q.db.Query(ctx, listAllChannelMessagesByIATAs, arg.Iatas, arg.CursorID, arg.PageLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListAllChannelMessagesByIATAsRow{}
|
|
for rows.Next() {
|
|
var i ListAllChannelMessagesByIATAsRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listAllChannelMessagesPlain = `-- name: ListAllChannelMessagesPlain :many
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.id < $1::bigint
|
|
AND EXISTS (SELECT 1 FROM packet_observations po WHERE po.packet_hash = cm.packet_hash)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $2
|
|
`
|
|
|
|
type ListAllChannelMessagesPlainParams struct {
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListAllChannelMessagesPlainRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// Every channel, unfiltered: a walk by id; a message counts once it has a stored observation.
|
|
func (q *Queries) ListAllChannelMessagesPlain(ctx context.Context, arg ListAllChannelMessagesPlainParams) ([]ListAllChannelMessagesPlainRow, error) {
|
|
rows, err := q.db.Query(ctx, listAllChannelMessagesPlain, arg.CursorID, arg.PageLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListAllChannelMessagesPlainRow{}
|
|
for rows.Next() {
|
|
var i ListAllChannelMessagesPlainRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listAllChannelMessagesPlainScoped = `-- name: ListAllChannelMessagesPlainScoped :many
|
|
WITH hits AS MATERIALIZED (
|
|
SELECT DISTINCT i.message_id
|
|
FROM channel_message_iatas i
|
|
WHERE i.scope_id = $1::int AND i.message_id < $2::bigint
|
|
ORDER BY i.message_id DESC
|
|
LIMIT $3)
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM hits h
|
|
JOIN channel_messages cm ON cm.id = h.message_id
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY cm.id DESC
|
|
`
|
|
|
|
type ListAllChannelMessagesPlainScopedParams struct {
|
|
ScopeID int32 `json:"scope_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListAllChannelMessagesPlainScopedRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// Every channel filtered by scope alone: the scope's newest message ids from the read model.
|
|
func (q *Queries) ListAllChannelMessagesPlainScoped(ctx context.Context, arg ListAllChannelMessagesPlainScopedParams) ([]ListAllChannelMessagesPlainScopedRow, error) {
|
|
rows, err := q.db.Query(ctx, listAllChannelMessagesPlainScoped, arg.ScopeID, arg.CursorID, arg.PageLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListAllChannelMessagesPlainScopedRow{}
|
|
for rows.Next() {
|
|
var i ListAllChannelMessagesPlainScopedRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listAllChannelMessagesSince = `-- name: ListAllChannelMessagesSince :many
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.sent_at >= $1::timestamptz
|
|
AND ($2::int IS NULL OR p.scope_id = $2::int)
|
|
AND cm.id < $3::bigint
|
|
AND EXISTS (SELECT 1 FROM packet_observations po WHERE po.packet_hash = cm.packet_hash)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $4
|
|
`
|
|
|
|
type ListAllChannelMessagesSinceParams struct {
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListAllChannelMessagesSinceRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// Every channel with since, optionally by scope (no sites): a walk by id.
|
|
func (q *Queries) ListAllChannelMessagesSince(ctx context.Context, arg ListAllChannelMessagesSinceParams) ([]ListAllChannelMessagesSinceRow, error) {
|
|
rows, err := q.db.Query(ctx, listAllChannelMessagesSince,
|
|
arg.Since,
|
|
arg.ScopeID,
|
|
arg.CursorID,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListAllChannelMessagesSinceRow{}
|
|
for rows.Next() {
|
|
var i ListAllChannelMessagesSinceRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessageIATASamples = `-- name: ListChannelMessageIATASamples :many
|
|
SELECT i.channel_id, i.iata, count(*)::bigint AS n
|
|
FROM channel_message_iatas i
|
|
WHERE i.message_id > $1::bigint
|
|
GROUP BY i.channel_id, i.iata
|
|
ORDER BY n DESC, i.channel_id, i.iata
|
|
LIMIT $2
|
|
`
|
|
|
|
type ListChannelMessageIATASamplesParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
RowLimit int32 `json:"row_limit"`
|
|
}
|
|
|
|
type ListChannelMessageIATASamplesRow struct {
|
|
ChannelID int32 `json:"channel_id"`
|
|
Iata string `json:"iata"`
|
|
N int64 `json:"n"`
|
|
}
|
|
|
|
// The busiest (channel, site) pairs among recent messages; the first-time verification compares their pages.
|
|
func (q *Queries) ListChannelMessageIATASamples(ctx context.Context, arg ListChannelMessageIATASamplesParams) ([]ListChannelMessageIATASamplesRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessageIATASamples, arg.AfterID, arg.RowLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessageIATASamplesRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessageIATASamplesRow
|
|
if err := rows.Scan(&i.ChannelID, &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 listChannelMessageScopeSamples = `-- name: ListChannelMessageScopeSamples :many
|
|
SELECT i.channel_id, i.scope_id, ts.name, count(*)::bigint AS n
|
|
FROM channel_message_iatas i
|
|
JOIN transport_scopes ts ON ts.id = i.scope_id
|
|
WHERE i.message_id > $1::bigint AND i.scope_id IS NOT NULL
|
|
GROUP BY i.channel_id, i.scope_id, ts.name
|
|
ORDER BY n DESC, i.channel_id, i.scope_id
|
|
LIMIT $2
|
|
`
|
|
|
|
type ListChannelMessageScopeSamplesParams struct {
|
|
AfterID int64 `json:"after_id"`
|
|
RowLimit int32 `json:"row_limit"`
|
|
}
|
|
|
|
type ListChannelMessageScopeSamplesRow struct {
|
|
ChannelID int32 `json:"channel_id"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
Name string `json:"name"`
|
|
N int64 `json:"n"`
|
|
}
|
|
|
|
// The busiest (channel, scope) pairs among recent messages, with the scope name today's statement filters on.
|
|
func (q *Queries) ListChannelMessageScopeSamples(ctx context.Context, arg ListChannelMessageScopeSamplesParams) ([]ListChannelMessageScopeSamplesRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessageScopeSamples, arg.AfterID, arg.RowLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessageScopeSamplesRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessageScopeSamplesRow
|
|
if err := rows.Scan(
|
|
&i.ChannelID,
|
|
&i.ScopeID,
|
|
&i.Name,
|
|
&i.N,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessagesByIATAs = `-- name: ListChannelMessagesByIATAs :many
|
|
WITH hits AS MATERIALIZED (
|
|
SELECT m.message_id
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT i.message_id FROM channel_message_iatas i
|
|
WHERE i.channel_id = $2::int AND i.iata = req.iata
|
|
AND i.message_id < $3::bigint
|
|
ORDER BY i.message_id DESC
|
|
LIMIT $4) m
|
|
GROUP BY m.message_id
|
|
ORDER BY m.message_id DESC
|
|
LIMIT $4)
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM hits h
|
|
JOIN channel_messages cm ON cm.id = h.message_id
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY cm.id DESC
|
|
`
|
|
|
|
type ListChannelMessagesByIATAsParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListChannelMessagesByIATAsRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// One channel filtered by sites: each site's newest message ids from the read model, merged.
|
|
func (q *Queries) ListChannelMessagesByIATAs(ctx context.Context, arg ListChannelMessagesByIATAsParams) ([]ListChannelMessagesByIATAsRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessagesByIATAs,
|
|
arg.Iatas,
|
|
arg.ChannelID,
|
|
arg.CursorID,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessagesByIATAsRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessagesByIATAsRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessagesByIATAsScoped = `-- name: ListChannelMessagesByIATAsScoped :many
|
|
WITH hits AS MATERIALIZED (
|
|
SELECT m.message_id
|
|
FROM unnest($1::bpchar[]) AS req(iata)
|
|
CROSS JOIN LATERAL (
|
|
SELECT i.message_id FROM channel_message_iatas i
|
|
WHERE i.channel_id = $2::int AND i.iata = req.iata AND i.scope_id = $3::int
|
|
AND i.message_id < $4::bigint
|
|
ORDER BY i.message_id DESC
|
|
LIMIT $5) m
|
|
GROUP BY m.message_id
|
|
ORDER BY m.message_id DESC
|
|
LIMIT $5)
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM hits h
|
|
JOIN channel_messages cm ON cm.id = h.message_id
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY cm.id DESC
|
|
`
|
|
|
|
type ListChannelMessagesByIATAsScopedParams struct {
|
|
Iatas []string `json:"iatas"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
ScopeID int32 `json:"scope_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListChannelMessagesByIATAsScopedRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// One channel filtered by sites and scope.
|
|
func (q *Queries) ListChannelMessagesByIATAsScoped(ctx context.Context, arg ListChannelMessagesByIATAsScopedParams) ([]ListChannelMessagesByIATAsScopedRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessagesByIATAsScoped,
|
|
arg.Iatas,
|
|
arg.ChannelID,
|
|
arg.ScopeID,
|
|
arg.CursorID,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessagesByIATAsScopedRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessagesByIATAsScopedRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessagesPlain = `-- name: ListChannelMessagesPlain :many
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.channel_id = $1::int AND cm.id < $2::bigint
|
|
AND EXISTS (SELECT 1 FROM packet_observations po WHERE po.packet_hash = cm.packet_hash)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $3
|
|
`
|
|
|
|
type ListChannelMessagesPlainParams struct {
|
|
ChannelID int32 `json:"channel_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListChannelMessagesPlainRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// One channel, unfiltered: a walk of the channel by id; a message counts once it has a stored observation.
|
|
func (q *Queries) ListChannelMessagesPlain(ctx context.Context, arg ListChannelMessagesPlainParams) ([]ListChannelMessagesPlainRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessagesPlain, arg.ChannelID, arg.CursorID, arg.PageLimit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessagesPlainRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessagesPlainRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessagesPlainScoped = `-- name: ListChannelMessagesPlainScoped :many
|
|
WITH hits AS MATERIALIZED (
|
|
SELECT DISTINCT i.message_id
|
|
FROM channel_message_iatas i
|
|
WHERE i.channel_id = $1::int AND i.scope_id = $2::int AND i.message_id < $3::bigint
|
|
ORDER BY i.message_id DESC
|
|
LIMIT $4)
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM hits h
|
|
JOIN channel_messages cm ON cm.id = h.message_id
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
ORDER BY cm.id DESC
|
|
`
|
|
|
|
type ListChannelMessagesPlainScopedParams struct {
|
|
ChannelID int32 `json:"channel_id"`
|
|
ScopeID int32 `json:"scope_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListChannelMessagesPlainScopedRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// One channel and scope: the read model has a row exactly when the message has a stored observation.
|
|
func (q *Queries) ListChannelMessagesPlainScoped(ctx context.Context, arg ListChannelMessagesPlainScopedParams) ([]ListChannelMessagesPlainScopedRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessagesPlainScoped,
|
|
arg.ChannelID,
|
|
arg.ScopeID,
|
|
arg.CursorID,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessagesPlainScopedRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessagesPlainScopedRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelMessagesSinceWalk = `-- name: ListChannelMessagesSinceWalk :many
|
|
SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at,
|
|
encode(cm.packet_hash, 'hex') AS packet_hash_hex, c.channel_hash, ts.name AS scope_name,
|
|
p.transport_codes_present, p.observation_count AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.channel_id = $1::int
|
|
AND cm.sent_at >= $2::timestamptz
|
|
AND EXISTS (SELECT 1 FROM packet_observations po WHERE po.packet_hash = cm.packet_hash)
|
|
AND (COALESCE(cardinality($3::bpchar[]), 0) = 0
|
|
OR EXISTS (SELECT 1 FROM channel_message_iatas i WHERE i.message_id = cm.id AND i.iata = ANY($3::bpchar[])))
|
|
AND ($4::int IS NULL OR p.scope_id = $4::int)
|
|
AND cm.id < $5::bigint
|
|
ORDER BY cm.id DESC
|
|
LIMIT $6
|
|
`
|
|
|
|
type ListChannelMessagesSinceWalkParams struct {
|
|
ChannelID int32 `json:"channel_id"`
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
Iatas []string `json:"iatas"`
|
|
ScopeID *int32 `json:"scope_id"`
|
|
CursorID int64 `json:"cursor_id"`
|
|
PageLimit int32 `json:"page_limit"`
|
|
}
|
|
|
|
type ListChannelMessagesSinceWalkRow struct {
|
|
ID int64 `json:"id"`
|
|
ChannelID int32 `json:"channel_id"`
|
|
PacketHash []byte `json:"packet_hash"`
|
|
SenderName *string `json:"sender_name"`
|
|
SenderPubkey []byte `json:"sender_pubkey"`
|
|
Content *string `json:"content"`
|
|
SentAt pgtype.Timestamptz `json:"sent_at"`
|
|
PacketHashHex string `json:"packet_hash_hex"`
|
|
ChannelHash []byte `json:"channel_hash"`
|
|
ScopeName *string `json:"scope_name"`
|
|
TransportCodesPresent *bool `json:"transport_codes_present"`
|
|
ObservationCount int64 `json:"observation_count"`
|
|
}
|
|
|
|
// One channel with since (no client sends it): a channel walk; sites filter through the read model.
|
|
func (q *Queries) ListChannelMessagesSinceWalk(ctx context.Context, arg ListChannelMessagesSinceWalkParams) ([]ListChannelMessagesSinceWalkRow, error) {
|
|
rows, err := q.db.Query(ctx, listChannelMessagesSinceWalk,
|
|
arg.ChannelID,
|
|
arg.Since,
|
|
arg.Iatas,
|
|
arg.ScopeID,
|
|
arg.CursorID,
|
|
arg.PageLimit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListChannelMessagesSinceWalkRow{}
|
|
for rows.Next() {
|
|
var i ListChannelMessagesSinceWalkRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.ChannelID,
|
|
&i.PacketHash,
|
|
&i.SenderName,
|
|
&i.SenderPubkey,
|
|
&i.Content,
|
|
&i.SentAt,
|
|
&i.PacketHashHex,
|
|
&i.ChannelHash,
|
|
&i.ScopeName,
|
|
&i.TransportCodesPresent,
|
|
&i.ObservationCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const maxChannelMessageID = `-- name: MaxChannelMessageID :one
|
|
SELECT COALESCE(max(id), 0)::bigint AS id FROM channel_messages
|
|
`
|
|
|
|
func (q *Queries) MaxChannelMessageID(ctx context.Context) (int64, error) {
|
|
row := q.db.QueryRow(ctx, maxChannelMessageID)
|
|
var id int64
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
const maxObservationID = `-- name: MaxObservationID :one
|
|
SELECT COALESCE(max(id), 0)::bigint AS id FROM packet_observations
|
|
`
|
|
|
|
func (q *Queries) MaxObservationID(ctx context.Context) (int64, error) {
|
|
row := q.db.QueryRow(ctx, maxObservationID)
|
|
var id int64
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
const nextChannelMessageID = `-- name: NextChannelMessageID :one
|
|
SELECT COALESCE(min(id), 0)::bigint AS id FROM channel_messages WHERE id > $1::bigint
|
|
`
|
|
|
|
// The first message id above after_id, or 0 when there is none.
|
|
func (q *Queries) NextChannelMessageID(ctx context.Context, afterID int64) (int64, error) {
|
|
row := q.db.QueryRow(ctx, nextChannelMessageID, afterID)
|
|
var id int64
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|