beacon-server/db/store.go

// Package db implements the ingest.DB interface using sqlc-generated queries
// over a pgx/v5 connection pool. Each method is a thin mapping layer between
// the ingest param structs and the sqlc-generated param structs.
package db

import (
	"context"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"time"

	sqlc "github.com/MeshCore-Tower/tower-server/db/sqlc"
	"github.com/MeshCore-Tower/tower-server/internal/api"
	"github.com/MeshCore-Tower/tower-server/internal/ingest"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Store struct {
	q *sqlc.Queries
}

func New(pool *pgxpool.Pool) *Store {
	return &Store{q: sqlc.New(pool)}
}

// UpsertObserver upserts the observers row keyed on pubkey.
func (s *Store) UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUID, string, error) {
	row, err := s.q.UpsertObserver(ctx, pubkey)
	if err != nil {
		return uuid.Nil, "", err
	}
	displayName := ""
	if row.DisplayName != nil {
		displayName = *row.DisplayName
	}
	return row.ID, displayName, err
}

// UpsertObserverBroker records that this observer was seen on brokerName.
func (s *Store) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error {
	params := sqlc.UpsertObserverBrokerParams{
		ObserverID: observerID,
		BrokerName: brokerName,
	}
	return s.q.UpsertObserverBroker(ctx, params)
}

// UpsertIATA auto-creates an iata_codes row if it doesn't exist yet.
func (s *Store) UpsertIATA(ctx context.Context, iata string) error {
	return s.q.UpsertIATA(ctx, iata)
}

// UpsertPacket inserts or bumps the packets row. Returns (isNew, error).
func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (bool, error) {
	var regionCode, subRegionCode *int32
	hasTransportCodes := len(p.TransportCodes) == 4
	if hasTransportCodes {
		r := int32(binary.LittleEndian.Uint16(p.TransportCodes[0:2]))
		s := int32(binary.LittleEndian.Uint16(p.TransportCodes[2:4]))
		regionCode = &r
		subRegionCode = &s
	}
	params := sqlc.UpsertPacketParams{
		PacketHash:            p.PacketHash,
		PayloadType:           int16(p.PayloadType),
		PayloadVersion:        int16(p.PayloadVersion),
		RouteType:             int16(p.RouteType),
		TransportCodesPresent: &hasTransportCodes,
		RegionCode:            regionCode,
		SubRegionCode:         subRegionCode,
		OriginPubkey:          p.OriginPubkey,
		RawPayload:            p.RawPayload,
		ParsedPayload:         p.ParsedPayload,
		ChannelHash:           p.ChannelHash,
	}
	row, err := s.q.UpsertPacket(ctx, params)
	if err != nil {
		return false, err
	}
	return row.Inserted, nil
}

// InsertObservation inserts a packet_observations row.
// Returns (inserted, error); inserted=false means ON CONFLICT DO NOTHING fired.
func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservationParams) (bool, error) {
	params := sqlc.InsertObservationParams{
		PacketHash:        o.PacketHash,
		ObserverID:        o.ObserverID,
		Iata:              o.IATA,
		HeardAt:           pgtype.Timestamptz{Time: o.HeardAt, Valid: true},
		PathLengthByte:    int16(o.PathLengthByte),
		HashSize:          int16(o.HashSize),
		HopCount:          int16(o.HopCount),
		PathBytes:         o.PathBytes,
		Rssi:              &o.RSSI,
		Snr:               &o.SNR,
		PropagationTimeMs: &o.PropagationTimeMs,
		RadioFreqMhz:      &o.RadioFreqMHz,
		SpreadFactor:      &o.SpreadFactor,
		BandwidthKhz:      &o.BandwidthKHz,
		CodingRate:        &o.CodingRate,
		SourceBroker:      &o.SourceBroker,
	}
	row, err := s.q.InsertObservation(ctx, params)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, nil // conflict, not an error
	}
	if err != nil {
		return false, err
	}
	return row.ID != 0, nil
}

