// Copyright 2026 Beacon Contributors // SPDX-License-Identifier: AGPL-3.0-or-later package db import ( "context" "encoding/hex" "errors" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) func (s *Store) UpsertChannel(ctx context.Context, channelHash []byte, keyFingerprint []byte, name string, hashtag string) (int, error) { var namePtr, hashtagPtr *string if name != "" { namePtr = &name } if hashtag != "" { hashtagPtr = &hashtag } isHashtag := hashtag != "" row, err := s.q.UpsertChannel(ctx, sqlc.UpsertChannelParams{ ChannelHash: channelHash, Column2: keyFingerprint, // key_fingerprint Name: namePtr, Hashtag: hashtagPtr, IsHashtag: &isHashtag, }) if err != nil { return 0, err } return int(row.ID), nil } func (s *Store) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) { rowID, err := s.q.UpsertChannelHashOnly(ctx, channelHash) if err != nil { return 0, err } return int(rowID), nil } // ListUndecryptedGroupTextPackets returns GRP_TXT packets never successfully decrypted -- // see internal/ingest.BackfillChannelMessages, which retries these against the current // keystore at boot. func (s *Store) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ingest.UndecryptedPacket, error) { rows, err := s.q.ListUndecryptedGroupTextPackets(ctx) if err != nil { return nil, err } packets := make([]ingest.UndecryptedPacket, 0, len(rows)) for _, v := range rows { packets = append(packets, ingest.UndecryptedPacket{ PacketHash: v.PacketHash, RawPayload: v.RawPayload, }) } return packets, nil } // ListUndecryptedGroupTextPacketsByHash limits the boot backfill to channels that just gained a key. func (s *Store) ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]ingest.UndecryptedPacket, error) { rows, err := s.q.ListUndecryptedGroupTextPacketsByHash(ctx, hashes) if err != nil { return nil, err } packets := make([]ingest.UndecryptedPacket, 0, len(rows)) for _, v := range rows { packets = append(packets, ingest.UndecryptedPacket{PacketHash: v.PacketHash, RawPayload: v.RawPayload}) } return packets, nil } // SetChannelConfigScopes replaces every configured channel's region placement; // an empty region means Beacon-wide. func (s *Store) SetChannelConfigScopes(ctx context.Context, fingerprints [][]byte, regions []string) error { if err := s.q.DeleteChannelConfigScopes(ctx); err != nil { return err } if len(fingerprints) == 0 { return nil } return s.q.AddChannelConfigScopes(ctx, sqlc.AddChannelConfigScopesParams{Fingerprints: fingerprints, Regions: regions}) } func (s *Store) UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error { return s.q.UpsertChannelIATA(ctx, sqlc.UpsertChannelIATAParams{ ChannelHash: channelHash, Iata: iata, LastHeard: pgtype.Timestamptz{Time: heardAt, Valid: true}, }) } func (s *Store) DeleteOldChannelIATAs(ctx context.Context, cutoff time.Time) error { return s.q.DeleteOldChannelIATAs(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true}) } func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, keyKnown *bool, cursor int64, pageCursor *api.ChannelCursor) (api.ChannelPage, error) { var cursorTS pgtype.Timestamptz if cursor > 0 { cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} } var rows []sqlc.Channel var err error if pageCursor != nil { rows, err = s.q.ListChannelsAfter(ctx, sqlc.ListChannelsAfterParams{ ChannelHash: hash, Iatas: iatas, KeyKnown: keyKnown, PageLimit: limit + 1, CursorTs: pgtype.Timestamptz{Time: pageCursor.LastSeen, Valid: true}, CursorID: pageCursor.ID, }) } else { rows, err = s.q.ListChannels(ctx, sqlc.ListChannelsParams{ ChannelHash: hash, Iatas: iatas, KeyKnown: keyKnown, CursorTs: cursorTS, PageLimit: limit + 1, }) } if err != nil { return api.ChannelPage{}, err } hasMore := len(rows) > int(limit) if hasMore { rows = rows[:limit] } items := make([]api.ChannelSummary, 0, len(rows)) for _, v := range rows { items = append(items, api.ChannelSummary{ ID: int(v.ID), Name: v.Name, ChannelHash: hex.EncodeToString(v.ChannelHash), LastSeen: v.LastSeen.Time.UnixMilli(), IsHashtag: v.IsHashtag != nil && *v.IsHashtag, KeyKnown: v.KeyKnown != nil && *v.KeyKnown, }) } var nextCursor *int64 var nextPageCursor *string if hasMore { last := items[len(items)-1].LastSeen nextCursor = &last row := rows[len(rows)-1] precise := (api.ChannelCursor{LastSeen: row.LastSeen.Time, ID: row.ID}).String() nextPageCursor = &precise } return api.ChannelPage{ Page: api.Page[api.ChannelSummary]{Items: items, NextCursor: nextCursor, HasMore: hasMore}, NextPageCursor: nextPageCursor, }, nil } func (s *Store) GetChannel(ctx context.Context, channelID int32) (*api.Channel, error) { row, err := s.q.GetChannelByID(ctx, channelID) if err != nil { return nil, err } channel := api.Channel{ ChannelSummary: api.ChannelSummary{ ID: int(row.ID), Name: row.Name, ChannelHash: hex.EncodeToString(row.ChannelHash), LastSeen: row.LastSeen.Time.UnixMilli(), IsHashtag: row.IsHashtag != nil && *row.IsHashtag, KeyKnown: row.KeyKnown != nil && *row.KeyKnown, }, Hashtag: row.Hashtag, MessageCount: row.MessageCount, } if row.IsHashtag != nil && *row.IsHashtag && row.KeyFingerprint != nil { fp := hex.EncodeToString(row.KeyFingerprint) channel.KeyFingerprint = &fp } return &channel, nil } func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) (*ingest.InsertedChannelMessage, error) { params := sqlc.InsertChannelMessageParams{ChannelID: int32(m.ChannelID), PacketHash: m.PacketHash, SenderName: &m.SenderName, Content: &m.Content, SentAt: pgtype.Timestamptz{Time: m.SentAt, Valid: true}, Column6: m.Historical} row, err := s.q.InsertChannelMessage(ctx, params) if errors.Is(err, pgx.ErrNoRows) { return nil, nil // duplicate } if err != nil { return nil, err } return &ingest.InsertedChannelMessage{ID: row.ID, Scope: row.ScopeName, ScopeStatus: api.RecordedChannelScopeStatus(row.ScopeName, row.TransportCodesPresent)}, nil } func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) { ts := pgtype.Timestamptz{Time: since, Valid: !since.IsZero()} var messages []api.ChannelMessage var hasMore bool if channelID == nil { rows, err := s.q.ListAllChannelMessages(ctx, sqlc.ListAllChannelMessagesParams{ Column1: ts, Column2: iatas, Column3: scope, Column4: cursor, Limit: limit + 1, }) if err != nil { return api.Page[api.ChannelMessage]{}, err } hasMore = len(rows) > int(limit) if hasMore { rows = rows[:limit] } messages = make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent)) } } else { rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{ ChannelID: *channelID, Column2: ts, Column3: iatas, Column4: scope, Column5: cursor, Limit: limit + 1, }) if err != nil { return api.Page[api.ChannelMessage]{}, err } hasMore = len(rows) > int(limit) if hasMore { rows = rows[:limit] } messages = make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent)) } } var nextCursor *int64 if hasMore && len(messages) > 0 { last := messages[len(messages)-1].ID nextCursor = &last } return api.Page[api.ChannelMessage]{ Items: messages, NextCursor: nextCursor, HasMore: hasMore, }, nil } func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) { rows, err := s.q.ListChannelMessagesByHash(ctx, sqlc.ListChannelMessagesByHashParams{ ChannelHash: hash, Column2: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()}, Column3: iatas, Column4: scope, Column5: cursor, Limit: limit + 1, }) if err != nil { return api.Page[api.ChannelMessage]{}, err } hasMore := len(rows) > int(limit) if hasMore { rows = rows[:limit] } messages := make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { messages = append(messages, toChannelMessage(v.ID, hex.EncodeToString(v.PacketHash), v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent)) } var nextCursor *int64 if hasMore && len(messages) > 0 { last := messages[len(messages)-1].ID nextCursor = &last } return api.Page[api.ChannelMessage]{ Items: messages, NextCursor: nextCursor, HasMore: hasMore, }, nil } func (s *Store) ListMessagesAfterID(ctx context.Context, afterID int64, iatas []string, scope string, limit int32) ([]api.ChannelMessage, error) { rows, err := s.q.ListMessagesAfterID(ctx, sqlc.ListMessagesAfterIDParams{ ID: afterID, Column2: iatas, Column3: scope, Limit: limit, }) if err != nil { return nil, err } items := make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { items = append(items, toChannelMessage( v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent, )) } return items, nil }