// SetNodeCapability flips supports_multibyte_paths or supports_multibyte_traces
// for a node, never downgrading an existing TRUE.
func (s *Store) SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, traces bool) error {
	var errs []error
	if paths {
		errs = append(errs, s.q.SetNodeMultibytePaths(ctx, nodeID))
	}
	if traces {
		errs = append(errs, s.q.SetNodeMultibyteTraces(ctx, nodeID))
	}
	return errors.Join(errs...)
}

// UpsertNode upserts a nodes row from an advert payload.
func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams) (uuid.UUID, error) {
	params := sqlc.UpsertNodeParams{
		PublicKey: n.PublicKey,
		NodeType:  int16(n.NodeType),
		Name:      &n.Name,
		Latitude:  n.Latitude,
		Longitude: n.Longitude,
	}
	row, err := s.q.UpsertNode(ctx, params)
	if err != nil {
		return uuid.Nil, err
	}
	return row.ID, nil
}

// UpsertNodeIATA upserts a node_iatas row.
func (s *Store) UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, iata string) error {
	params := sqlc.UpsertNodeIATAParams{NodeID: nodeID, Iata: iata}
	return s.q.UpsertNodeIATA(ctx, params)
}

// InsertChannelMessage stores a decrypted group text message.
func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) 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}}
	return s.q.InsertChannelMessage(ctx, params)
}

// UpdateObserverStatus updates the observer row from a /status message.
// Column2 = display_name, Column3 = observer_type (sqlc loses names inside CASE expressions).
func (s *Store) UpdateObserverStatus(ctx context.Context, p ingest.UpdateObserverStatusParams) (uuid.UUID, error) {
	params := sqlc.UpdateObserverStatusParams{PublicKey: p.PublicKey, Column2: p.DisplayName, Column3: p.ObserverType, SoftwareVersion: &p.SoftwareVersion, HardwareModel: &p.HardwareModel, FirmwareVersion: &p.FirmwareVersion, FirmwareBuild: &p.FirmwareBuild, RadioFreqMhz: &p.RadioFreqMHz, RadioSf: &p.RadioSF, RadioBwKhz: &p.RadioBWKHz, RadioCr: &p.RadioCR, BatteryLevel: p.BatteryLevel, UptimeSeconds: p.UptimeSeconds, StatusMetadata: p.StatusMetadata}
	return s.q.UpdateObserverStatus(ctx, params)
}

// GetObserverLastIATA returns the IATA from the most recent observation for the given observer.
func (s *Store) GetObserverLastIATA(ctx context.Context, observerID uuid.UUID) (string, error) {
	return s.q.GetObserverLastIATA(ctx, observerID)
}

// GetObserverRadio returns the current radio settings for the given observer.
func (s *Store) GetObserverRadio(ctx context.Context, observerID uuid.UUID) (ingest.RadioSettings, error) {
	row, err := s.q.GetObserverRadio(ctx, observerID)
	if err != nil {
		return ingest.RadioSettings{}, err
	}
	var settings ingest.RadioSettings
	if row.RadioFreqMhz != nil {
		settings.FreqMHz = *row.RadioFreqMhz
	}
	if row.RadioSf != nil {
		settings.SF = *row.RadioSf
	}
	if row.RadioBwKhz != nil {
		settings.BWKHz = *row.RadioBwKhz
	}
	if row.RadioCr != nil {
		settings.CR = *row.RadioCr
	}
	return settings, nil
}

// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
// Hash size is inferred from the length of the first element in hashes.
func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) ([]uuid.UUID, error) {
	rows, err := s.q.ResolvePathHashes(ctx, sqlc.ResolvePathHashesParams{
		Iata:    iata,
		Column2: hashes,
	})
	if err != nil {
		return nil, err
	}

	ids := make([]uuid.UUID, len(rows))
	copy(ids, rows)
	return ids, nil
}

// UpsertChannel upserts a channel row by (hash, keyFingerprint) and returns its integer ID.
// Pass nil keyFingerprint to record a hash-only row when the key is unknown.
// name and hashtag are optional metadata stored on the channel row.
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,
		MessageCount: nil, // message count bumped separately by InsertChannelMessage
	})
	if err != nil {
		return 0, err
	}
	return int(row.ID), nil
}

// ListIATAs returns all known IATA codes with display name and coordinates.
// IATAs are auto-created on first packet arrival from that location.
func (s *Store) ListIATAs(ctx context.Context) ([]api.IATA, error) {
	rows, err := s.q.ListIATAs(ctx)
	if err != nil {
		return nil, err
	}
	iatas := make([]api.IATA, 0, len(rows))
	for _, v := range rows {
		iatas = append(iatas, api.IATA{
			IATA:        v.Iata,
			DisplayName: v.DisplayName,
			Lat:         v.ApproxLat,
			Lng:         v.ApproxLng,
		})
	}
	return iatas, nil
}

// GetIATA returns a single IATA code by its 3-letter identifier.
// Returns nil, error if the IATA code is not found.
func (s *Store) GetIATA(ctx context.Context, iata string) (*api.IATA, error) {
	i, err := s.q.GetIATA(ctx, iata)
	if err != nil {
		return nil, err
	}
	return &api.IATA{
		IATA:        i.Iata,
		DisplayName: i.DisplayName,
		Lat:         i.ApproxLat,
		Lng:         i.ApproxLng,
	}, nil
}

// ListRegions returns a summary list of all regions ordered by display_order then name.
// Use GetRegion for full detail including associated IATAs.
func (s *Store) ListRegions(ctx context.Context) ([]api.RegionSummary, error) {
	rows, err := s.q.ListRegions(ctx)
	if err != nil {
		return nil, err
	}
	regions := make([]api.RegionSummary, 0, len(rows))
	for _, v := range rows {
		regions = append(regions, api.RegionSummary{
			ID:   int(v.ID),
			Slug: v.Slug,
			Name: v.Name,
		})
	}
	return regions, nil
}

// GetRegion returns full detail for a single region including its associated IATA codes.
// Returns nil, pgx.ErrNoRows if the region is not found.
func (s *Store) GetRegion(ctx context.Context, regionID int32) (*api.Region, error) {
	region, err := s.q.GetRegion(ctx, regionID)
	if err != nil {
		return nil, err
	}
	result := api.Region{
		RegionSummary: api.RegionSummary{
			ID:   int(region.ID),
			Slug: region.Slug,
			Name: region.Name,
		},
		Description: region.Description,
		CenterLat:   region.CenterLat,
		CenterLng:   region.CenterLng,
	}
	var zoomLevel *int
	if region.ZoomLevel != nil {
		z := int(*region.ZoomLevel)
		zoomLevel = &z
	}
	result.ZoomLevel = zoomLevel
	iatas, err := s.q.GetRegionIATAs(ctx, regionID)
	if err != nil {
		return nil, err
	}
	result.IATAs = iatas
	return &result, nil
}

// UpsertIATADetails updates an existing iata_codes row with display name and coordinates.
// The row must already exist (auto-created on first packet arrival).
// Safe to call on startup — does nothing if the IATA has not been seen yet.
func (s *Store) UpsertIATADetails(ctx context.Context, iata string, name string, lat, lng *float64) error {
	return s.q.UpsertIATADetails(ctx, sqlc.UpsertIATADetailsParams{
		Iata:        iata,
		DisplayName: &name,
		ApproxLat:   lat,
		ApproxLng:   lng,
	})
}

// UpsertRegion inserts or updates a region row by slug. Returns the region ID.
func (s *Store) UpsertRegion(ctx context.Context, slug, name, description string, displayOrder int, centerLat, centerLng *float64, zoomLevel *int) (int32, error) {
	var zl *int32
	if zoomLevel != nil {
		z := int32(*zoomLevel)
		zl = &z
	}
	do := int32(displayOrder)
	return s.q.UpsertRegion(ctx, sqlc.UpsertRegionParams{
		Slug:         slug,
		Name:         name,
		Description:  &description,
		DisplayOrder: &do,
		CenterLat:    centerLat,
		CenterLng:    centerLng,
		ZoomLevel:    zl,
	})
}

// UpsertRegionIATA adds an IATA code to a region. Safe to call repeatedly.
func (s *Store) UpsertRegionIATA(ctx context.Context, regionID int32, iata string) error {
	return s.q.UpsertRegionIATA(ctx, sqlc.UpsertRegionIATAParams{
		RegionID: regionID,
		Iata:     iata,
	})
}

// ListChannels returns a summary list of all known channels ordered by last seen.
// Includes both hashtag-derived and explicit key channels.
// Channels with unknown keys are included with KeyKnown=false.
// Filters on hash if provided, this is the hex channel hash
func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte) ([]api.ChannelSummary, error) {
	var rows []sqlc.Channel
	if hash != nil {
		r, err := s.q.GetChannelsByHash(ctx, sqlc.GetChannelsByHashParams{
			ChannelHash: hash,
			Limit:       limit,
		})
		if err != nil {
			return nil, err
		}
		rows = r
	} else {
		r, err := s.q.ListChannels(ctx, limit)
		if err != nil {
			return nil, err
		}
		rows = r
	}
	channels := make([]api.ChannelSummary, 0, len(rows))
	for _, v := range rows {
		channels = append(channels, api.ChannelSummary{
			ID:          int(v.ID),
			Name:        v.Name,
			ChannelHash: hex.EncodeToString(v.ChannelHash),
			LastSeen:    v.LastSeen.Time.Format(time.RFC3339),
			IsHashtag:   v.IsHashtag != nil && *v.IsHashtag,
			KeyKnown:    v.KeyKnown != nil && *v.KeyKnown,
		})
	}

	return channels, nil
}

// GetChannel returns full detail for a single channel by its integer ID.
// Returns nil, pgx.ErrNoRows if the channel is not found.
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.Format(time.RFC3339),
			IsHashtag:   row.IsHashtag != nil && *row.IsHashtag,
			KeyKnown:    row.KeyKnown != nil && *row.KeyKnown,
		},
		Hashtag:      row.Hashtag,
		MessageCount: 0,
	}
	if row.MessageCount != nil {
		channel.MessageCount = *row.MessageCount
	}
	if row.IsHashtag != nil && *row.IsHashtag && row.KeyFingerprint != nil {
		fp := hex.EncodeToString(row.KeyFingerprint)
		channel.KeyFingerprint = &fp
	}
	return &channel, nil
}

// ListChannelMessages returns paginated messages for a channel identified by its integer ID.
// Used by the /channels/{id}/messages endpoint.
// Pass a zero time.Time for since to return all messages up to limit.
func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32) ([]api.ChannelMessage, error) {
	var messages []api.ChannelMessage
	if channelID == nil {
		rows, err := s.q.ListAllChannelMessages(ctx, sqlc.ListAllChannelMessagesParams{
			Column1: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()},
			Limit:   limit,
		})
		messages = make([]api.ChannelMessage, 0, len(rows))
		if err != nil {
			return nil, err
		}
		for _, v := range rows {
			messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt))
		}
	} else {
		rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{
			ChannelID: *channelID,
			Column2:   pgtype.Timestamptz{Time: since, Valid: !since.IsZero()},
			Limit:     limit,
		})
		messages = make([]api.ChannelMessage, 0, len(rows))
		if err != nil {
			return nil, err
		}
		for _, v := range rows {
			messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt))
		}
	}
	return messages, nil
}

// ListChannelMessagesByHash returns paginated messages for all channels matching the given hash.
// Used by the /messages?hash= endpoint. May return messages from multiple channels
// if the hash collides across different keys.
// Pass a zero time.Time for since to return all messages up to limit.
func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32) ([]api.ChannelMessage, error) {
	rows, err := s.q.ListChannelMessagesByHash(ctx, sqlc.ListChannelMessagesByHashParams{
		ChannelHash: hash,
		Column2:     pgtype.Timestamptz{Time: since, Valid: !since.IsZero()},
		Limit:       limit,
	})
	if err != nil {
		return nil, err
	}
	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))
	}
	return messages, nil
}

func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz) api.ChannelMessage {
	sn := ""
	if senderName != nil {
		sn = *senderName
	}
	ct := ""
	if content != nil {
		ct = *content
	}
	return api.ChannelMessage{
		ID:          id,
		PacketHash:  packetHashHex,
		ChannelHash: hex.EncodeToString(channelHash),
		SenderName:  sn,
		Content:     ct,
		SentAt:      sentAt.Time.Format(time.RFC3339),
	}
}